Capstone: Building the Enterprise Customer Resolution System

Every module in this course built one real capability. This final module builds nothing new. It assembles everything — state, nodes, conditional edges, reducers, Command, Send, subgraphs, interrupts, checkpointing, threads, and tests — into one complete, realistic application, exactly the shape a genuine, production customer-resolution system actually takes.

The architecture

flowchart TD
    START --> understand
    understand --> classify
    classify -->|billing| billing_subgraph
    classify -->|technical| technical_subgraph
    billing_subgraph --> risk_check
    technical_subgraph --> risk_check
    risk_check -->|high risk| approval
    risk_check -->|low risk| execute
    approval -->|approved| execute
    approval -->|rejected| notify_rejection
    execute --> END
    notify_rejection --> END

Project structure

customer-agent/
├── app/
│   ├── graph.py
│   ├── state.py
│   ├── nodes/
│   │   ├── __init__.py
│   │   ├── classify.py
│   │   ├── billing.py
│   │   ├── technical.py
│   │   └── approval.py
│   ├── tools/
│   │   ├── __init__.py
│   │   └── customer_tools.py
│   ├── config.py
│   └── main.py
├── tests/
│   ├── test_nodes.py
│   └── test_graph.py
├── .env.example
└── requirements.txt

config.py — recall Module 3 (LangChain course)

from dotenv import load_dotenv
import os

load_dotenv()

OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("Missing OPENAI_API_KEY — check your .env file.")

state.py — recall Modules 3, 11, 12

from typing import TypedDict, Annotated, Literal
import operator
from langgraph.graph.message import add_messages

class ResolutionState(TypedDict):
    messages: Annotated[list, add_messages]
    query: str
    category: Literal["billing", "technical", ""]
    findings: Annotated[list[str], operator.add]
    risk_level: Literal["low", "high", ""]
    approved: bool
    resolution: str

Notice findings uses operator.add, exactly like Module 23’s shared multi-agent state — different specialist subgraphs will genuinely accumulate their own real findings into this one, shared field.

tools/customer_tools.py — recall Module 12 (LangChain course)

from langchain.tools import tool

ORDERS = {"o_1": {"customer_id": "c_1", "amount": 45.0, "days_since_purchase": 12}}

@tool
def get_order(order_id: str) -> str:
    """Look up order details by order ID."""
    order = ORDERS.get(order_id)
    return str(order) if order else f"No order found with ID {order_id}."

@tool
def retry_payment(order_id: str) -> str:
    """Retry a failed payment for an order."""
    return f"Payment for order {order_id} retried successfully."

@tool
def issue_refund(order_id: str, amount: float) -> str:
    """Issue a refund for an order."""
    return f"Refund of ${amount} issued for order {order_id}."

nodes/classify.py — recall Module 8

from langchain.chat_models import init_chat_model
from pydantic import BaseModel
from typing import Literal

class Classification(BaseModel):
    category: Literal["billing", "technical"]

model = init_chat_model("openai:gpt-4o-mini").with_structured_output(Classification)

def understand(state: dict) -> dict:
    return {"query": state["messages"][-1].content}

def classify(state: dict) -> dict:
    result = model.invoke(f"Classify this customer request: {state['query']}")
    return {"category": result.category}

def route_by_category(state: dict) -> str:
    return state["category"]

nodes/billing.py — recall Modules 22, 25

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from app.tools.customer_tools import get_order, retry_payment

class BillingState(TypedDict):
    query: str
    findings: list[str]

def check_order(state: BillingState) -> dict:
    try:
        result = get_order.invoke({"order_id": "o_1"})
        return {"findings": [f"Order lookup: {result}"]}
    except Exception as e:
        return {"findings": [f"Order lookup failed: {e}"]}

def attempt_retry(state: BillingState) -> dict:
    result = retry_payment.invoke({"order_id": "o_1"})
    return {"findings": [result]}

billing_builder = StateGraph(BillingState)
billing_builder.add_node("check_order", check_order)
billing_builder.add_node("attempt_retry", attempt_retry)
billing_builder.add_edge(START, "check_order")
billing_builder.add_edge("check_order", "attempt_retry")
billing_builder.add_edge("attempt_retry", END)

billing_subgraph = billing_builder.compile()

nodes/technical.py — a second, parallel specialist

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class TechnicalState(TypedDict):
    query: str
    findings: list[str]

def diagnose(state: TechnicalState) -> dict:
    return {"findings": [f"Diagnosis complete for: {state['query']}"]}

technical_builder = StateGraph(TechnicalState)
technical_builder.add_node("diagnose", diagnose)
technical_builder.add_edge(START, "diagnose")
technical_builder.add_edge("diagnose", END)

technical_subgraph = technical_builder.compile()

nodes/approval.py — recall Modules 19, 20

from langgraph.types import interrupt

def assess_risk(state: dict) -> dict:
    return {"risk_level": "high" if "refund" in state["query"].lower() else "low"}

def request_approval(state: dict) -> dict:
    decision = interrupt({"question": f"Approve action for: {state['query']}?"})
    return {"approved": decision}

def route_after_risk(state: dict) -> str:
    return "request_approval" if state["risk_level"] == "high" else "execute"

def route_after_approval(state: dict) -> str:
    return "execute" if state["approved"] else "notify_rejection"

def execute(state: dict) -> dict:
    return {"resolution": f"Resolved: {'; '.join(state['findings'])}"}

def notify_rejection(state: dict) -> dict:
    return {"resolution": "Action was reviewed and not approved."}

graph.py — assembling everything

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from app.state import ResolutionState
from app.nodes.classify import understand, classify, route_by_category
from app.nodes.billing import billing_subgraph
from app.nodes.technical import technical_subgraph
from app.nodes.approval import (
    assess_risk, request_approval, execute, notify_rejection,
    route_after_risk, route_after_approval,
)

builder = StateGraph(ResolutionState)
builder.add_node("understand", understand)
builder.add_node("classify", classify)
builder.add_node("billing", billing_subgraph)
builder.add_node("technical", technical_subgraph)
builder.add_node("assess_risk", assess_risk)
builder.add_node("request_approval", request_approval)
builder.add_node("execute", execute)
builder.add_node("notify_rejection", notify_rejection)

builder.add_edge(START, "understand")
builder.add_edge("understand", "classify")
builder.add_conditional_edges("classify", route_by_category, {"billing": "billing", "technical": "technical"})
builder.add_edge("billing", "assess_risk")
builder.add_edge("technical", "assess_risk")
builder.add_conditional_edges("assess_risk", route_after_risk, {"request_approval": "request_approval", "execute": "execute"})
builder.add_conditional_edges("request_approval", route_after_approval, {"execute": "execute", "notify_rejection": "notify_rejection"})
builder.add_edge("execute", END)
builder.add_edge("notify_rejection", END)

graph = builder.compile(checkpointer=InMemorySaver())

Read this against the module’s own opening diagram directly — every real edge here maps to exactly one arrow in that diagram. billing and technical are real, independently testable subgraphs from Module 22. assess_risk and request_approval form the exact real interrupt pattern from Modules 19-20. Nothing here is new; it’s every mechanism from this course, wired together.

main.py — running it end to end

from app.graph import graph
from langgraph.types import Command

def main():
    config = {"configurable": {"thread_id": "customer-session-1"}}

    result = graph.invoke(
        {"messages": [{"role": "user", "content": "I need a refund for order o_1"}], "findings": [], "category": "", "risk_level": "", "approved": False, "resolution": "", "query": ""},
        config=config,
    )

    if "__interrupt__" in result:
        print("Paused for human approval...")
        result = graph.invoke(Command(resume=True), config=config)

    print(result["resolution"])

if __name__ == "__main__":
    main()

tests/test_nodes.py — recall Module 27

from app.nodes.approval import route_after_risk, route_after_approval

def test_route_after_risk_high():
    assert route_after_risk({"risk_level": "high"}) == "request_approval"

def test_route_after_approval_rejected():
    assert route_after_approval({"approved": False}) == "notify_rejection"

tests/test_graph.py

from langgraph.types import Command
from app.graph import graph

def test_full_flow_pauses_and_resumes_for_refund():
    config = {"configurable": {"thread_id": "test-thread"}}
    result = graph.invoke(
        {"messages": [{"role": "user", "content": "I need a refund for order o_1"}], "findings": [], "category": "", "risk_level": "", "approved": False, "resolution": "", "query": ""},
        config=config,
    )
    assert "__interrupt__" in result

    final = graph.invoke(Command(resume=True), config=config)
    assert "Resolved" in final["resolution"]

Tracing the real, complete data flow

A real customer message enters at understand, extracting the raw query. classify genuinely reads it and routes to billing or technical — each a real, independently testable subgraph, whose own real findings accumulate into the shared findings field via Module 11’s reducer. assess_risk genuinely decides whether this specific resolution is high-stakes enough to need request_approval’s real interrupt() — and if the graph pauses there, main.py’s own real check for "__interrupt__" is what tells your application a human is genuinely needed before execute can run. Every one of these steps is checkpointed, meaning this entire flow could genuinely pause for a human review that takes two real days, and resume exactly where it left off, on a completely different machine, without losing a single piece of real progress.

Common mistakes worth avoiding, at the level of a real, complete application

Forgetting that billing and technical subgraphs need field names aligned with the parent’s shared state. Recall Module 22’s real distinction — BillingState and TechnicalState both share query and findings with ResolutionState deliberately, so they can be added directly as nodes without a translation wrapper. Changing a subgraph’s field names without updating this alignment silently breaks the handoff.

Deploying this exact application with InMemorySaver, unchanged. Recall Module 28’s own direct warning — the version in this module’s graph.py is genuinely correct for learning and local testing; a real deployment needs PostgresSaver in its place before it ever serves a real, paying customer.

Skipping the "__interrupt__" check in main.py. Recall Module 19’s own core lesson — if main.py didn’t check for it explicitly, a paused, high-risk refund would appear to simply hang, when it’s actually working exactly as designed, waiting for a real human who hasn’t been notified to look.

Closing this entire course

You began Module 1 watching a LangChain agent’s system prompt ask, uselessly, for a human to “pause and wait.” Twenty-nine modules later, you’ve built a real system where that pause is a structural guarantee — checkpointed, resumable, testable, and genuinely production-shaped.

The honest measure of this course was never “can you name LangGraph’s APIs.” It’s this: given a real, unfamiliar agent architecture — drawn as a diagram, described in a meeting, sketched on a whiteboard — can you model its state, draw its real workflow, implement it as nodes and edges, control its routing and loops deliberately, persist its execution, pause it for a human, recover it from a real failure, compose it from genuine subgraphs, and build a reliable, tested, multi-agent application from it. That’s what you now have.