"""
Conversation history and analytics for the Enhanced Matrix integration in OllamaTools.
This module provides functionality for tracking conversation history,
analyzing topics, and generating performance metrics for the Enhanced Matrix client.
Dependencies:
- matrix-nio: For Matrix protocol support
- ollama: For Ollama API access
"""
import os
import json
import time
import logging
from typing import Dict, List, Optional, Any, Set, Tuple
from collections import defaultdict
# Configure logging
logger = logging.getLogger(__name__)
def load_conversation_history(history_file: str) -> List[Dict[str, Any]]:
"""
Load conversation history from file.
Args:
history_file: Path to the history file
Returns:
List[Dict[str, Any]]: The conversation history
"""
if os.path.exists(history_file):
try:
with open(history_file, 'r') as f:
history = json.load(f)
logger.info(f"Loaded {len(history)} conversations from history")
return history
except Exception as e:
logger.error(f"Failed to load conversation history: {str(e)}")
return []
else:
logger.info("No conversation history file found")
return []
def save_conversation_history(history: List[Dict[str, Any]], history_file: str) -> None:
"""
Save conversation history to file.
Args:
history: The conversation history
history_file: Path to save the history to
"""
try:
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(history_file), exist_ok=True)
with open(history_file, 'w') as f:
json.dump(history, f, indent=2)
logger.debug(f"Saved {len(history)} conversations to history")
except Exception as e:
logger.error(f"Failed to save conversation history: {str(e)}")
def record_conversation(
history: List[Dict[str, Any]],
user_message: str,
assistant_response: str,
sender: str,
room_id: str,
response_time: float,
history_file: str,
max_history: int = 100
) -> None:
"""
Record a conversation in the history.
Args:
history: The conversation history list to update
user_message: The user's message
assistant_response: The assistant's response
sender: The Matrix user ID of the sender
room_id: The ID of the room where the conversation took place
response_time: The time taken to generate the response
history_file: Path to save the history to
max_history: Maximum number of conversations to keep in history
"""
# Create a conversation record
conversation = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"user_message": user_message,
"assistant_response": assistant_response,
"sender": sender,
"room_id": room_id,
"response_time": response_time,
"tokens_estimated": len(user_message.split()) + len(assistant_response.split())
}
# Add to conversation history
history.append(conversation)
# Trim conversation history if it gets too long
if len(history) > max_history:
history = history[-max_history:]
# Save conversation history
save_conversation_history(history, history_file)
def analyze_conversation_topics(conversations: List[Dict[str, Any]]) -> Dict[str, int]:
"""
Analyze common topics in conversation history.
Args:
conversations: List of conversation records
Returns:
Dict[str, int]: Dictionary of topics and their frequency
"""
if not conversations:
return {}
# Simple keyword-based topic analysis
topics = {
"help": 0,
"search": 0,
"file": 0,
"command": 0,
"code": 0,
"error": 0,
"issue": 0,
"question": 0,
"other": 0
}
for conv in conversations:
message = conv.get("user_message", "").lower()
# Check for keywords
if "help" in message or "how" in message:
topics["help"] += 1
if "search" in message or "find" in message or "look" in message:
topics["search"] += 1
if "file" in message or "read" in message or "write" in message:
topics["file"] += 1
if "command" in message or "run" in message or "execute" in message:
topics["command"] += 1
if "code" in message or "program" in message or "script" in message:
topics["code"] += 1
if "error" in message or "bug" in message or "fix" in message:
topics["error"] += 1
if "issue" in message or "problem" in message:
topics["issue"] += 1
if "?" in message:
topics["question"] += 1
# If no keywords matched, count as other
if not any(count > 0 for count in topics.values()):
topics["other"] += 1
# Remove topics with zero count
return {k: v for k, v in topics.items() if v > 0}
def get_performance_metrics(conversations: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Get performance metrics from conversation history.
Args:
conversations: List of conversation records
Returns:
Dict[str, Any]: Performance metrics
"""
if not conversations:
return {"error": "No conversation history available"}
# Calculate metrics
total_conversations = len(conversations)
total_response_time = sum(conv.get("response_time", 0) for conv in conversations)
avg_response_time = total_response_time / total_conversations if total_conversations > 0 else 0
# Calculate token usage
total_tokens = sum(conv.get("tokens_estimated", 0) for conv in conversations)
avg_tokens_per_conversation = total_tokens / total_conversations if total_conversations > 0 else 0
# Find slowest and fastest responses
slowest = max(conversations, key=lambda x: x.get("response_time", 0)) if conversations else None
fastest = min(conversations, key=lambda x: x.get("response_time", 0)) if conversations else None
return {
"total_conversations": total_conversations,
"total_response_time": total_response_time,
"avg_response_time": avg_response_time,
"total_tokens_estimated": total_tokens,
"avg_tokens_per_conversation": avg_tokens_per_conversation,
"slowest_response": {
"time": slowest.get("response_time", 0) if slowest else 0,
"message": slowest.get("user_message", "")[:50] + "..." if slowest and len(slowest.get("user_message", "")) > 50 else slowest.get("user_message", "") if slowest else ""
},
"fastest_response": {
"time": fastest.get("response_time", 0) if fastest else 0,
"message": fastest.get("user_message", "")[:50] + "..." if fastest and len(fastest.get("user_message", "")) > 50 else fastest.get("user_message", "") if fastest else ""
}
}
async def handle_history_command(client: Any, count: int, room_id: str) -> None:
"""
Handle the /history command.
Args:
client: The EnhancedMatrixClient instance
count: Number of recent conversations to show
room_id: The ID of the room where the command was sent
"""
if not client.enable_conversation_tracking:
await client.send_message(room_id, "Conversation tracking is not enabled.")
return
if not client.conversation_history:
await client.send_message(room_id, "No conversation history available.")
return
# Get the most recent conversations for this room
room_history = [conv for conv in client.conversation_history if conv.get("room_id") == room_id]
if not room_history:
await client.send_message(room_id, "No conversation history available for this room.")
return
# Get the most recent conversations
recent = room_history[-count:] if len(room_history) >= count else room_history
# Format the history
formatted_history = ["Recent Conversations:"]
for i, conv in enumerate(recent, 1):
timestamp = conv.get("timestamp", "Unknown time")
user_message = conv.get("user_message", "No message")
sender = conv.get("sender", "Unknown sender")
# Truncate message if it's too long
if len(user_message) > 100:
user_message = user_message[:100] + "..."
formatted_history.append(f"{i}. {timestamp}")
formatted_history.append(f" From: {sender}")
formatted_history.append(f" Message: {user_message}")
formatted_history.append("")
# Send the formatted history
await client.send_message(room_id, "\n".join(formatted_history))
async def handle_topics_command(client: Any, room_id: str) -> None:
"""
Handle the /topics command.
Args:
client: The EnhancedMatrixClient instance
room_id: The ID of the room where the command was sent
"""
if not client.enable_conversation_tracking:
await client.send_message(room_id, "Conversation tracking is not enabled.")
return
if not client.conversation_history:
await client.send_message(room_id, "No conversation history available.")
return
# Get the conversations for this room
room_history = [conv for conv in client.conversation_history if conv.get("room_id") == room_id]
if not room_history:
await client.send_message(room_id, "No conversation history available for this room.")
return
# Analyze topics
topics = analyze_conversation_topics(room_history)
if not topics:
await client.send_message(room_id, "No topics found in conversation history.")
return
# Format the topics
formatted_topics = ["Conversation Topics:"]
for topic, count in sorted(topics.items(), key=lambda x: x[1], reverse=True):
formatted_topics.append(f"- {topic}: {count} mentions")
# Send the formatted topics
await client.send_message(room_id, "\n".join(formatted_topics))
async def handle_metrics_command(client: Any, room_id: str) -> None:
"""
Handle the /metrics command.
Args:
client: The EnhancedMatrixClient instance
room_id: The ID of the room where the command was sent
"""
if not client.enable_conversation_tracking:
await client.send_message(room_id, "Conversation tracking is not enabled.")
return
if not client.conversation_history:
await client.send_message(room_id, "No conversation history available.")
return
# Get the conversations for this room
room_history = [conv for conv in client.conversation_history if conv.get("room_id") == room_id]
if not room_history:
await client.send_message(room_id, "No conversation history available for this room.")
return
# Calculate metrics
metrics = get_performance_metrics(room_history)
# Format the metrics
formatted_metrics = ["Performance Metrics:"]
formatted_metrics.append(f"- Total conversations: {metrics['total_conversations']}")
formatted_metrics.append(f"- Average response time: {metrics['avg_response_time']:.2f}s")
formatted_metrics.append(f"- Estimated total tokens: {metrics['total_tokens_estimated']}")
formatted_metrics.append(f"- Average tokens per conversation: {metrics['avg_tokens_per_conversation']:.1f}")
if metrics.get("slowest_response"):
formatted_metrics.append(f"- Slowest response: {metrics['slowest_response']['time']:.2f}s for '{metrics['slowest_response']['message']}'")
if metrics.get("fastest_response"):
formatted_metrics.append(f"- Fastest response: {metrics['fastest_response']['time']:.2f}s for '{metrics['fastest_response']['message']}'")
# Send the formatted metrics
await client.send_message(room_id, "\n".join(formatted_metrics))