Three patterns for agent orchestration that survived production

The agent orchestration landscape

Over the past 14 months, we've deployed LLM agent systems across three domains:

1. Customer support automation — 12,000 tickets/month, 8 integrated tools

2. Business intelligence analysis — 200 analysts, 15 data sources

3. Legal document processing — 50k documents/month, 6 extraction pipelines

Each deployment taught us which orchestration patterns work in theory versus which hold up under production stress — tool timeouts, API rate limits, ambiguous queries, and user expectations for sub-3-second responses.

This article catalogs three patterns that survived: Router, Planner-Executor, and Critic.

Pattern 1: Router (Simple Dispatch)

Architecture

                User Query
                    ↓
            ┌───────────────┐
            │  Router LLM   │  "Which tool handles this?"
            └───────┬───────┘
                    │
        ┌───────────┼───────────┬───────────┐
        ↓           ↓           ↓           ↓
    [Tool A]    [Tool B]    [Tool C]    [Tool D]
    Search      Calculator   Weather     Calendar
        ↓           ↓           ↓           ↓
                Response (from single tool)

When to use

• Multiple specialized tools with clear, non-overlapping domains.

• Single-step tasks (one tool call → result).

• Latency-sensitive applications (<1s response time).

Implementation

router_agent.py
from typing import Dict, Callable
class RouterAgent:
    """Simple routing agent — dispatch to one tool."""
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
        self.tool_descriptions = self._generate_tool_docs()
    def _generate_tool_docs(self) -> str:
        """Generate tool documentation for router prompt."""
        docs = []
        for name, tool in self.tools.items():
            docs.append(f"- {name}: {tool.__doc__}")
        return "\n".join(docs)
    async def route(self, query: str) -> str:
        """Route query to appropriate tool."""
        router_prompt = f"""
You are a tool router. Given a user query, select the single best tool to answer it.
Available tools:
{self.tool_descriptions}
User query: {query}
Respond with JSON: {{"tool": "tool_name", "reasoning": "why this tool"}}
"""
        routing_decision = await llm.generate(router_prompt)
        tool_name = json.loads(routing_decision)['tool']
        # Execute selected tool
        if tool_name not in self.tools:
            return f"Error: Unknown tool {tool_name}"
        return await self.toolstool_name
Usage
tools = {
    "search": search_knowledge_base,
    "calculator": calculate_expression,
    "weather": get_weather_forecast,
    "calendar": check_calendar_availability
}
agent = RouterAgent(tools)
response = await agent.route("What's the weather in Paris tomorrow?")

Production data (Customer support)

Metric	Value
Queries handled	12,000/month
Correct routing	94%
Avg latency	820ms
p95 latency	1.2s
Ambiguous routing	6% (fallback to human)

Strengths

✅ Low latency — Single LLM call + one tool execution ✅ Predictable — Linear execution, easy to reason about ✅ Debuggable — Simple to log: "Query → Routing decision → Tool → Result" ✅ Cost-effective — Minimal LLM calls

Weaknesses

❌ No tool chaining — Can't combine tools ("Search for X, then calculate Y") ❌ Routing errors are fatal — Wrong tool selection = wrong answer ❌ Ambiguous queries fail — "Book a meeting if it's not raining" requires two tools

Production lessons

Lesson 1: Build a fallback classifier

When routing confidence is low (<70%), escalate to human:

routing_confidence = routing_decision['confidence']
if routing_confidence < 0.70:
    return escalate_to_human(query, reason="ambiguous routing")

Lesson 2: Cache routing decisions

Common queries ("Check order status") route the same way every time:

@cache(ttl=3600)
def route_query(query: str):
    # Cache routing for 1 hour
    return router.route(query)

Lesson 3: Monitor routing accuracy

Track which tools are selected vs. what users actually needed:

Log routing decisions
log_routing_decision(
    query=query,
    selected_tool=tool_name,
    user_satisfaction=feedback  # Collect post-interaction
)
Weekly analysis
routing_errors = query_logs.filter(user_satisfaction < 3)
print(f"Top misrouted queries: {routing_errors.most_common(10)}")

Result: We improved routing accuracy from 87% → 94% by retraining on misrouted queries.

Pattern 2: Planner-Executor (Multi-Step Reasoning)

Architecture

            User Query: "Compare Q1 vs Q2 revenue"
                    ↓
            ┌───────────────┐
            │  Planner LLM  │  Generate execution plan
            └───────┬───────┘
                    ↓
            Plan: [Step 1, Step 2, Step 3]
            1. Fetch Q1 revenue from DB
            2. Fetch Q2 revenue from DB
            3. Calculate difference
                    ↓
            ┌───────────────┐
            │   Executor    │  Run plan sequentially
            └───────┬───────┘
                    ↓
        ┌───────────┼───────────┐
        ↓           ↓           ↓
    [Query DB]  [Query DB]  [Calculate]
        ↓           ↓           ↓
    $120K       $145K        +$25K (+21%)
                    ↓
            Final Response

When to use

• Multi-step workflows requiring tool composition.

• Dynamic tool selection (can't predict tool sequence upfront).

• Structured tasks (data analysis, report generation).

Implementation

planner_executor_agent.py
from typing import List, Dict
import json
class PlannerExecutorAgent:
    """Agent that plans before executing."""
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
    async def plan(self, query: str) -> List[Dict]:
        """Generate execution plan."""
        planner_prompt = f"""
You are a task planner. Break down this query into executable steps using available tools.
Available tools:
{self._tool_docs()}
User query: {query}
Generate a plan as JSON array:
[
  {{"step": 1, "tool": "tool_name", "input": "...", "output_var": "var1"}},
  {{"step": 2, "tool": "tool_name", "input": "use {{var1}}", "output_var": "var2"}},
  ...
]
"""
        plan_json = await llm.generate(planner_prompt)
        return json.loads(plan_json)
    async def execute(self, plan: List[Dict]) -> Dict:
        """Execute plan step by step."""
        context = {}  # Store intermediate results
        for step in plan:
            tool_name = step['tool']
            tool_input = step['input']
            # Substitute variables from context
            for var, value in context.items():
                tool_input = tool_input.replace(f"{{{var}}}", str(value))
            # Execute tool
            result = await self.toolstool_name
            # Store result in context
            output_var = step.get('output_var', f"step_{step['step']}")
            context[output_var] = result
            print(f"Step {step['step']}: {tool_name}({tool_input}) → {result}")
        return context
    async def run(self, query: str) -> str:
        """Plan and execute."""
        plan = await self.plan(query)
        context = await self.execute(plan)
        # Generate final response using context
        final_prompt = f"""
User query: {query}
Execution results:
{json.dumps(context, indent=2)}
Provide a natural language response to the user.
"""
        return await llm.generate(final_prompt)
Usage
tools = {
    "sql_query": execute_sql,
    "calculator": calculate,
    "search_docs": search_documentation,
    "send_email": send_email
}
agent = PlannerExecutorAgent(tools)
response = await agent.run("Compare Q1 vs Q2 revenue and email summary to CFO")

Production data (Business intelligence)

Metric	Value
Queries handled	1,200/month
Successful completions	89%
Avg latency	3.2s
p95 latency	8.4s
Plan errors	11% (invalid tool, wrong order)

Strengths

✅ Handles complex workflows — Multi-tool composition ✅ Flexible — Adapts to query complexity dynamically ✅ Transparent — Plan is human-readable, debuggable ✅ Recoverable — Can retry individual steps on failure

Weaknesses

❌ Higher latency — N+1 LLM calls (1 for planning, N for execution) ❌ Plans can be wrong — Invalid tool selection, incorrect order, missing steps ❌ Error propagation — Early step failure breaks entire plan ❌ Cost scales with steps — 5-step plan = 6 LLM calls

Production lessons

Lesson 1: Validate plans before execution

Don't blindly trust LLM-generated plans:

def validate_plan(plan: List[Dict]) -> bool:
    """Check plan for common errors."""
    for step in plan:
        # Check tool exists
        if step['tool'] not in self.tools:
            raise PlanError(f"Unknown tool: {step['tool']}")
        # Check variable dependencies
        required_vars = extract_variables(step['input'])
        available_vars = [s['output_var'] for s in plan[:step['step']-1]]
        for var in required_vars:
            if var not in available_vars:
                raise PlanError(f"Variable {var} not available at step {step['step']}")
    return True

Lesson 2: Add step-level retries

Network errors and rate limits happen. Retry individual steps:

from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def execute_step(tool_name: str, tool_input: str):
    return await self.toolstool_name

Lesson 3: Implement plan caching for similar queries

Queries like "Compare Q1 vs Q2 revenue" have similar plans:

Cache plan templates
plan_template = cached_plans.get(query_category)
if plan_template:
    plan = instantiate_template(plan_template, query_params)
else:
    plan = await self.plan(query)

Result: Reduced planning latency by 40% for repeated query patterns.

Pattern 3: Critic (Iterative Refinement)

Architecture

            User Query: "Draft a professional apology email"
                    ↓
            ┌───────────────┐
            │ Generator LLM │  Generate initial response
            └───────┬───────┘
                    ↓
            Draft v1: "Sorry for the issue..."
                    ↓
            ┌───────────────┐
            │  Critic LLM   │  Evaluate quality
            └───────┬───────┘
                    ↓
        [Pass: score ≥ 8/10] ────→ Return response
                    │
        [Fail: score < 8/10]
                    ↓
            Feedback: "Too casual. Add specific details."
                    ↓
            ┌───────────────┐
            │  Generator    │  Regenerate with feedback
            └───────┬───────┘
                    ↓
            Draft v2: "We sincerely apologize for [specific issue]..."
                    ↓
            [Repeat up to max_iterations=3]

When to use

• Quality-critical outputs (legal docs, customer comms).

• Iterative refinement needed.

• Latency tolerance (users expect 3-10s for complex tasks).

Implementation

critic_agent.py
from typing import Tuple
class CriticAgent:
    """Agent with self-critique loop."""
    def __init__(self, max_iterations: int = 3):
        self.max_iterations = max_iterations
    async def generate(self, query: str, feedback: str = None) -> str:
        """Generate response (with optional feedback)."""
        if feedback:
            prompt = f"""
User request: {query}
Previous attempt received this feedback:
{feedback}
Generate an improved response addressing the feedback.
"""
        else:
            prompt = f"User request: {query}\n\nGenerate a response."
        return await llm.generate(prompt)
    async def critique(self, query: str, response: str) -> Tuple[float, str]:
        """Critique response quality (score 0-10, feedback)."""
        critic_prompt = f"""
Evaluate this response for quality, accuracy, and professionalism.
User request: {query}
Response: {response}
Provide:
1. Score (0-10)
2. Specific feedback for improvement
Format: {{"score": X, "feedback": "..."}}
"""
        critique = await llm.generate(critic_prompt)
        result = json.loads(critique)
        return result['score'], result['feedback']
    async def run(self, query: str, min_score: float = 8.0) -> Dict:
        """Generate with iterative refinement."""
        history = []
        for iteration in range(self.max_iterations):
            # Generate response (with feedback from previous iteration)
            feedback = history[-1]['feedback'] if history else None
            response = await self.generate(query, feedback)
            # Critique response
            score, feedback = await self.critique(query, response)
            history.append({
                "iteration": iteration + 1,
                "response": response,
                "score": score,
                "feedback": feedback
            })
            # Check if quality threshold met
            if score >= min_score:
                return {
                    "response": response,
                    "iterations": iteration + 1,
                    "final_score": score,
                    "history": history
                }
        # Max iterations reached, return best attempt
        best = max(history, key=lambda x: x['score'])
        return {
            "response": best['response'],
            "iterations": self.max_iterations,
            "final_score": best['score'],
            "history": history,
            "warning": "Max iterations reached without meeting quality threshold"
        }
Usage
agent = CriticAgent(max_iterations=3)
result = await agent.run("Draft a professional apology for delayed shipment")
print(f"Final response (score: {result['final_score']}):\n{result['response']}")

Production data (Legal document generation)

Metric	Value
Documents generated	800/month
First-pass success	62% (score ≥ 8/10)
Second-pass success	89%
Third-pass success	96%
Avg latency	4.2s
p95 latency	11.8s

Strengths

✅ Higher quality outputs — Self-correction catches errors ✅ Adaptable — Learns from own mistakes within session ✅ Transparent — Critique feedback explains quality issues ✅ Graceful degradation — Returns best attempt if threshold not met

Weaknesses

❌ High latency — 2-6 LLM calls (2x per iteration) ❌ Expensive — Cost scales with iterations ❌ Can loop indefinitely — Must set max_iterations ❌ Critic can be wrong — False negatives (good response scored low)

Production lessons

Lesson 1: Set aggressive max_iterations limit

Our initial limit was 5. Observed 12% of queries hit this limit (wasting 10 LLM calls). Reduced to 3:

Cost analysis
avg_cost_per_llm_call = $0.02
max_iterations = 5 → avg_cost = $0.20 (10 calls)
max_iterations = 3 → avg_cost = $0.12 (6 calls)
40% cost reduction with minimal quality impact

Lesson 2: Use fast models for critique

Critic doesn't need frontier model intelligence. We use GPT-4 for generation, GPT-3.5-turbo for critique:

async def critique(self, query: str, response: str):
    # Use cheaper, faster model for critique
    critique = await llm.generate(critic_prompt, model="gpt-3.5-turbo")
    # ...

Result: Reduced critique latency by 60% (600ms → 240ms) with same accuracy.

Lesson 3: Add early stopping on "perfect" scores

If first attempt scores 9.5/10, skip further iterations:

if score >= 9.5:  # "Perfect" threshold
    return early_with_success(response, score)

Latency comparison: Real production data

Pattern	Avg Latency	p95 Latency	p99 Latency	LLM Calls
Router	820ms	1.2s	1.8s	1
Planner-Executor (3 steps)	3.2s	8.4s	14.1s	4
Critic (avg 1.8 iterations)	4.2s	11.8s	18.5s	3.6

Cost comparison

Assumptions:

• GPT-4 input: $0.01/1K tokens.

• GPT-4 output: $0.03/1K tokens.

• Avg query: 200 input tokens.

• Avg response: 500 output tokens.

Pattern	LLM Calls	Avg Cost
Router	1	$0.017
Planner-Executor	4	$0.068
Critic	3.6	$0.061

Decision matrix: Which pattern to use?

Choose Router if:

✅ Single-tool dispatch is sufficient ✅ Latency <1s is required ✅ Query routing is unambiguous ✅ Cost per query matters

Choose Planner-Executor if:

✅ Multi-step workflows needed ✅ Tool composition required ✅ Latency <5s is acceptable ✅ Transparency (visible plan) is valuable

Choose Critic if:

✅ Output quality is mission-critical ✅ Latency <10s is acceptable ✅ Self-correction adds value ✅ First-draft quality is insufficient

Hybrid patterns we've tested

Pattern 4: Router + Planner-Executor

Route simple queries to single tools, complex queries to planner:

if query_complexity(query) < 0.5:
    return router.route(query)  # Fast path
else:
    return planner_executor.run(query)  # Slow path

Result: 70% of queries take fast path (avg 850ms), 30% take slow path (avg 3.5s). Overall avg: 1.6s.

Pattern 5: Planner-Executor + Critic

Plan, execute, then critique final output:

context = await planner_executor.execute(query)
final_response = await generate_response(context)
score, feedback = await critic.critique(query, final_response)
if score < 8.0:
    final_response = await regenerate_with_feedback(context, feedback)

Result: Used for high-stakes reports. Latency: 8-12s. Quality: 98% user satisfaction.

Conclusion

After 18+ months in production:

1. Router handles 80% of queries with excellent latency

2. Planner-Executor excels for multi-step workflows but requires plan validation

3. Critic improves quality by 15-20% but doubles cost and latency

Our default recommendation:

• Start with Router for MVP.

• Add Planner-Executor when users request multi-step tasks.

• Reserve Critic for quality-critical outputs (legal, financial, medical).

The best pattern depends on your latency budget, quality requirements, and cost constraints. Don't over-engineer — deploy simple first, scale complexity as needed.

---

Building agent systems? Contact us to discuss your architecture. We offer agent design consulting, implementation support, and production optimization services.