Over the past eighteen months we have deployed AI agents into production for customer service, document processing and internal operations systems. Every one of those deployments started with the same optimistic assumption: a well-crafted prompt chain would be enough. Every one of them eventually required us to rebuild around a proper agent architecture. This post documents what we learned and why LangGraph became our default orchestration layer for agentic systems.

Why Simple Prompt Chains Break

A prompt chain is a linear sequence: take user input, pass it through prompt A, feed the result into prompt B, return the output. This works well for structured transformations — summarization, translation, format conversion. It fails the moment your system needs to make decisions about what to do next.

In our retail customer service deployment, the initial chain handled refund requests cleanly. But real conversations are not linear. A customer might start asking about a refund, pivot to asking where their package is, then circle back to the refund but for a different item. A chain cannot handle this because:

The breaking point for us was when a customer service agent needed to call three different backend APIs (order lookup, inventory check, refund processor), handle partial failures from any of them, and maintain a coherent conversation through all of it. That is not a chain. That is a state machine.

Agent Architecture: The State Machine Mental Model

Once you accept that an agent is a state machine, the design follows naturally. Each node in the graph represents a discrete action: calling an LLM, invoking a tool, validating output, waiting for human approval. Edges represent transitions — and critically, those transitions can be conditional.

The core components of our agent architecture:

LangGraph: Nodes, Edges, State, Checkpoints

LangGraph implements this state machine pattern with a clean API. Here is the skeleton of a customer service agent we deploy in production:

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.redis import RedisSaver
from typing import TypedDict, Annotated, Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
import operator


class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    intent: str
    order_data: dict | None
    requires_approval: bool
    retry_count: int
    tool_results: Annotated[list, operator.add]


def classify_intent(state: AgentState) -> dict:
    """Classify the customer's intent from conversation history."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm.invoke([
        {"role": "system", "content": INTENT_CLASSIFIER_PROMPT},
        *state["messages"]
    ])
    return {"intent": response.content.strip().lower()}


def route_by_intent(state: AgentState) -> Literal[
    "handle_refund", "track_order", "general_inquiry", "escalate"
]:
    """Conditional edge: route to the appropriate handler."""
    intent = state["intent"]
    if intent in ("refund", "return", "exchange"):
        return "handle_refund"
    elif intent in ("tracking", "shipping", "delivery"):
        return "track_order"
    elif state["retry_count"] > 2:
        return "escalate"
    return "general_inquiry"


def handle_refund(state: AgentState) -> dict:
    """Process refund with tool calls and validation."""
    llm = ChatOpenAI(model="gpt-4o").bind_tools([
        lookup_order, check_refund_eligibility, process_refund
    ])
    response = llm.invoke(state["messages"])
    # Tool calls are handled by LangGraph's ToolNode
    return {"messages": [response]}


# Build the graph
builder = StateGraph(AgentState)
builder.add_node("classify", classify_intent)
builder.add_node("handle_refund", handle_refund)
builder.add_node("track_order", track_order)
builder.add_node("general_inquiry", general_inquiry)
builder.add_node("escalate", escalate_to_human)
builder.add_node("respond", generate_response)

builder.set_entry_point("classify")
builder.add_conditional_edges("classify", route_by_intent)
builder.add_edge("handle_refund", "respond")
builder.add_edge("track_order", "respond")
builder.add_edge("general_inquiry", "respond")
builder.add_edge("escalate", END)
builder.add_edge("respond", END)

# Redis checkpointing for persistence and recovery
checkpointer = RedisSaver(redis_url="redis://localhost:6379")
graph = builder.compile(checkpointer=checkpointer)

The key insight: routing logic lives in Python, not in prompts. The LLM classifies intent, but the conditional edge function is deterministic code. This means routing is testable, debuggable, and does not hallucinate.

Checkpointing and Human-in-the-Loop

Redis-backed checkpointing gives us two critical capabilities. First, if the agent crashes mid-execution (an API timeout, a rate limit, a deployment), it resumes from the last checkpoint rather than restarting. Second, we can pause execution at any node and wait for human approval:

from langgraph.graph import StateGraph
from langgraph.checkpoint.redis import RedisSaver


# Mark the refund node as requiring human approval for amounts > $200
builder.add_node("human_review", human_review_node)
builder.add_conditional_edges(
    "handle_refund",
    lambda state: "human_review" if state.get("refund_amount", 0) > 200 else "respond"
)

# When execution hits human_review, it checkpoints and stops.
# A webhook from our internal tool resumes it:
config = {"configurable": {"thread_id": conversation_id}}
graph.invoke({"messages": [HumanMessage("approved")]}, config=config)

In our retail deployment, this pattern handles approximately 15% of conversations — the ones involving high-value refunds or edge cases that fall outside policy. The agent does not block; it checkpoints state to Redis and resumes when the human operator makes a decision.

Multi-Agent Patterns

Single-agent graphs handle most workflows. But when the domain grows complex enough — when you need specialized expertise in different areas — multi-agent patterns become necessary. We have deployed three patterns in production:

Supervisor Pattern

One orchestrator agent delegates to specialist agents. The supervisor maintains conversation state and decides which specialist handles each turn. We use this for customer service systems where a single conversation might span orders, billing, and technical support:

def supervisor_node(state: AgentState) -> dict:
    """Supervisor decides which specialist agent handles the next step."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm.invoke([
        {"role": "system", "content": SUPERVISOR_PROMPT},
        *state["messages"],
        {"role": "system", "content": f"Available specialists: {SPECIALIST_LIST}"}
    ])
    return {"next_agent": response.content.strip()}


def route_to_specialist(state: AgentState) -> str:
    return state["next_agent"]


builder.add_conditional_edges("supervisor", route_to_specialist, {
    "orders": "orders_agent",
    "billing": "billing_agent",
    "technical": "technical_agent",
    "end": END
})

Hierarchical Pattern

When specialists themselves need sub-specialists. Our document processing pipeline uses this: a top-level agent classifies the document type, delegates to a domain-specific extraction agent, which in turn delegates to field-level validators. Three levels of the graph, each with their own state and tool sets.

Collaborative Pattern

Multiple agents working on the same state without a central supervisor. Each agent has a defined scope and triggers based on state conditions. We use this for internal analysis workflows where a research agent, a fact-checker, and a writer all contribute to the same output document.

Production Concerns

Error Recovery

Every external call fails eventually. Our standard pattern wraps tool nodes with retry logic and graceful degradation:

def resilient_tool_node(state: AgentState) -> dict:
    """Execute tool with retry, backoff, and graceful degradation."""
    max_retries = 3
    retry_count = state.get("retry_count", 0)

    try:
        result = execute_tool(state)
        return {"tool_results": [result], "retry_count": 0}
    except RateLimitError:
        if retry_count < max_retries:
            return {"retry_count": retry_count + 1}  # loops back
        return {"tool_results": [FALLBACK_RESPONSE], "retry_count": 0}
    except TimeoutError:
        # Checkpoint and alert — do not retry indefinitely
        return {"requires_escalation": True}

The graph structure makes this clean: a retry is just an edge back to the same node, with a counter in state preventing infinite loops.

Timeout Handling

Agents without time bounds are a production risk. We enforce timeouts at three levels: individual LLM calls (30 seconds), individual node execution (60 seconds), and total graph execution (5 minutes for customer-facing, 30 minutes for background processing). LangGraph's async support makes this straightforward with asyncio.wait_for wrappers.

Cost Control

An agent that loops indefinitely burns tokens. We enforce hard limits in state:

def cost_gate(state: AgentState) -> Literal["continue", "terminate"]:
    """Kill the agent if it exceeds token or iteration budgets."""
    if state.get("total_tokens", 0) > 50_000:
        return "terminate"
    if state.get("iteration_count", 0) > 15:
        return "terminate"
    return "continue"


builder.add_conditional_edges("tool_executor", cost_gate, {
    "continue": "next_step",
    "terminate": "budget_exceeded_response"
})

In practice, we track token usage per conversation and per customer. If a single conversation exceeds our budget threshold, the agent gracefully hands off to a human rather than continuing to spend.

Observability with LangSmith

You cannot operate what you cannot observe. Every production agent deployment includes LangSmith tracing. This gives us:

We run LangSmith in a dedicated project per agent deployment. Weekly reviews of traces surface the most common failure modes, which directly inform the next iteration of routing logic and prompts.

Guardrails: Keeping Agents Safe

Output Validation

Every agent response passes through a validation layer before reaching the user. We use Pydantic models to enforce structure on LLM outputs, and a separate lightweight model (GPT-4o-mini) to check responses against policy rules:

from pydantic import BaseModel, field_validator


class AgentResponse(BaseModel):
    message: str
    actions_taken: list[str]
    confidence: float

    @field_validator("message")
    @classmethod
    def no_hallucinated_policies(cls, v: str) -> str:
        """Reject responses that reference non-existent policies."""
        forbidden = ["lifetime guarantee", "unlimited refund", "free replacement"]
        for phrase in forbidden:
            if phrase.lower() in v.lower():
                raise ValueError(f"Hallucinated policy: {phrase}")
        return v


def validate_response(state: AgentState) -> dict:
    """Validate agent output before sending to user."""
    try:
        validated = AgentResponse.model_validate_json(
            state["draft_response"]
        )
        return {"final_response": validated.message}
    except ValidationError as e:
        return {"requires_regeneration": True, "validation_errors": str(e)}

Hallucination Detection

For our retail agent, hallucinated order numbers or tracking codes are a critical failure. We cross-reference every claim against the actual API response stored in state. If the agent says "your order #12345 shipped on Tuesday" but the order data in state shows a different status, we catch it and regenerate.

Scope Limiting

Agents should refuse to act outside their defined scope. We implement this with a combination of system prompt instructions and a classification gate. Before executing any action, a lightweight model classifies whether the request falls within the agent's defined capability set. Out-of-scope requests get a polite redirect rather than an attempted (and likely incorrect) answer.

Real Patterns from Our Retail Customer Service Agent

Our most mature agent deployment handles customer service for a retail company processing thousands of conversations monthly. Here are the architectural decisions that survived contact with production:

The Production Stack

For teams evaluating this architecture, here is the full stack we deploy:

Lessons Learned

After eighteen months of iterating on this architecture, these are the principles that hold:

  1. Start with a graph, not a chain. Even if your initial flow is linear, model it as a graph. The first edge case will require branching, and refactoring a chain into a graph is more work than starting with one.
  2. Put routing logic in code, not prompts. LLMs are good at classification. They are unreliable at self-directing complex workflows. Let the LLM classify; let Python route.
  3. Checkpoint everything. Failures are normal. The question is whether you can resume or must restart. Redis checkpointing costs microseconds per node and saves minutes of recomputation.
  4. Budget your agents. Token limits, iteration limits, time limits. An unbounded agent is a production incident waiting to happen.
  5. Observe before you optimize. LangSmith traces tell you where time and tokens are actually spent. Optimize the 20% of nodes that consume 80% of resources.
  6. Test with production traces. The best evaluation dataset is real conversations from production. Build regression tests from actual failures, not synthetic examples.

AI agents are not research projects anymore. They are production software systems that happen to include LLM calls. The engineering discipline we apply to any production system — error handling, observability, graceful degradation, testing — applies equally here. LangGraph gives us the structure to apply that discipline without fighting the framework.