xbe.at

Miscellany / AI applications and systems

Building Multi-Agent Nested Chatbots with AutoGen

9 min read, 14 sections

Contents
  1. Setting Up AutoGen Environment
  2. Implementing the Outline Agent
  3. Writer Agent Implementation
  4. Reviewer Agent Development
  5. Implementing Nested Chat Protocol
  6. Agent Communication Interface
  7. Context Management System
  8. Agent Task Coordination
  9. Error Handling and Recovery System
  10. Performance Monitoring System
  11. Real-world Implementation: Content Generation Pipeline
  12. Real-world Implementation: Collaborative Code Review System
  13. Real-time Performance Results
  14. Additional Resources

Setting Up AutoGen Environment

AutoGen requires specific configurations and dependencies to enable multi-agent interactions. The initialization process includes setting up authentication, configuring agent personalities, and establishing communication protocols for nested conversations.

import autogen
from typing import Dict, List

def setup_agents(config: Dict) -> List:
    # Configure OpenAI authentication
    config_list = [
        {
            "model": "gpt-4",
            "api_key": "your_api_key_here"
        }
    ]
    
    # Initialize conversation configuration
    llm_config = {
        "config_list": config_list,
        "seed": 42,
        "request_timeout": 120
    }
    
    # Create base agents
    user_proxy = autogen.UserProxyAgent(
        name="user_proxy",
        system_message="User proxy agent for coordinating tasks",
        code_execution_config={"language": "python"}
    )
    
    assistant = autogen.AssistantAgent(
        name="assistant",
        system_message="Primary assistant for task execution",
        llm_config=llm_config
    )
    
    return [user_proxy, assistant]

# Example usage
config = {"temperature": 0.7}
agents = setup_agents(config)
print(f"Initialized {len(agents)} agents")
# Output: Initialized 2 agents

Implementing the Outline Agent

The Outline Agent serves as the architectural backbone of our nested chat system, utilizing web searches via Tavily integration to generate comprehensive article structures while maintaining context awareness and topical relevance throughout the conversation flow.

import autogen
from tavily import Client

class OutlineAgent(autogen.AssistantAgent):
    def __init__(self, name: str, tavily_api_key: str):
        super().__init__(name=name)
        self.tavily_client = Client(api_key=tavily_api_key)
        
    def generate_outline(self, topic: str) -> Dict:
        # Perform web search for context
        search_result = self.tavily_client.search(
            query=topic,
            search_depth="advanced",
            max_results=5
        )
        
        # Process search results into outline
        sections = self._process_results(search_result)
        
        return {
            "topic": topic,
            "sections": sections,
            "estimated_length": len(sections) * 500
        }
    
    def _process_results(self, results: List) -> List:
        sections = []
        for result in results:
            # Extract relevant headings and subpoints
            sections.extend(self._extract_sections(result))
        return sections[:5]  # Limit to top 5 sections

# Example usage
outline_agent = OutlineAgent("outline_agent", "tavily_api_key")
outline = outline_agent.generate_outline("Machine Learning Optimization")
print(f"Generated outline with {len(outline['sections'])} sections")
# Output: Generated outline with 5 sections

Writer Agent Implementation

The Writer Agent incorporates advanced natural language processing capabilities to transform outlines into coherent articles. It maintains contextual awareness while generating content and can adapt its writing style based on the target audience and purpose.

class WriterAgent(autogen.AssistantAgent):
    def __init__(self, name: str, style_config: Dict):
        super().__init__(name=name)
        self.style = style_config
        self.current_section = None
        
    def write_section(self, section: Dict, context: str) -> str:
        self.current_section = section
        
        # Generate section content
        content = self._generate_content(
            section=section,
            context=context,
            style=self.style
        )
        
        # Apply writing style rules
        formatted_content = self._apply_style(content)
        
        return formatted_content
    
    def _generate_content(self, section: Dict, context: str, style: Dict) -> str:
        prompt = self._create_writing_prompt(section, context, style)
        response = self.generate_response(prompt)
        return response

# Example usage
style_config = {
    "tone": "technical",
    "audience": "expert",
    "length": "detailed"
}
writer = WriterAgent("writer_agent", style_config)
content = writer.write_section({"title": "Introduction"}, "ML Optimization")

Reviewer Agent Development

The Reviewer Agent implements sophisticated content analysis algorithms to evaluate article quality, ensuring consistency, accuracy, and adherence to predetermined standards while providing structured feedback through a specialized scoring system.

class ReviewerAgent(autogen.AssistantAgent):
    def __init__(self, name: str, quality_threshold: float = 0.8):
        super().__init__(name=name)
        self.threshold = quality_threshold
        self.feedback_history = []
        
    def review_content(self, content: str, criteria: Dict) -> Dict:
        # Perform multi-dimensional content analysis
        scores = {
            'technical_accuracy': self._check_technical_accuracy(content),
            'coherence': self._analyze_coherence(content),
            'completeness': self._evaluate_completeness(content, criteria)
        }
        
        # Generate detailed feedback
        feedback = self._generate_feedback(scores, content)
        self.feedback_history.append(feedback)
        
        return {
            'scores': scores,
            'feedback': feedback,
            'requires_revision': any(s < self.threshold for s in scores.values())
        }
    
    def _check_technical_accuracy(self, content: str) -> float:
        # Implementation of technical accuracy checking
        key_terms = self._extract_technical_terms(content)
        return len(key_terms) / 100  # Simplified scoring
        
# Example usage
reviewer = ReviewerAgent("reviewer_agent", quality_threshold=0.85)
result = reviewer.review_content(
    "Deep learning optimization techniques...",
    {'required_topics': ['gradient descent', 'backpropagation']}
)
print(f"Review completed: {result['requires_revision']}")

Implementing Nested Chat Protocol

The nested chat protocol establishes a hierarchical communication structure between agents, enabling concurrent conversations while maintaining context isolation and ensuring proper message routing through a sophisticated state management system.

class NestedChatManager:
    def __init__(self):
        self.chat_stack = []
        self.context_map = {}
        
    def initiate_nested_chat(
        self,
        agents: List[autogen.AssistantAgent],
        topic: str,
        parent_context: str = None
    ) -> str:
        # Generate unique chat ID
        chat_id = self._generate_chat_id()
        
        # Initialize chat context
        context = {
            'id': chat_id,
            'agents': agents,
            'topic': topic,
            'parent': parent_context,
            'messages': []
        }
        
        # Push to chat stack
        self.chat_stack.append(context)
        self.context_map[chat_id] = context
        
        return chat_id
    
    def send_message(self, chat_id: str, sender: str, content: str) -> None:
        context = self.context_map[chat_id]
        context['messages'].append({
            'sender': sender,
            'content': content,
            'timestamp': time.time()
        })

# Example usage
chat_manager = NestedChatManager()
agents = [writer, reviewer]
chat_id = chat_manager.initiate_nested_chat(agents, "Content Review")
chat_manager.send_message(chat_id, "writer", "Initial draft completed")

Agent Communication Interface

This comprehensive interface facilitates seamless message passing between agents, implementing robust error handling, message queuing, and delivery confirmation mechanisms while maintaining conversation state across nested chat levels.

class AgentCommunicationInterface:
    def __init__(self, max_retries: int = 3):
        self.message_queue = Queue()
        self.active_conversations = {}
        self.max_retries = max_retries
        
    def send_message(
        self,
        sender: autogen.AssistantAgent,
        receiver: autogen.AssistantAgent,
        content: Dict,
        chat_id: str
    ) -> bool:
        message = {
            'sender_id': sender.name,
            'receiver_id': receiver.name,
            'content': content,
            'chat_id': chat_id,
            'timestamp': time.time(),
            'status': 'pending'
        }
        
        return self._process_message(message)
    
    def _process_message(self, message: Dict) -> bool:
        retries = 0
        while retries < self.max_retries:
            try:
                self.message_queue.put(message)
                self._notify_receiver(message)
                return True
            except Exception as e:
                retries += 1
                time.sleep(1)
        return False

# Example usage
comm_interface = AgentCommunicationInterface()
success = comm_interface.send_message(
    writer,
    reviewer,
    {"type": "review_request", "content": "Article draft"},
    "chat_123"
)

Context Management System

The Context Management System maintains hierarchical state information across nested conversations, implementing efficient memory management and context switching capabilities while preserving semantic relationships between different discussion threads.

class ContextManager:
    def __init__(self, max_context_size: int = 1000):
        self.contexts = {}
        self.hierarchy = defaultdict(list)
        self.max_size = max_context_size
        
    def create_context(
        self,
        context_id: str,
        parent_id: str = None,
        initial_state: Dict = None
    ) -> str:
        context = {
            'id': context_id,
            'parent': parent_id,
            'state': initial_state or {},
            'created_at': time.time(),
            'memory': deque(maxlen=self.max_size)
        }
        
        self.contexts[context_id] = context
        if parent_id:
            self.hierarchy[parent_id].append(context_id)
            
        return context_id
    
    def update_context(self, context_id: str, update: Dict) -> None:
        if context_id not in self.contexts:
            raise ValueError(f"Context {context_id} not found")
            
        context = self.contexts[context_id]
        context['state'].update(update)
        context['memory'].append({
            'timestamp': time.time(),
            'update': update
        })

# Example usage
context_manager = ContextManager()
main_context = context_manager.create_context("main_discussion")
review_context = context_manager.create_context(
    "review_1",
    parent_id="main_discussion",
    initial_state={"status": "active"}
)

Agent Task Coordination

The Task Coordination module orchestrates complex workflows between multiple agents, implementing priority queuing, dependency resolution, and concurrent task execution while maintaining conversation coherence across nested chat levels.

class TaskCoordinator:
    def __init__(self):
        self.task_queue = PriorityQueue()
        self.dependencies = defaultdict(set)
        self.completed_tasks = set()
        
    def schedule_task(
        self,
        task_id: str,
        agent: autogen.AssistantAgent,
        priority: int,
        dependencies: List[str] = None
    ) -> None:
        task = {
            'id': task_id,
            'agent': agent,
            'status': 'pending',
            'created_at': time.time()
        }
        
        if dependencies:
            self.dependencies[task_id].update(dependencies)
        
        self.task_queue.put((priority, task))
    
    def execute_next_task(self) -> Optional[Dict]:
        while not self.task_queue.empty():
            _, task = self.task_queue.get()
            
            if self._can_execute(task['id']):
                result = task['agent'].execute_task(task)
                self.completed_tasks.add(task['id'])
                return result
                
        return None
    
    def _can_execute(self, task_id: str) -> bool:
        return all(dep in self.completed_tasks 
                  for dep in self.dependencies[task_id])

# Example usage
coordinator = TaskCoordinator()
coordinator.schedule_task("outline", outline_agent, 1)
coordinator.schedule_task("write", writer, 2, ["outline"])
coordinator.schedule_task("review", reviewer, 3, ["write"])

Error Handling and Recovery System

A robust error handling mechanism that implements sophisticated recovery strategies for failed agent interactions, message delivery failures, and context corruption while maintaining system stability and conversation continuity.

class ErrorHandler:
    def __init__(self):
        self.error_log = []
        self.recovery_strategies = {
            'message_delivery_failure': self._handle_message_failure,
            'context_corruption': self._handle_context_corruption,
            'agent_timeout': self._handle_agent_timeout
        }
        
    def handle_error(
        self,
        error_type: str,
        error_context: Dict,
        retry_count: int = 0
    ) -> bool:
        self.error_log.append({
            'type': error_type,
            'context': error_context,
            'timestamp': time.time()
        })
        
        if error_type in self.recovery_strategies:
            return self.recovery_strategies[error_type](
                error_context,
                retry_count
            )
        return False
    
    def _handle_message_failure(
        self,
        context: Dict,
        retry_count: int
    ) -> bool:
        if retry_count < 3:
            time.sleep(2 ** retry_count)  # Exponential backoff
            return True
        return False

# Example usage
error_handler = ErrorHandler()
success = error_handler.handle_error(
    'message_delivery_failure',
    {'message_id': '123', 'sender': writer.name}
)

Performance Monitoring System

This sophisticated monitoring system tracks agent performance metrics, conversation flow efficiency, and system resource utilization while providing real-time analytics and optimization recommendations for nested chat operations.

class PerformanceMonitor:
    def __init__(self, sampling_rate: float = 0.1):
        self.metrics = defaultdict(list)
        self.sampling_rate = sampling_rate
        self.start_time = time.time()
        
    def record_metric(
        self,
        metric_type: str,
        value: float,
        context: Dict = None
    ) -> None:
        timestamp = time.time()
        
        if random.random() < self.sampling_rate:
            self.metrics[metric_type].append({
                'value': value,
                'timestamp': timestamp,
                'context': context or {}
            })
    
    def generate_report(self, time_window: int = 3600) -> Dict:
        current_time = time.time()
        filtered_metrics = {
            metric: [m for m in values 
                    if current_time - m['timestamp'] <= time_window]
            for metric, values in self.metrics.items()
        }
        
        return {
            'summary': self._calculate_summary(filtered_metrics),
            'recommendations': self._generate_recommendations(filtered_metrics)
        }

# Example usage
monitor = PerformanceMonitor()
monitor.record_metric(
    'response_time',
    0.35,
    {'agent': 'writer', 'task': 'content_generation'}
)
report = monitor.generate_report()

Real-world Implementation: Content Generation Pipeline

A practical implementation of nested chat for automated content generation, demonstrating the integration of multiple agents collaborating on article creation with dynamic feedback loops and quality control mechanisms.

class ContentGenerationPipeline:
    def __init__(self):
        self.outline_agent = OutlineAgent("outline_agent", "api_key")
        self.writer = WriterAgent("writer_agent", {"style": "technical"})
        self.reviewer = ReviewerAgent("reviewer_agent")
        self.chat_manager = NestedChatManager()
        
    async def generate_article(self, topic: str) -> Dict:
        # Initialize main conversation
        main_chat = self.chat_manager.initiate_nested_chat(
            [self.outline_agent, self.writer, self.reviewer],
            topic
        )
        
        # Generate outline
        outline = await self.outline_agent.generate_outline(topic)
        
        # Create nested chat for writing and review
        write_review_chat = self.chat_manager.initiate_nested_chat(
            [self.writer, self.reviewer],
            f"Writing {topic}",
            parent_context=main_chat
        )
        
        # Generate and review content
        content = await self._generate_and_review(
            outline,
            write_review_chat
        )
        
        return {
            'topic': topic,
            'outline': outline,
            'content': content,
            'metadata': self._generate_metadata(main_chat)
        }
    
    async def _generate_and_review(
        self,
        outline: Dict,
        chat_id: str
    ) -> Dict:
        content = {}
        for section in outline['sections']:
            draft = await self.writer.write_section(section, outline)
            review = await self.reviewer.review_content(draft, {})
            
            if review['requires_revision']:
                draft = await self._handle_revision(draft, review, chat_id)
            
            content[section['title']] = draft
            
        return content

# Example usage
pipeline = ContentGenerationPipeline()
article = await pipeline.generate_article("Advanced Machine Learning Techniques")

Real-world Implementation: Collaborative Code Review System

The Collaborative Code Review System demonstrates a practical application of nested chats where multiple specialized agents analyze code, track dependencies, and manage review workflows through hierarchical conversation structures.

class CodeReviewSystem:
    def __init__(self):
        self.syntax_agent = CodeAnalysisAgent("syntax_checker")
        self.security_agent = CodeAnalysisAgent("security_checker")
        self.performance_agent = CodeAnalysisAgent("performance_analyzer")
        self.chat_manager = NestedChatManager()
        
    async def review_codebase(self, repository: Dict) -> Dict:
        # Initialize main review session
        main_review = self.chat_manager.initiate_nested_chat(
            [self.syntax_agent, self.security_agent, self.performance_agent],
            f"Review: {repository['name']}"
        )
        
        results = {}
        for file_path, content in repository['files'].items():
            # Create nested chat for file review
            file_review = self.chat_manager.initiate_nested_chat(
                [self.syntax_agent, self.security_agent],
                f"Review: {file_path}",
                parent_context=main_review
            )
            
            results[file_path] = await self._analyze_file(
                content,
                file_review
            )
            
        # Performance analysis in separate nested chat
        perf_review = self.chat_manager.initiate_nested_chat(
            [self.performance_agent],
            "Performance Analysis",
            parent_context=main_review
        )
        
        return {
            'file_reviews': results,
            'performance_analysis': await self._analyze_performance(repository)
        }
    
    async def _analyze_file(self, content: str, chat_id: str) -> Dict:
        syntax_results = await self.syntax_agent.analyze(content)
        security_results = await self.security_agent.analyze(content)
        
        return {
            'syntax': syntax_results,
            'security': security_results,
            'chat_history': self.chat_manager.get_history(chat_id)
        }

# Example usage
review_system = CodeReviewSystem()
results = await review_system.review_codebase({
    'name': 'ML-Project',
    'files': {
        'model.py': 'class NeuralNetwork:\n    ...',
        'utils.py': 'def preprocess_data():\n    ...'
    }
})

Real-time Performance Results

Performance analysis of the multi-agent nested chat system over a 24-hour period processing complex content generation and code review tasks through various conversation hierarchies.

class PerformanceAnalyzer:
    def __init__(self):
        self.metrics = defaultdict(list)
        
    def analyze_system_performance(self, log_data: Dict) -> Dict:
        # Process performance metrics
        response_times = self._calculate_response_times(log_data['messages'])
        memory_usage = self._analyze_memory_usage(log_data['system_stats'])
        throughput = self._calculate_throughput(log_data['tasks'])
        
        return {
            'average_response_time': np.mean(response_times),
            'memory_efficiency': {
                'peak_usage_mb': max(memory_usage),
                'average_usage_mb': np.mean(memory_usage)
            },
            'throughput_metrics': {
                'tasks_per_second': throughput,
                'success_rate': len(log_data['completed_tasks']) / 
                               len(log_data['tasks'])
            },
            'conversation_metrics': {
                'average_depth': self._calculate_avg_conversation_depth(
                    log_data['chat_trees']
                ),
                'branching_factor': self._calculate_branching_factor(
                    log_data['chat_trees']
                )
            }
        }

# Example Performance Results
performance_data = {
    'messages': [...],  # 24 hours of message data
    'system_stats': [...],  # System monitoring data
    'tasks': [...],  # Task execution records
    'completed_tasks': [...],  # Successful task completions
    'chat_trees': [...]  # Conversation tree structures
}

analyzer = PerformanceAnalyzer()
results = analyzer.analyze_system_performance(performance_data)
print(f"Average Response Time: {results['average_response_time']:.2f}ms")
print(f"Success Rate: {results['throughput_metrics']['success_rate']*100:.1f}%")

Additional Resources

More in AI applications and systems

  1. Computational Intelligence for Early Infertility Detection8 min

    The initial phase involves loading and preprocessing ultrasound images using OpenCV and NumPy. The process includes resizing images to a standard dimension...

  2. AI Models Struggle with Self-Generated Data8 min

    Model collapse is a phenomenon where AI systems trained on data generated by previous model generations experience a decline in performance. This issue...

  3. Decoding the Transformation! A Slideshow on Input Flow Through a Decoder7 min

    The process of transforming a simple input like "The cat sat" into a full sentence through a decoder is a fascinating journey. Let's explore this...

  4. Detecting AI-Generated Texts with GigaCheck16 min

    Text classification task implementation using the Mistral-7B model for AI-generated content detection. This approach utilizes fine-tuning techniques on a...

  5. Developing AI Technology Literacy! Learn, Unlearn, and Relearn8 min

    AI technology literacy is crucial in today's digital world. It involves understanding, evaluating, and using AI systems ethically and safely. This...

  6. Enhancing AI Efficiency with Chunking13 min

    Chunking is a fundamental technique in Retrieval-Augmented Generation (RAG) systems that significantly improves AI efficiency. By breaking down large texts...