Multi-Agent Orchestration with LangGraph: Patterns for Enterprise AI Systems

Introduction

Single-agent LLM systems hit a ceiling quickly in enterprise environments. Complex business processes — document review workflows, multi-step compliance checks, incident response — require coordination between specialized agents that each handle a distinct part of the problem.

LangGraph provides a graph-based framework for building these multi-agent systems with explicit state management and conditional execution. After deploying LangGraph-based orchestration across two enterprise platforms, I want to share the patterns that survived contact with production workloads.

Why Multi-Agent Architecture?

The fundamental limitation of monolithic prompting is context window economics. A single agent tasked with understanding compliance regulations, analyzing financial documents, generating reports, and routing approvals will exhaust its context budget and produce inconsistent results.

Multi-agent systems solve this by:

• Decomposing complexity — each agent has a focused system prompt and limited responsibility
• Enabling specialization — agents can use different models (GPT-4 for reasoning, GPT-3.5 for classification) based on task requirements
• Supporting parallelism — independent analysis steps execute concurrently
• Improving observability — each agent's inputs and outputs are individually traceable

LangGraph Core Concepts

LangGraph models agent orchestration as a directed graph where nodes are computation steps and edges define transitions. State flows through the graph and accumulates results.

Defining the State Schema

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

class AgentState(TypedDict):
    """Shared state passed between agents in the orchestration graph."""
    messages: Annotated[list, add_messages]
    documents: list[dict]
    classification: str | None
    analysis_results: dict
    approval_status: str
    current_step: str
    error_log: list[str]

The state schema is critical. It defines the contract between agents — what each agent can read and what it must produce. In enterprise systems, I include error tracking and step identification directly in state for observability.

Building the Orchestration Graph

from langgraph.graph import END
from langgraph.checkpoint.memory import MemorySaver

def build_document_review_graph() -> StateGraph:
    """Build a multi-agent document review orchestration graph."""
    workflow = StateGraph(AgentState)

    # Add agent nodes
    workflow.add_node("classifier", classify_document)
    workflow.add_node("compliance_checker", check_compliance)
    workflow.add_node("financial_analyst", analyze_financials)
    workflow.add_node("summarizer", generate_summary)
    workflow.add_node("router", route_for_approval)
    workflow.add_node("human_review", human_in_the_loop)

    # Define edges with conditional routing
    workflow.set_entry_point("classifier")

    workflow.add_conditional_edges(
        "classifier",
        route_by_classification,
        {
            "financial": "financial_analyst",
            "compliance": "compliance_checker",
            "general": "summarizer",
        },
    )

    workflow.add_edge("financial_analyst", "summarizer")
    workflow.add_edge("compliance_checker", "summarizer")
    workflow.add_edge("summarizer", "router")

    workflow.add_conditional_edges(
        "router",
        check_approval_required,
        {
            "requires_human": "human_review",
            "auto_approved": END,
        },
    )

    workflow.add_edge("human_review", END)

    return workflow.compile(checkpointer=MemorySaver())

Production Patterns

Pattern 1: Supervisor Agent with Specialized Workers

The most robust pattern for enterprise workloads is a supervisor-worker topology. A supervisor agent receives the initial request, decomposes it into subtasks, delegates to specialized workers, and synthesizes results.

from langchain_openai import AzureChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

class SupervisorAgent:
    def __init__(self):
        self.llm = AzureChatOpenAI(
            deployment_name="gpt-4",
            temperature=0,
            max_tokens=1000,
        )
        self.system_prompt = SystemMessage(content="""
            You are a workflow supervisor. Analyze the incoming request and
            determine which specialist agents need to be invoked. Return a
            JSON plan with ordered steps and agent assignments.
        """)

    async def plan(self, state: AgentState) -> AgentState:
        """Decompose the request into a multi-step execution plan."""
        response = await self.llm.ainvoke([
            self.system_prompt,
            HumanMessage(content=f"Request: {state['messages'][-1].content}"),
        ])

        plan = parse_plan(response.content)
        state["analysis_results"]["plan"] = plan
        state["current_step"] = plan["steps"][0]["agent"]

        return state

Pattern 2: Conditional Routing with Fallbacks

Enterprise workflows need graceful degradation. When an agent fails or produces low-confidence results, the graph should route to a fallback path rather than failing entirely.

def route_with_fallback(state: AgentState) -> str:
    """Route to next agent with fallback handling."""
    confidence = state["analysis_results"].get("confidence", 0)
    error_count = len(state["error_log"])

    if error_count > 2:
        return "human_escalation"

    if confidence < 0.7:
        return "secondary_analysis"

    if state["classification"] == "high_risk":
        return "compliance_checker"

    return "summarizer"

Pattern 3: Human-in-the-Loop Checkpoints

Certain enterprise decisions cannot be fully automated. LangGraph supports interrupt points where execution pauses for human input before continuing.

from langgraph.types import interrupt

def human_in_the_loop(state: AgentState) -> AgentState:
    """Pause execution for human review and approval."""
    review_payload = {
        "document_summary": state["analysis_results"].get("summary"),
        "risk_flags": state["analysis_results"].get("risks", []),
        "recommended_action": state["approval_status"],
    }

    # This interrupts graph execution until human provides input
    human_decision = interrupt(review_payload)

    state["approval_status"] = human_decision["decision"]
    state["messages"].append(
        HumanMessage(content=f"Human reviewer decided: {human_decision['decision']}")
    )

    return state

State Management at Scale

Persistent Checkpointing

In production, graph execution can span minutes or hours (waiting for human input, external API calls, batch processing). Persistent checkpointing ensures execution resumes correctly after interruptions.

import { StateGraph } from "@langchain/langgraph";
import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";

const checkpointer = new PostgresSaver({
  connectionString: process.env.DATABASE_URL,
  tableName: "agent_checkpoints",
});

const graph = workflow.compile({ checkpointer });

// Resume from checkpoint
const config = { configurable: { thread_id: "workflow-123" } };
const result = await graph.invoke(currentState, config);

State Isolation Between Tenants

Enterprise multi-tenant deployments require strict state isolation. Each tenant's agent executions must be completely independent:

from dataclasses import dataclass

@dataclass
class TenantConfig:
    tenant_id: str
    thread_prefix: str
    model_deployment: str
    max_concurrent_workflows: int

def get_thread_id(tenant: TenantConfig, workflow_id: str) -> str:
    """Generate tenant-isolated thread identifier."""
    return f"{tenant.thread_prefix}:{tenant.tenant_id}:{workflow_id}"

Error Handling and Resilience

Retry with Exponential Backoff

LLM API calls are inherently unreliable. Every agent node should implement retry logic:

import asyncio
from functools import wraps

def with_retry(max_attempts: int = 3, base_delay: float = 1.0):
    """Decorator for retrying agent operations with exponential backoff."""
    def decorator(func):
        @wraps(func)
        async def wrapper(state: AgentState) -> AgentState:
            for attempt in range(max_attempts):
                try:
                    return await func(state)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        state["error_log"].append(
                            f"{func.__name__} failed after {max_attempts} attempts: {str(e)}"
                        )
                        raise
                    delay = base_delay * (2 ** attempt)
                    await asyncio.sleep(delay)
            return state
        return wrapper
    return decorator

@with_retry(max_attempts=3)
async def classify_document(state: AgentState) -> AgentState:
    """Classify document type with retry logic."""
    # Agent implementation here
    pass

Circuit Breaker for External Dependencies

When downstream services degrade, agent graphs should fail fast rather than accumulating timeouts:

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold: int = 5, reset_timeout: int = 60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = timedelta(seconds=reset_timeout)
        self.last_failure_time: datetime | None = None
        self.state = "closed"  # closed, open, half-open

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if datetime.now() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                return True
            return False
        return True  # half-open allows one attempt

    def record_failure(self) -> None:
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = "closed"

Observability and Monitoring

Tracing Agent Execution

Every production multi-agent system needs distributed tracing. LangGraph integrates with LangSmith for full execution visibility:

import os

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "enterprise-document-review"

# Every node execution, LLM call, and state transition is automatically traced

Key metrics to monitor in multi-agent systems:

1. End-to-end latency — total time from request to final output
2. Per-agent latency — time spent in each graph node
3. Token consumption — cumulative tokens across all agent invocations
4. Routing distribution — which conditional paths execute most frequently
5. Error rate by agent — identifies which specialized agents are least reliable

Lessons From Production

After running multi-agent LangGraph systems in production for over a year, these are the lessons that matter most:

• Start with two agents, not ten. The coordination overhead of multi-agent systems is significant. Add agents only when a single agent demonstrably cannot handle the task.
• State schemas are your API contract. Changing state shape after deployment is painful. Invest in getting the schema right early.
• Deterministic routing beats LLM-based routing for critical paths. Use rule-based conditional edges for high-stakes decisions; reserve LLM routing for ambiguous cases.
• Checkpointing is non-negotiable. Any workflow that can take more than 30 seconds needs persistent state. Users will close browsers, networks will drop, and services will restart.
• Test with adversarial inputs. Multi-agent systems amplify edge cases. A misclassification in agent 1 cascades through all downstream agents.

Conclusion

LangGraph provides the right abstraction level for enterprise multi-agent systems — explicit enough for debugging and monitoring, flexible enough for complex conditional workflows. The patterns described here have proven reliable across document processing, compliance automation, and knowledge management workloads at enterprise scale.

The key insight is that multi-agent orchestration is fundamentally a distributed systems problem with AI characteristics, not an AI problem with distributed systems characteristics. Bring your systems engineering discipline first, and the AI pieces follow naturally.