agent-orchestration

Agent Orchestration

Design and implement multi-agent systems where an orchestrator coordinates specialized subagents through task planning, intelligent routing, compliance validation, and human-in-the-loop approval.

When to Use

• Coordinating multiple specialized agents (frontend, backend, devops)
• Implementing task decomposition and delegation workflows
• Building a LangGraph StateGraph with routing logic
• Adding human-in-the-loop approval gates for destructive operations
• Creating agent pipelines that validate compliance before merging work

Available Operations

1. Define orchestrator state schema
2. Build the orchestrator graph with nodes and edges
3. Implement intelligent task routing
4. Delegate tasks to specialized subagents
5. Add compliance validation nodes
6. Wire up human-in-the-loop interrupt gates
7. Merge work from multiple agent branches

Multi-Step Workflow

Step 1: Define the Orchestrator State

Create a typed state schema that tracks the full lifecycle of a demand.

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

class OrchestratorState(TypedDict):
    messages: Annotated[list, add_messages]
    demand: dict
    task_plan: list[dict]
    current_task_index: int
    assigned_agents: dict[str, str]  # task_id -> agent_type
    branch_status: dict[str, str]    # agent -> branch_name
    compliance_results: list[dict]
    needs_human_review: bool
    final_status: str

Step 2: Build the Graph Structure

Wire up the orchestrator nodes and conditional edges.

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver

graph = StateGraph(OrchestratorState)

# Add nodes
graph.add_node("analyze_demand", analyze_demand_node)
graph.add_node("plan_tasks", plan_tasks_node)
graph.add_node("route_to_agent", route_to_agent_node)
graph.add_node("execute_agent", execute_agent_node)
graph.add_node("validate_compliance", validate_compliance_node)
graph.add_node("human_review", human_review_node)
graph.add_node("merge_work", merge_work_node)

# Wire edges
graph.add_edge(START, "analyze_demand")
graph.add_edge("analyze_demand", "plan_tasks")
graph.add_edge("plan_tasks", "route_to_agent")
graph.add_conditional_edges("route_to_agent", should_continue, {
    "execute": "execute_agent",
    "all_done": "validate_compliance"
})
graph.add_edge("execute_agent", "route_to_agent")
graph.add_conditional_edges("validate_compliance", needs_review, {
    "review": "human_review",
    "pass": "merge_work"
})
graph.add_edge("human_review", "merge_work")
graph.add_edge("merge_work", END)

app = graph.compile(checkpointer=PostgresSaver(conn))