# P1 · Pipecat Flows 兼容性 Spike 结论

> 日期：2026-09-16
> 状态：**spike 完成，7 项验证全部通过**
> 代码：`p1/flows_spike.py`（探索脚本）、`app/tests/test_flows_spike.py`（断言版）
> 前置：[P1-Pipecat Flows Agent化与记忆改造交接](P1-Pipecat%20Flows%20Agent化与记忆改造交接.md)

交接文档规定：**未经 spike 结论，不进入 Phase 3**。本文即该结论。

---

## 一、结论摘要

| 问题 | 结论 |
|---|---|
| 兼容的 flows 版本 | **不装独立包**。Pipecat 1.10.0 **内置** Flows（`pipecat.flows`） |
| LLMService 选择 | 复用 Pipecat 的 `DeepSeekLLMService`；**代价是失去本地 vLLM 降级** |
| token 级文本 | ✅ 实测 14–22 块/回复，满足网页显示与分句 TTS |
| 工具调用 | ✅ 参数聚合完整，无重复拼接；工具结果回灌后能出最终回答 |
| 取消 | ✅ `worker.cancel()` 能立即终止 LLM 流 |
| context | ⚠️ **会产生两条 system 消息**（见 §五） |
| 是否破坏现有功能 | 见 §六——**有两处需要处理** |

---

## 二、版本：不需要独立包，而且装了有害

**这是最重要的一条发现。**

`pipecat-ai 1.10.0` **已经内置了 Pipecat Flows**，模块路径 `pipecat.flows`。
装独立的 `pipecat-ai-flows` 包时，Pipecat 会直接报错提示：

```
ERROR | pipecat:_warn_if_standalone_flows_installed:45 -
The separate `pipecat-ai-flows` package is installed alongside a version of
Pipecat that already includes Pipecat Flows as `pipecat.flows`.
You do not need both — uninstall `pipecat-ai-flows` and import Flows from `pipecat.flows`.
```

而且独立包的版本区间有陷阱：

| 独立包版本 | 要求的 pipecat-ai | 后果 |
|---|---|---|
| 1.4.0 | `>=1.4.0,<1.5.0` | **会把 pipecat 从 1.10.0 降到 1.4.0** |
| 1.3.0 | `>=1.4.0,<2` | 可安装，pytest dry-run 只新增自己 |

（1.4.0 是"更新"的版本，却钉死在更老的 pipecat 区间——如果按"装最新版"的直觉操作，
会静默把整个框架降级。）

**决定**：

- `requirements.txt` **不添加** `pipecat-ai-flows`
- 代码 `from pipecat.flows import FlowManager, NodeConfig, FlowsFunctionSchema`
- 内置版与独立包 **API 一致**（FlowManager / NodeConfig / FlowsFunctionSchema 签名相同），
  所以这个切换不影响任何写法

内置版的导出比独立包更多（多出 `Flow`、`FlowConfig`、`FlowProblem`、`FlowReferenceError`），
像是更演进的 API，但 spike 只用到了两者共有的部分。

---

## 三、LLMService 选择

**结论：复用 Pipecat 的 `DeepSeekLLMService`，不写 `FallbackProvider → LLMService` 适配器。**

理由：

1. **它自带 `DeepSeekLLMAdapter`**，源码注释明写 "Supplies the `reasoning_content`
   DeepSeek requires on assistant messages in thinking mode"——**这正是我们在 P0
   阶段踩过的坑**（见 `P0-DeepSeek验证报告.md`：开思考时多轮/工具回灌一律 400）。
   框架已经处理了。
2. `thinking` 默认就是 `disabled`，与 P0 的结论一致。
3. 支持自定义 `base_url`，将来要接本地 vLLM 也能用。
4. **无法适配 `FallbackProvider`**：Flows 要求的是 `LLMService` 实例
   （`LLMService | LLMSwitcher`），而 `pipecat_flows` 里那个 `LLMAdapter`
   **只管摘要生成**，不是 provider 适配器。要保留 fallback 必须自己写
   `LLMService` 子类，成本中等。

**必须覆盖模型名**：默认是 `deepseek-v4-flash`，**该模型不存在**，用默认值会 400。
必须显式设 `DeepSeekLLMService.Settings(model="deepseek-flash")`。

**失去本地 vLLM 降级的代价**：目前本机 27B 的 vLLM 已停用，fallback 实际上没有可用的
后备目标。所以 Phase 3 先只用 DeepSeek 是合理的。将来若要恢复双后端，
Pipecat 有 `LLMSwitcher` 可用，不必自写。

---

## 四、实测结果（7 项，9 个断言全通过）

```
=== 2. DeepSeek 在 Flows 中流式输出中文 ===
      回复: '我是你的桌面陪伴小助手，平时就待在这儿陪你聊聊天、说说话。'
      分块数: 18
  [PASS] 有中文回复
  [PASS] 回复不含 Markdown
  [PASS] 7. token 级文本事件（>1 块） — 18 块

=== 3+4. 流式工具参数聚合 / 工具结果回灌 ===
      工具调用参数: [{'city': '北京'}]
      最终回复: '北京现在是下午三点半。'
  [PASS] 3. 工具被调用 — 1 次
  [PASS] 3. 参数聚合完整（无重复拼接） — {"city": "北京"}
  [PASS] 4. 工具结果回灌后有最终回答

=== 5. 取消 ===
      取消前 13 块 → 取消后 1.5s 内 13 块（又收到 0 块）
  [PASS] 5. 取消后停止输出

=== 6. system message ===
      context 消息角色序列: ['system', 'user', 'assistant']
  [PASS] 6. system 消息不超过 1 条
  [PASS] 6. 有 user 与 assistant 轮次
```

补充观察：
- **role_message 生效**：设"你是桌面陪伴助手"后，模型自称"桌面陪伴小助手"，
  不再是通用的"DeepSeek AI 助手"
- **工具调用只有 1 次**（修好 runner 之前会失败重试 2 次，见 §五）

---

## 五、暴露的问题（4 个坑，都已定位）

### 坑 1：`FlowManager.initialize` 是协程，漏 `await` 会静默失效

```python
inspect.signature(FlowManager.initialize)   # (self, initial_node=None) -> None
inspect.iscoroutinefunction(...)            # True  ← 签名上看不出来
```

漏 `await` 时**不报错**，只是整个 flow 从不初始化：role_message、task_messages、
工具注册**全部没生效**，而 LLM 照样会回复（因为 `run_llm=True` 直接触发推理）。

现象极具误导性：会让人误判成"Flows 不支持这些特性"。实测中我一度得出
"role_message 不生效"的结论，就是这个原因。

### 坑 2：必须用 `PipelineRunner` 跑 worker，不能直接 `worker.run()`

直接 `worker.run(...)` 时 runner 没挂上，**工具执行会失败**：

```
Error executing function call [get_time]:
Worker 'PipelineWorker#0': runner is not set; call attach() first.
```

这个异常被 Pipecat 转成给 LLM 的通用错误串
`"The function X failed and returned no result."`，**handler 从不执行**，
日志里默认看不到真实原因。要开 DEBUG 才看得到。

正确写法：`await PipelineRunner().run(worker)`

### 坑 3：`respond_immediately` 默认 `True`，initialize 会立刻触发一轮回复

`FlowManager.initialize()` 会 queue 一个 `LLMRunFrame`，**在用户说话之前**
就产生一轮 LLM 回复。

做测试/归因时如果不清空观察窗口，会把"节点初始化的自动回复"当成
"对用户问题的回复"——我第一版 spike 就误判了第 2 项。

生产上这通常是想要的行为（节点进入即主动开口），但**必须知道它存在**。

### 坑 4：会产生两条 system 消息

实际发给 DeepSeek 的请求里：

```
[system] '你是桌面陪伴助手……'          ← role_message（走 system_instruction）
[system] '当前任务：普通陪伴对话。'      ← task_messages
[user] '……'
```

`role_message` 通过 `LLMUpdateSettingsFrame(system_instruction=...)` 下发，
而 `task_messages` 里的 system 消息留在 context 里，**两者都进请求**。

交接文档 §九 Phase 1 第 6 项要求"context aggregator 不产生第二条 system message"——
严格说，**context 里确实只有一条**（`task_messages` 那条），
`role_message` 走的是 LLM settings 而非 context。但**发出去的请求有两条 system**。

风险：
- 多数模型能处理多条 system，但**优先级不保证**（谁覆盖谁取决于实现）
- 我们之前在**本机 vLLM** 上就吃过"多条 system 直接 400"的亏
  （见 `P0-本地LLM验证报告.md`）——如果将来把 LLM 切回本地 vLLM，这条会立刻炸

**建议**：Phase 3 集成时**只用一种**。要么把节点任务说明也写进 `role_message`，
要么不用 `role_message`、全部走 `task_messages`。不要两个都用。

### 坑 5（环境）：NLTK `punkt_tab` 本地也要装

Pipecat 启动会预热 NLTK 分句器，缺 `punkt_tab` 会尝试联网下载并被安全策略拦
（`SSRF attempt to restricted IP`），setup 超时后管线卡住。

服务器上已装（`~/.local/share/ai-companion/nltk_data`），**本地开发机也要装**，
否则 spike 跑不起来。`p1/flows_spike.py` 已加 fail-fast 检查与拷贝命令提示。

---

## 六、对现有功能的影响

| 现有功能 | 是否受影响 | 说明 |
|---|---|---|
| 分句 TTS | **不受影响** | Flows 输出 token 级 `TextFrame`（实测 18 块），自有分句逻辑照用 |
| RTVI 网页显示 | **不受影响** | token 增量与完整文本都能拿到 |
| 记忆 / 身份 / SQLite | 不受影响 | Flows 的 state 只存会话控制状态，长期数据仍走 `MemoryStore` |
| SearXNG 搜索 | 需迁移 | 现在的 `web_search` 手写循环要改成 Flows 工具 |
| **本地 vLLM fallback** | **会失去** | `FallbackProvider` 不能被 Flows 直接使用（见 §三） |
| 两条 system 消息 | **需要注意** | 见坑 4；切回本地 vLLM 时会 400 |

---

## 七、建议的最小生产集成 diff（Phase 3 用）

不改动现有 `EngineProcessor` 的音频路径，只在**文字路径**上分流：

1. `app/config.py` 加 `AI_AGENT_RUNTIME=legacy|flows`，**默认 legacy**
2. `app/agent/` 新增 `tools.py`（工具定义）、`memory_policy.py`（记忆策略）、
   `flows_runtime.py`（FlowManager 装配）
3. `services/bot_server.py` 里，RTVI `send-text` 按 `AI_AGENT_RUNTIME` 分流：
   `legacy` → 现有 `EngineProcessor`；`flows` → 新建的 Flows 路径
4. 语音路径**暂不动**，留到 Phase 5
5. `requirements.txt` **不加** `pipecat-ai-flows`

关键注意点：
- 用 `PipelineRunner().run(worker)`，不要直接 `worker.run()`
- `await flow.initialize(node)`
- `role_message` 与 `task_messages` 的 system 消息**二选一**
- `DeepSeekLLMService.Settings(model="deepseek-flash")` 必须显式覆盖

---

## 八、下一步

spike 已给出结论，**可以进入 Phase 2**（抽出工具与策略层）：

- 把现有 `web_search` 执行从 `ConversationEngine` 抽成可复用函数
- 实现记忆工具，先用 fake store 单测
- legacy 路径继续工作，Flows 不是唯一调用者

Phase 3（接网页文字路径）等 Phase 2 完成标准达成后再开始。

---

## 九、复现

```bash
cd ~/workspace/ai-companion

# 探索脚本（带请求体抓取与逐项诊断）
~/venvs/ai_assistant/bin/python p1/flows_spike.py
~/venvs/ai_assistant/bin/python p1/flows_spike.py --only 3

# 断言版（接 CI 用）
~/venvs/ai_assistant/bin/python app/tests/test_flows_spike.py

# 需要 DEBUG 日志排查时
SPIKE_DEBUG=1 ~/venvs/ai_assistant/bin/python p1/flows_spike.py --only 3
```