Example: Sequential pipeline with AutoGen [chapter] deterministic
from autogen import Agent, ConversableAgent planner = ConversableAgent( name="Planner", llm_config={"model": "gpt-4"}, system_message="You are a planner. Produce a step-by-step plan." ) executor = Con
from autogen import Agent, ConversableAgent planner = ConversableAgent( name="Planner", llm_config={"model": "gpt-4"}, system_message="You are a planner. Produce a step-by-step plan." ) executor = ConversableAgent( name="Executor", llm_config={"model": "gpt-3.5"}, system_message="You are an executor. Carry out the plan." ) planner.register_reply(executor, lambda agent, messages: planner.generate_reply(messages)) executor.register_reply(planner, lambda agent, messages: executor.generate_reply(messages)) planner.initiate_chat(executor, message="Plan a data extraction pipeline for CSV files.")
`
Parallel Branches When subtasks are independent, you can spawn parallel branches to reduce overall latency. For instance, two agents might simultaneously search different databases, then a third agent merges the results. Parallelism introduces concurrency challenges: you must handle race conditions, ensure consistent **Conversation History**, and manage resource limits.
Hierarchical Coordination In hierarchical setups, a higher-level agent delegates tasks to subordinate agents, which in turn may delegate further. This pattern mirrors organizational structures and is useful for large-scale projects where domain-specific expertise is required. Hierarchies also simplify debugging because the flow of control is explicit.
Decentralized Swarms A swarm architecture treats agents as peers that exchange information without a central coordinator. This can be highly resilient but is difficult to control, especially when agents have conflicting goals. Swarms are best suited for exploratory tasks where emergent behavior is desirable.
Implementing Agent Communication Protocols Agents do not exist in isolation; they must communicate to coordinate actions, share state, and resolve conflicts. In local-first systems, communication protocols must be lightweight, secure, and easy to embed in existing codebases. Below are the core components of an effective communication protocol.
Message Formats A well-defined message format ensures that agents can parse each other’s outputs reliably. JSON is a common choice because it is language-agnostic and supports nested structures. For larger payloads, consider protobuf or Avro, which provide schema validation and binary efficiency.
```json { "type": "task", "id": "task-001", "payload": { "action": "extract", "source": "https://example.com/data", "format": "csv" }, "timestamp": "2025-10-10T12:00:00Z" }
`
Conversation History
Maintaining a **Conversation History** is essential for agents that need context from prior exchanges. Each message should carry a MessageRole (e.g., `, , `) to indicate the sender. This metadata enables downstream agents to reconstruct the dialogue and reason about prior decisions.
```python class Message: def __init__(self, role: str, content: str, timestamp: datetime): self.role = role self.content = content self.timestamp = timestamp
`
Tool Execution and ToolResult Agents often need to invoke external tools—databases, APIs, or local scripts—to perform actions. The outcome of a tool execution is captured in a **ToolResult**, which includes the return value, any error messages, and metadata such as execution time. This result can be fed back into the conversation history, allowing other agents to react to the outcome.
```python class ToolResult: def __init__(self, success: bool, output: str, error: str = None): self.success = success self.output = output self.error = error
`
Confidentiality and Security When agents exchange sensitive data, **Confidentiality** must be preserved. In local-first deployments, you can achieve this by encrypting messages at rest and in transit using TLS or symmetric encryption (e.g., AES-256). Additionally, limit the exposure of **ToolResult** payloads by stripping out any personally identifiable information (PII) before sharing.
Orchestrating Agents with Microsoft AutoGen Microsoft AutoGen is an open-source framework for building multi-agent systems with minimal boilerplate. It provides built-in support for conversation-driven agents, tool integration, and hierarchical orchestration. Below is a walkthrough of how to set up AutoGen and use it to coordinate a small team of agents.
Installation and Setup
```bash pip install autogen
`
After installation, import the necessary modules:
```python from autogen import Agent, ConversableAgent, AssistantAgent, GroupChat, GroupChatManager
`
Defining Agents
Create individual agents, each with a distinct role and prompt. For example, a Planner agent might generate a plan, an Executor agent might carry it out, and a Validator agent might check the output.
```python planner = ConversableAgent( name="Planner", llm_config={"model": "gpt-4"}, system_message="You are a planner. Produce a step-by-step plan." ) executor = ConversableAgent( name="Executor", llm_config={"model": "gpt-3.5"}, system_message="You are an executor. Carry out the plan." ) validator = ConversableAgent( name="Validator", llm_config={"model": "gpt-4"}, system_message="You are a validator. Review the output and suggest improvements." )
`
Creating a Group Chat
AutoGen’s GroupChat allows you to define a set of agents that can converse with each other. You specify the participants and optionally a speaker selection method (e.g., round-robin, random, or LLM-based).
```python groupchat = GroupChat( agents=[planner, executor, validator], messages=[], max_round=5 )
`
Managing the Group Chat
A GroupChatManager oversees the conversation, ensuring that each agent gets a turn and that the dialogue stays within bounds. You can also provide a message to guide the manager’s behavior.
```python manager = GroupChatManager( groupchat=groupchat, llm_config={"model": "gpt-4"}, system_message="You are the manager. Keep the conversation on track." )
`
Initiating the Conversation Finally, kick off the interaction by sending an initial message to the manager. The manager will route it to the appropriate agent, and the conversation will unfold.
```python manager.initiate_chat( recipient=executor, message="Start the pipeline: extract data from the CSV source." )
`
Adding Tools Agents can be equipped with custom tools by registering reply functions that invoke external code. For example, you might add a tool that queries a local SQLite database.
```python import sqlite3 def query_db(sql: str) -> str: conn = sqlite3.connect("local.db") cursor = conn.cursor() cursor.execute(sql) result = cursor.fetchall() conn.close() return str(result) executor.register_tool("query_db", query_db)
`
When the executor needs data, it can call query_db, receive a **ToolResult**, and pass it back to the conversation history.
Handling Errors and Retries
In a multi-agent setting, failures are inevitable. AutoGen provides a retry mechanism that you can wrap around agent interactions. For example, if the executor fails to retrieve data, the planner can be prompted to adjust the plan.
```python def safe_execute(agent, message): try: return agent.generate_reply(message) except Exception as e: return f"Error: {e}"
`
This pattern ensures that the remains robust and that other agents can react to failures.