AgentFlow

Orchestrate Codex, Claude, Kimi, and Pi agents in dependency graphs with parallel fanout, iterative cycles, and execution on local Docker, SSH, EC2, or ECS targets.

Typed actor nodes, explicit agent profiles, credential-safe Docker preparation, and the Harbor Terminus 2 adapter are described in Actors and profiles.

94-node pipeline: plan → 64 workers → 8 batch merges → 16 reviews → 4 review merges → synthesis

Install / Upgrade

One line:

curl -fsSL https://raw.githubusercontent.com/shouc/agentflow/master/install.sh | bash

This installs agentflow, adds it to PATH, and installs the skill for Codex and Claude Code.

Or manually:

python3 -m venv .venv && . .venv/bin/activate
pip install -e .[dev]

Quick Start

from agentflow import Graph, codex, claude

with Graph("my-pipeline", concurrency=3) as g:
    plan = codex(task_id="plan", prompt="Inspect the repo and plan the work.", tools="read_only")
    impl = claude(task_id="impl", prompt="Implement the plan:\n{{ nodes.plan.output }}", tools="read_write")
    review = codex(task_id="review", prompt="Review:\n{{ nodes.impl.output }}")
    plan >> impl >> review

print(g.to_json())

agentflow run pipeline.py --output summary

Or just ask Codex (the agentflow skill is auto-installed):

codex "Use agentflow to fan out 10 codex agents, each telling a unique joke, then merge their outputs and pick the funniest one. Write the pipeline and run it."

Parallel Fanout

Fan a node into many parallel copies with fanout():

from agentflow import Graph, codex, fanout, merge

with Graph("code-review", concurrency=8) as g:
    scan = codex(task_id="scan", prompt="List the top 5 files to review.")
    review = fanout(
        codex(task_id="review", prompt="Review {{ item.file }}:\n{{ nodes.scan.output }}"),
        [{"file": "api.py"}, {"file": "auth.py"}, {"file": "db.py"}],
    )
    summary = codex(task_id="summary", prompt=(
        "Merge findings:\n{% for r in fanouts.review.nodes %}{{ r.output }}\n{% endfor %}"
    ))
    scan >> review >> summary

print(g.to_json())

fanout(node, source) dispatches on type:

• int -- N identical copies: fanout(node, 128)
• list -- one per item: fanout(node, [{"repo": "api"}, ...])
• dict -- cartesian product: fanout(node, {"axis1": [...], "axis2": [...]})

Reduce with merge(node, source, size=N) (batch) or merge(node, source, by=["field"]) (group).

Iterative Cycles

Loop until a stop condition with on_failure:

from agentflow import Graph, codex, claude

with Graph("iterative-impl", max_iterations=5) as g:
    write = codex(
        task_id="write",
        prompt="Write a Python email validator.\n{% if nodes.review.output %}Fix: {{ nodes.review.output }}{% endif %}",
        tools="read_write",
    )
    review = claude(
        task_id="review",
        prompt="Review:\n{{ nodes.write.output }}\nIf complete, say LGTM. Otherwise list issues.",
        success_criteria=[{"kind": "output_contains", "value": "LGTM"}],
    )
    write >> review
    review.on_failure >> write  # loop until LGTM or max_iterations

print(g.to_json())

Local & External Models via Pi

Use the pi coding agent as a target alongside codex and claude. Pi routes to Anthropic, OpenAI, Groq, Cerebras, xAI, DeepSeek, Gemini, OpenRouter, Bedrock, etc., and to local endpoints (LMStudio, Ollama) via its OpenAI-compatible or Anthropic-compatible wire protocols.

from agentflow import Graph, codex, pi

with Graph("mixed") as g:
    # External: Claude via Pi
    review = pi(
        task_id="review",
        prompt="Review {{ nodes.impl.output }}",
        model="anthropic/claude-sonnet-4-6:high",
    )

    # Local: LMStudio (add the provider once in ~/.pi/agent/models.json)
    scan = pi(
        task_id="scan",
        prompt="Scan the repo for TODOs.",
        model="lmstudio/qwen/qwen3.6-27b",
        tools="read_only",
    )

For one-off inline provider configs (e.g. a remote LMStudio box), pass a full ProviderConfig via provider={...} and AgentFlow materializes a scoped models.json for the run. See examples/pi_local_lmstudio.py.

Inference via SkyPilot

Launch a vLLM or SGLang OpenAI-compatible endpoint on SkyPilot-supported clouds:

agentflow inference Qwen/Qwen2.5-0.5B-Instruct \
  --gpu aws:1xl4@us-east-1

The command prints a base_url and api_key that can be passed to AgentFlow nodes through a structured provider config. Use --mode batch for explicit JSONL batch jobs.

For graph runs, attach the service directly to the pipeline. AgentFlow launches one shared SkyPilot service before scheduling nodes, then injects the resolved OpenAI-compatible provider into PI nodes that do not already set provider:

from agentflow import Graph, InferenceSetup, pi

with Graph(
    "my-pipeline",
    concurrency=3,
    inference=InferenceSetup(
        gpu="aws:8x8xb200@us-east-2",
        model="Qwen/Qwen2.5-0.5B-Instruct",
        engine="sglang",
    ),
) as g:
    pi(task_id="answer", prompt="Use the shared inference service.")

GPU selectors support single-node and multi-node shapes, including aws:8xb200@us-east-1 and aws:8x8xb200@us-east-2. Spot is enabled by default; use --no-spot to disable it. On AWS B200, AgentFlow resolves the current Blackwell-capable DLAMI from AWS SSM unless --image-id is supplied.

Execution Targets

Docker

Build the bundled agent image once. It contains AgentFlow plus the Codex, Claude, Kimi, and Pi CLIs, and includes both the Docker client and daemon for nested Docker workloads:

docker build -t agentflow-agents:latest .

kind: "docker" uses that image by default:

以上为 README 的开头部分。其余内容请阅读原文。 查看完整 README ↗