精华:把"结束回合"做成类型化工具 + agentlet 事件流

30 秒导读: Marvin 最巧妙的一招——它不靠"解析模型输出"来拿到任务结果,而是把"完成任务"本身做成一个工具。这个工具的参数类型,就是第 01 章里 Task 声明的那个类型。模型想结束这一回合,唯一的办法就是调用这个工具、并交出一个合法的结果对象——类型不对,pydantic 直接打回让它重试。本章拆开这套机制:工厂函数如何为每个 Task 动态造一个专属"完成工具",Agent 如何把它塞进 pydantic-ai 当 output_type,以及底层事件流如何被翻译成 Marvin 自己的事件。

本章是第 02 章"编排器与回合循环"里"跑一次 agentlet"那一步的放大镜。第 02 章讲"agentlet 是怎样被一次次驱动的";这一章只讲agentlet 内部长什么样、结束回合这件事怎么被类型化。回合调度的循环不再重复。

---

1. 先建立直觉:为什么"结束回合"要做成工具

1.1 一个反直觉的设计

大多数框架拿 LLM 的结构化输出,是这样的:让模型吐一段 JSON → 自己去 parse → 校验 → 塞进对象。出错了(少个字段、类型不对)得自己写重试逻辑。

Marvin 反过来。它的核心洞察是:

"我做完了,结果是 X" 和 "我要调用工具 search(query)",对模型来说是同一种动作——都是一次工具调用。既然如此,何不把"完成任务"也做成一个工具?

于是 Marvin 造了一族叫 EndTurn(结束回合) 的工具。它们不是给模型用来"干活"的普通工具,而是给模型用来**声明"这一回合该怎么收尾"**的:

EndTurn 工具	模型用它来说
MarkTaskSuccessful	"这个任务我做完了,结果是这个"
MarkTaskFailed	"这个任务我做不了,原因是……"
MarkTaskSkipped	"这个任务跳过"
PostMessage	"我先往对话里说句话(不收尾任务)"
DelegateToActor	"这活我转交给另一个 actor"
PlanSubtasks	"这任务太大,我先拆成几个子任务"

1.2 妙在哪:类型安全"免费"了

关键的巧妙之处在 MarkTaskSuccessful 上:它的 result 参数类型,不是写死的,而是等于那个具体 Task 声明的结果类型。

举例:你有个任务 result_type=list[int]。Marvin 会为它现造一个工具,这个工具的 result 字段类型就是 list[int]。模型想收工,就得调用这个工具并传一个 list[int]。传错了——比如传了字符串——pydantic-ai 会在校验阶段直接生成一个"重试提示",退回给模型让它改。

你没写一行 parse 代码,没写一行重试逻辑,类型安全就有了。 这就是本章要深挖的那个"全库最巧妙处"。

              ┌─────────────────────────────────────────────┐
   Task       │  result_type = list[int]                     │
  (声明要什么) └───────────────────┬─────────────────────────┘
                                  │ 工厂函数读取 get_result_type()
                                  ▼
              ┌─────────────────────────────────────────────┐
  EndTurn 工具│  MarkTaskSuccessful_<id>                     │
 (动态生成)   │    result: list[int]   ← 字段类型 = 上面那个 │
              └───────────────────┬─────────────────────────┘
                                  │ 塞进 pydantic-ai 当 output_type
                                  ▼
              ┌─────────────────────────────────────────────┐
   模型       │ 想收工 → 必须调用它 → 必须交出合法 list[int] │
              │ 类型不对 → pydantic 打回 → 重试              │
              └─────────────────────────────────────────────┘

---

2. 顶层全景:一次回合里这三个文件怎么协作

结束回合这件事,横跨三个文件。先看它们各自的职责:

部件	干什么	在哪个文件
EndTurn 家族 + 工厂函数	定义"结束回合"有哪几种,并为每个 Task 动态造出专属工具类	engine/end_turn.py
Task.get_end_turn_tools	按任务配置(允许失败?允许跳过?要规划?)组装出这个任务能用的 EndTurn 工具	tasks/task.py:517
Agent.get_agentlet	把 EndTurn 工具塞进 pydantic-ai 当 output_type,把普通工具包一层错误处理,造出一个一次性 agentlet	agents/agent.py:139
handle_agentlet_events	边跑 agentlet,边把 pydantic-ai 的底层事件翻译成 Marvin 自己的 Event	engine/streaming.py:44

主线走一遍(高层,不进代码):

编排器(第 02 章)
   │  ① 从每个 assigned task 收集 EndTurn 工具
   │      Task.get_end_turn_tools()  →  [MarkTaskSuccessful_<id>, ...]
   ▼
Agent.get_agentlet(tools, end_turn_tools)
   │  ② 普通工具 wrap_tool_errors 包一层;EndTurn 工具塞进 ToolOutput 当 output_type
   │  ③ 现场 new 一个 pydantic_ai.Agent(一次性,用完即弃)
   ▼
agentlet.iter(user_prompt)  →  run  (第 02 章的循环入口)
   │
   ▼
handle_agentlet_events(agentlet, actor, run)
   │  ④ 遍历 pydantic-ai 的节点,把底层事件翻译成 Marvin Event 往外 yield
   ▼
run.result.output  →  一个 EndTurn 实例
   │  ⑤ 编排器调用 result.output.run(...) 真正落地(mark_successful / delegate / ...)
   ▼
任务状态更新

注意 ③:agentlet 是每回合新建一次的一次性对象,不是长生命周期的。名字里的 "-let"(小字缀,表"小的、临时的")就是这个意思。原因很直接——每个回合能用的 EndTurn 工具、结果类型都可能不同,所以每回合都得按当前这批任务现造一个。

---

3. 核心机制一:工厂函数 + 闭包,为每个 Task 现造一个"完成工具"

3.1 它要解决的小问题

MarkTaskSuccessful 的 result 字段类型,必须等于"当前这个具体 Task"的结果类型。但类型在写代码时是不知道的——它是运行时才从 Task 对象上读出来的。怎么在运行时,动态造一个"某个字段是运行时才定的类型"的 dataclass?

答案:工厂函数 + 闭包。函数内部定义一个 dataclass,用参数 mark_task 把 Task 闭包进去,字段类型直接写成 mark_task.get_result_type()。

3.2 真实实现

看 create_mark_task_successful,end_turn.py:36-57:

def create_mark_task_successful(mark_task: "Task[Any]") -> type[MarkTaskSuccessful]:
    @dataclass(kw_only=True)
    class _MarkTaskSuccessful(MarkTaskSuccessful):
        """Mark a task successful and provide a result."""

        result: mark_task.get_result_type()  # type: ignore   ← 字段类型 = 运行时读出的类型
        task: ClassVar["Task[Any]"] = field(default=mark_task, init=False)
        name: ClassVar[str] = field(
            default=f"Mark {mark_task.friendly_name()} successful", init=False
        )

        def __post_init__(self):
            mark_task.validate_result(self.result)   # ← 构造完立刻再校验一次

        async def run(self, thread: Thread, actor: "Actor") -> None:
            await mark_task.mark_successful(self.result, thread=thread)

    _MarkTaskSuccessful.__name__ = f"MarkTaskSuccessful_{mark_task.id}"
    return _MarkTaskSuccessful

四个精华点,逐个看:

① 字段类型是运行时求值的。 result: mark_task.get_result_type() 这行,在 Python 里 dataclass 的注解是可以是任意表达式的——这里它是一次函数调用,返回值(比如 list[int])就成了字段的类型。get_result_type() 定义在 task.py:411,对分类任务还会把类型换成整数索引类型(见第 01 章)。

② 每个类的名字带上 task.id。 _MarkTaskSuccessful.__name__ = f"MarkTaskSuccessful_{mark_task.id}"。为什么要改名?因为这个名字最终会变成模型看到的工具名。如果同一回合里有多个任务,就会有多个 MarkTaskSuccessful_*,靠 id 后缀区分——模型才知道"我在给哪个任务交结果"。

③ __post_init__ 里再校验一次。 mark_task.validate_result(self.result)(task.py:435)。pydantic-ai 在按类型解析参数时已经校验过一遍了,这里是第二道保险——尤其能触发 Task 上自定义的 result_validator(比如"必须是正数")。校验失败会抛异常,pydantic-ai 捕获后转成对模型的重试提示。

④ run 方法把"落地"封进对象里。 工具被造出来的时候什么都不做;等编排器最后拿到这个 EndTurn 实例,调用它的 run,才真正去 mark_task.mark_successful(...) 改任务状态。"声明"和"执行"被拆开了——这让编排器可以在中间插入事件、审计、持久化。

3.3 同一个模式的其它变体

这个"工厂 + 闭包 + ClassVar 绑定"的模式,在 end_turn.py 里被复用了好几次。差别只在闭包进去的是什么、run 干什么:

工厂函数	闭包进去	额外字段	run 干什么	位置
create_mark_task_successful	mark_task	result(类型化)	mark_task.mark_successful	end_turn.py:36
create_mark_task_failed	mark_task	message: str | None	mark_task.mark_failed	end_turn.py:64
create_mark_task_skipped	mark_task	无	mark_task.mark_skipped	end_turn.py:94
create_delegate_to_actor	delegate_actor, team	message: str | None	切换 team.active_member,发消息	end_turn.py:131
create_plan_subtasks	parent_task	instructions: str	调 marvin.plan_async 生成子任务	end_turn.py:162

不走工厂的只有 PostMessage(end_turn.py:114)——它不绑定任何具体 Task,是个普通的静态 dataclass,run 就往 thread 里加一条消息。

PlanSubtasks 值得单独看一眼(end_turn.py:193-205):它的 run 会开一个新的 Thread,在里面调 marvin.plan_async 让模型生成子任务——"注释里写着 create tasks in a new thread to avoid interrupting the function call",即避免把子任务规划的中间消息混进当前这次工具调用里。

3.4 这些工厂是谁在调

回到 Task 一侧。Task.get_end_turn_tools(task.py:517-530)是组装入口——它按任务的配置决定放哪几个工具进去:

def get_end_turn_tools(self) -> list[type["EndTurn"]]:
    tools: list[...] = []
    tools.append(self.mark_successful_tool())            # 永远有
    if self.allow_fail:
        tools.append(self.mark_failed_tool())            # 允许失败才给
    if self.allow_skip:
        tools.append(self.mark_skipped_tool())           # 允许跳过才给
    if self.plan:
        tools.append(create_plan_subtasks(parent_task=self))  # 要规划才给
    return tools

mark_successful_tool 等(task.py:532-545)只是薄薄一层,转手去调上面那些工厂函数。注意:每次调用 get_end_turn_tools 都会现造一批新的工具类——这也是为什么 agentlet 必须每回合重建。

---

4. 核心机制二:get_agentlet —— 把 EndTurn 塞进 pydantic-ai 的 output_type

4.1 它要解决的小问题

pydantic-ai 的 Agent 有个概念叫 output_type:你告诉它"这次运行的最终产物应该是什么类型",它就会在内部生成一个特殊工具,并且当模型调用这个工具时,把它当作"运行结束、产出最终结果"。

Marvin 要做的事,正好能对上:把 EndTurn 工具当作 pydantic-ai 的 output_type。这样,模型调用 EndTurn 工具 = pydantic-ai 认为"这次 run 结束了,产出就是这个 EndTurn 实例"。

4.2 分开两堆:普通工具 vs EndTurn 工具

get_agentlet(agent.py:139)拿到一大堆混在一起的东西(来自任务的工具、agent 自己的工具、EndTurn 工具),第一步是分堆(agent.py:148-169):

all_potential_items = (
    list(tools) + self.get_tools() + list(end_turn_tools) + self.get_end_turn_tools()
)
...
for item in all_potential_items:
    ...
    if issubclass_safe(item, EndTurn):     # 是 EndTurn 的子类(类)
        final_end_turn_defs.append(item)
    elif isinstance(item, EndTurn):        # 是 EndTurn 的实例(如 PostMessage())
        final_end_turn_defs.append(item)
    elif callable(item):                   # 普通可调用 → 当工具
        marvin_tool_callables.append(item)

用 processed_ids(一个 id() 集合)去重,避免同一个工具重复注册。

普通工具接着被 wrap_tool_errors 包一层(agent.py:171):

unique_marvin_tools = [wrap_tool_errors(tool) for tool in marvin_tool_callables]

wrap_tool_errors 的作用是:普通工具里抛的异常不要直接把整个 run 炸掉,而是转成一个能返回给模型的错误信息,让模型有机会重试或换个做法。

4.3 关键取舍:单个 EndTurn 直接用,多个退化成 None

这是本章最值得玩味的一段(agent.py:177-196)。EndTurn 工具进 output_type 时,pydantic-ai 一次只接受一个输出类型。可是一个回合里可能有好几个 EndTurn 工具(成功/失败/跳过全都在)。怎么办?

if len(final_end_turn_defs) == 1:
    output_type_for_tool_output = final_end_turn_defs[0]   # 单个:直接用它
    tool_output_name = getattr(output_type_for_tool_output, "__name__", tool_output_name)
else:
    # Use None if zero or multiple EndTurn tools are present
    # This avoids schema issues but might prevent multi-turn scenarios?
    # TODO: Revisit handling of multiple EndTurn tools / Union[EndTurn]
    output_type_for_tool_output = type(None)               # 多个:退化成 None
    if len(final_end_turn_defs) > 1:
        logger.warning("Multiple EndTurn tools detected, output validation might be limited.")

final_tool_output = ToolOutput(
    type_=output_type_for_tool_output,
    name=tool_output_name,
    description=tool_output_description,
)

读懂这个取舍:

• 恰好一个 EndTurn 工具 → 直接把它当 output_type,类型安全全程生效,工具名就用它的 __name__(即那个带 id 后缀的名字)。
• 零个或多个 → 退化,output_type 设成 type(None)。代码注释自己承认这是权衡:"This avoids schema issues but might prevent multi-turn scenarios?",并挂了个 TODO: Revisit handling of multiple EndTurn tools / Union[EndTurn]。

这里有个耐人寻味的错位。 类里另有一个 _determine_result_type(agent.py:129-137),它写的是"多个就返回 Union[...]":

def _determine_result_type(self, end_turn_tools: list[Any]) -> type:
    if len(end_turn_tools) == 1:
        return end_turn_tools[0]
    else:
        from typing import Union
        return Union[tuple(end_turn_tools)]   # 理想:多个用 Union

但 get_agentlet 并没有调用 _determine_result_type——它自己内联了一套逻辑,而且选择了更保守的"退化成 None"而非 Union。看起来 _determine_result_type 是"本该这样做"的理想版,get_agentlet 里的内联版是"当前先这样"的务实版,两者尚未收敛(这正是那个 TODO 指向的地方)(inferred)。

给读者的提醒: 别被"退化成 None"吓到。实践中,pydantic-ai 仍然会把每个 EndTurn 工具作为可调用的工具注册进去(而不只是通过 output_type),所以模型照样能调用它们;区别只在于"输出类型层面的强校验"在多工具时被弱化了。真正的落地在编排器的 end_turn(orchestrator.py:233)里通过 isinstance(result.output, EndTurn) 兜底完成。

4.4 造出一次性 agentlet,挂上两个私有属性

最后一步,现场 new 一个 pydantic-ai Agent,并挂两个私有属性供后面事件流用(agent.py:198-217):

agentlet = pydantic_ai.Agent[Any, Any](**agent_kwargs)

# for internal use
agentlet._marvin_tools = combined_tools            # 普通工具(已包错误处理)
agentlet._marvin_end_turn_tools = final_end_turn_defs   # EndTurn 定义(原样)

这两个 _marvin_* 属性是 Marvin 偷偷挂在 pydantic-ai 对象上的书签,下一节的事件翻译要靠它们把"工具名"映射回"是哪个工具 / 是不是 end-turn 工具"。

---

5. 核心机制三:handle_agentlet_events —— 把底层事件翻译成 Marvin Event

5.1 它要解决的小问题

编排器想在模型跑的过程中实时知道:模型在打字了、模型要调工具了、工具返回了、模型要收尾了……但 pydantic-ai 吐的是它自己的一套底层事件。Marvin 需要一个翻译层,把底层事件转成自己的一套语义清晰的 Event(供 handler 消费:打印、持久化、UI 更新)。

handle_agentlet_events(streaming.py:44)就是这个翻译层,它是个异步生成器,边遍历边 yield Marvin Event。

5.2 先建两张映射表

进入循环前,它先用上一节挂的两个私有属性建两张表(streaming.py:77-90):

tools_map = {}          # 工具名 → 工具对象  (来自 _marvin_tools)
end_turn_tools_map = {} # 工具名 → EndTurn 定义 (来自 _marvin_end_turn_tools)
for t in agentlet._marvin_end_turn_tools:
    end_turn_tools_map[t.__name__] = t

有了这两张表,后面每看到一个"工具名",就能查出它是普通工具还是 end-turn 工具——这是区分两类工具调用的关键。

5.3 遍历 pydantic-ai 的节点

pydantic-ai 的一次 run 由若干**节点(node)**串成。handle_agentlet_events 用 async for node in run 遍历,并按节点类型分派(streaming.py:92-182):

async for node in run:
   ├─ is_user_prompt_node   → yield UserMessageEvent(用户输入)
   │
   ├─ is_model_request_node → ① 扫 RetryPromptPart:若是 end-turn 工具被打回
   │                             → yield ToolRetryEvent
   │                          ② 流式读模型 token
   │                             → 逐事件交给 _process_pydantic_event 翻译
   │
   ├─ is_call_tools_node     → 模型要调工具了,流式处理
   │                             → 逐事件交给 _process_pydantic_event 翻译
   │
   └─ is_end_node            → 到达终点(某 end-turn 工具被认定为最终输出)
                                → 从 parts_manager 里找回对应的 ToolCallPart
                                → yield EndTurnToolResultEvent(带上 node.data 结果)

其中 is_model_request_node 分支里有个精细处理(streaming.py:104-109):end-turn 工具的重试不会走普通 FunctionToolResultEvent 通道,而是表现为一个 RetryPromptPart。所以这里专门扫一遍 node.request.parts,凡是名字命中 end_turn_tools_map 的 RetryPromptPart,就补发一个 ToolRetryEvent——这正是"类型不对就退回重试"在事件流里的可见形态。

5.4 逐个事件翻译:_process_pydantic_event

真正的一对一翻译在 _process_pydantic_event(streaming.py:186)。它是一长串 isinstance 分派。核心对照表:

pydantic-ai 底层事件	翻成 Marvin Event	含义
PartStartEvent / PartDeltaEvent(text)	ActorMessageDeltaEvent	模型正在流式打字
PartStartEvent / PartDeltaEvent(tool-call)	ToolCallDeltaEvent	工具调用参数正在流式拼出来
FunctionToolCallEvent	ToolCallEvent	一次(普通)工具调用拼完了,准备执行
FunctionToolResultEvent(ToolReturnPart)	ToolResultEvent	工具正常返回
FunctionToolResultEvent(RetryPromptPart)	ToolRetryEvent	工具让模型重试
FinalResultEvent	EndTurnToolCallEvent	模型选定了 end-turn 工具(见下)

边翻译边维护快照。 翻译时它同时喂一个 ModelResponsePartsManager(parts_manager),把流式 delta 累积成完整的 part。ActorMessageDeltaEvent / ToolCallDeltaEvent 都带一个 snapshot=,让消费者既能拿到"这次新增的一小段",也能拿到"到目前为止的完整内容"。有个注释解释了它为什么直接读私有的 parts_manager._parts 而非 get_parts():"get_parts() filters out ToolCallPartDelta objects which causes index mismatch"(streaming.py:195-197)——即为了让 event.index 对得上,必须用未过滤的内部列表。

5.5 FinalResultEvent:整套机制的高潮

最值得看的翻译是 FinalResultEvent(streaming.py:305-321):

elif isinstance(event, FinalResultEvent):
    tool_call_part = None
    for part in parts_manager.get_parts():
        if isinstance(part, ToolCallPart) and part.tool_name == event.tool_name:
            tool_call_part = part
            break
    ...
    return EndTurnToolCallEvent(
        actor=actor,
        event=event,
        tool_call_id=tool_call_part.tool_call_id,
        tool=end_turn_tools_map.get(event.tool_name),   # ← 查表:是哪个 end-turn 工具
    )

代码上方的注释点破了触发时机(streaming.py:302-304):

"This fires as soon as Pydantic AI recognizes that the tool call is an end turn tool (i.e. as soon as the name is recognized, but before the args are returned)"

翻译:pydantic-ai 一旦认出"模型这次调的是那个 output_type 工具",就立刻发 FinalResultEvent——早在参数还没流完之前。这就是"结束回合"这件事在事件层面的确认信号。Marvin 借它 emit EndTurnToolCallEvent,并用 end_turn_tools_map 把工具名映射回具体的 EndTurn 定义。

至此闭环:

模型决定收工
   │
   ▼ 调用 MarkTaskSuccessful_<id>(result=...)
pydantic-ai 认出这是 output_type 工具
   │
   ▼ 发 FinalResultEvent(仅凭名字,参数还没齐)
_process_pydantic_event 翻成 EndTurnToolCallEvent
   │
   ▼ (参数流完 → 构造 EndTurn 实例 → __post_init__ 再校验)
到达 is_end_node → yield EndTurnToolResultEvent(带最终结果)
   │
   ▼ 编排器:run.result.output 是个 EndTurn 实例
orchestrator.end_turn: isinstance(output, EndTurn) → output.run(...)
   │
   ▼ mark_task.mark_successful(result) → 任务状态变 SUCCESSFUL

最后那两步在编排器里(orchestrator.py:233-235):

async def end_turn(self, result: AgentRunResult, actor: Actor):
    if isinstance(result.output, EndTurn):
        await result.output.run(thread=self.thread, actor=actor)   # ← 真正落地
    ...

这就接回了第 02 章——回合结束,任务状态更新,循环判断要不要再转一圈。

---

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

1. 把"产出结果"降维成"调用工具"。 结构化输出 = 一次工具调用,于是复用了工具调用的全套基础设施(流式、重试、校验),不必单独写一套输出解析器。核心在 create_mark_task_successful(end_turn.py:36)+ ToolOutput(agent.py:192)。

2. 运行时求值的 dataclass 字段类型。 result: mark_task.get_result_type()(end_turn.py:41)让"字段类型"来自运行时的 Task 声明——这是把第 01 章的"类型声明"和本章的"类型校验"焊在一起的那颗钉子。

3. 闭包 + ClassVar 绑定 + 改 __name__。 工厂函数把 Task/Actor 闭包进去当 ClassVar,再把类名改成带 id 的唯一名(end_turn.py:56),让模型在多任务场景里能分清"给哪个任务交结果"。

4. 声明与执行分离。 EndTurn 工具的 run 方法把"落地动作"封装起来,但直到编排器最后才调用(orchestrator.py:235),中间留出插入事件/审计/持久化的空间。

5. 一次性 agentlet + 私有书签属性。 每回合现造 agentlet(agent.py:211),并偷挂 _marvin_tools / _marvin_end_turn_tools(agent.py:214-215)当事件翻译的映射源——用完即弃,天然隔离每回合的工具集差异。

6. 诚实的 TODO。 多 EndTurn 工具退化成 type(None)(agent.py:186)是个有意识的权衡,代码里用注释和 logger.warning 明确标了出来,还留了 _determine_result_type(agent.py:129)这个"理想版"作为将来收敛的方向。

---

7. 边界与局限

• 多 EndTurn 工具时,output_type 层的强类型校验被弱化。 退化成 type(None)(agent.py:186),此时类型安全主要靠工具本身的 __post_init__ 校验(end_turn.py:47)和编排器的 isinstance 兜底(orchestrator.py:234),而非 pydantic-ai 的 output schema。这也是那个 TODO 未了结的地方。

• _determine_result_type 与 get_agentlet 尚未统一。 前者写的是"多个用 Union",后者内联实现却选了"退化成 None",两者未收敛(inferred,依据是 get_agentlet 中并未调用 _determine_result_type)。

• 依赖 pydantic-ai 的内部结构。 事件翻译层直接读了 parts_manager._parts(streaming.py:197)、依赖 is_*_node / FinalResultEvent 等 pydantic-ai 语义,上游若改这些内部形态,翻译层需要跟着改。

• 本章不覆盖回合调度本身。 agentlet 被"一次次驱动"的循环、任务依赖、多 actor 轮转,见第 02 章;多智能体/委派的放大见第 04 章。

---

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

主题	文件路径	符号名
EndTurn 基类(有 name / run)	src/marvin/engine/end_turn.py	EndTurn
造"标记成功"工具(result 类型化)	src/marvin/engine/end_turn.py	create_mark_task_successful
造"标记失败"工具	src/marvin/engine/end_turn.py	create_mark_task_failed
造"标记跳过"工具	src/marvin/engine/end_turn.py	create_mark_task_skipped
委派给别的 actor	src/marvin/engine/end_turn.py	create_delegate_to_actor
规划子任务(开新 thread)	src/marvin/engine/end_turn.py	create_plan_subtasks
往 thread 发消息(不收尾)	src/marvin/engine/end_turn.py	PostMessage
组装某任务能用的 EndTurn 工具	src/marvin/tasks/task.py	Task.get_end_turn_tools
读出任务的有效结果类型	src/marvin/tasks/task.py	Task.get_result_type
第二道结果校验(含自定义 validator)	src/marvin/tasks/task.py	Task.validate_result
造一次性 agentlet,EndTurn 进 output_type	src/marvin/agents/agent.py	Agent.get_agentlet
"理想版"结果类型判定(未被调用)	src/marvin/agents/agent.py	Agent._determine_result_type
普通工具错误包装	src/marvin/utilities/tools.py	wrap_tool_errors
遍历节点、翻译事件流	src/marvin/engine/streaming.py	handle_agentlet_events
一对一事件翻译	src/marvin/engine/streaming.py	_process_pydantic_event
回合收尾:调用 EndTurn.run 落地	src/marvin/engine/orchestrator.py	Orchestrator.end_turn