数据截至 (上游 commit 689ca048bb0a)
可扩展性接缝:注册表驱动的供应商与工具
30 秒导读: 一个平台 要活得久,难点不在「今天支持 Twilio」,而在「明天要接 Plivo、后天要接一个客户自建的 Asterisk,而这不能把核心代码改成一堆
if provider == "twilio"」。Dograh 的做法是把三处天然会变的东西——电话供应商、对话图的节点类型、LLM 能调的工具——都做成注册表插槽:每样新东西只写自己的文件夹、加一行 import 就接入,核心的编排/管线/路由代码永远只对着一个抽象基类和一张注册表说话。本章讲清这套「接缝」怎么设计、为什么这么设计。
本章是 Dograh 系列的收尾章。前面几章讲的是「系统怎么跑」: 对话即图是数据模型,实时语音管线是帧的流动, PipecatEngine是把图变成工具调用的状态机, 一次通话的编排是端到端串起来。本章讲的是「系统怎么长大」—— 新供应商、新节点、新工具从哪个缝里塞进去,而不用动上面这些核心。
1. 这是什么(零基础也能懂)
先说要解决的痛
假设你在做一个语音 AI 平台。第一版只接了 Twilio 打电话。很快你会遇到:
- 有客户在印度,要用 Plivo;有客户要 Telnyx 的呼叫控制;有客户干脆自建 Asterisk。
- 每家供应商的凭证字段不一样(Twilio 是
account_sid+auth_token,Vonage 是 JWT), 音频采样率不一样(Twilio 8kHz,Vonage 16kHz),回话格式不一样(TwiML vs NCCO JSON), 连配置表单都得为每家单独画。
最容易写坏的写法是让核心到处长出分支:
# 反面教材,非源码:每加一家供应商,这些 if 都要改一遍
def create_provider(name, config):
if name == "twilio":
return TwilioProvider(...)
elif name == "plivo":
return PlivoProvider(...)
elif name == "vonage": # 又要来改这里
return VonageProvider(...)
这种代码的病根:一个变化点(新供应商)散落在 N 个文件里(工厂、音频配置、schema、路由、前端表单), 加一家漏改一处就出 bug。
「接缝」的思路:让核心只认抽象和注册表
Dograh 的解法可以一句话概括:
把「会变的东西」收进一个不可变的描述对象(spec),丢进一张全局注册表;核心代码只查注册表、 只调抽象基类,永远不认识任何一家具体供应商的名字。
加一家新供应商 = 在 providers/<名字>/ 下写自己那份 spec + 实现类,并在一个 import 列表里加一行。
核心的工厂、管线、路由一个字都不用改——它们遍历注册表就自动看见了新成员。
三处接缝
Dograh 里用了同一套模式的地方有三个,面向三类「想加东西」的读者:
| 你想加的东西 | 接缝在哪 | 面向谁 |
|---|---|---|
| 新电话供应商(接个新运营商) | services/telephony/ provider 注册表 | 要接 Twilio/Plivo 之外的运营商 |
| 新对话节点类型(图里的新积木) | services/workflow/node_specs/ + services/integrations/ | 要给工作流加自定义节点 |
| 新工具(LLM 能调用的能力) | services/workflow/tools/ + MCP 会话 | 要给 agent 加计算器/知识库/外部 API/MCP |
三处长得几乎一样,学会一处就懂三处。本章以最完整的电话供应商为主线讲透模式, 再用节点和工具说明「同一套模式在不同场景怎么变形」。
2. 顶层全景(这套接缝大概怎么转)
一张图:注册表插槽的通用形状
三处接缝都是这个形状。以电话供应商为例,把抽象名字换成节点/工具也成立:
┌─────────────────────────────────────────────┐
加东西的人 ──▶ │ providers/<名字>/ (只碰这个文件夹) │
│ __init__.py ── 造一个 ProviderSpec 并 │
│ register(SPEC) ①自注册 │
│ provider.py ── 实现抽象基类的方法 │
│ config.py ── Pydantic 请求/响应模型 │
│ transport.py ── 造 pipecat transport │
└───────────────────┬─────────────────────────┘
│ import 触发 register()
▼
┌─────────────────────────────────────────────┐
│ registry._REGISTRY : {name → ProviderSpec} │ ②全局注册表
└───────────────────┬─────────────────────────┘
│ get(name) / all_specs()
┌────────────────────────────┼────────────────────────────┐
▼ ▼ ▼
factory.py run_pipeline.py routes/telephony.py
查 provider_cls 查 transport_factory 遍历 all_specs()
造实例 起 transport 按名字挂载路由
└──────────── 核心代码:只认抽象基类 + 注册表,不认名字 ──────────┘
│
▼ ③从 spec 生成
UI 表单 (ProviderUIField) · 校验 schema · 掩码规则 · 音频配置
怎么读这张图: 上半是「加东西的人」的活,全在自己文件夹里;中间是那张注册表;
下半是核心代码——它们只通过注册表接口(get/all_specs) 拿到成员,从不写供应商名字。
最下面一行是「白拿的红利」:UI、校验、掩码、音频参数都从同一个 spec 派生,不用另写。
三个不变量(整套模式的骨架)
| 编号 | 名字 | 是什么 | 电话供应商里的体现 |
|---|---|---|---|
| ① | 自注册 | 成员在 import 时把自己登记进注册表 | 每个 providers/<名字>/__init__.py 调 register(SPEC) |
| ② | 全局注册表 | 一个 {名字 → spec} 字典 + 查询函数 | registry._REGISTRY、get()、all_specs() |
| ③ | 从 spec 生成 | UI/schema/校验/参数都从 spec 派生,不重复写 | ui_metadata 生成表单、config_request_cls 校验、transport_sample_rate 定音频 |
记住这三点,后面每一节都是它们的具体化。
3. 接缝一:电话供应商(最完整的样板)
这节讲最全的一处。看懂它,节点和工具就是「同一套模式的简化版」。
3.1 抽象基类:核心眼里「一家供应商」长什么样
核心代码不认识 Twilio,只认识一个抽象类 TelephonyProvider——它规定了「任何一家供应商必须能做的事」。
真实实现见 api/services/telephony/base.py:118 的 TelephonyProvider(ABC)。它用 @abstractmethod
钉死了一组必须实现的方法(下面挑几个有代表性的):
| 抽象方法 | 干什么 | 为什么必须抽象 |
|---|---|---|
initiate_call | 发起一通外呼 | 各家 REST API 完全不同 |
parse_inbound_webhook | 把入站 webhook 解析成标准结构 | 各家 payload 字段名不同 |
verify_inbound_signature | 验签保安全 | 各家 签名方案不同(HMAC/JWT/body 签名) |
start_inbound_stream | 接起入站呼叫、起媒体流 | 有的回 TwiML,有的发 REST(见下) |
can_handle_webhook(classmethod) | 「这条 webhook 是我的吗?」 | 入站分发时用来认领 |
关键设计:标准化 DTO(数据传输对象)。 各家 API 五花八门,但核心不想处理这种差异, 于是基类定义了一组「归一化」的 dataclass,让所有供应商的输出都长成同一个样子:
CallInitiationResult(base.py:16)——外呼结果,统一成call_id/status/caller_number…NormalizedInboundData(base.py:44)——入站数据,统一成from_number/to_number/account_id…ProviderSyncResult(base.py:31)——把「DB 写成功但供应商 API 拒绝了」表达成一个非致命警告 (ok=False, message=...),而不是抛异常炸掉流程。
有了这些 DTO,核心的编排代码(见 04)处理的永远是标准形状, 「哪家供应商」的差异被挡在了基类实现里。
一个体现「抽象要贴合现实」的细节:
start_inbound_stream(base.py:321)的 docstring 明确区分两类供应商—— 标记响应型(Twilio/Plivo,直接返回 TwiML/XML)和呼叫控制型(Telnyx,发 REST 调用去接起并起流)。 抽象方法的返回值被设计成「可以是 Response 对象,也可以是 JSON」,正是为了同时容纳这两种截然不同的交互。
3.2 ProviderSpec:把「一家供应商」打包成一个不可变描述
光有实现类还不够。核心还需要知道「这家的采样率多少、凭证怎么归一化、表单长啥样」。
这些元信息被收进一个冻结的 dataclass ProviderSpec(api/services/telephony/registry.py:247)。
它是整套模式的核心数据结构。字段一览:
| 字段 | 类型 | 作用 |
|---|---|---|
name | str | 注册表的键,也是存进 DB 的鉴别符(discriminator) |
provider_cls | Type[TelephonyProvider] | 那个实现类,工厂用它造实例 |
config_loader | ConfigLoader | 把 DB 里的原始凭证 dict 归一化成构造器要的形状 |
transport_factory | TransportFactory | 为接受的 WebSocket 造 pipecat transport 的异步函数 |
transport_sample_rate | int | 线路音频采样率(Twilio 8000,Vonage 16000);pipecat 据此推出整份 AudioConfig |
config_request_cls / config_response_cls | Type[BaseModel] | 存/取配置的 Pydantic 模型(校验 + 掩码响应) |
ui_metadata | Optional[ProviderUIMetadata] | 驱动前端配置表单(见 3.5) |
account_id_credential_field | str | 入站 webhook 匹配到哪个 org 配置的凭证字段;"" 表示该供应商没有账号概念(如 ARI) |
preprocess_credentials_on_save | Optional[CredentialsPreprocessor] | 存盘前对凭证做 I/O 改写的可选钩子 |
@dataclass(frozen=True)(registry.py:76)——注册后不可变,谁也别想在运行时偷偷改它。
一个刻意的「不放进 spec」决定: spec 不带路由。
ProviderSpec的 docstring(registry.py:99) 解释:路由(webhook、状态回调、应答 URL)住在providers/<名字>/routes.py,靠importlib按需加载—— 因为路由处理器往往牵连很深(campaign、db 代码),不能让「有人 import 了一个 TelephonyProvider 类型」 就把整条依赖链拖进来。这是「spec 里放什么、不放什么」的一次精心权衡,3.6 会看到它的另一半。
3.3 注册表:一张字典 + 几个查询函数
注册表本体朴素得几乎不像「架构」——就是一个模块级字典和几个函数(registry.py:126 起):
# 真实源码骨架,api/services/telephony/registry.py:126
_REGISTRY: Dict[str, ProviderSpec] = {}
def register(spec: ProviderSpec) -> None: ... # 登记一家,registry.py:129
def get(name: str) -> ProviderSpec: ... # 按名查,查不到抛错,registry.py:140
def get_optional(name) -> Optional[...]: ... # 按名查,查不到返回 None,registry.py:148
def all_specs() -> List[ProviderSpec]: ... # 全部,按名排序稳定迭代,registry.py:153
两处值得看的细节:
- 重复注册是防呆的。
register(registry.py:129)里,如果同名 spec 是同一个实例再登记, 静默放过(import 可能被触发多次);如果是不同实例同名,直接raise ValueError——这才是真 bug (两家抢一个名字)。 - 迭代是稳定的。
all_specs()(registry.py:153)按名字排序返回,保证核心遍历供应商的顺序确定, 不受 import 顺序影响。
3.4 自注册 + 一行 import:成员怎么进注册表
现在把「加一家供应商」的动作走一遍。以 Twilio 为例,它的 providers/twilio/__init__.py 做三件事:
- 造 spec——
api/services/telephony/providers/twilio/__init__.py:64处SPEC = ProviderSpec(name="twilio", ...)。 - 登记自己——
twilio/__init__.py:76处一行register(SPEC)。 - 导出——
__all__里带上 SPEC 和实现类。
那么 register() 是什么时候被调用的?答案是 import 时的副作用。看
api/services/telephony/providers/__init__.py:9:
# api/services/telephony/providers/__init__.py:9
from api.services.telephony.providers import ( # noqa: F401 — import 为了副作用(注册)
ari, cloudonix, plivo, telnyx, twilio, vobiz, vonage,
)
这就是「加一家 只改一行」的那一行。 import 这些包 → 触发每个包的 __init__.py 执行 →
register(SPEC) 被调 → 注册表里就有了这家。等到工厂、音频配置、路由去查注册表时,
包早已 import 完、登记完了。
对照两家 spec 看差异有多小。 Twilio 和 ARI(Asterisk)的 __init__.py 几乎一模一样,
只是字段值不同:
Twilio (twilio/__init__.py:64) | ARI (ari/__init__.py:215) | |
|---|---|---|
config_loader 归一化的字段 | account_sid/auth_token/amd_enabled | ari_endpoint/app_name/app_password |
account_id_credential_field | "account_sid" | 缺省 ""(ARI 没账号概念) |
transport_sample_rate | 8000 | 8000 |
写一家新供应商,本质就是填一张这样的表。
3.5 从 spec 生成 UI:一家新供应商不用写前端
这是「从 spec 生成」最亮的一处。看 Twilio 的 _config_loader 上面那段
_UI_METADATA = ProviderUIMetadata(...)(twilio/__init__.py:27)——它是一串 ProviderUIField:
# 真实源码节选,api/services/telephony/providers/twilio/__init__.py:31
ProviderUIField(
name="account_sid", # 必须和 Pydantic 字段名一致
label="Account SID",
type="text",
sensitive=True, # 存储值展示时打码
description="Twilio Account SID (starts with AC)",
),
ProviderUIField(registry.py:34)和 ProviderUIMetadata(registry.py:52)描述了「一个表单字段长啥样」:
名字、标签、控件类型(text/password/textarea/string-array/number/boolean)、是否必填、是否敏感。
红利有两层:
- 前端表单自动生成。 前端拉一个
GET .../telephony-providers/metadata接口,拿到所有供应商的ui_metadata,通用地渲染成表单。加一家供应商,前端一行不改。 - 掩码规则复用同一处。 哪些字段读取时要打码,不是另写一张清单,而是直接看
ui_metadata里sensitive=True的字段(ProviderUIField.sensitive,registry.py:46)。同一份声明,喂两个用途。
这就是「一处声明,处处派生」——spec 是唯一真源,UI、掩码、校验都从它长出来。
3.6 核心怎么用注册表(而不是 if/elif)
回到核心侧,看它如何只查注册表。三个地方:
工厂造实例。 api/services/telephony/factory.py:353 的 _instantiate:
# api/services/telephony/factory.py:218
def _instantiate(config: Dict[str, Any]) -> TelephonyProvider:
spec = registry.get(config["provider"]) # 查注册表,不写名字
return spec.provider_cls(config) # 用 spec 里的类造实例
而 _normalize_with_phone_numbers(factory.py:203)则用 spec.config_loader(raw) 把 DB 凭证归一化——
旧代码里那条 if/elif 链,被换成了「查 spec 拿 config_loader」(见 ProviderSpec.config_loader 的 docstring,registry.py:83)。
管线起 transport。 api/services/pipecat/run_pipeline.py:364:
# api/services/pipecat/run_pipeline.py:260
spec = telephony_registry.get(provider_name)
audio_config = create_audio_config(provider_name) # 内部也读 transport_sample_rate
transport = await spec.transport_factory(websocket, workflow_run_id, audio_config, ...)
路由按需挂载。 api/routes/telephony.py:1286 的 _mount_provider_routers 遍历 all_specs(),
用 importlib.import_module(f"...providers.{spec.name}.routes") 尝试加载每家的路由,把
ModuleNotFoundError 当成「这家没有路由」(如 ARI 只有 WebSocket):
# api/routes/telephony.py:1063
for spec in _telephony_registry.all_specs():
try:
module = importlib.import_module(f"api.services.telephony.providers.{spec.name}.routes")
except ModuleNotFoundError:
continue # 这家没路由,跳过
router.include_router(module.router)
这正是 3.2 提到的「spec 不带路由」的另一半:注册表给出名字,importlib 按名字懒加载路由模块, 既保持了「加一家只碰自己文件夹」,又避免把沉重的路由依赖链在 import provider 类型时就拖进来。
主线走一遍(外呼): 编排层拿到 provider_name → registry.get(name) 拿 spec →
spec.provider_cls(config) 造实例发起呼叫 → WebSocket 接上后 spec.transport_factory(...) 起管线 →
全程核心没有一个 if name == ...。
4. 接缝二:节点类型(同一套模式,换个场景)
对话工作流是一张图(见 01),图里每种节点都有一份 NodeSpec—— 描述这个节点有哪些属性、怎么渲染、给 LLM 看的说明文字。加一种自定义节点,用的还是「注册表 + 从 spec 生成」。
两级来源:核心节点自动生成,集成节 点自注册
节点注册表在 api/services/workflow/node_specs/__init__.py:27(REGISTRY: dict[str, NodeSpec])。
它的取数逻辑分两支(get_spec,node_specs/__init__.py:43):
# api/services/workflow/node_specs/__init__.py:40
def get_spec(name: str) -> NodeSpec | None:
_ensure_core_registered() # ① 核心节点:从 DTO 模型生成
if name in REGISTRY:
return REGISTRY[name]
from api.services.integrations import get_node_spec
return get_node_spec(name) # ② 集成节点:去集成注册表找
- 核心节点(
_ensure_core_registered,node_specs/__init__.py:84)——遍历_CORE_NODE_DATA_CLASSES,对每个 DTO 模型调build_spec(model_cls)自动生成 NodeSpec。 也就是说,核心节点的 spec 不是手写的,是从数据模型 + 挂在模型上的元数据派生的 (build_spec,api/services/workflow/node_specs/model_spec.py:124)。 - 第三方集成节点——住在
api/services/integrations/<名字>/,通过集成注册表登记,get_spec在核心里找不到时兜底去那儿找。加集成节点,不用改node_specs/__init__.py(见文件顶部 docstring,node_specs/__init__.py:1)。
all_specs()(node_specs/__init__.py:53)把两支合并、按名排序返回——和电话供应商的 all_specs() 神似。
集成包:比 provider 更大的插槽
集成注册表在 api/services/integrations/registry.py。它的 spec 叫 IntegrationPackageSpec
(api/services/integrations/base.py:63,冻结 dataclass),比 provider 管得更宽——一个包可以同时带节点、路由、运行时会话、通话结束后的收尾处理:
IntegrationPackageSpec 字段 | 作用 |
|---|---|
nodes | 一串 IntegrationNodeRegistration(每项:类型名 + 数据模型 + node_spec + 敏感字段) |
routers | 这个集成挂的 FastAPI 路由 |
create_runtime_sessions | 通话运行时要起的会话(可选) |
run_completion | 通话结束后跑的收尾处理器(可选) |
自注册的写法和 provider 一模一样。看 tuner 集成的 __init__.py:10:
# api/services/integrations/tuner/__init__.py:10
PACKAGE = register_package(
IntegrationPackageSpec(
name="tuner",
nodes=(NODE,), # NODE 见 tuner/node.py:131
create_runtime_sessions=create_runtime_sessions,
run_completion=run_completion,
)
)
其中那个 NODE(tuner/node.py:217,IntegrationNodeRegistration)里的 node_spec=SPEC,而
SPEC = build_spec(TunerNodeData)(tuner/node.py:214)——集成节点也用和核心节点一样的 build_spec
从数据模型生成 spec,只是登记走的是集成注册表。
自注册的触发:这次是 pkgutil 扫目录
电话供应商靠「手写一行 import」触发注册;集成这边更自动——ensure_integrations_loaded
(api/services/integrations/loader.py:10)用 pkgutil.iter_modules 遍历 integrations/ 目录,
跳过内部模块(base/loader/registry),把其余子包逐个 importlib.import_module,触发它们的
register_package。所以加一个集成包,连那行 import 都省了——放进目录即可。
NodeSpec 与 provider spec 的一处不同:
NodeSpec不是冻结 dataclass,而是 PydanticBaseModel(node_specs/_base.py:289,model_config = ConfigDict(extra="forbid"))。因为 NodeSpec 是要序列化 发给前端、MCP 工具、SDK 的线上契约(见_base.py:1docstring),用 Pydantic 更合适。 模式相同(spec + registry + 从模型生成),载体按用途选——这正是「同一套模式在不同场景变形」。
5. 接缝三:工具系统(LLM 能调的能力)
第三处接缝是工具——LLM 在对话里能调用的函数:算个数、查知识库、打个外部 API、连一台 MCP 服务器。 这里的「注册」略有不同:工具不是启动时登进全局字典,而是一通电话开始时,按这通电话用到的节点动态装配。 但「从 spec 生成 schema」的内核完全一致。
5.1 工具的四种来源
api/services/workflow/tools/ 下每个文件是一类工具:
| 文件 | 工具 | 形态 |
|---|---|---|
calculator.py | 安全算术 | 内置,固定 schema(get_calculator_tools,calculator.py:31) |
timezone.py | 时区/时间转换 | 内置,固定 schema(get_time_tools,timezone.py:149) |
knowledge_base.py | 知识库检索 | 内置 |
custom_tool.py | 用户自定义 HTTP API | 从 DB 里用户配置生成 schema |
mcp_tool.py | MCP 工具 | 连外部 MCP 服务器,schema 由服务器返回 |
前三种是「写死的能力」,后两种才是真正的可扩展插槽:用户在界面上配一个 HTTP 工具或一台 MCP 服务器, 就等于给 agent 加了个新能力,不改一行后端代码。
5.2 共同货币:FunctionSchema
不管工具从哪来,最终都要变成 LLM 能理解的函数 schema。统一的转换点是
get_function_schema(api/services/workflow/pipecat_engine_custom_tools.py:71):
# api/services/workflow/pipecat_engine_custom_tools.py:41
def get_function_schema(function_name, description, *, properties=None, required=None):
return FunctionSchema(name=function_name, description=description,
properties=properties or {}, required=required or [])
FunctionSchema 就是工具系统的「标准 DTO」,等价于电话那边的 NormalizedInboundData——
把千差万别的来源归一成一个形状,后面 pipecat 再把它转成各家 LLM(OpenAI/Gemini)的具体格式。
自定义 HTTP 工具怎么变成 schema?看 custom_tool.py:60 的 tool_to_function_schema——
它读用户存在 DB 里的 tool.definition,把每个 parameter 的类型(经 TYPE_MAP,custom_tool.py:30)
映射成 JSON schema 的 properties/required。用户配的参数表,就是工具的 spec,schema 从它生成。
5.3 CustomToolManager:一通电话的工具装配台
真正把「这通电话用哪些工具」装配起来的是 CustomToolManager
(pipecat_engine_custom_tools.py:94)。两个主方法,分工清晰:
get_tool_schemas(:125)——给 LLM 看的:按 tool_uuid 从 DB 取工具,逐个转成FunctionSchema。register_handlers(:205)——给 LLM 调的:为每个工具在engine.llm上注册一个执行处理器。
它内部按类别分派(用 ToolCategory 枚举),这本身就是个小注册表模式:
tool.category == CALCULATOR → 内置 schema + calculate_func 处理器 (:151, :308)
tool.category == MCP → 找到该工具的 live MCP 会话,取它的 schemas (:165, :237)
其它(HTTP/END_CALL/TRANSFER) → tool_to_function_schema + 对应处理器 (:181, :282)
注意 MCP 那支:schema 不是本地造的,而是问活着的 MCP 会话要
(session.function_schemas(allowed),:178)——工具的能力清单来自远端服务器。
5.4 MCP 会话:持久连接 + 优雅降级
最能体现「可扩展性要为失败设计」的是 McpToolSession
(api/services/workflow/mcp_tool_session.py:49)。它是「一通电话期间对一台 MCP 服务器的活连接」。
装配时机: PipecatEngine 在初始化时调 _open_mcp_sessions
(api/services/workflow/pipecat_engine.py:1060),把这通电话所有节点引用到的 MCP 工具连上,
存进 self._mcp_sessions(pipecat_engine.py:150)。连上后 McpToolSession.start()
(mcp_tool_session.py:81)拉取工具列表、缓存成 FunctionSchema。
命名空间防撞: 多台 MCP 服务器可能都有个 echo 工具。会话用 namespace_function_name
(api/services/workflow/tools/mcp_tool.py:64)把它们改名成 mcp__<slug>__echo,slug 从 Dograh 里
的工具名派生——避免 LLM 看到两个同名函数。
核心亮点——降级而非崩溃: MCP 服务器可能是死的、连不上的。设计要求是
「一通电话必须能在 MCP 服务器挂掉时活下来」。start()(mcp_tool_session.py:81)在连接失败时
不抛异常,而是把会话标记 available = False(_degrade,:146),清空 schema,这通电话就当没有这些工具继续跑。
start() 里有一段极其克制、注释极长的异常处理(mcp_tool_session.py:120-144),值得一提:
经验上,一台连不上的 MCP 服务器不会以普通
Exception冒出来。真正的失败是httpx.ConnectError, 但 anyio 的 task group 在拆除时,会把它重新包装成一个内部的CancelledError,携带签名消息"Cancelled via cancel scope <id>"。而真正的外部取消(通话结束/关机)是一个消息不同的CancelledError。 两者类型、MRO、上下文链都一样,唯一能区分的就是这条 anyio 的签名消息。于是代码只在消息以"Cancelled via cancel scope"开头时降级,否则重新抛出以保住结构化并发的正确性。
这段是「优雅降级」的教科书:该活下去的失败(服务器连不上)吞掉降级,不该吞的信号(真取消)老实放行。
调用时(_create_mcp_handler,pipecat_engine_custom_tools.py:470)同样把任何异常兜成
结构化的错误文本回给 LLM,让 agent 用嘴巴挽回(「抱歉这个功能暂时不可用」),而不是让整通电话崩掉。
连 call_timeout_secs(mcp_tool_session.py:171)都特意设得比传输读超时长 5 秒,让慢调用表现为一个可处理的工具错误,
而不是一次硬性的管线超时。
6. 提炼:同一套模式,三处变形
把三处接缝并排看,骨架完全一致,只在「载体」和「注册触发方式」上按场景微调:
| 维度 | 电话供应商 | 节点类型 | 工具 |
|---|---|---|---|
| spec 对象 | ProviderSpec(冻结 dataclass) | NodeSpec(Pydantic 模型)/ IntegrationPackageSpec(冻结 dataclass) | FunctionSchema + DB 里的 tool.definition |
| 注册表 | telephony/registry._REGISTRY | node_specs.REGISTRY + integrations 包注册表 | CustomToolManager 按类别 + engine._mcp_sessions |
| 注册触发 | providers/__init__.py 手写一行 import | pkgutil 扫目录自动 import | 按这通电话的节点动态装配 |
| 从 spec 生成什么 | UI 表单 / 校验 / 掩码 / 音频参数 | 前端渲染 / SDK / MCP 契约 | LLM function schema |
| 查询接口 | get() / all_specs() | get_spec() / all_specs() | get_tool_schemas() / register_handlers() |
| 为失败设计 | ProviderSyncResult(ok=False) 非致命警告 | spec lint 校验描述/示例非空 | MCP 会话 available=False 降级 |
一句话的共同模式:
**不可变的 spec(把「会变的东西」声明成数据)+ 一张全局注册表(自注册进来,按名/全量查出去)
- 从 spec 派生一切(UI、schema、校验、参数都不重复写)。** 核心代码只对着抽象和注册表编程, 永远不认识 任何一个具体成员的名字——于是「加东西」变成「填一张表 + 放进目录」,而不是「到处改 if」。
7. 边界与局限(诚实)
- 「只改一行/一个文件夹」有前提。 电话供应商加一家,除了自己文件夹,仍要手动在
providers/__init__.py加一行 import(节点集成靠pkgutil扫目录省了这行,但 provider 没有)。 代码里也可见:provider 的config_request_cls/config_response_cls要接进api/schemas/telephony_config.py的鉴别联合(discriminated union)——这一步是本章读的文件之外的, 属于「加供应商」的完整流程,本章未展开。 - 抽象基类是「大而全」的。
TelephonyProvider(base.py:68)有十几个@abstractmethod, 一家新供应商即便不需要转接、AMD,也得把这些方法实现掉(哪怕抛NotImplementedError)。 基类用了一些非抽象的默认实现(如configure_inbound默认 no-op,base.py:374;supports_answering_machine_detection默认False,base.py:204)来缓解,但接口面依然偏重。 - MCP 降级的区分逻辑依赖字符串匹配。 6 里那段靠
"Cancelled via cancel scope"消息前缀区分 「连接失败」与「真取消」(mcp_tool_session.py:137),注释自己也承认这是唯一可靠的判别器—— 一旦上游 anyio/pipecat 改了这条消息文案,判别就会失效。代码留了个except Exception兜底(:140)防未来变化, 但这终究是与上游实现细节耦合的脆弱点。 - 工具「注册」不是全局静态的。 和电话/节点不同,工具是每通电话动态装配的
(
_open_mcp_sessions在引擎初始化时跑,pipecat_engine.py:218)。好处是隔离、可按节点过滤; 代价是没有一张「系统支持哪些工具」的全局静态表可查,能力清单散在 DB 配置和 live 会话里。
8. 代码地图(导航索引)
用符号名 grep 比行号抗漂移。以下均相对克隆根 api/。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 供应商 spec 定义(冻结 dataclass) | services/telephony/registry.py | ProviderSpec |
| 表单字段 / 表单元数据 | services/telephony/registry.py | ProviderUIField · ProviderUIMetadata |
| 注册 / 查询接口 | services/telephony/registry.py | register · get · get_optional · all_specs |
| 供应商抽象基类 | services/telephony/base.py | TelephonyProvider |
| 标准化 DTO | services/telephony/base.py | CallInitiationResult · NormalizedInboundData · ProviderSyncResult |
| 自注册样板(最全) | services/telephony/providers/twilio/__init__.py | SPEC · _config_loader · _UI_METADATA |
| 自注册样板(最简) | services/telephony/providers/ari/__init__.py | SPEC |
| 一行 import 触发注册 | services/telephony/providers/__init__.py | (import for side effects) |
| 核心按注册表造实例 | services/telephony/factory.py | _instantiate · _normalize_with_phone_numbers |
| 管线按 spec 起 transport | services/pipecat/run_pipeline.py | spec.transport_factory(:260) |
| 路由按 importlib 懒加载 | routes/telephony.py | _mount_provider_routers |
| 节点 spec 注册表 | services/workflow/node_specs/__init__.py | REGISTRY · get_spec · all_specs · _ensure_core_registered |
| 从模型生成 node spec | services/workflow/node_specs/model_spec.py | build_spec |
| node spec 线上契约 | services/workflow/node_specs/_base.py | NodeSpec · PropertyType |
| 集成包 spec | services/integrations/base.py | IntegrationPackageSpec · IntegrationNodeRegistration |
| 集成注册表 | services/integrations/registry.py | register_package · get_node_spec · all_node_specs |
| 集成自动加载(扫目录) | services/integrations/loader.py | ensure_integrations_loaded |
| 集成样板 | services/integrations/tuner/__init__.py | PACKAGE |
| 工具 → function schema | services/workflow/tools/custom_tool.py | tool_to_function_schema · execute_http_tool |
| 内置工具 schema | services/workflow/tools/calculator.py · tools/timezone.py | get_calculator_tools · get_time_tools |
| MCP 定义校验 / 命名空间 | services/workflow/tools/mcp_tool.py | validate_mcp_definition · namespace_function_name |
| 工具装配台 | services/workflow/pipecat_engine_custom_tools.py | CustomToolManager · get_function_schema |
| MCP 持久会话 + 降级 | services/workflow/mcp_tool_session.py | McpToolSession · start · _degrade |
| 引擎侧开 MCP 会话 | services/workflow/pipecat_engine.py | _open_mcp_sessions · _mcp_sessions |
相关章节: 节点数据模型见 01 对话即图;transport 如何在管线里流帧见 02 实时语音管线;工具如何被引擎当作图的边来调用见 03 PipecatEngine;供应商实例在一次通话里的编排见 04 一次通话的编排。