数据截至 (上游 commit 689ca048bb0a)
一次通话的端到端编排与引擎旁路能力
30 秒导读: 前三章分别讲 了"图长什么样"(01)、"帧怎么在处理器间流"(02)、"图怎么变成状态机工具"(03)。这一章把它们串成一根线:一个 WebSocket / WebRTC / 电话连接进来,系统怎么一步步把它变成一通真实通话,跑完,再干净地收尾。核心是
run_pipeline.py里的_run_pipeline_impl—— 通话的"总装配车间"。
本章覆盖三件事:
- 入口:连接从哪几个口进来,进来后各自做什么(建 run、校配额、装 transport)。
- 编排主体:
_run_pipeline_impl如何用运行时快照装配一整套 services→引擎→管线→Worker,注册观测器,跑完,收尾。 - 旁路/带外能力:变量抽取、上下文压缩、知识库检索、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:549 |
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:245 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/{provider_name}/{workflow_uuid} 是一个"万能口":provider 名是 URL 路径段,provider 特有的通话元数据(主被叫号、凭证等)则从该 provider 自己的流协议里读,不需要在组织里预存 TelephonyConfigurationModel 行(agent_stream.py:1-11 的模块 docstring 点明了这层区别)。
它进来后按顺序做四件事(agent_stream.py:72-127):
- 建 workflow_run ——
db_client.create_workflow_run(...),把 provider、direction="inbound"存进initial_context(主被叫号等元数据由 provider 协议解析,agent_stream.py:72-92)。 - 设运行上下文 ——
set_current_run_id/set_current_org_id,让后续日志和 trace 带上 run/org(agent_stream.py:93-94)。 - 校配额 ——
authorize_workflow_run_start(...);has_quota为假就用错误信息关闭 WebSocket(agent_stream.py:96-113)。 - 派发 ——
provider_instance.handle_external_websocket(...),把这个 socket 交给注册表里对应 provider 的实现(agent_stream.py:116-127),后者最终会调run_pipeline_telephony。
注意边界: provider 名是必填路径段,未注册的 provider 直接以 1008 关闭(
agent_stream.py:48-53);OLD 版"无?provider=的裸音频分支"随路由改版一并消失。
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:515):
- 收到
offer后设 run/org 上下文,并先校配额authorize_workflow_run_start(webrtc_signaling.py:552-569)。 - 新建
SmallWebRTCConnection,用get_ice_servers(user_id=...)塞入按用户生成的时限 TURN 凭证(webrtc_signaling.py:669)。 - 注册 WS 反馈通道
register_ws_sender(workflow_run_id, ws_sender)—— 这就是后面观测器把实时反馈推给浏览器的那根管子(webrtc_signaling.py:683)。 - 后台起管线
asyncio.create_task(run_pipeline_smallwebrtc(...)),不阻塞信令(webrtc_signaling.py:708);随后把 answer 发回浏览器,ICE 候选另行 trickle。
还有一个 public/signaling/{session_token} 公开口(embed 嵌入用),多一层 token 校验 + 来源域校验 validate_origin,防止泄露的 token 从任意站点接入(webrtc_signaling.py:874-951)。
3.3 电话入口:经 telephony providers
七家运营商的 provider 各自处理完自己的信令握手后,统一调 run_pipeline_telephony(如 services/telephony/providers/twilio/provider.py:328、plivo/provider.py:350 等,共七家)。配额校验发生在更上游的运营商回调里;这里只负责把 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:268-270)。
三个包装(run_pipeline_telephony:165、run_pipeline_smallwebrtc:292、_run_pipeline:386)各自解析出 transport 后,都汇入下面的 _run_pipeline_impl。
4. 编排主体:_run_pipeline_impl 主流程
这是本章的心脏(run_pipeline.py:549)。它拿到 transport + workflow_run_id,把一通通话需要的一切从零装好。按代码顺序,它是这样一步步走的:
4.1 幂等闸门 + 运行时快照(最关键的设计)
先挡掉重复运行:workflow_run 已完成就抛 400(run_pipeline.py:592-593)。
然后取 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:650-653),图模型细节见 01-workflow-model。
4.2 解析配置 + 有效模型配置
从 run_configs(即快照里的 workflow_configurations)里抠出运行参数(run_pipeline.py:614-630):
| 参数 | 默认 | 作用 |
|---|---|---|
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:638-648)。入口层若已算过就直接复用(resolved_user_config),避免二次拉取。
4.3 判定 is_realtime,装配 services
is_realtime = user_config.is_realtime and user_config.realtime is not None(run_pipeline.py:670)—— 决定走"语音到语音"(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:673-697。这个 inference_llm(旁路文本 LLM)是后面变量抽取、上下文压缩、语音信箱检测的共同"侧信道大脑"—— 记住它,4.7 节全靠它。
装完 services 后,把本次真正用的 provider/model 盖章进 initial_context.runtime_configuration 并落库,供事后分析用(run_pipeline.py:713-741)。
4.4 造引擎:PipecatEngine
把图、上下文变量、各种回调注入,造出状态机引擎(run_pipeline.py:844-860):
# 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:1042-1066),再 create_pipeline_task 造出 PipelineWorker(run_pipeline.py:1075)。轮次策略(realtime / 外部信号 / 智能轮次 / 转写)在此按 STT 和配置分流(run_pipeline.py:893-938)。管线内部结构见 02-voice-pipeline。
集成运行时会话在此 attach(task)(run_pipeline.py:1077-1083),然后把 task 和 transport 输出回填给引擎(run_pipeline.py:1086-1087)。
4.6 初始化引擎:系统提示 + 工具 + MCP
# run_pipeline.py:1096 真实源码
# Initialize the engine to set the initial context with System Prompt and Tools
await engine.initialize()
engine.initialize() 在这里做两件对收尾至关重要的事:装好起始节点的系统提示与工具、打开本次通话的持久 MCP 会话(pipecat_engine.py:217 _open_mcp_sessions)。记住"MCP 会话在 initialize() 里打开",4.9 节的清理谜题就靠它。
4.7 带外能力在装配期接线(引擎旁路)
这是本章的另一个重点。 有一批能力不在主音频帧流里,而是在通话生命周期的关键点被"叫出来"跑一趟。它们大多在 _run_pipeline_impl 装配期接线、由引擎在节点转移或通话结束时触发。逐一说明。
| 能力 | 接线位置 | 触发时机 | 用哪个"大脑" |
|---|---|---|---|
| pre-call fetch | run_pipeline.py:743 | 通话开始前(并发预取) | 无(HTTP) |
| 变量抽取 | 引擎持有 manager | 节点转移前 / 通话结束 | inference_llm |
| 上下文压缩 | 引擎持有 manager | 节点转移后(后台) | inference_llm |
| 知识库检索 | 按节点注册为 LLM 工具 | 模型主动调用 | embeddings 服务 |
| 语音信箱检测 | run_pipeline.py:981 | 长时无人语音时 | 独立 voicemail LLM |
| 录音路由 | run_pipeline.py:1027 | 非实时且有录音 | 无(音频路由) |
(a) Pre-call fetch:通话开始前预热上下文
起始节点若开了 pre_call_fetch_enabled + URL,就尽早并发发一个 HTTP POST 去外部系统(CRM/ERP)拉数据(run_pipeline.py:743-768),用 asyncio.create_task 让它与后续装配并行跑。
真正的请求在 execute_pre_call_fetch(pre_call_fetch.py:53):发标准化 payload(含 agent_id、主被叫号),从响应里抽 initial_context(兼容旧的 dynamic_variables 键),任何失败都返回 {} 绝不抛异常(pre_call_fetch.py:171)—— 外部系统挂了不能拖垮通话。
它的结果在哪合并?在 event_handlers.py 的 maybe_trigger_initial_response(event_handlers.py:105-170):管线和客户端都就绪后,若 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:433):节点转移前(pipecat_engine.py:266,transition_func 里)和通话结束时(pipecat_engine.py:554)。默认后台 fire-and-forget。
(c) 上下文压缩:节点转移后压缩历史
只在 context_compaction_enabled 且非实时时启用(实时服务自己在服务端管会话状态,run_pipeline.py:835-842 明确关掉它)。ContextSummarizationManager(pipecat_engine_context_summarizer.py:22)在节点转移后由引擎调 start() 起一个后台任务(pipecat_engine.py:662-663),用 inference_llm._generate_summary 把旧消息(含前一节点遗留的孤儿工具调用)换成一条摘要,只保留"系统消息 + 摘要 + 最近若干条"(pipecat_engine_context_summarizer.py:140-161)。
关键细节:在应用时(而非请求时)重新快照消息,以保住摘要生成期间新加进来的消息(pipecat_engine_context_summarizer.py:135-138);超时 30s / 被新转移取消都安全降级(pipecat_engine_context_summarizer.py:166-171)。目标 token 4000(pipecat_engine_context_summarizer.py:32-36)。
(d) 知识库检索:注册成 LLM 工具
不同于上面几个"系统主动叫",知识库是模型主动调用的工具。引擎按当前节点的 document_uuids 把 retrieve_from_knowledge_base 注册成一个函数(pipecat_engine.py:380 _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:981-1022):开关开且非实时时,单开一个 LLM 子管线(不能和主管线共享,否则搞乱帧链接),检测到语音信箱就engine.end_call_with_reason(VOICEMAIL_DETECTED, abort_immediately=True)—— 直接把这通挂掉。 - RecordingRouterProcessor(
run_pipeline.py:1024-1039):有预录音时,在"播放预录音频"与"动态 TTS"之间路由;并后台warm_recording_cache预热,保证首次播放不卡。实时 LLM 直接产音频,故用不上。
4.8 注册观测器:全程留痕
引擎初始化后,编排挂上一排观测器,把通话过程实时落到内存 buffer,并在有 WS 时推给前端(run_pipeline.py:1086-1094):
| 观测器 | 作用 | 位置 |
|---|---|---|
RealtimeFeedbackObserver | 实时反馈事件 → buffer + WS | run_pipeline.py:1090-1094 |
| 延迟观测器 | 测"用户→机器人"响应延迟 | run_pipeline.py:1101-1126 |
| turn log handlers | 每轮用户/助手消息落 log | run_pipeline.py:1128-1133 |
node_transition_callback | 节点切换事件 → buffer(+WS) | run_pipeline.py:776-807 |
InMemoryLogsBuffer | 承接以上所有事件的内存缓冲 | run_pipeline.py:771 |
register_event_handlers | transport/task 生命周期事件 | run_pipeline.py:1139-1153 |
其中 node_transition_callback(send_node_transition)每次都更新 buffer 的"当前节点",于是后续所有事件都自动带上节点标签(run_pipeline.py:784-785)。
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:1163-1168):
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:245。它是"清洁工"的主体(注意:这跑在pipecat 事件处理器的 task里,所以 MCP 不在 这关):
按顺序(event_handlers.py:245-440):
- 停录音,汇总
gathered_context,补上 trace URL、call tags、disposition code。 - 逐个集成会话
on_call_finished,收集集成日志。 engine.cleanup()(此处会 cleanup 上下文压缩 manager、语音信箱检测器等)。- 关智能轮次分析器的 WebSocket(若有)。
- 汇总用量,
update_workflow_run(is_completed=True, state=COMPLETED)落库。 - 发 PostHog
CALL_COMPLETED,保存实时反馈日志。 - 把混音/用户/机器人音频、转写各写到临时文件,
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 —— provider 名是必填路径段,未注册的 provider 以 1008 关闭(
agent_stream.py:48-53);"裸音频"连接形态已随路由改版移除。 - 实时模式主动砍掉多项带外能力 —— 上下文压缩(
run_pipeline.py:840-842)、语音信箱检测(: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(见 APIAGENTS.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:549 | _run_pipeline_impl |
| 电话包装(drain 登记) | api/services/pipecat/run_pipeline.py:257 | run_pipeline_telephony |
| WebRTC 包装 | api/services/pipecat/run_pipeline.py:401 | run_pipeline_smallwebrtc |
| 运行时快照定义 | api/services/pipecat/run_pipeline.py:608 | run_definition = workflow_run.definition |
| 引擎初始化(系统提示+工具+MCP) | api/services/pipecat/run_pipeline.py:1096 | engine.initialize |
| finally 关 MCP(cancel-scope) | api/services/pipecat/run_pipeline.py:1163 | engine.close_mcp_sessions |
| Worker 生命周期 | api/services/pipecat/worker_runner.py:7 | run_pipeline_worker / WorkerRunner |
| 通用入口 | api/routes/agent_stream.py:33 | agent_stream_websocket |
| 配额校验 | api/routes/agent_stream.py:96 | authorize_workflow_run_start |
| WebRTC 信令 offer | api/routes/webrtc_signaling.py:515 | _handle_offer |
| WS 反馈通道注册 | api/routes/webrtc_signaling.py:687 | 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:53 | execute_pre_call_fetch |
| 开场白触发 + fetch 合并 | api/services/pipecat/event_handlers.py:105 | maybe_trigger_initial_response |
| 收尾落库 + 后处理 | api/services/pipecat/event_handlers.py:245 | on_pipeline_finished |
| 引擎触发抽取 | api/services/workflow/pipecat_engine.py:433 | _perform_variable_extraction_if_needed |
| 引擎触发压缩 | api/services/workflow/pipecat_engine.py:662 | _context_summarization_manager.start |
| 引擎注册 KB 工具 | api/services/workflow/pipecat_engine.py:380 | _register_knowledge_base_function |