Back to Details

orchestrate

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Orchestrate

编排

Write a Python program that drives agent execution. The program is your body — agents are your hands. You write the loop, they do the work.
A single program can coordinate 100+ agents — fan out work with 
asyncio.gather
, manage worker pools with queues, and stay in control through structured decisions.
编写用于驱动Agent执行的Python程序。程序就是你的躯干——Agent就是你的双手：你编写循环逻辑，它们负责完成具体工作。
单个程序可以协调100个以上的Agent——通过
asyncio.gather
分发任务，用队列管理工作池，通过结构化决策始终掌握控制权。

Launch

启动步骤

  1. Write a Python program
  2. Run: 
    python <file.py>
CRITICAL: After launching with 
python
, STOP. Do not monitor, tail logs, or wait. Your turn is done. The program handles the rest.
  1. 编写Python程序
  2. 运行命令：
    python <file.py>
重要提示：使用
python
启动后就停止操作，不要监控、追踪日志或是等待。你的环节已经结束，剩下的流程会由程序自动处理。

Agent API

Agent API

python
from orchestrate import Agent
python
from orchestrate import Agent

Create agents — loads config from ~/.claude/agents/ if available

Create agents — loads config from ~/.claude/agents/ if available

researcher = Agent("research") # loads ~/.claude/agents/research.md dev = Agent("frontend_dev", prompt="React expert") # inline config reviewer = Agent("reviewer") # loads ~/.claude/agents/reviewer.md
researcher = Agent("research") # loads ~/.claude/agents/research.md dev = Agent("frontend_dev", prompt="React expert") # inline config reviewer = Agent("reviewer") # loads ~/.claude/agents/reviewer.md

Run a task

Run a task

result = await researcher.arun("analyze the codebase")
result = await researcher.arun("analyze the codebase")

With explicit context

With explicit context

impl = await dev.arun("implement feature", context=[result])
impl = await dev.arun("implement feature", context=[result])

With schema for structured output

With schema for structured output

review = await reviewer.arun("review this", context=[impl], schema={ "status": "str", "issues": "list" }) print(review["status"]) # "APPROVED"
review = await reviewer.arun("review this", context=[impl], schema={ "status": "str", "issues": "list" }) print(review["status"]) # "APPROVED"

Spawn a child agent — inherits parent config

Spawn a child agent — inherits parent config

helper = dev.spawn("css_helper", prompt="Handle CSS only") await helper.arun("fix spacing", context=[impl])
helper = dev.spawn("css_helper", prompt="Handle CSS only") await helper.arun("fix spacing", context=[impl])

Close when done

Close when done

await researcher.aclose()
undefined
await researcher.aclose()
undefined

Agent creation

Agent创建方式

python
undefined
python
undefined

From ~/.claude/agents/ definition (auto-loads prompt, tools, model)

From ~/.claude/agents/ definition (auto-loads prompt, tools, model)

researcher = Agent("research")
researcher = Agent("research")

Inline definition

Inline definition

dev = Agent("my_dev", prompt="You are a developer.", model="sonnet")
dev = Agent("my_dev", prompt="You are a developer.", model="sonnet")

Override a .claude/agents/ definition

Override a .claude/agents/ definition

fast = Agent("research", model="haiku") # keeps prompt/tools, overrides model
fast = Agent("research", model="haiku") # keeps prompt/tools, overrides model

Explicit server URL (default: ORCHESTRATE_API_URL env or localhost:7777)

Explicit server URL (default: ORCHESTRATE_API_URL env or localhost:7777)

agent = Agent("research", api_url="http://localhost:7777")
undefined
agent = Agent("research", api_url="http://localhost:7777")
undefined

Concurrency — just asyncio

并发处理——直接使用asyncio

python
undefined
python
undefined

Parallel fan-out

Parallel fan-out

r1, r2 = await asyncio.gather( Agent("research").arun("check GitHub"), Agent("research").arun("check papers"), )
r1, r2 = await asyncio.gather( Agent("research").arun("check GitHub"), Agent("research").arun("check papers"), )

Sequential — agent accumulates session context

Sequential — agent accumulates session context

await dev.arun("set up project") await dev.arun("add auth") # dev remembers the setup await dev.arun("write tests") # dev remembers everything
undefined
await dev.arun("set up project") await dev.arun("add auth") # dev remembers the setup await dev.arun("write tests") # dev remembers everything
undefined

Legacy API (still works)

旧版API（仍可正常使用）

The 
Orchestrate
class still works for backward compat:
python
from orchestrate import Orchestrate

async def main(orch):
    await orch.agent("coder", prompt="...")
    result = await orch.run("do X", to="coder")
为了向后兼容，
Orchestrate
类仍然可以正常使用：
python
from orchestrate import Orchestrate

async def main(orch):
    await orch.agent("coder", prompt="...")
    result = await orch.run("do X", to="coder")

Writing effective programs

编写高效程序的技巧

Stay in the loop

保持在循环中掌控进度

You are the brain. Agents are the hands. If you don't check their work, you're blind.
python
undefined
你是大脑，Agent是双手。如果你不检查它们的输出，就等于闭着眼工作。
python
undefined

Bad — fire and forget

Bad — fire and forget

for task in tasks: await dev.arun(task)
for task in tasks: await dev.arun(task)

Good — check results, steer

Good — check results, steer

for task in tasks: result = await dev.arun(task) review = await reviewer.arun(f"Review: {result.text}", schema={"approved": "bool", "feedback": "str"}) if not review["approved"]: await dev.arun(f"Fix: {review['feedback']}", context=[review])
undefined
for task in tasks: result = await dev.arun(task) review = await reviewer.arun(f"Review: {result.text}", schema={"approved": "bool", "feedback": "str"}) if not review["approved"]: await dev.arun(f"Fix: {review['feedback']}", context=[review])
undefined

Use schema for program logic

配合Schema实现程序逻辑

python
undefined
python
undefined

Structured — reliable

Structured — reliable

result = await agent.arun("check build status", schema={ "passing": "bool", "failures": "int", "summary": "str", }) if result["passing"]: print("Ship it")
undefined
result = await agent.arun("check build status", schema={ "passing": "bool", "failures": "int", "summary": "str", }) if result["passing"]: print("Ship it")
undefined

Use context to chain agent work

使用上下文串联Agent工作流

Every 
arun()
returns a 
ContextResult
— pass results between agents with 
context=
:
python
research = await researcher.arun("research X")
impl = await dev.arun("implement based on research", context=[research])
review = await reviewer.arun("review", context=[research, impl])
ContextResult
behaves like a dict for schema fields:
python
r = await agent.arun("analyze", schema={"score": "int", "reason": "str"})
print(r["score"])   # 87
print(r.text)       # full response
print(r.summary)    # one-line summary
print(r.file)       # ~/.orchestrate/context/42.md
每个
arun()
都会返回一个
ContextResult
对象——通过
context=
参数在不同Agent之间传递结果：
python
research = await researcher.arun("research X")
impl = await dev.arun("implement based on research", context=[research])
review = await reviewer.arun("review", context=[research, impl])
ContextResult
在处理Schema字段时的行为和字典一致：
python
r = await agent.arun("analyze", schema={"score": "int", "reason": "str"})
print(r["score"])   # 87
print(r.text)       # full response
print(r.summary)    # one-line summary
print(r.file)       # ~/.orchestrate/context/42.md

CRITICAL: Always validate context before passing

重要提示：传递上下文前必须先校验

Never blindly pass 
recall()
results. Inspect first:
python
past = await orch.recall(agent="researcher", limit=10)
for c in past:
    print(f"  [{c.id}] {c.agent}: {c.summary[:80]}")
relevant = [c for c in past if "architecture" in c.summary.lower()]
await dev.arun("build on this", context=relevant)
永远不要盲目传递
recall()
的结果，先检查内容是否相关：
python
past = await orch.recall(agent="researcher", limit=10)
for c in past:
    print(f"  [{c.id}] {c.agent}: {c.summary[:80]}")
relevant = [c for c in past if "architecture" in c.summary.lower()]
await dev.arun("build on this", context=relevant)

Implement → Review → Fix loop

实现→评审→修复循环

Every implementation should be reviewed by a separate agent (not self-review):
python
researcher = Agent("research")
dev = Agent("implementer")
reviewer = Agent("reviewer")
每个实现产出都应该由独立的Agent评审（不要自评审）：
python
researcher = Agent("research")
dev = Agent("implementer")
reviewer = Agent("reviewer")

Research

Research

research = await researcher.arun("analyze codebase")
research = await researcher.arun("analyze codebase")

Implement

Implement

impl = await dev.arun("implement feature", context=[research])
impl = await dev.arun("implement feature", context=[research])

Review → Fix loop (max 2 rounds)

Review → Fix loop (max 2 rounds)

for attempt in range(3): review = await reviewer.arun("review this", context=[impl], schema={"status": "str", "issues": "list"}) if review["status"] == "APPROVED": break if attempt < 2: issues = "\n".join(f"- {i}" for i in review["issues"]) impl = await dev.arun(f"Fix:\n{issues}", context=[review])
undefined
for attempt in range(3): review = await reviewer.arun("review this", context=[impl], schema={"status": "str", "issues": "list"}) if review["status"] == "APPROVED": break if attempt < 2: issues = "\n".join(f"- {i}" for i in review["issues"]) impl = await dev.arun(f"Fix:\n{issues}", context=[review])
undefined

Full example

完整示例

python
import asyncio
from orchestrate import Agent

async def main():
    researcher = Agent("research")
    dev = Agent("implementer")
    reviewer = Agent("reviewer")

    # Phase 1: Parallel research
    r1, r2 = await asyncio.gather(
        researcher.arun("research approach A"),
        researcher.arun("research approach B"),
    )

    # Phase 2: Implement
    impl = await dev.arun("implement the best approach", context=[r1, r2])

    # Phase 3: Review loop
    for attempt in range(3):
        review = await reviewer.arun("review", context=[impl], schema={"status": "str", "issues": "list"})
        if review["status"] == "APPROVED":
            break
        impl = await dev.arun(f"fix: {review['issues']}", context=[review])

    print(f"Done! Final review: {review['status']}")

    await researcher.aclose()
    await dev.aclose()
    await reviewer.aclose()

if __name__ == "__main__":
    asyncio.run(main())
python
import asyncio
from orchestrate import Agent

async def main():
    researcher = Agent("research")
    dev = Agent("implementer")
    reviewer = Agent("reviewer")

    # Phase 1: Parallel research
    r1, r2 = await asyncio.gather(
        researcher.arun("research approach A"),
        researcher.arun("research approach B"),
    )

    # Phase 2: Implement
    impl = await dev.arun("implement the best approach", context=[r1, r2])

    # Phase 3: Review loop
    for attempt in range(3):
        review = await reviewer.arun("review", context=[impl], schema={"status": "str", "issues": "list"})
        if review["status"] == "APPROVED":
            break
        impl = await dev.arun(f"fix: {review['issues']}", context=[review])

    print(f"Done! Final review: {review['status']}")

    await researcher.aclose()
    await dev.aclose()
    await reviewer.aclose()

if __name__ == "__main__":
    asyncio.run(main())