Files

T

ylhan f0dce05ac1 新增 Pipecat Flows 兼容性 spike 结论

交接文档规定"未经 spike 结论，不进入 Phase 3"。本次完成 Phase 1 spike，
7 项验证全部通过，并给出可进入 Phase 2 的结论。

新增 P1-Flows兼容性Spike结论.md

核心结论：
- **不装 pipecat-ai-flows**。Pipecat 1.10 内置 Flows（pipecat.flows），
  装独立包会报错提示卸载；独立包 1.4.0 会把 pipecat 降到 1.4.0
- **复用 DeepSeekLLMService**，不写 FallbackProvider 适配器。它自带
  DeepSeekLLMAdapter 处理 reasoning_content 回传（P0 踩过的坑），
  thinking 默认 disabled。代价是失去本地 vLLM 降级
- 实测：中文流式 14–22 块/回复、工具参数无重复拼接、回灌后出最终回答、
  worker.cancel() 立即终止、role_message 生效

定位了 5 个坑（4 个 Flows 相关 + 1 个环境）：
1. FlowManager.initialize 是协程但签名显示 -> None，漏 await 静默失效，
   role_message/task_messages/工具全部不生效而 LLM 照样回复，极易误判
2. 必须用 PipelineRunner 跑 worker；直接 run() 会让工具执行失败，
   异常被转成通用错误串，handler 从不执行，默认日志看不到
3. respond_immediately 默认 True，initialize 会立刻触发一轮回复，
   做归因测试时不清空观察窗口会归因错
4. role_message 与 task_messages 各产生一条 system 消息，
   两者不要同时用（切回本地 vLLM 时会 400）
5. NLTK punkt_tab 本地也需预装，否则 setup 超时卡住

对现有功能的影响：分句 TTS、RTVI、记忆身份不受影响；
SearXNG 工具需迁移；本地 vLLM fallback 会失去。
第四节给出 Phase 3 的最小生产集成 diff 要点。

2026-09-16 13:13:52 +08:00

19 KiB

Raw Blame History

P1 · Pipecat Flows Agent 化与记忆改造交接

日期：2026-09-16
状态：方案已选定，等待实现
实现仓库：~/workspace/ai-companion
部署目录：yonglin@cadadmin-server.taild0f91d.ts.net:~/ai-companion

一、给 Claude Code 的任务摘要

在不推翻现有实时语音链路的前提下，渐进引入 Pipecat Flows，把当前手写的单工具循环升级为受策略约束的 Agent 工具层，并补齐通用长期记忆的确认、读取、修改和忘记流程。

必须遵守以下边界：

保留现有 MemoryStore、SQLite 数据、身份隔离、DeepSeek fallback、SearXNG、 RTVI 网页显示和流式 TTS。
先做 Pipecat Flows 与当前 ConversationEngine 的兼容性 spike；验证通过前，不得替换线上默认路径。
使用 AI_AGENT_RUNTIME=legacy|flows 保留可立即回滚的 legacy 路径。
模型不得决定或传入 owner_user_id；用户身份必须由服务端会话上下文注入。
用户明确要求记忆与模型推测必须分开。模型推测不得直接成为 confirmed。
不把 Pipecat Flow state 当成长期记忆；长期数据仍通过 MemoryStore 持久化。
不顺手实现唤醒词、硬件 AEC、人脸自动切换、向量数据库或 systemd。

二、开始前必须知道的仓库状态

代码和部署

Git 已提交基线包含：
- c3d2c7f：修复 Pipecat 实时链路并补齐网页测试
- 719fc13：修复网页对话并接入 SearXNG 搜索
2026-09-16 的姓名档案修复、疑问句防误写和测试目前仍是未提交改动。
线上服务已运行这些未提交改动，不等于服务器代码与 Git HEAD 相同。
开始工作前必须运行 git status --short 和 git diff；不得 reset、checkout 或覆盖现有改动。
线上健康接口：http://127.0.0.1:7860/health。
网页调试入口通过 SSH 隧道访问：http://localhost:7860/client/。

当前已验证能力

SmallWebRTC 信令、ICE、data channel、音频输入输出已闭环。
浏览器真人麦克风 → STT → DeepSeek → Qwen3-TTS → 浏览器播放正常。
网页文字输入、用户文本、助手文本和语音回复正常。
DeepSeek 能自主调用 web_search，SearXNG 工具结果能回灌后继续回答。
用户自报“我叫林”会同步写入 confirmed 姓名档案，disconnect/reconnect 后恢复。
姓名疑问句不会再把“谁吗”之类内容覆盖为姓名。
当前专项回归：
- 姓名档案 22/22
- 记忆存储 30/30
- 身份 19/19
- 搜索 12/12
- 传输 4/4

线上姓名数据库备份

首次档案迁移前： /home/yonglin/.local/share/ai_assistant/memory.db.before-name-profile-20260916
疑问句误写修复前： /home/yonglin/.local/share/ai_assistant/memory.db.before-name-question-20260916-1140

不要删除这些备份。后续若改 schema，必须另建带时间戳的备份。

三、当前系统到底是什么

当前不是完整通用 Agent，而是：

SmallWebRTC / RTVI
        ↓
EngineProcessor
        ↓
SimpleVAD → StreamingSTT
        ↓
ConversationEngine
  身份 → 自动记忆检索 → DeepSeek/搜索工具循环 → 分句 TTS
        ↓
WebRTC 音频与 RTVI 文本

ConversationEngine._stream_and_speak() 手写了最多两轮的 web_search Function Calling 循环。模型能自主决定是否搜索，但还不能自主调用记忆工具。

Pipecat 当前主要负责传输、Frame、PipelineWorker 和 RTVI；STT、LLM、TTS 语义链路仍由自有引擎调度。

四、当前记忆决策与真实缺口

写入

输入	当前行为	实际是否可复用
“我叫林”	同步更新 `users.display_name`，写 confirmed 姓名	是
“记住我不吃香菜”	回复后规则抽取，写 candidate	否，默认检索排除
“我喜欢咖啡”	回复后规则抽取，写 candidate	否，默认检索排除
普通聊天	只保留当前内存历史和 user turn	重连后不可用

当前没有 candidate 的确认、晋升或管理入口。因此除姓名外，“写进数据库”和“助手会在以后使用”不是一回事。

读取

每条已确认身份的用户消息都会尝试检索，不是模型按需决定。
默认只读取 confirmed、private/shared、未过期记忆。
最多 5 条，超时 300ms 后无记忆继续回答。
当前是中文字符/二元组字面相似度；记忆总量很小时，无命中会退回最近记忆。
用户姓名作为档案固定注入，不依赖语义检索。

必须达到的目标语义

稳定档案：姓名、称呼、语言和明确的稳定偏好；连接开始时加载。
显式长期记忆：用户明确说“记住/以后/别忘了”，经策略校验后 confirmed。
候选观察：普通聊天中推断出的可能信息只能 candidate，不直接影响回答。
事件/摘要：会话事件与逐字稿不冒充永久事实，可过期、可删除。
矛盾更新：同 owner + kind + key 的 confirmed 新值覆盖或 supersede 旧值。
透明控制：支持查看、确认、修改、忘记；模糊删除必须追问。

五、为什么选择 Pipecat Flows

候选比较

框架	优点	与本项目的主要问题	结论
Pipecat Flows	已在 Pipecat 生态；状态、节点、函数注册和校验；支持 DeepSeek 等 OpenAI-compatible 服务	期待 Pipecat `LLMService + context aggregator`，与当前自有引擎有接口差异	选择，渐进集成
Agno	内置 user/session/memory，最接近轻量版 OpenClaw	会与现有会话、历史和记忆层重叠	暂不采用
PydanticAI	类型安全、依赖注入、流式工具事件、较轻	长期记忆策略仍要自建；与现有工具循环收益有限	备选
LangGraph	持久状态、checkpoint、长任务和人工审批成熟	对当前实时陪伴链路过重	暂不采用
LiveKit Agents	完整实时语音 Agent、工具、handoff	相当于替换已跑通的 Pipecat/WebRTC；长期记忆仍需外接	不迁移
smolagents	代码少、多步工具 Agent	偏任务型，多步延迟高，API仍偏实验	不用于主链路
OpenClaw	通用 Agent runtime、分层记忆和后台整理完整	运行时过重、默认个人 Agent 假设、实时语音和身份边界仍需定制	不整体迁移

官方资料：

选型原则

选择 Pipecat Flows，不是为了追求“框架化”，而是为了：

复用现有实时媒体栈；
把工具 schema、执行、结果回灌和状态切换从手写循环中抽离；
给记忆确认/删除等多轮流程提供显式状态；
保留产品特有的身份、隐私和低延迟策略。

六、最重要的集成风险

Pipecat Flows 的正常构造需要：

PipelineTask
Pipecat LLMService
context aggregator
由 Pipeline Frame 驱动的函数调用和消息更新

当前项目则是：

PipelineWorker + 自定义 EngineProcessor
自有 FallbackProvider，不是 Pipecat LLMService
自有消息历史与 prompt 构造
自有 STT/TTS 调用，不是标准 Pipecat service processor
RTVI 事件由 EngineProcessor 手工补发

因此不能直接“安装 Flows，然后包住 ConversationEngine”。如果忽略这一点，很容易出现：

两套消息历史同时增长；
工具调用重复执行；
token 既被自有引擎又被 Pipecat aggregator 消费；
RTVI 文本重复或消失；
TTS 重复播报；
取消只停止其中一套任务；
身份切换后旧 context 泄漏。

七、目标架构

                    ┌─────────────────────────────┐
WebRTC / RTVI ─────▶│ Pipecat Pipeline            │
                    │ transport + turn + metrics  │
                    └──────────────┬──────────────┘
                                   │
                    ┌──────────────▼──────────────┐
                    │ Flows Runtime（可切换）      │
                    │ node / state / tools        │
                    └──────────────┬──────────────┘
                                   │
          ┌────────────────────────┼────────────────────────┐
          ▼                        ▼                        ▼
   LLMService 适配器          Memory Policy             Web Search
   DeepSeek/Fallback          + MemoryStore              SearXNG
          │                        │
          ▼                        ▼
   流式文本与 TTS            SQLite（继续使用）

Flow state 只保存会话控制状态

建议字段：

session_id
服务端绑定的 user_id（内部使用，不展示给模型）
display_name
identity_version
pending_memory_action
pending_confirmation
当前 node

不要把全部记忆内容复制进 Flow state，也不要把 state 当数据库。

初始节点

第一版只需要两个节点，不要设计庞大状态图：

conversation
- 普通陪伴对话；
- 可用只读搜索和受限记忆工具；
- 默认节点。
confirm_memory_change
- 只在修改、删除目标不明确或敏感写入时进入；
- 用户确认/取消后返回 conversation。

八、工具设计

全局工具

`web_search(query)`

复用 app/search.py 与 SearXNG。
保留最多两次连续搜索限制。
搜索失败返回工具错误结果，不让整轮异常退出。
来源 URL 保留给网页，TTS 继续不朗读来源段。

`memory_search(query)`

owner_user_id 从会话注入，禁止模型传入。
只返回 confirmed、未过期、授权 scope。
不确认身份时返回不可用，不搜索任何私人记忆。
结果条数和超时沿用配置上限。

`memory_remember(kind, key, value)`

仅确认身份可用。
只有当前用户明确表达“记住/以后/别忘了/我的偏好是”等意图时，才能 confirmed。
模型推断或总结只能写 candidate。
owner_user_id、source session/turn、scope 由服务端填充。
默认 private；模型不得扩大成 shared。
同 key confirmed 写入保持幂等并更新旧值。
姓名确定性快路径暂时保留，不立即交给模型。

`memory_forget(selector)`

仅确认身份可用。
精确匹配一条时可返回待删除摘要；删除前应有明确用户意图。
多条或模糊匹配进入 confirm_memory_change，不得猜测删除目标。
删除必须同时匹配 owner；不得只凭 memory_id。
返回“已删除/没有找到/需要确认”，不能把“请求已接收”说成“已经删除”。

`memory_update(key, value)`

用于用户明确纠正已有事实或偏好。
更新必须保留来源和更新时间。
若未来增加 superseded 状态，先做向后兼容 migration，不删旧 schema。

`profile_update(display_name)`

第一阶段可继续使用现有确定性姓名提取，不必立即暴露给模型。
如果以后工具化，也必须经过同一疑问句和角色描述过滤。

记忆执行策略必须独立于 prompt

不要仅依赖“请不要误记”的 system prompt。新增类似：

app/agent/memory_policy.py

策略层输入至少包含：

当前原始 user text
工具请求
已确认身份
当前 session/turn
已有同 key 记忆

策略层输出：

allow_confirmed
allow_candidate
needs_confirmation
reject

模型可以提出动作，策略层拥有最终决定权。

九、分阶段实施

Phase 0：保护基线

运行 git status --short、保存 git diff。
运行当前相关测试。
记录网页文字、网页语音、搜索、姓名重连的基线延迟与行为。
不修改线上数据库 schema。

完成标准：现有未提交修复没有丢失，基线测试结果有记录。

Phase 1：独立兼容性 spike

新增独立实验，不先改生产入口，例如：

p1/flows_spike.py
app/tests/test_flows_spike.py

需要验证：

找到与 pipecat-ai==1.10.0 兼容的 pipecat-ai-flows 版本并精确 pin。
DeepSeek OpenAI-compatible 接口能在 Flows 中流式输出中文内容。
流式 function-call arguments 能被正确聚合一次，不重复拼接。
工具结果回灌后能生成最终回答。
任务取消能终止 LLM 请求和后续工具执行。
context aggregator 不产生第二条 system message。
能获得 token 级文本事件，满足网页显示和分句 TTS。

若任何一项不成立，停止生产集成，先记录结论；不得为了使用框架而绕过现有正确行为。

Phase 2：抽出工具和策略层

建议新增：

app/agent/
├── __init__.py
├── tools.py
├── memory_policy.py
├── flows_runtime.py
└── llm_service.py

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

完成标准：工具层可被 legacy 和 flows 两条路径复用，策略测试不依赖真实 LLM。

Phase 3：先接网页文字路径

新增 AI_AGENT_RUNTIME，默认仍为 legacy。
只让 RTVI send-text 在显式 flows 配置下走新路径。
保持网页用户文本、token 增量、完整 assistant 文本和可选 TTS。
语音路径暂时保持 legacy。

完成标准：同一服务可切换两条 runtime；legacy 回归无变化；flows 文字路径通过工具、姓名、记忆和搜索验收。

Phase 4：补齐通用记忆闭环

至少覆盖：

“记住我不吃香菜” → confirmed。
disconnect/reconnect → “我不吃什么”能召回香菜。
“我今天吃了香菜”不得自动变成稳定偏好。
“其实我现在吃香菜了”能明确更新旧偏好，而不是保留冲突 active 值。
“忘掉我的饮食偏好”能删除；目标模糊时先确认。
访客无法读取或写入私人记忆。
用户 A 的记忆绝不出现在用户 B 的 prompt、工具结果或日志中。

Phase 5：迁移语音路径

只有 Phase 3、4 稳定后才迁移语音：

保持 STT transcript 和原始 user text 一致；
保持首句尽快进入 TTS；
工具调用期间不得把工具 JSON 念出来；
来源段不进入 TTS；
用户打断时取消 Flows、LLM、工具和 TTS 的同一轮 execution；
ConversationEngine 中已经稳定的身份与记忆接口可以拆分复用，不要求一次删除旧类。

Phase 6：打断控制链

当前 SimpleVAD 只做 RMS 话段切分。下一步先实现控制闭环，不在本阶段承诺真实硬件声学效果：

user-started-speaking
  ├─ 发出 InterruptionFrame / 客户端停播信号
  ├─ 取消当前 Flows/LLM/TTS execution
  ├─ 清空未发送和客户端已缓冲音频
  └─ 保留新用户话语的 pre-roll 并继续收集

要求：

不等待 700ms 尾静音才停止助手；
使用开始阈值、连续帧或正式 VAD，避免单个噪声帧误触发；
_bot_speaking 不能只代表服务端推帧结束，要考虑客户端仍在播放；
文字请求期间的麦克风回采不得错误抢占；
先做合成 Frame 回归；真实 AEC、扬声器回声和远场误触发留到硬件 P2。

Phase 7：部署与清理

flows 默认值只有在线上验收后才能从 legacy 改为 flows。
更新 README、部署说明和本交接文档。
在修改数据库前创建新备份。
确认 /health、网页文字、网页语音、搜索和重连记忆。
确认日志没有 API key、私人全文记忆或内部 ID 泄漏。

十、测试与验收矩阵

必须新增的单元测试

工具参数 schema 与非法参数。
模型不能传 owner_user_id/scope 越权。
未确认身份拒绝 private memory。
明确“记住”写 confirmed。
推断信息只写 candidate。
同 key 更新幂等。
模糊忘记返回 needs_confirmation。
精确忘记只删除当前 owner。
Flows node 切换后不会重复执行工具。
取消后不再写记忆或发送音频。

必须保留的现有测试

.venv/bin/python app/tests/test_profile_memory.py
.venv/bin/python app/tests/test_memory.py
.venv/bin/python app/tests/test_identity.py
.venv/bin/python app/tests/test_search.py
.venv/bin/python app/tests/test_transport.py
.venv/bin/python app/tests/test_engine_e2e.py

真人网页验收

文字：“记住我不吃香菜。”
disconnect/reconnect。
文字：“我不吃什么？”
语音重复 1–3。
问实时问题，确认会搜索且显示来源。
助手播长回答时插话，确认音频停止、插话未丢失。
问“我是谁”，只能回答用户称呼，不能回答内部 ID。

性能验收

无工具普通对话不能因为 Flows 固定增加一次额外 LLM 调用。
记忆检索仍有超时和失败降级。
分别记录 VAD、STT、记忆检索、LLM 首 token、工具、TTS 首段和首音。
对比 legacy/flows 的 P50/P95；无工具首音出现明显回退时不得切默认。

十一、明确非目标

本次不要做：

唤醒词和 20 秒连续对话窗口；
N5095 产品客户端、声卡 AEC 或阵列波束形成；
人脸自动确认说话者；
Mem0、Chroma、Graphiti、LanceDB 等新存储；
OpenClaw、Agno、LangGraph 或 LiveKit 迁移；
多 Agent、子 Agent 或复杂任务规划；
米家/Home Assistant 工具；
systemd、自启动和公网暴露；
清理或重写所有 legacy 代码。

十二、回滚要求

AI_AGENT_RUNTIME=legacy 必须能绕过全部 Flows 代码。
新数据库字段必须向后兼容，禁止 drop/rename 现有表或列。
工具执行要有幂等键或 source turn，避免重试重复写入。
新路径异常时，本轮可以无记忆继续，但不能自动切换后重复执行有副作用工具。
部署前保留数据库与上一版本代码；回滚不能依赖重新生成用户数据。

十三、建议 Claude Code 的第一轮动作

cd ~/workspace/ai-companion
git status --short
git diff --check

# 先读这些文件，不要先改
sed -n '1,240p' services/bot_server.py
sed -n '240,560p' services/bot_server.py
sed -n '180,500p' app/engine.py
sed -n '1,220p' app/memory/base.py
sed -n '220,380p' app/memory/sqlite_store.py

然后只完成 Phase 1 spike，提交一份结论：

兼容的 flows 版本；
选择复用 Pipecat OpenAI-compatible LLMService，还是实现 FallbackProvider → LLMService 适配器；
token、tool-call、取消和 context 的实测结果；
建议的最小生产集成 diff；
是否会破坏现有分句 TTS、RTVI 和 fallback。

未经 spike 结论，不进入 Phase 3。

19 KiB Raw Blame History Unescape Escape