langgraph-state-graph

name	langgraph-state-graph
description	LangGraph StateGraph builder with state schema design. Create stateful agent workflows with cycles, conditionals, and persistence.
allowed-tools	Read, Grep, Write, Edit, Bash, Glob, WebFetch
graph	{"domains":["domain:software-engineering"],"specializations":["specialization:ai-agents-conversational"],"skillAreas":["skill-area:agentic-loops","skill-area:agent-planning-reasoning"],"roles":["role:ml-engineer","role:backend-engineer"],"workflows":["workflow:feature-development","workflow:ml-model-lifecycle"],"topics":["topic:design-patterns","topic:event-driven-architecture"]}

langgraph-state-graph

Build stateful agent workflows using LangGraph's StateGraph pattern. Design state schemas, create nodes, define edges with conditional routing, and enable persistence.

Overview

LangGraph is a library for building stateful, multi-actor applications with LLMs. The StateGraph is the core abstraction that enables:

• Cyclical computation graphs (unlike DAGs)
• State persistence and checkpointing
• Human-in-the-loop interaction patterns
• Conditional branching and routing
• Multi-agent coordination

Capabilities

State Schema Design

• Define typed state schemas with TypedDict or Pydantic
• Configure state channels for message passing
• Set up reducer functions for state updates
• Design accumulator patterns for conversation history

Graph Construction

• Create nodes as functions or runnables
• Define edges (normal, conditional, entry points)
• Configure start and end nodes
• Implement routing logic for conditional edges

Persistence & Checkpointing

• Configure checkpoint backends (SQLite, PostgreSQL, Redis)
• Enable state snapshots at each step
• Support for resuming interrupted workflows
• Thread-based conversation persistence

Human-in-the-Loop

• Insert interrupt points in workflows
• Collect human feedback before continuing
• Support approval gates and input collection
• Resume from interrupt with updated state

Usage

Basic StateGraph Pattern

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

# Define state schema
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_step: str
    iteration: int

# Create nodes
def agent_node(state: AgentState) -> AgentState:
    # Process state and return updates
    return {"current_step": "processed", "iteration": state["iteration"] + 1}

def tool_node(state: AgentState) -> AgentState:
    # Execute tools based on agent decisions
    return {"current_step": "tools_executed"}

# Build graph
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# Define edges
graph.set_entry_point("agent")
graph.add_edge("agent", "tools")
graph.add_conditional_edges(
    "tools",
    lambda state: "end" if state["iteration"] >= 3 else "continue",
    {"end": END, : }
)


app = graph.()

Conditional Routing

def router(state: AgentState) -> str:
    """Route based on state conditions."""
    last_message = state["messages"][-1]

    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    elif state["iteration"] >= state.get("max_iterations", 10):
        return "end"
    else:
        return "agent"

graph.add_conditional_edges(
    "agent",
    router,
    {
        "tools": "tool_executor",
        "agent": "agent",
        "end": END
    }
)

Persistence with Checkpointing

from langgraph.checkpoint.sqlite import SqliteSaver

# Configure checkpointer
memory = SqliteSaver.from_conn_string(":memory:")

# Compile with persistence
app = graph.compile(checkpointer=memory)

# Run with thread_id for persistence
config = {"configurable": {"thread_id": "conversation-1"}}
result = app.invoke(initial_state, config=config)

# Resume from checkpoint
result = app.invoke(None, config=config)  # Continues from last state

Human-in-the-Loop

from langgraph.graph import StateGraph

graph = StateGraph(AgentState)
# ... add nodes ...

# Compile with interrupt points
app = graph.compile(
    checkpointer=memory,
    interrupt_before=["tool_executor"]  # Pause before tool execution
)

# First invocation - pauses at interrupt
result = app.invoke(initial_state, config)

# After human approval, resume
result = app.invoke(None, config)  # Continues past interrupt

Task Definition

const langgraphStateGraphTask = defineTask({
  name: 'langgraph-state-graph-design',
  description: 'Design and implement a LangGraph StateGraph workflow',

  inputs: {
    workflowName: { type: 'string', required: true },
    stateSchema: { type: 'object', required: true },
    nodes: { type: 'array', required: true },
    edges: { type: 'array', required: true },
    enablePersistence: { type: 'boolean', default: true },
    interruptPoints: { type: 'array', default: [] }
  },

  outputs: {
    graphCode: { type: 'string' },
    stateSchemaCode: { type: 'string' },
    compiledGraph: { type: 'boolean' },
    artifacts: { type: 'array' }
  },

  async run(inputs, taskCtx) {
    return {
      : ,
      : ,
      : {
        : ,
        : {
          : inputs.,
          : inputs.,
          : inputs.,
          : inputs.,
          : inputs.,
          : inputs.,
          : [
            ,
            ,
            ,
            ,
            ,
            ,
            
          ]
        }
      },
      : {
        : ,
        : 
      }
    };
  }
});

Applicable Processes

• langgraph-workflow-design
• multi-agent-system
• plan-and-execute-agent
• conversational-memory-system

External Dependencies

• langgraph Python package
• langchain-core
• Optional: langgraph-checkpoint-sqlite, langgraph-checkpoint-postgres

References

• LangGraph Documentation
• Awesome LangGraph
• LangGraph RAG MCP
• LangGraph MCP Agents

Related Skills

• SK-LG-002 langgraph-checkpoint
• SK-LG-003 langgraph-hitl
• SK-LG-004 langgraph-routing
• SK-LG-005 langgraph-subgraph

Related Agents

• AG-AA-004 langgraph-workflow-designer
• AG-MEM-004 state-machine-designer