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

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

## 一、给 Claude Code 的任务摘要

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

必须遵守以下边界：

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

```text
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 后无记忆继续回答。
- 当前是中文字符/二元组字面相似度；记忆总量很小时，无命中会退回最近记忆。
- 用户姓名作为档案固定注入，不依赖语义检索。

### 必须达到的目标语义

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

## 五、为什么选择 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 Overview](https://docs.pipecat.ai/api-reference/pipecat-flows/overview)
- [Pipecat Function Calling](https://docs.pipecat.ai/pipecat/learn/function-calling)
- [Agno Memory](https://docs.agno.com/examples/memory/overview)
- [PydanticAI Agents](https://ai.pydantic.dev/agents/)
- [LangGraph Memory](https://docs.langchain.com/oss/python/langgraph/add-memory)
- [LiveKit Chat Context](https://docs.livekit.io/agents/logic/chat-context/)
- [OpenClaw Memory](https://docs.openclaw.ai/concepts/memory)

### 选型原则

选择 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 泄漏。

## 七、目标架构

```text
                    ┌─────────────────────────────┐
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 当数据库。

### 初始节点

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

1. `conversation`
   - 普通陪伴对话；
   - 可用只读搜索和受限记忆工具；
   - 默认节点。
2. `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。新增类似：

```text
app/agent/memory_policy.py
```

策略层输入至少包含：

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

策略层输出：

- allow_confirmed
- allow_candidate
- needs_confirmation
- reject

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

## 九、分阶段实施

### Phase 0：保护基线

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

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

### Phase 1：独立兼容性 spike

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

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

需要验证：

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

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

### Phase 2：抽出工具和策略层

建议新增：

```text
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：补齐通用记忆闭环

至少覆盖：

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

### Phase 5：迁移语音路径

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

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

### Phase 6：打断控制链

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

```text
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 切换后不会重复执行工具。
- 取消后不再写记忆或发送音频。

### 必须保留的现有测试

```bash
.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
```

### 真人网页验收

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

## 十二、回滚要求

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

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

```bash
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。