Chapter 18 · Deployment & Production
Part of Part IV · Production
"A demo is not a product. This chapter is about the unglamorous work that turns your Jupyter notebook into something people can actually use."
Your agent is brilliant in a notebook. You show it to a colleague. They ask: "How do I use it?" You realize: there's no API, no authentication, no error handling, no monitoring. It's not a product. It's a demo.
Productionizing an agent is 80% standard software engineering and 20% agent-specific concerns. This chapter covers both: APIs, streaming, scaling, cost management, and the resilience patterns that keep your agent running when things go wrong.
Section 1: The Agent API
Wrapping your agent in a web API. FastAPI is the standard choice — async-native, auto-generated docs, great performance.
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
import json
app = FastAPI(title="Research Agent API")
class RunRequest(BaseModel):
query: str
max_steps: int = 20
stream: bool = False
class RunResponse(BaseModel):
run_id: str
status: str # "running", "completed", "failed"
result: str | None = None
# Store for running/past runs
runs: dict[str, dict] = {}
@app.post("/agent/run", response_model=RunResponse)
async def run_agent(request: RunRequest):
"""Run the agent synchronously — wait for complete result."""
run_id = generate_run_id()
runs[run_id] = {"status": "running", "result": None}
try:
agent = ResearchAgent()
result = await agent.run(request.query, max_steps=request.max_steps)
runs[run_id] = {"status": "completed", "result": result}
return RunResponse(run_id=run_id, status="completed", result=result)
except Exception as e:
runs[run_id] = {"status": "failed", "result": str(e)}
raise HTTPException(status_code=500, detail=str(e))
@app.post("/agent/run/stream")
async def run_agent_streaming(request: RunRequest):
"""Run the agent with Server-Sent Events — real-time progress."""
async def event_stream():
agent = ResearchAgent()
async for event in agent.run_streaming(request.query):
yield f"data: {json.dumps(event)}\n\n"
yield f"data: {json.dumps({'type': 'done'})}\n\n"
return StreamingResponse(
event_stream(),
media_type="text/event-stream",
headers={"X-Accel-Buffering": "no"}, # Disable nginx buffering
)
@app.get("/agent/run/{run_id}")
async def get_run_status(run_id: str):
"""Check the status of a running or completed task."""
if run_id not in runs:
raise HTTPException(status_code=404, detail="Run not found")
return runs[run_id]
@app.post("/agent/run/{run_id}/cancel")
async def cancel_run(run_id: str):
"""Cancel a running task."""
if run_id not in runs:
raise HTTPException(status_code=404, detail="Run not found")
runs[run_id]["status"] = "cancelled"
return {"status": "cancelled"}
Section 2: Streaming Agent Output
Agents can take minutes. Users need to see progress. Server-Sent Events (SSE) is the standard.
What to stream: Each thought, each tool call, each tool result, intermediate findings, and the final answer.
async def run_streaming(self, query: str):
"""Run the agent, yielding events as they happen."""
yield {"type": "status", "message": "Planning research..."}
plan = await self._plan(query)
yield {"type": "plan", "steps": plan}
for i, step in enumerate(plan):
yield {"type": "status", "message": f"Step {i+1}/{len(plan)}: {step['description']}"}
yield {"type": "thought", "content": step["reasoning"]}
yield {"type": "tool_call", "tool": "search_web", "args": step["search_query"]}
results = await self._search(step["search_query"])
yield {"type": "tool_result", "tool": "search_web", "result_count": len(results)}
yield {"type": "thought", "content": "Analyzing results..."}
analysis = await self._analyze(results)
yield {"type": "analysis", "content": analysis}
yield {"type": "status", "message": "Writing final report..."}
report = await self._write_report()
yield {"type": "done", "report": report}
Client-side consumption (JavaScript):
const eventSource = new EventSource('/agent/run/stream');
eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === 'thought') {
console.log('Agent thinking:', data.content);
} else if (data.type === 'tool_call') {
console.log('Calling tool:', data.tool, data.args);
} else if (data.type === 'done') {
console.log('Complete!', data.report);
eventSource.close();
}
};
Section 3: Scaling Agents
Agents are resource-intensive. Each run may involve 5-50 LLM calls, each taking 1-10 seconds. A hundred simultaneous users means thousands of API calls in flight.
The scaling stack:
import asyncio
from asgiref.sync import async_to_sync
class AgentWorker:
"""Async agent worker that handles concurrent runs efficiently."""
def __init__(self, max_concurrency: int = 10):
self.semaphore = asyncio.Semaphore(max_concurrency)
self.active_runs: dict[str, asyncio.Task] = {}
async def submit(self, run_id: str, query: str) -> None:
"""Submit a task. It will run when capacity is available."""
async with self.semaphore:
task = asyncio.create_task(self._run_agent(run_id, query))
self.active_runs[run_id] = task
try:
await task
finally:
del self.active_runs[run_id]
async def _run_agent(self, run_id: str, query: str):
agent = ResearchAgent()
result = await agent.run(query)
runs[run_id] = {"status": "completed", "result": result}
async def cancel(self, run_id: str):
if run_id in self.active_runs:
self.active_runs[run_id].cancel()
For heavier workloads, use a task queue:
# Celery worker for background agent execution
from celery import Celery
celery_app = Celery('agent_tasks', broker='redis://localhost:6379/0')
@celery_app.task(bind=True)
def run_agent_task(self, query: str, max_steps: int = 20):
agent = ResearchAgent()
result = agent.run_sync(query, max_steps)
return result
Key scaling principles:
- Async everywhere. Use async/await for all I/O — LLM calls, tool execution, database queries. Never block the event loop.
- Connection pooling. Reuse HTTP connections to LLM APIs. Don't create a new connection per request.
- Rate limit awareness. Respect LLM API rate limits. Implement client-side throttling with token buckets.
- Horizontal scaling. Run multiple agent workers behind a load balancer. Each worker handles a subset of requests.
Section 4: Cost Management
Agent costs can spiral fast. Here's the math:
50 LLM calls/run × $0.015/call × 1,000 users/day = $750/day = $22,500/month
Cost optimization strategies:
class CostOptimizedAgent:
def __init__(self):
self.cost_tracker = CostTracker()
self.semantic_cache = SemanticCache()
async def think(self, messages: list[dict]) -> str:
# Strategy 1: Semantic cache — skip LLM call if we've seen this before
cache_key = self._hash_messages(messages)
cached = self.semantic_cache.get(cache_key)
if cached:
self.cost_tracker.record_saved(cost=0.015) # What we saved
return cached
# Strategy 2: Model routing — cheap model for simple decisions
complexity = self._assess_complexity(messages)
if complexity == "simple":
model = "claude-haiku-4-5-20251001" # ~$0.001/call
elif complexity == "medium":
model = "claude-sonnet-4-20250514" # ~$0.003/call
else:
model = "claude-opus-4-20250514" # ~$0.015/call
response = await self._call_llm(messages, model)
self.cost_tracker.record_call(model, response.usage)
self.semantic_cache.set(cache_key, response.content)
return response.content
class CostTracker:
def __init__(self, daily_budget: float = 100.0):
self.daily_budget = daily_budget
self.calls: list[dict] = []
def record_call(self, model: str, usage):
cost = self._calculate_cost(model, usage)
self.calls.append({"model": model, "cost": cost, "time": datetime.now()})
daily_total = sum(c["cost"] for c in self.calls
if c["time"].date() == datetime.now().date())
if daily_total > self.daily_budget * 0.8:
print(f"WARNING: 80% of daily budget used (${daily_total:.2f}/${self.daily_budget})")
if daily_total > self.daily_budget:
raise BudgetExceededError(f"Daily budget exceeded: ${daily_total:.2f}")
def _calculate_cost(self, model: str, usage) -> float:
prices = {
"claude-haiku-4-5-20251001": (0.80, 4.00), # per 1M tokens
"claude-sonnet-4-20250514": (3.00, 15.00),
"claude-opus-4-20250514": (15.00, 75.00),
}
input_price, output_price = prices.get(model, (0, 0))
return (usage.input_tokens * input_price + usage.output_tokens * output_price) / 1_000_000
The cost optimization checklist:
- Model routing: Cheap models for simple decisions, expensive models only when needed
- Semantic caching: Don't call the LLM twice for the same question
- Prompt caching: Cache static system prompts and tool definitions (Anthropic)
- Token optimization: Shorter prompts, fewer examples, concise tool descriptions
- Conversation limits: Summarize old messages instead of keeping them all
- Tool call limits: Cap the number of tool calls per run
Section 5: LLM Fallbacks and Resilience
LLM APIs fail. They rate limit. They return errors. They have outages. Your agent must handle this.
import random
class ResilientLLMClient:
def __init__(self):
self.circuit_breakers: dict[str, CircuitBreaker] = {}
self.models = [
"claude-sonnet-4-20250514", # Primary
"gpt-4o", # Fallback 1
"claude-haiku-4-5-20251001", # Fallback 2 (cheaper, faster)
]
async def call(self, messages: list[dict], **kwargs) -> dict:
last_error = None
for model in self.models:
try:
# Check circuit breaker
if self._is_circuit_open(model):
continue
response = await self._call_with_retry(model, messages, **kwargs)
self._circuit_success(model)
return response
except RateLimitError:
last_error = "Rate limited"
await asyncio.sleep(1 + random.random() * 2) # Jitter
except APIError as e:
last_error = str(e)
self._circuit_failure(model)
except TimeoutError:
last_error = "Timeout"
self._circuit_failure(model)
raise AllModelsFailedError(f"All models failed. Last error: {last_error}")
async def _call_with_retry(self, model: str, messages: list[dict], max_retries: int = 3, **kwargs):
for attempt in range(max_retries):
try:
return await call_llm(model, messages, **kwargs)
except (RateLimitError, TimeoutError) as e:
if attempt == max_retries - 1:
raise
wait = (2 ** attempt) + random.random()
await asyncio.sleep(wait)
The resilience stack:
- Retry with exponential backoff — transient errors resolve quickly
- Model fallback — if primary model fails, try alternatives
- Circuit breaker — if a model fails repeatedly, stop trying for a while
- Graceful degradation — if all models fail, return a cached or partial response
- Timeout — never wait more than N seconds for an LLM response
Section 6: Prompt and Agent Versioning
Your agent's behavior is determined by: system prompt, model version, tool definitions, temperature, and other parameters. When you change any of these, you've created a new version.
Version everything:
# prompts/research_agent/v2.yaml
version: "2.0.0"
date: "2026-07-27"
model: "claude-sonnet-4-20250514"
temperature: 0.1
system_prompt: |
You are a research assistant. Your task is to research topics thoroughly
and produce well-sourced, accurate reports.
CAPABILITIES:
- Search the web for current information
- Analyze and synthesize information from multiple sources
- Write clear, structured reports with citations
CONSTRAINTS:
- Never fabricate information. If you don't know, say so.
- Always cite your sources.
- If search results are insufficient, broaden your query.
OUTPUT FORMAT:
Respond with a structured report containing:
1. Executive Summary
2. Key Findings (with citations)
3. Analysis
4. Sources
tools:
- name: search_web
description: Search the web for current information
parameters:
query:
type: string
description: The search query
- name: fetch_page
description: Fetch and read a web page
parameters:
url:
type: string
description: The URL to fetch
eval_baseline:
pass_rate: 0.87
avg_accuracy: 4.2
avg_latency: 45.3
The deploy checklist:
- Eval suite passes (no regression > 5%)
- Prompt is versioned in git
- Model version is pinned (not "latest")
- Deploy to staging, run smoke tests
- Deploy to production, monitor metrics for 30 minutes
- If error rate spikes or latency degrades → rollback
Section 7: Monitoring in Production
What to monitor:
| Metric | What It Tells You | Alert Threshold |
|---|---|---|
| Success rate | Are agents completing tasks? | < 85% |
| P95 latency | How long do users wait? | > 120s |
| Error rate | Are LLMs/tools failing? | > 5% |
| Cost per run | Are costs under control? | > $2.00 |
| Tool call distribution | Are tools being used as expected? | Unused tool > 7 days |
| Hallucination rate | Are agents making things up? | > 10% |
Setting up monitoring with LangFuse:
from langfuse import Langfuse
langfuse = Langfuse()
# Track each agent run
def monitor_agent_run(func):
async def wrapper(query: str, *args, **kwargs):
trace = langfuse.trace(
name="research_agent",
input={"query": query},
metadata={"model": "claude-sonnet-4-20250514"},
)
start = time.time()
try:
result = await func(query, *args, **kwargs)
trace.update(
output={"result": result},
metadata={"latency": time.time() - start, "success": True},
)
return result
except Exception as e:
trace.update(
output={"error": str(e)},
metadata={"latency": time.time() - start, "success": False},
)
raise
finally:
trace.end()
return wrapper
Section 8: The Complete Production Architecture
┌──────────────┐
│ CDN/Proxy │
└──────┬───────┘
│
┌──────▼───────┐
│ FastAPI │
│ Server │
└──────┬───────┘
│
┌────────────┼────────────┐
│ │ │
┌─────▼─────┐ ┌───▼────┐ ┌────▼─────┐
│ Agent │ │ Agent │ │ Agent │
│ Worker 1 │ │Worker 2│ │ Worker N │
└─────┬─────┘ └───┬────┘ └────┬─────┘
│ │ │
└────────────┼────────────┘
│
┌──────────────────┼──────────────────┐
│ │ │
┌─────▼─────┐ ┌──────▼──────┐ ┌──────▼──────┐
│ Claude │ │ GPT-4 │ │ Gemini │
│ API │ │ API │ │ API │
└───────────┘ └─────────────┘ └─────────────┘
│ │ │
└──────────────────┼──────────────────┘
│
┌────────────┼────────────┐
│ │ │
┌─────▼─────┐ ┌───▼────┐ ┌────▼─────┐
│ Redis │ │Postgres│ │ Vector │
│ (Cache) │ │ (State)│ │ Store │
└───────────┘ └────────┘ └──────────┘
│
┌──────▼───────┐
│ Monitoring │
│ (LangFuse) │
└──────────────┘
The Turn
You now understand that productionizing an agent is not magic. It's standard software engineering applied to a new domain. APIs, streaming, scaling, cost management, monitoring — these are the same patterns you'd use for any production service, with agent-specific adaptations.
The difference between a demo and a product is not the quality of the agent's reasoning. It's the quality of the infrastructure around it. A brilliant agent that's down half the time is worse than a mediocre agent that's always available.
In the next chapter: You now have all the pieces. You can build agents, give them tools and memory, make them reason, secure them, evaluate them, and deploy them to production. In the final two chapters, you'll put everything together in two capstone projects: a research assistant that reads, understands, and synthesizes information from dozens of sources, and a coding agent that writes, tests, and deploys software.