Dograh — 架构与原理
30 秒导读: Dograh 是一个开源、可自托管的语音 AI 平台(对标 Vapi / Retell)。 你在网页上用拖拽画一张对话流程图,Dograh 就把它变成一个能打电话、能网页通话的语音机器人。 它最核心、也最巧的一招:把你画的图"编译"成给 LLM 的一组函数——图上每条箭头(出边) 都变成一个 LLM 可以调用的
transition工具,LLM 靠"调用哪条边对应的函数"来决定对话往哪走。 底下是一条 pipecat 实时语音管线在推着帧走:麦克风音频 → 语音转文字(STT)→ LLM → 文字转语音(TTS)→ 扬声器。
1. 这是什么(零基础也能懂)
一句话定义: Dograh 让你不写代码、用画流程图的方式搭一个能实时对话的电话/网页语音机器人。
解决谁的什么问题。 假设你要做一个"保险线索资格筛查"外呼机器人:它得先打招呼、问几个问题、 根据回答走不同分支、合格就转人工、不合格就礼貌挂断。传统上这要写一堆状态机代码 + 拼 STT/LLM/TTS 管线。Dograh 把这件事拆成两层:
- 给业务的人: 在画布上拖节点、连箭头、每个节点写一段提示词(prompt),每条箭头写一句 "什么情况下走这条"(condition)。
- 给平台: 把这张图翻译成 LLM 能执行的东西,再挂到一条实时语音管线上跑起来。
它能做什么(功能):
- 拖拽式工作流构建器:开始节点、Agent 节点、结束节点、全局节点。
- 实时语音通话:网页 WebRTC 通话、电话(Twilio/Vonage/Telnyx 等),可转人工。
- 自带整套栈:开箱即用的 LLM / TTS / STT,也可换成自己的任意供应商。
- 变量抽取、知识库检索(RAG)、自定义工具 / MCP 工具、通话录音回放、QA 质检节点。
用起来什么样。 自托管一条命令起服务后,打开 http://localhost:3010,选 Inbound/Outbound、
给机器人起个名、用一句话描述用途,点 Web Call 就能直接和它对话(README.md:102-108)。
一句话直觉/类比。 把它想成:你画的流程图 = 一张"对话地图";LLM = 一个只能沿地图走的司机; 每条路口(出边)= 一个写着"满足 X 条件时走这里"的路牌(函数)。 司机每到一个路口,读路牌、 判断当前对话满没满足条件,满足就"按下"那条边对应的函数,地图随之切到下一个节点——换新的提示词、 换新的一组路牌。
本节不涉及底层实现。目标:读完你能对同事讲清"Dograh 是干嘛的"。
2. 顶层全景(它大概怎么转)
怎么读下面这张图: 从上到下是两层。上层是控制层——PipecatEngine 状态机拿着你画的
WorkflowGraph,决定"现在在哪个节点、给 LLM 什么提示词和哪些工具"。下层是数据层——一条
pipecat 管线,音频/文本帧从左到右在处理器间流动。两层通过"LLM 调用 transition 函数"这一个动作耦合。
你在网页画布上拖出的对话流程图 (ReactFlow JSON)
│ 校验 + 建图
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 控制层 · PipecatEngine (services/workflow/pipecat_engine.py) │
│ │
│ WorkflowGraph ── 当前节点 ──► 为该节点组装: │
│ nodes / edges · 系统提示词 = 全局prompt + 节点prompt │
│ · 工具集 = 每条出边→一个 transition 函数 │
│ (+ 知识库/自定义/MCP 工具) │
│ │
│ LLM 调用某个 transition 函数 ──► set_node(目标节点) ──► 换提示词+换工具 │
└───────────────────────────────┬────────────────────────────────────────┘
│ 注册函数 / 更新 system prompt+tools
▼
┌──────────────────────────────────────────────────────────────────────┐
│ 数据层 · pipecat 语音管线 (services/pipecat/pipeline_builder.py) │
│ │
│ 麦克风 ─► 传输输入 ─► STT ─► 用户聚合器 ─► LLM ─► 引擎回调 ─► TTS ─► │
│ │ 传输输出─►扬声器│
│ └─(可选)录音/语音信箱检测 │
│ │
│ realtime 模式:STT+LLM+TTS 合成一个"语音到语音"服务,管线更短 │
└──────────────────────────────────────────────────────────────────────┘
主要部件一句话职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
WorkflowGraph / Node / Edge | 把校验过的画布 JSON 建成邻接表;每条边知道自己的函数名和跳转条件 | services/workflow/workflow_graph.py:173 |
PipecatEngine | 对话状态机:决定当前节点、组装提示词、把出边注册成函数、执行跳转 | services/workflow/pipecat_engine.py:59 |
| 上下文组装器 | 为节点拼系统提示词、把每条出边转成函数 schema | services/workflow/pipecat_engine_context_composer.py:49 |
| 管线构建器 | 按标准 / realtime 两种布局把处理器串成 pipecat Pipeline | services/pipecat/pipeline_builder.py:28 |
| 通话编排 | 从触发到建管线、初始化引擎、跑起来、收尾 | services/pipecat/run_pipeline.py:386 |
| 服务工厂 | 按供应商名产出具体的 LLM/STT/TTS/realtime 服务 | services/pipecat/service_factory.py:1066 |
主线走一遍(高层,不进代码):
- 建图。 前端画布导出的
ReactFlowDTO经校验后建成WorkflowGraph——节点连成邻接表, 每条边算出自己的函数名(把 label 转成合法标识符)与跳转条件。 - 接通即入场。 WebRTC/电话接通后,引擎
set_node(start_node),为开始节点装好提示词和一组 transition 函数,并把开场白(问候语或首次 LLM 生成)排进管线。 - 一个回合。 用户说话 → STT 转文字 → LLM 拿着"当前节点提示词 + 出边函数列表"生成回复; 若判断应当推进,LLM 调用某条出边对应的 transition 函数。
- 跳转。 该函数执行
set_node(目标节点):抽取本节点变量、(可选)播一句过渡语、换成新节点的 提示词与出边函数。跳到结束节点则排入EndFrame,通话收尾。 - 收尾。 做最终变量抽取、记录 disposition/标签、关闭 MCP 会话与录音。
目标:看懂"大盘"。想深入任一环,去下面第 3 节挑对应章节。
3. 阅读地图(从哪读起)
这是一个中等偏复杂的项目,已拆成多文件。建议顺序如下(由浅入深,每章都先点题再上源码):
-
对话即图:工作流数据模型与校验 — 先搞懂"图"长什么样: 四种节点类型(
startCall/agentNode/endCall/globalNode,dto.py:25)、边如何携带condition与transition_speech、Edge.get_function_name()怎样把 label 变成函数名、WorkflowGraph做哪些校验(连接数、节点基数)。这是后面一切的地基。 -
实时语音管线:帧如何在处理器间流动 — 看数据层:标准管线
transport→STT→聚合器→LLM→引擎回调→TTS→输出与 realtime(语音到语音)管线的不同布局, 以及为什么语音信箱检测器在两种布局里位置不对称(pipeline_builder.py:28与:97)。 -
图即工具:PipecatEngine 状态机(核心) — 全项目最精华的一章。 讲清
_setup_llm_context如何为节点注册出边函数、_create_transition_func里一次跳转的完整 时序(变量抽取→过渡语→set_node→on_context_updated回调触发新生成),以及结束节点如何收尾。 -
一次通话的端到端编排与引擎旁路能力 —
run_pipeline.py如何拼装 整通电话:选标准/realtime、造服务、建管线、engine.initialize()、接通后触发首个节点、finally里为何要在这里(而非cleanup())关 MCP 会话;以及"绕过 LLM 直接播音频"的旁路(问候/过渡录音)。 -
可扩展性接缝:注册表驱动的供应商与工具 — 想加一个新 LLM/STT/TTS 供应商、新节点类型、新工具时改哪里:
ServiceProviders枚举 +service_factory分派、 节点规格注册表(node_specs.register)、自定义/知识库/MCP 工具如何进入节点的函数集。
4. 巧妙之处(读者要带走的精华)
① 每条出边 = 一个无参 transition 函数,条件 = 函数描述。
这是整个项目的题眼。组装节点上下文时,引擎为每条出边生成一个函数 schema:函数名取边的 label
(经 re.sub(r"[^a-z0-9]", "_", label.lower()) 清洗,workflow_graph.py:53 Edge.get_function_name),
函数描述就是边上写的跳转条件 condition,且不带任何参数(pipecat_engine_context_composer.py:126-130
compose_functions_for_node;schema 由 get_function_schema 造,pipecat_engine_custom_tools.py:41)。
于是 LLM 的"函数选择"能力被直接用作"对话分支选择"——不需要另写路由逻辑,LLM 调哪个函数就走哪条边。
② 跳转在"函数调用结果写回上下文之后"才触发新生成。
_create_transition_func 把 set_node(目标) 排好后,通过 FunctionCallResultProperties(on_context_updated=…)
把"结束判断/新生成"挂在 pipecat 的回调上(pipecat_engine.py:288-316)。这样能 保证:等函数调用结果
落进 LLM 上下文、且新节点的提示词与工具已就位,才跑下一轮生成——避免"用旧提示词生成"的竞态。
③ 控制层与数据层用同一条管线,但引擎能"旁路"LLM 直接播音频。
过渡语和问候既可以是 LLM 生成的 TTS 文本,也可以是预录音频:引擎持有 transport.output(),
把音频帧直接排进传输输出、绕过 STT 和 LLM(pipecat_engine.py:255-267 的 play_audio,
set_transport_output 于 :852)。这让"固定话术"零延迟、可控。
④ realtime 与标准模式共用一套引擎,只换管线布局和一个旁路推理 LLM。
语音到语音模型(OpenAI Realtime、Gemini Live 等)自己吞 STT+LLM+TTS,管线更短
(build_realtime_pipeline,pipeline_builder.py:97);但变量抽取/摘要仍需一个文本 LLM,于是引擎
额外持有 inference_llm(pipecat_engine.py:89)。同一状态机、两种下层,复用度很高。
⑤ 校验驱动的图约束单一真源。 节点的入/出度、基数上限都由 NodeSpec.graph_constraints 描述,
WorkflowGraph._assert_connection_counts 统一执行(workflow_graph.py:306),加新节点类型时改 spec 即可。
5. 代码地图(导航索引)
下表给"人/agent 直接跳进源码"的入口。路径相对克隆根,均 as-of
sourceCommit。行号可能随上游漂移, 优先用符号名 grep 定位。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 建图 / 邻接表 / 校验 | api/services/workflow/workflow_graph.py | WorkflowGraph、Node、Edge |
| 边→函数名(核心转换) | api/services/workflow/workflow_graph.py:53 | Edge.get_function_name |
| 节点/边数据模型、节点类型枚举 | api/services/workflow/dto.py:25 | NodeType、EdgeDataDTO、ReactFlowDTO、AgentNodeData |
| 对话状态机(核心) | api/services/workflow/pipecat_engine.py:59 | PipecatEngine |
| 为节点装提示词+注册出边函数 | api/services/workflow/pipecat_engine.py:506 | PipecatEngine._setup_llm_context |
| 一次跳转的完整时序 | api/services/workflow/pipecat_engine.py:224 | PipecatEngine._create_transition_func |
| 切换当前节点 | api/services/workflow/pipecat_engine.py:549 | PipecatEngine.set_node |
| 节点系统提示词 / 出边→函数 schema | api/services/workflow/pipecat_engine_context_composer.py:49 | compose_system_prompt_for_node、compose_functions_for_node |
| 函数 schema 构造 | api/services/workflow/pipecat_engine_custom_tools.py:41 | get_function_schema |
| 标准 / realtime 管线布局 | api/services/pipecat/pipeline_builder.py:28 | build_pipeline、build_realtime_pipeline |
| 通话端到端编排 | api/services/pipecat/run_pipeline.py:386 | _run_pipeline、_run_pipeline_impl |
| 接通后触发首个节点 | api/services/pipecat/event_handlers.py:99 | maybe_trigger_initial_response(engine.set_node / queue_node_opening) |
| 供应商→具体服务分派 | api/services/pipecat/service_factory.py:1066 | create_llm_service、create_stt_service、create_tts_service、create_realtime_llm_service |
| 供应商枚举 | api/services/configuration/registry.py:64 | ServiceProviders、ServiceType |
| 节点规格注册表 | api/services/workflow/node_specs/__init__.py:28 | register、get_spec、all_specs |
入口速记(自检友好): 想理解"图怎么变成对话"读 §4-①(get_function_name + compose_functions_for_node);
想理解"一通电话怎么跑起来"从 _run_pipeline(run_pipeline.py:386)顺到 maybe_trigger_initial_response
(event_handlers.py:157)再进 PipecatEngine.set_node。