# LangGraph

Production-grade stateful graphs for agents. Built on LangChain, used by Replit, Uber, Elastic, Cognition.

---

## Core Concepts

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # Auto-merge
    user_input: str
    context: dict
    iterations: int

graph = StateGraph(AgentState)

# Add nodes
graph.add_node("reasoning", reasoning_node)
graph.add_node("tools", tool_node)

# Add edges
graph.add_edge(START, "reasoning")
graph.add_conditional_edges("reasoning", should_continue)
graph.add_edge("tools", "reasoning")

# Compile
app = graph.compile(checkpointer=MemorySaver())
```

### Key Components
| Component | Purpose |
|---|---|
| **State** | TypedDict with reducers (merge strategies) |
| **Nodes** | Functions: `state -> partial_state` |
| **Edges** | Fixed: `A -> B` |
| **Conditional Edges** | `state -> next_node` (dynamic routing) |
| **Checkpointer** | Persistence: `MemorySaver`, `SqliteSaver`, `PostgresSaver` |
| **Interrupt** | Human-in-the-loop: `interrupt()` + `resume()` |

---

## State Design

### Reducers (Merge Strategies)
```python
from typing import Annotated
import operator

class State(TypedDict):
    # Append (default for lists)
    messages: Annotated[list[BaseMessage], operator.add]
    
    # Replace (last write wins)
    user_input: str
    
    # Custom reducer
    context: Annotated[dict, lambda old, new: {**old, **new}]
    
    # Accumulate (numeric)
    tokens_used: Annotated[int, operator.add]
```

### Best Practices
- **Minimal state** — Only what's needed for decisions
- **Immutable updates** — Return partial dict, not mutated state
- **Serializable** — JSON-compatible for checkpointing
- **Versioned** — Add `schema_version` for migrations

---

## Control Flow Patterns

### 1. Sequential
```python
graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", "node_c")
```

### 2. Branching (Conditional)
```python
def route(state) -> str:
    if state["needs_tool"]:
        return "tools"
    return "final"

graph.add_conditional_edges("reasoning", route, {
    "tools": "tools",
    "final": "final_answer"
})
```

### 3. Parallel (Fan-out/Fan-in)
```python
# Fan-out: single node -> multiple parallel
graph.add_edge("start", "parallel_a")
graph.add_edge("start", "parallel_b")

# Fan-in: multiple -> single (using reducer)
# Both write to same state key with operator.add
graph.add_edge("parallel_a", "join")
graph.add_edge("parallel_b", "join")
```

### 4. Loops
```python
def should_continue(state) -> str:
    if state["iterations"] >= 5:
        return "final"
    return "reasoning"

graph.add_conditional_edges("tools", should_continue)
```

### 5. Dynamic Routing (LLM-based)
```python
def router(state) -> str:
    decision = llm.invoke(ROUTER_PROMPT.format(state=state))
    return parse_action(decision)

graph.add_conditional_edges("reasoning", router)
```

---

## Human-in-the-Loop (Interrupt)

### Basic Pattern
```python
from langgraph.graph import interrupt

def human_review(state):
    # This PAUSES execution
    feedback = interrupt({
        "question": "Approve this action?",
        "context": state["proposed_action"]
    })
    # Execution resumes HERE after resume()
    return {"human_feedback": feedback}

graph.add_node("human_review", human_review)
```

### Resume
```python
# From client
config = {"configurable": {"thread_id": "123"}}
app.invoke(None, config=config)  # Runs until interrupt

# Human provides feedback
app.resume(config, {"approved": True, "notes": "Looks good"})
```

### Approval UI Integration
```python
# Slack bot example
@app.route("/slack/interactive", methods=["POST"])
def slack_interactive():
    payload = json.loads(request.form["payload"])
    thread_id = payload["actions"][0]["value"]
    approved = payload["actions"][0]["action_id"] == "approve"
    
    # Resume the graph
    app.resume(
        {"configurable": {"thread_id": thread_id}},
        {"approved": approved}
    )
    return "", 200
```

---

## Checkpointing & Persistence

### Checkpointer Options
| Checkpointer | Backend | Use Case |
|---|---|---|
| `MemorySaver()` | In-memory | Dev, testing |
| `SqliteSaver("checkpoints.db")` | SQLite | Single-node, dev/prod |
| `PostgresSaver(pool)` | PostgreSQL | Production, multi-node |
| `AsyncPostgresSaver(pool)` | PostgreSQL (async) | High-throughput async |

### Configuration
```python
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    conninfo="postgresql://user:pass@host/db",
    max_size=20,
    kwargs={"autocommit": True}
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()  # Creates tables

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

### Thread Management
```python
# Each conversation = thread
config = {"configurable": {"thread_id": "user-123-session-456"}}

# Resume from checkpoint
state = app.get_state(config)
print(state.values)  # Current state
print(state.next)    # Next node to execute
print(state.tasks)   # Pending tasks

# Time travel
history = list(app.get_state_history(config))
app.update_state(config, {"messages": [...]} )  # Fork from past
```

### Migration
```python
# Add schema_version to state
class State(TypedDict):
    schema_version: int
    # ... other fields

def migrate_state(state):
    if state.get("schema_version", 0) < 2:
        state["new_field"] = default_value
        state["schema_version"] = 2
    return state

# Apply on resume
def load_state(config):
    state = checkpointer.get(config)
    return migrate_state(state)
```

---

## Subgraphs (Multi-Agent)

### Subgraph as Node
```python
# Sub-agent graph
subgraph = StateGraph(SubState)
subgraph.add_node("researcher", researcher)
subgraph.add_node("writer", writer)
subgraph.add_edge(START, "researcher")
subgraph.add_edge("researcher", "writer")
subgraph.add_edge("writer", END)
compiled_sub = subgraph.compile()

# Main graph uses subgraph as node
main = StateGraph(State)
main.add_node("research_team", compiled_sub)  # Treated as single node
main.add_edge(START, "research_team")
```

### Inter-Agent Communication (send)
```python
from langgraph.graph import send

def coordinator(state):
    # Send to researcher subgraph
    send("researcher", {"task": "Find X", "thread_id": "research-1"})
    # Send to coder subgraph
    send("coder", {"task": "Implement Y", "thread_id": "code-1"})
    return {"status": "delegated"}

# Subgraphs receive via special input
def researcher(state):
    task = state.get("task")  # From send()
    return {"result": do_research(task)}
```

---

## Streaming

### Token Streaming
```python
async for chunk in app.astream({"input": "Hello"}, config, stream_mode="messages"):
    if chunk[0].content:
        print(chunk[0].content, end="", flush=True)
```

### Stream Modes
| Mode | Yields |
|---|---|
| `"values"` | Full state after each node |
| `"updates"` | Node name + partial state |
| `"messages"` | Individual message chunks |
| `"debug"` | Detailed execution trace |
| `"custom"` | Custom from `stream_mode="custom"` in node |

### Custom Streaming
```python
def streaming_node(state):
    for token in llm.stream(state["prompt"]):
        yield {"token": token}  # Custom stream mode
```

---

## Observability (LangSmith)

### Automatic Tracing
```python
import langsmith
client = langsmith.Client()

# Automatic with LangGraph
app = graph.compile(checkpointer=checkpointer)
# Traces automatically sent if LANGSMITH_API_KEY set
```

### Custom Metadata
```python
config = {
    "configurable": {"thread_id": "123"},
    "metadata": {
        "user_id": "user-123",
        "session_type": "coding",
        "model": "gpt-4o"
    },
    "tags": ["production", "v2.1"]
}
```

### Evaluation Integration
```python
from langsmith.evaluation import evaluate

def target(inputs):
    return app.invoke(inputs, config={"configurable": {"thread_id": "eval"}})

results = evaluate(
    target,
    data="my-dataset",
    evaluators=[accuracy, latency, cost],
    experiment_prefix="agent-v2"
)
```

---

## Deployment

### LangGraph Platform (Managed)
```yaml
# langgraph.yaml
dependencies: ["."]
graphs:
  my_agent: "./agent.py:graph"
env:
  - OPENAI_API_KEY
  - LANGSMITH_API_KEY
```

### Self-Hosted (Docker)
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]
```

### FastAPI Server
```python
from fastapi import FastAPI
from langgraph.pregel.remote import RemoteGraph

app = FastAPI()

# Expose compiled graph as API
graph = RemoteGraph("my_agent", url="http://langgraph-server:8000")

@app.post("/invoke")
async def invoke(input: dict, thread_id: str):
    return await graph.ainvoke(input, config={"configurable": {"thread_id": thread_id}})

@app.post("/stream")
async def stream(input: dict, thread_id: str):
    async for chunk in graph.astream(input, config={"configurable": {"thread_id": thread_id}}):
        yield chunk
```

---

## Testing

### Unit Testing Nodes
```python
def test_reasoning_node():
    state = {"messages": [HumanMessage("Hello")], "context": {}}
    result = reasoning_node(state)
    assert "messages" in result
    assert len(result["messages"]) > 0
```

### Integration Testing
```python
def test_full_graph():
    app = graph.compile(checkpointer=MemorySaver())
    config = {"configurable": {"thread_id": "test"}}
    result = app.invoke({"input": "Test question"}, config)
    assert "answer" in result
```

### Time-Travel Testing
```python
def test_replay():
    # Run to checkpoint
    app.invoke({"input": "Step 1"}, config)
    state = app.get_state(config)
    
    # Fork and test alternative
    app.update_state(config, {"messages": [AlternativeMessage()]})
    result = app.invoke(None, config)
    assert result["path"] == "alternative"
```

---

## Performance Tips

1. **Minimize state size** — Only essential data
2. **Batch LLM calls** — Use `abatch` for parallel
3. **Cache checkpointer** — Reuse connection pools
4. **Stream early** — `stream_mode="messages"` for UX
5. **Async everywhere** — `ainvoke`, `astream`, `abatch`

---

## Common Patterns

### ReAct in LangGraph
```python
def reasoning_node(state):
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state):
    tool_calls = state["messages"][-1].tool_calls
    results = []
    for tc in tool_calls:
        tool = tools_by_name[tc["name"]]
        results.append(tool.invoke(tc["args"]))
    return {"messages": results}

def should_continue(state):
    return "tools" if state["messages"][-1].tool_calls else "end"
```

### Plan-and-Execute
```python
def planner(state):
    plan = llm.invoke(PLAN_PROMPT.format(goal=state["goal"]))
    return {"plan": parse_plan(plan), "step": 0}

def executor(state):
    step = state["plan"][state["step"]]
    result = execute_step(step)
    return {"results": [result], "step": state["step"] + 1}

def should_replan(state):
    if state["step"] >= len(state["plan"]):
        return "final"
    if step_failed(state):
        return "replan"
    return "executor"
```

---

## Related Notes

- [[Frameworks/Framework-Comparison.md]] — vs other frameworks
- [[Architectures/Control-Flow.md]] — Patterns implemented
- [[Architectures/Human-in-the-Loop.md]] — `interrupt()` deep dive
- [[StateManagement/Checkpointing.md]] — Persistence patterns
- [[MultiAgent/Coordination-Patterns.md]] — Subgraphs, `send()`
- [[Production/Observability.md]] — LangSmith integration
- [[Production/Deployment-Architecture.md]] — Deployment patterns
- [[Evaluation/Agent-Benchmarks.md]] — Testing with LangGraph

---

*Sources: LangGraph docs (langchain-ai.github.io/langgraph), LangGraph Platform docs, LangSmith docs, GitHub langgraph repo, production case studies (Replit, Uber, Elastic, Cognition), LangChain blog, LangGraph Academy.*