Agent Orchestration Patterns — The $4k Mistake We Made With LLM-Controlled Handoffs

from openai import OpenAI
from agents import Agent, Runner, function_tool

client = OpenAI()

# Step 1: Define a simple sequential pipeline
@function_tool()
def extract_keywords(text: str) -> list:
    """Extract keywords from user query."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"Extract keywords from: {text}"}]
    )
    return response.choices[0].message.content.split(",")

keyword_agent = Agent(
    name="keyword_extractor",
    instructions="Extract keywords from the user query. Return as comma-separated.",
    tools=[extract_keywords]
)

search_agent = Agent(
    name="search",
    instructions="Search for products matching the keywords.",
    tools=[function_tool(lambda keywords: f"Searching for {keywords}")]
)

# Sequential orchestration: keyword_agent -> search_agent
# Note: Runner.run() returns a RunResult with the final agent's output
result = Runner.run(keyword_agent, "Find me a red dress")
keywords = result.final_output  # e.g., "red, dress"

result2 = Runner.run(search_agent, keywords)
print(result2.final_output)

import asyncio
from agents import Agent, Runner

# Define three agents in a pipeline
validate_agent = Agent(
    name="validator",
    instructions="Check if the input is valid. Return 'VALID' or 'INVALID'."
)

process_agent = Agent(
    name="processor",
    instructions="Process the validated input and return a summary."
)

report_agent = Agent(
    name="reporter",
    instructions="Generate a final report from the processed data."
)

async def sequential_pipeline(user_input: str) -> str:
    try:
        # Step 1: Validate
        result = await Runner.run(validate_agent, user_input, timeout=10)
        if result.final_output.strip() != "VALID":
            return "Invalid input. Please try again."
        
        # Step 2: Process
        result = await Runner.run(process_agent, user_input, timeout=10)
        processed = result.final_output
        
        # Step 3: Report
        result = await Runner.run(report_agent, processed, timeout=10)
        return result.final_output
    except asyncio.TimeoutError:
        return "Pipeline timed out. Check agent responsiveness."
    except Exception as e:
        return f"Pipeline error: {str(e)}"

# Bad: Multi-agent for a simple classification
# This costs 2 LLM calls instead of 1
classifier_agent = Agent(
    name="classifier",
    instructions="Classify the email as spam or not spam."
)

report_agent = Agent(
    name="reporter",
    instructions="Report the classification."
)

# This is wasteful
result = Runner.run(classifier_agent, email)
classification = result.final_output
result2 = Runner.run(report_agent, classification)

# Good: Single agent with a clear prompt
single_agent = Agent(
    name="spam_classifier",
    instructions="Classify the email. Return 'SPAM' or 'NOT SPAM'."
)
result = Runner.run(single_agent, email)
print(result.final_output)

import asyncio
from agents import Agent, Runner

# Token bucket for rate limiting
class TokenBucket:
    def __init__(self, rate: int, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill = asyncio.get_event_loop().time()
    
    async def acquire(self):
        now = asyncio.get_event_loop().time()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_refill = now
        if self.tokens < 1:
            await asyncio.sleep(0.1)  # Wait for tokens
            return await self.acquire()
        self.tokens -= 1
        return True

bucket = TokenBucket(rate=10, capacity=10)  # 10 calls per second

async def safe_agent_call(agent: Agent, input: str) -> str:
    await bucket.acquire()
    try:
        result = await asyncio.wait_for(
            Runner.run(agent, input),
            timeout=5.0
        )
        return result.final_output
    except asyncio.TimeoutError:
        return "Agent timed out. Please try again."
    except Exception as e:
        return f"Agent error: {str(e)}"

# Mistake 1: No context passed on handoff
agent_a = Agent(name="a", instructions="Extract keywords.")
agent_b = Agent(name="b", instructions="Search for products.")

result = Runner.run(agent_a, "red dress")
# B gets no context about the original query
result2 = Runner.run(agent_b, result.final_output)  # B only sees keywords, not the original intent

# Fix: Pass context explicitly
result2 = Runner.run(agent_b, result.final_output, context={"original_query": "red dress"})

# Mistake 2: No max handoff depth
agent = Agent(name="triage", instructions="Route to the best agent.", handoffs=[agent_b])
# If triage routes to itself, infinite loop

# Fix: Set max_handoff_depth
agent = Agent(name="triage", instructions="Route to the best agent.", handoffs=[agent_b], max_handoff_depth=2)

# Mistake 3: Generic prompt
agent = Agent(name="helper", instructions="Help the user.")
# This agent might try to do everything and do nothing well

# Fix: Specific prompt
agent = Agent(name="billing", instructions="You handle billing questions only. If asked about anything else, say 'I cannot help with that.'")

from agents import Agent, Runner

# Handoff pattern: specialist takes over
specialist = Agent(
    name="billing",
    instructions="Answer billing questions.",
    handoffs=[]  # no further handoffs
)

triage = Agent(
    name="triage",
    instructions="Route billing questions to the billing agent.",
    handoffs=[specialist]
)

# Tool pattern: manager keeps control
summarizer = Agent(
    name="summarizer",
    instructions="Summarize the text."
)

manager = Agent(
    name="manager",
    instructions="You are a manager. Use the summarizer tool to summarize user input.",
    tools=[summarizer.as_tool(
        tool_name="summarize_text",
        tool_description="Summarize a given text."
    )]
)

# Usage
result = Runner.run(manager, "This is a long text to summarize...")
print(result.final_output)  # Manager's final answer, not the summarizer's

import json
import logging
from datetime import datetime
from agents import Runner, Agent

# Structured logging setup
logger = logging.getLogger("agent_orchestration")
handler = logging.StreamHandler()
formatter = logging.Formatter(
    '{"timestamp": "%(asctime)s", "level": "%(levelname)s", "message": %(message)s}'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Custom wrapper to log orchestration events
async def monitored_agent_call(agent: Agent, input: str, context: dict = None):
    start = datetime.now()
    logger.info(json.dumps({
        "event": "agent_call_start",
        "agent": agent.name,
        "input_length": len(input),
        "context_keys": list(context.keys()) if context else None
    }))
    
    try:
        result = await Runner.run(agent, input, context=context)
        elapsed = (datetime.now() - start).total_seconds()
        logger.info(json.dumps({
            "event": "agent_call_end",
            "agent": agent.name,
            "elapsed_seconds": elapsed,
            "output_length": len(result.final_output),
            "handoffs": result.handoffs  # list of handoff targets
        }))
        return result
    except Exception as e:
        elapsed = (datetime.now() - start).total_seconds()
        logger.error(json.dumps({
            "event": "agent_call_error",
            "agent": agent.name,
            "elapsed_seconds": elapsed,
            "error": str(e)
        }))
        raise

// io.thecodeforge
from strands.multiagent.graph import GraphBuilder

# Define nodes with explicit input/output schemas
search_agent = Agent("search_flights", tools=[flight_api])
book_agent = Agent("book_flight", tools=[booking_api], wait_for=[search_agent])
notify_agent = Agent("notify_user", tools=[email_api], wait_for=[book_agent])

# Build and validate DAG at init time
graph = GraphBuilder()\
    .add_node(search_agent)\
    .add_node(book_agent)\
    .add_node(notify_agent)\
    .build()

# Raises GraphValidationError if cycles or missing edges detected
assert graph.is_dag(), "Orchestration must be a DAG"

// io.thecodeforge
from strands.multiagent import ReWOOPipeline
from strands.agents import Toolbelt

planner = Agent(
    "planner",
    tools=None,  # Planner never calls tools
    system_prompt="Output steps as: PLAN: <step>:<tool>(<args>)"
)
worker = Toolbelt([search_api, book_api])  # Pure executor, no LLM
solver = Agent("solver", tools=None)  # Synthesizer

pipeline = ReWOOPipeline(planner, worker, solver)
result = pipeline.run("Book a flight from SFO to LHR on Dec 10")

print(result.token_efficiency)  # 0.4 (40% of equivalent ReAct)