Langgraph Python Expert

LangGraph Python Expert

Comprehensive expert for building sophisticated stateful applications with LangGraph, focusing on production-ready workflows, state management, and agent orchestration.

📚 Official Source Documentation

This skill includes access to the official LangGraph source code through the source/langgraph/ directory (managed as git submodule with sparse-checkout), which contains:

• Core Libraries: libs/langgraph/, libs/prebuilt/, libs/checkpoint*/
• Official Examples: examples/ - Up-to-date examples and tutorials
• Complete Documentation: docs/docs/ - Latest documentation and API references

Source Structure (66MB with sparse-checkout)

source/langgraph/
├── libs/
│   ├── langgraph/          # Core StateGraph, nodes, edges
│   ├── prebuilt/           # create_react_agent, ToolNode
│   ├── checkpoint/         # Base checkpoint classes
│   ├── checkpoint-sqlite/  # SQLite persistence
│   └── checkpoint-postgres/# PostgreSQL persistence
├── examples/               # Official examples and tutorials
├── docs/docs/              # Documentation (concepts, how-tos, reference)
├── README.md               # Project overview
├── CLAUDE.md               # Claude Code instructions
└── AGENTS.md               # Agent development guide

Updating Source Code

cd source/langgraph
git pull origin main

For detailed structure, see SOURCE_STRUCTURE.md.

Quick Start

Installation

pip install langgraph langchain-core langchain-openai

Basic Concepts

StateGraph: The core component for building workflows with state persistence Nodes: Functions that process the state and return updates Edges: Define the flow between nodes (conditional or direct) State: TypedDict that holds conversation/application state Persistence: Checkpointing for memory and conversation history

Core Components

1. State Definition

from typing import TypedDict, List, Optional
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: List[BaseMessage]
    current_user: Optional[str]
    step_count: int
    requires_action: bool

2. Node Functions

from langchain_core.messages import HumanMessage, AIMessage

def llm_node(state: AgentState) -> AgentState:
    """Process messages with LLM and return updated state"""
    messages = state["messages"]
    response = llm.invoke(messages)
    return {
        "messages": messages + [response],
        "step_count": state["step_count"] + 1
    }

def router_node(state: AgentState) -> str:
    """Decide next node based on state"""
    last_message = state["messages"][-1]
    if "tool_call" in last_message.additional_kwargs:
        return "tool_executor"
    return "end"

3. Graph Construction

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# Create graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("agent", agent_node)
workflow.add_node("tool_executor", tool_node)
workflow.add_node("router", router_node)

# Add edges
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    router_node,
    {
        "tool_executor": "tool_executor",
        "end": END
    }
)
workflow.add_edge("tool_executor", "agent")

# Memory
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

4. Running the Graph

from langchain_core.messages import HumanMessage

# Run with thread_id for conversation memory
config = {"configurable": {"thread_id": "conversation-1"}}
response = app.invoke(
    {"messages": [HumanMessage(content="Hello!")]},
    config=config
)

5. Prebuilt Agents

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

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

response = agent.invoke(
    {"messages": [HumanMessage(content="What's 2+2?")]}
)

Testing and Debugging

Testing Individual Nodes

def test_llm_node():
    state = {
        "messages": [HumanMessage(content="Test")],
        "step_count": 0
    }
    result = llm_node(state)
    assert len(result["messages"]) == 2
    assert result["step_count"] == 1

Graph Visualization

from IPython.display import Image, display

try:
    display(Image(app.get_graph().draw_mermaid_png()))
except Exception:
    pass  # Requires graphviz

Debug Mode

from langgraph.types import interrupt

def debug_node(state):
    result = process(state)
    interrupt({"debug": result})  # Pause execution
    return result

Troubleshooting

Common Issues:

• State not updating: Ensure nodes return state updates, not mutate state directly
• Memory not persisting: Check thread_id is consistent across invocations
• Conditional edges not working: Verify router function returns exact node names
• Checkpoint errors: Ensure checkpoint saver is initialized before compile()

Debug Techniques:

• Use print() statements in nodes to trace state
• Check state type hints match actual data
• Verify tool schemas are properly defined
• Test nodes individually before integrating

Requirements

Essential Packages

# Core
pip install langgraph langchain-core

# LLM Integrations
pip install langchain-openai    # OpenAI
pip install langchain-anthropic # Anthropic

# Optional Checkpointers
pip install langgraph-checkpoint-sqlite
pip install langgraph-checkpoint-postgres

Optional Dependencies

# Visualization
pip install graphviz matplotlib

# Async Support
pip install aiohttp

# Development
pip install pytest pytest-asyncio

Quick Reference

Essential APIs

• StateGraph(state_schema) - Create graph with typed state
• add_node(name, func) - Register node function
• set_entry_point(node) - Set starting node
• add_edge(from, to) - Add direct connection
• add_conditional_edges(node, router, mapping) - Conditional routing
• compile(checkpointer=...) - Build executable graph
• invoke(state, config) - Run graph synchronously
• stream(state, config) - Stream outputs
• astream(state, config) - Async streaming

Common Patterns

• Agent Loop: agent → router → [tools/end] → agent
• Router: Return string key from conditional edge mapping
• State Updates: Return partial dict from node functions
• Memory: Use thread_id in config for persistence

Graph Methods

# Execution
app.invoke(state, config)
app.stream(state, config, stream_mode="values")
app.batch([state1, state2], config)

# Inspection
app.get_graph()
app.get_graph().print_ascii()

See references/ADVANCED.md for advanced patterns including human-in-the-loop, async operations, and multi-agent systems. See references/DEPLOYMENT.md for production deployment, monitoring, and optimization strategies.