142 lines
6.0 KiB
Python
142 lines
6.0 KiB
Python
# --- START OF FILE conversation_history.py ---
|
|
|
|
import os
|
|
import json
|
|
from collections import defaultdict, deque
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, Any, List
|
|
import logging # Added logging
|
|
# Import global config defaults, channel specific config might override lengths elsewhere
|
|
from config import BOT_USER_ID, CONVERSATION_TIMEOUT_MINUTES, MAX_HISTORY_LENGTH as GLOBAL_MAX_HISTORY_LENGTH
|
|
|
|
# Conversation (messages section of the prompt) history storage
|
|
# Keyed by channel_id
|
|
conversation_histories = defaultdict(lambda: deque(maxlen=GLOBAL_MAX_HISTORY_LENGTH)) # Use deque for efficiency
|
|
|
|
# Last message times for each channel, for conversation timeout
|
|
last_message_times = {}
|
|
|
|
def update_llm_conversation_history(inputmessage, bot_identifier="unknown", channel_max_length: Optional[int] = None):
|
|
"""
|
|
Update the conversation history for a specific channel.
|
|
Handles both regular text messages and tool-related messages with structured content.
|
|
Uses global timeout and default max length, but deque handles automatic trimming.
|
|
|
|
Args:
|
|
inputmessage (dict): Message object from the user or LLM.
|
|
bot_identifier (str): Identifier for debug logging (e.g., "techsupport").
|
|
"""
|
|
channel = inputmessage.get('channel')
|
|
if not channel:
|
|
logging.error("Attempted to update history with no channel ID.")
|
|
return None # Cannot proceed without channel
|
|
|
|
user = inputmessage.get('user')
|
|
current_time = datetime.now()
|
|
|
|
# Check for timeout and clear history if needed
|
|
if channel in last_message_times:
|
|
last_time = last_message_times[channel]
|
|
if (current_time - last_time) > timedelta(minutes=CONVERSATION_TIMEOUT_MINUTES):
|
|
logging.info(f"[{bot_identifier}/{channel}] Conversation timed out. Clearing history.")
|
|
conversation_histories[channel].clear() # Clear the deque
|
|
|
|
# Update last message time
|
|
last_message_times[channel] = current_time
|
|
|
|
# Determine the role based on the user ID
|
|
role = "assistant" if user == BOT_USER_ID else "user"
|
|
|
|
new_message = None # Initialize
|
|
|
|
# Check message structure to determine format
|
|
content = inputmessage.get('content')
|
|
if isinstance(content, list):
|
|
# This could be a tool_use request from assistant OR tool_result from user simulation
|
|
# The role should already be correctly determined above based on 'user' field
|
|
new_message = {
|
|
"role": role,
|
|
"content": content # Keep the list structure
|
|
}
|
|
elif isinstance(inputmessage.get('text'), str):
|
|
# Regular text message from user or assistant
|
|
text = inputmessage.get('text', '')
|
|
new_message = {
|
|
"role": role,
|
|
"content": text
|
|
}
|
|
else:
|
|
logging.warning(f"[{bot_identifier}/{channel}] Unrecognized message format for history: {inputmessage}")
|
|
# Optionally create a placeholder or skip
|
|
new_message = {
|
|
"role": role,
|
|
"content": "[Unrecognized message format]"
|
|
}
|
|
|
|
# --- START NEW TRIM LOGIC ---
|
|
# Use the specific channel length if provided, otherwise fallback to global default (though ideally it's always provided now)
|
|
current_maxlen = channel_max_length if channel_max_length is not None else GLOBAL_MAX_HISTORY_LENGTH
|
|
|
|
# Ensure deque uses the correct maxlen (it might have been created with the default)
|
|
# This seems inefficient to do every time, but deque doesn't easily allow changing maxlen.
|
|
# A potential optimization is to store deques with specific maxlens from the start,
|
|
# but let's try the simpler approach first.
|
|
# We'll manually trim *before* appending if needed.
|
|
while len(conversation_histories[channel]) >= current_maxlen:
|
|
try:
|
|
conversation_histories[channel].popleft() # Remove the oldest message
|
|
logging.debug(f"[{bot_identifier}/{channel}] History limit ({current_maxlen}) reached. Popped oldest message.")
|
|
except IndexError: # Should not happen with deque, but safety first
|
|
break
|
|
# --- END NEW TRIM LOGIC ---
|
|
|
|
# Add new message to history deque (automatically handles max length)
|
|
if new_message:
|
|
conversation_histories[channel].append(new_message)
|
|
logging.debug(f"[{bot_identifier}/{channel}] History updated. New length: {len(conversation_histories[channel])}")
|
|
|
|
|
|
# write the updated conversation history to a file for debugging.
|
|
debug_dir = "debug"
|
|
os.makedirs(debug_dir, exist_ok=True) # Ensure debug directory exists
|
|
debug_history_filename = os.path.join(debug_dir, f'conversation_history-{bot_identifier}.txt')
|
|
try:
|
|
with open(debug_history_filename, 'w', encoding='utf-8') as f:
|
|
# Convert deque to list for JSON serialization
|
|
history_list = list(conversation_histories[channel])
|
|
f.write(json.dumps(history_list, indent=2))
|
|
except Exception as e:
|
|
logging.error(f"[{bot_identifier}/{channel}] Failed to write debug history file {debug_history_filename}: {e}")
|
|
|
|
|
|
# Return the current history as a list
|
|
return list(conversation_histories[channel])
|
|
|
|
def get_conversation_history(channel_id):
|
|
"""
|
|
Get the current conversation history for a specific channel as a list.
|
|
|
|
Args:
|
|
channel_id (str): The channel ID to get conversation history for
|
|
|
|
Returns:
|
|
list: The conversation history for the specified channel (empty list if none)
|
|
"""
|
|
return list(conversation_histories.get(channel_id, []))
|
|
|
|
def clear_conversation_history(channel_id):
|
|
"""
|
|
Clear the conversation history for a specific channel.
|
|
|
|
Args:
|
|
channel_id (str): The channel ID to clear conversation history for
|
|
"""
|
|
if channel_id in conversation_histories:
|
|
conversation_histories[channel_id].clear()
|
|
logging.info(f"Conversation history cleared for channel {channel_id}")
|
|
if channel_id in last_message_times:
|
|
del last_message_times[channel_id] # Reset timeout timer too
|
|
return True
|
|
return False
|
|
|
|
# --- END OF FILE conversation_history.py --- |