corpus/langgraph_docs.md

LangGraph — Complete Documentation Reference

Source: https://langchain-ai.github.io/langgraph/

---

What is LangGraph?

LangGraph is a low-level framework for building stateful, multi-actor applications with LLMs. It extends LangChain by adding support for stateful, cyclic computation — essential for building agents with loops, retries, and branching logic.

LangGraph is used by companies including Klarna, Uber, JP Morgan, and LinkedIn for production agent systems.

Core design goals

• Stateful: Maintain state across multiple LLM calls.
• Cyclic: Support loops and retries (unlike simple DAGs).
• Multi-actor: Coordinate multiple agents or sub-graphs.
• Controllable: Fine-grained control over agent behavior.
• Streaming: Native support for token streaming.
• Persistence: Built-in checkpointing and memory.

---

Installation

pip install langgraph
pip install langchain-openai  # or langchain-google-genai, etc.

---

Core Concepts

StateGraph

StateGraph is the primary graph type in LangGraph. It maintains a State dict that is passed between nodes and updated at each step.

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # Messages accumulate
    question: str
    answer: str | None

Nodes

Nodes are Python functions (sync or async) that receive the current State and return a partial update:

def my_node(state: AgentState) -> dict:
    # Do something with state
    return {"answer": "computed result"}  # Partial update to state

Edges

Edges define the flow between nodes:

# Simple edge: always go from A to B
graph.add_edge("node_a", "node_b")

# Conditional edge: choose next node based on state
graph.add_conditional_edges(
    "node_a",
    route_function,   # Returns the name of the next node
    {
        "path_1": "node_b",
        "path_2": "node_c",
        "end": END
    }
)

---

Building a Simple Graph

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

class State(TypedDict):
    input: str
    output: str

def process(state: State) -> dict:
    return {"output": f"Processed: {state['input']}"}

def check(state: State) -> str:
    if "error" in state["output"]:
        return "retry"
    return "done"

# Build the graph
graph_builder = StateGraph(State)
graph_builder.add_node("process", process)
graph_builder.set_entry_point("process")
graph_builder.add_conditional_edges("process", check, {
    "retry": "process",
    "done": END
})

graph = graph_builder.compile()

# Invoke
result = graph.invoke({"input": "hello", "output": ""})

---

State Management

TypedDict State

from typing import TypedDict, Annotated
import operator

class RAGState(TypedDict):
    question: str                          # Simple field — last write wins
    documents: list[str]                   # Simple field — last write wins
    messages: Annotated[list, operator.add] # Reducer — messages accumulate
    retry_count: int
    generation: str | None

Annotated fields with reducers

When multiple nodes update the same field, use Annotated to define a reducer:

• Annotated[list, operator.add] — appends new items to the list
• Annotated[int, max] — keeps the maximum value
• Custom reducer functions for complex merge logic

---

Conditional Edges and Routing

def grade_documents(state: RAGState) -> str:
    """Route based on document quality."""
    relevant = [d for d in state["documents"] if d["grade"] == "relevant"]
    
    if not relevant:
        return "rewrite"     # No relevant docs — rewrite query
    elif len(relevant) >= 3:
        return "generate"    # Enough context — generate answer
    else:
        return "web_search"  # Some context — augment with web search

graph.add_conditional_edges(
    "grade_documents",
    grade_documents,
    {
        "rewrite": "rewrite_query",
        "generate": "generate",
        "web_search": "web_search"
    }
)

---

Checkpointing and Persistence

LangGraph supports persistent state via checkpointers:

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.memory import MemorySaver

# In-memory (for testing)
memory = MemorySaver()

# SQLite (persistent)
with SqliteSaver.from_conn_string("./checkpoints.db") as checkpointer:
    graph = graph_builder.compile(checkpointer=checkpointer)

# Invoke with a thread_id (enables persistence and resumability)
config = {"configurable": {"thread_id": "user-123"}}
result = graph.invoke({"question": "What is RAG?"}, config=config)

---

Streaming

# Stream individual node outputs
for chunk in graph.stream({"question": "What is FastAPI?"}, config=config):
    for node_name, node_output in chunk.items():
        print(f"Node: {node_name}")
        print(f"Output: {node_output}")

# Stream LLM tokens (requires astream_events)
async for event in graph.astream_events({"question": "..."}, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)

---

Human-in-the-Loop

LangGraph supports pausing execution for human approval:

from langgraph.graph import StateGraph, END, interrupt

def human_approval(state: State) -> dict:
    # Pause and wait for human input
    decision = interrupt("Do you approve this action? (yes/no)")
    return {"approved": decision == "yes"}

graph_builder.add_node("human_approval", human_approval)

---

Multi-Agent Systems

LangGraph supports multiple agents collaborating:

from langgraph.graph import StateGraph

# Each agent is a node in the parent graph
def researcher_agent(state):
    # Agent 1: searches for information
    ...

def writer_agent(state):
    # Agent 2: writes based on research
    ...

def supervisor(state) -> str:
    # Decides which agent to call next
    ...

main_graph = StateGraph(State)
main_graph.add_node("researcher", researcher_agent)
main_graph.add_node("writer", writer_agent)
main_graph.add_conditional_edges("supervisor", supervisor, {...})

---

Prebuilt Agents

LangGraph includes prebuilt agent architectures:

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, tools=[search])

result = agent.invoke({"messages": [("user", "What is the weather in Paris?")]})

---

RAG with LangGraph — Corrective RAG Pattern

from langgraph.graph import StateGraph, END

# Nodes
graph.add_node("retrieve", retrieve_documents)
graph.add_node("grade_documents", grade_documents)
graph.add_node("generate", generate_answer)
graph.add_node("rewrite_query", rewrite_query)
graph.add_node("web_search", web_search)

# Edges
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "grade_documents")
graph.add_conditional_edges(
    "grade_documents",
    decide_next_step,
    {
        "generate": "generate",
        "rewrite": "rewrite_query",
        "web_search": "web_search",
    }
)
graph.add_edge("rewrite_query", "retrieve")   # Loop back
graph.add_edge("web_search", "generate")
graph.add_edge("generate", END)

---

LangGraph vs Traditional LangChain Chains

Feature	LangChain Chains	LangGraph
Execution flow	Linear DAG	Cyclic graph
State management	Limited	Full TypedDict state
Loops/retries	Not supported	Native
Branching	Limited	Conditional edges
Human-in-loop	Not supported	Built-in interrupt
Streaming	Partial	Full token streaming
Persistence	External	Built-in checkpointing
Debugging	Limited	LangSmith integration

---

LangGraph Key Components Summary

Component	Description
StateGraph	Main graph class with state management
TypedDict	Define the state schema
Annotated	Add reducers for accumulating state fields
add_node()	Register a node function
add_edge()	Define deterministic transitions
add_conditional_edges()	Define dynamic routing
set_entry_point()	Set the starting node
compile()	Build the executable graph
invoke()	Run graph synchronously
ainvoke()	Run graph asynchronously
stream()	Stream node outputs
MemorySaver	In-memory checkpointing
SqliteSaver	SQLite-based persistence
END	Special node indicating graph completion
interrupt()	Pause for human-in-the-loop