主线:一次运行如何跑完 24 步管线(核心)

30 秒导读: 你调用 agent.do(task),Upsonic 内部把这一次运行拆成一条从上到下的线性管线—— 24 个小步骤(缓存检查、建 prompt、装工具、调模型、跑工具、存记忆……)依次执行。每一步都改写同一个共享上下文 AgentRunOutput。真正"思考+动手"的活都挤在第 14 步 ModelExecutionStep 里;其余 23 步是它的前置准备和善后。本章讲清这条管线怎么被驱动、怎么在中途暂停又原地续跑。

本章是全库最核心的一章。读完你应该能回答:一次 agent 运行,从 do() 到拿到结果,中间到底发生了什么、按什么顺序、谁在指挥。上层的 Agent / Task 抽象见 01-agent-and-task.md; 模型怎么被调用见 03-model-layer.md;工具与 HITL 的细节见 04-tools-system.md;两个安全步骤见 05-safety-engine.md。

1. 先建立直觉:什么是"管线"

一次 agent 运行不是一个大函数,而是一串小步骤。 Upsonic 把运行拆成很多个独立的 Step,排成一条队列,由一个"调度员"PipelineManager 从头到尾依次调用。

为什么要这么拆?三个好处,后面会一一兑现:

可观察 —— 每步有名字、有状态、有耗时,debug 时能看到"卡在哪一步"。
可中断可续跑 —— 需要等人确认时,记住"停在第几步",人回话后从那一步继续,不重头再来。
可组合 —— 普通调用、超时调用、流式调用只是换一份步骤清单,主循环不变。

一个生活化的类比:把它当成工厂流水线。原料(你的 Task)从传送带一头进去,依次经过 24 个工位 (每个工位就是一个 Step),每个工位在同一块"工单"(AgentRunOutput)上盖章、补信息,最后一头出来就是成品(结果文本 / 结构化对象)。

三个最基础的概念

管线的地基是三个类型,都在 agent/pipeline/step.py:

概念	是什么	源码
`Step`	一个工位的抽象基类。子类实现 `execute()` 干活	`agent/pipeline/step.py:91` `Step`
`StepResult`	一步跑完的收据:名字、序号、状态、耗时、消息	`agent/pipeline/step.py:59` `StepResult`
`StepStatus`	这一步的结局枚举:`RUNNING/COMPLETED/PAUSED/CANCELLED/ERROR/SKIPPED`	`agent/pipeline/step.py:35` `StepStatus`

StepStatus 有个细节值得记住:它能映射回运行级的状态 RunStatus,其中 SKIPPED 会被当成 completed(跳过 = 正常),而 PAUSED 会保留成暂停——这为后面的"暂停续跑"埋了伏笔 (agent/pipeline/step.py:44 StepStatus.to_run_status)。

2. 顶层全景:Pipeline 架构三件套

整条管线只由三样东西撑起来:

Step 基类(step.py)—— 定义"一个工位长什么样、生命周期怎么走"。
PipelineManager(manager.py)—— 调度员,持有步骤清单 + task/agent/model,按序驱动, 并统一处理取消 / 暂停 / 错误。
agent.py 的组装 & 分发 —— 三个 _create_*_pipeline_steps 决定清单里放哪些步骤, _do_async_pipeline 决定用哪条执行路径(普通 / 超时 / 流式)。

一句话概括它们的关系:

agent.py 决定"放哪些步骤、走哪条路"  ──►  PipelineManager 按序驱动  ──►  每个 Step 改写共享 context
   (_create_*_steps / _do_async_pipeline)     (execute / execute_stream)        (AgentRunOutput)

贯穿始终的"单一事实源":AgentRunOutput

没有参数在步骤间传来传去。 所有步骤共享同一个 context: AgentRunOutput 对象——它既是输入也是输出。 ModelSelectionStep 把选好的模型写进去,SystemPromptBuildStep 把 prompt 写进去,ModelExecutionStep 把模型回复写进 context.response 和 context.chat_history……下一步直接从 context 读。

Step.execute() 的签名把这点写得很直白——每个步骤都拿到全套 context/task/agent/model (agent/pipeline/step.py:147 Step.execute):

# 示意,非源码 —— 每个步骤的统一入口
async def execute(self, context, task, agent, model, step_number, pipeline_manager=None) -> StepResult:
    ...              # 从 context 读上一步的产物
    context.xxx = y  # 把自己的产物写回 context
    return StepResult(name=self.name, status=StepStatus.COMPLETED, ...)

3. 一次运行从上到下:24 步全景图

先看整条直连(direct)管线的分阶段骨架(这张图只画 5 个大阶段,方便记忆;逐步清单在下方表里):

   agent.do(task)
        │
        ▼
 ┌──────────────────┐
 │ ① 准备环境        │  Init → Storage连接 → 缓存命中检查 → 选模型 → 装工具
 └──────────────────┘
        │
        ▼
 ┌──────────────────┐
 │ ② 组装这次请求    │  载入记忆 → 建系统prompt → 建任务上下文(RAG) → 载入历史
 │                  │  → 用户安全策略 → 建用户输入 → 拼成 ModelRequest
 └──────────────────┘
        │
        ▼
 ┌──────────────────┐
 │ ③ 调模型(核心)  │  CallManager就绪 → ModelExecution
 │                  │      └─ 内含"调模型→跑工具→再调模型"的循环(见 §7)
 └──────────────────┘
        │
        ▼
 ┌──────────────────┐
 │ ④ 加工输出        │  记录消息 → 反思(Reflection) → 任务管理 → 可靠性校验 → agent安全策略
 └──────────────────┘
        │
        ▼
 ┌──────────────────┐
 │ ⑤ 收尾落库        │  写缓存 → Finalization → 存记忆 → 记账(task_end)
 └──────────────────┘
        │
        ▼
     结果返回

怎么读这张图: 从上到下就是真实执行顺序,命中缓存 / 被策略拦截时,中间步骤会被"跳过" (SKIPPED)但仍然依次经过。第 ③ 阶段的 ModelExecution 是唯一会自己内部循环、也是唯一会"暂停" 整条管线的步骤。

直连管线的完整 24 步

以下是 _create_direct_pipeline_steps 里真实的步骤顺序与索引(agent/agent.py:4650)。注意:实际组装顺序和直觉略有出入(比如 StorageConnection 排在 CacheCheck 前、UserPolicy 排在 ChatHistory 之后),下表以代码为准:

#	步骤类	name	干什么(一句话)
0	`InitializationStep`	initialization	初始化本次运行的 `AgentRunOutput`、run_id、统计
1	`StorageConnectionStep`	storage_connection	连接会话存储后端
2	`CacheCheckStep`	cache_check	查缓存;命中则后续步骤大多 `SKIPPED`
3	`LLMManagerStep`	llm_manager	准备 LLM 层管理器
4	`ModelSelectionStep`	model_selection	解析 `provider/model`,定下本次用的 `Model`
5	`ToolSetupStep`	tool_setup	收集/注册工具(自定义 + MCP),生成工具定义
6	`MemoryPrepareStep`	memory_prepare	建 `MemoryManager`,载入历史/画像/元数据
7	`SystemPromptBuildStep`	system_prompt_build	组装系统 prompt
8	`ContextBuildStep`	context_build	建任务上下文(知识库 / RAG 检索)
9	`ChatHistoryStep`	chat_history	载入对话历史,并标记本次运行边界
10	`UserPolicyStep`	user_policy	在建输入前跑用户侧安全策略(见第5章)
11	`UserInputBuildStep`	user_input_build	拼用户输入(prompt + 上下文 + 附件)
12	`MessageAssemblyStep`	message_assembly	组 `ModelRequest`,过上下文管理中间件
13	`CallManagerSetupStep`	call_manager_setup	建并注册 `CallManager`(负责打印/记录响应)
14	`ModelExecutionStep`	model_execution	核心:调模型 + 工具调用循环;外部工具从这里续跑
15	`ResponseProcessingStep`	response_processing	把消息记录进 `AgentRunOutput`
16	`ReflectionStep`	reflection	可选的自我反思/改写
17	`TaskManagementStep`	task_management	任务级状态管理
18	`ReliabilityStep`	reliability	可靠性层:校验/清洗输出
19	`AgentPolicyStep`	agent_policy	agent 侧安全策略(见第5章)
20	`CacheStorageStep`	cache_storage	把结果写入缓存
21	`FinalizationStep`	finalization	收尾整理
22	`MemorySaveStep`	memory_save	完成态下存会话(历史/摘要/画像),并最终化消息
23	`CallManagementStep`	call_management	最后一步:调 `task_end()`、打印指标、记账

这份清单是硬编码的,不是配置驱动的——想改顺序就改这个方法。PipelineManager 只负责"照单执行"。

4. Step 基类:一个工位的生命周期

Step 是抽象基类,子类只必须实现两样:name 属性和 execute() 方法 (agent/pipeline/step.py:91 Step)。基类提供四个执行入口,分工如下:

方法	用途	返回
`execute()`	子类实现的核心逻辑,含 try-except-finally 生命周期	`StepResult`
`run()`	非流式的极薄包装,直接调 `execute()`	`StepResult`
`execute_stream()`	流式步骤覆写它,`yield` 出 `AgentEvent` 事件	事件异步流
`run_stream()`	流式包装:先发 `StepStartEvent`,跑 `execute_stream()`,再发 `StepEndEvent`	事件异步流

execute() 的固定套路:try-except-finally

每个真实步骤的 execute() 都长一个样(以 MemoryPrepareStep 为例,agent/pipeline/steps.py:760):

# 示意,非源码 —— 所有步骤共用的骨架
async def execute(self, context, task, agent, model, step_number, pipeline_manager=None):
    step_result = None
    try:
        raise_if_cancelled(agent.run_id)          # 1. 先检查有没有被取消
        if task._cached_result: ...return SKIPPED # 2. 缓存命中就跳过
        if task._policy_blocked: ...return SKIPPED # 3. 被策略拦截也跳过
        ...做正事...                               # 4. 干活,写回 context
        step_result = StepResult(status=COMPLETED, ...)
        return step_result
    except RunCancelledException: ...; raise       # 取消:标 CANCELLED 后上抛
    except Exception: ...; raise                   # 出错:标 ERROR 后上抛
    finally:
        if step_result:
            self._finalize_step_result(step_result, context)  # 5. 收据入账

重点看两处:

步首三连检查 —— "被取消了吗?缓存命中吗?被策略拦了吗?"。这让每个步骤天然支持"提前跳过", 缓存命中时几乎所有中间步骤都会 SKIPPED(steps.py:779 起的 _cached_result 分支)。
finally 里必然入账 —— 不管成功、跳过、报错,都会把 StepResult 交给 _finalize_step_result:它把收据追加进 context.step_results,并更新统计 (agent/pipeline/step.py:124 _finalize_step_result)。正是这份收据记录了"停在第几步、什么状态", 成了后面续跑的路标。

5. PipelineManager:调度员怎么按序驱动 + 出事怎么办

PipelineManager 持有步骤清单和 task/agent/model,核心是两个方法:execute()(非流式)和 execute_stream()(流式)。还有一个管理器注册表 _managers,让步骤之间共享像 memory_manager、 call_manager 这样的重对象(manager.py:871 set_manager / get_manager)。

execute():主循环极其朴素

去掉遥测和日志,execute() 的骨架就是一个 for 循环(agent/pipeline/manager.py:230 execute):

# 示意,非源码 —— 主循环的本质
for step_index in range(start_step_index, len(self.steps)):
    step = self.steps[step_index]
    result = await step.run(context, self.task, self.agent, self.model, step_index, pipeline_manager=self)
    # result 已经被 step 自己 finalize 进 context 了

注意 start_step_index 参数:默认 0(从头跑),续跑时传一个中间索引,从那一步开始跑——这是整个 "暂停续跑"能力的物理基础(manager.py:256 记录 resumed_from)。

出事怎么办:异常分诊(本章精华之一)

主循环外裹着一个大 try/except。Upsonic 用异常类型来区分四种"非正常结局",并给出完全不同的处理 (agent/pipeline/manager.py:405 起):

抛出的异常	含义	处理方式	是否重新抛出
`ConfirmationPause`	工具需人工确认	建 requirement,标 `paused("confirmation")`,存档	否,正常返回
`UserInputPause`	工具需人补参数	建 requirement,标 `paused("user_input")`,存档	否,正常返回
`ExternalExecutionPause`	工具要外部执行	建 requirement,标 `paused("external_tool")`,存档	否,正常返回
`RunCancelledException`	运行被取消	`_ahandle_cancellation`:标 `cancelled`,存档	否,正常返回
其它任意 `Exception`	真出错	`_ahandle_durable_execution_error`:标 `error`,存档	是,再抛给调用方

关键洞察:前四种不是"失败",而是"暂停/挂起"——管线把当前进度存进会话(_save_session),然后 正常 return context,让调用方拿到一个"带 requirements 的暂停态输出"。只有最后一种"真错误"会在存档后把异常继续抛出去。

三种 HITL 暂停的处理器长得几乎一样(manager.py:749 _handle_external_tool_pause / manager.py:792 _handle_confirmation_pause / manager.py:831 _handle_user_input_pause):都是把 e.paused_calls 里每个挂起的工具调用,转成一条 RunRequirement 挂到 output,再 mark_paused(...)。

这四条处理器都会调 _save_session(output) 把检查点落库(manager.py:113)。即便是"真错误",也先存档再抛——这就是注释里说的 durable execution(可持久化执行):进程崩了也能从存档恢复。

execute_stream():多了什么

流式版本(agent/pipeline/manager.py:490 execute_stream)在结构上和 execute 同源,区别是:

开跑先发 RunStartedEvent + PipelineStartEvent,每步前后发 StepStartEvent/StepEndEvent。
对支持流式的步骤(step.supports_streaming 为真,主要是 StreamModelExecutionStep)调 run_stream(),把模型吐字的 delta 事件实时 yield 出去(manager.py:584)。
其余步骤仍走 run(),把它们期间往 context.events 追加的事件补发出去。
结尾在 finally 里发 PipelineEndEvent,完成态再发 RunCompletedEvent(manager.py:693)。

6. agent.py 怎么组装 & 选路

PipelineManager 只管"照单执行";"单子里放什么、走哪条执行路"由 agent.py 决定。

三份步骤清单

组装方法	用于	步数	关键差异
`_create_direct_pipeline_steps`	`do()` / `do_async()` 普通调用	24	用 `ModelExecutionStep`,含 `MemorySave` + `CallManagement`
`_create_streaming_pipeline_steps`	`stream()` 流式	22	换成 `StreamModelExecutionStep`,用 `StreamFinalization` + `StreamMemoryMessageTracking`
`_create_full_pipeline_steps(is_streaming)`	上面两者的分发入口	—	`is_streaming` 决定返回哪一份

_create_full_pipeline_steps 就是个二选一开关(agent/agent.py:4749):

# 示意,非源码
def _create_full_pipeline_steps(self, is_streaming=False):
    if is_streaming:
        return self._create_streaming_pipeline_steps()   # 22 步,流式
    return self._create_direct_pipeline_steps()          # 24 步,直连

流式清单把核心执行步换成 StreamModelExecutionStep(name=stream_model_execution, supports_streaming=True,steps.py:2977),收尾则用 StreamFinalizationStep(steps.py:3588)与 StreamMemoryMessageTrackingStep(steps.py:3421,负责存会话 + task_end)。

_do_async_pipeline:三条执行路径

选好清单后,_do_async_pipeline 根据 timeout / partial_on_timeout 走三条不同的路 (agent/agent.py:3898):

              partial_on_timeout=True 且有 timeout ?
                       │是                              │否
                       ▼                                ▼
        ┌──────────────────────────┐         timeout is not None ?
        │ 路径A:流式 + 超时容错     │          │是            │否
        │ 用 streaming 清单         │          ▼             ▼
        │ asyncio.wait_for 包住     │   ┌────────────┐  ┌────────────┐
        │ 超时→返回已生成的部分文本 │   │ 路径B:     │  │ 路径C:     │
        └──────────────────────────┘   │ 直连+硬超时 │  │ 直连普通    │
                                        │ 超时→抛     │  │ 直接 execute│
                                        │ TimeoutError│  └────────────┘
                                        └────────────┘

路径 C(默认) —— 最朴素:建 PipelineManager(直连清单),await pipeline.execute(...),取结果 (agent/agent.py:3995)。
路径 B(硬超时) —— 同样直连,但 asyncio.wait_for(pipeline.execute(...), timeout) 包住;超时就取消运行并抛 ExecutionTimeoutError(agent/agent.py:3965)。
路径 A(流式+部分结果) —— 改用流式清单,边消费流边把文本累积进 accumulated_text;超时不报错,而是把已经生成的半截文本当结果返回,并在 metadata 里标 timeout=True(agent/agent.py:3911)。因为流式清单里没有 CallManagementStep,这条路会手动补跑一次记账(_run_call_management_step,agent/agent.py:4011)。

三条路都传 start_step_index:普通运行是 0,HITL 续跑时是"上次暂停的那一步"(见 §8)。

7. 核心中的核心:ModelExecutionStep 与工具调用循环

第 14 步 ModelExecutionStep 是整条管线唯一"真正思考+动手"的地方 (agent/pipeline/steps.py:1570 ModelExecutionStep)。它做的事:

步首三连检查(取消 / 缓存命中 / 策略拦截 / 上下文已满),命中就跳过。
从注册表取 memory_manager、call_manager(取不到就兜底新建,这正是续跑场景:管线从第 14 步起跑, 前面建这些管理器的步骤没跑过)(steps.py:1640、steps.py:1656)。
调模型:response = await model.request(messages=context.chat_history, ...) (steps.py:1708)。
把响应交给 agent._handle_model_response(response, context.chat_history) —— 工具调用循环就藏在这里。
把最终响应写回 context.response / context.chat_history,收工。

工具调用循环:一次 Step 内的"自我递归"

_handle_model_response 是 agentic loop 的心脏(agent/agent.py:2662)。它的逻辑:

         模型响应 response
              │
              ▼
   response 里有"常规工具调用"吗?（排除结构化输出的 output 工具）
       │否                         │是
       ▼                           ▼
   直接返回 response      _execute_tool_calls(工具们)  ← 真正执行工具
   (本轮结束)                      │
                                   ▼
                    把工具结果拼成 ModelRequest 追加进 messages
                                   │
                                   ▼
                    再调一次 model.request(...)  →  follow_up_response
                                   │
                                   ▼
              递归:_handle_model_response(follow_up_response, messages)

也就是说:"调模型 → 模型要求用工具 → 跑工具 → 把结果喂回去 → 再调模型",这个循环全部发生在第 14 步内部,通过 _handle_model_response 的自我递归实现,直到某轮模型不再要工具、或触发停止条件才返回 (agent/agent.py:2937 递归调用)。

循环里几个真实的收口分支(都在 _handle_model_response 内):

工具数量到顶 —— _tool_limit_reached 为真时,追加一条"工具已达上限"的系统提示,让模型基于现有信息给最终答复,不再递归(agent/agent.py:2819)。
max_tokens 截断 —— finish_reason == 'length' 且有工具调用时,参数可能不完整,于是回一条"你的输出被截断,请重试"给模型,重来而不是执行残缺的调用(agent/agent.py:2723)。
提前停止 —— 某个工具结果里带 _stop_execution 标记(或是结构化输出 output 工具),直接把它当最终文本返回,不再递归(agent/agent.py:2866 起的 should_stop)。

_execute_tool_calls 负责真正跑工具(agent/agent.py:2311):按工具定义分成顺序执行和可并行两组, 跑前还会过一道 post-execution 工具策略校验,并累加 _tool_call_count、检查 tool_call_limit。工具本身怎么定义 / MCP 怎么接入,见 04-tools-system.md。

8. HITL 暂停:管线怎么"停下来等人",又原地续跑

HITL(Human-In-The-Loop,人在环)是本章第二个精华。核心问题:模型想调一个"需要人确认 / 需要人补参数 / 要外部系统执行"的工具时,管线怎么优雅地停在半路,等人回话后从原地继续,而不是从头重跑?

停:一路异常上抛,在 PipelineManager 变成检查点

暂停靠异常实现。工具执行层在遇到这三类工具时直接抛异常(tools/execution.py:83-89):

# 示意,非源码 —— tools/execution.py 里的分诊
if 需要确认:      raise ConfirmationPause()
elif 需要用户输入: raise UserInputPause()
elif 需要外部执行: raise ExternalExecutionPause()

这个异常的传播链:

tools/execution.py 抛 XxxPause
        │  向上冒泡
        ▼
_execute_tool_calls → _handle_model_response  (不拦截,继续上抛)
        ▼
ModelExecutionStep.execute 的 except (捕获→把 StepResult 标成 PAUSED→再抛)  [steps.py:1785]
        ▼
PipelineManager.execute 的 except (按类型分诊)                            [manager.py:410]
        ▼
_handle_xxx_pause:为每个挂起工具建 RunRequirement,mark_paused,存档,正常 return

注意 ModelExecutionStep 在 except 里先把自己的 StepResult 状态设成 PAUSED 再重新抛 (steps.py:1785)——这条 PAUSED 收据会进 context.step_results,记下"停在第 14 步",成为续跑路标。

续:从暂停的那一步重新开跑

人处理完(设好外部工具结果 / 确认 / 补参数)后调 continue_run_async。它的关键动作 (agent/agent.py:5247 continue_run_async → _prepare_continuation_context,agent/agent.py:5047):

从存档取回暂停态的 output,找到那条"有问题的步骤"记录: resume_step_index = problematic_step.step_number(agent/agent.py:5204)。对外部工具暂停来说,这个索引就是 model_execution(14)。
把人给的结果注入回 chat_history(_inject_hitl_results,agent/agent.py:4772):外部结果直接用,确认通过就真跑工具、拒绝就注入一句拒绝说明,用户输入则用补的参数跑工具。
带着 _resume_output=output 和 _resume_step_index=resume_step_index 再调 do_async (agent/agent.py:5367)——最终传给 pipeline.execute(context, start_step_index=resume_step_index), 管线直接从第 14 步开始跑,前 14 步一概不重来。

有个边界处理很讲究:如果续跑点在 chat_history 步(第 9 步)之后,ChatHistoryStep 被跳过了,于是 _prepare_continuation_context 会手动 output.start_new_run() 补上运行边界标记,保证注入的工具结果也被算进这次运行的消息里(agent/agent.py:5206)。

continue_run 的外层还有个最多 10 轮的循环(_continue_run_direct_impl,agent/agent.py:5339),让"确认→ 又要确认→再确认"这种连续 HITL 能自动接力。

9. 记忆的两步:prepare 与 save

本章不单开记忆章,这里顺带交代管线里的记忆两步(细节属于记忆子系统):

第 6 步 MemoryPrepareStep(load) —— 建 MemoryManager,aprepare() 载入历史 / 用户画像 / 元数据, 并把 manager 放进注册表供后续步骤(尤其第 14 步)取用(agent/pipeline/steps.py:749)。
第 22 步 MemorySaveStep(save) —— 仅在完成态下经 MemoryManager.afinalize() 存会话(含摘要 / 画像更新),并在此最终化消息——刻意放到这么靠后,是为了等 AgentPolicyStep 等可能追加消息的步骤都跑完再抽取新消息(agent/pipeline/steps.py:2323 类注释)。

而暂停 / 取消 / 出错这些非完成态,会话不是这一步存的,而是前面 §5 里 PipelineManager 的异常处理器调 _save_session 存的检查点。两条存档路径分工明确:正常收尾走 MemorySaveStep,异常挂起走 manager。

10. 巧妙之处(可借鉴)

用异常类型做控制流分诊 —— 暂停 / 取消 / 错误不是靠返回码,而是靠不同的异常类,在 PipelineManager.execute 一处集中分诊(manager.py:405)。三种 HITL 暂停被当成"正常返回",只有真错误才继续上抛,语义干净。
共享 context,零参数穿线 —— 24 个步骤不互相传参,全部读写同一个 AgentRunOutput,新增步骤几乎零接口成本(step.py:147 统一签名)。
StepResult 收据 = 续跑路标 —— 每步的状态 / 序号都记进 context.step_results,续跑时靠 get_problematic_step().step_number 精确定位从哪一步重开(agent/agent.py:5200),不需要额外的状态机。
兜底新建管理器 —— 续跑从第 14 步起跑,前面建 memory_manager / call_manager 的步骤没执行, ModelExecutionStep 于是内置了"取不到就新建"的兜底(steps.py:1656),让"从半路开跑"成立。
换清单即换模式 —— 流式 / 非流式只是两份步骤清单 + 一个 is_streaming 开关,主循环完全复用 (agent/agent.py:4749)。

11. 边界与局限

步骤顺序硬编码 —— 清单写死在 _create_*_pipeline_steps 里,不是配置或插件式;要增删步骤得改这两个方法(agent/agent.py:4650 / :4724)。
续跑锚在 step_number(索引) —— 依赖清单顺序稳定。如果直连与流式两份清单的索引含义不同,续跑必须用与暂停时同一种清单;代码用 _get_step_index_by_name 按名字换算来抵消一部分脆弱性 (agent/agent.py:4677)。
流式清单缺 CallManagementStep —— 所以 partial_on_timeout 路径要手动补跑记账 (agent/agent.py:4011),否则用量统计会漏。
HITL 自动接力上限 10 轮 —— _continue_run_direct_impl 写死 max_rounds = 10,超过就不再自动续 (agent/agent.py:5357)。
工具调用循环深度受 tool_call_limit 约束 —— 没设上限时,_handle_model_response 的递归深度由模型何时停止决定,理论上可能很深(agent/agent.py:2778)。

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

主题	文件路径	符号名
步骤基类与生命周期	`agent/pipeline/step.py`	`Step`、`Step.execute`、`Step.run_stream`
步骤收据 / 状态枚举	`agent/pipeline/step.py`	`StepResult`、`StepStatus`、`_finalize_step_result`
调度主循环(非流式)	`agent/pipeline/manager.py`	`PipelineManager.execute`
调度主循环(流式)	`agent/pipeline/manager.py`	`PipelineManager.execute_stream`
异常分诊 / 取消 / 错误存档	`agent/pipeline/manager.py`	`_ahandle_cancellation`、`_ahandle_durable_execution_error`
三种 HITL 暂停处理器	`agent/pipeline/manager.py`	`_handle_external_tool_pause`、`_handle_confirmation_pause`、`_handle_user_input_pause`
步骤间共享管理器	`agent/pipeline/manager.py`	`set_manager`、`get_manager`、`_save_session`
直连 24 步清单	`agent/agent.py`	`_create_direct_pipeline_steps`
流式 22 步清单	`agent/agent.py`	`_create_streaming_pipeline_steps`
清单分发开关	`agent/agent.py`	`_create_full_pipeline_steps`
三条执行路径(普通/超时/流式)	`agent/agent.py`	`_do_async_pipeline`
核心执行步	`agent/pipeline/steps.py`	`ModelExecutionStep`
工具调用循环(自我递归)	`agent/agent.py`	`_handle_model_response`
真正执行工具	`agent/agent.py`	`_execute_tool_calls`
HITL 续跑准备	`agent/agent.py`	`continue_run_async`、`_prepare_continuation_context`、`_inject_hitl_results`
暂停异常定义	`tools/hitl.py`	`ExternalExecutionPause`、`ConfirmationPause`、`UserInputPause`
暂停异常抛出点	`tools/execution.py`	`raise ConfirmationPause / UserInputPause / ExternalExecutionPause`
记忆载入 / 存档两步	`agent/pipeline/steps.py`	`MemoryPrepareStep`、`MemorySaveStep`

1. 先建立直觉:什么是"管线"​

三个最基础的概念​

2. 顶层全景:Pipeline 架构三件套​

贯穿始终的"单一事实源":AgentRunOutput​

3. 一次运行从上到下:24 步全景图​

直连管线的完整 24 步​

4. Step 基类:一个工位的生命周期​

execute() 的固定套路:try-except-finally​

5. PipelineManager:调度员怎么按序驱动 + 出事怎么办​

execute():主循环极其朴素​

出事怎么办:异常分诊(本章精华之一)​

execute_stream():多了什么​

6. agent.py 怎么组装 & 选路​

三份步骤清单​

_do_async_pipeline:三条执行路径​

7. 核心中的核心:ModelExecutionStep 与工具调用循环​

工具调用循环:一次 Step 内的"自我递归"​

8. HITL 暂停:管线怎么"停下来等人",又原地续跑​

停:一路异常上抛,在 PipelineManager 变成检查点​

续:从暂停的那一步重新开跑​

9. 记忆的两步:prepare 与 save​

10. 巧妙之处(可借鉴)​

11. 边界与局限​

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