Lesson 69 — Beginner

Building Agents with LangGraph

LangGraph ek low-level orchestration framework hai jo stateful, long-running agents ko explicit graph ke roop mein define karta hai — nodes (kaam), edges (raste), state (shared memory). Is lesson mein tool-using agent ka graph banayenge, conditional edges se loop chalayenge, checkpointer se run resume karenge aur interrupt se human approval lagayenge.

LangGraph kis liye — state, nodes, edges

                 +-------+
   START ------->| agent |<-------------------+
                 +-------+                    |
                    |  |  tool_calls hai?     |
                    |  +-------> tools -------+
                    |               (no tool) |
                    v                          |
                   END                        |

state = {messages: [...] }  -- nodes ise padhte/likhte hain, reducers merge karte hain

Graph banana — nodes, edges, conditional routing

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

class State(TypedDict):
    messages: Annotated[list, add_messages]   # reducer — naye messages append

def call_agent(state):
    reply = llm.bind_tools(TOOLS).invoke(state["messages"])
    return {"messages": [reply]}

def call_tools(state):
    outs = [run_tool(c) for c in state["messages"][-1].tool_calls]
    return {"messages": outs}                  # tool results state mein jodo

def route(state):
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else "end"

g = StateGraph(State)
g.add_node("agent", call_agent)
g.add_node("tools", call_tools)
g.add_edge(START, "agent")
g.add_conditional_edges("agent", route, {"tools": "tools", "end": END})
g.add_edge("tools", "agent")                   # yahi loop hai — wapas agent
app = g.compile()

• Nodes: plain functions jo state lekar partial update return karein. Model call, tool execution, processing, decision — sab isi shakal mein.
• Conditional edges: ek function jo next node ka naam decide kare — route() ki tarah. Yahi hai "agla step kaun tay karta hai" wala point.
• Loops: tools se wapas agent ki edge — retry, iteration, multi-turn tool use isi se bante hain. Cycle ka matlab infinite nahi — stopping condition route() mein hi hai.
• Reducers: Annotated[list, add_messages] jaise — do nodes same state par likhein toh merge kaise hoga, yehi decide karta hai.

Checkpointing aur resume

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("agent_state.db")
app = g.compile(checkpointer=checkpointer)

cfg = {"configurable": {"thread_id": "refund-case-42"}}   # run ki identity
app.invoke({"messages": [user_msg("Mera refund kahan hai?")]}, cfg)

# beech mein process crash ho gaya?  Wahi thread_id dobara chalao —
# checkpointer har step ka state save karta hai, wahi jagah se aage badhega.
# pura restart nahi — yahi hai resume ka mechanism.

• Checkpointer = state ka har step snapshot (SQLite/Postgres/memory) — thread_id key hai, wahi purana db session-id wala concept.
• Resume: wahi thread_id dobara invoke — graph us step se aage badhta hai jahan tha.
• Human approval ke liye bhi isi ki zarurat: interrupt thamta hai, approval ke baad wahi thread resume hota hai.
• State inspect: app.get_state(cfg) se dekho current values aur pending steps — path kyun liya, yahin se pata chalta hai.

Human-in-the-loop interrupt

from langgraph.types import interrupt, Command

def approval(state):
    decision = interrupt({
        "question": "Refund approve karein?",
        "order_id": state["order_id"],
        "amount": state["amount"],
    })
    return {"approved": bool(decision)}

# interrupt() yahin graph ko rok deta hai — state checkpoint mein save,
# execution user ke paas chala jaata hai.

# user ne jawab de diya:
app.invoke(Command(resume=True), cfg)     # ya Command(resume=False)
# graph thama hua node wahi se dobara chalega

Yehi hai consequential tools ka graph-version gate (L5.3): action se pehle ruko, approval lo, phir aage. Interrupt ke bina approval sirf prompt mein likhi hui umeed hai — model usey kabhi kabhi todta hai.

State se debug aur kab graph, kab plain loop

• Debug flow: thread_id lo → get_state() → messages/decisions dekho → pata chalega kaunsa node kaunsa raasta kyun liya. Bina state dump ke andaza mat lagao.
• Graph chahiye jab: explicit branches ho, fan-out (ek node se kai parallel) ho, HITL chahiye, persistence/rerun chahiye, ya review mein graph dikhana ho.
• Plain loop kaafi jab: single agent sirf tools call karta hai — while loop hi saaf hai. Loop mein daal kar graph banana over-engineering hai.
• Migration: loop likh chuke ho? Nodes mein todo functions, route() mein tumhara if-else — wahi code graph ban jaata hai, logic dobara nahi.

Exercise

Common mistakes

• Simple loop ko graph mein over-engineer karna: Ek tool-loop ke liye while loop hi saaf hai. Graph tab banao jab branches, persistence, fan-out ya HITL genuinely chahiye.
• State ko manually mutate karna: Reducer (add_messages jaisa) ke bina do nodes ke writes takrayenge aur updates khayab ho jaayenge. State update partial return karo, mutation nahi.
• thread_id ke bina checkpointer: Har invoke naya run ban jaata hai — resume kaam nahi karega. Config stable thread_id har call par pass karo.
• Tool results ko state format na dena: Graph mein sab kuch messages channel par travel karta hai — tool result ko bhi message ki shakal do, warna agla node usey padh hi nahi payega.

sadak:   A -> B -> C (jab tak fuel hai, wahi raasta)
junction: START -signal-> agent -signal-> tools / END
                  ^_____________________________|  (loop track)
resume:  crash ke baad bhi stationmaster ke record se wahi gaadi wahi se chale