Building with Frameworks

from langchain.chat_models import init_chat_model
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def get_order(order_id: int, config: RunnableConfig) -> str:
    """Look up one of the customer's orders: total, delivery fee, how late it
    arrived. Use before any claim about an order."""
    customer = config["configurable"]["customer_id"]      # from the session, not the model
    return lookup_order(order_id, customer)

@tool
def refund_item(order_id: int, amount: int, config: RunnableConfig) -> str:
    """Refund part of an order to the customer's wallet. Use only after
    get_order confirms the complaint. Amount in rupees."""
    customer = config["configurable"]["customer_id"]
    return issue_refund(order_id, amount, customer)        # enforces its own limits

TOOLS = [get_order, refund_item]
model = init_chat_model("anthropic:claude-opus-5-5").bind_tools(TOOLS)
tool_node = ToolNode(TOOLS)            # runs the tool calls in the last message

from typing import Annotated, Literal, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]      # reducer: append, do not replace
    route: str
    steps: int

def route(state: State) -> dict:
    verdict = small_model.with_structured_output(Route).invoke(state["messages"][-1].content)
    return {"route": verdict.category, "steps": 0}

def agent(state: State) -> dict:
    reply = model.invoke([SYSTEM_MESSAGE, *state["messages"]])
    return {"messages": [reply], "steps": state["steps"] + 1}

def after_agent(state: State) -> Literal["tools", "approve", "reply"]:
    last = state["messages"][-1]
    if not last.tool_calls or state["steps"] >= 8:           # answer, or the step limit
        return "reply"
    needs_person = any(c["name"] == "refund_item" and c["args"]["amount"] > 500
                       for c in last.tool_calls)
    return "approve" if needs_person else "tools"

graph = StateGraph(State)
graph.add_node("route", route)
graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
graph.add_node("approve", approve)               # defined in the approval card
graph.add_node("reply", reply)

graph.add_edge(START, "route")
graph.add_conditional_edges("route", lambda s: "reply" if s["route"] == "other" else "agent",
                            ["agent", "reply"])
graph.add_conditional_edges("agent", after_agent, ["tools", "approve", "reply"])
graph.add_edge("tools", "agent")                 # the agent loop
graph.add_edge("reply", END)

from langgraph.checkpoint.memory import InMemorySaver
# Production: a database-backed saver, e.g. PostgresSaver, so any server can resume any thread.

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

config = {"configurable": {"thread_id": "ticket-4411", "customer_id": 412}}

app.invoke({"messages": [("user", "Was order 88213 late?")]}, config)
app.invoke({"messages": [("user", "Then refund the delivery fee.")]}, config)
# The second call sees the first exchange: both ran on thread ticket-4411.

for step in app.get_state_history(config):        # every checkpoint, newest first
    print(step.next, len(step.values["messages"]))

from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore      # production: a database-backed store

def recall(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    namespace = ("customers", str(config["configurable"]["customer_id"]))
    found = store.search(namespace, query=state["messages"][-1].content, limit=3)
    facts = "\n".join(f"- {item.value['fact']}" for item in found)
    return {"messages": [("system", f"Known about this customer:\n{facts}")]} if facts else {}

def remember(state: State, config: RunnableConfig, *, store: BaseStore) -> dict:
    namespace = ("customers", str(config["configurable"]["customer_id"]))
    for fact in extract_lasting_facts(state["messages"]):     # a small-model call
        store.put(namespace, key=fact.key, value={"fact": fact.text})   # same key overwrites
    return {}

app = graph.compile(checkpointer=InMemorySaver(), store=InMemoryStore())

from langchain_core.messages import ToolMessage
from langgraph.types import Command, interrupt

def approve(state: State) -> Command:
    call = next(c for c in state["messages"][-1].tool_calls if c["name"] == "refund_item")
    decision = interrupt({                            # pauses here; nothing above has side effects
        "question": "Approve this refund?",
        "order_id": call["args"]["order_id"],
        "amount": call["args"]["amount"],
    })
    if decision["approved"]:
        return Command(goto="tools")                  # let the tool node run the refund
    declined = ToolMessage(tool_call_id=call["id"],
                           content=f"Declined by a supervisor: {decision['reason']}. Do not retry.")
    return Command(goto="agent", update={"messages": [declined]})

# 1. The customer's request runs until the approval node pauses it.
result = app.invoke({"messages": [("user", "Refund ₹1,800 for order 88213")]}, config)
print(result["__interrupt__"])                        # payload for the approval screen

# 2. Later, the supervisor's decision resumes the same thread.
app.invoke(Command(resume={"approved": False, "reason": "order was on time"}), config)

from langchain.agents import create_agent

policy_agent = create_agent(
    model="anthropic:claude-haiku-4-5",
    tools=[search_help_centre],
    system_prompt="Find the exact policy for the question. Quote it and give its title.",
)
orders_agent = create_agent(
    model="anthropic:claude-haiku-4-5",
    tools=[get_order, get_recent_orders],
    system_prompt="Answer questions about this customer's orders from the order records only.",
)

@tool
def ask_policy_specialist(brief: str) -> str:
    """Find what Kirana Kart's policy says. Give a self-contained brief: the
    situation and the exact question. The specialist sees nothing else."""
    result = policy_agent.invoke({"messages": [("user", brief)]})     # fresh context
    return result["messages"][-1].content                              # only the answer returns

@tool
def ask_orders_specialist(brief: str, config: RunnableConfig) -> str:
    """Look up facts about the customer's orders. Include the order id if known."""
    result = orders_agent.invoke({"messages": [("user", brief)]}, config)
    return result["messages"][-1].content

supervisor = create_agent(
    model="anthropic:claude-opus-5-5",
    tools=[ask_policy_specialist, ask_orders_specialist, refund_item],
    system_prompt="Resolve the customer's request. Use the specialists for facts; "
                  "brief them fully. Decide and act yourself.",
    checkpointer=InMemorySaver(),
)

print(app.get_graph().draw_mermaid())                 # paste into any Mermaid viewer

for update in app.stream({"messages": [("user", "Refund ₹300 for order 88213")]},
                         config, stream_mode="updates"):
    print(update)                                     # {'route': {...}} then {'agent': {...}} ...

def test_large_refund_needs_approval():
    last = ai_message(tool_calls=[{"name": "refund_item", "id": "1",
                                   "args": {"order_id": 88213, "amount": 1800}}])
    assert after_agent({"messages": [last], "route": "refund", "steps": 1}) == "approve"

def test_small_refund_does_not():
    last = ai_message(tool_calls=[{"name": "refund_item", "id": "1",
                                   "args": {"order_id": 88213, "amount": 300}}])
    assert after_agent({"messages": [last], "route": "refund", "steps": 1}) == "tools"

Q: How would you structure an agent in LangGraph?

Define a typed state with a message list and counters; nodes for the model call, tool execution and any fixed steps; conditional edges for tool use, approval and finishing; an edge from tools back to the model as the loop; compile with a checkpointer.

Q: Where does the agent loop go in a graph?

It is a cycle: the model node, a conditional edge to the tool node when tools are requested, and an edge back. The exit is the conditional edge choosing the end when there are no tool calls or a limit is reached.

Q: How is memory handled?

Short-term memory is the thread's checkpointed state, continued by reusing the thread id. Long-term memory is a separate store, namespaced per user, read and written by nodes.

Q: How do you add human approval?

A node calls interrupt with what needs deciding; the state is saved and the call returns. The decision arrives as a resume command on the same thread. The trigger is code on the tool call's arguments, and the node has no side effects before the interrupt.

Q: How do you build a supervisor with workers?

Create each worker as its own agent, wrap each in a tool that takes a brief and returns only the final answer, and give the supervisor those tools. Workers run in fresh contexts and can use smaller models.

Q: How do you pass the user's identity to tools?

Through the run configuration set by the server, read inside the tool. It is never a tool argument, so the model cannot supply or change it.

Q: How do you test a graph-based agent?

Unit-test nodes and routing functions with hand-made states and no model; run end-to-end cases against fake tools asserting on end state and path; keep the evaluation set independent of the framework.

Q: What did the framework not do for you?

Tool design, business rules, what to remember, context compaction, evaluation and security. It supplied structure and durability, not judgement.

Your task. Build the graph from this module end to end and prove each capability with a test or a demonstration.

Steps

1. Install pinned versions and build the state, the tools and the agent-tools cycle. Run the late-delivery task.
2. Add the routing node and send 'thank you' down the short path.
3. Compile with a checkpointer. Show a second message on the same thread sees the first, and a new thread does not.
4. Add the approval node. Show a ₹300 refund runs straight through and a ₹1,800 one pauses; resume it with approve and with decline.
5. Add a store, a recall node and a remember node. Tell the agent a preference in one thread and show it is used in another thread for the same customer and not for a different customer.
6. Wrap a policy specialist and an orders specialist as tools for a supervisor.
7. Write unit tests for after_agent and draw the graph.

You are done when

• The generated diagram matches the six nodes in this module.
• Killing the process during a tool call and re-invoking the same thread continues the run.
• Routing and approval are covered by tests that call no model.
• No tool takes the customer id as a model-supplied argument.

Stretch. Rebuild the triage and hand-off part in a hand-off-based SDK and list what was easier and what was harder.