跳到主要内容

数据截至 (上游 commit eaf1040642b2)

编排器与回合循环:agentlet 是怎样被一次次驱动的

30 秒导读: 上一章讲了 Task 是"你想要什么类型的结果"的声明。但声明本身不会自己跑。真正把一堆 Task 变成"模型一步步动起来"的,是 Orchestrator(编排器)——Marvin 的执行引擎。它是一台"回合驱动机":每一回合挑一个演员(actor)、凑齐它这一步能用的工具和记忆、组装消息、跑一次底层 LLM agent(Marvin 叫它 agentlet),然后把结果落库,循环往复直到任务全部完成。本章讲透这台机器的主循环和一个回合的六个步骤。

本章是全库的主线。前一章 01-task-and-result-type.md 讲"要什么";本章讲"怎么一步步去要"。至于"结束回合"这件事具体怎么被做成类型化工具、事件流怎么流,是精华,留给下一章 03-end-turn-and-agentlet.md。

1. 这是什么(零基础也能懂)​

一句话定义: Orchestrator 是一个 while 循环,反复驱动 AI 演员"走一步、再走一步",直到它声明的所有任务都有了结果。

它要解决的问题: 一次 LLM 调用往往不够。模型要先看任务、调个工具、看工具返回、再调一个、最后才敢说"我做完了"。这中间的排程(该谁上场、这一步给它哪些工具、上一步说的话记在哪、什么时候该停)必须有人管。Orchestrator 就是这个管家。

一句话直觉: 把它想成一个回合制游戏的裁判。每一回合,裁判点一个玩家(actor)上场,告诉它"这回合你能用这些道具(tools)、这是你之前的记录(消息历史)",让它行动一次,然后把这回合发生的事记进档案(落库)。裁判一直数回合数,直到"所有目标达成"或"回合用尽"才吹哨结束。

用起来什么样: 大多数时候你不直接碰 Orchestrator,你调 marvin.run(...),它内部替你建了一台。但你也可以直接用:

# 示意,非源码
from marvin.engine.orchestrator import Orchestrator

orchestrator = Orchestrator(tasks=[my_task], thread=thread)
results = await orchestrator.run(max_turns=5)   # 最多跑 5 个回合

src/marvin/engine/orchestrator.py:60 的 Orchestrator 只装三样东西:一组 tasks、一个 thread(对话线程,消息都往这里存)、一组 handlers(事件观察者,如打印/流式)。

2. 顶层全景(它大概怎么转)​

Orchestrator 有两层循环:外层 run() 数回合、管生命周期;内层 run_once() 跑一个回合。

怎么读这张图: 从上往下是一次 run();中间那个方框是 while 主循环,每转一圈就是一个 run_once();跳出循环有两个出口(任务全完 / 回合用尽)。

                    Orchestrator.run()
                          │
             设 contextvar(_current_orchestrator=self)
                          │
                  进入 Thread 上下文
                          │
               manage_mcp_servers(actor)  ← 启动/复用 MCP 服务
                          │
   ┌──────────────────────▼──────────────────────────┐
   │  while  还有未完成任务  且  turns < max_turns     │
   │  ────────────────────────────────────────────    │
   │        result = await run_once()   ← 跑一个回合    │
   │        turns += 1                                 │
   │        有任务失败? → 抛错(可选)                  │
   │        重算"未完成任务"集合                        │
   └──────────────────────┬──────────────────────────┘
                          │
          任务全完 → 正常返回 results
          回合用尽 → raise "Max agent turns reached"
                          │
              finally: 复位 contextvar
                        + 若为最外层 Thread → 清理 MCP

部件一句话职责:

部件干什么在哪
run()外层主循环:数回合、判终止、管 contextvar 与 MCP 生命周期orchestrator.py:240
run_once()一个回合的六步(下面详解)orchestrator.py:136
get_all_tasks()拓扑收集任务(含依赖/子任务),可按 ready/incomplete 过滤orchestrator.py:92
start_turn / end_turn回合的开场/收场;end_turn 里执行 EndTurn.runorchestrator.py:229 / :233
_get_messages组装系统提示 + 历史,拆出 user_promptorchestrator.py:339
_check_memories按最近消息自动检索记忆,注入线程orchestrator.py:316
agentlet底层 pydantic-ai 的 Agent,真正调模型的那层由 actor.get_agentlet() 造

一句话主线: run() 起一台裁判,进入 while;每圈 run_once() 挑 actor → 凑工具 → 检索记忆 → 组消息 → 跑 agentlet → 落库 → 结束回合;直到任务全完或回合耗尽。

3. 外层:run() 主循环与终止条件​

这节讲外层循环"什么时候转、什么时候停"。

3.1 终止条件:两个退出口​

主循环的判据在 orchestrator.py:277:

while incomplete_tasks and (max_turns is None or turns < max_turns):
    result = await self.run_once(actor=actor, active_mcp_servers=active_mcp_servers)
    results.append(result)
    turns += 1

两个退出口:

  • 任务全完 → incomplete_tasks 变空,循环自然结束,正常返回 results。
  • 回合用尽 → turns >= max_turns,循环结束后在 :299 主动 raise ValueError("Max agent turns reached")。

max_turns 的解析在 :245:显式传参优先,否则取 marvin.settings.max_agent_turns,再没有就是 math.inf(无上限)。

3.2 关键细节:"未完成"只算"交给我的那些任务"​

incomplete_tasks 在 :251 初始化时,只从 self.tasks(你交给编排器的顶层任务)里挑,不是从 get_all_tasks() 的全集里挑。源码注释(:270-276)特意点明了这个设计:

一个任务如果有未完成的依赖,那些依赖会在编排逻辑里被求值,但不算进终止条件。

直觉: 你说"我要 A 完成";A 依赖 B。裁判会顺带把 B 也调度出来跑,但它只盯着"A 完了没"来决定收工。B 是手段,A 才是目的。

每圈末尾在 :295 重算这个集合:incomplete_tasks = {t for t in self.tasks if t.is_incomplete()}。is_incomplete()(task.py:629)的定义是状态为 PENDING 或 RUNNING。

3.3 失败处理​

若 raise_on_failure=True(默认),每圈跑完后在 :288-293 扫一遍:任一顶层任务 is_failed() 就立即抛错,把失败往上冒,不再空转。

3.4 生命周期:contextvar 与 MCP 服务​

run() 用两层上下文把"当前编排器"和"MCP 服务"管起来。

当前编排器(contextvar): :252 用 _current_orchestrator.set(self) 把自己塞进一个 ContextVar(orchestrator.py:45),finally(:308)再 reset。这样代码任何角落都能用 get_current_orchestrator()(:378)拿到"现在是哪台编排器在跑",不必层层传参。

MCP 服务生命周期: :263 用 async with manage_mcp_servers(actor) as active_mcp_servers 包住整个主循环。这个上下文管理器(_internal/integrations/mcp.py:329)的巧思是跨回合复用:MCP 服务在同一个 Thread 上下文里持久存在,避免每个回合都启停一遍。真正的清理推迟到最外层 Thread 退出——run() 的 finally(:311)判断 get_current_thread() is None(说明没有外层 Thread 了),才调 cleanup_thread_mcp_servers()。

为什么这么设计: 嵌套调用很常见(一个任务里又调 marvin.run)。若每层都启停 MCP,开销大且易乱。用"最外层才清理"的判据,让内层复用外层已经起好的服务。

4. 内层:run_once() 一个回合的六步​

这节是本章核心。run_once()(orchestrator.py:136)把"一个回合"拆成清晰的六步。下面逐步走。

   run_once()
   ─────────
   ① 收集 ready 任务      get_all_tasks("ready")
   ② 选 actor + 派任务    独立任务每回合只派一个
   ③ 汇集 tools           普通 tools + end_turn_tools
   ④ 检索 memories        _check_memories
   ⑤ 组装消息             _get_messages + system.jinja
   ⑥ 跑 agentlet          agentlet.iter → 落库 → end_turn

4.1 步骤①:收集 ready 任务(拓扑收集)​

回合开头(:141)调 get_all_tasks(_filter="ready")。

get_all_tasks(:92)不是简单列出 self.tasks,而是递归展开整张任务图:对每个任务,先收集它的 subtasks(子任务)、再收集 depends_on(依赖),在依赖都入列之后才把自己 append 进 ordered_tasks,最后收集 parent。看 :104-123 的 collect_tasks:

def collect_tasks(task):
    if task in all_tasks:      # 去重,防环
        return
    all_tasks.add(task)
    for subtask in task.subtasks:   # 先子任务
        collect_tasks(subtask)
    for dep in task.depends_on:     # 再依赖
        collect_tasks(dep)
    ordered_tasks.append(task)      # 依赖之后才轮到自己 ← 拓扑序
    if task.parent:
        collect_tasks(task.parent)

这是拓扑收集(topological collect): 保证"被依赖的先出现在列表前面"。这样上层逻辑遍历时,先看到能先跑的任务。

_filter="ready" 再过滤出"现在就能跑"的:is_ready()(task.py:637)的定义是——任务本身未完成,且它所有 depends_on 和 subtasks 都已完成。换句话说,依赖没满足的任务不会进入这一批。

若一个 ready 任务都没有,:143 直接抛 ValueError("No tasks to run")。

4.2 步骤②:选 actor 与 assigned_tasks(独立任务每回合只派一个)​

选 actor: 若外部没传,:147 取 tasks[0].get_actor()——第一个 ready 任务的负责演员。然后 :150 挑出所有"归这个 actor 管"的 ready 任务,记为 potential_tasks。

关键细节——独立任务每回合只派一个: 若 potential_tasks 超过一个,:153-164 会检查它们之间有没有依赖关系:

if len(potential_tasks) > 1:
    has_deps = any(
        t2 in t1.depends_on or t1 in t2.depends_on
        for t1 in potential_tasks
        for t2 in potential_tasks
        if t1 != t2
    )
    # 独立 → 一次只派一个;有依赖 → 一起派
    assigned_tasks = [potential_tasks[0]] if not has_deps else potential_tasks

为什么? 源码注释(:152)说得直白:"For independent tasks, only assign one per turn to avoid EndTurn conflicts"。原因是下一章会讲的 EndTurn 机制——模型这一回合只能产出一个结束动作(如"标记某任务成功")。若你把两个互不相干的任务塞进同一回合,模型没法在一次输出里同时"结束两个任务",会打架。所以独立任务被拆开,一回合专心推进一个;而有依赖关系的任务是相关的,反而适合放一起让模型统筹。

标记运行中: :167 把还是 pending 的 assigned 任务标成 running,mark_running(task.py:582)顺手把任务的 prompt 作为一条 user 消息写进 thread——这就是"任务被摆上桌面"的那句话。随后 :170 调 start_turn(actor)。

4.3 步骤③:汇集 tools 与 end_turn_tools​

:172-180 把这一回合 assigned 任务的工具汇成两个集合:

  • 普通 tools(:173):t.get_tools() 并集——模型这一步能调的函数。
  • end_turn_tools(:178):t.get_end_turn_tools() 并集——"结束回合"的类型化工具(标记成功/失败/跳过/委派等)。

用 set 去重,因为多个任务可能共享同一工具。这两类工具的构造细节(尤其 EndTurn 怎么被做成类型化工具)是下一章 03-end-turn-and-agentlet.md 的主题,本章不展开。

4.4 步骤④:检索 memories​

:183 调 _check_memories(actor, assigned_tasks)。它的逻辑(:316)是:

  1. 收集 assigned 任务里 auto_use=True 的记忆;若 actor 是 Agent,再并上它自己 auto_use 的记忆(:322)。
  2. 若有这类记忆,:325 取最近 3 条消息,序列化成查询串(:329)。
  3. 对每块记忆 search(query, n=3),把结果作为 info 消息注入线程(:334),前缀标"Automatically recalled memories"。

直觉: 上场前,裁判先替演员翻一翻"和眼下话题相关的旧记忆",把它铺到桌面上,模型这一步就能"记起来"。记忆系统本身的持久化细节在 04-actors-threads-scaling.md。

4.5 步骤⑤:组装消息(_get_messages + system.jinja)​

:186 调 _get_messages。它做三件事(:339):

其一,系统提示。 :342 用 SystemPrompt(orchestrator.py:51,一个绑到 system.jinja 模板的 Template)渲染出系统消息,喂进三样东西:当前 actor、全局 get_instructions()、以及 assigned_tasks。模板 src/marvin/templates/system.jinja 很短:

<system>
    Complete tasks to the best of your ability by using the appropriate tool...
    {{ actor.get_prompt() | indent(4) }}
    {% if instructions %}
    <instructions> ... {% for instruction in instructions %} ... {% endfor %} </instructions>
    {% endif %}
</system>

它把"演员是谁(actor.get_prompt())"和"必须遵守的指令"拼进一个 <system> 块。演员的自我介绍由 actor.get_prompt()(actor.py:133)渲染。

其二,历史消息。 :350 从 thread 拉出历史(不含系统消息)。

其三,拆出 user_prompt。 :355-363 有个小技巧:若最后一条消息是"用户提问"(kind == "request" 且首段是 user-prompt),就把它从历史里摘出来,单独作为本回合的 user_prompt;否则(:368)退化成一个空格 " "。注释(:365)解释原因:pydantic-ai 要求必须有 user prompt,而有些模型供应商不允许空串。

返回值是 (user_prompt, [system_prompt] + message_history)。

4.6 步骤⑥:跑 agentlet,落库,结束回合​

造 agentlet 并跑。 :191 让 actor 用汇好的 tools/end_turn_tools/MCP 服务造一个 agentlet(底层 pydantic-ai 的 Agent)。然后 :197-207 在 with actor: 上下文里,用 agentlet.iter(user_prompt, message_history=...) 流式驱动一次模型调用,期间产生的事件交给 handle_agentlet_events → self.handle_event 分发给各 handler。事件流的细节属于下一章末尾。

落库 + usage 记录。 回合跑完,:210 取 run.result.new_messages(),:211 写进 thread——注意 new_messages[1:] 跳过了首条(它要么来自历史、要么是那个占位空串)。紧接着 :217 建一条 DBLLMCall:

await DBLLMCall.create(
    thread_id=self.thread.id,
    usage=run.usage(),                    # token 用量
    prompt_messages=prompt_messages,      # 这回合喂进去的
    completion_messages=completion_messages,  # 模型产出的
)

这是每回合的审计记录: 谁在哪个线程、这一步花了多少 token、喂了什么、答了什么,全部落库。

结束回合。 :225 调 end_turn(result, actor)。它的关键一步(:233)是:

async def end_turn(self, result, actor):
    if isinstance(result.output, EndTurn):
        await result.output.run(thread=self.thread, actor=actor)   # ← 执行结束动作
    await actor.end_turn(result=result, thread=self.thread)
    await self.handle_event(ActorEndTurnEvent(actor=actor))

若模型本回合的输出是一个 EndTurn 工具,就调它的 .run(...)——真正去"标记任务成功/失败/跳过/委派"。这是任务状态如何被改写的关键接缝,也是外层 while 能收敛的原因:回合里模型调了个 EndTurn,任务状态变了,外层重算 incomplete_tasks 才会真的缩小。EndTurn 类型体系的构造,是下一章的正题。

对称地,start_turn(:229)在回合开头调 actor.start_turn(thread) 并发 ActorStartTurnEvent;actor 默认实现(actor.py:105/:113)在 verbose 时往线程写一条"开始/结束回合"的 info 消息。

5. 一个岔路:独立任务的并发分支(run_tasks_async)​

前面讲的都是"进 Orchestrator、一回合一回合来"。但有个更高层的入口 src/marvin/fns/run.py 里藏着一条旁路:多个互相独立的任务可以不进这套回合循环,而是并发跑。

看 run_tasks_async(run.py:34):

if len(tasks) > 1 and _tasks_are_independent(tasks):
    await asyncio.gather(*[task.run_async() for task in tasks])   # 并发
    return tasks
else:
    orchestrator = Orchestrator(tasks=tasks, thread=thread, handlers=handlers)
    await orchestrator.run(raise_on_failure=raise_on_failure)     # 走回合循环
    return tasks

判据 _tasks_are_independent(run.py:18)检查任意两个任务之间是否有依赖、父子、共享子任务的关系;全无,才算独立。

两条路的分工:

情况走哪条路为什么
多个独立任务asyncio.gather,各自 task.run_async()互不相干,并发跑最快,每个任务其实各起一台自己的 Orchestrator
有依赖 / 只有一个任务一台 Orchestrator + run()需要统一排程、按依赖顺序推进

注意区分两个"独立判断": 这里(run.py)的判断决定"要不要开并发、各跑各的编排器";而 §4.2 里 run_once 那个判断,是在同一台编排器内、同一个 actor 名下,决定"这一回合派几个任务"。前者是宏观分流,后者是回合内防 EndTurn 打架。二者都源于"独立任务不该在一次结束动作里被搅在一起"这个约束,只是作用在不同层。

6. 巧妙之处(可带走的精华)​

  • 拓扑收集,依赖后置。 get_all_tasks 用一个带去重的递归,把"依赖先于自己入列"变成天然拓扑序,还顺手防了环(orchestrator.py:104-123)。调度不必再单独排序。
  • 终止只盯顶层任务。 依赖被调度但不进终止条件(:270-276)——"手段可以很多,目的只有几个",循环收敛判据干净。
  • 独立任务每回合只派一个。 用"防 EndTurn 冲突"这一条约束,反推出"独立就拆、相关就合"的派任务策略(:152-164)。一个很实在的工程折衷。
  • MCP 服务按最外层 Thread 清理。 跨回合、跨嵌套调用复用服务,只在真正最外层退出时才清理(:311),省掉反复启停。
  • contextvar 传递"当前编排器"。 免去层层传参,任何深处都能 get_current_orchestrator()(:378)。
  • 每回合一条 DBLLMCall。 token 用量与前后消息全落库(:217),天然的可观测与审计。

7. 边界与局限(诚实)​

  • 多演员尚未完备。 run() 里 :257 有一条 # TODO: Handle multi-actor scenarios properly,当前直接取 self.tasks[0].get_actor() 作为整场的 actor。真正的多智能体协作(委派、切换)见 04-actors-threads-scaling.md。
  • 回合内一次结束动作。 一个回合模型只产出一个 EndTurn,所以独立任务必须拆到不同回合——这是设计约束,不是 bug,但意味着"一回合推进多个独立目标"做不到(要么走 §5 的并发旁路)。
  • 无 ready 任务即报错。 run_once 在没有 ready 任务时直接 raise(:143),而非静默等待;调度前提是"至少有一个能跑的"。

8. 代码地图(导航索引)​

主题文件路径符号名
编排器类与构造src/marvin/engine/orchestrator.pyOrchestrator, Orchestrator.__init__
外层主循环 / 终止条件src/marvin/engine/orchestrator.pyOrchestrator.run
一个回合的六步src/marvin/engine/orchestrator.pyOrchestrator.run_once
拓扑收集任务src/marvin/engine/orchestrator.pyOrchestrator.get_all_tasks, collect_tasks
回合开场/收场 + 执行 EndTurnsrc/marvin/engine/orchestrator.pyOrchestrator.start_turn, Orchestrator.end_turn
组装消息 / 拆 user_promptsrc/marvin/engine/orchestrator.pyOrchestrator._get_messages
自动检索记忆src/marvin/engine/orchestrator.pyOrchestrator._check_memories
系统提示模板对象src/marvin/engine/orchestrator.pySystemPrompt
当前编排器 contextvarsrc/marvin/engine/orchestrator.py_current_orchestrator, get_current_orchestrator
系统提示模板src/marvin/templates/system.jinja(jinja 模板)
独立任务并发分支src/marvin/fns/run.pyrun_tasks_async, _tasks_are_independent
MCP 服务生命周期src/marvin/_internal/integrations/mcp.pymanage_mcp_servers, cleanup_thread_mcp_servers
任务就绪/未完成判定src/marvin/tasks/task.pyTask.is_ready, Task.is_incomplete, Task.mark_running
演员的 agentlet / 回合钩子src/marvin/agents/actor.pyActor.get_agentlet, Actor.start_turn, Actor.end_turn