Distributed Agent Orchestration: Implementing a Hierarchical Planner in Python

# Distributed Agent Orchestration: Implementing a Hierarchical Planner in Python

*By an OnlineCloud Principal AI Engineer — 9 min read*


Every serious agent system I’ve reviewed in production eventually hits the same wall: a single prompt — no matter how well-crafted — cannot reliably execute a task with more than a few dependent steps. The fix isn’t a better prompt. It’s a better control structure. This post walks through the architecture and Python implementation of a **hierarchical planner** — a Planner–Worker–Verifier control loop that decomposes work, executes it in parallel, verifies it, and heals itself when the world disagrees with the plan.

## 1. The Bottleneck of Linear Planning

The default agent design is a loop: model → tool call → observation → model. For a two-step task like “look up an order and summarize its status,” that loop is fine. For anything real — a logistics reroute, a multi-system data migration, a claims workflow — the loop becomes a liability. Three failure modes dominate:

**Error compounding.** Each step consumes context and injects drift. If step three of eight produces a subtly wrong intermediate value, steps four through eight are all downstream of garbage. A linear chain has no structural way to notice, because nothing is *checking* intermediate outputs against their specification.

**Lost locality.** A single model context holds the entire task, so every decision is entangled with every other. The model reasoning about carrier APIs is also holding route geometry, pricing data, and customer constraints in one window. Context limits force truncation, and truncation is where silent mistakes are born.

**Zero recovery semantics.** When a linear loop fails, the only move is “try the whole thing again” — replaying cost, latency, and the same odds of failure. There’s no notion of *which subgoal* failed, no notion of *retrying only that*, and no budget except your own patience.

The pattern that fixes all three is older than LLMs: **divide and conquer, with inspection**. Hierarchical task decomposition. What LLMs add is that the decomposition — and the inspection — can be generated dynamically per request instead of hard-coded.

## 2. The Hierarchical Pattern: Planner–Worker–Verifier

The architecture has three roles and one shared contract. The **Planner** takes a goal and a world snapshot and produces a structured plan: an ordered list of typed subgoals, each with an owner, inputs, and acceptance criteria. The **Workers** execute subgoals against real systems — they are narrow, single-purpose, and stateless with respect to the overall goal. The **Verifier** is the quality gate: it checks each worker’s output against the subgoal’s acceptance criteria and returns a verdict with *machine-readable feedback*, not just a score.

“`
┌─────────────┐
goal ───────────────▶│ Planner │◀───── revision context
└──────┬──────┘
│ plan (typed subgoals)
┌────────────────┼─────────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ Worker A│ │ Worker B│ │ Worker C│ ← parallel fan-out
└────┬────┘ └────┬────┘ └────┬────┘
└────────────────┼─────────────────┘
▼
┌─────────────┐
│ Verifier │── ACCEPT ──▶ consolidate / ship
└──────┬──────┘
│ REVISE + feedback (bounded loop)
▼
Planner re-plans failed subgoals only
“`

Three properties make this work in practice. First, **the plan is data, not prose**: typed subgoal objects with explicit acceptance criteria, so verification is a deterministic check layered over an LLM judgment call. Second, **workers are interchangeable**: because each subgoal declares its type and inputs, you can swap an LLM worker for a deterministic function or a human approval step without touching the orchestration layer. Third, **feedback flows up, not just down**: the verifier’s critique returns to the planner as structured context, which is what converts a blind retry into a *directed* revision.

## 3. Stateful Planning in Practice: LangGraph vs. PydanticAI

The hierarchy only works if the runtime gives you three things: explicit state, controllable control flow, and durable checkpoints. As of this writing, the two Python frameworks I’d put in front of a client are **LangGraph** and **PydanticAI**.

**LangGraph** models the agent as a state machine: nodes are Python functions, edges are explicit transitions, and a typed state object flows through every node. Its superpowers are *conditional edges* (the verifier verdict literally routes the graph) and *checkpointing* — every state transition can be persisted, so a crashed run resumes mid-plan rather than restarting. It’s the right choice when your orchestration is genuinely complex: branching, loops, parallel fan-out, human-in-the-loop interrupts.

**PydanticAI** is the opposite philosophy: agents are plain typed functions with dependency injection, and multi-agent work is *delegation* — one agent calls another as a tool, with the model deciding when. It’s dramatically more readable for the 80% case, and its `RunContext` gives you typed deps without globals. The tradeoff: the control flow lives inside the model’s tool-calling loop, so bounded retry loops and checkpoints are something you build rather than inherit.

My rule of thumb: if your hierarchy is **two levels and mostly linear**, use PydanticAI and keep the code boring. If you have **verifier-gated revision loops, parallel fan-out, or long-running stateful tasks**, use LangGraph — because those are exactly the features you shouldn’t hand-roll. The walkthrough below uses LangGraph, with the hierarchical *structure* portable to either.

## 4. Orchestration: Multi-Agent Communication and State Consistency

Once you have multiple agents mutating shared state, the hard problem stops being intelligence and starts being *consistency*. Five disciplines keep distributed agent state from rotting:

**One shared state schema, owned by the graph.** Agents never pass unstructured strings to each other. Everything they exchange — plans, results, feedback, spend — is a typed field on a single state object. In LangGraph this is the state schema; in any design it should be a versioned Pydantic model, because the schema *is* the contract between agents.

**Reducers for everything append-only.** Messages, task results, and feedback accumulate via merge functions, never via overwrite. If two workers finish concurrently, their results append into an ordered log — no last-writer-wins races. This is `operator.add` or a custom reducer over an annotated channel, and it eliminates the most common class of heisenbugs in multi-agent systems.

**Idempotency keys at the system boundary.** The real world is not idempotent: booking a carrier, writing a row, sending a notification. Every worker that touches an external system carries a client-generated request ID (`order_id + task_id + attempt`), so a retried execution cannot double-book. If the downstream API doesn’t support idempotency keys, wrap it in a small outbox table that does — dedupe on write, exactly-once on retry.

**Checkpoints, not just logs.** A log tells you what happened; a checkpoint lets you *resume* from it. LangGraph’s checkpointer serializes full graph state after every super-step. MemorySaver is fine for demos; production wants Postgres or Redis-backed persistence so a worker crash — or a full process restart — resumes the exact plan with spend and attempts intact.

**Tracing with a correlation ID.** Every agent appends to a `trace` list on its result: `(node, step, model, cost_ms)`. Ship those into OpenTelemetry with the run’s correlation ID, and you can answer the two questions every incident starts with: *which subgoal failed, and how much did we spend before it did?*

## 5. Error Recovery: Self-Healing Loops and Backtracking

The verifier is what makes recovery *surgical* instead of *nuclear*. The control loop is deliberately small and brutal:

1. Worker output fails verification (schema-invalid, acceptance criteria unmet, or external call errored).
2. The verifier returns structured feedback: *which subgoal, which acceptance criterion, what evidence is missing*.
3. The router decides the recovery granularity: **retry the worker** (transient failure, same plan), **revise the plan** (the plan itself was wrong — planner gets the feedback and re-plans only the failed subgoals), or **escalate** (budget exhausted, or two consecutive revisions failed — hand to a human with the full trace).

Three guardrails keep the self-healing loop from becoming a self-immolating one:

**Bounded attempts, counted in state.** `attempts` and `revision_count` live in the state schema and are checked by the router. There is no unbounded “keep trying” — there is a counter, a limit, and an escalation path.

**Budget as a first-class state field.** Track `spent` against `budget` on every LLM call (input + output tokens priced per model). When spend crosses the ceiling, the router must escalate, not retry. Cost ceilings are a *control-flow* concern, not an accounting one.

**Backoff with jitter on external calls.** Transient API failures are the most common error in real systems. Workers retry with exponential backoff plus jitter; the *orchestrator* never retries — it just marks the subgoal failed and lets the router decide. That separation keeps failure handling at exactly one layer.

This gives you the property that matters most in production: **a failed run terminates in one of two states — a verified result, or an escalation with a readable trace.** Never a silent half-done state.

## 6. Code Walkthrough: A Hierarchical Logistics Agent

Let’s build it. The scenario: a customer requests a *priority reroute* for an in-transit shipment. The planner decomposes the goal into typed subgoals — validate the order, replan the route, match carriers with live rates, run a compliance check, estimate ETA and cost. Workers execute each subgoal against mocked external systems, and a verifier gates acceptance. The revision loop re-plans only the failed subgoal.

“`python
# agent_state.py
from __future__ import annotations
import operator
from typing import Annotated, Any, Literal, TypedDict

from pydantic import BaseModel, Field

# —- Shared contract: the typed plan ————————————-

class TaskSpec(BaseModel):
“””One subgoal. The acceptance_criteria field is what makes
verification deterministic instead of vibes-based.”””
task_id: str
kind: Literal[“validate_order”, “route_plan”, “carrier_match”,
“compliance_check”, “eta_cost”]
goal: str
inputs: dict[str, Any] = Field(default_factory=dict)
acceptance_criteria: list[str] = Field(default_factory=list)

class TaskResult(BaseModel):
task_id: str
kind: str
status: Literal[“ok”, “error”, “needs_input”]
output: dict[str, Any] = Field(default_factory=dict)
error: str | None = None
trace: list[str] = Field(default_factory=list) # correlation trail

class Verdict(BaseModel):
accepted: bool
issues: list[str] = Field(default_factory=list) # machine-readable feedback

# —- Graph state: single source of truth for every agent ——————

class LogisticsState(TypedDict, total=False):
order: dict[str, Any] # immutable request payload
plan: list[TaskSpec] # produced by the Planner
results: Annotated[list[TaskResult], operator.add] # append-only channel
feedback: Annotated[list[str], operator.add] # verifier critiques
attempts: int # bounded self-healing counter
budget: float # hard cost ceiling (USD)
spent: float # accumulated token cost
accepted_plan: dict[str, Any] # verified, ready to ship
needs_human: bool
“`

The state schema is the contract: workers read `order` and write `results`; the verifier reads `results` and writes `feedback`; only the router reads the counters. Nothing is shared outside the graph, and every channel that multiple nodes write to is append-only.

“`python
# graph.py
import asyncio, json, uuid
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command # noqa (used with interrupt for humans)

from agent_state import LogisticsState, TaskResult, TaskSpec, Verdict

MAX_ATTEMPTS = 3
REVISION_BUDGET_USD = 1.50 # hard ceiling per run

# —- Workers: narrow, single-purpose, registered by kind ——————

WORKERS: dict[str, callable] = {} # populated by @worker decorator below

def worker(kind: str):
def deco(fn):
WORKERS[kind] = fn
return fn
return deco

@worker(“validate_order”)
async def validate_order(order: dict) -> dict:
# Deterministic gate: no LLM needed where a schema check suffices.
if order.get(“status”) != “in_transit”:
raise ValueError(f”order {order[‘order_id’]} is not in transit”)
if not order.get(“items”):
raise ValueError(“order has no line items”)
return {“valid”: True, “shipment_id”: order[“shipment_id”]}

@worker(“route_plan”)
async def route_plan(order: dict, spent: float) -> dict:
# Mocked as deterministic for the walkthrough; in production this is an
# LLM call priced against the shared budget.
return {“segments”: [“MEM → DFW”, “DFW → LAX”], “via”: “air”}

@worker(“carrier_match”)
async def carrier_match(order: dict) -> dict:
# Idempotency: same (order, task) always sends the same idempotency key.
idem = f”carrier:{order[‘order_id’]}:{uuid.uuid5(uuid.NAMESPACE_URL, json.dumps(order, sort_keys=True))}”
# …POST /carriers/quote with Idempotency-Key: idem…
return {“carrier”: “AeroFreight”, “rate”: 412.50, “confirmed_slots”: 1}

@worker(“compliance_check”)
async def compliance_check(order: dict) -> dict:
return {“hazmat”: “ok”, “export_control”: “ok”, “blocked_regions”: []}

@worker(“eta_cost”)
async def eta_cost(order: dict, segments: list[str]) -> dict:
return {“eta_days”: 2, “total_cost_usd”: 1340.0, “currency”: “USD”}
“`

Notice what the workers are *not* doing: none of them decide the next step, none of them read the verifier’s feedback, and none of them know a plan exists. That ignorance is the point — it makes them trivially testable and swappable.

“`python
# —- Planner: LLM call that turns a goal + feedback into typed subgoals —-

async def call_llm_json(system: str, user: str, budget: dict) -> dict:
# Real implementation: your provider SDK + token accounting that
# increments budget[“spent”]. Kept abstract so the graph logic stays
# framework-agnostic.
…

async def planner_node(state: LogisticsState) -> dict:
“””Decompose the goal into typed subgoals. On revision rounds, the
verifier’s feedback is injected so the planner only re-plans what broke.”””
if not state[“plan”]:
system = “You are the planning layer of a logistics agent. ” \
“Emit subgoals as JSON only.”
else:
system = “The plan below failed verification. Re-plan ONLY the ” \
“subgoals implicated by feedback. Keep accepted subgoals.”
payload = await call_llm_json(
system=system,
user=json.dumps({“order”: state[“order”],
“feedback”: state.get(“feedback”, [])}),
budget=state)
tasks = [TaskSpec(**t) for t in payload[“subgoals”]]
return {“plan”: tasks,
“attempts”: state[“attempts”] + 1,
“spent”: payload[“cost”]}
“`

The planner node’s only job is to turn *goal + world + feedback* into a typed plan. Every subsequent node treats that plan as data, which means the verifier can hold the plan and the results side by side — that’s the whole trick.

“`python
# —- Executor: fan out to workers, gather, and trace ———————–

async def executor_node(state: LogisticsState) -> dict:
“””Run every planned subgoal concurrently. Each worker writes one
TaskResult into the append-only results channel.”””

async def run_one(spec: TaskSpec) -> TaskResult:
try:
fn = WORKERS[spec.kind]
out = await fn(**spec.inputs, spent=state[“spent”])
return TaskResult(task_id=spec.task_id, kind=spec.kind,
status=”ok”, output=out,
trace=[f”{spec.kind}:ok”])
except Exception as exc: # surface, don’t swallow
return TaskResult(task_id=spec.task_id, kind=spec.kind,
status=”error”, error=str(exc),
trace=[f”{spec.kind}:error:{exc!r}”])

results = await asyncio.gather(*(run_one(t) for t in state[“plan”]))
return {“results”: list(results)}
“`

Errors are captured into `TaskResult` objects, never raised through the graph. A raised exception would kill the whole run; a captured error is *just another input to the router*, which is exactly where failure-handling belongs.

“`python
# —- Verifier: deterministic checks layered over one LLM judgment ———-

async def verifier_node(state: LogisticsState) -> dict:
“””Gate every subgoal against its acceptance criteria.”””
by_id = {r.task_id: r for r in state[“results”]}
issues: list[str] = []

for spec in state[“plan”]:
res = by_id.get(spec.task_id)
if res is None:
issues.append(f”{spec.task_id}: no result produced”)
elif res.status == “error”:
issues.append(f”{spec.task_id}: {res.error}”)
elif res.status == “ok”:
# 1) Structural check — deterministic, zero token cost.
missing = [c for c in spec.acceptance_criteria
if not _criterion_satisfied(c, res.output)]
issues += [f”{spec.task_id}: {m}” for m in missing]

if not issues:
return {“accepted_plan”: _consolidate(state), “needs_human”: False}

verdict = await call_llm_json(
system=”You are the verifier. Given failing subgoals, decide: ”
“RETRY_WORKER (transient) or REVISE_PLAN (plan was wrong).”,
user=json.dumps({“issues”: issues, “attempts”: state[“attempts”]}),
budget=state)
return {“feedback”: verdict[“issues”],
“spent”: state[“spent”] + verdict[“cost”]}
“`

“`python
# —- Router: the self-healing control flow ——————————–

def router(state: LogisticsState) -> str:
“””Where the graph goes next depends on counters in state — the whole
recovery policy lives in this one pure function.”””
if state.get(“accepted_plan”):
return “accept”
exhausted_budget = state[“spent”] >= state[“budget”]
out_of_attempts = state[“attempts”] >= MAX_ATTEMPTS
if exhausted_budget or out_of_attempts:
return “escalate”
# REVISE_PLAN ⇒ loop back through the planner with feedback attached;
# RETRY_WORKER would route back to the executor instead. Both are just
# conditional edges — no bespoke retry framework required.
return “revise”

# —- Assemble the graph —————————————————-

builder = StateGraph(LogisticsState)
builder.add_node(“planner”, planner_node)
builder.add_node(“execute”, executor_node)
builder.add_node(“verify”, verifier_node)
builder.add_node(“human_review”, lambda s: {“needs_human”: True})

builder.add_edge(START, “planner”)
builder.add_edge(“planner”, “execute”)
builder.add_edge(“execute”, “verify”)
builder.add_conditional_edges(“verify”, router, {
“accept”: END,
“revise”: “planner”, # ← backtrack: only failed subgoals re-plan
“escalate”: “human_review”, # ← human checkpoint with full trace
})
builder.add_edge(“human_review”, END)

graph = builder.compile(checkpointer=MemorySaver()) # PostgresSaver in prod

async def main() -> None:
order = {“order_id”: “SO-88231”, “status”: “in_transit”,
“shipment_id”: “SH-441”, “items”: [{“sku”: “A1”, “qty”: 40}]}
thread = {“configurable”: {“thread_id”: “run-” + uuid.uuid4().hex[:8]}}
final = await graph.ainvoke(
{“order”: order, “plan”: [], “attempts”: 0,
“budget”: REVISION_BUDGET_USD, “spent”: 0.0},
config=thread)
print(json.dumps(final.get(“accepted_plan”) or
{“escalated”: True, “trace”: final[“results”]}, indent=2))

if __name__ == “__main__”:
asyncio.run(main())
“`

Three details in this walkthrough carry the whole design. **The verifier is layered**: deterministic structural checks run first at zero token cost, and the LLM verdict is only consulted when structure fails — that’s how you keep verification cheap enough to run on every attempt. **The router is a pure function over state counters**: recovery policy is testable without any LLM in the loop (feed it a state dict, assert the edge). **The revision loop backtracks to the planner, not to the start**: `attempts` is bounded at three and `budget` at $1.50 per run, and crossing either one routes to a human checkpoint with every worker trace intact — OnlineCloud’s “plan/act/reflect with cost ceilings and human checkpoints” pattern, exactly.

If you squint, the whole graph reads like a state machine version of what a good senior dispatcher does: break the job into pieces someone can own, check the pieces before you commit, and when something breaks, fix the *piece* — not the whole shipment.

## Building This for Your Systems

The hierarchy above is deliberately framework-light at the edges: workers are plain functions, the verifier is a JSON contract, and only the control flow is graph-shaped. That’s what makes it portable across LangGraph, PydanticAI, or a hand-rolled asyncio loop — and what makes it auditable, because every transition is a checkpoint and every decision is a state field.

A hierarchical planner is the difference between an agent that *looks* autonomous in a demo and one that survives contact with real databases, real auth boundaries, and real cost ceilings. When you’re ready to move past the demo, the architects at OnlineCloud design and ship exactly this class of system — MCP-native orchestration with evaluated, observable, human-checkpointed control loops, in production in 6–10 weeks.

**Have a multi-step agent workload that needs real orchestration? Talk to an OnlineCloud architect** — [contact@onlinecloud.dev](mailto:contact@onlinecloud.dev), or reach out via [onlinecloud.dev](https://www.onlinecloud.dev/). Free initial consultation, US-based senior engineers, response within 24 hours.