Chapter 12 — Deployment & Production

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    thread_id: str = "default"

class ChatResponse(BaseModel):
    response: str
    tool_calls: list[dict] = []

@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    try:
        result = await agent.ainvoke(
            {"messages": [("user", req.message)]},
            config={"configurable": {"thread_id": req.thread_id}},
        )
        return ChatResponse(
            response=result["messages"][-1].content,
            tool_calls=[],
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    """Stream agent responses for real-time UX."""
    async def generate():
        async for event in agent.astream_events(
            {"messages": [("user", req.message)]},
            config={"configurable": {"thread_id": req.thread_id}},
            version="v2",
        ):
            if event["event"] == "on_chat_model_stream":
                chunk = event["data"]["chunk"]
                if chunk.content:
                    yield f"data: {chunk.content}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(generate(), media_type="text/event-stream")

# Just set environment variables — LangGraph auto-traces
export LANGCHAIN_TRACING_V2=true
export LANGCHAIN_API_KEY="ls__..."
export LANGCHAIN_PROJECT="my-research-agent"

# Every agent run now shows in LangSmith:
# - Full message trace (user → LLM → tool → LLM → ...)
# - Token counts and costs per step
# - Latency breakdown
# - Tool inputs/outputs

import logging
import time

logger = logging.getLogger("agent")

class AgentLogger:
    def on_llm_start(self, prompt, **kwargs):
        logger.info(f"LLM call started | model={kwargs.get('model')}")
        self.start_time = time.time()
    
    def on_llm_end(self, response, **kwargs):
        duration = time.time() - self.start_time
        tokens = response.usage
        logger.info(
            f"LLM call completed | "
            f"duration={duration:.2f}s | "
            f"input_tokens={tokens.prompt_tokens} | "
            f"output_tokens={tokens.completion_tokens}"
        )
    
    def on_tool_start(self, tool_name, tool_input, **kwargs):
        logger.info(f"Tool call: {tool_name}({tool_input})")
    
    def on_tool_error(self, error, **kwargs):
        logger.error(f"Tool error: {error}")

# 1. Use cheaper models for simple routing/classification
router_llm = ChatOpenAI(model="gpt-4o-mini")  # Cheap for classification
worker_llm = ChatOpenAI(model="gpt-4o")        # Powerful for complex work

# 2. Set token budgets
MAX_TOKENS_PER_REQUEST = 50_000
MAX_COST_PER_REQUEST = 1.00  # dollars

# 3. Cache tool results
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_search(query: str) -> str:
    return tavily.search(query)

# 4. Summarize long tool outputs before injecting into context
def truncate_tool_output(output: str, max_chars: int = 2000) -> str:
    if len(output) > max_chars:
        return output[:max_chars] + "\n...[truncated]"
    return output

import tenacity

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=1, min=1, max=10),
    retry=tenacity.retry_if_exception_type((RateLimitError, TimeoutError)),
)
async def call_llm_with_retry(messages, tools):
    return await client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        timeout=30,
    )