Agent Engineering

# LangGraph — stateful agent with checkpointing and crash recovery
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver

class AgentState(TypedDict):
    messages: list[BaseMessage]
    plan: list[str]
    current_step: int
    context: dict

graph = StateGraph(AgentState)
graph.add_node("plan", create_plan)
graph.add_node("execute", execute_step)
graph.add_node("evaluate", evaluate_result)
graph.add_node("replan", replan_if_needed)

graph.add_edge("plan", "execute")
graph.add_edge("execute", "evaluate")
graph.add_conditional_edges("evaluate", should_continue, {
    "continue": "execute",
    "replan": "replan",
    "done": END,
})

# Persistent checkpointing — survives crashes, enables time-travel debugging
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
agent = graph.compile(checkpointer=checkpointer)

# Claude Agent SDK — native shell + filesystem tools, hooks for interception
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher

async def block_destructive_bash(input_data, tool_use_id, context):
    cmd = input_data["tool_input"].get("command", "")
    if any(p in cmd for p in ("rm -rf", "DROP TABLE", "TRUNCATE")):
        return {"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": "Destructive command blocked.",
        }}
    return {}

options = ClaudeAgentOptions(
    system_prompt="""You are a customer operations agent.
    Query the database, analyze patterns, take action.
    Always explain what you're about to do before doing it.""",
    # Read/Write/Edit/Bash are NATIVE — no MCP server needed for filesystem
    allowed_tools=["Read", "Bash", "mcp__customer_db__lookup"],
    mcp_servers={
        "customer_db": {"type": "stdio", "command": "customer-db-server"},
    },
    # Hooks intercept every tool call before execution — distinctive to this SDK
    hooks={"PreToolUse": [HookMatcher(matcher="Bash", hooks=[block_destructive_bash])]},
    permission_mode="acceptEdits",
    max_turns=10,
)

async with ClaudeSDKClient(options=options) as client:
    await client.query("Investigate why churn spiked 15% last week")
    async for msg in client.receive_response():
        print(msg)

# OpenAI Agents SDK — handoffs as the primary primitive for routing
from agents import Agent, Runner, function_tool

@function_tool
def lookup_invoice(customer_id: str) -> str:
    """Lookup invoice by customer ID."""
    return fetch_invoice(customer_id)

@function_tool
def restart_service(service_name: str) -> str:
    """Restart a customer's service."""
    return restart(service_name)

@function_tool
def update_account_email(customer_id: str, new_email: str) -> str:
    """Update the account holder's email."""
    return update_email(customer_id, new_email)

billing_agent = Agent(
    name="Billing Agent",
    handoff_description="Handles billing, invoices, refunds, payment methods.",
    instructions="Handle billing inquiries. Access invoices and payment history.",
    tools=[lookup_invoice],
    input_guardrails=[detect_prompt_injection],
    output_guardrails=[no_pii_in_output],
)

technical_agent = Agent(
    name="Technical Agent",
    handoff_description="Handles outages, errors, restarts, debugging.",
    instructions="Diagnose technical issues. Restart services if needed.",
    tools=[restart_service],
)

account_agent = Agent(
    name="Account Agent",
    handoff_description="Handles account changes — email, password, profile.",
    instructions="Handle account management requests.",
    tools=[update_account_email],
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="Route the customer to the right specialist based on their question.",
    handoffs=[billing_agent, technical_agent, account_agent],  # typed Agent objects
)

# Each handoff becomes a typed tool call with full tracing built in
result = await Runner.run(triage_agent, "I was charged twice this month")
print(f"Resolved by: {result.last_agent.name}")  # → "Billing Agent"
print(result.final_output)

# Context budget allocation — a well-engineered agent
context_window: 200k tokens

allocation:
  system_prompt:     2%    # ~4K tokens — role, boundaries, output format
  tool_definitions:  8%    # ~16K tokens — MCP tools, schemas, examples
  memory:           15%    # ~30K tokens — conversation + long-term
  retrieved_context: 35%   # ~70K tokens — RAG results, documents
  working_space:    40%    # ~80K tokens — reasoning, tool outputs

compression_thresholds:
  tool_output: 20k         # Offload to filesystem with 10-line preview
  total_window: 85%        # Trigger structured summarization
  stale_context: 3_turns   # Compress conversation older than 3 turns

# Critical: place high-priority context at the END of the window
# Transformer attention has recency bias — middle content gets ignored

// MCP server — connecting an agent to your database (SDK v2 API)
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";

const server = new McpServer({ name: "customer-db", version: "1.0.0" });

server.registerTool(
  "lookup_customer",
  {
    description:
      "Find customer by email or ID. Returns profile, plan, and recent activity.",
    inputSchema: {
      email: z.string().email().optional(),
      id: z.string().optional(),
    },
  },
  async ({ email, id }) => {
    const customer = await db.customers.findOne(
      email ? { email } : { id }
    );
    // Return only what the agent needs — not the full record
    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          name: customer.name,
          plan: customer.plan,
          status: customer.status,
          recentTickets: customer.tickets.slice(0, 5),
        }),
      }],
    };
  }
);

# Agentic RAG — the agent decides retrieval strategy, not a fixed pipeline
from langgraph.graph import StateGraph

class RAGState(TypedDict):
    query: str
    documents: list[Document]
    answer: str
    confidence: float
    retrieval_attempts: int

def route_query(state: RAGState) -> str:
    """Agent decides retrieval strategy based on query type."""
    if is_entity_lookup(state["query"]):
        return "graph_rag"       # Entity relationships → knowledge graph
    if is_simple_factual(state["query"]):
        return "naive_retrieval" # Simple facts → fastest pipeline
    return "hybrid_retrieval"    # Complex → vector + BM25 + reranking

def evaluate_sufficiency(state: RAGState) -> str:
    """Agent evaluates whether retrieved docs answer the query."""
    if state["confidence"] > 0.85:
        return "generate_answer"
    if state["retrieval_attempts"] < 3:
        return "rewrite_and_retry"  # Query transformation loop
    return "escalate_to_human"

# Voice agent latency budget — cascading architecture
pipeline:
  target_total: 2000ms     # Maximum acceptable end-to-end

  stt:                      # Speech-to-Text
    provider: deepgram-nova-3
    latency: 250ms
    wer: 6.84%

  llm:                      # Reasoning + tool use
    provider: gemini-2.5-flash  # Fastest for voice
    ttft: 300ms
    max_tokens: 200         # Short, conversational responses

  tts:                      # Text-to-Speech
    provider: elevenlabs-flash-v2.5
    latency: 75ms
    streaming: true         # Start playing before full generation

  overhead:                 # Network + processing
    budget: 375ms

# Memory is critical for voice — users cannot scroll back
# Mem0 integration for cross-session context persistence

# Agent evaluation — testing the full trajectory, not just the answer
from deepeval import evaluate
from deepeval.test_case import ConversationalTestCase, Turn
from deepeval.metrics import (
    ConversationCompletenessMetric,  # User intentions satisfied across dialogue?
    RoleAdherenceMetric,             # Agent stays in character?
)

# Golden dataset: multi-turn conversation with expected behavior
convo_test_case = ConversationalTestCase(
    turns=[
        Turn(role="user", content="I need to cancel my subscription"),
        Turn(role="assistant",
             content="I've cancelled it. Your refund will process in 3–5 days."),
        Turn(role="user", content="Actually, can I downgrade instead?"),
        Turn(role="assistant",
             content="Done — downgraded to Basic and the cancellation is reversed."),
    ],
)

# Keyword signature: evaluate(test_cases=[...], metrics=[...])
evaluate(
    test_cases=[convo_test_case],
    metrics=[
        ConversationCompletenessMetric(threshold=0.7),
        RoleAdherenceMetric(
            threshold=0.8,
            role="You are a helpful customer support agent.",
        ),
    ],
)

# Three-layer guardrail architecture
guardrails:
  layer_1_rules:          # < 10ms — runs on every action
    - block_pii_in_output
    - enforce_response_length
    - validate_tool_arguments
    - check_rate_limits

  layer_2_classifiers:    # 50-200ms — runs on user-facing output
    - topic_boundary_check
    - sentiment_detection
    - toxicity_filter

  layer_3_semantic:       # 300-2000ms — runs on high-risk actions
    - policy_compliance_check
    - factual_grounding_verification
    - escalation_decision

  routing:
    low_risk:  [layer_1]                    # Fast path
    medium:    [layer_1, layer_2]           # Standard
    high_risk: [layer_1, layer_2, layer_3]  # Full verification

  fallback:
    on_uncertainty: escalate_to_human
    on_failure: return_safe_default
    on_timeout: retry_once_then_escalate

# Long-running agent in sandbox — overnight task execution
$ claude -p "Migrate all API handlers from Express to Hono. Run tests after each file." \
    --sandbox                     # Isolated execution environment
    --permission-mode default     # Ask before destructive actions
    --output-pr                   # Create PR on completion

# Skills — domain expertise loaded contextually
# .claude/skills/database-migrations/SKILL.md activates when
# the agent touches migration files, injecting project-specific
# patterns without bloating base context

# Agent Teams — parallel sandboxed agents
$ claude --agent test-writer "Write integration tests for auth module"
$ claude --agent security-audit "Scan for OWASP Top 10 vulnerabilities"
# Each agent runs in isolation with restricted tool access