Imported from selvarajmurugesan90/ops-engineering-skills (plugins/ai-agent/skills/langchain-and-langgraph-agent-orchestration/SKILL.md). Install upstream with npx skills add selvarajmurugesan90/ops-engineering-skills --skill langchain-and-langgraph-agent-orchestration. Copyright stays with the author (Apache-2.0).

LangChain and LangGraph Agent Orchestration

Purpose

LangChain provides two things that are easy to conflate: a set of composable building blocks (prompt templates, model wrappers, retrievers, output parsers, chained via LangChain Expression Language / Runnable), and a higher-level AgentExecutor that wraps those blocks into a single-loop, tool-calling agent. LangGraph is a separate, lower-level runtime built by the same project specifically for agents whose control flow is not a single linear loop — it models the agent as an explicit graph of nodes and edges, supports cycles (a node can route back to an earlier node), and adds two capabilities AgentExecutor does not have out of the box: durable checkpointed state (so a run can pause, crash, and resume from its last checkpoint) and first-class human-in-the-loop interrupts (a graph can pause at a named node until a human approves, edits, or rejects the pending state). This skill covers choosing between plain LangChain composition, AgentExecutor, and LangGraph, and operating LangGraph's persistence and interrupt features correctly. It is a framework-specific complement to agent-architecture-design, which covers the underlying control-flow patterns (ReAct loop, plan-and-execute, finite-state/graph) in a vendor-neutral way — LangGraph is one concrete runtime that implements the finite-state/graph pattern described there. For tool access, LangChain/LangGraph agents can call tools defined directly in Python or exposed via an MCP server; see mcp-server-development for building the tool-serving side, which this skill treats as an external dependency rather than repeating.

When to use

Deciding whether a task needs plain LangChain chain composition (a fixed pipeline, no branching), AgentExecutor (a single ReAct-style tool-calling loop), or a LangGraph graph (multi-step, branching, cyclical, or needing persistence/human approval).
Building an agent whose steps depend on prior results in ways a single linear chain can't express — retries, conditional branches, or loops back to an earlier step.
Adding durable state to a LangChain/LangGraph agent so a long-running or multi-session workflow survives a process restart or crash mid-run.
Adding a human-in-the-loop approval checkpoint before an irreversible tool call in an existing LangGraph graph.
An agent built with AgentExecutor loses context, re-does completed work, or can't be paused/resumed, and the team is evaluating migrating it to LangGraph.
Debugging a LangGraph graph that loops indefinitely, gets stuck at an interrupt, or fails to restore state correctly from a checkpoint.

Prerequisites & environment

Python (LangChain/LangGraph's primary, most mature ecosystem) or JavaScript/TypeScript (langchain/langgraph npm packages, closely mirroring the Python API but with some feature lag) — pick one and check current package versions before starting, since both projects have had breaking changes across major versions (notably the LangChain 0.1 → 0.2/0.3 restructuring that split core, community, and partner packages).
An LLM provider integration package (e.g. langchain-anthropic, langchain-openai) installed separately from langchain-core — recent LangChain versions moved provider-specific code out of the core package.
For LangGraph persistence: a checkpointer backend — the in-memory MemorySaver for local development/testing only (state is lost on process exit), or a durable checkpointer (SQLite, Postgres, or a managed backend) for anything that must survive a restart.
Tool definitions the agent will call, either as plain Python functions decorated with LangChain's @tool, or proxied from an MCP server via an MCP-to-LangChain adapter — confirm which integration path your LangChain version currently supports before assuming API shape.
Clarity on which parts of the workflow are genuinely cyclical/branching (justifying LangGraph) versus a fixed sequence (better served by a plain LCEL chain) — reach for LangGraph only once a chain's limitations are concrete, mirroring the "justify the split" discipline in agent-architecture-design.

Step-by-step guidance

Start with the simplest abstraction that fits the task's shape. A fixed sequence (retrieve → prompt → parse) is a plain LCEL chain:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_anthropic import ChatAnthropic

prompt = ChatPromptTemplate.from_messages([
    ("system", "Summarize the following ticket in one sentence."),
    ("human", "{ticket_text}"),
])
model = ChatAnthropic(model="claude-sonnet-4-5", temperature=0)
chain = prompt | model | StrOutputParser()
result = chain.invoke({"ticket_text": ticket_body})

No loop, no tool calls, no branching — a chain is the right tool and adding AgentExecutor or LangGraph here is unjustified complexity.

Use AgentExecutor only for a single, bounded ReAct-style loop with no need for persistence, human interrupts, or branching beyond tool-call/no-tool-call:

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool

@tool
def search_tickets(query: str) -> str:
    """Search support tickets by keyword. Returns matching ticket IDs."""
    return ticketing_backend.search(query)

agent = create_tool_calling_agent(model, [search_tickets], prompt)
executor = AgentExecutor(
    agent=agent, tools=[search_tickets],
    max_iterations=8, max_execution_time=60,   # bound the loop explicitly
)
result = executor.invoke({"input": "find open billing tickets"})

max_iterations/max_execution_time are AgentExecutor's equivalent of the hard iteration cap and timeout described in agent-architecture-design — set both explicitly; the defaults are more permissive than most production use cases want.

Move to LangGraph once the task needs cycles, branching, persistence, or a human checkpoint. Model the agent as a typed state object and a graph of nodes:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

class TicketState(TypedDict):
    ticket_id: str
    category: str
    draft_reply: str
    approved: bool

def triage(state: TicketState) -> TicketState:
    category = classify(state["ticket_id"])
    return {**state, "category": category}

def draft(state: TicketState) -> TicketState:
    reply = generate_draft(state["ticket_id"], state["category"])
    return {**state, "draft_reply": reply}

def route_after_triage(state: TicketState) -> str:
    return "escalate" if state["category"] == "legal" else "draft"

graph = StateGraph(TicketState)
graph.add_node("triage", triage)
graph.add_node("draft", draft)
graph.add_node("escalate", lambda s: {**s, "approved": False})
graph.set_entry_point("triage")
graph.add_conditional_edges("triage", route_after_triage, {"draft": "draft", "escalate": "escalate"})
graph.add_edge("draft", END)
graph.add_edge("escalate", END)

app = graph.compile(checkpointer=MemorySaver())

This is the finite-state/graph pattern from agent-architecture-design expressed directly in LangGraph's API — nodes are states, edges (plain or conditional) are transitions.

Add a human-in-the-loop interrupt at the highest-leverage node, not everywhere, using interrupt_before/interrupt_after at compile time:
```
app = graph.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["send_reply"],   # pause here every run until resumed
)

config = {"configurable": {"thread_id": "ticket-8842"}}
app.invoke(initial_state, config=config)   # runs up to send_reply, then pauses

# ... a human reviews the checkpointed state out-of-band ...

app.invoke(None, config=config)            # resumes from the paused checkpoint
```
Passing None as input on resume is deliberate — it tells LangGraph to continue from the last checkpoint rather than starting a new run; passing a real input restarts the thread instead.

Warning: compiling a graph that reaches an irreversible-write node (sending a message, executing a payment, deleting a resource) with no interrupt_before on that node means it will execute automatically the first time the graph reaches it, with no human checkpoint at all. Do not rely on prompt wording alone to prevent an irreversible action — gate it structurally with interrupt_before, the same discipline described for any irreversible tool in agent-architecture-design.
Choose a checkpointer backend deliberately for the deployment target. MemorySaver is fine for local development and tests; anything long-running or multi-process needs a durable checkpointer:
```
from langgraph.checkpoint.sqlite import SqliteSaver
# or, for production multi-instance deployments:
# from langgraph.checkpoint.postgres import PostgresSaver

with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)
```
Every checkpointed run needs a stable thread_id in config; reusing a thread_id across unrelated tasks corrupts that thread's history.

Bound cycles explicitly. A conditional edge that can route back to an earlier node (e.g. draft -> review -> draft on rejection) needs an explicit counter in state and a hard cap, or a rejection loop can run indefinitely:

def route_after_review(state: TicketState) -> str:
    if state.get("revision_count", 0) >= 3:
        return "escalate"          # fail closed after 3 rejected drafts
    return "draft" if not state["approved"] else END

Stream intermediate state for observability, rather than only consuming the final result — both AgentExecutor and LangGraph support streaming (.stream()/.astream()), which surfaces each tool call and state transition as it happens, matching the "instrument every loop iteration" guidance in agent-architecture-design.
Compose multiple LangGraph graphs for multi-agent topologies rather than hand-rolling a supervisor loop, when the task genuinely needs multiple specialized roles — a compiled graph can itself be a node in a parent graph. Confirm this split is justified per multi-agent-orchestration before introducing it; LangGraph makes multi-agent easy to wire, not automatically the right call.

Best practices

Default to the least powerful abstraction that fits: LCEL chain < AgentExecutor < LangGraph. Reaching for LangGraph on day one for a fixed three-step pipeline adds state-management overhead with no payoff.
Keep LangGraph node functions small, pure, and independently testable — each node should be callable and assertable in isolation, the same way you would unit-test any function, rather than only testable end-to-end through the compiled graph.
Treat thread_id as a real identifier with a lifecycle (create, resume, archive/expire), not an incidental config parameter — losing track of which thread_id corresponds to which real-world task makes checkpointed state unrecoverable in practice even though it's durably stored.
Put interrupt_before only on nodes that perform irreversible or high-consequence actions (sending a message, executing a payment, deleting a resource), mirroring the "human checkpoint at the highest-leverage point" guidance in agent-architecture-design — an interrupt on every node defeats the purpose and trains reviewers to rubber-stamp.
Pin langchain-core, langgraph, and provider integration package versions together and upgrade them as a set — LangChain's ecosystem has a history of breaking changes across the core/community/partner package split, and mismatched versions fail in confusing ways (missing attributes, import errors) rather than a clear version-conflict message.
Prefer typed state (TypedDict, a Pydantic model, or a dataclass) over a loose dict for LangGraph state — type errors in state shape are a common source of node-to-node bugs that a type checker catches before runtime.

Common pitfalls

Symptom: An AgentExecutor-based agent "forgets" earlier context or redoes work across what should be one continuous multi-turn session. Fix: AgentExecutor has no built-in cross-invocation persistence — each .invoke() call is independent unless you manage message history yourself. If the task genuinely needs durable, resumable state across turns or a process restart, migrate to LangGraph with a persistent checkpointer rather than hand-rolling message-history plumbing on top of AgentExecutor.
Symptom: A LangGraph graph with a conditional edge routing back to an earlier node runs far longer than expected, or never terminates. Fix: Add an explicit counter field to state (e.g. revision_count) and check it in the conditional-edge function with a hard cap that routes to an escape/escalate node — do not rely on the model's own judgment about when to stop looping, per the loop-bounding guidance in agent-architecture-design.
Symptom: Resuming a LangGraph run after an interrupt_before pause restarts the whole graph from the entry point instead of continuing from where it paused. Fix: Confirm you invoked with the same thread_id in config and passed None (not a fresh input dict) on resume — passing a new input value is interpreted as starting a new run on that thread, not continuing the paused one.
Symptom: State checkpointed with MemorySaver disappears between requests in what looks like a production deployment. Fix: MemorySaver is process-local, in-memory state — it does not survive a process restart and does not work correctly behind multiple server instances/replicas. Swap in a durable, shared checkpointer (SQLite for single-instance durability, Postgres or another shared backend for multi-instance deployments) before treating persistence as production-ready.
Symptom: Upgrading langchain breaks imports (ModuleNotFoundError, moved classes) with no code logic changes. Fix: Check whether the class moved to a separate langchain-community or provider-specific package (langchain-anthropic, langchain-openai) as part of LangChain's core/community/partner package split; pin langchain-core, langchain, and each provider package to versions tested together rather than upgrading one package in isolation.

Worked example

Scenario: An internal support-ticket agent triages a ticket, drafts a reply, and requires human approval before sending — the same workflow used as the finite-state example in agent-architecture-design, built concretely in LangGraph with durable persistence and a human interrupt.

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver

class TicketState(TypedDict):
    ticket_id: str
    category: str
    draft_reply: str
    revision_count: int

def triage(state: TicketState) -> TicketState:
    return {**state, "category": classify_ticket(state["ticket_id"])}

def draft_reply(state: TicketState) -> TicketState:
    reply = generate_reply(state["ticket_id"], state["category"])
    return {**state, "draft_reply": reply}

def send_reply(state: TicketState) -> TicketState:
    ticketing_backend.send(state["ticket_id"], state["draft_reply"])
    return state

graph = StateGraph(TicketState)
graph.add_node("triage", triage)
graph.add_node("draft_reply", draft_reply)
graph.add_node("send_reply", send_reply)
graph.set_entry_point("triage")
graph.add_edge("triage", "draft_reply")
graph.add_edge("draft_reply", "send_reply")
graph.add_edge("send_reply", END)

with SqliteSaver.from_conn_string("tickets_checkpoints.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer, interrupt_before=["send_reply"])

    config = {"configurable": {"thread_id": "ticket-8842"}}
    app.invoke({"ticket_id": "8842", "category": "", "draft_reply": "", "revision_count": 0}, config=config)
    # graph pauses before send_reply; state.draft_reply is now available for a
    # human reviewer to read via app.get_state(config)

    # ... later, after a reviewer approves in a separate process/session ...
    app.invoke(None, config=config)   # resumes and executes send_reply

Because SqliteSaver persists to disk, the pause can span a process restart — a reviewer approving the draft an hour later (or after a deployment) still resumes correctly from the same thread_id, which a MemorySaver-backed graph could not survive.

Cross-references

agent-architecture-design — the vendor-neutral control-flow patterns (ReAct loop, plan-and-execute, finite-state/graph) that LangGraph implements concretely; read this first for the underlying design principles.
multi-agent-orchestration — when to compose multiple LangGraph graphs into a supervisor/worker or pipeline topology, and the coordination pitfalls that apply regardless of framework.
mcp-server-development — building the tool-serving side an agent calls into; this skill covers wiring those tools into a LangChain/LangGraph agent, not building the MCP server itself.
crewai-and-autogen-multi-agent-frameworks — alternative higher-level multi-agent frameworks with a role-based abstraction, contrasted with LangGraph's lower-level graph model.
agent-tool-use-patterns — general tool-design principles (single-purpose tools, schema clarity) that apply to @tool-decorated functions in LangChain the same as any other agent tool surface.

langchain-and-langgraph-agent-orchestration

About