| Competitive analysis |
Research ...5. AI Orchestrator Patterns5.1 Sequential Orchestration
In sequential orchestration, sub-tasks are executed one after another in a predefined order. Each task's output feeds into the next task as input.
Goal ā Task A ā Task B ā Task C ā Task D ā Output
When to use:
- Tasks have strict dependencies (each step requires the previous step's output)
- The workflow is linear and predictable
- Simpler debugging and error tracing is desired
Example:
Write Specification ā Generate Code ā Run Tests ā Fix Issues ā Deploy
Implementation approach:
async def sequential_orchestrator(tasks, agents):
results = {}
for task in tasks:
agent = select_agent(task, agents)
result = await agent.run(task, context=results)
results[task.id] = result
return results
Advantages: Simple to implement, easy to debug, predictable state
Disadvantages: Slow (no parallelism), underutilizes available resources
5.2 Parallel Orchestration
In parallel orchestration, independent tasks are executed concurrently. The orchestrator waits for all tasks to complete (or a sufficient subset) before moving to the synthesis phase.
Goal ā Task A āā
Task B āāā¤ā Synthesis ā Output
Task C āā
When to use:
- Multiple independent research questions
- Data collection from different sources
- Any workflow where tasks don't depend on each other
Example:
Market Research Request:
āā Find competitor prices āā
āā Analyze customer reviews āā¤ā Combine ā Report
āā Search industry news āā
āā Gather financial data āā
Implementation approach:
async def parallel_orchestrator(tasks, agents):
async def run_task(task):
agent = select_agent(task, agents)
return task.id, await agent.run(task)
results = await asyncio.gather(*[run_task(t) for t in tasks])
return dict(results)
Advantages: Faster execution, better resource utilization
Disadvantages: Requires careful dependency analysis, more complex error handling, potential for resource contention
5.3 Hierarchical Orchestration
Hierarchical orchestration introduces a tree structure where a manager agent oversees multiple worker agents, which may themselves have sub-workers.
Orchestrator
/ | \
Manager A Manager B Manager C
/ \ | |
Worker1 Worker2 Worker3 Worker4
When to use:
- The problem has natural hierarchical structure (e.g., organization reporting, project management)
- Different levels of abstraction require different reasoning approaches
- You need delegation chains with escalation paths
Example ā Enterprise Report Generation:
Report Orchestrator
āāā Research Manager
ā āāā Web Researcher Agent
ā āāā Database Analyst Agent
ā āāā Interview Agent
āāā Writing Manager
ā āāā Executive Summary Writer
ā āāā Technical Writer
ā āāā Editor/Reviewer Agent
āāā QA Manager
āāā Fact-Checker Agent
āāā Compliance Reviewer Agent
āāā Format Validator Agent
Implementation approach:
class HierarchicalOrchestrator:
async def execute(self, goal):
plan = await self.master_planner.create_plan(goal)
for phase in plan.phases:
manager = self.get_manager(phase.manager_type)
phase_result = await manager.execute(phase, self.worker_registry)
self.state.store(phase_result)
return self.synthesizer.synthesize(self.state)
Advantages: Scalable to very large workflows, clear responsibility boundaries, natural delegation
Disadvantages: Higher latency due to multi-layer coordination, more complex to debug and monitor
5.4 Recursive Orchestration
Recursive orchestration is used when a sub-task is discovered to be too complex for a single agent. The orchestrator recursively decomposes the sub-task into smaller pieces, treating it as a new mini-goal.
Goal ā Task A ā Task B ā B is too complex ā Decompose B ā
āā B1 ā B2 ā B3 ā Reassemble B ā Continue ā Output
When to use:
- The complexity of sub-tasks is unknown upfront
- Tasks have variable granularity (some are simple, some turn out to be complex)
- Self-improving systems where the orchestrator learns to decompose over time
Example ā Complex Code Generation:
Goal: "Build a recommendation engine"
ā Design architecture (easy, one shot)
ā Implement collaborative filtering (medium, one shot)
ā Implement real-time personalization (too complex!)
ā Recursively decompose:
ā Implement user session tracking
ā Implement feature extraction pipeline
ā Implement model inference endpoint
ā Implement A/B testing framework
ā Implement caching layer
ā Reassemble into personalization module
ā Write tests
ā Integration and final output
Implementation approach:
async def recursive_orchestrator(goal, depth=0, max_depth=5):
if depth > max_depth:
raise ValueError("Max recursion depth exceeded")
complexity = await estimate_complexity(goal)
if complexity <= THRESHOLD:
agent = select_best_agent(goal)
return await agent.run(goal)
sub_tasks = await decompose(goal)
results = await asyncio.gather(
*[recursive_orchestrator(st, depth+1) for st in sub_tasks]
)
return await synthesize(sub_tasks, results)
Advantages: Handles unknown complexity gracefully, adapts to task difficulty, no wasted overhead on simple tasks
Disadvantages: Risk of infinite recursion, unpredictable execution time, harder to estimate cost
5.5 Hybrid Orchestration
Hybrid orchestration combines multiple patterns based on the characteristics of each sub-task. Most real-world orchestrators are hybrid.
Goal Analysis ā Task Decomposition ā
āā Branch 1: Sequential (linear dependency chain)
ā āā A ā B ā C ā D
āā Branch 2: Parallel (independent tasks)
ā āā E āā
ā āā F āā¤ā G
ā āā H āā
āā Branch 3: Hierarchical (sub-manager)
ā āā Manager ā Workers
āā Branch 4: Recursive (if complex)
āā Decompose on demand
When to use:
- Virtually all production systems. Pure patterns are rare in practice.
- The workflow has a mix of sequential, parallel, and conditional elements
Example ā Multi-Article Research System:
async def hybrid_orchestrator(goal):
# Phase 1: Goal Analysis (sequential ā simple)
goal_spec = await analyze_goal(goal)
# Phase 2: Research (parallel ā independent topics)
research_tasks = decompose_into_topics(goal_spec)
research_results = await asyncio.gather(
*[research_agent.run(t) for t in research_tasks]
)
# Phase 3: Drafting (sequential ā builds on research)
draft = await writer_agent.run("Write draft", context=research_results)
# Phase 4: Review (parallel ā independent reviewers)
reviews = await asyncio.gather(
fact_checker.run(draft),
style_editor.
š Multi-Agent Systems ā 2723 linesComprehensive reference on multi-agent architectures, communication patterns, coordination mechanisms, consensus algorithms, task decomposition, error handling, memory sharing, topology patterns, and real-world implementations. ACPAgentAlignmentBenchmarkDeploymentEnsembleEvaluationGPTInferenceLLM Multi-Agent ArchitecturesMulti-agent systems (MAS) organize multiple AI agents to collaborate on complex tasks. The architecture defines how agents are structured, how they communicate, and how decisions are made. Below are the primary architectural patterns, each suited to different problem domains.
Hierarchical: Manager-Worker Architecture
The hierarchical architecture introduces a tiered structure where a manager agent supervises and delegates tasks to worker agents. This pattern mirrors organizational hierarchies in human teams.
Structure:
Manager Agent
āāā Worker Agent 1 (specialist)
āāā Worker Agent 2 (specialist)
āāā Worker Agent 3 (specialist)
āāā Worker Agent N (specialist)
Key Characteristics:
- Manager Agent: Responsible for task decomposition, assignment, result aggregation, and quality control. The manager maintains the global context and decides which worker handles which subtask.
- Worker Agents: Execute specific subtasks. Workers may be generalists or specialists with domain expertise. They report results back to the manager.
- Escalation Path: When a worker fails or encounters ambiguity, the manager intervenes or reassigns the task.
- Scalability: Adding more workers scales horizontally, but the manager can become a bottleneck.
Implementation Variants:
- Strict Hierarchy: Workers cannot communicate directly; all coordination flows through the manager.
- Relaxed Hierarchy: Workers can communicate peer-to-peer for subtask coordination, but final decisions rest with the manager.
- Dynamic Hierarchy: The manager role can be reassigned based on task requirements or agent capabilities.
Use Cases:
- Complex document generation (manager outlines, workers write sections)
- Code generation (manager designs architecture, workers implement modules)
- Multi-step research tasks (manager formulates questions, workers search and analyze)
- Customer support escalation (triage agent ā specialist agents ā supervisor)
Implementation Example (Pseudocode):
class ManagerAgent:
def orchestrate(self, task):
subtasks = self.decompose(task)
results = []
for subtask in subtasks:
worker = self.select_worker(subtask)
result = worker.execute(subtask)
results.append(result)
return self.aggregate(results)
class WorkerAgent:
def execute(self, subtask):
# Domain-specific execution logic
return self.process(subtask)
Advantages:
- Clear responsibility boundaries
- Natural fit for hierarchical organizations
- Simplified debugging and monitoring
- Modular specialization
Disadvantages:
- Single point of failure (manager)
- Communication overhead
- Potential for bottleneck at manager level
- Less flexible for dynamic task re-prioritization
Code Example: CrewAI Hierarchical:
from crewai import Agent, Task, Crew, Process
manager = Agent(
role="Project Manager",
goal="Coordinate the team to build a complete application",
backstory="Experienced engineering manager",
allow_delegation=True
)
developer = Agent(
role="Senior Developer",
goal="Write high-quality code",
backstory="Expert Python developer"
)
tester = Agent(
role="QA Engineer",
goal="Ensure code quality and test coverage",
backstory="Detail-oriented tester"
)
crew = Crew(
agents=[developer, tester],
manager_agent=manager,
process=Process.hierarchical
)
Sequential: Pipeline Architecture
The pipeline architecture chains agents in a sequence where each agent's output becomes the next agent's input. This is analogous to Unix pipes or assembly lines.
Structure:
Agent 1 ā Agent 2 ā Agent 3 ā ... ā Agent N
Key Characteristics:
- Linear Flow: Data passes through agents in a predetermined order.
- Specialization: Each agent performs a specific transformation on the data.
- Deterministic: The execution order is fixed and predictable.
- Bottleneck Awareness: The slowest agent determines overall throughput.
Variants:
- Simple Pipeline: Fixed order, all agents execute in sequence.
- Conditional Pipeline: Some agents may be skipped based on conditions.
- Dynamic Pipeline: The pipeline structure can be modified at runtime based on intermediate results.
Use Cases:
- Data processing ETL pipelines
- Document generation (research ā outline ā draft ā review ā finalize)
- Code review pipelines (lint ā test ā build ā deploy)
- Content moderation pipeline
Implementation Example:
class PipelineAgent:
def __init__(self, name, process_fn):
self.name = name
self.process_fn = process_fn
def run(self, input_data):
print(f"{self.name}: processing")
return self.process_fn(input_data)
class Pipeline:
def __init__(self, agents):
self.agents = agents
def execute(self, initial_input):
data = initial_input
for agent in self.agents:
data = agent.run(data)
return data
Advantages:
- Simple to design and debug
- Predictable execution order
- Easy to add, remove, or replace stages
- Natural parallelism between stages (if buffered)
Disadvantages:
- Inflexible for dynamic task structures
- Error in one stage breaks the entire chain
- Limited to linear workflows
- No feedback loops without additional complexity
Parallel: Fan-Out Architecture
The fan-out architecture dispatches tasks to multiple agents simultaneously, then aggregates results. This pattern is essential for throughput-sensitive applications.
Structure:
Task Dispatcher
āāā Agent 1 (parallel)
āāā Agent 2 (parallel)
āāā Agent 3 (parallel)
āāā Agent N (parallel)
ā
Result Aggregator
Key Characteristics:
- Concurrent Execution: Agents run in parallel, potentially on different threads/processes/machines.
- Data Independence: Each agent works on a subset of data or an independent aspect of the task.
- Synchronization Barrier: Results are collected and merged at a synchronization point.
- Scaling: Near-linear speedup for embarrassingly parallel workloads.
Variants:
- Pure Fan-Out: All agents execute the same function on different data shards.
- Specialized Fan-Out: Different agents with different capabilities work on different aspects.
- Competitive Fan-Out: Multiple agents tackle the same task; best result wins (see voting/ensemble).
Use Cases:
- Parallel document chunk processing
- Multi-source research and aggregation
- Ensemble model inference
- Batch data validation
- Web scraping across multiple sources
Implementation Example:
import asyncio
class FanOutOrchestrator:
def __init__(self, agents):
self.agents = agents
async def fan_out(self, task_data):
tasks = [agent.process(task_data) for agent in self.agents]
results = await asyncio.gather(*tasks)
return self.aggregate(results)
def aggregate(self, results):
# Merge or select best result
return results
Advantages:
- High throughput for independent subtasks
- Resilient to individual agent failures (if designed properly)
- Efficient resource utilization
- Scales horizontally with available compute
Disadvantages:
- Requires merge/aggregation logic
- Increased complexity for shared-state coordination
- Diminishing returns with too many parallel agents
- Debugging parallel execution is harder
Swarm: Homogeneous Architecture
Swarm architecture involves many identical or similar agents that collectively solve problems through emergent behavior, inspired by biological swarms (ants, bees, birds).
Structure:
Agent Pool (homogeneous)
āāā Agent (identical)
āāā Agent (identical)
āāā Agent (identical)
āāā Agent (identical)
āāā Agent (identical)
Coordinator (lightweight)
Key Characteristics:
- Homogeneous Agents: All agents have the same capabilities and prompts.
- Emergent Behavior: Complex outcomes arise from simple individual rules and local interactions.
- Decentralized: No single point of control; coordination is implicit.
- Scalability: Can scale to hundreds or thousands of agents.
Variants:
- Stochastic Swarm: Agents introduce randomness in their behavior for exploration.
- Greedy Swarm: Agents always take the locally optimal action.
- Role-Switching Swarm: Agents can adopt different roles dynamically.
Use Cases:
- Idea generation and brainstorming
- Diverse solution exploration
- Content variation generation
- Data augmentation
- Parallel hypothesis testing
Implementation Patterns:
class SwarmAgent:
def __init__(self, agent_id, system_prompt):
self.id = agent_id
self.system_prompt =
Communication PatternsCommunication is the backbone of multi-agent systems. The choice of communication pattern affects scalability, latency, reliability, and the types of coordination possible.
Direct Messaging
Point-to-point communication between agents, where one agent sends a message directly to another.
Characteristics:
- Addressed: Messages have a specific recipient.
- Synchronous or Asynchronous: Can be blocking or non-blocking.
- Private: Only the sender and recipient see the message.
- Direct: No intermediary or routing logic needed.
Implementation Approaches:
class Agent:
def __init__(self, name):
self.name = name
self.mailbox = []
def send_message(self, recipient, message):
recipient.receive_message(self.name, message)
def receive_message(self, sender, message):
self.mailbox.append((sender, message))
Use Cases:
- Task assignment from manager to worker
- Peer-to-peer coordination between specialized agents
- Query-response patterns between agents
Broadcast
One agent sends a message to all other agents in the system.
Characteristics:
- One-to-Many: A single sender reaches all recipients.
- Efficient Dissemination: Useful for global updates.
- No Targeting: All agents receive the message regardless of relevance.
- Potential for Information Overload: Agents must filter relevant messages.
Implementation Approaches:
class MessageBus:
def __init__(self):
self.subscribers = []
def subscribe(self, agent):
self.subscribers.append(agent)
def broadcast(self, sender, message):
for agent in self.subscribers:
if agent != sender:
agent.receive_broadcast(sender, message)
class Agent:
def receive_broadcast(self, sender, message):
if self.is_interested(message):
self.process_message(sender, message)
Use Cases:
- System-wide announcements (shutdown, configuration changes)
- Global state updates
- Emergency or alert messages
- Discovery and registration events
Blackboard / Shared Memory
A shared repository where agents read and write information, enabling indirect communication and collaboration.
Structure:
āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
ā Blackboard ā
ā āāāāāāāāāāāāāāāāāāāāāāāāā ā
ā ā Problem Description ā ā
ā āāāāāāāāāāāāāāāāāāāāāāāāā¤ ā
ā ā Partial Solutions ā ā
ā āāāāāāāāāāāāāāāāāāāāāāāāā¤ ā
ā ā Constraints ā ā
ā āāāāāāāāāāāāāāāāāāāāāāāāā¤ ā
ā ā Shared Knowledge ā ā
ā āāāāāāāāāāāāāāāāāāāāāāāāā ā
āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
ā ā ā
Agent 1 Agent 2 Agent 3
Characteristics:
- Decoupled: Agents don't need to know about each other.
- Shared State: All agents contribute to and read from a common repository.
- Opportunistic Contribution: Agents can contribute when they have relevant knowledge.
- Iterative Refinement: Solutions improve as agents add information.
Implementation:
class Blackboard:
def __init__(self):
self.data = {}
self.lock = asyncio.Lock()
async def write(self, key, value, agent_name):
async with self.lock:
self.data[key] = {"value": value, "author": agent_name, "timestamp": time.time()}
async def read(self, key):
async with self.lock:
return self.data.get(key)
async def query(self, filter_fn):
async with self.lock:
return {k: v for k, v in self.data.items() if filter_fn(k, v)}
Use Cases:
- Collaborative problem solving
- Incremental document building
- Shared research findings
- Multi-agent planning
Tool-Mediated Communication
Agents communicate indirectly through shared tools and their outputs.
Characteristics:
- Artifact-Based: Communication happens through tool outputs.
- Asynchronous: Agents can work at different times.
- Persistent: Tool outputs persist beyond the conversation.
- Auditable: All tool usage is logged.
Examples:
- Agent A writes to a shared database; Agent B reads from it.
- Agent A creates a file; Agent B processes that file.
- Agent A issues a API call; Agent B reads the API response.
- Agent A updates a shared document; Agent B reviews the changes.
Advantages:
- Natural integration with existing tools and APIs
- Persistent communication artifacts
- Parallel execution possible
- Interoperable with non-agent systems
Structured ACP (Agent Communication Protocol)
ACP defines formal message formats, ontologies, and interaction protocols for agent communication.
Characteristics:
- Standardized Format: Messages follow a defined schema (e.g., FIPA ACL, JSON-RPC).
- Speech Acts: Messages have performative types (request, inform, query, propose, accept, reject).
- Conversation Protocols: Predefined interaction patterns (FIPA Contract Net, English Auction).
- Content Languages: Shared ontology for unambiguous meaning.
FIPA ACL Message Structure:
class ACLMessage:
def __init__(self, performative, sender, receivers, content, ontology, language):
self.performative = performative # request, inform, query, propose, etc.
self.sender = sender
self.receivers = receivers
self.content = content
self.ontology = ontology
self.language = language
self.conversation_id = str(uuid.uuid4())
self.reply_with = None
self.in_reply_to = None
Common Performatives:
- INFORM: Agent shares information.
- REQUEST: Agent asks another to perform an action.
- QUERY-REF: Agent asks for information.
- PROPOSE: Agent proposes a course of action.
- ACCEPT-PROPOSAL: Agent accepts a proposal.
- REJECT-PROPOSAL: Agent rejects a proposal.
- CANCEL: Agent cancels a previous request.
- CONFIRM: Agent confirms receipt or agreement.
Use Cases:
- Enterprise multi-agent systems requiring standardization
- Interoperable agent platforms
- Formal verification of agent interactions
Coordination MechanismsCoordination mechanisms govern how agents align their actions to achieve coherent outcomes.
Consensus Building
Agents iteratively negotiate to reach agreement on a course of action or answer.
Process:
1. Agents express initial positions
2. Agents share reasoning and evidence
3. Agents update positions based on new information
4. Iterate until convergence or timeout
5. Final position is adopted as consensus
Implementation:
class ConsensusBuilder:
def __init__(self, agents, max_rounds=5, threshold=0.8):
self.agents = agents
self.max_rounds = max_rounds
self.threshold = threshold
def build_consensus(self, question):
positions = [agent.initial_position(question) for agent in self.agents]
for round in range(self.max_rounds):
agreement = self.measure_agreement(positions)
if agreement >= self.threshold:
return self.aggregate_positions(positions), "consensus"
# Share reasoning
for i, agent in enumerate(self.agents):
positions[i] = agent.update_position(question, positions)
return self.aggregate_positions(positions), "majority"
def measure_agreement(self, positions):
# Measure similarity or overlap in positions
pass
Voting
Agents cast votes on options, and the aggregated vote determines the outcome.
See detailed coverage in Consensus Algorithms section below.
MoE / Mixture of Agents
MoE architecture routes tasks to specialized sub-agents (experts) based on the input, using a gating/routing mechanism.
Architecture:
Input
ā
Router/Gate
ā
āāā Expert 1
āāā Expert 2
āāā Expert 3
āāā Expert N
ā
Top-k Aggregation
ā
Output
Key Concepts:
- Gating Network: Learns to route inputs to the most relevant experts.
- Sparse Activation: Only a subset of experts (top-k) are activated per input.
- Load Balancing: Ensures all experts receive sufficient training/usage.
- Shared Context: Experts may share some layers or parameters.
In Multi-Agent Context:
- Each "expert" is a full agent with specialized knowledge
- The router is typically an LLM or classification model
- Top-1 or top-2 routing is common
- Experts can be fine-tuned on domain-specific data
Code Example:
class MoERouter:
def __init__(self, agents, routing_fn=None):
self.agents = agents # List of specialized agents
self.routing_fn = routing_fn or self.default_router
def route(self, task):
# Get routing weights for all agents
weights = self.routing_fn(task)
# Select top-k agents
top_k = sorted(
enumerate(weights), key=lambda x: x[1], reverse=True
)[:self.k]
results = []
for idx, weight in top_k:
result = self.agents[idx].process(task)
results.append((result, weight))
return self.aggregate(results)
Debate
Structured argumentation where agents argue for different positions and a resolution mechanism determines the outcome. See Debate Architecture for details.
Referee / Arbitrator
A neutral agent that mediates disputes, ensures fairness, and makes final decisions when agents disagree.
Role of the Referee:
- Rule Enforcement: Ensures agents follow established protocols
- Conflict Resolution: Makes binding decisions when agents disagree
- Quality Assessment: Evaluates agent outputs for correctness
- Process Management: Controls turn-taking and speaking order
Implementation:
class RefereeAgent:
def __init__(self, rules):
self.rules = rules
def evaluate_submissions(self, agents, task):
submissions = [agent.produce_output(task) for agent in agents]
scores = []
for submission in submissions:
score = self.evaluate_quality(submission, task)
scores.append(score)
winner_idx = scores.index(max(scores))
return agents[winner_idx], submissions[winner_idx], scores
def resolve_dispute(self, agent_a, agent_b, issue):
# Make binding decision
return self.llm_judge(issue, agent_a.argument, agent_b.argument)
Use Cases:
- Tournament-style agent evaluation
- Quality-controlled generation
- Fair resource allocation
- Multi-agent learning with rewards
Auction-Based Coordination
Agents bid on tasks or resources, and an auctioneer allocates based on bids.
Types of Auctions:
- English Auction: Ascending bid, highest wins.
- Dutch Auction: Descending bid, first to accept wins.
- Vickrey Auction: Sealed bid, highest wins but pays second-highest price.
- Combinatorial Auction: Bids on bundles of items.
- Reverse Auction: Buyers bid for services (lowest wins).
Implementation Pattern:
class Auctioneer:
def __init__(self, auction_type="english"):
self.auction_type = auction_type
self.bids = {}
def announce_task(self, task, min_bid=0):
self.current_task = task
self.min_bid = min_bid
self.bids = {}
# Notify agents of new task
def receive_bid(self, agent, amount, capabilities):
self.bids[agent] = {"amount": amount, "capabilities": capabilities}
def resolve(self):
if self.auction_type == "english":
winner = max(self.bids, key=lambda a: self.bids[a]["amount"])
elif self.auction_type == "reverse":
winner = min(self.bids, key=lambda a: self.bids[a]["amount"])
return winner, self.current_task
Role-Based Coordination
Agents adopt specific roles with associated responsibilities and interaction protocols.
Common Roles in Multi-Agent Systems:
- Orchestrator: Coordinates agent activities and workflow
- Executor: Performs specific tasks
- Monitor: Observes system behavior and raises alerts
- Gatekeeper: Controls access to resources
- Validator: Verifies outputs and enfor... Agent Specialization PatternsAgent specialization enables efficient task allocation and high-quality outputs by matching tasks to agents with appropriate capabilities.
Role Assignment
Agents are assigned specific roles that define their responsibilities, authority, and interaction patterns.
Role Definition Structure:
Role:
name: "Senior Python Developer"
responsibilities:
- Write production-ready Python code
- Implement unit and integration tests
- Code review and refactoring
authorities:
- Modify source code in /src
- Approve pull requests
capabilities:
- Python, FastAPI, SQLAlchemy
- Docker, Kubernetes
- pytest, unittest
constraints:
- Must not modify database schema without review
- Must follow coding standards
Role Assignment Strategies:
- Hard-Coded: Roles are predefined in the system configuration.
- Capability-Based: Agents are assigned roles matching their demonstrated capabilities.
- Preference-Based: Agents express role preferences.
- Load-Balanced: Roles are assigned to balance workload across agents.
Skill-Based Routing
Tasks are routed to agents based on their demonstrated or declared skills.
Skill Taxonomy:
- Hard Skills: Technical abilities (coding, data analysis, API integration)
- Soft Skills: Communication, negotiation, empathy
- Domain Skills: Legal, medical, financial expertise
- Tool Skills: Proficiency with specific tools or APIs
Routing Strategies:
class SkillBasedRouter:
def __init__(self, agent_registry):
self.agent_registry = agent_registry
def find_best_agent(self, task_requirements):
candidates = []
for agent in self.agent_registry:
match_score = self.calculate_match(
task_requirements, agent.skills
)
candidates.append((agent, match_score))
candidates.sort(key=lambda x: x[1], reverse=True)
return candidates[0][0] if candidates else None
def calculate_match(self, requirements, skills):
# Weighted Jaccard similarity or learned embedding similarity
score = 0
for req, weight in requirements.items():
if req in skills:
score += weight * skills[req]
return score
Domain Expertise
Agents are specialized in specific knowledge domains with tailored system prompts, knowledge bases, and tools.
Domain Expert Design:
class DomainExpert(Agent):
def __init__(self, name, domain, knowledge_base=None, tools=None):
super().__init__(name)
self.domain = domain
self.knowledge_base = knowledge_base # Domain-specific vector store
self.tools = tools or [] # Domain-specific tools
self.system_prompt = self.build_system_prompt()
def build_system_prompt(self):
return f"""You are a {self.domain} expert.
You have access to domain-specific knowledge and tools.
Always ground your responses in domain expertise.
{self.get_domain_guidelines()}"""
def process(self, task):
context = self.knowledge_base.query(task.query)
return self.llm.generate(task, context, tools=self.tools)
Domain Boundaries:
- Clear definition of what each domain covers
- Confidence thresholds for domain relevance
- Escalation procedures for cross-domain queries
- Fallback to generalist when domain match is uncertain
Context Specialization
Agents specialize based on the context of the interaction, such as conversation history, user profile, or environmental conditions.
Context Dimensions:
- User Context: User preferences, history, language
- Temporal Context: Time of day, season, urgency
- Task Context: Complexity, domain, required output format
- System Context: Available resources, load, constraints
Implementation:
class ContextSpecializedAgent:
def __init__(self, context_profile, base_agent):
self.context_profile = context_profile
self.base_agent = base_agent
def matches_context(self, current_context):
return all(
current_context.get(k) == v
for k, v in self.context_profile.items()
)
def process(self, task, context):
if self.matches_context(context):
# Use specialized behavior
return self.specialized_process(task, context)
else:
# Fall back to base behavior
return self.base_agent.process(task)
Consensus AlgorithmsConsensus algorithms enable multiple agents to agree on answers, decisions, or courses of action.
Majority Voting
The simplest consensus method: each agent produces an answer, and the answer with the most votes wins.
Characteristics:
- Simple: Easy to implement and understand.
- Fair: Each agent has equal weight.
- Requires N ā„ 3: At least 3 agents needed for meaningful majority.
- Tie-Breaking: Random selection or fallback for ties.
Implementation:
def majority_vote(responses):
"""Standard majority voting."""
from collections import Counter
vote_counts = Counter(responses)
max_votes = max(vote_counts.values())
winners = [ans for ans, count in vote_counts.items() if count == max_votes]
if len(winners) == 1:
return winners[0]
return random.choice(winners) # Tie-breaking
Variants:
- Plurality Voting: Most votes wins even without absolute majority.
- Super-Majority: Requires > 66% or > 75% agreement.
- Unanimous: All agents must agree.
Weighted Voting
Agents have different voting weights based on expertise, historical accuracy, or confidence.
Weight Assignment:
- Confidence-Based: Agent's self-reported confidence score
- Accuracy-Based: Historical correctness rate
- Expertise-Based: Domain relevance weight
- Seniority-Based: Experience level or authority
Implementation:
def weighted_vote(responses, weights):
"""Weighted voting where each agent has a weight."""
score_map = {}
for response, weight in zip(responses, weights):
score_map[response] = score_map.get(response, 0) + weight
winner = max(score_map, key=score_map.get)
return winner, score_map[winner]
def confidence_weighted_vote(responses, confidences):
"""Vote weighted by agent confidence scores."""
return weighted_vote(responses, confidences)
Use Cases:
- Expert panels with varying expertise levels
- Systems where some agents are known to be more reliable
- When agents report confidence in their outputs
Borda Count
A ranked voting method where agents rank options, and points are assigned based on rank position.
Method:
1. Each agent ranks N options from best to worst.
2. The top-ranked option receives N-1 points, second receives N-2, etc.
3. Points are summed across all agents.
4. The option with the highest total wins.
Implementation:
def borda_count(rankings):
"""Borda count consensus.
rankings: list of lists, each sublist is an agent's ranking [best, ..., worst]
"""
N = len(rankings[0]) # Number of options
scores = {option: 0 for option in rankings[0]}
for agent_ranking in rankings:
for position, option in enumerate(agent_ranking):
scores[option] += (N - 1 - position) # N-1 for 1st, 0 for last
return max(scores, key=scores.get), scores
Advantages:
- Considers full preference ordering, not just top choice
- Less susceptible to strategic voting
- Produces more representative consensus
Sequential Voting
Multiple rounds of voting where options are eliminated or refined between rounds.
Variants:
- Runoff Voting: Bottom option eliminated each round until majority is achieved.
- Instant-Runoff: Voters rank options; lowest-ranked is eliminated; votes redistributed.
- Approval Voting: Multiple rounds approve/disapprove options.
Implementation:
def instant_runoff_voting(rankings):
"""Instant-runoff (ranked-choice) voting."""
active_options = list(rankings[0]) # All options initially active
while len(active_options) > 1:
# Count first-choice votes among active options
votes = {opt: 0 for opt in active_options}
for ranking in rankings:
for opt in ranking:
if opt in active_options:
votes[opt] += 1
break
# Eliminate lowest vote-getter
min_votes = min(votes.values())
eliminated = [opt for opt, v in votes.items() if v == min_votes]
if len(eliminated) == len(active_options):
break # All tied
active_options = [opt for opt in active_options if opt not in eliminated]
return active_options[0]
Pairwise Comparison
Options are compared in head-to-head matchups to determine the overall winner.
Methods:
- Condorcet Method: An option that beats all others in pairwise comparisons wins.
- Copeland's Method: Score based on number of pairwise wins minus losses.
- Schulze Method: Beatpaths to resolve cycles in pairwise preferences.
Characteristics:
- Condorcet Winner: May not always exist (voting cycles).
- Complete Ranking: Produces a full ordered ranking.
- Computationally Intensive: O(NĀ²) comparisons for N options.
Debate with Referee
Agents argue for their positions, and a referee/judge agent evaluates the arguments to determine the winner.
Process:
1. Agents present initial arguments.
2. Agents rebut opposing arguments (multiple rounds).
3. Referee evaluates arguments based on evidence, reasoning, and relevance.
4. Referee declares winner or produces synthesized answer.
Implementation:
class DebateWithReferee:
def __init__(self, agents, referee, rounds=3):
self.agents = agents
self.referee = referee
self.rounds = rounds
def resolve(self, question):
arguments = {agent: agent.initial_argument(question) for agent in self.agents}
for r in range(self.rounds):
for agent in self.agents:
# Respond to other agents' arguments
others = [a for a in self.agents if a != agent]
for other in others:
arguments[agent] = agent.rebut(
question, arguments[other], arguments[agent]
)
# Referee makes final judgment
return self.referee.judge(question, arguments)
Mixture of Agents (TogetherAI)
TogetherAI's MoA architecture uses multiple LLMs as "proposers" and "aggregators" in a layered architecture.
Architecture:
Layer 1 (Proposers):
GPT-4o, Claude 3.5, Llama 3.1, Mixtral, Qwen 2
Each produces an initial response
Layer 2 (Aggregators):
GPT-4o (aggregator) reviews and synthesizes layer 1 outputs
May...
Task DecompositionTask decomposition breaks complex tasks into smaller, manageable subtasks that can be distributed among agents.
Hierarchical Decomposition
Tasks are broken down recursively into a tree of subtasks, where each level represents increasing granularity.
Structure:
Main Task
āāā Subtask 1
ā āāā Subtask 1.1
ā āāā Subtask 1.2
ā āāā Subtask 1.3
āāā Subtask 2
ā āāā Subtask 2.1
ā āāā Subtask 2.2
āāā Subtask 3
āāā Subtask 3.1
āāā Subtask 3.2
āāā Subtask 3.3
Decomposition Strategies:
- Top-Down: Start with the main task and recursively decompose.
- Bottom-Up: Identify atomic tasks and group them.
- Pattern-Based: Use known decomposition patterns for common task types.
Implementation:
class HierarchicalDecomposer:
def __init__(self, llm):
self.llm = llm
def decompose(self, task, depth=0, max_depth=3):
if depth >= max_depth or self.is_atomic(task):
return [task]
subtasks = self.llm.decompose_task(task)
result = []
for subtask in subtasks:
result.extend(self.decompose(subtask, depth + 1, max_depth))
return result
def is_atomic(self, task):
# Check if task is simple enough for single agent
return task.complexity_score < self.threshold
DAG Decomposition
Tasks are decomposed into a directed acyclic graph (DAG) where nodes are subtasks and edges represent dependencies.
Characteristics:
- Dependency-Aware: Captures which subtasks must complete before others.
- Parallelism: Independent subtasks can execute in parallel.
- Flexible Ordering: Any topological ordering is valid.
- Comprehensive: Captures complex relationships beyond sequential or tree structures.
DAG Representation:
class DAGTask:
def __init__(self, task_id, description):
self.id = task_id
self.description = description
self.dependencies = [] # List of DAGTask IDs
self.subtasks = []
class DAGDecomposer:
def decompose_to_dag(self, task):
# Use LLM to identify subtasks and their dependencies
subtasks = self.llm.identify_subtasks(task)
dependencies = self.llm.identify_dependencies(subtasks)
dag = {}
for st in subtasks:
dag[st.id] = DAGTask(st.id, st.description)
dag[st.id].dependencies = dependencies.get(st.id, [])
return dag
def get_execution_order(self, dag):
"""Return topological ordering of tasks."""
visited = set()
order = []
def dfs(node_id):
if node_id in visited:
return
visited.add(node_id)
node = dag[node_id]
for dep_id in node.dependencies:
dfs(dep_id)
order.append(node_id)
for node_id in dag:
dfs(node_id)
return order
Sequential Decomposition
Tasks are decomposed into a linear sequence of steps.
Characteristics:
- Linear Order: Each step depends on the previous.
- Simple: Easy to understand and implement.
- Predictable: Fixed execution order.
Use Cases:
- Recipe following
- Step-by-step procedures
- Document template filling
Parallel Decomposition
Tasks are decomposed into independent subtasks that can execute simultaneously.
Characteristics:
- No Dependencies: Subtasks are independent of each other.
- Maximum Parallelism: All subtasks can run concurrently.
- Merge Required: Results must be combined at the end.
Use Cases:
- Processing independent data shards
- Multi-source fact-checking
- Multiple solution generation
Recursive Decomposition
Tasks are decomposed using the same decomposition strategy applied recursively.
Characteristics:
- Self-Similar: Decomposition pattern repeats at each level.
- Natural Fit: Works well for hierarchical problems.
- Potential for Deep Nesting: Needs depth limits.
Feedback-Driven Decomposition
Decomposition is refined based on intermediate results and feedback.
Process:
1. Initial decomposition of the main task.
2. Execute subtasks and collect results.
3. Analyze intermediate results for issues or gaps.
4. Refine decomposition based on feedback.
5. Re-execute modified subtasks.
6. Iterate until satisfactory completion.
Implementation:
class FeedbackDrivenDecomposer:
def __init__(self, llm, max_iterations=5):
self.llm = llm
self.max_iterations = max_iterations
def execute_with_feedback(self, main_task):
plan = self.llm.create_initial_plan(main_task)
for iteration in range(self.max_iterations):
results = self.execute_plan(plan)
feedback = self.analyze_results(results, main_task)
if feedback.is_complete:
return self.synthesize(results)
plan = self.llm.refine_plan(plan, results, feedback)
return self.synthesize(results)
Error Handling in Multi-Agent SystemsMulti-agent systems introduce unique failure modes beyond single-agent errors. Robust error handling is essential for production deployments.
Retry with Different Agent
When an agent fails or produces low-quality output, the task is retried with a different agent.
Strategy:
class RetryWithDifferentAgent:
def __init__(self, agents, fallback_order=None):
self.agents = agents
self.fallback_order = fallback_order or list(agents)
def execute_with_retry(self, task, max_retries=3):
errors = []
for i in range(min(max_retries, len(self.fallback_order))):
agent = self.fallback_order[i]
try:
result = agent.process(task)
if self.validate_result(result):
return result
else:
errors.append((agent.name, "Validation failed"))
except Exception as e:
errors.append((agent.name, str(e)))
continue
raise MultiAgentError(f"All agents failed: {errors}")
def validate_result(self, result):
# Quality checks, format validation, etc.
return result is not None and result.quality_score > 0.7
Considerations:
- Agent ordering: Try specialized agents first, fall back to generalists
- Timeout management: Set per-agent timeouts to avoid cascading delays
- Error context: Pass error information to fallback agents
Fallback Chain
A predefined chain of fallback strategies or agents, tried in order until one succeeds.
Fallback Chain Structure:
Primary Strategy
ā Fallback 1 (different approach)
ā Fallback 2 (simplified approach)
ā Fallback 3 (default response)
Example Chain:
class FallbackChain:
def __init__(self, strategies):
self.strategies = strategies # Ordered list of (name, handler_fn)
def execute(self, task):
errors = []
for name, handler in self.strategies:
try:
result = handler(task)
if result is not None:
return result
except Exception as e:
errors.append(f"{name}: {e}")
continue
return self.default_response(task)
Fallback Strategies (Ordered):
1. Primary agent with full context
2. Different agent with simplified context
3. Single-step generation (no decomposition)
4. Rule-based template response
5. Graceful error message to user
Circuit Breaker
Prevents cascading failures by detecting when a component is failing and temporarily stopping requests to it.
States:
- CLOSED: Normal operation, requests pass through.
- OPEN: Failures detected, requests are rejected immediately.
- HALF-OPEN: Testing if service has recovered.
Implementation:
class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=30):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED"
def call(self, agent_fn, task):
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "HALF-OPEN"
else:
raise CircuitBreakerOpen("Agent temporarily unavailable")
try:
result = agent_fn(task)
self.on_success()
return result
except Exception as e:
self.on_failure()
raise
def on_success(self):
if self.state == "HALF-OPEN":
self.state = "CLOSED"
self.failure_count = 0
def on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
Consensus Verification
Multiple agents independently verify outputs to catch errors.
Process:
1. Primary agent produces output.
2. Verification agents independently check the output.
3. Consensus among verifiers determines if output is accepted.
4. If verification fails, output is rejected and re-generated.
Implementation:
class ConsensusVerifier:
def __init__(self, verifiers, required_approvals=2):
self.verifiers = verifiers
self.required_approvals = required_approvals
def verify(self, output, context):
approvals = 0
rejections = []
for verifier in self.verifiers:
result = verifier.check(output, context)
if result.approved:
approvals += 1
else:
rejections.append(result.reason)
if approvals >= self.required_approvals:
return True, []
return False, rejections
Verification Strategies:
- Factual Verification: Check claims against known facts
- Logical Verification: Check reasoning consistency
- Format Verification: Check output format compliance
- Style Verification: Check tone and style guidelines
Human-in-the-Loop Escalation
When agents cannot resolve an issue, the system escalates to a human operator.
Escalation Criteria:
- Low confidence scores across all agents
- Repeated failures or timeouts
- Novel or ambiguous queries
- Sensitive topics requiring human judgment
- Edge cases outside agent training
Escalation Flow:
class HITLEScalation:
def __init__(self, auto_resolve_threshold=0.9):
self.auto_resolve_threshold = auto_resolve_threshold
def process_with_escalation(self, task):
# Try automatic processing
result = self.auto_process(task)
if result.confidence >= self.
Agent Topology PatternsThe topology defines how agents are connected and interact within the system.
Linear Chain
Agents are arranged in a line, with each agent communicating only with its immediate neighbors.
Structure:
A1 ā A2 ā A3 ā ... ā An
Characteristics:
- Simple communication paths
- Clear data flow direction
- Limited connectivity
- Single path for information
Use Cases:
- Sequential processing pipelines
- Transformation chains
- Simple workflows
Tree Topology
Agents are organized hierarchically in a tree structure.
Structure:
Root
/ \
A2 A3
/ \ / \
A4 A5 A6 A7
Characteristics:
- Hierarchical control flow
- Parent-child relationships
- Broadcast from root to leaves
- Aggregation from leaves to root
Use Cases:
- Hierarchical task decomposition
- Organizational hierarchies
- Decision trees
Star / Hub-and-Spoke Topology
A central hub agent communicates with all peripheral agents.
Structure:
Characteristics:
- Centralized coordination
- Hub is single point of communication
- Spokes only communicate through hub
- Easy to monitor and control
Use Cases:
- Manager-worker patterns
- Centralized information routing
- Load balancing
Mesh / Fully Connected Topology
Every agent can communicate directly with every other agent.
Structure:
A1 ---- A2
| \ / |
| X |
| / \ |
A3 ---- A4
Characteristics:
- Maximum connectivity
- Redundant communication paths
- High flexibility
- High complexity for N agents (NĆ(N-1)/2 connections)
Use Cases:
- Small teams requiring full collaboration
- Debate and discussion systems
- Peer review networks
DAG Topology
Agents are arranged in a directed acyclic graph, allowing parallel paths and dependencies.
Structure:
A1
/ \
A2 A3
| |
A4 A5
\ /
A6
Characteristics:
- Supports parallelism and dependencies
- No cycles (enforces acyclic flow)
- Flexible routing
- Complex to validate
Use Cases:
- Complex workflows with branching
- Multi-stage processing
- Dependency-aware task execution
Modular Topology
The topology is composed of reusable modules, each containing multiple agents with internal topology.
Structure:
Module1 Module2
āāāāāāāā āāāāāāāā
ā A1-A2āāāāāāāā A3-A4ā
ā ā ā ā ā ā
ā A5 ā ā A6 ā
āāāāāāāā āāāāāāāā
Characteristics:
- Encapsulated agent groups
- Well-defined interfaces between modules
- Reusable and composable
- Scalable through module addition
Use Cases:
- Large-scale enterprise systems
- Microservice-based agent architectures
- Domain-separated systems
Dynamic Topology
The topology can change at runtime based on task requirements, agent availability, or performance.
Characteristics:
- Self-Organizing: Agents form and break connections as needed.
- Adaptive: Topology adjusts to changing conditions.
- Resilient: Can route around failed agents.
- Complex: Requires monitoring and reconfiguration logic.
Implementation Considerations:
- Topology discovery protocol
- Health monitoring and heartbeat
- Reconfiguration triggers and strategies
- Consistency guarantees during transitions
Memory SharingMemory sharing enables agents to access and contribute to shared knowledge, enabling coherent collaboration.
Shared Vector Store
A shared vector database where agents store and retrieve embeddings of knowledge, context, and intermediate results.
Architecture:
Agents ā Shared Vector Store (Chroma/Pinecone/Qdrant)
āāā Knowledge embeddings
āāā Conversation history
āāā Task results
āāā Shared context
Implementation:
class SharedVectorStore:
def __init__(self, collection_name, embedding_fn):
self.collection = chroma_client.get_or_create_collection(
name=collection_name,
embedding_function=embedding_fn
)
def store_agent_output(self, agent_name, content, metadata=None):
self.collection.add(
documents=[content],
metadatas=[{"agent": agent_name, **(metadata or {})}],
ids=[f"{agent_name}_{uuid.uuid4()}"]
)
def query_shared_knowledge(self, query, n_results=5):
return self.collection.query(
query_texts=[query],
n_results=n_results
)
Use Cases:
- Cross-agent retrieval augmented generation
- Long-term memory for agent conversations
- Shared knowledge base
- Task result caching
Shared Message Bus
A publish-subscribe message bus that agents use to exchange messages asynchronously.
Components:
- Topics/Channels: Named channels for message categories.
- Publishers: Agents that produce messages.
- Subscribers: Agents that consume messages.
- Broker: Routes messages from publishers to subscribers.
Implementation Pattern:
class MessageBus:
def __init__(self):
self.topics = defaultdict(list)
self.history = defaultdict(list)
def publish(self, topic, message, sender):
self.history[topic].append({"sender": sender, "message": message, "timestamp": time.time()})
for subscriber in self.topics[topic]:
subscriber.receive(topic, message, sender)
def subscribe(self, topic, agent):
self.topics[topic].append(agent)
def get_history(self, topic, since=None):
if since:
return [m for m in self.history[topic] if m["timestamp"] > since]
return self.history[topic]
Shared State
A central state store that agents can read from and write to, enabling coordinated state management.
Characteristics:
- Consistent: All agents see the same state (subject to consistency guarantees).
- Transactional: Updates can be atomic and isolated.
- Observable: Agents can watch for state changes.
Implementation:
class SharedState:
def __init__(self, backend="redis"):
self.store = RedisClient() if backend == "redis" else InMemoryStore()
self.watchers = defaultdict(list)
def set(self, key, value, agent_name):
old_value = self.store.get(key)
self.store.set(key, {"value": value, "updated_by": agent_name, "timestamp": time.time()})
self.notify_watchers(key, value, old_value)
def get(self, key):
return self.store.get(key)
def watch(self, key, callback):
self.watchers[key].append(callback)
def transaction(self, updates):
"""Atomic batch update."""
with self.lock:
for key, value in updates:
self.set(key, value)
Blackboard Pattern
A shared knowledge repository where agents contribute partial solutions and retrieve needed information. See Blackboard / Shared Memory section.
Advanced Blackboard Features:
- Structured Blackboard: Organized into panels/levels for different types of information.
- Source Tracking: Each entry records which agent contributed it.
- Confidence Scoring: Entries have associated confidence levels.
- Lifecycle Management: Entries can be proposed, verified, accepted, or rejected.
Cross-Agent RAG
Retrieval-Augmented Generation where agents query each other's knowledge stores or a shared knowledge base.
Architecture:
Agent A (query)
ā
āāāā Shared Vector Store (documents)
āāāā Agent B's Knowledge (via tool call)
āāāā Agent C's Knowledge (via tool call)
ā
āāāā Aggregate retrieved context ā Generate response
Implementation:
class CrossAgentRAG:
def __init__(self, shared_store, peer_agents):
self.shared_store = shared_store
self.peer_agents = peer_agents # Other agents that can provide knowledge
def retrieve(self, query):
# Retrieve from shared store
shared_results = self.shared_store.query(query)
# Retrieve from peer agents
peer_results = []
for agent in self.peer_agents:
try:
result = agent.provide_knowledge(query)
peer_results.append(result)
except Exception:
continue
# Aggregate and deduplicate
all_results = self.merge_results(shared_results, peer_results)
return all_results
Tool Sharing ModelsTools enable agents to interact with external systems. Tool sharing models determine how agents discover and use tools.
Global Tool Registry
A centralized registry where all tools are registered and any agent can discover and use any tool.
Architecture:
Global Tool Registry
āāā Tool: WebSearch
ā āāā Agent A can use
ā āāā Agent B can use
ā āāā Agent C can use
āāā Tool: CodeInterpreter
ā āāā Agent A can use
ā āāā Agent B can use
āāā Tool: DatabaseQuery
āāā Agent A can use
āāā Agent C can use
Implementation:
class GlobalToolRegistry:
def __init__(self):
self.tools = {}
def register_tool(self, name, tool_fn, description, schema):
self.tools[name] = {
"fn": tool_fn,
"description": description,
"schema": schema,
"usage_count": 0
}
def get_tool(self, agent, tool_name):
if tool_name in self.tools:
self.tools[tool_name]["usage_count"] += 1
return self.tools[tool_name]["fn"]
return None
def list_tools(self, agent=None):
return {name: info["description"] for name, info in self.tools.items()}
Advantages:
- Single source of truth for tools
- Easy to add, update, or remove tools
- Usage tracking and monitoring
- Access control at registry level
Disadvantages:
- Centralized point of failure
- Registry can become a bottleneck
- All agents see all tools (potential confusion)
Per-Agent Tools
Each agent has its own set of tools, configured specifically for its role and capabilities.
Configuration:
class PerAgentToolConfig:
def __init__(self):
self.agent_tools = {}
def configure_agent(self, agent_name, tools):
self.agent_tools[agent_name] = [
{"name": name, "config": config}
for name, config in tools.items()
]
def get_agent_tools(self, agent_name):
return self.agent_tools.get(agent_name, [])
# Example configuration
config = PerAgentToolConfig()
config.configure_agent("research_agent", {
"web_search": {"engine": "bing", "max_results": 10},
"fetch_page": {"timeout": 5}
})
config.configure_agent("code_agent", {
"python_repl": {},
"file_ops": {"allowed_paths": ["/workspace"]},
"git_ops": {}
})
Advantages:
- Clean separation of concerns
- Agents only see relevant tools
- Reduced prompt complexity
- Security isolation between agents
Disadvantages:
- Tool duplication across agents
- Configuration overhead
- Harder to share tools dynamically
Capability-Based Routing
Tools are matched to tasks based on the capabilities they provide, independent of which agent uses them.
Capability Model:
Tools:
WebSearchTool:
capabilities:
- web_search
- information_retrieval
input: query string
output: search results
CodeInterpreter:
capabilities:
- code_execution
- data_analysis
- visualization
input: code string
output: execution result
DatabaseQuery:
capabilities:
- data_query
- database_access
input: SQL query
output: query results
Routing Logic:
class CapabilityRouter:
def __init__(self, tool_registry):
self.tool_registry = tool_registry
def find_tools_for_capability(self, required_capabilities):
matching_tools = []
for tool_name, tool_info in self.tool_registry.items():
if any(cap in tool_info["capabilities"] for cap in required_capabilities):
matching_tools.append(tool_name)
return matching_tools
def route_task(self, task, agent):
required_caps = task.required_capabilities
available_tools = self.find_tools_for_capability(required_caps)
# Filter to tools agent is allowed to use
agent_tools = [t for t in available_tools if agent.can_use(t)]
return agent_tools
Tool Output Caching
Results from tool executions are cached so that repeated calls with the same or similar inputs return cached results.
Caching Strategies:
- Exact Match Cache: Identical inputs return cached result.
class ExactMatchCache:
def __init__(self, ttl=300):
self.cache = {}
self.ttl = ttl
def get_or_execute(self, tool_fn, tool_args):
cache_key = self.make_key(tool_fn.__name__, tool_args)
if cache_key in self.cache:
entry = self.cache[cache_key]
if time.time() - entry["timestamp"] < self.ttl:
return entry["result"]
result = tool_fn(**tool_args)
self.cache[cache_key] = {"result": result, "timestamp": time.time()}
return result
def make_key(self, fn_name, args):
return f"{fn_name}:{hash(frozenset(args.items()))}"
- Semantic Cache: Similar inputs return cached result based on embedding similarity.
class SemanticCache:
def __init__(self, embedding_fn, similarity_threshold=0.95):
self.entries = []
self
Real-World Patterns and FrameworksAutoGen Group Chat
Microsoft's AutoGen provides a flexible group chat pattern for multi-agent conversations.
Architecture:
GroupChatManager
āāā Agent A (UserProxy)
āāā Agent B (Assistant)
āāā Agent C (Critic)
āāā Agent D (Executor)
āāā Agent E (Custom)
Key Concepts:
- GroupChat: Manages a conversation among multiple agents.
- GroupChatManager: Orchestrates the group chat, selecting the next speaker.
- ConversableAgent: Base class for all agents in AutoGen.
- Speaker Selection: Configurable strategies (round-robin, auto, manual, custom).
Speaker Selection Methods:
- Round Robin: Agents speak in fixed order.
- Auto: LLM decides who speaks next based on conversation context.
- Random: Random selection for exploratory conversations.
- Manual: Human selects the next speaker.
- Custom: User-defined selection logic.
Implementation Pattern:
import autogen
# Define agents
planner = autogen.AssistantAgent(
name="Planner",
system_message="You are a planner. Break down tasks into steps.",
llm_config={"config_list": [{"model": "gpt-4", "api_key": "..."}]}
)
executor = autogen.AssistantAgent(
name="Executor",
system_message="You execute tasks and produce results.",
llm_config={"config_list": [{"model": "gpt-4", "api_key": "..."}]}
)
critic = autogen.AssistantAgent(
name="Critic",
system_message="You review and critique work.",
llm_config={"config_list": [{"model": "gpt-4", "api_key": "..."}]}
)
# Create group chat
groupchat = autogen.GroupChat(
agents=[planner, executor, critic],
messages=[],
max_round=10,
speaker_selection_method="auto"
)
manager = autogen.GroupChatManager(
groupchat=groupchat,
llm_config={"config_list": [{"model": "gpt-4", "api_key": "..."}]}
)
CrewAI Hierarchical / Sequential
CrewAI provides both hierarchical and sequential process patterns.
Sequential Process:
from crewai import Crew, Process
crew = Crew(
agents=[agent1, agent2, agent3],
tasks=[task1, task2, task3],
process=Process.sequential
)
# Tasks execute in order: task1 ā task2 ā task3
Hierarchical Process:
crew = Crew(
agents=[developer, tester, writer],
tasks=[main_task],
manager_agent=manager,
process=Process.hierarchical
)
# Manager decomposes main_task, assigns to workers
Key Features:
- Role Assignment: Each agent has a role, goal, and backstory.
- Task Delegation: Manager can delegate subtasks to workers.
- Tool Sharing: Agents can share tools defined at crew level.
- Memory: Crew-level memory for context across tasks.
- Callbacks: Hooks for logging, monitoring, and custom actions.
LangGraph Graphs
LangGraph provides a graph-based approach to building agent workflows.
Key Concepts:
- StateGraph: Graph where nodes modify shared state.
- MessageGraph: Specialized for chat message passing.
- Nodes: Functions or agents that process state.
- Edges: Define transitions between nodes.
- Conditional Edges: Dynamic routing based on state.
- Checkpointing: Save and restore state for persistence.
- Human-in-the-Loop: Interrupt execution for human input.
Implementation Pattern:
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import operator
class AgentState(TypedDict):
messages: List
next_agent: str
def agent_a(state):
# Process and return updated state
return {"messages": state["messages"] + ["A processed"]}
def agent_b(state):
return {"messages": state["messages"] + ["B processed"]}
def router(state):
# Conditional routing logic
if len(state["messages"]) < 3:
return "agent_b"
return END
# Build graph
graph = StateGraph(AgentState)
graph.add_node("agent_a", agent_a)
graph.add_node("agent_b", agent_b)
graph.set_entry_point("agent_a")
graph.add_edge("agent_a", "agent_b")
graph.add_conditional_edges("agent_b", router)
graph.set_finish_point(END)
# Compile and run
app = graph.compile()
result = app.invoke({"messages": []})
Semantic Kernel Planner
Microsoft's Semantic Kernel provides planning capabilities for multi-step tasks.
Planner Types:
- HandlebarsPlanner: Uses Handlebars templates for prompt-based planning.
- FunctionCallingStepwisePlanner: Step-by-step planning with function calling.
- StepwisePlanner: Creates sequential plans from available functions.
Architecture:
Kernel
āāā Plugins (collections of functions)
ā āāā Native Functions (C#/Python code)
ā āāā Semantic Functions (AI prompts)
āāā Planner (creates execution plan)
āāā Memory (vector store integration)
āāā Connectors (OpenAI, HuggingFace, etc.)
smolagents Code Agents
HuggingFace's smolagents framework provides code-first agent architecture.
Key Concepts:
- CodeAgent: Agent that writes and executes Python code.
- ToolCallingAgent: Agent that calls tools via function calling.
- ManagedAgent: Sub-agent managed by a main agent.
- Tool Sharing: Tools can be shared across agents.
Architecture:
from smolagents import CodeAgent, HfApiModel, ManagedAgent
# Create tools
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Results for {query}"
# Create agent
agent = CodeAgent(
tools=[search_web],
model=HfApiModel(),
add_base_tools=True
)
# Multi-agent setup
managed_agent = ManagedAgent(
agent=agent,
name="researcher",
description="Handles research tasks"
)
main_agent = CodeAgent(
tools=[],
model=HfApiModel(),
managed_agents=[managed_agent]
)
OpenAI Swarm
OpenAI's Swarm framework provides lightweight multi-agent orchestration.
Key Concepts:
- Routines: Agents with specific instructions and functions.
- Handoffs: Agents can transfer control to other agents.
- Context Variables: Shared state across agents.
Implementation:
from swarm import Swarm, Agent
client = Swarm()
def transfer_to_spanish_agent():
return spanish_agent
english_agent = Agent(
name="English Agent",
instructions="You speak English.",
functions=[transfer_to_spanish_agent]
)
spanish_agent = Agent(
name="Spanish Agent",
instructions="You speak Spanish."
)
# Run with handoff
response = client.run(
agent=
Performance Metrics for Multi-Agent SystemsMeasuring multi-agent system performance requires metrics at multiple levels: system-level, task-level, and agent-level.
System-Level Metrics
| Metric |
Description |
Measurement |
| Throughput |
Tasks completed per unit time |
tasks/second |
| Latency (p50/p95/p99) |
Time from task submission to completion |
milliseconds |
| Scalability |
Performance change with additional agents |
Īthroughput / Īagents |
| Availability |
System uptime / total time |
percentage |
| Cost Efficiency |
Quality per unit cost |
score / $ |
| Resource Utilization |
Agent idle vs active time |
percentage |
| Fault Tolerance |
Performance degradation under failures |
graceful degradation |
Task-Level Metrics
| Metric |
Description |
| Task Completion Rate |
Percentage of tasks completed successfully |
| Task Accuracy |
Correctness of outputs |
| Task Quality Score |
Human or automated quality rating |
| Task Decomposition Efficiency |
Overhead of task decomposition |
| First-Time Success Rate |
Tasks completed without retry |
| Time to Resolution |
End-to-end task completion time |
Agent-Level Metrics
| Metric |
Description |
| Individual Accuracy |
Per-agent output correctness |
| Contribution Value |
Impact on final outcome |
| Response Time |
Per-agent processing time |
| Tool Utilization |
Frequency and efficiency of tool use |
| Error Rate |
Frequency of agent failures |
| Collaboration Score |
Effectiveness in multi-agent context |
Communication Metrics
| Metric |
Description |
| Message Volume |
Total messages exchanged |
| Messages per Task |
Communication overhead |
| Average Message Size |
Information density |
| Coordination Overhead |
Time spent in coordination vs execution |
| Convergence Time |
Time to reach consensus |
| Agreement Rate |
How often agents agree |
Quality Metrics
| Metric |
Description |
Method |
| Consistency |
Output consistency across similar inputs |
Variance analysis |
| Robustness |
Performance under noisy inputs |
Stress testing |
| Diversity |
Variance in agent outputs |
Entropy measurement |
| Novelty |
Unique insights produced |
Human evaluation |
| Coherence |
Logical consistency of multi-agent output |
Automated checks |
Cost Metrics
| Metric |
Description |
| Token Cost per Task |
Total LLM tokens consumed |
| API Call Volume |
Number of external API calls |
| Compute Cost |
Infrastructure cost per task |
| Human Oversight Cost |
Time spent by humans reviewing outputs |
| Storage Cost |
Memory and state storage costs |
Monitoring Dashboard Example
class MASMonitor:
def __init__(self):
self.metrics = defaultdict(list)
def record_task_completion(self, task_id, duration, success, cost):
self.metrics["task_duration"].append(duration)
self.metrics["task_success"].append(success)
self.metrics["task_cost"].append(cost)
def record_message(self, sender, receiver, size):
self.metrics["messages"].append({
"sender": sender,
"receiver": receiver,
"size": size,
"timestamp": time.time()
})
def summary(self):
return {
"total_tasks": len(self.metrics["task_duration"]),
"avg_duration": mean(self.metrics["task_duration"]),
"p95_duration": percentile(self.metrics["task_duration"], 95),
"success_rate": mean(self.metrics["task_success"]),
"total_cost": sum(self.metrics["task_cost"]),
"total_messages": len(self.metrics["messages"]),
"agents_active": len(set(m["sender"] for m in self.metrics["messages"]))
}
š Agentic Frameworks ā 2514 linesComprehensive reference and comparison of major agentic frameworks: LangGraph, CrewAI, AutoGen, Semantic Kernel, smolagents, DSPy, OpenAI Agents SDK, Anthropic MCP, Google ADK, Amazon Bedrock Agents, Agno/Phidata, and Pydantic AI. AgentDPODeploymentEvaluationGPTLLMLoRAMCPMulti-AgentOrchestrator LangGraphLangGraph is a framework developed by LangChain for building stateful, multi-agent applications using graph-based state machines. It enables developers to define agent workflows as directed graphs where nodes are computation steps and edges define control flow.
Core Architecture
LangGraph models agent workflows as a state machine where each node in the graph represents a step that transforms the shared state.
StateGraph
StateGraph is the primary graph type. It operates on a shared state object that persists across nodes.
Key Components:
- State Schema: A TypedDict or Pydantic model defining the structure of the shared state.
- Nodes: Functions or runnable objects that take state as input and return state updates.
- Edges: Define the flow between nodes (directed connections).
- Conditional Edges: Dynamic routing based on state content.
- Entry Point: The first node to execute.
- Finish Points: Nodes where execution terminates.
Basic StateGraph Example:
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import operator
class GraphState(TypedDict):
input: str
messages: List[str]
next_step: str
def node_a(state: GraphState) -> GraphState:
# Process and return updates (only returning changes)
return {"messages": state["messages"] + ["Node A processed"]}
def node_b(state: GraphState) -> GraphState:
return {"messages": state["messages"] + ["Node B processed"]}
def router(state: GraphState) -> str:
"""Conditional edge: decide which node to go to next."""
if "error" in state:
return "error_handler"
elif state["next_step"] == "b":
return "node_b"
return END
# Build the graph
builder = StateGraph(GraphState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("error_handler", error_handler)
builder.set_entry_point("node_a")
builder.add_edge("node_a", "node_b")
builder.add_conditional_edges("node_b", router, {
"node_a": "node_a",
"error_handler": "error_handler",
END: END
})
# Compile into an executable app
app = builder.compile()
MessageGraph
MessageGraph is a specialized graph for chat-based applications. It operates on lists of messages rather than arbitrary state.
from langgraph.graph import MessageGraph, END
from langchain_core.messages import HumanMessage, AIMessage
def assistant_node(state):
messages = state["messages"]
response = llm.invoke(messages)
return {"messages": messages + [response]}
def human_review_node(state):
# Interrupt for human review
return {"messages": state["messages"]}
# MessageGraph is simpler - just message passing
graph = MessageGraph()
graph.add_node("assistant", assistant_node)
graph.add_node("human_review", human_review_node)
graph.set_entry_point("assistant")
graph.add_edge("assistant", "human_review")
graph.add_conditional_edges("human_review",
lambda state: "assistant" if state["approved"] else END)
Node Types and Patterns
Nodes in LangGraph are the fundamental building blocks. They can be:
- Function Nodes: Simple Python functions that take state and return state updates.
def my_node(state: dict) -> dict:
# Process state
return {"key": "value"}
- Runnable Nodes: LangChain Runnable objects (chains, LCEL pipelines).
from langchain_core.runnables import RunnableLambda
chain = prompt | llm | parser
builder.add_node("llm_call", chain)
- Subgraph Nodes: Entire graphs used as nodes within a larger graph.
subgraph = create_subgraph().compile()
builder.add_node("sub_task", subgraph)
- Agent Nodes: Full agent loops that can make multiple tool calls.
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(model, tools)
builder.add_node("agent_step", agent)
Edges and Control Flow
LangGraph provides several types of edges for different control flow patterns:
Direct Edges
Simple, unconditional transitions between nodes.
graph.add_edge("node_a", "node_b")
Conditional Edges
Edges that route based on a function evaluating the current state.
def route_decision(state: dict) -> str:
if state["quality"] > 0.8:
return "finalize"
elif state["attempts"] < 3:
return "retry"
return "fallback"
graph.add_conditional_edges(
"evaluator",
route_decision,
{
"finalize": "output_node",
"retry": "generator",
"fallback": "human_review"
}
)
Entry and Finish Points
graph.set_entry_point("start_node")
graph.set_finish_point("end_node")
# Can also set conditional entry:
graph.set_conditional_entry_point(entry_router)
Checkpointing and Persistent State
LangGraph provides built-in checkpointing for state persistence, enabling pause/resume, human-in-the-loop, and fault recovery.
from langgraph.checkpoint import MemorySaver
# In-memory checkpointing
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
# SQLite checkpointing (persistent)
from langgraph.checkpoint import SqliteSaver
with SqliteSaver.from_conn_string("checkpoints.db") as saver:
app = graph.compile(checkpointer=saver)
# Run with thread ID for conversation continuity
config = {"configurable": {"thread_id": "conversation_1"}}
result = app.invoke({"input": "Hello"}, config)
Checkpointing Features:
- Automatic State Saving: State is saved after each node execution.
- Thread-Based Isolation: Each conversation/run gets a unique thread_id.
- Pause/Resume: Execution can be paused and resumed later.
- State Inspection: View state at any checkpoint.
- Branching: Fork from any checkpoint for exploration.
Streaming
LangGraph supports multiple streaming modes for real-time output:
CrewAICrewAI is a framework for orchestrating role-based AI agents that work together on complex tasks. It provides a high-level abstraction for multi-agent systems.
Core Architecture
CrewAI is built around four core entities: Agents, Tasks, Crews, and Processes.
Crew
āāā Process (sequential / hierarchical)
āāā Agents
ā āāā Agent 1 (role, goal, backstory, tools)
ā āāā Agent 2 (role, goal, backstory, tools)
ā āāā Agent N (...)
āāā Tasks
ā āāā Task 1 (description, agent, tools)
ā āāā Task 2 (description, agent, tools)
ā āāā Task N (...)
āāā Memory (optional)
āāā Short-term
āāā Long-term
āāā Entity memory
Agents
Agents in CrewAI are defined by their role, goal, backstory, and available tools.
from crewai import Agent
researcher = Agent(
role="Senior Research Analyst",
goal="Uncover cutting-edge developments in AI and technology",
backstory="""You are a seasoned research analyst with over 15 years
of experience in technology analysis. Known for deep dives into
complex topics and identifying emerging trends.""",
allow_delegation=False,
verbose=True,
tools=[search_tool, scrape_tool],
llm="gpt-4-turbo",
max_iter=15,
max_rpm=10,
memory=True
)
Agent Parameters:
| Parameter | Description |
|-----------|-------------|
| role | Agent's function within the crew |
| goal | Individual objective the agent is working towards |
| backstory | Context and personality for the agent |
| allow_delegation | Whether agent can delegate tasks to other agents |
| tools | List of tools available to the agent |
| llm | Language model to use (default: gpt-4) |
| max_iter | Maximum iterations before forced answer |
| max_rpm | Maximum requests per minute (rate limiting) |
| verbose | Enable detailed logging |
| memory | Enable memory for this agent |
Tasks
Tasks define what needs to be done and can be assigned to specific agents.
from crewai import Task
research_task = Task(
description="""Research the latest developments in {topic}.
Focus on major breakthroughs in the last 6 months.""",
expected_output="""A comprehensive 3-page report covering:
- Key breakthroughs and their implications
- Companies and researchers leading the work
- Timeline of major developments""",
agent=researcher,
tools=[search_tool], # Override tools for this specific task
context=[], # Context from other tasks
async_execution=False, # Whether to run asynchronously
callback=task_callback_function,
human_input=False
)
Task Features:
- Context Passing: Tasks can receive context from previous tasks.
- Async Execution: Tasks can run in parallel.
- Callbacks: Hooks for task lifecycle events.
- Human Input: Option to request human input during task execution.
- Output Customization: Define expected output format.
Process Types
CrewAI supports two primary process types:
Sequential Process
Tasks are executed in order, with each task receiving context from the previous.
from crewai import Crew, Process
crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, writing_task, editing_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff()
Execution Flow:
Task 1 (Research) ā Task 2 (Writing) ā Task 3 (Editing)
ā ā ā
Output Output Final Output
When to Use:
- Simple linear workflows
- Tasks with clear dependencies
- When order of execution matters
- Prototyping and simple automation
Hierarchical Process
A manager agent orchestrates worker agents, decomposing tasks and assigning them dynamically.
manager = Agent(
role="Project Manager",
goal="Efficiently manage the team to achieve the project goals",
backstory="Experienced project manager with a track record of delivering complex projects",
allow_delegation=True,
llm="gpt-4-turbo"
)
crew = Crew(
agents=[researcher, writer, editor, designer],
tasks=[project_task],
manager_agent=manager,
process=Process.hierarchical,
verbose=True
)
Execution Flow:
Manager Agent
āāā Decomposes project into subtasks
āāā Assigns subtasks to appropriate agents
āāā Monitors progress and quality
āāā Synthesizes final output
Worker Agents
āāā Receive assignments from manager
āāā Execute their specific tasks
āāā Report results back to manager
When to Use:
- Complex projects requiring planning
- When agent specialization is important
- Dynamic task allocation needed
- Quality control through manager oversight
Role Assignment
CrewAI's role system provides structured agent definition:
# Predefined Roles
from crewai.roles import Role
custom_role = Role(
name="Data Scientist",
goal="Extract insights from data",
backstory="Expert in statistical analysis and machine learning",
tools=[python_repl,ę°ę®åę_tools]
)
Role Assignment Strategies:
- Single Role: Each agent has one fixed role.
- Context-Enhanced Roles: Roles are augmented with task-specific context.
- Dynamic Role Adjustment: Agent behavior adjusts based on conversation.
Task Delegation
Agents can delegate subtasks to other agents when allow_delegation=True.
researcher = Agent(
role="Lead Researcher",
goal="Coordinate research efforts",
allow_delegation=True,
tools=[delegation_tool]
)
# The researcher agent can request help from other agents
# during task execution
Delegation Flow:
1. Agent receives a task.
2. Agent identifies sub-parts that require other expertise.
3. Agent delegates sub-tasks to appropriate agents.
4. Delegated agents return results.
5. Original agent synthesizes all results.
Tool Sharing
Tools can be shared across agents at the crew level or agent level.
from crewai_tools import tool
@tool("WebSearch")
def web_search(query: str) -> str:
"""Search the web for information."""
return search_results
@tool("Calculator")
def calculator(expression: str) -> str:
"""Evaluate mathematical expressions."""
return str(eval(expression))
# Agent-level tools
AutoGen / AutoGen StudioAutoGen is a multi-agent conversation framework developed by Microsoft Research. AutoGen Studio provides a visual interface for designing and debugging agent workflows.
Core Concepts
ConversableAgent
The base class for all agents in AutoGen. It defines the fundamental communication interface.
import autogen
class ConversableAgent:
"""
Base class for agents that can converse with each other.
"""
def __init__(self, name, system_message, llm_config, human_input_mode="NEVER"):
self.name = name
self.system_message = system_message
self.llm_config = llm_config
self.human_input_mode = human_input_mode
Key Methods:
- send(message, recipient, request_reply): Send a message to another agent.
- receive(message, sender, request_reply): Receive a message from another agent.
- generate_reply(messages, sender): Generate a reply based on conversation history.
- initiate_chat(recipient, message): Start a conversation with another agent.
AssistantAgent
An agent that uses an LLM to generate responses.
assistant = autogen.AssistantAgent(
name="assistant",
system_message="You are a helpful assistant.",
llm_config={
"config_list": [
{
"model": "gpt-4",
"api_key": os.environ["OPENAI_API_KEY"]
}
],
"temperature": 0.7
}
)
UserProxyAgent
An agent that can act on behalf of a human user, including executing code and providing feedback.
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
human_input_mode="TERMINATE",
max_consecutive_auto_reply=10,
code_execution_config={
"work_dir": "coding",
"use_docker": False, # Set to True for sandboxed execution
}
)
Two-Agent Chat
The simplest pattern: two agents converse to solve a task.
# Create agents
assistant = autogen.AssistantAgent(
name="assistant",
llm_config=llm_config
)
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
code_execution_config={"work_dir": "coding"}
)
# Initiate conversation
user_proxy.initiate_chat(
assistant,
message="Write a Python script to calculate Fibonacci numbers"
)
Chat Flow:
UserProxy ā Assistant: "Write a Python script..."
Assistant ā UserProxy: "Here's a script..."
UserProxy ā Assistant: Executes script, returns output
Assistant ā UserProxy: "The result is..."
... continues until termination condition
Group Chat
Multiple agents converse in a group chat managed by a GroupChatManager.
from autogen import GroupChat, GroupChatManager
# Define multiple agents
planner = autogen.AssistantAgent(
name="Planner",
system_message="You are a planner. Break down tasks.",
llm_config=llm_config
)
coder = autogen.AssistantAgent(
name="Coder",
system_message="You are a Python programmer.",
llm_config=llm_config
)
critic = autogen.AssistantAgent(
name="Critic",
system_message="Review and critique code.",
llm_config=llm_config
)
# Create group chat
groupchat = GroupChat(
agents=[planner, coder, critic, user_proxy],
messages=[],
max_round=12,
speaker_selection_method="auto", # Options: auto, round_robin, random, manual
allow_repeat_speaker=False, # Prevent same speaker twice in a row
)
# Create manager
manager = GroupChatManager(
groupchat=groupchat,
llm_config=llm_config
)
# Start conversation
user_proxy.initiate_chat(
manager,
message="Create a web scraper for news articles"
)
Speaker Selection Methods:
| Method | Description |
|--------|-------------|
| auto | LLM decides next speaker based on context |
| round_robin | Speakers take turns in fixed order |
| random | Random selection |
| manual | Human selects next speaker |
| Custom | User-defined selection function |
Function Calling and Tool Registration
AutoGen supports function calling for tool integration.
# Define a function/tool
def search_web(query: str) -> str:
"""Search the web for information."""
import requests
response = requests.get(f"https://api.search.com?q={query}")
return response.text
# Register with assistant
assistant = autogen.AssistantAgent(
name="assistant",
llm_config={
"config_list": [...],
"functions": [{
"name": "search_web",
"description": "Search the web",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query"
}
},
"required": ["query"]
}
}]
}
)
# Register function with user proxy
user_proxy.register_function(
function_map={
"search_web": search_web
}
)
Function Registration for Tool Calling:
# Alternative: function registration with tool calling
@user_proxy.register_for_execution()
@assistant.register_for_llm(description="Search the web")
def web_search(query: str) -> str:
return search_api(query)
Code Execution Sandbox
AutoGen provides configurable code execution environments:
# Local execution
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
code_execution_config={
"work_dir": "coding",
"use_docker": False,
"timeout": 60
}
)
# Docker sandbox (secure)
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
code_execution_config={
"work_dir": "coding",
"use_docker": True,
"docker_container_name": "autogen-sandbox",
"timeout": 120
}
)
# Custom executor
user_proxy = autogen.UserProxyAgent(
name="user_proxy",
code_execution_config={
"executor": custom_executor
}
)
Sequential, Parallel, and Debate Patterns
Sequential Pattern:
Semantic KernelSemantic Kernel (SK) is an SDK by Microsoft that integrates LLMs with conventional programming languages.
Core Architecture
Semantic Kernel
āāā Kernel (central orchestrator)
āāā Plugins (collections of functions)
ā āāā Native Functions (code)
ā āāā Semantic Functions (prompts)
āāā Planners (auto function chaining)
āāā Memory (semantic memory / vector store)
āāā Connectors (AI services, memory stores)
āāā Filters (content moderation, logging)
Kernel
The kernel is the central orchestrator that manages plugins, memory, and AI service connections.
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion
# Create kernel
kernel = Kernel()
# Add AI service
kernel.add_service(
AzureChatCompletion(
deployment_name="gpt-4",
endpoint="https://your-endpoint.openai.azure.com",
api_key="your-api-key"
)
)
# The kernel connects all components
result = await kernel.invoke_prompt_async(
"What is the capital of France?",
stream=True
)
Plugins
Plugins are collections of functions that the AI can use.
Native Functions (Code)
from semantic_kernel.plugin import kernel_function
class EmailPlugin:
@kernel_function(
description="Send an email to a recipient",
name="send_email"
)
def send_email(self, to: str, subject: str, body: str) -> str:
"""Send an email using the SMTP server."""
# Email sending logic
return f"Email sent to {to}"
@kernel_function(
description="Get unread email count",
name="get_unread_count"
)
def get_unread_count(self) -> int:
"""Get the number of unread emails."""
return 42
# Add plugin to kernel
email_plugin = EmailPlugin()
kernel.add_plugin(email_plugin, "Email")
Semantic Functions (Prompts)
from semantic_kernel.functions import KernelFunctionFromPrompt
# Define prompt-based function
summarize_function = kernel.create_function_from_prompt(
function_name="Summarize",
plugin_name="TextPlugin",
prompt="""Summarize the following text in 3 sentences:
{{$input}}
Summary:"""
)
# Or use YAML/Prompt template files
# Save as TextPlugin/Summarize/skprompt.txt
kernel.add_plugin_from_directory("./plugins/TextPlugin")
Planners
Planners automatically chain functions together to accomplish complex goals.
HandlebarsPlanner
Uses Handlebars templates for plan generation.
from semantic_kernel.planners import HandlebarsPlanner
planner = HandlebarsPlanner(
kernel=kernel,
prompt="""Create a plan to accomplish: {{$goal}}
Available functions: {{$functions}}"""
)
plan = await planner.create_plan_async(
goal="Send a summary email to the team"
)
FunctionCallingStepwisePlanner
Iterative planner that uses function calling to build plans step by step.
from semantic_kernel.planners import FunctionCallingStepwisePlanner
planner = FunctionCallingStepwisePlanner(
kernel=kernel,
max_tokens=2000,
max_iterations=10
)
result = await planner.execute_plan_async(
goal="Research latest AI news and create a report"
)
Planning Process:
1. Decompose the goal into steps.
2. Match steps to available functions.
3. Execute functions in order.
4. Adapt plan based on intermediate results.
5. Return final output.
StepwisePlanner
The original planner that creates step-by-step execution plans.
from semantic_kernel.planners import StepwisePlanner
planner = StepwisePlanner(kernel)
plan = planner.create_plan(goal="Compare stock prices of Apple and Microsoft")
Connectors
Connectors bridge Semantic Kernel with external services.
from semantic_kernel.connectors.ai.open_ai import (
AzureChatCompletion,
OpenAIChatCompletion
)
from semantic_kernel.connectors.ai.anthropic import AnthropicChatCompletion
from semantic_kernel.connectors.ai.google import GooglePalmChatCompletion
from semantic_kernel.connectors.memory.chroma import ChromaMemoryStore
from semantic_kernel.connectors.memory.qdrant import QdrantMemoryStore
# Multiple AI connectors
kernel.add_service(AzureChatCompletion(...))
kernel.add_service(OpenAIChatCompletion(...))
kernel.add_service(AnthropicChatCompletion(...))
# Memory connectors
kernel.add_memory_store(ChromaMemoryStore(persist_directory="./chroma"))
kernel.add_memory_store(QdrantMemoryStore(host="localhost", port=6333))
Filters
Filters provide middleware-like capabilities for the kernel pipeline.
from semantic_kernel.filters import (
FunctionFilter,
PromptFilter,
KernelFilter
)
class LoggingFilter(FunctionFilter):
async def on_function_invoking(self, context):
print(f"Invoking function: {context.function.name}")
async def on_function_invoked(self, context):
print(f"Function completed: {context.function.name}")
print(f"Duration: {context.duration}")
class SafetyFilter(PromptFilter):
async def on_prompt_rendering(self, context):
if contains_harmful_content(context.prompt):
context.cancel = True
# Add filters to kernel
kernel.add_filter(LoggingFilter())
kernel.add_filter(SafetyFilter())
Memory and Vector Store Integration
from semantic_kernel.memory import SemanticTextMemory
from semantic_kernel.memory.memory_store import MemoryStore
# Configure memory
memory = SemanticTextMemory(
storage=ChromaMemoryStore(),
embeddings_generator=kernel.get_service("text_embedding")
)
# Store information
await memory.save_information_async(
collection="facts",
text="The Eiffel Tower is in Paris",
id="fact_1"
)
# Retrieve information
result = await memory.search_async(
collection="facts",
query="Where is the Eiffel Tower?",
limit=3
)
OpenAI / Anthropic / Google Integration
# OpenAI
kernel.add_service(
OpenAIChatCompletion(
model_id="gpt-4",
api_key="...",
org_id="..."
)
)
# Azure OpenAI
kernel.add_service(
smolagentssmolagents is HuggingFace's framework for building AI agents that can write and execute code.
Core Philosophy
smolagents distinguishes between two types of agents:
- CodeAgent: Writes and executes Python code to accomplish tasks.
- ToolCallingAgent: Uses tool calling APIs (function calling).
Both are built on a shared base architecture.
CodeAgent
The CodeAgent writes Python code to solve tasks, then executes it.
from smolagents import CodeAgent, HfApiModel
# Create a CodeAgent
agent = CodeAgent(
model=HfApiModel(),
tools=[], # Custom tools
add_base_tools=True, # Include base tools like Python interpreter
max_steps=10,
verbosity_level=1
)
# Run the agent
result = agent.run(
"Calculate the 50th Fibonacci number and check if it's prime."
)
How CodeAgent Works:
1. Receives a task.
2. Generates Python code to solve it.
3. Executes the code in a sandboxed Python environment.
4. If execution fails, the agent can fix the code and retry.
5. Returns the final result.
ToolCallingAgent
The ToolCallingAgent uses function calling APIs to invoke tools.
from smolagents import ToolCallingAgent, tool
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
return str(eval(expression))
agent = ToolCallingAgent(
model=HfApiModel(),
tools=[search_web, calculate],
max_steps=5
)
result = agent.run("What is 2^10? And search for 'AI news'")
Multi-Agent Hierarchy
smolagents supports hierarchical multi-agent setups through ManagedAgent.
from smolagents import CodeAgent, ManagedAgent, HfApiModel
# Create specialized sub-agents
research_agent = CodeAgent(
model=HfApiModel(),
tools=[search_tool],
add_base_tools=True
)
code_agent = CodeAgent(
model=HfApiModel(),
tools=[python_repl],
add_base_tools=True
)
# Wrap as managed agents
managed_researcher = ManagedAgent(
agent=research_agent,
name="researcher",
description="Researchs topics online and returns findings"
)
managed_coder = ManagedAgent(
agent=code_agent,
name="coder",
description="Writes and executes Python code"
)
# Main agent with sub-agents
main_agent = CodeAgent(
model=HfApiModel(),
managed_agents=[managed_researcher, managed_coder],
tools=[],
add_base_tools=True
)
# Run
result = main_agent.run(
"Research quantum computing and write a simulation"
)
Agent Hierarchy:
Main Agent
āāā Managed Agent: Researcher
ā āāā Tools: web_search, fetch_page
āāā Managed Agent: Coder
āāā Tools: python_repl, file_ops
Tool Sharing
Tools can be defined and shared across agents:
from smolagents import tool, CodeAgent, ToolCollection
# Define tools
@tool
def get_weather(location: str) -> str:
"""Get current weather for a location."""
return f"Weather in {location}: Sunny, 72Ā°F"
@tool
def get_time(timezone: str) -> str:
"""Get current time in a timezone."""
return f"Time in {timezone}: 14:30"
# Create a tool collection
tools = ToolCollection([get_weather, get_time])
# Share across agents
agent1 = CodeAgent(tools=tools, ...)
agent2 = CodeAgent(tools=tools, ...)
Hub Integration
smolagents integrates with the HuggingFace Hub for model and tool sharing:
from smolagents import HfApiModel, CodeAgent, ToolCollection
# Use models from HuggingFace Hub
model = HfApiModel(
model_id="Qwen/Qwen2.5-72B-Instruct",
token="hf_..."
)
# Load tools from Hub
tools = ToolCollection.from_hub(
"huggingface/tools-web-search",
token="hf_..."
)
agent = CodeAgent(model=model, tools=tools)
Gradio UI
smolagents includes a Gradio-based UI for interactive agent usage:
from smolagents import CodeAgent, HfApiModel
from smolagents.gradio_ui import GradioUI
agent = CodeAgent(
model=HfApiModel(),
tools=[...],
add_base_tools=True
)
# Launch UI
GradioUI(agent).launch()
DSPyDSPy is a framework for programmatically composing and optimizing language model programs.
Core Philosophy
DSPy treats LMs as programmable components. Instead of hand-crafting prompts, you define signatures and modules, then use optimizers to automatically improve performance.
Signatures
Signatures define the input/output behavior of an LM call.
import dspy
# Define a signature
class GenerateAnswer(dspy.Signature):
"""Answer questions based on context."""
context = dspy.InputField(desc="Contains relevant facts")
question = dspy.InputField(desc="The question to answer")
answer = dspy.OutputField(desc="The answer")
# Use the signature
module = dspy.Predict(GenerateAnswer)
result = module(context="Paris is the capital of France.", question="What is the capital of France?")
print(result.answer) # "Paris"
Signature Fields:
- InputField: Marks input parameters with optional descriptions.
- OutputField: Marks output parameters.
- desc: Natural language description of the field.
Modules
DSPy modules encapsulate LM calls with signatures.
# Basic Predict module
predict = dspy.Predict(GenerateAnswer)
# Chain of Thought
cot = dspy.ChainOfThought(GenerateAnswer)
# ReAct (Reasoning + Acting)
react = dspy.ReAct(GenerateAnswer, tools=[search_tool])
# Multi-Chain Comparison
multi = dspy.MultiChainComparison(GenerateAnswer, M=3) # Compare 3 chains
# Retrieve then Read (RAG)
rag = dspy.RetrieveThenRead(GenerateAnswer, retriever=retriever)
# Program of Thought (PoT)
pot = dspy.ProgramOfThought(GenerateAnswer)
Optimizers (Teleprompters)
DSPy optimizers automatically tune prompts and few-shot examples.
BootstrapFewShot
from dspy.teleprompt import BootstrapFewShot
# Configure optimizer
optimizer = BootstrapFewShot(
metric=answer_accuracy,
max_bootstrapped_demos=8,
max_labeled_demos=16,
max_rounds=3
)
# Compile (optimize) the program
compiled_program = optimizer.compile(
student=program,
teacher=program,
trainset=training_data,
valset=validation_data
)
MIPRO (Multi-Instruction Proposal Optimizer)
from dspy.teleprompt import MIPRO
optimizer = MIPRO(
metric=answer_accuracy,
num_bayes_trials=10,
num_candidates=5,
verbose=True
)
compiled = optimizer.compile(
program,
trainset=training_data,
num_trials=30
)
MIPRO Features:
- Proposes multiple instruction variations.
- Uses Bayesian optimization to find best combinations.
- Tunes both instructions and few-shot examples.
- More thorough optimization than BootstrapFewShot.
BayesianSignatureOptimizer
from dspy.teleprompt import BayesianSignatureOptimizer
optimizer = BayesianSignatureOptimizer(
metric=answer_accuracy,
n_trials=20
)
compiled = optimizer.compile(
program,
trainset=training_data
)
Compiled Programs
Once optimized, DSPy programs can be saved and loaded:
# Save compiled program
compiled_program.save("optimized_rag_program.dspy")
# Load and use
loaded_program = dspy.load("optimized_rag_program.dspy")
result = loaded_program(context="...", question="...")
Multi-Stage Pipelines
DSPy can compose multiple modules into pipelines:
class RAGPipeline(dspy.Module):
def __init__(self, retriever):
super().__init__()
self.retriever = retriever
self.generate_answer = dspy.ChainOfThought(GenerateAnswer)
self.check_facts = dspy.Predict(VerifyFacts)
def forward(self, question):
# Stage 1: Retrieve
context = self.retriever(question)
# Stage 2: Generate answer
answer = self.generate_answer(context=context, question=question)
# Stage 3: Verify facts
verification = self.check_facts(answer=answer.answer, context=context)
return dspy.Prediction(
answer=answer.answer,
confidence=verification.confidence
)
# Create and optimize
pipeline = RAGPipeline(retriever=retriever)
optimized = optimizer.compile(pipeline, trainset=trainset)
Evaluation
DSPy provides built-in evaluation tools:
from dspy.evaluate import Evaluate
# Create evaluator
evaluator = Evaluate(
devset=test_data,
metric=answer_accuracy,
num_threads=4,
display_progress=True
)
# Evaluate
score = evaluator(compiled_program)
print(f"Accuracy: {score}")
# Custom metrics
def exact_match(example, prediction, trace=None):
return example.answer.strip() == prediction.answer.strip()
evaluator = Evaluate(devset=test_data, metric=exact_match)
OpenAI Agents SDKThe OpenAI Agents SDK is a lightweight framework for building agent-based applications using OpenAI models.
Core Architecture
OpenAI Agents SDK
āāā Agent (runnable agent instance)
āāā Handoffs (agent-to-agent transfer)
āāā Guardrails (input/output validation)
āāā Tracing (observability)
āāā Agent Loop (execution engine)
Agent Loop
The core execution loop:
from agents import Agent, Runner
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant.",
model="gpt-4o"
)
result = Runner.run_sync(agent, "What is the capital of France?")
print(result.final_output)
Agent Loop Process:
1. Receive user input.
2. Process through model.
3. Execute any tool calls.
4. Continue until final output.
5. Return result.
Handoffs
Agents can transfer control to specialized sub-agents:
from agents import Agent, Runner
# Specialized agents
spanish_agent = Agent(
name="Spanish Agent",
instructions="You speak Spanish. Respond in Spanish.",
)
french_agent = Agent(
name="French Agent",
instructions="You speak French. Respond in French.",
)
# Main agent with handoff functions
def transfer_to_spanish():
return spanish_agent
def transfer_to_french():
return french_agent
main_agent = Agent(
name="Main Agent",
instructions="Route users to the right language specialist.",
functions=[transfer_to_spanish, transfer_to_french]
)
result = Runner.run_sync(main_agent, "Parlez-vous franƧais?")
Guardrails
Guardrails validate inputs and outputs:
from agents import Agent, Runner, Guardrail
# Input guardrail
class TopicGuardrail(Guardrail):
async def check_input(self, agent, input_text):
if "illegal" in input_text.lower():
return GuardrailResult(blocked=True, message="Cannot process this request")
return GuardrailResult(blocked=False)
# Output guardrail
class SafetyGuardrail(Guardrail):
async def check_output(self, agent, output):
if contains_harmful_content(output):
return GuardrailResult(blocked=True, message="Output blocked")
return GuardrailResult(blocked=False)
agent = Agent(
name="Safe Agent",
instructions="Be helpful and safe.",
input_guardrails=[TopicGuardrail()],
output_guardrails=[SafetyGuardrail()]
)
Tracing
Built-in tracing for observability:
from agents import Agent, Runner, trace
# Automatic tracing
with trace("my_app"):
result = Runner.run_sync(agent, "Question 1")
result2 = Runner.run_sync(agent, "Question 2")
# Manual tracing
trace.add_event("custom_event", {"key": "value"})
Comparison with Older Function Calling
| Aspect |
Agents SDK |
Older Function Calling |
| Abstraction |
Full agent lifecycle |
Raw function calling |
| Handoffs |
Built-in |
Manual implementation |
| Guardrails |
First-class |
Custom validation |
| Tracing |
Built-in |
None |
| Multi-turn |
Managed loop |
Manual conversation management |
| State |
Context management |
Manual state tracking |
Anthropic MCP ClientAnthropic's Model Context Protocol (MCP) client enables standardized tool and resource access.
Architecture
MCP Client
āāā MCP Server Connection
āāā Tool Discovery
āāā Resource Access
āāā Prompt Management
āāā Sampling (optional)
Key Concepts
- MCP Server: Provides tools and resources.
- Client: Connects to servers and exposes capabilities to LLMs.
- Tools: Functions the model can call.
- Resources: Data sources the model can read.
- Prompts: Reusable prompt templates.
Client Implementation
import asyncio
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
async def main():
# Connect to MCP server
server_params = StdioServerParameters(
command="python",
args=["mcp_server.py"]
)
async with ClientSession(server_params) as session:
# Initialize connection
await session.initialize()
# List available tools
tools = await session.list_tools()
# Call a tool
result = await session.call_tool(
"search_web",
{"query": "latest AI news"}
)
# Access resources
resource = await session.read_resource("database://users")
Google Agent Development Kit (ADK)Google's ADK is a framework for building agents using Google's AI models and tools.
Architecture
Google ADK
āāā Agent (base agent class)
āāā Tools (function, API, code execution)
āāā Memory (conversation, session)
āāā Auth (Google OAuth, service accounts)
āāā Deployment (Cloud Run, Cloud Functions)
Key Features
- Gemini Integration: First-class support for Gemini models.
- Google Workspace Tools: Gmail, Calendar, Drive integration.
- Vertex AI Agent Builder: Visual agent builder.
- Enterprise Security: IAM, VPC-SC, CMEK.
- Streaming Support: Real-time response streaming.
Agent Definition
from google.adk import Agent
from google.adk.tools import FunctionTool
def search_flights(origin, destination, date):
"""Search for flights."""
return flight_api.search(origin, destination, date)
agent = Agent(
model="gemini-2.0-flash",
tools=[
FunctionTool(search_flights),
GoogleWorkspaceToolkit()
],
system_instruction="You are a travel assistant."
)
Amazon Bedrock AgentsAmazon Bedrock Agents enable building agents using foundation models on AWS.
Architecture
Amazon Bedrock Agent
āāā Foundation Model (Claude, Llama, Mistral, etc.)
āāā Action Groups (Lambda functions)
āāā Knowledge Base (RAG with vector store)
āāā Guardrails (content filtering)
āāā Orchestration (step-by-step planning)
Key Features
- Managed Infrastructure: Fully managed by AWS.
- Multi-Model Support: Choose from various foundation models.
- Knowledge Base: Built-in RAG with Aurora/OpenSearch.
- Guardrails: Content filtering and safety controls.
- CloudWatch Integration: Monitoring and logging.
- IAM Security: AWS IAM integration.
- Cost Management: Pay-per-use pricing.
Agent Configuration
import boto3
bedrock_agent = boto3.client("bedrock-agent")
# Create agent
response = bedrock_agent.create_agent(
agentName="customer-support-agent",
foundationModel="anthropic.claude-3-sonnet-20240229",
instruction="You are a helpful customer support agent.",
actionGroups=[{
"actionGroupName": "ticket-actions",
"description": "Actions for managing support tickets",
"actionGroupExecutor": {
"lambda": "arn:aws:lambda:...:function:ticket-handler"
},
"functionSchema": {
"functions": [
{
"name": "create_ticket",
"description": "Create a support ticket",
"parameters": {
"customer_id": {"type": "string", "description": "Customer ID"},
"issue": {"type": "string", "description": "Issue description"}
}
}
]
}
}],
knowledgeBases=[{
"knowledgeBaseId": "KB123",
"description": "Product documentation"
}]
)
Knowledge Base Integration
# Create knowledge base
bedrock_agent.create_knowledge_base(
name="product-docs",
roleArn="arn:aws:iam::...:role/bedrock-knowledge-base",
knowledgeBaseConfiguration={
"type": "VECTOR",
"vectorKnowledgeBaseConfiguration": {
"embeddingModelArn": "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v2"
}
},
storageConfiguration={
"type": "OPENSEARCH_SERVERLESS",
"opensearchServerlessConfiguration": {
"collectionArn": "arn:aws:aoss:...:collection/docs",
"vectorIndexName": "bedrock-knowledge-base-index",
"fieldMapping": {
"metadataField": "metadata",
"textField": "text"
}
}
}
)
Agno / PhidataAgno (formerly Phidata) is a framework for building agent-as-function applications.
Core Architecture
Agno Agent
āāā Model (LLM backend)
āāā Tools (function registry)
āāā Knowledge Base (vector store)
āāā Memory (conversation history)
āāā Storage (persistent state)
āāā Monitor (observability)
Agent-as-Function
Agno treats agents as composable functions:
from agno import Agent
# Define agent
agent = Agent(
name="research-agent",
model="gpt-4o",
description="Research assistant that searches the web",
instructions=[
"Search the web for relevant information",
"Cite your sources",
"Provide comprehensive answers"
],
tools=[web_search, fetch_page],
show_tool_calls=True,
markdown=True
)
# Call like a function
result = agent.run("What are the latest developments in quantum computing?")
print(result.content)
Knowledge Base Integration
from agno.knowledge import KnowledgeBase
from agno.vectordb import QdrantDb
# Create knowledge base
knowledge_base = KnowledgeBase(
vector_db=QdrantDb(
collection="documents",
url="localhost:6333"
),
num_documents=5
)
# Add documents
knowledge_base.load_directory("./documents/", formats=[".pdf", ".txt", ".md"])
# Create agent with knowledge base
agent = Agent(
knowledge_base=knowledge_base,
search_knowledge=True,
add_references=True
)
Tool Registry
from agno.tools import tool
@tool(show_result=True)
def get_stock_price(symbol: str) -> dict:
"""Get current stock price for a symbol."""
return {"symbol": symbol, "price": 150.25, "currency": "USD"}
@tool
def send_email(to: str, subject: str, body: str) -> bool:
"""Send an email."""
email_client.send(to, subject, body)
return True
# Tools are automatically registered and available to the agent
agent = Agent(tools=[get_stock_price, send_email])
Pydantic AIPydantic AI is a framework for building typed, production-ready AI agents with validation.
Core Architecture
Pydantic AI Agent
āāā Model (OpenAI, Anthropic, Gemini, etc.)
āāā Tools (typed function tools)
āāā Result Types (Pydantic models)
āāā Dependency Injection
āāā Run Context
āāā OpenTelemetry Tracing
āāā Logfire Monitoring
Typed Agents
Agents are strongly typed using Pydantic models:
from pydantic_ai import Agent
from pydantic import BaseModel
# Define output type
class CityLocation(BaseModel):
city: str
country: str
latitude: float
longitude: float
# Agent with typed output
agent = Agent(
'openai:gpt-4o',
result_type=CityLocation,
system_prompt="Extract location information from queries."
)
# Run returns typed result
result = agent.run_sync("What is the capital of France?")
print(result.data) # CityLocation(city='Paris', country='France', ...)
print(result.data.latitude) # 48.8566
Dependency Injection
Pydantic AI supports dependency injection for clean code:
from dataclasses import dataclass
from pydantic_ai import Agent, RunContext
@dataclass
class Database:
connection: str
async def query(self, sql: str) -> list:
return [{"name": "Alice"}, {"name": "Bob"}]
agent = Agent(
'openai:gpt-4o',
deps_type=Database,
system_prompt="Query the database to answer questions."
)
@agent.tool
async def get_users(ctx: RunContext[Database]) -> str:
"""Get all users from the database."""
users = await ctx.deps.query("SELECT * FROM users")
return f"Found {len(users)} users"
# Inject dependencies at runtime
db = Database(connection="postgresql://...")
result = agent.run_sync("How many users do we have?", deps=db)
Result Validation
from pydantic import BaseModel, Field, validator
class SearchResult(BaseModel):
title: str
url: str = Field(..., pattern=r'^https?://')
relevance_score: float = Field(..., ge=0, le=1)
@validator('relevance_score')
def score_must_be_valid(cls, v):
if v < 0 or v > 1:
raise ValueError('Score must be between 0 and 1')
return v
agent = Agent(
'openai:gpt-4o',
result_type=list[SearchResult],
system_prompt="Search and return structured results."
)
# Results are automatically validated
result = agent.run_sync("Search for Pydantic AI framework")
for item in result.data:
print(f"{item.title}: {item.relevance_score}")
OpenTelemetry Tracing
from pydantic_ai import Agent
from opentelemetry import trace
# Automatic OpenTelemetry instrumentation
agent = Agent(
'openai:gpt-4o',
instrument=True # Enable tracing
)
# Custom spans
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("custom_span"):
result = agent.run_sync("Some task")
Logfire Monitoring
from pydantic_ai import Agent
import logfire
# Pydantic Logfire integration
logfire.configure()
agent = Agent(
'openai:gpt-4o',
name="my-agent",
instrument=True
)
# All agent runs are automatically logged to Logfire
result = agent.run_sync("Monitor this run")
š MCP (Model Context Protocol) and ACP (Agent Communication Protocol) ā 1122 linesACPAgentDPODeploymentGPTHermes AgentLLMLoRAMCPMulti-Agent OverviewThe emergence of Large Language Models (LLMs) as general-purpose reasoning engines has created a new challenge: how do these models interact with the outside world? Two protocols have emerged to address this question from different angles:
- Model Context Protocol (MCP) ā An open protocol developed by Anthropic that standardizes how LLMs connect to external tools, data sources, and services.
- Agent Communication Protocol (ACP) ā An emerging specification that defines how autonomous AI agents discover each other and communicate in a multi-agent ecosystem.
While both protocols deal with AI system interoperability, they serve fundamentally different purposes and operate at different layers of the stack. This document provides a comprehensive examination of each protocol, their architectures, use cases, and relationship to one another.
Part I: Model Context Protocol (MCP)What is MCP?
The Model Context Protocol (MCP) is an open-standard, client-server protocol developed by Anthropic that provides a standardized interface for LLMs to interact with external systems ā databases, file systems, APIs, search engines, code repositories, and any other tool or data source. It was first announced in November 2024 and released as an open specification and reference implementation.
MCP solves a fundamental problem in the LLM ecosystem: every integration between an LLM and an external system has historically been a bespoke, one-off implementation. Before MCP, connecting an LLM to a GitHub repository required custom code ā and connecting the same LLM to a database required entirely different custom code. This created a fragmented landscape where each integration was brittle, non-portable, and costly to maintain.
MCP addresses this by introducing a universal protocol layer analogous to how USB standardized peripheral connections for computers. Just as USB allows any compliant device (keyboard, mouse, storage) to connect to any compliant host (laptop, desktop, tablet) through a standard interface, MCP allows any LLM application (the "host") to connect to any tool or data source (the "server") through a standard protocol.
Key characteristics of MCP:
- Open standard: The specification, SDKs, and reference implementations are publicly available under permissive licenses.
- Protocol-agnostic transport: Supports both local (stdio) and remote (SSE) communication channels.
- Provider-neutral: Designed to work with any LLM provider, not just Anthropic's Claude.
- Bidirectional: Servers can both respond to requests and push updates to clients.
- Extensible: The primitive set (tools, resources, prompts) can be extended as the ecosystem evolves.
Historical Context and Motivation
MCP emerged from a specific pain point at Anthropic. As Claude gained capabilities and users began integrating it into their workflows, Anthropic observed that each integration required a custom adapter layer. This was unsustainable for three reasons:
- Duplication of effort: Every MCP-like capability (database access, file operations, API calls) was being reimplemented by every integrator.
- Fragmentation: No standard way for tools to declare their capabilities or for models to discover what's available.
- Context window pressure: Without a structured protocol, tool descriptions and invocation patterns consumed precious context tokens with inconsistent formatting.
The protocol was designed to be universal ā not tied to any specific LLM or provider. While Anthropic developed it, the specification is vendor-neutral and can be implemented by any model provider. This is a deliberate architectural choice: the protocol sits between the application layer and the tool layer, not inside any particular model.
Architecture: Host, Client, Server
MCP uses a three-tier architecture with clearly defined roles:
+----------------------------------+
| HOST APPLICATION |
| (Claude Desktop, IDE, CLI, |
| custom application) |
| |
| +----------------------------+ |
| | MCP CLIENT | |
| | (embedded in host) | |
| +----------+----------------+ |
+-------------|-------------------+
| MCP Protocol
| (JSON-RPC over stdio/SSE)
|
+-------------|-------------------+
| +----------+----------------+ |
| | MCP SERVER | |
| | (adapter for a tool or | |
| | data source) | |
| +----------------------------+ |
| |
| +----------------------------+ |
| | EXTERNAL SYSTEM | |
| | (database, API, files, | |
| | search engine, etc.) | |
| +----------------------------+ |
+----------------------------------+
Host
The host is the application or environment where the LLM operates and the user interacts. It is the top-level orchestrator that:
- Manages the lifecycle of MCP client connections
- Handles user authentication and authorization
- Controls which servers are available to the LLM
- Processes the LLM's decisions about which tools to invoke
- Presents results back to the user
Examples of hosts include:
- Claude Desktop ā Anthropic's desktop application
- IDE plugins ā VS Code extension, JetBrains plugin, etc.
- CLI tools ā Command-line interfaces that wrap LLM interactions
- Custom applications ā Any application that embeds an LLM
The host is responsible for security policy enforcement. It decides which MCP servers a given LLM session can access, under what conditions, and with what permissions.
Client
The client is a protocol-level component embedded within the host application. It establishes and manages a 1:1 connection with an MCP server. The client:
- Initiates the connection using the configured transport
- Negotiates protocol version and capabilities during initialization
- Sends requests on behalf of the LLM (tool calls, resource reads, prompt retrievals)
- Receives responses and optional notifications from the server
- Manages connection state, reconnection, and error handling
Important architectural note: The MCP client is not the LLM itself. The client is a protocol adapter within the host application that translates between the LLM's decisions and the MCP protocol messages. The LLM operates at a higher level ā it decides which tool to call based on descriptions provided through the protocol, but the actual protocol negotiation and message formatting is handled by the client.
A single host may run multiple clients simultaneously, each connected to a different server. For example, Claude Desktop might maintain concurrent connections to a filesystem server, a database server, and a GitHub server.
Server
The server is a lightweight adapter that exposes a specific tool, data source, or service through the MCP protocol. Each server:
- Implements the MCP protocol specification
- Registers its capabilities (tools, resources, prompts) during initialization
- Handles incoming requests from the client
- Communicates with the underlying external system
- Can send proactive notifications to the client
Servers are designed to be single-purpose and independently deployable. A typical MCP server wraps one underlying service ā for example:
mcp-server-filesystem ā Exposes local file system operationsmcp-server-postgres ā Exposes PostgreSQL database queriesmcp-server-github ā Exposes GitHub API operationsmcp-server-slack ā Exposes Slack messaging operationsmcp-server-puppeteer ā Exposes browser automation
Servers can run locally (started as subprocesses by the host) or remotely (accessed over the network). The server is stateless with respect to protocol state ā each request contains enough information to be processed independently.
Initialization Sequence
The MCP connection lifecycle follows a structured handshake:
- Client sends
initialize request ā includes protocol version, client capabilities, and client metadata
- Server responds with
initialized ā includes protocol version, server capabilities (list of supported features), and server metadata
- Client sends
initialized notification ā acknowledges the server's response
- Normal operation begins ā tools, resources, and prompts can be listed and used
- Termination ā either side can close the connection gracefully
During initialization, both sides negotiate the protocol version (using semantic versioning) and declare which capabilities they support. This allows the protocol to evolve while maintaining backward compatibility.
Transport Layer
MCP defines two primary transport mechanisms, plus extensibility for custom transports.
1. Stdio Transport (Standard Input/Output)
The stdio transport is used for local, subprocess-based communication between a host and a server running on the same machine.
How it works:
- The host launches the MCP server as a child process
- The host writes JSON-RPC messages to the server's stdin
- The server writes JSON-RPC responses to its stdout
- The server writes log/debug information to stderr (which the host can capture separately)
- The host monitors the child process for termination
Characteristics:
- Low latency: No network overhead; communication happens via OS pipes
- High security: The server runs as a local process with the host's permissions; no network exposure
- Simple deployment: No network configuration, authentication, or TLS setup
- Process isolation: Each server runs in its own process; crashes are contained
- Single connection: One client connects to one server per process
Use cases:
- Desktop applications connecting to local tools (file system, local database, local services)
- Development tools (IDE plugins connecting to code analyzers, linters, local git)
- CLI-based LLM workflows where all tools run on the local machine
Example configuration (from Claude Desktop's config.json):
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/path/to/allowed/directory"
]
},
"github": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-github"
],
"env": {
"GITHUB_TOKEN": "<your-token>"
}
}
}
}
2. SSE Transport (Server-Sent Events)
The SSE transport enables remote communication between a client and a server over HTTP. It uses a combination of:
- SSE (Server-Sent Events) ā for the server-to-client direction (notifications, responses)
- HTTP POST ā for the client-to-server direction (requests)
How it works:
- The client connects to the server's SSE endpoint
- The server sends an endpoint event containing the URL for client-to-server requests
- The client sends JSON-RPC requests via HTTP POST to that endpoint
- The server streams responses and notifications back through the SSE connection
Characteristics:
- Remote access: Servers can be anywhere on the network
- Persistent connection: The SSE channel stays open for the session duration
- Standard web infrastructure: Works through proxies, load balancers, and CDNs
- HTTP-based security: Can leverage TLS, API keys, OAuth, and other web security mechanisms
- Multiple clients: A remote server can serve multiple clients simultaneously
Use cases:
- Connecting to cloud-based services (SaaS tools, cloud databases, web APIs)
- Multi-user or shared server deployments
- Scenarios where the tool runs in a different environment than the LLM
3. Custom Transports
The protocol specification allows for custom transports beyond stdio and SSE. The SDK provides interfaces for implementing alternative transports, which could include:
- WebSocket-based transport
- gRPC-based transport (for high-performance scenarios)
- Message queue transports (RabbitMQ, Kafka for enterprise deployments)
- Unix domain sockets (for local inter-process communication with better lifecycle management than pipes)
Message Format
All MCP messages use JSON-RPC 2.0 as the wire format. This is a lightweight, well-established RPC protocol that uses JSON for data serialization. The protocol defines three types of messages:
- Requests ā Client sends to server, expects a response (e.g.,
tools/call, resources/read)
- Responses ā Server replies to a request (success with result, or error)
- Notifications ā Either side sends without expecting a response (e.g.,
notifications/initialized, notifications/cancelled)
Example JSON-RPC message (tool call request):
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "read_file",
"arguments": {
"path": "/path/to/document.txt"
}
}
}
Example JSON-RPC response:
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "Contents of the file..."
}
],
"isError": false
}
}
Protocol Primitives
MCP defines three core primitives that servers can expose: Tools, Resources Part II: Agent Communication Protocol (ACP)What is ACP?
The Agent Communication Protocol (ACP) is an emerging open specification that defines how autonomous AI agents discover each other, exchange messages, coordinate actions, and share information in a multi-agent environment. While MCP standardizes LLM-to-tool communication, ACP standardizes agent-to-agent communication.
ACP addresses a different but equally important problem: as AI agents become capable of performing complex, multi-step tasks, there is an increasing need for agents to work together. A single agent might be good at code generation but not at deployment; another might excel at data analysis but not at visualization. ACP provides the communication substrate that enables these agents to collaborate.
The protocol is being developed as a community effort with contributions from multiple organizations and researchers working on multi-agent systems. It is not owned by any single company, unlike MCP's origin at Anthropic.
Purpose and Goals
ACP is designed to address several fundamental challenges in multi-agent systems:
Primary Goals:
-
Agent Discovery ā Allow agents to find each other based on capabilities, without requiring a central registry or manual configuration. Agents should be able to join an ecosystem and announce "I can do X, Y, Z."
-
Standardized Communication ā Define a common message format and protocol for agent-to-agent interaction, regardless of what underlying LLM or technology each agent uses internally.
-
Task Delegation ā Enable one agent to request another agent's services. "Agent A, who is good at code generation, can ask Agent B, who is good at security auditing, to review the generated code."
-
State Sharing ā Allow agents to share context, intermediate results, and state information so that collaboration is coherent and agents don't need to rediscover information.
-
Error Handling ā Define how agents communicate failures, timeouts, and unexpected conditions to each other.
-
Interoperability ā Ensure that agents built with different frameworks, LLMs, or architectures can communicate productively.
Design Principles:
- Decentralized: No central authority or registry required; agents can form ad-hoc networks
- Extensible: The message types and interaction patterns can grow with the ecosystem
- Language-agnostic: Works with any programming language or AI framework
- Security-aware: Includes mechanisms for authentication, authorization, and trust
- Graceful degradation: Communication failures shouldn't crash the system
How Agents Discover and Communicate
ACP defines a multi-layered approach to agent interaction:
Agent Discovery
ACP supports several discovery mechanisms, ranging from simple to sophisticated:
1. Static Configuration (Direct Connection):
Agents are configured with the addresses of other agents they should communicate with. This is the simplest approach, suitable for small, controlled deployments.
Agent A -> configured to know about Agent B at agent-b.example.com:8080
Agent B -> configured to know about Agent A at agent-a.example.com:8080
2. Capability Registry (Centralized Discovery):
A discovery service maintains a registry where agents register their capabilities and connection endpoints. Other agents query the registry to find agents with specific skills.
Agent A registers: {"capabilities": ["code_generation", "python"], "endpoint": "..."}
Agent B queries: "Find agents with code_review capability"
Registry returns: [Agent C (code_review), Agent D (security_review)]
3. Peer-to-Peer Discovery (Decentralized):
Agents use gossip protocols or distributed hash tables to discover each other without a central point of failure. This is more complex but more resilient and scalable.
4. DNS-Based Discovery:
Using DNS SRV records or well-known endpoints to discover agent services, similar to how service mesh architectures work.
Communication Patterns
ACP defines several fundamental communication patterns:
1. Request-Response (Synchronous):
An agent sends a request and waits for a response. Used for queries, calculations, or any interaction that expects an immediate answer.
Agent A -> Agent B: "Analyze this code for security vulnerabilities"
Agent B -> Agent A: "Found 2 vulnerabilities: SQL injection at line 42, XSS at line 78"
2. Task Delegation (Asynchronous):
An agent delegates a task and receives a task ID. The delegating agent can check back later or receive a callback/notification when the task completes.
Agent A -> Agent B: {"action": "delegate", "task": "audit_code", "payload": {...}, "callback": "agent-a/callback"}
Agent B -> Agent A: {"task_id": "task-123", "status": "accepted"}
... later ...
Agent B -> Agent A: {"task_id": "task-123", "status": "completed", "result": {...}}
3. Broadcast/Multicast:
An agent sends a message to multiple agents simultaneously. Used for announcements, questions ("Does anyone know X?"), or coordination signals.
Agent A -> All: "Starting deployment of version 2.1.0, pausing all dependent operations"
Agent B, Agent C, Agent D -> Agent A: "Acknowledged"
4. Publish-Subscribe:
Agents subscribe to specific topics or event types and receive notifications when relevant events occur. This is useful for ongoing monitoring and alerting scenarios.
Agent A subscribes to: "security_events", "deployment_status"
Agent B publishes: {"topic": "security_events", "severity": "high", "description": "..."}
Agent A receives the notification and takes action
5. Negotiation Protocol:
Agents engage in multi-turn conversations to agree on plans, resolve conflicts, or coordinate complex workflows.
Agent A: "I propose we split the task: I'll generate the code, you test it, she deploys it."
Agent B: "I can test, but I need a test plan first."
Agent C: "I'll handle deployment, but only after tests pass with 90%+ coverage."
Agent A: "Agreed. Let's proceed with this plan."
Message Format
ACP uses a structured message format (typically JSON) with common fields:
{
"protocol": "acp/1.0",
"message_id": "msg-abc-123",
"sender_id": "agent-code-gen-01",
"receiver_id": "agent-security-audit-02",
"timestamp": "2026-05-31T10:30:00Z",
"message_type": "request",
"interaction_id": "interaction-456",
"payload": {
"action": "security_audit",
"parameters": {
"code": "def login(username, password): ...",
"language": "python",
"depth": "thorough"
}
},
"metadata": {
"ttl": 300,
"priority": "normal",
"signature": "base64signature..."
}
}
Key fields:
- protocol: Version of ACP being used
- message_id: Unique identifier for this message (for deduplication)
- sender_id/receiver_id: Agent identities
- interaction_id: Correlates messages in a multi-turn interaction
- message_type: request, response, notification, broadcast, etc.
- payload: The actual content, structured by the interaction type
- metadata: Routing, security, and lifecycle information
Relationship with MCP
ACP and MCP are complementary protocols that operate at different layers of an AI system architecture. They are not competitors ā they solve different problems.
Architectural Layering
+----------------------------------------------------+
| USER INTERFACE |
+----------------------------------------------------+
| AGENT COMMUNICATION LAYER |
| (ACP Layer) |
| Multi-agent coordination, task delegation, |
| knowledge sharing between agents |
+----------------------------------------------------+
| TOOL INTEGRATION LAYER |
| (MCP Layer) |
| Tool discovery, resource access, prompt templates |
+----------------------------------------------------+
| AI MODELS |
| (LLMs, vision models, specialized models) |
+----------------------------------------------------+
| INFRASTRUCTURE |
| (Compute, storage, networking) |
+----------------------------------------------------+
In this architecture:
- MCP handles the vertical integration: how an AI agent talks to databases, APIs, files, and other external tools.
- ACP handles the horizontal integration: how multiple AI agents talk to each other.
How They Can Work Together
A realistic multi-agent system might use both protocols:
- A coordinator agent
Part III: Comparative AnalysisWhen to Use Which Protocol
| Scenario |
Use MCP |
Use ACP |
Use Both |
| Single agent needs database access |
Yes |
No |
No |
| Single agent needs to read/write files |
Yes |
No |
No |
| Multiple agents collaborating on a task |
No |
Yes |
Yes |
| Agent needs to query an external API |
Yes |
No |
No |
| Agent network with specialized roles |
No |
Yes |
Yes |
| Building a coding assistant |
Yes |
No |
No |
| Building a multi-agent coding team |
Yes |
Yes |
Yes |
| Connecting LLM to a CRM system |
Yes |
No |
No |
| Coordinating across 10+ agents |
No |
Yes |
Yes |
| Rapid prototyping of tool access |
Yes |
No |
No |
Maturity and Risk Assessment
MCP:
- Maturity: High (production-ready SDK, clear specification, growing ecosystem)
- Risk: Low (backed by Anthropic, open source, active community)
- Investment readiness: Ready for production use today
- Learning curve: Moderate (understand JSON-RPC, the three primitives, and transport options)
ACP:
- Maturity: Low (early specification, limited implementations)
- Risk: High (no standard implementation, potential for fragmentation)
- Investment readiness: Experimental only; expect breaking changes
- Learning curve: Steep (multiple design dimensions still being explored)
Ecosystem Comparison
| Metric |
MCP |
ACP |
| Official servers |
~20+ reference implementations |
None |
| Community servers |
200+ (and growing rapidly) |
A few experimental |
| SDK languages |
TypeScript, Python (official) |
Python, TypeScript (experimental) |
| Integration platforms |
Claude Desktop, VS Code, CLI |
None mainstream |
| Documentation |
Comprehensive, with tutorials |
Limited, academic |
| Conference presence |
Multiple talks at AI/developer conferences |
Research workshops |
Part IV: Future DirectionsMCP Evolution
-
Broader Model Support: While MCP works with any LLM via function calling, deeper integration with more model providers' native tool formats will reduce friction.
-
Enhanced Security Model: Expect more sophisticated authorization frameworks, including role-based access control (RBAC) for MCP servers and fine-grained permission scopes.
-
Streaming Responses: Ability for tools to stream partial results back to the LLM (useful for long-running operations, progress updates, and real-time data).
-
Transactional Operations: Support for multi-step atomic operations across multiple servers (e.g., "read from DB, write to file, post to Slack" as a single transaction).
-
Caching and Optimization: Standardized caching mechanisms for resources and tool results to reduce latency and save API calls.
-
Multi-Server Coordination: Better primitives for MCP servers to coordinate with each other, potentially via the host as an intermediary.
-
Mobile and Edge Support: Adaptations of the stdio transport for mobile environments and the SSE transport for edge computing scenarios.
ACP Evolution
-
Standardization: The most critical path for ACP is convergence around a single specification. This may happen organically through community consensus or through adoption by a major AI platform.
-
Reference Implementation: A robust, well-documented reference implementation in one or two languages is essential for adoption.
-
Integration with MCP: As both protocols mature, expect tighter integration ā perhaps ACP messages carrying MCP-style tool descriptions, or MCP servers being wrapped as ACP-capable agents.
-
Identity and Trust Frameworks: Development of practical, cryptographically sound approaches to agent identity and trust, potentially leveraging existing standards like DID (Decentralized Identifiers) and Verifiable Credentials.
-
Capability Ontologies: Standardized vocabularies for describing agent capabilities, making discovery more meaningful and automated.
-
Testing and Verification Tools: Frameworks for testing multi-agent interactions, simulating failures, and verifying protocol compliance.
-
Enterprise Features: Audit logging, compliance tracking, SLA management, and billing integration for agent-to-agent interactions.
Convergence Possibilities
The long-term trajectory suggests a convergence where:
-
MCP becomes the standard tool-access layer for virtually all LLM applications, analogous to how HTTP/SSL became the standard for web communication.
-
ACP builds on MCP's patterns but extends them to the multi-agent domain, possibly reusing MCP's primitive definitions for agent capability descriptions.
-
A unified ecosystem emerges where tools are discoverable via MCP, agents are discoverable via ACP, and they share common patterns for security, identity, and message serialization.
-
Protocol gateways may emerge that translate between MCP and ACP, allowing single tools to be accessed by both individual LLMs (via MCP) and agent collectives (via ACP).
Part V: Implementation NotesGetting Started with MCP
Minimum setup to use MCP (as a user):
1. Install Claude Desktop (or another MCP-compatible host)
2. Configure MCP servers in the host's configuration file
3. Start interacting with tools through natural language
Minimum setup to create an MCP server:
1. Choose an SDK (TypeScript or Python)
2. Import the MCP SDK and create a server instance
3. Define tools using the tool() decorator or addTool() method
4. Implement the handler functions that execute the tool operations
5. Run the server with stdio or SSE transport
Example (Python, minimal server):
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
app = Server("my-server")
@app.list_tools()
async def list_tools():
return [
{
"name": "greet",
"description": "Greet a person by name",
"inputSchema": {
"type": "object",
"properties": {
"name": {"type": "string", "description": "Name to greet"}
},
"required": ["name"]
}
}
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "greet":
return {"content": [{"type": "text", "text": f"Hello, {arguments['name']}!"}]}
raise ValueError(f"Unknown tool: {name}")
if __name__ == "__main__":
from mcp.server.stdio import stdio_server
import anyio
anyio.run(stdio_server(app))
Getting Started with ACP
Since ACP is still in an experimental phase, the best approach is:
- Follow the community discussions: Watch for developments on GitHub and AI research forums
- Experiment with early implementations: Try proof-of-concept agent coordination systems
- Contribute to the specification: The protocol is still being shaped; community input is valuable
- Build on MCP first: For practical needs today, build MCP servers for your tools; the multi-agent layer can be added later as ACP matures
|