Lesson 16.24: Multi-Agent Collaboration & Orchestration

01 What Truly Constitutes a Multi-Agent System?

In previous lessons, we examined how a single autonomous AI agent operates. An individual agent typically contains an LLM, targeted instructions, tools, working state, persistent memory, planning logic, and output validation. However, as enterprise applications grow in scope, forcing a single agent to handle all disparate responsibilities inflates the prompt and fractures focus.

"Find our company's Leave Policy, analyze the specific rules for probationary hires, detect any conflicts with Finance policies, and produce an audited report complete with source citations."

Notice that this single user request demands at least 5 distinct specialized capabilities:

Dumping 20 tools and 5 complex roles into a single prompt degrades reasoning and spikes hallucination rates. Instead, we decompose the workload across specialized agents:

02 Single-Agent vs. Multi-Agent: When Should You Upgrade?

One of the most expensive engineering anti-patterns is premature multi-agent complexity. Multi-agent architectures introduce substantial token latency, network failure domains, and concurrency overhead.

OpenAI's official Agent Orchestration Guide emphasizes: "Always begin with a single well-structured agent. Add specialist agents only when doing so yields concrete gains in prompt clarity, capability isolation, security, or traceability."

03 Four Core Architectural Multi-Agent Patterns

Production enterprise systems rely primarily on four orchestration topologies. These patterns are complementary and can be composed within the same application:

04 Manager vs. Handoff — Which Pattern Should You Choose?

Selecting between Manager and Handoff patterns is the pivotal design choice in multi-agent routing. The table below provides a rigorous comparison:

05 Agent Orchestration Control Flow: LLM vs. Code

Production architectures implement three tiers of control flow across agent components:

06 Agent Contracts — How to Structure Specialist Tasks

Assigning vague tasks—such as "Review this policy and tell me something"—invites hallucination and unformatted outputs. Every specialist requires a rigid architectural contract:

07 Shared Context vs. Independent Context

Should every specialist receive the full conversational transcript? In production, doing so leaks tenant data and wastes token capacity.

08 Output Schema & Validation: Structure vs. Truth

If a specialist reports "Confidence: 99%" while providing an empty evidence array, accepting it is disastrous. Pydantic v2 enforces structural contracts:

from pydantic import BaseModel, Field

class EvidenceReference(BaseModel):
    document_id: str
    passage_id: str
    excerpt: str = Field(description="Exact snippet supporting the claim")

class SpecialistReport(BaseModel):
    summary: str
    evidence: list[EvidenceReference] = Field(default_factory=list)
    limitations: list[str] = Field(default_factory=list)
    confidence_score: float = Field(ge=0.0, le=1.0)

Caution: Schema validation guarantees structural compliance, not factual veracity. The backend must independently query the database to verify that document_id exists and is accessible to the authenticated tenant.

09–13 Practical Project: Byteion Multi-Agent Knowledge Assistant

We construct a complete multi-agent system using the OpenAI Agents SDK, consisting of a Manager Agent, Research Specialist, Policy Specialist, and Citation Reviewer:

byteion-multi-agent/
├── app/
│   ├── __init__.py
│   ├── knowledge_tools.py   # Step 2: Search Tool
│   ├── agents.py            # Step 3 & 4: Specialists & Manager
│   └── main.py              # Step 5: Execution Runner
├── requirements.txt
└── .env

Step 2: Knowledge Search Tool (app/knowledge_tools.py)

# app/knowledge_tools.py
from agents import function_tool

# DEMO FIXTURES ONLY (In production, replace with vector DB / Hybrid retrieval)
@function_tool
def search_internal_knowledge(query: str) -> str:
    """Search sample internal knowledge base records with source references."""
    query_lower = query.lower()
    if "probation" in query_lower or "leave" in query_lower:
        return """
DEMO SOURCE: HR-POLICY-014
PASSAGE: chunk_104
Text: "Probationary employees are entitled to 1.25 days of earned leave per month completed."
"""
    if "citation" in query_lower or "source" in query_lower:
        return """
DEMO SOURCE: KNOWLEDGE-GUIDE-002
PASSAGE: chunk_022
Guidance: "Every policy answer must cite authorized internal source IDs."
"""
    return "No matching sample record found in knowledge base."

Step 3 & 4: Specialists & Manager Agent (app/agents.py)

# app/agents.py
from agents import Agent
from app.knowledge_tools import search_internal_knowledge

# Specialist 1: Document Researcher
research_agent = Agent(
    name="Knowledge Research Specialist",
    instructions=(
        "You are a document research specialist. Search the available internal knowledge tool. "
        "Return concise summaries and identify exact document and passage references. "
        "Do not invent documents. If no evidence is found, say so explicitly."
    ),
    tools=[search_internal_knowledge],
)

# Specialist 2: Policy Analyst
policy_agent = Agent(
    name="Policy Analysis Specialist",
    instructions=(
        "You analyze corporate policies from evidence. Separate explicit rules from assumptions. "
        "Do not invent policy rules not present in source text. State limitations clearly."
    ),
    tools=[search_internal_knowledge],
)

# Specialist 3: Citation Reviewer
citation_agent = Agent(
    name="Citation Review Specialist",
    instructions=(
        "You review draft claims against source citations. Flag missing references and unsupported assertions. "
        "Verify citation integrity before final publication."
    ),
)

# Manager Agent: Coordinates specialists using Agent.as_tool()
manager_agent = Agent(
    name="Byteion AI Knowledge Manager",
    instructions=(
        "You are the coordinator for the Byteion enterprise knowledge assistant.
"
        "1. Understand the incoming request.
"
        "2. Delegate document retrieval to the research specialist.
"
        "3. Delegate policy interpretation to the policy specialist.
"
        "4. Review claims using the citation reviewer.
"
        "5. Synthesize a unified, transparent, cited final answer."
    ),
    tools=[
        research_agent.as_tool(
            tool_name="research_internal_knowledge",
            tool_description="Research relevant internal documents and return evidence with sources."
        ),
        policy_agent.as_tool(
            tool_name="analyze_policy_evidence",
            tool_description="Analyze company policy evidence and identify supported rules."
        ),
        citation_agent.as_tool(
            tool_name="review_citations",
            tool_description="Review draft conclusions for unsupported assertions and missing citations."
        ),
    ],
)

Step 5: Runner Execution (app/main.py)

# app/main.py
import asyncio
from dotenv import load_dotenv
from agents import Runner
from app.agents import manager_agent

load_dotenv()

async def main() -> None:
    question = (
        "What is the leave entitlement for probationary employees according to our company handbook? "
        "Cite the document ID and passage."
    )
    result = await Runner.run(manager_agent, question)
    print("
=== FINAL SYNTHESIZED ANSWER ===")
    print(result.final_output)
    print(f"
Last Active Agent: {result.last_agent.name}")

if __name__ == "__main__":
    asyncio.run(main())

14 Handoff Pattern Implementation & Security Boundaries

Routing user support inquiries directly from a triage router to a dedicated billing or technical specialist:

# app/handoff_demo.py
import asyncio
from agents import Agent, Runner

billing_agent = Agent(
    name="Billing Support Specialist",
    instructions="Handle billing questions and refunds. Always verify invoice access."
)
technical_agent = Agent(
    name="Technical Troubleshooting Specialist",
    instructions="Handle application bugs, API errors, and runtime diagnosis."
)

triage_agent = Agent(
    name="Support Triage Router",
    instructions="Classify user query and hand off to either Billing or Technical support.",
    handoffs=[billing_agent, technical_agent]
)

async def test_handoff():
    res = await Runner.run(triage_agent, "I have an unexplained charge on invoice INV-9021.")
    print("Handed off to:", res.last_agent.name)
    print("Response:", res.final_output)

if __name__ == "__main__":
    asyncio.run(test_handoff())

15 Parallel Agent Execution & asyncio.gather()

Executing three independent specialist evaluations (Technical, Documentation, Support) concurrently in parallel:

# app/parallel_audit.py
import asyncio
from agents import Agent, Runner

tech_agent = Agent(name="Tech Analyst", instructions="Identify latency and infrastructure bottlenecks.")
doc_agent = Agent(name="Doc Analyst", instructions="Identify missing API documentation and specs.")
support_agent = Agent(name="Support Analyst", instructions="Predict common customer confusion points.")
synth_agent = Agent(name="Lead Auditor", instructions="Synthesize findings and flag conflicting points.")

async def run_parallel_audit(proposal: str):
    # Concurrently execute 3 independent specialist runs
    tech_res, doc_res, sup_res = await asyncio.gather(
        Runner.run(tech_agent, proposal),
        Runner.run(doc_agent, proposal),
        Runner.run(support_agent, proposal),
    )
    # Assemble into synthesis prompt
    combined_brief = f"""
Tech Findings: {tech_res.final_output}
Doc Findings: {doc_res.final_output}
Support Findings: {sup_res.final_output}
Synthesize a single executive readiness report.
"""
    return await Runner.run(synth_agent, combined_brief)

Reminder: asyncio.gather() handles in-process concurrency, but it is not a distributed task queue. Long-running tasks require the Checkpoint Store introduced in Lesson 16.23.

16 & 17 Shared State Ownership & Multi-Agent Failure Handling

Never pool all mutable state into a shared bag. Assign strict ownership to individual sub-task outputs:

18 & 19 Agent Loop Bounding & Multi-Agent Security Boundaries

Without strict guardrails, a supervisor can loop indefinitely invoking specialists. Hard production constraints:

20, 21 & 22 Cost Optimization, Observability & Testing Strategy

Prudent model sizing and structured telemetry drastically curb production costs and latency:

23 Common Multi-Agent Pitfalls to Avoid

24 Complete Production Architecture

End-to-end component topology and control flow for the enterprise assistant:

25 Knowledge Check: Multiple-Choice Assessment

Test your technical grasp of multi-agent orchestration, handoff mechanics, and security boundaries across these 10 real-world scenarios: