Langgraph Orchestrator

LangGraph Orchestrator Skill

A skill that governs how to build, extend, and debug stateful agent workflows using LangGraph.

When to use LangGraph vs simple agents:

| Use Simple Agent (ReAct) | Use LangGraph | |---|---| | Single task, linear flow | Multi-step with branching logic | | No state between steps | State must persist across steps | | No human approval needed | Human-in-the-loop required | | One agent, few tools | Multiple agents coordinating | | No cycles or retries | Retry loops, self-correction |

---

Operation: BUILD — Creating a LangGraph Workflow

Core Concepts

StateGraph — A graph where nodes are functions and edges define the flow. State is a typed dictionary that flows through the graph.

Nodes — Python functions that receive state and return updates. Each node does ONE thing.

Edges — Connections between nodes. Can be unconditional (always follow) or conditional (choose based on state).

State — A TypedDict that accumulates data as it flows through nodes. Use Annotated with reducers for list fields.

Step 1 — Define State

State is the data contract for your entire graph. Define it upfront.

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

class AgentState(TypedDict):
    """State that flows through the graph."""
    messages: Annotated[list, add_messages]  # conversation history
    current_step: str                         # tracks progress
    result: str                               # final output

State design rules:

• Include only fields that nodes need to read or write
• Use Annotated[list, add_messages] for message history (appends, not replaces)
• Use plain fields for values that get overwritten
• Keep state flat — avoid deep nesting

Step 2 — Define Nodes

Each node is a function that takes state and returns a partial state update.

from langchain_core.messages import SystemMessage, HumanMessage
from common.llm import create_llm

llm = create_llm()

def analyze(state: AgentState) -> dict:
    """Analyze the input and determine next action."""
    messages = [
        SystemMessage(content="You are an analyst. Classify the input."),
        *state["messages"],
    ]
    response = llm.invoke(messages)
    return {"messages": [response], "current_step": "analyze"}

def generate_report(state: AgentState) -> dict:
    """Generate a report from the analysis."""
    messages = [
        SystemMessage(content="Generate a concise report from the analysis."),
        *state["messages"],
    ]
    response = llm.invoke(messages)
    return {"messages": [response], "result": response.content}

Node rules:

• One task per node (the "one task, one agent" principle)
• Return a dict with only the fields being updated
• Never mutate state directly — return updates
• Keep nodes stateless — all state lives in the TypedDict

Step 3 — Define Edges and Routing

def should_continue(state: AgentState) -> str:
    """Route based on the last message content."""
    last_message = state["messages"][-1]
    if "NEEDS_REVIEW" in last_message.content:
        return "review"
    return "report"

Step 4 — Build the Graph

from langgraph.graph import StateGraph, START, END

# Create the graph
graph = StateGraph(AgentState)

# Add nodes
graph.add_node("analyze", analyze)
graph.add_node("review", human_review)
graph.add_node("report", generate_report)

# Add edges
graph.add_edge(START, "analyze")
graph.add_conditional_edges("analyze", should_continue, {
    "review": "review",
    "report": "report",
})
graph.add_edge("review", "analyze")  # cycle: review → re-analyze
graph.add_edge("report", END)

# Compile
app = graph.compile()

Step 5 — Run the Graph

from langchain_core.messages import HumanMessage

result = app.invoke({
    "messages": [HumanMessage(content="Analyze this log file...")],
    "current_step": "",
    "result": "",
})
print(result["result"])

---

Common Patterns

Pattern 1 — ReAct Agent with Tools

The simplest LangGraph pattern. Use create_react_agent for a single agent with tools.

from langgraph.prebuilt import create_react_agent
from common.llm import create_llm

llm = create_llm()
tools = [search_logs, read_file, run_query]

agent = create_react_agent(llm, tools)
result = agent.invoke({
    "messages": [HumanMessage(content="Find errors in today's logs")]
})

Pattern 2 — Sequential Pipeline

Nodes execute in a fixed order.

graph = StateGraph(PipelineState)
graph.add_node("parse", parse_input)
graph.add_node("validate", validate_data)
graph.add_node("transform", transform_data)
graph.add_node("output", format_output)

graph.add_edge(START, "parse")
graph.add_edge("parse", "validate")
graph.add_edge("validate", "transform")
graph.add_edge("transform", "output")
graph.add_edge("output", END)

Pattern 3 — Conditional Branching

Route to different nodes based on state.

def classify_input(state) -> str:
    """Classify input type for routing."""
    content = state["messages"][-1].content.lower()
    if "bug" in content:
        return "bug_handler"
    elif "feature" in content:
        return "feature_handler"
    return "general_handler"

graph.add_conditional_edges("classifier", classify_input, {
    "bug_handler": "handle_bug",
    "feature_handler": "handle_feature",
    "general_handler": "handle_general",
})

Pattern 4 — Self-Correction Loop

Agent checks its own work and retries if needed.

def check_quality(state) -> str:
    """Evaluate output quality."""
    if state.get("retry_count", 0) >= 3:
        return "accept"  # give up after 3 retries
    if quality_score(state["result"]) < 0.8:
        return "retry"
    return "accept"

graph.add_conditional_edges("checker", check_quality, {
    "retry": "generator",   # loop back
    "accept": "output",
})

Pattern 5 — Human-in-the-Loop

Pause the graph for human approval before continuing.

from langgraph.checkpoint.memory import MemorySaver

# Compile with checkpointing (required for interrupts)
checkpointer = MemorySaver()
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["dangerous_action"],  # pause before this node
)

# Run until interrupt
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke(input_data, config)
# → Graph pauses before "dangerous_action"

# User reviews, then resume
result = app.invoke(None, config)  # continues from checkpoint

When to use interruptbefore vs interruptafter:

• interrupt_before — pause BEFORE the node runs (for approval gates)
• interrupt_after — pause AFTER the node runs (for review of output)

Pattern 6 — Multi-Agent Graph

Each node is a different specialist agent.

from langgraph.prebuilt import create_react_agent

# Create specialist agents
researcher = create_react_agent(
    create_llm(), [web_search, read_document],
    prompt="You are a research specialist..."
)
writer = create_react_agent(
    create_llm(temperature=0.7), [write_file, format_text],
    prompt="You are a technical writer..."
)
reviewer = create_react_agent(
    create_llm(), [read_file, check_style],
    prompt="You are a code reviewer..."
)

# Wire into graph
graph = StateGraph(TeamState)
graph.add_node("research", researcher)
graph.add_node("write", writer)
graph.add_node("review", reviewer)

graph.add_edge(START, "research")
graph.add_edge("research", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", needs_revision, {
    "revise": "write",
    "approve": END,
})

Pattern 7 — Subgraphs

Nest one graph inside another for modular design.

# Define inner graph
inner_graph = StateGraph(InnerState)
inner_graph.add_node("step_a", do_a)
inner_graph.add_node("step_b", do_b)
inner_graph.add_edge(START, "step_a")
inner_graph.add_edge("step_a", "step_b")
inner_graph.add_edge("step_b", END)
inner_compiled = inner_graph.compile()

# Use as a node in outer graph
outer_graph = StateGraph(OuterState)
outer_graph.add_node("preprocess", preprocess)
outer_graph.add_node("inner_workflow", inner_compiled)  # subgraph as node
outer_graph.add_node("postprocess", postprocess)

---

Checkpointing and Persistence

In-Memory (Development)

from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

SQLite (Production — Single Instance)

from langgraph.checkpoint.sqlite import SqliteSaver
checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
app = graph.compile(checkpointer=checkpointer)

Thread Management

Every conversation gets a unique thread_id for state isolation:

config = {"configurable": {"thread_id": "unique-conversation-id"}}
result = app.invoke(input_data, config)

---

Streaming

Stream Node Outputs

for event in app.stream(input_data, config):
    for node_name, output in event.items():
        print(f"[{node_name}] {output}")

Stream LLM Tokens

for event in app.stream(input_data, config, stream_mode="messages"):
    # event is (message_chunk, metadata)
    print(event[0].content, end="", flush=True)

---

Operation: DEBUG — Diagnosing Graph Issues

Visualization

# Print the graph structure
print(app.get_graph().draw_mermaid())

Common Issues

| Symptom | Cause | Fix | |---|---|---| | Graph runs forever | Missing END edge or broken conditional | Ensure every path reaches END, add recursion limit | | State not updating | Node returns wrong keys | Return dict with exact TypedDict field names | | Messages duplicating | Not using add_messages reducer | Use Annotated[list, add_messages] for message fields | | Checkpoint errors | No checkpointer configured | Add MemorySaver() to compile() | | Human-in-the-loop not pausing | Missing checkpointer | Interrupts require a checkpointer | | Wrong node executes | Conditional edge returns wrong key | Print routing decisions, verify return values match edge map |

Recursion Limit

Prevent infinite loops:

app = graph.compile()
result = app.invoke(
    input_data,
    config={"recursion_limit": 25},  # default is 25
)

---

When NOT to Use LangGraph

• Simple Q&A — use a direct LLM call
• Linear pipeline with no branching — use a simple chain (prompt | llm | parser)
• Single agent with tools — use create_react_agent directly (still LangGraph, but no custom graph needed)
• No state between calls — use stateless LangChain chains

Start with the simplest approach. Escalate to a custom StateGraph only when you need branching, cycles, human-in-the-loop, or multi-agent coordination.

---

Post-Build Verification

• [ ] Every path through the graph reaches END
• [ ] State TypedDict includes all fields nodes read/write
• [ ] List fields use Annotated with appropriate reducers
• [ ] Conditional edges return values that match the edge map exactly
• [ ] Recursion limit is set to prevent infinite loops
• [ ] Checkpointer is configured if using interrupts or persistence
• [ ] Graph visualization matches intended flow (draw_mermaid())
• [ ] Each node does one thing and returns a partial state update

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: ravi2799
• Source: ravi2799/ai-agent-skills
• License: MIT
• Homepage: https://skills.sh

Install and usage instructions live in the source repository linked above.