Agent Orchestration with LangGraph

Turn the execution flow, state, and failure recovery of a multi-agent system into a LangGraph StateGraph.

The financial task names, latency targets, and code in this article are illustrative design examples. Execution and financial performance were not validated, and this is not a reproduction of a particular paper.

Why orchestration is needed

In a multi-agent system, communication between agents, state management, and control of the execution flow matter as much as the reasoning ability of any individual agent. Wiring nine agents into a plain sequential for-loop hits a set of hard limits.

LangGraph addresses all of these with a graph-based state machine (StateGraph).

Core LangGraph concepts

StateGraph

LangGraph models an agent workflow as a stateful, controlled execution graph. A node is one agent's unit of execution, and an edge defines a state transition. The example below uses a directed acyclic graph (DAG), but LangGraph itself can also represent cycles.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
 
class PipelineState(TypedDict):
    project_id: str
    market: str
    config: dict
    scan_results: list[ScanResult]
    news_items: list[NewsItem]
    position_alerts: list[PositionAlert]
    signals: list[Signal]
    debate_result: dict | None
    final_decision: dict | None
    errors: list[str]
 
graph = StateGraph(PipelineState)

PipelineState is the shared state for the whole workflow. Each node reads and updates it, and the checkpointer saves a snapshot after every step.

Core building blocks

StateGraph DAG

Mapping onto the three Sense-Think-Act layers

The nine agents split across the three Sense-Think-Act layers, and the LangGraph StateGraph wires them into a DAG.

flowchart TD
    subgraph Sense["Sense Layer"]
        DQG[Data Quality Gate]
        VS[Volume Scanner]
        NC[News Collector]
        PS[Position Sensor]
    end

    subgraph Think["Think Layer"]
        QA[Quant Analyst]
        LA[LLM Analyst]
        SR[Sector Rotation]
        MERGE[Analysis Merge]
        BD[Bull/Bear Debate]
        EV[Ensemble Voter]
    end

    subgraph Act["Act Layer"]
        RS[Risk Sentinel]
        TE[Trade Executor]
    end

    DQG -->|PASS| VS
    DQG -->|FAIL| END_NODE[END]
    VS --> NC --> PS

    PS --> QA
    PS --> LA
    PS --> SR

    QA --> MERGE
    LA --> MERGE
    SR --> MERGE

    MERGE --> BD --> EV --> RS

    RS -->|APPROVE| TE
    RS -->|REJECT| END_NODE
    TE --> END_NODE

Graph definition

graph = StateGraph(PipelineState)
 
# --- Sense Layer ---
graph.add_node("data_quality_gate", data_quality_gate)
graph.add_node("volume_scanner", volume_scanner)
graph.add_node("news_collector", news_collector)
graph.add_node("position_sensor", position_sensor)
 
# --- Think Layer (fan-out / fan-in) ---
graph.add_node("quant_analyst", quant_analyst)
graph.add_node("llm_analyst", llm_analyst)
graph.add_node("sector_rotation", sector_rotation)
graph.add_node("analysis_merge", merge_analysis_results)
graph.add_node("bull_bear_debate", bull_bear_debate)
graph.add_node("ensemble_voter", ensemble_voter)
 
# --- Act Layer ---
graph.add_node("risk_sentinel", risk_sentinel)
graph.add_node("trade_executor", trade_executor)
 
# Edge definitions
graph.set_entry_point("data_quality_gate")
 
# Data Quality Gate: conditional branch
graph.add_conditional_edges(
    "data_quality_gate",
    lambda s: "continue" if not s["errors"] else "end",
    {"continue": "volume_scanner", "end": END},
)
 
# Sense Layer runs sequentially
graph.add_edge("volume_scanner", "news_collector")
graph.add_edge("news_collector", "position_sensor")
 
# Fan-out: parallel analysis
graph.add_edge("position_sensor", "quant_analyst")
graph.add_edge("position_sensor", "llm_analyst")
graph.add_edge("position_sensor", "sector_rotation")
 
# Fan-in: results converge
graph.add_edge("quant_analyst", "analysis_merge")
graph.add_edge("llm_analyst", "analysis_merge")
graph.add_edge("sector_rotation", "analysis_merge")
 
# Second half of the Think Layer
graph.add_edge("analysis_merge", "bull_bear_debate")
graph.add_edge("bull_bear_debate", "ensemble_voter")
graph.add_edge("ensemble_voter", "risk_sentinel")
 
# Risk Sentinel: conditional branch
graph.add_conditional_edges(
    "risk_sentinel",
    lambda s: "execute" if s["final_decision"].get("approved") else "end",
    {"execute": "trade_executor", "end": END},
)
graph.add_edge("trade_executor", END)

Fan-out / fan-in parallelism

What parallelism buys

Inside the decision loop, Quant Analyst (p95 < 300ms), Sector Rotation (p95 < 300ms), and LLM Analyst (p95 < 8s) are independent of one another. Here p95 means the 95th percentile latency: 95% of requests finish within that time. Individual p95 values cannot be added to obtain the overall p95. Measure the complete execution, including join costs, to establish the effect of parallelism.

gantt
    title Latency budget for the decision loop
    dateFormat HH:mm:ss
    axisFormat %S

    section Sense
    Ticket creation and preprocessing :a1, 00:00:00, 1s

    section Think (parallel)
    Quant/Sector in parallel          :a2, 00:00:01, 1s
    LLM news analysis                 :a3, 00:00:01, 6s
    Debate (1-2 rounds)               :a4, 00:00:07, 2s
    Voting + risk gate                :a5, 00:00:09, 1s

    section Act
    Order plan + submit               :a6, 00:00:10, 1s
    Monitor/control                   :a7, 00:00:11, 10s

How fan-out works in LangGraph

When one node has edges to several nodes, LangGraph fans out automatically and runs them in parallel. Fan-in happens when several nodes point at the same next node; that node waits until every predecessor has finished.

# position_sensor -> [quant, llm, sector] : fan-out
graph.add_edge("position_sensor", "quant_analyst")
graph.add_edge("position_sensor", "llm_analyst")
graph.add_edge("position_sensor", "sector_rotation")
 
# [quant, llm, sector] -> analysis_merge : fan-in
graph.add_edge("quant_analyst", "analysis_merge")
graph.add_edge("llm_analyst", "analysis_merge")
graph.add_edge("sector_rotation", "analysis_merge")

The analysis_merge node combines the three agents' outputs into one report. Because each agent's result accumulates in the shared state, the join node can read all of them.

Checkpointing and durable execution

How the checkpointer behaves

The LangGraph checkpointer saves state snapshots at super-step boundaries. That enables the following.

A PostgreSQL-backed checkpointer

from langgraph.checkpoint.postgres import PostgresSaver
 
checkpointer = PostgresSaver.from_conn_string(DATABASE_URL)
app = graph.compile(checkpointer=checkpointer)
 
# Run
result = await app.ainvoke(
    initial_state,
    config={"configurable": {"thread_id": "pipeline-run-uuid"}}
)

thread_id is the primary key for checkpoint lookups. Rerunning with the same thread_id resumes automatically from the step after the last checkpoint.

Restart sequence

sequenceDiagram
    participant LG as LangGraph
    participant CP as Checkpointer
    participant DB as PostgreSQL

    Note over LG: Normal execution
    LG->>LG: Node 1 (Scanner) done
    LG->>CP: save_checkpoint(thread_id, state)
    CP->>DB: INSERT checkpoint

    LG->>LG: Node 2 (News) done
    LG->>CP: save_checkpoint(thread_id, state)
    CP->>DB: INSERT checkpoint

    LG->>LG: Node 3 (PositionSensor) done
    LG->>CP: save_checkpoint(thread_id, state)
    CP->>DB: INSERT checkpoint

    LG->>LG: Node 4 (Quant) raises an error

    Note over LG: Restart
    LG->>CP: get_checkpoint(thread_id)
    CP->>DB: SELECT latest checkpoint
    CP-->>LG: state as of Node 3 completion

    LG->>LG: Rerun starting at Node 4 (Quant)
    Note over LG: Nodes 1-3 are skipped

Durable execution principles

Durable execution is a core LangGraph pattern, and it follows these principles.

In trading, exactly-once execution is hard to guarantee perfectly because of the broker and exchange boundary. So each order carries an idempotency_key, and on restart an idempotency check verifies whether an order with that key was already submitted.

sequenceDiagram
    participant LG as LangGraph
    participant CP as Checkpointer
    participant EX as Trade Executor

    LG->>CP: save(state) at super-step N

    Note over LG: Failure or redeploy

    LG->>CP: load(thread_id) → latest checkpoint
    LG->>LG: resume from node after checkpoint

    LG->>EX: before submit, check idempotency_key
    EX-->>LG: if used → skip / reconcile
    EX-->>LG: if new → submit order

Human-in-the-loop

When a human is needed

In automated trading, some situations call for human confirmation rather than a purely mechanical judgment.

• Additional trades when the daily loss limit is close
• Sudden regime shifts such as a VIX spike or a black swan event
• Large buys made while news reliability is very low
• The first live trade right after a new strategy or prompt change

Implementing human-in-the-loop in LangGraph

The Risk Sentinel output carries a requires_human_approval flag, and the LangGraph interrupt() function pauses execution.

from langgraph.types import interrupt
 
def risk_sentinel(state: PipelineState) -> PipelineState:
    intent = state["final_decision"]
    portfolio = get_portfolio_snapshot(state["project_id"])
 
    decision = evaluate_risk(intent, portfolio)
 
    if decision.requires_human_approval:
        # interrupt(): pause and wait for human approval. The return value is what the human sends on resume.
        approved = interrupt(
            f"High-risk trade detected: {decision.reasons}"
        )
        state["final_decision"]["approved"] = approved
        return state
 
    state["final_decision"]["approved"] = decision.approved
    return state

A node that calls interrupt() stops at that point. When the user resumes with Command(resume=value), the node runs again from its start. The older approach of raising NodeInterrupt was replaced by the interrupt() function in LangGraph v0.4. Thanks to checkpoints, none of the steps before the pause have to run again.

Branching with conditional edges

Data quality gate

The pipeline entry point validates data quality and aborts the whole run when the data falls below the bar.

graph.add_conditional_edges(
    "data_quality_gate",
    lambda s: "continue" if not s["errors"] else "end",
    {"continue": "volume_scanner", "end": END},
)

The data quality gate checklist:

Risk Sentinel branch

The risk verdict routes the run to execute or to end.

graph.add_conditional_edges(
    "risk_sentinel",
    lambda s: "execute" if s["final_decision"].get("approved") else "end",
    {"execute": "trade_executor", "end": END},
)

The Risk Sentinel verdict has five possible values.

Event-sourced state persistence

A three-tier storage layout

State persistence in the trading system uses three tiers.

flowchart LR
    subgraph Kafka["Kafka (event log, immutable)"]
        E1[TICKET_CREATED]
        E2[NEWS_INGESTED]
        E3[QUANT_DONE]
        E4[DEBATE_DONE]
        E5[INTENT_CREATED]
        E6[RISK_DECISION]
        E7[ORDER_SUBMITTED]
        E8[FILL_RECEIVED]
    end

    subgraph Redis["Redis (mutable state cache)"]
        R1[Quote cache]
        R2[Agent weights]
        R3[Rate limiter]
        R4[Kill switch flag]
    end

    subgraph PG["PostgreSQL (durable storage)"]
        P1[LangGraph checkpoints]
        P2[Trade history]
        P3[Reflection memos]
        P4[Weight snapshots]
    end

Event envelope schema

Every message between agents uses a common envelope.

{
  "event_id": "uuid-v7",
  "ts": "2026-02-21T09:15:00.123+09:00",
  "event_type": "QUANT_DONE",
  "correlation_id": "pipeline-run-uuid",
  "project_id": "project-uuid",
  "agent": "quant_analyst",
  "payload": {
    "symbol": "005930",
    "signal": 0.72,
    "confidence": 0.68,
    "feature_vector": { "rsi_14": 32.5, "macd_signal": 0.003 }
  },
  "schema_version": "v2",
  "idempotency_key": "uuid-for-dedup"
}

idempotency_key and correlation_id identify duplicate processing and related events, respectively. The first prevents duplicate processing; the second ties together every event belonging to a single pipeline run.

Framework selection criteria

LangGraph's official persistence documentation distinguishes checkpointers from long-term stores. It describes implementation features, not financial performance or general superiority over another framework.

Test the required state storage, pause and resume behavior, and external-operation deduplication in a small example. The design below does not reproduce a financial-agent paper's architecture or returns.

Per-agent SLOs and failure modes

Each agent gets a service level objective (SLO) expressed as a latency target. In the table, p99 means the 99th percentile latency: 99% of requests finish within that time.

Latency and throughput targets

Per-agent I/O flow

flowchart TD
    VS["Volume Scanner<br/>OUT: SignalTicket[]"]
    NC["News Collector<br/>OUT: NewsItem[]"]
    PS["Position Sensor<br/>OUT: PositionAlert[]"]
    QA["Quant Analyst<br/>OUT: QuantReport"]
    LA["LLM Analyst<br/>OUT: NarrativeReport"]
    SR["Sector Rotation<br/>OUT: Regime, Rankings"]
    BD["Bull/Bear Debate<br/>OUT: DebateTranscript"]
    EV["Ensemble Voter<br/>OUT: TradeIntent"]
    RS["Risk Sentinel<br/>OUT: RiskDecision"]
    TE["Trade Executor<br/>OUT: Fills, IS(bps)"]

    VS --> QA & SR
    NC --> LA
    PS --> BD & EV & RS
    QA --> BD
    LA --> BD
    SR --> BD
    BD --> EV
    EV --> RS
    RS --> TE

IS in the Trade Executor output is implementation shortfall, the gap between the target price and the actual fill price, expressed in basis points (1bps = 0.01%).

Operations

Verification procedure

Use a simulated environment to check normal execution, interruption and resumption, duplicate requests, and external API failures. This is a design proposal, not proof of regulatory compliance or safe live trading. Before executing external actions, separately establish the environment's permissions and approval conditions.

Kill switch

The kill switch is an emergency control that cancels every open order and halts the pipeline.

sequenceDiagram
    participant USER as User/system
    participant KS as Kill Switch
    participant R as Redis
    participant TE as Trade Executor
    participant KIS as KIS API

    USER->>KS: Trigger the kill switch
    KS->>R: SET kill_switch:{project_id} = true
    KS->>TE: Cancel all open orders

    loop Each open order
        TE->>KIS: Cancel order
        KIS-->>TE: Cancellation confirmed
    end

    Note over KS: Release is manual only

Every LangGraph node checks the Redis kill switch flag before it runs. If the flag is set, the node exits immediately.

Infrastructure stack summary

LangGraph answers the question of how agents should be wired together and executed. It bundles a graph-based DAG, checkpointing, conditional branching, parallel execution, and human-in-the-loop into one framework, to structure execution without hand-extending a sequential pipeline. Production readiness requires separate validation.

References

• LangGraph Persistence: https://docs.langchain.com/oss/python/langgraph/persistence
• LangGraph Durable Execution: https://docs.langchain.com/oss/python/langgraph/durable-execution
• Event Sourcing Pattern: https://learn.microsoft.com/en-us/azure/architecture/patterns/event-sourcing

Summary

A sequential for-loop leaves failure recovery, parallel execution, conditional branching, state tracking, and human intervention all as things you must build yourself. LangGraph folds them into a single StateGraph: nodes and edges declare the execution flow, and a TypedDict holds the shared state. The checkpointer saves state at super-step boundaries, so a failed run resumes from its last point. interrupt() collects human approval before a high-risk decision. Conditional edges model the data quality gate and the risk gate naturally.

LangGraph fits domains such as trading, where checkpointing, conditional branching, and human-in-the-loop are all required. Verify the selected framework's resumption behavior and external-action boundaries through actual execution.