LangGraph Skill

Graph-based agent framework from LangChain

id: langgraph name: LangGraph version: 1.0.0 layer: 2 # Integration layer

description: | Expert in LangGraph - the production-grade framework for building stateful, multi-actor AI applications. Covers graph construction, state management, cycles and branches, persistence with checkpointers, human-in-the-loop patterns, and the ReAct agent pattern. Used in production at LinkedIn, Uber, and 400+ companies. This is LangChain's recommended approach for building agents.

owns:

• Graph construction (StateGraph)
• State management and reducers
• Node and edge definitions
• Conditional routing
• Checkpointers and persistence
• Human-in-the-loop patterns
• Tool integration
• Streaming and async execution

pairs_with:

• crewai
• autonomous-agents
• langfuse
• structured-output

requires:

• Python 3.9+
• langgraph package
• LLM API access (OpenAI, Anthropic, etc.)
• Understanding of graph concepts

ecosystem: primary: - LangGraph - LangChain - LangSmith (observability) common_integrations: - OpenAI / Anthropic / Google - Tavily (search) - SQLite / PostgreSQL (persistence) - Redis (state store) platforms: - Python applications - FastAPI / Flask backends - Cloud deployments

prerequisites:

• Python proficiency
• LLM API basics
• Async programming concepts
• Graph theory fundamentals

limits:

• Python-only (TypeScript in early stages)
• Learning curve for graph concepts
• State management complexity
• Debugging can be challenging

tags:

• langgraph
• langchain
• agents
• state-machine
• workflow
• graph
• ai-agents
• orchestration

triggers:

• "langgraph"
• "langchain agent"
• "stateful agent"
• "agent graph"
• "react agent"
• "agent workflow"
• "multi-step agent"

history:

• version: "1.0.0" date: "2025-01" changes: "Initial skill covering LangGraph patterns"

contrarian_insights:

• claim: "Use LangChain for everything" counter: "LangChain for chains, LangGraph for agents - they're different tools" evidence: "LangChain team explicitly says 'Use LangGraph for agents, not LangChain'"
• claim: "Simpler is always better - avoid graphs" counter: "Graphs make complex flows explicit and debuggable" evidence: "Teams report easier debugging when agent logic is visible in graph structure"
• claim: "State management is overhead" counter: "Automatic state management prevents bugs in multi-step agents" evidence: "Manual state passing leads to subtle bugs; LangGraph eliminates entire class of errors"

identity: role: LangGraph Agent Architect personality: | 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. expertise: - Graph topology design - State schema patterns - Conditional branching - Persistence strategies - Human-in-the-loop - Tool integration - Error handling and recovery

patterns:

• name: Basic Agent Graph description: Simple ReAct-style agent with tools when_to_use: Single agent with tool calling implementation: | 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?")] })

• name: State with Reducers description: Complex state management with custom reducers when_to_use: Multiple agents updating shared state implementation: | 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

• name: Conditional Branching description: Route to different paths based on state when_to_use: Multiple possible workflows implementation: | 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()

• name: Persistence with Checkpointer description: Save and resume agent state when_to_use: Multi-turn conversations, long-running agents implementation: | from langgraph.graph import StateGraph from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.checkpoint.postgres import PostgresSaver

  SQLite for development

  memory = SqliteSaver.from_conn_string(":memory:")

  Or persistent file

  memory = SqliteSaver.from_conn_string("agent_state.db")

  PostgreSQL for production

  memory = PostgresSaver.from_conn_string(DATABASE_URL)

  Compile with checkpointer

  app = graph.compile(checkpointer=memory)

  Run with thread_id for conversation continuity

  config = {"configurable": {"thread_id": "user-123-session-1"}}

  First message

  result1 = app.invoke( {"messages": [("user", "My name is Alice")]}, config=config )

  Second message - agent remembers context

  result2 = app.invoke( {"messages": [("user", "What's my name?")]}, config=config )

  Agent knows name is Alice!

  Get conversation history

  state = app.get_state(config) print(state.values["messages"])

  List all checkpoints

  for checkpoint in app.get_state_history(config): print(checkpoint.config, checkpoint.values)

• name: Human-in-the-Loop description: Pause for human approval before actions when_to_use: Sensitive operations, review before execution implementation: | from langgraph.graph import StateGraph, START, END

  class ApprovalState(TypedDict): messages: Annotated[list, add_messages] pending_action: dict | None approved: bool

  def agent(state: ApprovalState) -> dict: # Agent decides on action action = {"type": "send_email", "to": "user@example.com"} return { "pending_action": action, "messages": [("assistant", f"I want to: {action}")] }

  def execute_action(state: ApprovalState) -> dict: action = state["pending_action"] # Execute the approved action result = f"Executed: {action['type']}" return { "messages": [("assistant", result)], "pending_action": None }

  def should_execute(state: ApprovalState) -> str: if state.get("approved"): return "execute" return END # Wait for approval

  Build graph

  graph = StateGraph(ApprovalState) graph.add_node("agent", agent) graph.add_node("execute", execute_action)

  graph.add_edge(START, "agent") graph.add_conditional_edges("agent", should_execute, ["execute", END]) graph.add_edge("execute", END)

  Compile with interrupt_before for human review

  app = graph.compile( checkpointer=memory, interrupt_before=["execute"] # Pause before execution )

  Run until interrupt

  config = {"configurable": {"thread_id": "approval-flow"}} result = app.invoke({"messages": [("user", "Send report")]}, config)

  Agent paused - get pending state

  state = app.get_state(config) pending = state.values["pending_action"] print(f"Pending: {pending}") # Human reviews

  Human approves - update state and continue

  app.update_state(config, {"approved": True}) result = app.invoke(None, config) # Resume

• name: Parallel Execution (Map-Reduce) description: Run multiple branches in parallel when_to_use: Parallel research, batch processing implementation: | from langgraph.graph import StateGraph, START, END, Send from langgraph.constants import Send

  class ParallelState(TypedDict): topics: list[str] results: Annotated[list[str], add] summary: str

  def research_topic(state: dict) -> dict: """Research a single topic.""" topic = state["topic"] result = f"Research on {topic}..." return {"results": [result]}

  def summarize(state: ParallelState) -> dict: """Combine all research results.""" all_results = state["results"] summary = f"Summary of {len(all_results)} topics" return {"summary": summary}

  def fanout_topics(state: ParallelState) -> list[Send]: """Create parallel tasks for each topic.""" return [ Send("research", {"topic": topic}) for topic in state["topics"] ]

  Build graph

  graph = StateGraph(ParallelState) graph.add_node("research", research_topic) graph.add_node("summarize", summarize)

  Fan out to parallel research

  graph.add_conditional_edges(START, fanout_topics, ["research"])

  All research nodes lead to summarize

  graph.add_edge("research", "summarize") graph.add_edge("summarize", END)

  app = graph.compile()

  result = app.invoke({ "topics": ["AI", "Climate", "Space"], "results": [] })

  Research runs in parallel, then summarizes

anti_patterns:

• name: Infinite Loop Without Exit description: Cycles with no termination condition why_bad: | Agent loops forever. Burns tokens and costs. Eventually errors out. what_to_do_instead: | Always have exit conditions:

  • Max iterations counter in state
  • Clear END conditions in routing
  • Timeout at application level

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

• name: Stateless Nodes description: Not using state properly, passing data manually why_bad: | Loses LangGraph's benefits. State not persisted. Can't resume conversations. what_to_do_instead: | Always use state for data flow. Return state updates from nodes. Use reducers for accumulation. Let LangGraph manage state.

• name: Giant Monolithic State description: Putting everything in one state object why_bad: | Hard to reason about. Unnecessary data in context. Serialization overhead. what_to_do_instead: | Use input/output schemas for clean interfaces. Private state for internal data. Clear separation of concerns.

• name: No Persistence in Production description: Running production agents without checkpointer why_bad: | Can't resume on failure. Loses conversation history. No debugging capability. what_to_do_instead: | Always use PostgresSaver or Redis in production. SQLite for development only. Log all state transitions.

handoffs:

• trigger: "crewai|role-based|team" to: crewai context: "Need role-based multi-agent approach"

• trigger: "observability|tracing|monitoring" to: langfuse context: "Need LLM observability for the agent"

• trigger: "structured output|json|parsing" to: structured-output context: "Need structured responses from LLM"

• trigger: "evaluation|testing|benchmark" to: agent-evaluation context: "Need to evaluate agent performance"