"""Multi-LLM Conversation Mesh — agents converse with each other to build skills. All 5 agents (planner, coder, researcher, reviewer, executor) can talk to each other and to the LLM directly. These conversations create: 1. Skill Building Pools — shared pools of skills built collaboratively - Agent A asks a question → Agent B answers → skill extracted from the exchange - Multiple agents contribute to a single skill - Skills are pooled by category and shared across all agents 2. Intelligence Expansion — each conversation generates new knowledge: - New facts → semantic memory - New patterns → skills - New categories → auto-added to skill category list 3. Auto Skill Category Adder — dynamically discovers new categories: - Analyzes conversation topics - Extracts new category keywords - Adds them to the skill category list automatically - No hardcoded category list — categories grow organically Conversation patterns: - Round-robin: each agent talks in turn - Pair-wise: two agents have a focused conversation - Brainstorm: all agents contribute to a single topic - Debate: agents argue different sides of a topic - Teaching: one agent teaches a topic to others """ from __future__ import annotations import hashlib import logging import random import threading import time from collections import defaultdict, deque from dataclasses import dataclass, field from typing import Any, Callable logger = logging.getLogger(__name__) @dataclass class ConversationMessage: """A single message in a multi-LLM conversation.""" agent_name: str content: str timestamp: float = field(default_factory=time.time) role: str = "speaker" # speaker, listener, summarizer @dataclass class SkillPool: """A pool of skills built collaboratively by multiple agents.""" id: str category: str name: str description: str contributions: list[dict] = field(default_factory=list) # which agents contributed skill_content: str = "" confidence: float = 0.0 created_at: float = field(default_factory=time.time) tags: list[str] = field(default_factory=list) class AutoCategoryManager: """Automatically discovers and adds new skill categories. Analyzes conversation topics and extracts new categories dynamically. No hardcoded category list — categories grow organically based on what the agents actually talk about. Category discovery: 1. Extract keywords from conversation topics 2. Cluster similar keywords into categories 3. Add new categories to the skill system 4. Track category health (how many skills, how active) """ # Seed categories — the list grows from here SEED_CATEGORIES = [ "conversation", "code", "speed", "voice", "tool", "math", "writing", "analysis", "debugging", "optimization", ] def __init__(self) -> None: self._categories: dict[str, dict] = {} self._keyword_to_category: dict[str, str] = {} self._category_keywords: dict[str, set[str]] = defaultdict(set) self._stats = { "categories_total": 0, "categories_auto_added": 0, "keywords_indexed": 0, "category_lookups": 0, } # Initialize with seed categories for cat in self.SEED_CATEGORIES: self._add_category(cat, auto=False) def _add_category(self, name: str, auto: bool = True) -> str: """Add a new category.""" name = name.lower().strip() if name in self._categories: return name self._categories[name] = { "name": name, "auto_added": auto, "skill_count": 0, "conversation_count": 0, "created_at": time.time(), "last_active": time.time(), } self._stats["categories_total"] += 1 if auto: self._stats["categories_auto_added"] += 1 logger.info("Added skill category: %s (auto=%s)", name, auto) return name def discover_from_topic(self, topic: str) -> list[str]: """Discover new categories from a conversation topic. Extracts keywords, checks if they match existing categories, and creates new categories for novel topics. """ words = self._extract_keywords(topic) discovered: list[str] = [] for word in words: # Check if this keyword maps to an existing category if word in self._keyword_to_category: cat = self._keyword_to_category[word] self._categories[cat]["last_active"] = time.time() self._categories[cat]["conversation_count"] += 1 discovered.append(cat) continue # Check if word is similar to an existing category matched = self._match_existing_category(word) if matched: self._keyword_to_category[word] = matched self._category_keywords[matched].add(word) self._stats["keywords_indexed"] += 1 discovered.append(matched) continue # New category discovered! cat = self._add_category(word, auto=True) self._keyword_to_category[word] = cat self._category_keywords[cat].add(word) self._stats["keywords_indexed"] += 1 self._categories[cat]["conversation_count"] += 1 discovered.append(cat) return discovered def _extract_keywords(self, text: str) -> list[str]: """Extract meaningful keywords from text.""" import re # Remove common stop words stop_words = { "the", "a", "an", "is", "are", "was", "were", "be", "been", "have", "has", "had", "do", "does", "did", "will", "would", "could", "should", "may", "might", "must", "can", "need", "how", "what", "when", "where", "why", "who", "which", "to", "of", "in", "on", "at", "by", "for", "with", "about", "from", "as", "into", "through", "during", "before", "after", "and", "or", "but", "not", "no", "yes", "if", "then", "else", "this", "that", "these", "those", "i", "you", "he", "she", "it", "we", "they", "me", "him", "her", "us", "them", "my", "your", "his", "its", "our", "their", } words = re.findall(r'\b[a-z]{3,20}\b', text.lower()) keywords = [w for w in words if w not in stop_words and len(w) >= 4] # Limit to top 5 unique keywords seen = set() result = [] for w in keywords: if w not in seen: seen.add(w) result.append(w) if len(result) >= 5: break return result def _match_existing_category(self, keyword: str) -> str | None: """Check if a keyword is similar to an existing category.""" # Simple prefix/suffix matching for cat in self._categories: if keyword.startswith(cat) or cat.startswith(keyword): return cat if keyword.endswith(cat) or cat.endswith(keyword): return cat # Check if keyword is in any category's keyword set for cat, keywords in self._category_keywords.items(): if keyword in keywords: return cat return None def get_categories(self) -> list[str]: """Get all category names.""" return list(self._categories.keys()) def get_auto_categories(self) -> list[str]: """Get only auto-discovered categories.""" return [name for name, info in self._categories.items() if info["auto_added"]] def increment_skill_count(self, category: str) -> None: """Increment the skill count for a category.""" if category in self._categories: self._categories[category]["skill_count"] += 1 self._categories[category]["last_active"] = time.time() def get_category_info(self, category: str) -> dict | None: """Get info about a specific category.""" return self._categories.get(category) def get_stats(self) -> dict[str, Any]: return { **self._stats, "categories": dict(self._categories), "seed_categories": self.SEED_CATEGORIES, } class ConversationMesh: """Multi-LLM conversation mesh — agents converse to build skills. All 5 agents can talk to each other and to the LLM. Conversations create skill building pools and discover new skill categories. Conversation modes: - round_robin: each agent speaks in turn on a topic - pairwise: two agents have a focused discussion - brainstorm: all agents contribute ideas on a topic - debate: agents argue different perspectives - teaching: one agent teaches a topic to others """ CONVERSATION_MODES = ["round_robin", "pairwise", "brainstorm", "debate", "teaching"] BRAINSTORM_TOPICS = [ "How to improve code quality", "Best practices for data processing", "Optimizing memory usage in AI systems", "New approaches to natural language understanding", "Efficient algorithms for pattern matching", "Strategies for error handling and recovery", "Building better user interfaces", "Improving search and retrieval accuracy", "Techniques for faster inference", "Methods for continuous learning", "Approaches to multi-agent coordination", "Enhancing creativity in problem solving", "Building robust testing frameworks", "Optimizing database queries", "Improving API design patterns", "Strategies for handling concurrency", "Methods for data compression", "Approaches to security and privacy", "Techniques for code refactoring", "Building scalable architectures", "Improving error messages and debugging", "Optimizing network communication", "Strategies for resource management", "Methods for adaptive learning", "Approaches to self-improvement", ] def __init__(self, harness: Any) -> None: self.harness = harness self.agent_names = ["planner", "coder", "researcher", "reviewer", "executor"] self.category_manager = AutoCategoryManager() self._skill_pools: dict[str, SkillPool] = {} self._conversations: deque[dict] = deque(maxlen=100) self._cascade_depth = 0 # track cascade depth to prevent infinite recursion self._stats = { "conversations_total": 0, "skills_pooled": 0, "categories_discovered": 0, "messages_exchanged": 0, "cross_agent_skills": 0, "cascade_depth": 0, "cascade_pools_created": 0, "cascade_conversations": 0, } def run_conversation(self, mode: str = "", topic: str = "") -> dict[str, Any]: """Run a multi-LLM conversation. Args: mode: conversation mode (round_robin, pairwise, brainstorm, debate, teaching) topic: conversation topic (random if not provided) Returns: Summary of the conversation and skills created """ mode = mode or random.choice(self.CONVERSATION_MODES) topic = topic or random.choice(self.BRAINSTORM_TOPICS) # Discover categories from the topic categories = self.category_manager.discover_from_topic(topic) if categories: self._stats["categories_discovered"] += len(categories) # Run the conversation based on mode if mode == "round_robin": messages = self._round_robin(topic) elif mode == "pairwise": messages = self._pairwise(topic) elif mode == "brainstorm": messages = self._brainstorm(topic) elif mode == "debate": messages = self._debate(topic) elif mode == "teaching": messages = self._teaching(topic) else: messages = self._round_robin(topic) # Build skill pool from the conversation pool = self._build_skill_pool(topic, messages, categories) # Cascade: when a skill pool is built, auto-generate related skill pools cascade_results = [] if pool and self._cascade_depth < 2: cascade_results = self._cascade_related_skills(topic, categories, max_depth=2) # Store conversation conv_record = { "mode": mode, "topic": topic, "categories": categories, "messages": [{"agent": m.agent_name, "content": m.content[:100]} for m in messages], "skill_pool_id": pool.id if pool else None, "timestamp": time.time(), } self._conversations.append(conv_record) self._stats["conversations_total"] += 1 self._stats["messages_exchanged"] += len(messages) # Store in persistent memory if self.harness: try: self.harness.persistent_memory.add_episodic( "mesh", f"[{mode}] {topic}: {len(messages)} messages, categories: {categories}", channel="mesh", importance=0.5, tags=["mesh", mode] + categories ) except Exception: pass return { "mode": mode, "topic": topic, "categories": categories, "messages": len(messages), "skill_pool": pool is not None, "pool_id": pool.id if pool else None, "cascaded": cascade_results, } def _cascade_related_skills(self, topic: str, categories: list[str], max_depth: int = 3) -> list[dict]: """When a skill pool is built, automatically generate related skill pools. For each new skill pool, the system: 1. Generates related topics based on the categories 2. Runs conversations on those related topics 3. Builds new skill pools from those conversations 4. Cascades further (up to max_depth) if new categories are discovered This creates a branching tree of related skills — one conversation on "database optimization" cascades into skills on "query performance", "indexing strategies", "cache invalidation", etc. Runs during idle time via the always-on daemon. """ if self._cascade_depth >= max_depth: return [] self._cascade_depth += 1 self._stats["cascade_depth"] = max(self._stats["cascade_depth"], self._cascade_depth) # Generate related topics from the categories related_topics = self._generate_related_topics(topic, categories) cascade_results = [] for related_topic in related_topics: try: # Run a conversation on the related topic (different mode for variety) mode = random.choice(self.CONVERSATION_MODES) result = self.run_conversation(mode=mode, topic=related_topic) if result.get("skill_pool"): self._stats["cascade_pools_created"] += 1 self._stats["cascade_conversations"] += 1 cascade_results.append({ "topic": related_topic, "mode": mode, "pool_id": result.get("pool_id"), "categories": result.get("categories", []), }) # Check if new categories were discovered → cascade further new_cats = [c for c in result.get("categories", []) if c not in categories] if new_cats: deeper = self._cascade_related_skills(related_topic, new_cats, max_depth) cascade_results.extend(deeper) except Exception as e: logger.debug("Cascade conversation failed: %s", e) self._cascade_depth -= 1 return cascade_results def _generate_related_topics(self, topic: str, categories: list[str]) -> list[str]: """Generate related topics from a topic and its categories. Uses category keywords to create variations and deeper explorations. """ related = [] # Topic variations based on categories for cat in categories[:3]: # top 3 categories related.extend([ f"Advanced {cat} techniques for {topic.lower()[:30]}", f"Common pitfalls in {cat} when working with {topic.lower()[:20]}", f"Best practices for {cat} in real-world scenarios", f"Optimizing {cat} performance and reliability", f"Testing strategies for {cat} implementations", ]) # Cross-category topics if len(categories) >= 2: related.append(f"Integrating {categories[0]} with {categories[1]}") related.append(f"Trade-offs between {categories[0]} and {categories[1]}") # Deduplicate and limit seen = set() unique = [] for t in related: t_lower = t.lower() if t_lower not in seen: seen.add(t_lower) unique.append(t) return unique[:3] # limit to 3 related topics per cascade level def _generate_response(self, agent_name: str, topic: str, context: str = "") -> str: """Generate a response from an agent.""" prompt = f"Agent {agent_name} discusses: {topic}" if context: prompt += f"\nContext: {context[:200]}" try: if self.harness: return self.harness._agent_generate(prompt)[:200] except Exception: pass return f"{agent_name}: I think {topic.lower()[:50]} is important to explore." def _round_robin(self, topic: str) -> list[ConversationMessage]: """Each agent speaks in turn on the topic.""" messages = [] context = "" for agent in self.agent_names: response = self._generate_response(agent, topic, context) msg = ConversationMessage(agent_name=agent, content=response) messages.append(msg) context = f"{agent} said: {response[:100]}" return messages def _pairwise(self, topic: str) -> list[ConversationMessage]: """Two agents have a focused discussion.""" messages = [] a1, a2 = random.sample(self.agent_names, 2) context = "" for i in range(4): # 2 exchanges each speaker = a1 if i % 2 == 0 else a2 response = self._generate_response(speaker, topic, context) msg = ConversationMessage(agent_name=speaker, content=response) messages.append(msg) context = f"{speaker} said: {response[:100]}" return messages def _brainstorm(self, topic: str) -> list[ConversationMessage]: """All agents contribute ideas on a topic.""" messages = [] for agent in self.agent_names: prompt = f"Brainstorm ideas for: {topic}" response = self._generate_response(agent, topic, "") msg = ConversationMessage(agent_name=agent, content=response, role="speaker") messages.append(msg) # Summarizer summary = f"Summary: We discussed {topic} and generated {len(messages)} ideas." msg = ConversationMessage(agent_name="reviewer", content=summary, role="summarizer") messages.append(msg) return messages def _debate(self, topic: str) -> list[ConversationMessage]: """Agents argue different perspectives on a topic.""" messages = [] for i, agent in enumerate(self.agent_names): side = "for" if i % 2 == 0 else "against" prompt = f"Argue {side}: {topic}" response = self._generate_response(agent, prompt, "") msg = ConversationMessage(agent_name=agent, content=response) messages.append(msg) return messages def _teaching(self, topic: str) -> list[ConversationMessage]: """One agent teaches a topic to others.""" messages = [] teacher = random.choice(self.agent_names) # Teacher explains response = self._generate_response(teacher, f"Teach: {topic}", "") messages.append(ConversationMessage(agent_name=teacher, content=response, role="speaker")) # Students ask questions students = [a for a in self.agent_names if a != teacher] for student in students[:3]: question = self._generate_response(student, f"Question about: {topic}", response[:100]) messages.append(ConversationMessage(agent_name=student, content=question, role="listener")) return messages def _build_skill_pool(self, topic: str, messages: list[ConversationMessage], categories: list[str]) -> SkillPool | None: """Build a skill pool from a conversation. A skill pool is a collaborative skill built from multiple agent contributions. """ if not messages or not categories: return None # Combine all messages into skill content content_parts = [f"Skill Pool: {topic}"] content_parts.append(f"Category: {categories[0]}") content_parts.append(f"Contributors: {', '.join(set(m.agent_name for m in messages))}") content_parts.append("Knowledge:") for msg in messages: content_parts.append(f" [{msg.agent_name}] {msg.content[:150]}") content_parts.append("Best practice: Apply this knowledge collaboratively.") skill_content = "\n".join(content_parts) # Create pool pool_id = hashlib.sha256(f"{topic}:{time.time()}".encode()).hexdigest()[:16] category = categories[0] pool = SkillPool( id=pool_id, category=category, name=f"{category}-pool-{pool_id[:8]}", description=f"Collaborative skill pool for {topic[:80]}", contributions=[ {"agent": m.agent_name, "content": m.content[:100], "role": m.role} for m in messages ], skill_content=skill_content, confidence=min(0.9, len(messages) / 10.0), tags=categories, ) self._skill_pools[pool_id] = pool self._stats["skills_pooled"] += 1 # Count cross-agent skills (skills where multiple agents contributed) unique_agents = set(m.agent_name for m in messages) if len(unique_agents) >= 2: self._stats["cross_agent_skills"] += 1 # Update category skill count self.category_manager.increment_skill_count(category) # Try to create an actual skill in the skill manager if self.harness: try: from ..skills.skills import Skill skill = Skill( id=pool_id, name=pool.name, description=pool.description, content=pool.skill_content[:500], category=category, trigger_conditions=categories + [topic.lower()[:20]], effectiveness_score=pool.confidence, confidence=pool.confidence, ) self.harness.skill_manager.create(skill) except Exception as e: logger.debug("Skill creation from pool failed: %s", e) return pool def get_skill_pools(self) -> list[dict]: """Get all skill pools.""" return [ { "id": p.id, "name": p.name, "category": p.category, "description": p.description, "contributors": list(set(c["agent"] for c in p.contributions)), "confidence": p.confidence, "tags": p.tags, } for p in self._skill_pools.values() ] def get_categories(self) -> list[str]: """Get all skill categories (including auto-discovered).""" return self.category_manager.get_categories() def get_auto_categories(self) -> list[str]: """Get auto-discovered categories.""" return self.category_manager.get_auto_categories() def get_stats(self) -> dict[str, Any]: return { **self._stats, "skill_pools": len(self._skill_pools), "categories": self.category_manager.get_stats(), "conversations_buffered": len(self._conversations), }