一次通话的端到端编排与引擎旁路能力

30 秒导读: 前三章分别讲了"图长什么样"(01)、"帧怎么在处理器间流"(02)、"图怎么变成状态机工具"(03)。这一章把它们串成一根线:一个 WebSocket / WebRTC / 电话连接进来,系统怎么一步步把它变成一通真实通话,跑完,再干净地收尾。核心是 run_pipeline.py 里的 _run_pipeline_impl —— 通话的"总装配车间"。

本章覆盖三件事:

1. 入口:连接从哪几个口进来,进来后各自做什么(建 run、校配额、装 transport)。
2. 编排主体:_run_pipeline_impl 如何用运行时快照装配一整套 services→引擎→管线→Worker,注册观测器,跑完,收尾。
3. 旁路/带外能力:变量抽取、上下文压缩、知识库检索、pre-call fetch、语音信箱检测、录音路由 —— 这些"不在主音频管线里"的能力,是在这条编排线的哪一步接上去的。

不重复的部分:管线内部帧结构看 02-voice-pipeline;图→工具的状态机机制看 03-pipecat-engine;供应商注册表看 05-extensibility-registries。

---

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

一句话定义: 这是 Dograh 的"通话主控程序" —— 从一个连接建立,到一通语音通话跑完并把录音、转写、用量落库,全程由它编排。

想象一家呼叫中心。接线员(入口路由)接起电话,核对工号和额度;调度台(_run_pipeline_impl)把这通电话需要的所有设备(麦克风、喇叭、翻译、大脑)一次性配齐、接线,然后按下"开始";质检员(观测器)全程记录;通话结束,清洁工(收尾逻辑)关掉设备、封存录音、结账。

这一章讲的就是调度台 + 接线员 + 清洁工,不是"大脑怎么想"(那是引擎,03 章)。

它要解决的真问题: 一通语音通话涉及十几个组件(STT、TTS、LLM、VAD、转写聚合、录音、观测、集成会话、MCP……),而且入口五花八门(浏览器 WebRTC、七八家电话运营商、通用 WebSocket)。如果每个入口各写一套装配,代码会爆炸。Dograh 把入口收敛成薄薄的适配层,把装配逻辑全部收敛到一个 _run_pipeline_impl。

一句话直觉: 入口负责"把连接变成一个 transport 对象 + 一个 workflow_run_id",剩下的所有脏活累活都交给同一个编排函数。换入口不改编排,加供应商不改编排。

---

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

一通通话的生命周期,是"连接 → 编排 → 观测 → 收尾"四段:

  ┌──────────── 入口层(薄适配) ────────────┐
  │  浏览器 Web Call    通用 WebSocket    电话  │
  │  webrtc_signaling   agent_stream    telephony/providers/* │
  └───────┬──────────────┬─────────────────┬──┘
          │   建 workflow_run + 校配额 + 装 transport
          └──────────────┼─────────────────┘
                         ▼
        ┌────────────────────────────────────────┐
        │   _run_pipeline_impl  (编排主体)          │
        │                                          │
        │  1 取运行时快照定义(pinned definition)     │
        │  2 解析 run_configs + effective model cfg  │
        │  3 判定 is_realtime,造 services           │
        │  4 造 PipecatEngine(注入图/上下文/回调)     │
        │  5 装管线 → PipelineWorker(task)          │
        │  6 engine.initialize()(系统提示+工具+MCP)  │
        │  7 注册观测器(反馈/延迟/turn log/buffer)   │
        │  8 run_pipeline_worker(task) ── 跑完一通   │
        │  9 finally: close_mcp_sessions + cleanup  │
        └────────────────────────────────────────┘
                         ▲  ▲  ▲
          带外能力在装配期接线(不进主音频帧流):
          变量抽取 · 上下文压缩 · 知识库 · pre-call fetch
          · 语音信箱检测 · 录音路由

部件一句话职责:

部件	干什么	在哪
入口路由	接连接、建 run、校配额、造 transport	routes/agent_stream.py、routes/webrtc_signaling.py、services/telephony/providers/*
_run_pipeline_impl	编排主体:装配→运行→收尾	services/pipecat/run_pipeline.py:415
register_active_call	通话计数,供发布时优雅排空(drain)	services/pipecat/active_calls.py
run_pipeline_worker	走 pipecat v1.3 的 WorkerRunner 生命周期	services/pipecat/worker_runner.py:7
观测器	反馈事件、延迟、turn log 落 buffer/WS	services/pipecat/event_handlers.py、realtime_feedback_observer.py
收尾	落库、封存录音/转写、enqueue 后处理	event_handlers.py:219 on_pipeline_finished

主线走一遍(高层): 连接进来 → 入口建 workflow_run 并校配额 → 入口造出 transport → 调 _run_pipeline_impl → 它按快照装好一整套 → engine.initialize() 备好系统提示和工具 → run_pipeline_worker 让音频真正开跑 → 用户挂断 / 到时 / 语音信箱触发结束 → 收尾落库 → finally 关 MCP。

---

3. 三个入口:连接怎么变成一通通话

所有入口最终都汇聚到 _run_pipeline_impl,但它们建 run、校配额、造 transport 的方式不同。

入口	路径	谁用	凭证从哪来
通用 agent-stream	routes/agent_stream.py	任意外部方,凭证内联在查询串	query string(含 provider 凭证)
WebRTC 信令	routes/webrtc_signaling.py	浏览器 Web Call(SmallWebRTC)	登录用户 / embed session token
电话	services/telephony/providers/*	Twilio/Plivo/Telnyx/Vonage/Vobiz/ARI/Cloudonix	运营商 config 行

3.1 通用入口 agent_stream.py:内联凭证

/agent-stream/{workflow_uuid} 是一个"万能口":调用方把 provider 名、主被叫号、甚至 provider 凭证全塞在查询串里,不需要在组织里预存 TelephonyConfigurationModel 行(agent_stream.py:1-13 的模块 docstring 点明了这层区别)。

它进来后按顺序做四件事(agent_stream.py:80-125):

1. 建 workflow_run —— db_client.create_workflow_run(...),把 provider、主被叫号、direction="inbound" 存进 initial_context(agent_stream.py:73-93)。
2. 设运行上下文 —— set_current_run_id / set_current_org_id,让后续日志和 trace 带上 run/org(agent_stream.py:95-96)。
3. 校配额 —— authorize_workflow_run_start(...);has_quota 为假就用错误信息关闭 WebSocket(agent_stream.py:98-110)。
4. 派发 —— provider_instance.handle_external_websocket(...),把这个 socket 交给注册表里对应 provider 的实现(agent_stream.py:116-125),后者最终会调 run_pipeline_telephony。

注意边界: 没有 ?provider= 的"裸音频"分支目前未实现,直接以 1011 关闭(agent_stream.py:50-56)—— 代码里明写"reserved for a future protocol decision"。

3.2 浏览器 Web Call webrtc_signaling.py:SmallWebRTC 信令

浏览器打电话走 WebRTC。这个文件用 WebSocket 信令 + ICE trickling(边收集边发候选)取代 HTTP PATCH,因为多 worker 部署下本地 _pcs_map 无法共享(webrtc_signaling.py:1-15)。

关键动作在 _handle_offer(webrtc_signaling.py:396-537):

1. 收到 offer 后设 run/org 上下文,并先校配额 authorize_workflow_run_start(webrtc_signaling.py:429-445)。
2. 新建 SmallWebRTCConnection,用 get_ice_servers(user_id=...) 塞入按用户生成的时限 TURN 凭证(webrtc_signaling.py:477-485)。
3. 注册 WS 反馈通道 register_ws_sender(workflow_run_id, ws_sender) —— 这就是后面观测器把实时反馈推给浏览器的那根管子(webrtc_signaling.py:491-495)。
4. 后台起管线 asyncio.create_task(run_pipeline_smallwebrtc(...)),不阻塞信令(webrtc_signaling.py:513-522);随后把 answer 发回浏览器,ICE 候选另行 trickle。

还有一个 public/signaling/{session_token} 公开口(embed 嵌入用),多一层 token 校验 + 来源域校验 validate_origin,防止泄露的 token 从任意站点接入(webrtc_signaling.py:643-697)。

3.3 电话入口:经 telephony providers

七家运营商的 provider 各自处理完自己的信令握手后,统一调 run_pipeline_telephony(如 services/telephony/providers/twilio/provider.py:323、plivo/provider.py:346 等,共七家)。配额校验发生在更上游的运营商回调里;这里只负责把 socket 变成 transport。供应商如何插进注册表,见 05-extensibility-registries。

3.4 三个入口的会合点:register_active_call

三条路各有一个薄包装函数,职责相同:在任何异步 setup 之前先登记这通活跃通话,finally 里注销。

# services/pipecat/run_pipeline.py:165  run_pipeline_telephony(节选,真实源码)
register_active_call(workflow_run_id)   # 先登记,再干活
try:
    await _run_pipeline_telephony_impl(...)   # 解析 run、算 is_realtime、造 transport
finally:
    unregister_active_call(workflow_run_id)   # 无论如何注销

为什么先登记? 注释点明:发布(deploy)时要优雅排空在跑的通话;必须让排空逻辑也看得见那些"还在解析 DB/config/transport 状态"的通话,所以登记要早于一切 async setup(run_pipeline.py:176-178)。

三个包装(run_pipeline_telephony:165、run_pipeline_smallwebrtc:292、_run_pipeline:386)各自解析出 transport 后,都汇入下面的 _run_pipeline_impl。

---

4. 编排主体:_run_pipeline_impl 主流程

这是本章的心脏(run_pipeline.py:415)。它拿到 transport + workflow_run_id,把一通通话需要的一切从零装好。按代码顺序,它是这样一步步走的:

4.1 幂等闸门 + 运行时快照(最关键的设计)

先挡掉重复运行:workflow_run 已完成就抛 400(run_pipeline.py:442-443)。

然后取 pinned definition —— 运行时快照定义。这是整个编排最值得记住的一点:

# run_pipeline.py:456  真实源码
# Use the run's pinned definition for graph + configs (not the workflow's current)
run_definition = workflow_run.definition
run_workflow_json = run_definition.workflow_json
run_configs = run_definition.workflow_configurations or {}

为什么要快照? 用户随时在编辑器里改工作流图。如果一通正在跑的通话读"当前的图",用户一保存,通话中途图就变了 —— 灾难。所以每个 workflow_run 在创建时就钉死了当时那一版定义(workflow_json),运行期只认这份快照,不认最新图。这份 JSON 随后喂给 WorkflowGraph(ReactFlowDTO.model_validate(run_workflow_json)) 构图(run_pipeline.py:576-579),图模型细节见 01-workflow-model。

4.2 解析配置 + 有效模型配置

从 run_configs(即快照里的 workflow_configurations)里抠出运行参数(run_pipeline.py:462-486):

参数	默认	作用
max_call_duration	300s	最长通话时长
max_user_idle_timeout	10.0s	用户静默多久算 idle
smart_turn_stop_secs	2.0s	智能轮次结束超时
turn_stop_strategy	transcription	轮次结束检测策略
dictionary	无	STT 关键词增强(逗号分隔→keyterms)

再算 effective AI model config:get_effective_ai_model_configuration_for_workflow(...) 把这一版工作流的 model_overrides 叠加到组织级用户配置上(run_pipeline.py:490-501)。入口层若已算过就直接复用(resolved_user_config),避免二次拉取。

4.3 判定 is_realtime,装配 services

is_realtime = user_config.is_realtime and user_config.realtime is not None(run_pipeline.py:517)—— 决定走"语音到语音"(OpenAI Realtime、Gemini Live)还是传统 STT→LLM→TTS 三段式。这个布尔值会贯穿后面几乎每个分支。

is_realtime = True                 is_realtime = False
  llm = 实时语音服务                   stt / tts / llm 三件套
  stt = tts = None                   inference_llm = None
  inference_llm = 独立文本 LLM  ◄── 关键:实时服务不实现
     (变量抽取/语音信箱用)              run_inference,带外推理要另开一个

见 run_pipeline.py:520-544。这个 inference_llm(旁路文本 LLM)是后面变量抽取、上下文压缩、语音信箱检测的共同"侧信道大脑"—— 记住它,4.7 节全靠它。

装完 services 后,把本次真正用的 provider/model 盖章进 initial_context.runtime_configuration 并落库,供事后分析用(run_pipeline.py:546-574)。

4.4 造引擎:PipecatEngine

把图、上下文变量、各种回调注入,造出状态机引擎(run_pipeline.py:677-692):

# run_pipeline.py:677  真实源码(节选)
engine = PipecatEngine(
    llm=llm,
    inference_llm=inference_llm,          # 旁路文本 LLM
    workflow=workflow_graph,              # 4.1 的快照图
    call_context_vars=merged_call_context_vars,
    node_transition_callback=node_transition_callback,  # 节点切换→观测
    has_recordings=has_recordings,
    context_compaction_enabled=context_compaction_enabled,
    ...
)

引擎内部如何把图变成 LLM 工具、如何驱动节点转移,是 03-pipecat-engine 的主题。这里只需知道:编排负责"造好并喂料",引擎负责"想"。

4.5 装管线 + 造 Worker

按 is_realtime 走 build_realtime_pipeline 或 build_pipeline(run_pipeline.py:882-906),再 create_pipeline_task 造出 PipelineWorker(run_pipeline.py:909)。轮次策略(realtime / 外部信号 / 智能轮次 / 转写)在此按 STT 和配置分流(run_pipeline.py:725-780)。管线内部结构见 02-voice-pipeline。

集成运行时会话在此 attach(task)(run_pipeline.py:911-917),然后把 task 和 transport 输出回填给引擎(run_pipeline.py:920-921)。

4.6 初始化引擎:系统提示 + 工具 + MCP

# run_pipeline.py:923  真实源码
# Initialize the engine to set the initial context with System Prompt and Tools
await engine.initialize()

engine.initialize() 在这里做两件对收尾至关重要的事:装好起始节点的系统提示与工具、打开本次通话的持久 MCP 会话(pipecat_engine.py:193 _open_mcp_sessions)。记住"MCP 会话在 initialize() 里打开",4.9 节的清理谜题就靠它。

4.7 带外能力在装配期接线(引擎旁路)

这是本章的另一个重点。 有一批能力不在主音频帧流里,而是在通话生命周期的关键点被"叫出来"跑一趟。它们大多在 _run_pipeline_impl 装配期接线、由引擎在节点转移或通话结束时触发。逐一说明。

能力	接线位置	触发时机	用哪个"大脑"
pre-call fetch	run_pipeline.py:581	通话开始前(并发预取)	无(HTTP)
变量抽取	引擎持有 manager	节点转移前 / 通话结束	inference_llm
上下文压缩	引擎持有 manager	节点转移后(后台)	inference_llm
知识库检索	按节点注册为 LLM 工具	模型主动调用	embeddings 服务
语音信箱检测	run_pipeline.py:823	长时无人语音时	独立 voicemail LLM
录音路由	run_pipeline.py:867	非实时且有录音	无(音频路由)

(a) Pre-call fetch:通话开始前预热上下文

起始节点若开了 pre_call_fetch_enabled + URL,就尽早并发发一个 HTTP POST 去外部系统(CRM/ERP)拉数据(run_pipeline.py:581-601),用 asyncio.create_task 让它与后续装配并行跑。

真正的请求在 execute_pre_call_fetch(pre_call_fetch.py:41):发标准化 payload(含 agent_id、主被叫号),从响应里抽 initial_context(兼容旧的 dynamic_variables 键),任何失败都返回 {} 绝不抛异常(pre_call_fetch.py:120-130)—— 外部系统挂了不能拖垮通话。

它的结果在哪合并?在 event_handlers.py 的 maybe_trigger_initial_response(event_handlers.py:99-162):管线和客户端都就绪后,若 fetch 还没回,就放一段振铃音等它,拿到结果 update 进 engine._call_context_vars 并落库,然后才设起始节点、开场白 —— 保证开口时上下文已完整。

(b) 变量抽取:把对话抽成结构化字段

VariableExtractionManager(pipecat_engine_variable_extractor.py:19)负责在合适时机把对话里的信息抽成变量。核心是"带外":

# pipecat_engine_variable_extractor.py:206  真实源码
# Use engine's LLM for out-of-band inference (no pipeline frames).
llm_response = await self._engine.inference_llm.run_inference(
    extraction_context, system_instruction=system_prompt
)

它先把对话历史(含格式化后的工具响应,跳过 transition 工具、超长截断)拼成文本(_build_conversation_history,:133),再让 inference_llm 跑一次独立推理,parse_llm_json 容错解析(:242)。因为不进管线帧流,它不会打断正在进行的语音。

触发时机由引擎控制 _perform_variable_extraction_if_needed(pipecat_engine.py:402):节点转移前(pipecat_engine.py:242,transition_func 里)和通话结束时(pipecat_engine.py:745)。默认后台 fire-and-forget。

(c) 上下文压缩:节点转移后压缩历史

只在 context_compaction_enabled 且非实时时启用(实时服务自己在服务端管会话状态,run_pipeline.py:668-675 明确关掉它)。ContextSummarizationManager(pipecat_engine_context_summarizer.py:22)在节点转移后由引擎调 start() 起一个后台任务(pipecat_engine.py:597-598),用 inference_llm._generate_summary 把旧消息(含前一节点遗留的孤儿工具调用)换成一条摘要,只保留"系统消息 + 摘要 + 最近若干条"(context_summarizer.py:140-161)。

关键细节:在应用时(而非请求时)重新快照消息,以保住摘要生成期间新加进来的消息(context_summarizer.py:135-138);超时 30s / 被新转移取消都安全降级(context_summarizer.py:166-171)。目标 token 4000(context_summarizer.py:32-36)。

(d) 知识库检索:注册成 LLM 工具

不同于上面几个"系统主动叫",知识库是模型主动调用的工具。引擎按当前节点的 document_uuids 把 retrieve_from_knowledge_base 注册成一个函数(pipecat_engine.py:349 _register_knowledge_base_function、:400 register_function、:534 按节点注册)。

实现 retrieve_from_knowledge_base(knowledge_base.py:20)做向量相似检索,并支持 full_document 模式直接返回全文(knowledge_base.py:237-256);全程带 OpenTelemetry span 供 Langfuse 观测(knowledge_base.py:57-180)。工具定义 get_knowledge_base_tool 会按是否限定文档动态改描述(knowledge_base.py:321)。

(e) 语音信箱检测 & (f) 录音路由

两者都只在非实时下有意义:

• VoicemailDetector(run_pipeline.py:823-862):开关开且非实时时,单开一个 LLM 子管线(不能和主管线共享,否则搞乱帧链接),检测到语音信箱就 engine.end_call_with_reason(VOICEMAIL_DETECTED, abort_immediately=True) —— 直接把这通挂掉。
• RecordingRouterProcessor(run_pipeline.py:864-879):有预录音时,在"播放预录音频"与"动态 TTS"之间路由;并后台 warm_recording_cache 预热,保证首次播放不卡。实时 LLM 直接产音频,故用不上。

4.8 注册观测器:全程留痕

引擎初始化后,编排挂上一排观测器,把通话过程实时落到内存 buffer,并在有 WS 时推给前端(run_pipeline.py:927-985):

观测器	作用	位置
RealtimeFeedbackObserver	实时反馈事件 → buffer + WS	run_pipeline.py:928-932
延迟观测器	测"用户→机器人"响应延迟	run_pipeline.py:935-960
turn log handlers	每轮用户/助手消息落 log	run_pipeline.py:962-965
node_transition_callback	节点切换事件 → buffer(+WS)	run_pipeline.py:609-640
InMemoryLogsBuffer	承接以上所有事件的内存缓冲	run_pipeline.py:604
register_event_handlers	transport/task 生命周期事件	run_pipeline.py:971-983

其中 node_transition_callback(send_node_transition)每次都更新 buffer 的"当前节点",于是后续所有事件都自动带上节点标签(run_pipeline.py:617-618)。

4.9 跑完一通 + finally 收尾(MCP 的 cancel-scope 谜题)

一切就绪,真正开跑:

# run_pipeline.py:987  真实源码
try:
    await run_pipeline_worker(task)     # 阻塞在此,直到通话结束
except asyncio.CancelledError:
    logger.warning("Received CancelledError in _run_pipeline")
finally:
    await engine.close_mcp_sessions()   # 注意:在这里关,不在 engine.cleanup()
    await feedback_observer.cleanup()

为什么 MCP 会话必须在这个 finally 关、而不是在 engine.cleanup() 里关? 注释给出了精确原因(run_pipeline.py:993-998):

MCPClient.start() 在 engine.initialize() 里打开的 anyio cancel scope 是 task-affine(和创建它的 task 绑定)。这个 finally 和 initialize() 跑在同一个 task;而 engine.cleanup() 是被 pipecat 的事件处理器(on_pipeline_finished)在另一个 task里调的。

anyio 的 cancel scope 有个硬规则:必须在打开它的同一个 task 里退出,跨 task 关闭会抛 "cancel scope in a different task" 错误。所以 MCP 会话的开(initialize,4.6)与关(这个 finally)被刻意安排在同一 task 内。这是本章最容易被忽略、也最能体现"编排要管 task 亲缘性"的细节。

---

5. WorkerRunner 生命周期

run_pipeline_worker 很薄(worker_runner.py:7-17),但它是通话真正"转起来"的地方:

# services/pipecat/worker_runner.py:7  真实源码
async def run_pipeline_worker(worker, *, handle_sigint=False, handle_sigterm=False, auto_end=True):
    runner = WorkerRunner(handle_sigint=handle_sigint, handle_sigterm=handle_sigterm)
    await runner.add_workers(worker)
    await runner.run(auto_end=auto_end)

它把 PipelineWorker 交给 pipecat v1.3 的 WorkerRunner 生命周期。默认不接管 SIGINT/SIGTERM(信号由上层 FastAPI 进程管)、auto_end=True(worker 结束即收工)。await runner.run(...) 一直阻塞到通话结束,这正是 4.9 里 await run_pipeline_worker(task) 停住的地方。

文件里还有个 wait_for_pipeline_worker_started(:20)供测试等待 worker 稳定启动,主流程不用。

---

6. 收尾:通话结束后发生了什么

通话结束由 pipecat 触发 on_pipeline_finished 事件,handler 在 event_handlers.py:219。它是"清洁工"的主体(注意:这跑在pipecat 事件处理器的 task里,所以 MCP 不在这关):

按顺序(event_handlers.py:219-408):

1. 停录音,汇总 gathered_context,补上 trace URL、call tags、disposition code。
2. 逐个集成会话 on_call_finished,收集集成日志。
3. engine.cleanup()(此处会 cleanup 上下文压缩 manager、语音信箱检测器等)。
4. 关智能轮次分析器的 WebSocket(若有)。
5. 汇总用量,update_workflow_run(is_completed=True, state=COMPLETED) 落库。
6. 发 PostHog CALL_COMPLETED,保存实时反馈日志。
7. 把混音/用户/机器人音频、转写各写到临时文件,enqueue_job(PROCESS_WORKFLOW_COMPLETION, ...) 把上传+集成+算成本交给后台任务。

与之配对的还有其它生命周期 handler:on_client_disconnected(用户挂断→USER_HANGUP 结束,:171)、on_pipeline_error(管线错→记录熔断+结束,:192)、on_pipeline_started / on_client_connected(两者都到齐才触发开场白,:186/:164)。

---

7. 边界与局限(诚实)

• agent-stream 裸音频分支未实现 —— 无 ?provider= 时直接 1011 关闭(agent_stream.py:50-56),这是明确的占位。
• 实时模式主动砍掉多项带外能力 —— 上下文压缩(run_pipeline.py:673-675)、语音信箱检测(:826-829)、录音路由(:867)在 is_realtime 下全部关闭,因为语音到语音服务自管会话状态、自出音频。若你依赖这些能力,实时模式下它们静默不生效。
• MCP 会话的 task 亲缘性是硬约束 —— 开/关必须同 task(§4.9)。任何把 close_mcp_sessions 挪进 engine.cleanup() 的"重构"都会踩 anyio cancel-scope 报错。
• pre-call fetch 失败静默降级 —— 返回 {} 不抛错(pre_call_fetch.py),超时 10s。外部系统挂了通话照常进行,但上下文会缺字段。
• 多 worker 下 WebRTC 状态本地化 —— SignalingManager 自维护连接映射而非依赖 pipecat 的 _pcs_map,以适配多 worker;跨 worker 的实时反馈靠 ws_sender_registry + Redis(见 API AGENTS.md 的 cross-worker 说明,属推断 (inferred))。

---

8. 横向对比

本章是把 01/02/03 串成端到端线的"胶水"。三者的分工:

章	管什么	与本章的接触面
01 workflow-model	图的数据模型与校验	§4.1 用快照 JSON 构 WorkflowGraph
02 voice-pipeline	帧在处理器间怎么流	§4.5 build_pipeline 装出来的东西
03 pipecat-engine	图→工具的状态机	§4.4 造引擎、§4.7 引擎触发带外能力
05 registries	供应商/工具注册表	§3.3 电话入口经注册表派发

一句话:本章是"控制流",02 是"数据流",03 是"决策流"。

---

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

主题	文件路径	符号名
编排主体	api/services/pipecat/run_pipeline.py:415	_run_pipeline_impl
电话包装(drain 登记)	api/services/pipecat/run_pipeline.py:165	run_pipeline_telephony
WebRTC 包装	api/services/pipecat/run_pipeline.py:292	run_pipeline_smallwebrtc
运行时快照定义	api/services/pipecat/run_pipeline.py:456	run_definition = workflow_run.definition
引擎初始化(系统提示+工具+MCP)	api/services/pipecat/run_pipeline.py:923	engine.initialize
finally 关 MCP(cancel-scope)	api/services/pipecat/run_pipeline.py:993	engine.close_mcp_sessions
Worker 生命周期	api/services/pipecat/worker_runner.py:7	run_pipeline_worker / WorkerRunner
通用入口	api/routes/agent_stream.py:31	agent_stream_websocket
配额校验	api/routes/agent_stream.py:98	authorize_workflow_run_start
WebRTC 信令 offer	api/routes/webrtc_signaling.py:396	_handle_offer
WS 反馈通道注册	api/routes/webrtc_signaling.py:495	register_ws_sender
变量抽取(带外)	api/services/workflow/pipecat_engine_variable_extractor.py:156	VariableExtractionManager._perform_extraction
上下文压缩(带外)	api/services/workflow/pipecat_engine_context_summarizer.py:58	ContextSummarizationManager._summarize_context_in_background
知识库检索工具	api/services/workflow/tools/knowledge_base.py:20	retrieve_from_knowledge_base
pre-call fetch	api/services/pipecat/pre_call_fetch.py:41	execute_pre_call_fetch
开场白触发 + fetch 合并	api/services/pipecat/event_handlers.py:99	maybe_trigger_initial_response
收尾落库 + 后处理	api/services/pipecat/event_handlers.py:219	on_pipeline_finished
引擎触发抽取	api/services/workflow/pipecat_engine.py:402	_perform_variable_extraction_if_needed
引擎触发压缩	api/services/workflow/pipecat_engine.py:597	_context_summarization_manager.start
引擎注册 KB 工具	api/services/workflow/pipecat_engine.py:349	_register_knowledge_base_function