LangGraph: Complete Interactive Notes

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

class State(TypedDict):                     # 1. the shared state every node reads and updates
    amount: int
    risk: str
    decision: str

def score(state: State):                    # 2. a node = a plain function: state in, partial update out
    return {"risk": "high" if state["amount"] > 50_000 else "low"}

def approve(state: State):
    return {"decision": f"approved ₹{state['amount']:,} ({state['risk']} risk)"}

g = StateGraph(State)
g.add_node("score", score)
g.add_node("approve", approve)
g.add_edge(START, "score")                  # 3. edges = the order of steps
g.add_edge("score", "approve")
g.add_edge("approve", END)
app = g.compile()                           # 4. compile -> a runnable app (invoke, stream, batch)

print(app.invoke({"amount": 20_000}))
print([l.strip() for l in app.get_graph().draw_mermaid().splitlines() if "-->" in l])   # diagram as Mermaid text

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

class State(TypedDict):
    amount: int
    risk: str
    decision: str

def score(s): return {"risk": "high" if s["amount"] > 50_000 else "low"}
def approve(s): return {"decision": "auto-approved"}
def review(s): return {"decision": "sent to a human reviewer"}

def route(s) -> Literal["approve", "review"]:   # a router returns the NAME of the next node
    return "review" if s["risk"] == "high" else "approve"

g = StateGraph(State)
for name, fn in [("score", score), ("approve", approve), ("review", review)]: g.add_node(name, fn)
g.add_edge(START, "score")
g.add_conditional_edges("score", route)        # branch!
g.add_edge("approve", END); g.add_edge("review", END)
app = g.compile()

for amt in [20_000, 80_000]:
    print(f"₹{amt:,} ->", app.invoke({"amount": amt})["decision"])

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

class State(TypedDict):
    query: str
    tries: int
    log: Annotated[list[str], operator.add]     # reducer: new items are APPENDED, not overwritten

KB = {"fixed deposit penalty": "Breaking an FD early reduces interest by 1%."}

def search(s):
    hit = KB.get(s["query"])
    return {"tries": s["tries"] + 1, "log": [f"try {s['tries'] + 1}: search '{s['query']}' -> {'found' if hit else 'nothing'}"]}

def rewrite(s):
    return {"query": s["query"].replace("FD", "fixed deposit"), "log": ["rewrote the query"]}

def answer(s):
    return {"log": ["answer: " + KB.get(s["query"], "Sorry, I could not find it.")]}

def next_step(s):
    if s["query"] in KB: return "answer"
    return "rewrite" if s["tries"] < 3 else "answer"

g = StateGraph(State)
for n, f in [("search", search), ("rewrite", rewrite), ("answer", answer)]: g.add_node(n, f)
g.add_edge(START, "search")
g.add_conditional_edges("search", next_step)
g.add_edge("rewrite", "search")                 # a CYCLE: go back and try again
g.add_edge("answer", END)
app = g.compile()

for line in app.invoke({"query": "FD penalty", "tries": 0, "log": []})["log"]: print(line)

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver   # use SqliteSaver / PostgresSaver in production

class State(TypedDict):
    items: Annotated[list[str], operator.add]

def add_item(s): return {}                               # the input itself is the update here
g = StateGraph(State); g.add_node("add", add_item); g.add_edge(START, "add"); g.add_edge("add", END)
app = g.compile(checkpointer=InMemorySaver())

cart = {"configurable": {"thread_id": "cart-42"}}
app.invoke({"items": ["gold loan form"]}, cart)
app.invoke({"items": ["KYC update"]}, cart)              # same thread: state continues
print("current state:", app.get_state(cart).values)

history = list(app.get_state_history(cart))              # every checkpoint is kept: "time travel"
print("checkpoints saved:", len(history))
print("oldest with items:", [h.values for h in history if h.values.get("items")][-1])

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    amount: int
    status: str

def ask_human(s):
    answer = interrupt({"question": f"Approve transfer of ₹{s['amount']:,}?"})   # pause here
    return {"status": "approved" if answer == "yes" else "rejected"}

def execute(s):
    return {"status": s["status"] + (" and sent" if s["status"] == "approved" else "")}

g = StateGraph(State)
g.add_node("ask_human", ask_human); g.add_node("execute", execute)
g.add_edge(START, "ask_human"); g.add_edge("ask_human", "execute"); g.add_edge("execute", END)
app = g.compile(checkpointer=InMemorySaver())             # interrupts need a checkpointer

cfg = {"configurable": {"thread_id": "tx-9"}}
first = app.invoke({"amount": 75_000, "status": ""}, cfg)
print("paused, asking:", first["__interrupt__"][0].value)
print("next node waiting:", app.get_state(cfg).next)

final = app.invoke(Command(resume="yes"), cfg)            # the human replies; the graph continues
print("result:", final["status"])

import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send

class State(TypedDict):
    docs: list[str]
    summaries: Annotated[list[str], operator.add]

class DocState(TypedDict):
    doc: str

def fan_out(s):                                  # Send = run one node per item, in parallel (map-reduce)
    return [Send("summarise", {"doc": d}) for d in s["docs"]]

def summarise(s: DocState):
    return {"summaries": [s["doc"].split(":")[0] + " ✔"]}

def combine(s):
    return {"summaries": [f"TOTAL {len(s['summaries'])} documents summarised"]}

g = StateGraph(State)
g.add_node("summarise", summarise); g.add_node("combine", combine)
g.add_conditional_edges(START, fan_out, ["summarise"])
g.add_edge("summarise", "combine"); g.add_edge("combine", END)
app = g.compile()

docs = ["KYC policy: ...", "FD policy: ...", "Loan policy: ..."]
for update in app.stream({"docs": docs, "summaries": []}, stream_mode="updates"):   # watch each step finish
    print(update)

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

class State(TypedDict):
    question: str
    trail: list[str]
    answer: str

def supervisor(s) -> Command[Literal["cards_agent", "loans_agent"]]:
    target = "cards_agent" if "card" in s["question"] else "loans_agent"
    return Command(goto=target, update={"trail": s["trail"] + [f"supervisor -> {target}"]})  # route AND update

def cards_agent(s):
    return {"trail": s["trail"] + ["cards_agent answered"], "answer": "Card blocked. A new one arrives in 5 days."}

def loans_agent(s):
    return {"trail": s["trail"] + ["loans_agent answered"], "answer": "Your EMI is due on the 5th."}

g = StateGraph(State)
for n, f in [("supervisor", supervisor), ("cards_agent", cards_agent), ("loans_agent", loans_agent)]: g.add_node(n, f)
g.add_edge(START, "supervisor"); g.add_edge("cards_agent", END); g.add_edge("loans_agent", END)
app = g.compile()

for q in ["Please block my debit card", "When is my home loan EMI due?"]:
    r = app.invoke({"question": q, "trail": [], "answer": ""})
    print(q, "\n   ", " | ".join(r["trail"]), "\n   ", r["answer"])

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool
from langchain_core.messages import AIMessage
from fake_llm import ScriptedModel

@tool
def fd_rate(years: int) -> str:
    """Return the fixed-deposit interest rate for a tenure in years."""
    return {1: "6.8%", 3: "7.1%", 5: "7.0%"}.get(years, "not offered")

model = ScriptedModel([AIMessage("", tool_calls=[{"name": "fd_rate", "args": {"years": 3}, "id": "t1"}]),
                       "The 3-year FD rate is 7.1%."]).bind_tools([fd_rate])

def call_model(state: MessagesState):              # MessagesState = {"messages": [...]} with add_messages
    return {"messages": [model.invoke(state["messages"])]}

g = StateGraph(MessagesState)
g.add_node("model", call_model)
g.add_node("tools", ToolNode([fd_rate]))           # runs whatever tools the model asked for
g.add_edge(START, "model")
g.add_conditional_edges("model", tools_condition)  # tool calls? -> "tools", else -> END
g.add_edge("tools", "model")                       # the ReAct loop
app = g.compile()

for m in app.invoke({"messages": [("user", "What is the 3-year FD rate?")]})["messages"]:
    print(f"{m.type:5s}", m.tool_calls if getattr(m, "tool_calls", None) else m.content)