#!/usr/bin/env python3 """ Collaboration Insights Module for HuggingClaw Cain Visualizes agent communication patterns between Adam, Eve, and Cain. Generates ASCII/text tree showing conversation flow and agent interactions. """ import json from typing import Dict, List, Any, Optional, Tuple from pathlib import Path from datetime import datetime from collections import defaultdict, Counter # Data paths - use nested openclaw/.openclaw structure OPENCLAW_AGENTS_DIR = Path(__file__).parent / "openclaw" / ".openclaw" / "agents" ANALYTICS_DIR = OPENCLAW_AGENTS_DIR / "analytics" CONVERSATION_HISTORY_FILE = ANALYTICS_DIR / "conversation_history.jsonl" ANALYTICS_FILE = ANALYTICS_DIR / "conversation_analytics.json" class CollaborationVisualizer: """Visualizes agent collaboration patterns from conversation data""" def __init__(self): self.interactions: List[Dict] = [] self.agent_stats: Dict[str, Dict] = defaultdict(lambda: { "messages_sent": 0, "messages_received": 0, "tools_used": Counter(), "avg_response_time": 0.0, "total_response_time": 0.0, "response_count": 0 }) def load_conversation_data(self) -> int: """Load conversation history from analytics files""" count = 0 # Load from conversation_history.jsonl if CONVERSATION_HISTORY_FILE.exists(): try: with open(CONVERSATION_HISTORY_FILE, 'r') as f: for line in f: if line.strip(): data = json.loads(line) self.interactions.append(data) count += 1 except Exception as e: print(f"[CollaborationInsights] Failed to load history: {e}") # Load from analytics file if ANALYTICS_FILE.exists(): try: with open(ANALYTICS_FILE, 'r') as f: data = json.load(f) convs = data.get("conversations", {}) for conv_id, conv_data in convs.items(): self.interactions.append({ "conversation_id": conv_id, "timestamp": conv_data.get("timestamp", ""), "message_count": conv_data.get("message_count", 0), "tool_calls": conv_data.get("tool_calls", {}), "status": conv_data.get("status", "unknown"), "avg_response_time": conv_data.get("avg_response_time", 0) }) count += 1 except Exception as e: print(f"[CollaborationInsights] Failed to load analytics: {e}") return count def _detect_agent_from_message(self, msg: Dict) -> str: """Detect which agent sent a message based on context""" role = msg.get("role", "unknown") content = msg.get("content", "").lower() tools = msg.get("tools_used", []) # Agent signatures in content if "adam" in content or "infrastructure" in content: return "adam" elif "eve" in content or "ui" in content or "design" in content: return "eve" elif "cain" in content or "chat" in content or "conversation" in content: return "cain" # Tool-based detection for tool in tools: if "adam" in tool.lower(): return "adam" elif "eve" in tool.lower(): return "eve" # Role-based mapping if role == "user": return "user" elif role == "assistant": return "cain" return "unknown" def _generate_interaction_tree(self) -> List[str]: """Generate ASCII tree of agent interactions""" lines = [ "## 🌳 Agent Collaboration Tree", "", "```\n" ] if not self.interactions: lines.extend([ " No interaction data available yet.", " Start a conversation to see the collaboration flow!", "" ]) lines.append("```\n") return lines # Build interaction chain for i, msg in enumerate(self.interactions[:50]): # Limit to 50 for readability agent = self._detect_agent_from_message(msg) timestamp = msg.get("timestamp", "")[:19] if msg.get("timestamp") else "" role = msg.get("role", "unknown") # Format based on agent if agent == "adam": prefix = " [Adam] 🔧" elif agent == "eve": prefix = " [Eve] 🎨" elif agent == "cain": prefix = " [Cain] 💬" elif agent == "user": prefix = " [User] 👤" else: prefix = " [?] ❓" # Add timestamp if available if timestamp: time_str = f" {timestamp}" else: time_str = "" # Add arrow for flow if i > 0: arrow = " └─>" else: arrow = " " lines.append(f"{prefix}{arrow}{role}{time_str}") # Show tools if present tools = msg.get("tools_used", []) if tools: for tool in tools[:3]: # Max 3 tools per message lines.append(f" └─ 🔧 {tool}") lines.append("\n```\n") return lines def _generate_agent_stats(self) -> List[str]: """Generate statistics about agent collaboration""" lines = [ "## 📊 Collaboration Statistics", "" ] # Calculate stats agent_counts = Counter() tool_usage = Counter() total_messages = len(self.interactions) for msg in self.interactions: agent = self._detect_agent_from_message(msg) agent_counts[agent] += 1 for tool in msg.get("tools_used", []): tool_usage[tool] += 1 if total_messages == 0: lines.append("No data available yet. Start a conversation!") lines.append("") return lines lines.append(f"**Total Interactions:** {total_messages}") lines.append("") lines.append("### Agent Activity") lines.append("") for agent, count in agent_counts.most_common(): percentage = (count / total_messages) * 100 agent_name = agent.capitalize() bar = "█" * int(percentage / 5) lines.append(f"- **{agent_name}:** {count} messages ({percentage:.1f}%) {bar}") lines.append("") if tool_usage: lines.append("### Top Tools Used") lines.append("") for tool, count in tool_usage.most_common(10): lines.append(f"- 🔧 `{tool}`: {count} uses") lines.append("") return lines def _generate_interaction_flow(self) -> List[str]: """Generate visualization of message flow between agents""" lines = [ "## 🔄 Interaction Flow", "", "```" ] if not self.interactions: lines.append("No flow data available yet.") lines.append("```") return lines # Track agent transitions flow_pattern = [] prev_agent = None for msg in self.interactions[:20]: # Last 20 messages agent = self._detect_agent_from_message(msg) if prev_agent and prev_agent != agent: flow_pattern.append(f"{prev_agent} → {agent}") prev_agent = agent if not flow_pattern: lines.extend([ " Start → [First Message]", "", " (More interactions needed for flow visualization)" ]) else: # Count unique flows flow_counts = Counter(flow_pattern) lines.append(" Most Common Flows:") for flow, count in flow_counts.most_common(10): arrow = " → ".join(f"[{f.capitalize()}]" for f in flow.split(" → ")) lines.append(f" {arrow} ({count}x)") lines.extend([ "", "```", "" ]) return lines def generate_insights(self) -> str: """Generate complete collaboration insights""" self.load_conversation_data() sections = [] sections.extend(self._generate_interaction_tree()) sections.extend(self._generate_agent_stats()) sections.extend(self._generate_interaction_flow()) return "\n".join(sections) def get_summary(self) -> str: """Get a quick summary of collaboration insights""" self.load_conversation_data() if not self.interactions: return "## 🤝 Collaboration Insights\n\nNo interaction data yet. Start chatting with Cain to see how Adam, Eve, and Cain work together!" agent_counts = Counter( self._detect_agent_from_message(msg) for msg in self.interactions ) lines = [ "## 🤝 Collaboration Insights", "", f"**Total Interactions Tracked:** {len(self.interactions)}", "", "### Agent Participation" ] for agent, count in agent_counts.most_common(): agent_name = agent.capitalize() emoji = { "adam": "🔧", "eve": "🎨", "cain": "💬", "user": "👤" }.get(agent.lower(), "❓") lines.append(f"- **{agent_name}** {emoji}: {count} interactions") lines.append("") lines.append("*> Open the Collaboration Insights tab for detailed visualizations*") return "\n".join(lines) # Global instance _visualizer: Optional[CollaborationVisualizer] = None def get_visualizer() -> CollaborationVisualizer: """Get the global collaboration visualizer instance""" global _visualizer if _visualizer is None: _visualizer = CollaborationVisualizer() return _visualizer def generate_collaboration_tree() -> str: """Generate the collaboration tree visualization""" return get_visualizer().generate_insights() def get_collaboration_summary() -> str: """Get a quick summary of collaboration insights""" return get_visualizer().get_summary() if __name__ == "__main__": print("=== Collaboration Insights Test ===\n") viz = CollaborationVisualizer() count = viz.load_conversation_data() print(f"Loaded {count} interactions\n") print(viz.generate_insights())