langchain-dynamic-workflow

English | 中文

Deterministic, scripted, resumable multi-agent orchestration for LangChain deepagents — a community port of Claude Code's Dynamic Workflows.

A normal agent decides its control flow turn by turn: every loop, branch, and fan-out lives in the model's context window, burning tokens and accumulating intermediate state. langchain-dynamic-workflow inverts that — a deterministic orchestration script owns the control flow, and only leaf agent() calls delegate to deepagents, each running in an isolated, discarded context, so only the final result reaches the caller's context.

	Normal agent	Dynamic workflow
Who decides the next step	the LLM, turn by turn	the script (deterministic code)
Where intermediate results live	the model's context window	script variables
What reaches the caller's context	the whole trajectory	only the final result

Table of contents

• Why
• Highlights
• Architecture
• Installation
• Quickstart
• Host agents and the meta layer
• Resume, budget, and observability
• Examples
• Public API
• Development
• Status
• License

Why

Turn-by-turn control flow has three costs that compound as a task grows: the context window fills with intermediate reasoning, the trajectory is non-deterministic, and an interrupted run cannot resume without replaying the model. Inverting control flow addresses all three at once — the script holds the loops and branches, intermediate results stay in plain variables, and a content-hash journal makes a resumed run replay completed work at zero model cost.

Reach for it when a task is fan-out heavy (research N angles, grade M candidates), long and multi-step (the trajectory would otherwise overflow the context), or needs deterministic resume and a shared token budget across many sub-agents.

Highlights

• Deterministic control flow — loops, branching, and fan-out live in code, not in the model's head.
• Context quarantine — each leaf runs in a fresh, discarded deepagents context; only its folded result returns.
• Parallel, pipeline, and race fan-out — parallel() (blocking barrier), pipeline() (no-barrier streaming), and race() (best-of-N early exit: the first result satisfying win wins, in-flight losers are cancelled, the decision is journaled so resume reproduces the winner) over a shared concurrency gate.
• Cross-leaf reduce — first-class helpers fold many leaf results: survives (refute-by-default voting), dedup, reconcile (double-blind reviewer reconciliation), and corroborate (judge-panel aggregation).
• Resumable by content hash — a success-only journal replays completed leaves on resume at zero model cost.
• Cross-session persistence — wire the optional [sqlite] store and a fresh process resumes a run by run_id, replaying completed leaves at zero model cost (a superset of Claude Code's same-session-only resume).
• Fail-loud determinism guard — a replay whose agent() call sequence diverges raises rather than serving a positionally misaligned cache entry.
• Shared token budget — one ceiling across every leaf, with a loop-until-budget idiom.
• Observability by default — every agent / parallel / pipeline / race call emits a span to an opt-in sink (zero cost when unset).
• Large fan-out batch map — ctx.batch_map streams an Iterable / AsyncIterable through a bounded admission window, so thousands of items never materialize thousands of tasks; results return in input order with automatic live count/ETA progress.
• Background-run live drill-in — a detached background run is no longer opaque: its runtime events are buffered on its slot and a live turn replays them as the same per-leaf cards an inline run streams, so a host can drill into a running background job and report back each run's result on demand.
• Per-leaf sandbox isolation + real execution — execution leaves lease isolated backends (a /shared/ route enables explicit producer→consumer hand-off); opt into LocalSubprocessSandbox for real subprocess build/test with exit-code gating, resource limits, and admission control.

Architecture

Three layers, with a one-directional dependency (Layer 2 → Layer 1 → Layer 0) that import-linter enforces mechanically:

• Layer 0 — substrate: LangGraph durable execution (@entrypoint + @task + checkpointer). Provides resume, replay, and cached-result skip.
• Layer 1 — orchestration runtime: the primitives — agent(), parallel() (barrier), pipeline() (no barrier), race() (best-of-N early exit), phase(), log(), budget, workflow() — plus the two patches LangGraph lacks: a content-hash journal (LangGraph's native cache is index-based) and a fail-loud determinism guard (LangGraph treats determinism as convention).
• Layer 2 — meta layer: an LLM authors a Python orchestration script; an AST gate validates it (no imports, dunders, or banned names) before a restricted-builtins exec.

Leaf agent() calls resolve a named deepagent from a registry (roster) and invoke it as a @task, reusing deepagents' context quarantine and sandbox backends.

Installation

uv sync   # install dependencies + create .venv

Python 3.12+. Dependency management via uv. The package is not yet published to PyPI; install it from a clone or as a git dependency.

Quickstart

Write an orchestration script (the ctx exposes the primitives), register your leaf agents in a roster, and run it. A leaf is any runnable whose state has a messages key — typically a deepagents.create_deep_agent(...):

import asyncio

from deepagents import create_deep_agent
from langchain_dynamic_workflow import Ctx, Roster, run_workflow


async def main() -> None:
    # 1. Register leaf agents by name (build-time wiring; the agent never does this).
    roster = Roster().register(
        "researcher",
        create_deep_agent(model="anthropic:claude-haiku-4-5"),
        description="Researches one topic",
    )

    # 2. The orchestration script owns the control flow; only leaf agent() calls
    #    delegate to deepagents. parallel() is a blocking barrier; a failed leaf
    #    lands as None (filter the holes) and never aborts the barrier.
    async def orchestrate(ctx: Ctx) -> str:
        ctx.phase("research")
        findings = await ctx.parallel(
            [
                lambda t=topic: ctx.agent(f"Research {t}", agent_type="researcher")
                for topic in ("batteries", "solar", "wind")
            ]
        )
        surviving = [f for f in findings if f is not None]
        return f"synthesized {len(surviving)} findings: " + " | ".join(surviving)

    # 3. Run it. Only the final result reaches you — not the whole trajectory.
    result = await run_workflow(orchestrate, roster=roster)
    print(result)


asyncio.run(main())

Host agents and the meta layer

To let a host agent drive workflows in the background, attach create_workflow_middleware(roster, workflows=...) to a host create_deep_agent. The agent then controls runs through a single workflow tool:

Command	Effect
run	Launch a registered workflow by name; returns a run_id immediately.
run_script	Launch an ad-hoc script the agent authors on the spot (the meta layer).
status	Poll a run and fetch its result (large results are offloaded behind a handle).
resume	Re-run against the same journal so completed leaves replay at zero cost.
cancel	Stop an in-flight run.

A run executes in the background and a completion notice is injected before the host's next turn, so launching never blocks the conversation.

The meta layer (run_script). The host writes an async def orchestrate(ctx, args) and submits the source. It passes an AST security gate (no imports, dunder access, banned builtins, or str.format injection) and then runs under a restricted-builtins namespace via a single, confined exec. A rejected script returns its exact violations so the host can fix them and resubmit. Build-time code can do the same programmatically with run_workflow_from_source(source, roster=...).

Security boundary. The gate plus the restricted namespace stop an honest model's slip — they are not a security sandbox, and a determined adversarial script can still escape. Submit only scripts the host authors itself; for adversarial input, run the engine behind an out-of-process isolation backend.

⚠️ Real shell execution is a DANGEROUS opt-in — NOT a security sandbox

By default, execution leaves use the offline in-memory backend and run no real shell. You can opt a SandboxManager into a real local-subprocess backend (SandboxManager(sandbox_factory=local_subprocess_factory(ExecPolicy(...)))), which runs each leaf's shell command in a private per-leaf temporary directory. This is not a security sandbox — read this before enabling it.

What a command CAN still do — it runs as you, on your host:

• Read and write any host path via absolute paths. The per-leaf temp directory bounds only the default working directory, not the reachable filesystem. cat ~/.ssh/id_rsa or rm -rf /some/path is fully within reach.
• Use the network with your user's permissions — exfiltrate, download, connect anywhere your account can.
• Consume host resources beyond the best-effort POSIX rlimits — the limits are a runaway guard, not a hard cgroup-grade containment.

What it DOES guarantee (resilience, not confinement):

• A private temp working directory per leaf — no accidental execution in the engine's own directory.
• The engine file APIs stay rooted, with .. traversal rejected.
• A bounded effective timeout with best-effort process-group kill (SIGTERM → grace → SIGKILL on POSIX).
• Bounded combined stdout+stderr output (flagged when truncated).
• A bounded count of concurrent executions.

What it deliberately does NOT provide (deferred sharp edges):

• Hard filesystem confinement (container / chroot / nsjail / firejail).
• Network-egress blocking.
• cgroup-grade CPU / memory / process containment.
• Strong orphan cleanup