truetype
eda22811ce
- Renamed all references from maxclaw to langcode across all files - Renamed maxclaw.py → langcode.py - Renamed completion scripts: maxclaw.bash/zsh → langcode.bash/zsh - Updated Dockerfile, README, all Python files, HTML, JSON, YAML - New Gitea repository: http://192.168.0.183:8085/truetype/langcode_agent.git
494 lines
20 KiB
Python
494 lines
20 KiB
Python
"""
|
|
Advanced analytics for LangCode Agent
|
|
Provides usage analytics, insights, and reporting
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Any, Optional
|
|
from pathlib import Path
|
|
from collections import defaultdict, Counter
|
|
|
|
# Add current directory to path
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
class AnalyticsEngine:
|
|
"""Advanced analytics engine for LangCode Agent"""
|
|
|
|
def __init__(self, data_dir: Optional[str] = None):
|
|
"""
|
|
Initialize analytics engine
|
|
|
|
Args:
|
|
data_dir: Directory for storing analytics data
|
|
"""
|
|
if data_dir is None:
|
|
self.data_dir = Path.home() / ".langcode" / "analytics"
|
|
else:
|
|
self.data_dir = Path(data_dir)
|
|
|
|
# Create data directory
|
|
self.data_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Analytics data
|
|
self.usage_data = {
|
|
"conversations": [],
|
|
"tool_usage": defaultdict(int),
|
|
"provider_usage": defaultdict(int),
|
|
"command_usage": defaultdict(int),
|
|
"error_patterns": defaultdict(int),
|
|
"response_times": [],
|
|
"user_sessions": []
|
|
}
|
|
|
|
# Load existing data
|
|
self._load_data()
|
|
|
|
def _load_data(self):
|
|
"""Load analytics data from file"""
|
|
data_file = self.data_dir / "analytics_data.json"
|
|
|
|
if data_file.exists():
|
|
try:
|
|
with open(data_file, 'r') as f:
|
|
data = json.load(f)
|
|
# Convert defaultdicts back
|
|
self.usage_data["tool_usage"] = defaultdict(int, data.get("tool_usage", {}))
|
|
self.usage_data["provider_usage"] = defaultdict(int, data.get("provider_usage", {}))
|
|
self.usage_data["command_usage"] = defaultdict(int, data.get("command_usage", {}))
|
|
self.usage_data["error_patterns"] = defaultdict(int, data.get("error_patterns", {}))
|
|
self.usage_data["conversations"] = data.get("conversations", [])
|
|
self.usage_data["response_times"] = data.get("response_times", [])
|
|
self.usage_data["user_sessions"] = data.get("user_sessions", [])
|
|
except Exception as e:
|
|
print(f"Warning: Could not load analytics data: {e}")
|
|
|
|
def _save_data(self):
|
|
"""Save analytics data to file"""
|
|
data_file = self.data_dir / "analytics_data.json"
|
|
|
|
try:
|
|
# Convert defaultdicts to regular dicts for JSON serialization
|
|
data_to_save = {
|
|
"conversations": self.usage_data["conversations"],
|
|
"tool_usage": dict(self.usage_data["tool_usage"]),
|
|
"provider_usage": dict(self.usage_data["provider_usage"]),
|
|
"command_usage": dict(self.usage_data["command_usage"]),
|
|
"error_patterns": dict(self.usage_data["error_patterns"]),
|
|
"response_times": self.usage_data["response_times"],
|
|
"user_sessions": self.usage_data["user_sessions"]
|
|
}
|
|
|
|
with open(data_file, 'w') as f:
|
|
json.dump(data_to_save, f, indent=2)
|
|
except Exception as e:
|
|
print(f"Warning: Could not save analytics data: {e}")
|
|
|
|
def record_conversation(self, user_message: str, agent_response: str,
|
|
provider: str, model: str, response_time: float):
|
|
"""Record a conversation for analytics"""
|
|
conversation = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"user_message": user_message[:100], # Truncate for privacy
|
|
"agent_response": agent_response[:100], # Truncate for privacy
|
|
"provider": provider,
|
|
"model": model,
|
|
"response_time": response_time,
|
|
"message_length": len(user_message),
|
|
"response_length": len(agent_response)
|
|
}
|
|
|
|
self.usage_data["conversations"].append(conversation)
|
|
self.usage_data["response_times"].append(response_time)
|
|
|
|
# Keep only last 1000 conversations
|
|
if len(self.usage_data["conversations"]) > 1000:
|
|
self.usage_data["conversations"] = self.usage_data["conversations"][-1000:]
|
|
|
|
# Keep only last 5000 response times
|
|
if len(self.usage_data["response_times"]) > 5000:
|
|
self.usage_data["response_times"] = self.usage_data["response_times"][-5000:]
|
|
|
|
self._save_data()
|
|
|
|
def record_tool_usage(self, tool_name: str, success: bool, duration: float):
|
|
"""Record tool usage"""
|
|
self.usage_data["tool_usage"][tool_name] += 1
|
|
self._save_data()
|
|
|
|
def record_provider_usage(self, provider: str, model: str):
|
|
"""Record provider usage"""
|
|
key = f"{provider}:{model}"
|
|
self.usage_data["provider_usage"][key] += 1
|
|
self._save_data()
|
|
|
|
def record_command_usage(self, command: str):
|
|
"""Record command usage"""
|
|
self.usage_data["command_usage"][command] += 1
|
|
self._save_data()
|
|
|
|
def record_error(self, error_type: str, error_message: str):
|
|
"""Record error patterns"""
|
|
# Simplify error message for pattern recognition
|
|
simplified = error_message[:50].lower()
|
|
self.usage_data["error_patterns"][simplified] += 1
|
|
self._save_data()
|
|
|
|
def record_user_session(self, session_id: str, duration: float,
|
|
messages_count: int, tools_used: List[str]):
|
|
"""Record user session"""
|
|
session = {
|
|
"session_id": session_id,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"duration": duration,
|
|
"messages_count": messages_count,
|
|
"tools_used": tools_used,
|
|
"tools_count": len(tools_used)
|
|
}
|
|
|
|
self.usage_data["user_sessions"].append(session)
|
|
|
|
# Keep only last 100 sessions
|
|
if len(self.usage_data["user_sessions"]) > 100:
|
|
self.usage_data["user_sessions"] = self.usage_data["user_sessions"][-100:]
|
|
|
|
self._save_data()
|
|
|
|
def get_usage_summary(self, days: int = 30) -> Dict[str, Any]:
|
|
"""
|
|
Get usage summary for the last N days
|
|
|
|
Args:
|
|
days: Number of days to include
|
|
|
|
Returns:
|
|
Dictionary with usage summary
|
|
"""
|
|
cutoff_time = datetime.now() - timedelta(days=days)
|
|
|
|
# Filter recent conversations
|
|
recent_conversations = [
|
|
c for c in self.usage_data["conversations"]
|
|
if datetime.fromisoformat(c["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
# Calculate statistics
|
|
total_conversations = len(recent_conversations)
|
|
total_messages = sum(c.get("message_length", 0) for c in recent_conversations)
|
|
total_responses = sum(c.get("response_length", 0) for c in recent_conversations)
|
|
|
|
# Provider usage
|
|
provider_counts = Counter()
|
|
for conv in recent_conversations:
|
|
provider = conv.get("provider", "unknown")
|
|
provider_counts[provider] += 1
|
|
|
|
# Model usage
|
|
model_counts = Counter()
|
|
for conv in recent_conversations:
|
|
model = conv.get("model", "unknown")
|
|
model_counts[model] += 1
|
|
|
|
# Response time statistics
|
|
recent_response_times = [
|
|
c.get("response_time", 0) for c in recent_conversations
|
|
if c.get("response_time")
|
|
]
|
|
|
|
if recent_response_times:
|
|
avg_response_time = sum(recent_response_times) / len(recent_response_times)
|
|
min_response_time = min(recent_response_times)
|
|
max_response_time = max(recent_response_times)
|
|
else:
|
|
avg_response_time = min_response_time = max_response_time = 0
|
|
|
|
# Tool usage
|
|
tool_usage = dict(self.usage_data["tool_usage"])
|
|
|
|
# Command usage
|
|
command_usage = dict(self.usage_data["command_usage"])
|
|
|
|
# Error patterns
|
|
error_patterns = dict(self.usage_data["error_patterns"])
|
|
|
|
# Session statistics
|
|
recent_sessions = [
|
|
s for s in self.usage_data["user_sessions"]
|
|
if datetime.fromisoformat(s["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
total_sessions = len(recent_sessions)
|
|
total_session_duration = sum(s.get("duration", 0) for s in recent_sessions)
|
|
avg_session_duration = total_session_duration / total_sessions if total_sessions > 0 else 0
|
|
|
|
return {
|
|
"period_days": days,
|
|
"conversations": {
|
|
"total": total_conversations,
|
|
"total_messages_chars": total_messages,
|
|
"total_responses_chars": total_responses,
|
|
"avg_message_length": total_messages / total_conversations if total_conversations > 0 else 0,
|
|
"avg_response_length": total_responses / total_conversations if total_conversations > 0 else 0
|
|
},
|
|
"providers": {
|
|
"usage": dict(provider_counts),
|
|
"most_used": provider_counts.most_common(1)[0][0] if provider_counts else "None"
|
|
},
|
|
"models": {
|
|
"usage": dict(model_counts),
|
|
"most_used": model_counts.most_common(1)[0][0] if model_counts else "None"
|
|
},
|
|
"performance": {
|
|
"avg_response_time": avg_response_time,
|
|
"min_response_time": min_response_time,
|
|
"max_response_time": max_response_time,
|
|
"total_response_time": sum(recent_response_times)
|
|
},
|
|
"tools": {
|
|
"usage": tool_usage,
|
|
"most_used": max(tool_usage.items(), key=lambda x: x[1])[0] if tool_usage else "None"
|
|
},
|
|
"commands": {
|
|
"usage": command_usage,
|
|
"most_used": max(command_usage.items(), key=lambda x: x[1])[0] if command_usage else "None"
|
|
},
|
|
"errors": {
|
|
"patterns": error_patterns,
|
|
"most_common": max(error_patterns.items(), key=lambda x: x[1])[0] if error_patterns else "None"
|
|
},
|
|
"sessions": {
|
|
"total": total_sessions,
|
|
"total_duration": total_session_duration,
|
|
"avg_duration": avg_session_duration,
|
|
"avg_messages_per_session": sum(s.get("messages_count", 0) for s in recent_sessions) / total_sessions if total_sessions > 0 else 0
|
|
},
|
|
"generated_at": datetime.now().isoformat()
|
|
}
|
|
|
|
def get_detailed_report(self, days: int = 30) -> str:
|
|
"""
|
|
Get detailed analytics report
|
|
|
|
Args:
|
|
days: Number of days to include
|
|
|
|
Returns:
|
|
Formatted report string
|
|
"""
|
|
summary = self.get_usage_summary(days)
|
|
|
|
report = f"📊 Advanced Analytics Report (Last {days} days)\n"
|
|
report += "=" * 60 + "\n\n"
|
|
|
|
# Conversation statistics
|
|
report += "💬 Conversation Statistics:\n"
|
|
report += f" Total Conversations: {summary['conversations']['total']}\n"
|
|
report += f" Total Message Characters: {summary['conversations']['total_messages_chars']:,}\n"
|
|
report += f" Total Response Characters: {summary['conversations']['total_responses_chars']:,}\n"
|
|
report += f" Average Message Length: {summary['conversations']['avg_message_length']:.0f} chars\n"
|
|
report += f" Average Response Length: {summary['conversations']['avg_response_length']:.0f} chars\n\n"
|
|
|
|
# Provider usage
|
|
report += "🤖 Provider Usage:\n"
|
|
for provider, count in summary['providers']['usage'].items():
|
|
percentage = (count / summary['conversations']['total'] * 100) if summary['conversations']['total'] > 0 else 0
|
|
report += f" {provider}: {count} ({percentage:.1f}%)\n"
|
|
report += f" Most Used: {summary['providers']['most_used']}\n\n"
|
|
|
|
# Model usage
|
|
report += "🧠 Model Usage:\n"
|
|
for model, count in summary['models']['usage'].items():
|
|
percentage = (count / summary['conversations']['total'] * 100) if summary['conversations']['total'] > 0 else 0
|
|
report += f" {model}: {count} ({percentage:.1f}%)\n"
|
|
report += f" Most Used: {summary['models']['most_used']}\n\n"
|
|
|
|
# Performance metrics
|
|
report += "⚡ Performance Metrics:\n"
|
|
report += f" Average Response Time: {summary['performance']['avg_response_time']:.3f}s\n"
|
|
report += f" Minimum Response Time: {summary['performance']['min_response_time']:.3f}s\n"
|
|
report += f" Maximum Response Time: {summary['performance']['max_response_time']:.3f}s\n"
|
|
report += f" Total Response Time: {summary['performance']['total_response_time']:.1f}s\n\n"
|
|
|
|
# Tool usage
|
|
report += "🔧 Tool Usage:\n"
|
|
if summary['tools']['usage']:
|
|
for tool, count in sorted(summary['tools']['usage'].items(), key=lambda x: x[1], reverse=True)[:10]:
|
|
report += f" {tool}: {count}\n"
|
|
report += f" Most Used: {summary['tools']['most_used']}\n"
|
|
else:
|
|
report += " No tool usage recorded\n"
|
|
report += "\n"
|
|
|
|
# Command usage
|
|
report += "⌨️ Command Usage:\n"
|
|
if summary['commands']['usage']:
|
|
for command, count in sorted(summary['commands']['usage'].items(), key=lambda x: x[1], reverse=True)[:10]:
|
|
report += f" {command}: {count}\n"
|
|
report += f" Most Used: {summary['commands']['most_used']}\n"
|
|
else:
|
|
report += " No command usage recorded\n"
|
|
report += "\n"
|
|
|
|
# Error patterns
|
|
report += "❌ Error Patterns:\n"
|
|
if summary['errors']['patterns']:
|
|
for pattern, count in sorted(summary['errors']['patterns'].items(), key=lambda x: x[1], reverse=True)[:5]:
|
|
report += f" {pattern}: {count}\n"
|
|
report += f" Most Common: {summary['errors']['most_common']}\n"
|
|
else:
|
|
report += " No errors recorded\n"
|
|
report += "\n"
|
|
|
|
# Session statistics
|
|
report += "👥 Session Statistics:\n"
|
|
report += f" Total Sessions: {summary['sessions']['total']}\n"
|
|
report += f" Total Session Duration: {summary['sessions']['total_duration']:.1f}s\n"
|
|
report += f" Average Session Duration: {summary['sessions']['avg_duration']:.1f}s\n"
|
|
report += f" Average Messages per Session: {summary['sessions']['avg_messages_per_session']:.1f}\n"
|
|
|
|
return report
|
|
|
|
def get_insights(self, days: int = 30) -> List[str]:
|
|
"""
|
|
Get insights and recommendations based on usage patterns
|
|
|
|
Args:
|
|
days: Number of days to analyze
|
|
|
|
Returns:
|
|
List of insights and recommendations
|
|
"""
|
|
summary = self.get_usage_summary(days)
|
|
insights = []
|
|
|
|
# Provider insights
|
|
if summary['providers']['usage']:
|
|
most_used_provider = summary['providers']['most_used']
|
|
provider_count = summary['providers']['usage'].get(most_used_provider, 0)
|
|
total_conversations = summary['conversations']['total']
|
|
|
|
if total_conversations > 0:
|
|
provider_percentage = (provider_count / total_conversations) * 100
|
|
if provider_percentage > 80:
|
|
insights.append(f"💡 You're heavily using {most_used_provider} ({provider_percentage:.0f}% of conversations). Consider diversifying with other providers for different use cases.")
|
|
elif provider_percentage < 20:
|
|
insights.append(f"💡 You're barely using {most_used_provider} ({provider_percentage:.0f}% of conversations). Consider if this provider is meeting your needs.")
|
|
|
|
# Performance insights
|
|
avg_response_time = summary['performance']['avg_response_time']
|
|
if avg_response_time > 5.0:
|
|
insights.append(f"⚠️ Average response time is high ({avg_response_time:.1f}s). Consider using faster models or optimizing your queries.")
|
|
elif avg_response_time < 1.0:
|
|
insights.append(f"✅ Great response times! Average is {avg_response_time:.1f}s.")
|
|
|
|
# Tool usage insights
|
|
if summary['tools']['usage']:
|
|
most_used_tool = summary['tools']['most_used']
|
|
tool_count = summary['tools']['usage'].get(most_used_tool, 0)
|
|
insights.append(f"🔧 You frequently use the '{most_used_tool}' tool ({tool_count} times). Consider creating shortcuts or aliases for it.")
|
|
|
|
# Error insights
|
|
if summary['errors']['patterns']:
|
|
most_common_error = summary['errors']['most_common']
|
|
error_count = summary['errors']['patterns'].get(most_common_error, 0)
|
|
if error_count > 5:
|
|
insights.append(f"❌ You're encountering a recurring error: '{most_common_error}' ({error_count} times). Consider investigating the root cause.")
|
|
|
|
# Session insights
|
|
avg_session_duration = summary['sessions']['avg_duration']
|
|
if avg_session_duration > 3600: # More than 1 hour
|
|
insights.append(f"⏱️ Your average session is long ({avg_session_duration/60:.0f} minutes). Consider breaking work into shorter sessions for better focus.")
|
|
|
|
# Usage pattern insights
|
|
total_conversations = summary['conversations']['total']
|
|
if total_conversations > 100:
|
|
insights.append(f"📈 You're a power user with {total_conversations} conversations in {days} days!")
|
|
elif total_conversations < 10:
|
|
insights.append(f"💡 You've only had {total_conversations} conversations in {days} days. Consider exploring more features.")
|
|
|
|
# Default insight if none generated
|
|
if not insights:
|
|
insights.append("📊 Your usage patterns look normal. Keep using LangCode Agent!")
|
|
|
|
return insights
|
|
|
|
def clear_old_data(self, days: int = 90):
|
|
"""
|
|
Clear data older than specified days
|
|
|
|
Args:
|
|
days: Number of days to keep data
|
|
"""
|
|
cutoff_time = datetime.now() - timedelta(days=days)
|
|
|
|
# Filter out old conversations
|
|
self.usage_data["conversations"] = [
|
|
c for c in self.usage_data["conversations"]
|
|
if datetime.fromisoformat(c["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
# Filter out old sessions
|
|
self.usage_data["user_sessions"] = [
|
|
s for s in self.usage_data["user_sessions"]
|
|
if datetime.fromisoformat(s["timestamp"]) > cutoff_time
|
|
]
|
|
|
|
# Save cleaned data
|
|
self._save_data()
|
|
|
|
print(f"✅ Cleared analytics data older than {days} days")
|
|
|
|
|
|
# Global analytics engine instance
|
|
analytics_engine = AnalyticsEngine()
|
|
|
|
|
|
def get_usage_summary(days: int = 30) -> Dict[str, Any]:
|
|
"""
|
|
Get usage summary
|
|
|
|
Args:
|
|
days: Number of days to include
|
|
|
|
Returns:
|
|
Usage summary dictionary
|
|
"""
|
|
return analytics_engine.get_usage_summary(days)
|
|
|
|
|
|
def get_analytics_report(days: int = 30) -> str:
|
|
"""
|
|
Get formatted analytics report
|
|
|
|
Args:
|
|
days: Number of days to include
|
|
|
|
Returns:
|
|
Formatted report string
|
|
"""
|
|
return analytics_engine.get_detailed_report(days)
|
|
|
|
|
|
def get_analytics_insights(days: int = 30) -> List[str]:
|
|
"""
|
|
Get analytics insights and recommendations
|
|
|
|
Args:
|
|
days: Number of days to analyze
|
|
|
|
Returns:
|
|
List of insights and recommendations
|
|
"""
|
|
return analytics_engine.get_insights(days)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Print analytics report when run directly
|
|
print(get_analytics_report())
|
|
print("\n" + "=" * 60)
|
|
print("💡 Insights and Recommendations:")
|
|
for insight in get_analytics_insights():
|
|
print(f" {insight}") |