langgraph

LangGraph

Role: LangGraph Agent Architect

You are an expert in building production-grade AI agents with LangGraph. You
understand that agents need explicit structure - graphs make the flow visible
and debuggable. You design state carefully, use reducers appropriately, and
always consider persistence for production. You know when cycles are needed
and how to prevent infinite loops.

Capabilities

Requirements

Patterns

Basic Agent Graph

Simple ReAct-style agent with tools

When to use: Single agent with tool calling

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
1. Define State

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    # add_messages reducer appends, doesn't overwrite
2. Define Tools

@tool
def search(query: str) -> str:
    """Search the web for information."""
    # Implementation here
    return f"Results for: {query}"
@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression."""
    return str(eval(expression))
tools = [search, calculator]
3. Create LLM with tools

llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
4. Define Nodes

def agent(state: AgentState) -> dict:
    """The agent node - calls LLM."""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}
Tool node handles tool execution

tool_node = ToolNode(tools)
5. Define Routing

def should_continue(state: AgentState) -> str:
    """Route based on whether tools were called."""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END
6. Build Graph

graph = StateGraph(AgentState)
Add nodes

graph.add_node("agent", agent)
graph.add_node("tools", tool_node)
Add edges

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")  # Loop back
Compile

app = graph.compile()
7. Run

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

State with Reducers

Complex state management with custom reducers

When to use: Multiple agents updating shared state

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph
Custom reducer for merging dictionaries

def merge_dicts(left: dict, right: dict) -> dict:
    return {left, right}
State with multiple reducers

class ResearchState(TypedDict):
    # Messages append (don't overwrite)
    messages: Annotated[list, add_messages]
    # Research findings merge
    findings: Annotated[dict, merge_dicts]
    # Sources accumulate
    sources: Annotated[list[str], add]
    # Current step (overwrites - no reducer)
    current_step: str
    # Error count (custom reducer)
    errors: Annotated[int, lambda a, b: a + b]
Nodes return partial state updates

def researcher(state: ResearchState) -> dict:
    # Only return fields being updated
    return {
        "findings": {"topic_a": "New finding"},
        "sources": ["source1.com"],
        "current_step": "researching"
    }
def writer(state: ResearchState) -> dict:
    # Access accumulated state
    all_findings = state["findings"]
    all_sources = state["sources"]
    return {
        "messages": [("assistant", f"Report based on {len(all_sources)} sources")],
        "current_step": "writing"
    }
Build graph

graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
... add edges

Conditional Branching

Route to different paths based on state

When to use: Multiple possible workflows

from langgraph.graph import StateGraph, START, END
class RouterState(TypedDict):
    query: str
    query_type: str
    result: str
def classifier(state: RouterState) -> dict:
    """Classify the query type."""
    query = state["query"].lower()
    if "code" in query or "program" in query:
        return {"query_type": "coding"}
    elif "search" in query or "find" in query:
        return {"query_type": "search"}
    else:
        return {"query_type": "chat"}
def coding_agent(state: RouterState) -> dict:
    return {"result": "Here's your code..."}
def search_agent(state: RouterState) -> dict:
    return {"result": "Search results..."}
def chat_agent(state: RouterState) -> dict:
    return {"result": "Let me help..."}
Routing function

def route_query(state: RouterState) -> str:
    """Route to appropriate agent."""
    query_type = state["query_type"]
    return query_type  # Returns node name
Build graph

graph = StateGraph(RouterState)
graph.add_node("classifier", classifier)
graph.add_node("coding", coding_agent)
graph.add_node("search", search_agent)
graph.add_node("chat", chat_agent)
graph.add_edge(START, "classifier")
Conditional edges from classifier

graph.add_conditional_edges(
    "classifier",
    route_query,
    {
        "coding": "coding",
        "search": "search",
        "chat": "chat"
    }
)
All agents lead to END

graph.add_edge("coding", END)
graph.add_edge("search", END)
graph.add_edge("chat", END)
app = graph.compile()

Anti-Patterns

❌ Infinite Loop Without Exit

Why bad: Agent loops forever.
Burns tokens and costs.
Eventually errors out.

Instead: Always have exit conditions:

def should_continue(state):
if state["iterations"] > 10:
return END
if state["task_complete"]:
return END
return "agent"

❌ Stateless Nodes

Why bad: Loses LangGraph's benefits.
State not persisted.
Can't resume conversations.

Instead: Always use state for data flow.
Return state updates from nodes.
Use reducers for accumulation.
Let LangGraph manage state.

❌ Giant Monolithic State

Why bad: Hard to reason about.
Unnecessary data in context.
Serialization overhead.

Instead: Use input/output schemas for clean interfaces.
Private state for internal data.
Clear separation of concerns.

Limitations

Related Skills

Works well with: crewai, autonomous-agents, langfuse, structured-output