LangGraph Agents Cheat Sheet

Installation & Setup

# Install LangGraph
pip install langgraph langchain langchain-openai

# Basic imports
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator

Define Agent State

class AgentState(TypedDict):
    # Messages list (accumulates conversation)
    messages: Annotated[list, operator.add]

    # Single values (gets overwritten)
    current_step: str
    tool_result: str

    # Counter (gets incremented)
    iteration_count: Annotated[int, operator.add]

# State is passed between nodes
# Annotations control how values merge

Create Basic Graph

from langgraph.graph import StateGraph, END

# Initialize graph with state type
workflow = StateGraph(AgentState)

# Define node functions
def call_model(state: AgentState):
    llm = ChatOpenAI(model="gpt-4")
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "continue"
    return "end"

# Add nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# Add edges
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile
app = workflow.compile()

Tool Integration

from langchain.tools import tool
from langgraph.prebuilt import ToolNode

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

@tool
def calculator(expression: str) -> str:
    """Calculate mathematical expressions."""
    return str(eval(expression))

# Create tool list
tools = [search_web, calculator]

# Bind tools to LLM
llm = ChatOpenAI(model="gpt-4")
llm_with_tools = llm.bind_tools(tools)

# Create tool execution node
tool_node = ToolNode(tools)

Running the Graph

# Simple invoke
result = app.invoke({
    "messages": [("user", "What's 25 * 4?")]
})

# Stream responses
for output in app.stream({
    "messages": [("user", "Search for AI news")]
}):
    print(output)

# Get final state
final_state = app.get_state()
print(final_state["messages"])

Conditional Routing

def route_by_intent(state: AgentState):
    """Route based on user intent."""
    last_msg = state["messages"][-1].content.lower()

    if "search" in last_msg or "find" in last_msg:
        return "search_node"
    elif "calculate" in last_msg or "math" in last_msg:
        return "calculator_node"
    else:
        return "general_node"

workflow.add_conditional_edges(
    "classifier",
    route_by_intent,
    {
        "search_node": "search",
        "calculator_node": "calc",
        "general_node": "agent"
    }
)

Human-in-the-Loop

from langgraph.checkpoint.sqlite import SqliteSaver

# Add memory/checkpointing
memory = SqliteSaver.from_conn_string(":memory:")

# Compile with checkpointer
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["human_approval"]
)

# Run until interrupt
config = {"configurable": {"thread_id": "1"}}
for output in app.stream(input_data, config):
    print(output)

# Resume after human input
app.update_state(
    config,
    {"approval": "approved"}
)

# Continue execution
for output in app.stream(None, config):
    print(output)

Multi-Agent Collaboration

class MultiAgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_agent: str

# Define specialized agents
def researcher_agent(state):
    llm = ChatOpenAI(model="gpt-4")
    prompt = "You are a research specialist..."
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "next_agent": "writer"
    }

def writer_agent(state):
    llm = ChatOpenAI(model="gpt-4")
    prompt = "You are a content writer..."
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],
        "next_agent": "reviewer"
    }

# Build collaboration graph
workflow = StateGraph(MultiAgentState)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("writer", writer_agent)
workflow.add_node("reviewer", reviewer_agent)

# Route between agents
workflow.add_conditional_edges(
    "researcher",
    lambda x: x["next_agent"]
)
workflow.add_conditional_edges(
    "writer",
    lambda x: x["next_agent"]
)

Persistence & Memory

from langgraph.checkpoint.sqlite import SqliteSaver

# SQLite checkpointer
memory = SqliteSaver.from_conn_string("agent_memory.db")

app = workflow.compile(checkpointer=memory)

# Each thread has isolated state
config1 = {"configurable": {"thread_id": "user-1"}}
config2 = {"configurable": {"thread_id": "user-2"}}

# Run separate conversations
app.invoke({"messages": [("user", "Hello")]}, config1)
app.invoke({"messages": [("user", "Hi")]}, config2)

# Get conversation history
history = app.get_state_history(config1)

Error Handling

def safe_tool_execution(state: AgentState):
    try:
        result = execute_tool(state["tool_input"])
        return {
            "tool_result": result,
            "error": None
        }
    except Exception as e:
        return {
            "tool_result": None,
            "error": str(e),
            "messages": [("system", f"Error: {e}")]
        }

def handle_error(state: AgentState):
    if state.get("error"):
        return "retry"
    return "success"

workflow.add_conditional_edges(
    "tool_executor",
    handle_error,
    {
        "retry": "agent",
        "success": "next_step"
    }
)

Streaming Tokens

# Stream individual tokens
async for event in app.astream_events(
    {"messages": [("user", "Tell me a story")]},
    version="v1"
):
    kind = event["event"]

    if kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="", flush=True)

Common Patterns

Debugging & Visualization

# Print graph structure
print(app.get_graph().draw_ascii())

# Get current state
state = app.get_state(config)
print("Current node:", state.next)
print("State values:", state.values)

# LangSmith tracing
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your_key"

# View traces in LangSmith dashboard

Best Practices