Agent Communication Patterns — How a Missing Timeout Caused a $12k/Hour Token Storm

import asyncio
from dataclasses import dataclass, field
from typing import Optional
from openai import AsyncOpenAI
import json

client = AsyncOpenAI()

@dataclass
class ConversationContext:
    messages: list = field(default_factory=list)
    handoff_count: int = 0
    max_handoffs: int = 3  # Hard limit to prevent loops
    token_budget: int = 32000
    trace_id: str = ""

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        # Track token usage roughly (1 token ~= 4 chars)
        if len(content) > 0:
            self.token_budget -= len(content) // 4
        if self.token_budget <= 0:
            raise RuntimeError(f"Token budget exceeded for trace {self.trace_id}")

    def copy_readonly(self) -> "ConversationContext":
        # Return a deep copy to prevent sub-agent mutation
        return ConversationContext(
            messages=json.loads(json.dumps(self.messages)),
            handoff_count=self.handoff_count,
            max_handoffs=self.max_handoffs,
            token_budget=self.token_budget,
            trace_id=self.trace_id
        )

async def run_agent_with_handoff(context: ConversationContext, agent_prompt: str, timeout: int = 30):
    if context.handoff_count >= context.max_handoffs:
        raise RuntimeError(f"Max handoffs reached ({context.max_handoffs}) for trace {context.trace_id}")
    
    context.handoff_count += 1
    
    try:
        response = await asyncio.wait_for(
            client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "system", "content": agent_prompt}] + context.messages,
                timeout=timeout
            ),
            timeout=timeout + 5  # Add buffer for network
        )
        context.add_message("assistant", response.choices[0].message.content)
        return response.choices[0].message.content
    except asyncio.TimeoutError:
        # Fallback: escalate to human
        context.add_message("system", "[Agent timeout - escalating to human]")
        return "I need to transfer you to a human agent. One moment please."

# Usage example
async def main():
    ctx = ConversationContext(trace_id="trace-123")
    ctx.add_message("user", "My internet is down and my bill is wrong.")
    
    router_prompt = "You are a router. If the user mentions billing, handoff to billing_agent."
    result = await run_agent_with_handoff(ctx, router_prompt)
    print(f"Router response: {result}")
    print(f"Handoff count: {ctx.handoff_count}")

asyncio.run(main())

import json
from openai import OpenAI
from typing import Literal, Optional

client = OpenAI()

Intent = Literal["billing", "technical", "account", "escalation", "unknown"]

class Router:
    def __init__(self, confidence_threshold: float = 0.7):
        self.threshold = confidence_threshold
        # Fast keyword-based routes for common cases
        self.keyword_routes = {
            "password": "technical",
            "login": "technical",
            "charge": "billing",
            "refund": "billing",
            "cancel": "account",
            "update": "account",
        }
    
    def keyword_classify(self, message: str) -> Optional[Intent]:
        message_lower = message.lower()
        for keyword, intent in self.keyword_routes.items():
            if keyword in message_lower:
                return intent
        return None
    
    def llm_classify(self, message: str) -> dict:
        # Returns {"intent": "billing", "confidence": 0.95}
        response = client.chat.completions.create(
            model="gpt-4o-mini",  # Cheaper model for classification
            messages=[
                {"role": "system", "content": "Classify the user's intent. Return JSON with 'intent' (billing/technical/account/escalation/unknown) and 'confidence' (0.0-1.0)."},
                {"role": "user", "content": message}
            ],
            response_format={"type": "json_object"}
        )
        return json.loads(response.choices[0].message.content)
    
    def route(self, message: str) -> tuple[Intent, float]:
        # Stage 1: fast keyword match
        keyword_intent = self.keyword_classify(message)
        if keyword_intent:
            return keyword_intent, 0.85  # Assume high confidence for keywords
        
        # Stage 2: LLM classification
        result = self.llm_classify(message)
        intent = result.get("intent", "unknown")
        confidence = result.get("confidence", 0.0)
        
        if confidence < self.threshold:
            return "escalation", confidence  # Force human handoff
        return intent, confidence

# Usage
router = Router(confidence_threshold=0.7)
intent, conf = router.route("I need to reset my password")
print(f"Intent: {intent}, Confidence: {conf}")
# Output: Intent: technical, Confidence: 0.85

# BAD: Over-engineered multi-agent system for a simple task
# This is what NOT to do

# Instead of this:
class WeatherRouter:
    def route(self, query):
        if "weather" in query:
            return "weather_agent"
        return "unknown"

class WeatherAgent:
    def get_weather(self, city):
        # LLM call to extract city, then API call
        return "It's sunny in Paris."

# Do this:
import requests

def get_weather_simple(city: str) -> str:
    """Single function call. No agents needed."""
    response = requests.get(f"https://api.weather.com/v1/{city}", timeout=5)
    data = response.json()
    return f"It's {data['condition']} in {city}."

# Usage
print(get_weather_simple("Paris"))
# Output: It's sunny in Paris.
# Latency: 1 second vs 8 seconds for the agent version

import asyncio
from typing import Any

async def run_agent(agent_name: str, task: str, timeout: int = 5) -> dict:
    """Simulate an agent that might be slow."""
    try:
        # Simulate variable latency
        delay = 2 if agent_name != "slow_agent" else 10
        await asyncio.sleep(delay)
        return {"agent": agent_name, "result": f"Analysis from {agent_name}", "status": "success"}
    except asyncio.TimeoutError:
        return {"agent": agent_name, "result": None, "status": "timeout"}

async def broadcast_with_partial_results(task: str, agents: list[str], timeout: int = 5) -> list[dict]:
    """Run all agents with a timeout. Return partial results if some fail."""
    tasks = [asyncio.create_task(run_agent(agent, task, timeout)) for agent in agents]
    
    # Wait for all tasks, but with a timeout per task
    done, pending = await asyncio.wait(tasks, timeout=timeout)
    
    results = []
    for task in done:
        try:
            result = task.result()
            results.append(result)
        except Exception as e:
            results.append({"agent": "unknown", "result": None, "status": f"error: {e}"})
    
    # Handle pending tasks (they timed out)
    for task in pending:
        task.cancel()
        # In a real system, you'd log this and retry later
        results.append({"agent": "unknown", "result": None, "status": "timeout"})
    
    return results

async def main():
    agents = ["fraud_check", "compliance_check", "risk_analysis", "slow_agent"]
    results = await broadcast_with_partial_results("Analyze transaction T-12345", agents, timeout=5)
    for r in results:
        print(f"{r['agent']}: {r['status']} -> {r['result']}")

asyncio.run(main())
# Output:
# fraud_check: success -> Analysis from fraud_check
# compliance_check: success -> Analysis from compliance_check
# risk_analysis: success -> Analysis from risk_analysis
# unknown: timeout -> None

# Mistake #1: Generic system prompt for all agents
# BAD:
# system_prompt = "You are a helpful assistant."

# GOOD: Specialized prompts with boundaries
BILLING_PROMPT = """You are a billing support agent. You ONLY handle questions about:
- Charges and invoices
- Payment methods
- Refunds and credits
- Subscription plans

If the user asks about technical issues (e.g., login problems, bugs), say:
"I can only help with billing questions. Let me transfer you to technical support."

Do NOT attempt to answer technical questions."""

TECHNICAL_PROMPT = """You are a technical support agent. You ONLY handle questions about:
- Login and account access
- Software bugs and errors
- Feature requests
- System configuration

If the user asks about billing, say:
"I can only help with technical issues. Let me transfer you to billing support."

Do NOT attempt to answer billing questions."""

# Mistake #2: Not logging handoffs
# BAD:
# def handoff(agent, message):
#     return agent.run(message)

# GOOD:
import logging
logger = logging.getLogger(__name__)

def handoff_with_logging(source_agent: str, target_agent: str, message: str, confidence: float):
    logger.info(f"HANDOFF: {source_agent} -> {target_agent} | confidence={confidence} | msg_preview={message[:100]}")
    return target_agent.run(message)

# Mistake #3: Using same model for routing and execution
# BAD:
# router = OpenAI(model="gpt-4o")
# executor = OpenAI(model="gpt-4o")

# GOOD:
router = OpenAI(model="gpt-4o-mini")  # Cheap for classification
executor = OpenAI(model="gpt-4o")  # Expensive for reasoning

# Router Pattern: Sub-agent takes over
async def router_pattern(user_message: str) -> str:
    intent = await classify_intent(user_message)
    if intent == "billing":
        return await billing_agent.run(user_message)  # Billing agent owns the conversation
    elif intent == "technical":
        return await technical_agent.run(user_message)

# Agent-as-Tool Pattern: Sub-agent returns result, main agent continues
async def agent_as_tool_pattern(user_message: str) -> str:
    main_agent = MainAgent()
    # Main agent decides to call fact-checker as a tool
    fact_check_result = await fact_checker_agent.run("Verify: The Eiffel Tower is in Rome.")
    # Main agent continues with the result
    response = await main_agent.run(f"User said: {user_message}. Fact check result: {fact_check_result}")
    return response

# Broadcast Pattern: Multiple agents run independently, results aggregated
async def broadcast_pattern(transaction_data: dict) -> dict:
    tasks = [
        fraud_agent.run(transaction_data),
        compliance_agent.run(transaction_data),
        risk_agent.run(transaction_data)
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return {
        "fraud": results[0],
        "compliance": results[1],
        "risk": results[2]
    }

import logging
import json
from datetime import datetime
from openai import OpenAI

# Structured logging setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent_communication")

class ObservableAgent:
    def __init__(self, name: str, model: str = "gpt-4o-mini"):
        self.name = name
        self.model = model
        self.client = OpenAI()
    
    async def run(self, messages: list, trace_id: str, parent_agent: str = None):
        start_time = datetime.now()
        
        # Log the handoff
        logger.info(json.dumps({
            "event": "handoff",
            "trace_id": trace_id,
            "source_agent": parent_agent,
            "target_agent": self.name,
            "timestamp": start_time.isoformat(),
            "message_count": len(messages),
            "last_message_preview": messages[-1]["content"][:200] if messages else ""
        }))
        
        # Make the LLM call
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages
        )
        
        end_time = datetime.now()
        latency_ms = (end_time - start_time).total_seconds() * 1000
        
        # Log the LLM call
        logger.info(json.dumps({
            "event": "llm_call",
            "trace_id": trace_id,
            "agent": self.name,
            "model": self.model,
            "latency_ms": latency_ms,
            "token_count": response.usage.total_tokens,
            "response_preview": response.choices[0].message.content[:500]
        }))
        
        # Alert on high latency
        if latency_ms > 30000:  # 30 seconds
            logger.error(json.dumps({
                "event": "high_latency_alert",
                "trace_id": trace_id,
                "agent": self.name,
                "latency_ms": latency_ms
            }))
        
        return response.choices[0].message.content