Skip to main content
Ecosystem & Marketplaces›References›LangGraph orchestration

LangGraph orchestration

TalkOps uses LangGraph to model multi-agent infrastructure workflows as stateful directed graphs. Rather than relying on unconstrained single-prompt loops, LangGraph gives TalkOps deterministic control flow, isolated subagent subgraphs, durable PostgreSQL checkpointing, and native interrupt() gates for Human-in-the-Loop approvals.

Typed State Schemas

Track messages, infrastructure plans, validation results, and approval flags in a strongly typed AgentState.

Hierarchical Subgraphs

Compose supervisor agents, domain coordinators, and isolated subagents into modular graphs.

Durable Checkpointing

Persist graph execution state to memory or PostgreSQL so long-running workflows survive restarts.

Human-in-the-Loop Interrupts

Pause graph execution before mutating infrastructure and resume seamlessly once an operator approves.

Core concepts​

State schema​

Every LangGraph workflow is backed by a TypedDict state schema. Reducers like add_messages append new messages automatically while scalar keys track current execution context:

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


class AgentState(TypedDict):
    # Conversation history with automatic message reducer
    messages: Annotated[list, add_messages]

    # Request metadata
    request_id: str
    user_id: str
    intent: str

    # Domain artifacts
    cloud_resources: Optional[dict]
    deployment_status: Optional[str]

    # Control flow and governance
    next_agent: str
    approval_pending: bool

Nodes​

Nodes are Python functions (or compiled subgraphs) that read the current AgentState, perform work or invoke MCP tools, and return a partial dictionary of state updates:

from langchain_core.messages import AIMessage


async def cloud_agent_node(state: AgentState) -> dict:
    """Execute cloud provisioning plan in an isolated node."""
    result = await provision_infrastructure(state["messages"])
    return {
        "cloud_resources": result,
        "messages": [AIMessage(content=f"Generated Terraform plan: {result['summary']}")],
    }

Conditional edges​

Conditional edges inspect the current state after a node finishes and return the name of the next node to execute:

from langgraph.graph import END


def route_to_next(state: AgentState) -> str:
    """Route workflow based on approval and resource state."""
    if state.get("approval_pending"):
        return "human_approval"
    if state.get("cloud_resources"):
        return "cicd_agent"
    return END

Building a state graph​

Combine state schemas, nodes, and conditional edges into a compiled StateGraph:

from langgraph.graph import END, StateGraph

# 1. Initialize graph with typed state
graph = StateGraph(AgentState)

# 2. Register nodes
graph.add_node("supervisor", supervisor_node)
graph.add_node("cloud_agent", cloud_agent_node)
graph.add_node("cicd_agent", cicd_agent_node)

# 3. Wire edges and conditional routing
graph.set_entry_point("supervisor")
graph.add_edge("supervisor", "cloud_agent")
graph.add_conditional_edges(
    "cloud_agent",
    route_to_next,
    {
        "cicd_agent": "cicd_agent",
        END: END,
    },
)

# 4. Compile executable graph
app = graph.compile()

Multi-agent patterns in TalkOps​

1. Supervisor routing with Command​

A central supervisor node analyzes the user's intent and uses LangGraph's Command primitive to update state and jump directly to the target specialist:

from langgraph.types import Command


async def supervisor_node(state: AgentState) -> Command:
    """Analyze request intent and route to the appropriate specialist."""
    response = await llm.ainvoke([
        SystemMessage(content="Classify the infrastructure domain: cloud, cicd, or end."),
        *state["messages"],
    ])
    decision = response.content.strip().lower()

    if "cloud" in decision:
        return Command(goto="cloud_agent", update={"intent": "cloud"})
    if "cicd" in decision:
        return Command(goto="cicd_agent", update={"intent": "cicd"})
    return Command(goto=END)

2. Hierarchical subgraphs​

Complex agents like the Kubernetes Agent and OpsCode compile domain-specific subgraphs and mount them as nodes inside a parent supervisor graph:

# Subgraph 1: Infrastructure provisioning
infra_graph = StateGraph(AgentState)
infra_graph.add_node("aws_agent", aws_node)
infra_graph.add_node("terraform_validator", validator_node)
infra_graph.add_edge("aws_agent", "terraform_validator")
infra_team = infra_graph.compile()

# Subgraph 2: Application delivery
delivery_graph = StateGraph(AgentState)
delivery_graph.add_node("helm_agent", helm_node)
delivery_graph.add_node("argocd_agent", argocd_node)
delivery_graph.add_edge("helm_agent", "argocd_node")
delivery_team = delivery_graph.compile()

# Parent supervisor graph
main_graph = StateGraph(AgentState)
main_graph.add_node("supervisor", main_supervisor_node)
main_graph.add_node("infra_team", infra_team)
main_graph.add_node("delivery_team", delivery_team)

Checkpointing and persistence​

Checkpointers persist the graph state after every super-step so conversations and multi-stage rollouts can pause and resume across process restarts:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

# Development: in-memory checkpointer
memory = MemorySaver()
dev_app = graph.compile(checkpointer=memory)

# Production: durable PostgreSQL checkpointer
async with AsyncPostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    prod_app = graph.compile(checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "deploy-eks-prod-01"}}
    result = await prod_app.ainvoke(initial_state, config=config)

Human-in-the-Loop interrupts​

TalkOps uses LangGraph's interrupt() primitive to pause execution before mutating live infrastructure. The graph state is saved to the checkpointer until the operator approves or rejects the plan:

from langgraph.types import Command, interrupt


def approval_node(state: AgentState) -> dict:
    """Pause graph execution and request human review of the deployment diff."""
    decision = interrupt({
        "type": "approval_request",
        "plan": state.get("cloud_resources"),
        "options": ["approve", "reject", "modify"],
    })

    if decision.get("action") == "approve":
        return {"approval_pending": False, "deployment_status": "approved"}
    return {"approval_pending": False, "deployment_status": "rejected"}


# Resume a paused thread after human approval
await app.ainvoke(
    Command(resume={"action": "approve"}),
    config={"configurable": {"thread_id": "deploy-eks-prod-01"}},
)

Streaming events​

Stream token chunks, tool executions, and node transitions in real time to your CLI or web console:

async for event in app.astream_events(input_state, config=config, version="v2"):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="", flush=True)
    elif event["event"] == "on_tool_end":
        print(f"\n[Tool Completed] {event['name']}")

Production best practices​

  • Keep state schemas flat and serializable: Store concise summaries and resource references in AgentState rather than multi-megabyte raw CLI logs.
  • Isolate noisy subagent loops: Use dedicated subgraphs (or BranchMemoryStore) for iterative tasks like terraform validate retries so intermediate errors don't pollute the supervisor's context window.
  • Always configure a checkpointer when using interrupt() gates—human approval requires persisted thread state to resume.
  • Bound cyclic retry loops: Track a retry_count integer in state and route to an error handler node once retry_count exceeds your threshold (typically 3).
On this page