diff --git a/app/agent/flows_runtime.py b/app/agent/flows_runtime.py index 1da78b8..810acdb 100644 --- a/app/agent/flows_runtime.py +++ b/app/agent/flows_runtime.py @@ -71,16 +71,21 @@ def set_last_user_text(text: str) -> None: #: 会话控制状态。**不要把长期记忆复制进来**——Flows 的 state 不是数据库， #: 长期数据一律走 MemoryStore（交接文档 §七）。 -def _conversation_node(*, display_name: str, tool_schemas: list[dict], +def _conversation_node(*, identity_decision, memories: list, + tool_schemas: list[dict], ctx_factory: Callable[[], agent_tools.ToolContext]) -> dict: """默认节点：普通陪伴对话 + 可用工具。 - 人格与任务都放在 task_messages 里——**不用 role_message**， - 避免两条 system 消息（见模块文档）。 + 人格与记忆都放进**同一条** system 消息（`task_messages`）—— + **不用 role_message**，避免两条 system 消息（见模块文档）。 + + 这里复用 legacy 的 `build_system_prompt`，保证两条路径： + - 人设与规则完全一致 + - 记忆注入格式一致 + - 身份未确认时的"按访客对待"口径一致 """ - persona = PERSONA_RULES - if display_name: - persona += f"\n\n当前用户希望被称为：{display_name}。" + from ..prompts import build_system_prompt + persona = build_system_prompt(identity_decision, memories) from pipecat.flows import FlowsFunctionSchema @@ -289,7 +294,8 @@ async def build_flows_parts( """ from .llm_service import create_llm_service - llm = create_llm_service(cfg.llm) + # 用 agent 的 max_tokens，不是语音的 200——带工具需要余量 + llm = create_llm_service(cfg.llm, max_tokens=cfg.agent.max_tokens) if llm is None: raise RuntimeError( f"Flows 路径不支持后端 {cfg.llm.name!r}（只支持 deepseek）。" @@ -326,8 +332,29 @@ def make_flow_manager(*, worker, llm, aggregator): return FlowManager(worker=worker, llm=llm, context_aggregator=aggregator) +async def load_stable_memories(memory, identity, *, limit: int = 8) -> list: + """加载"稳定档案"：已确认、未过期的记忆。 + + 方案 §四要求"稳定档案（姓名、称呼、语言、明确的稳定偏好） + **连接开始时加载**"，而不是等模型自己想起来查。 + + 身份未确认 → 返回空。访客不加载任何私人记忆（验收第 6 条）。 + """ + d = identity.current() + if not d.confirmed or memory is None: + return [] + try: + rows = await memory.list_memories(d.user_id, include_candidates=False) + except Exception as e: # noqa: BLE001 — 加载失败不阻塞会话 + log.warning("加载稳定记忆失败（继续）：%s", e) + return [] + # 最近更新的排前面 + rows.sort(key=lambda m: getattr(m, "updated_at", 0), reverse=True) + return rows[:limit] + + async def initialize_conversation_node( - flow, *, display_name: str, tool_schemas: list[dict], + flow, *, identity_decision, memories: list, tool_schemas: list[dict], ctx_factory: Callable[[], agent_tools.ToolContext], ) -> None: """设置初始节点。 @@ -337,7 +364,8 @@ async def initialize_conversation_node( Phase 1 spike 在这里栽过一次，表现极具误导性： 模型会正常回复（因为 LLM 照样被调用），但没有人格、没有工具。 """ - node = _conversation_node(display_name=display_name, + node = _conversation_node(identity_decision=identity_decision, + memories=memories, tool_schemas=tool_schemas, ctx_factory=ctx_factory) await flow.initialize(node) @@ -345,6 +373,6 @@ async def initialize_conversation_node( __all__ = [ "FlowsBridge", "UserTextRecorder", "build_flows_parts", "make_flow_manager", - "initialize_conversation_node", + "initialize_conversation_node", "load_stable_memories", "set_last_user_text", "PERSONA_RULES", ] diff --git a/app/agent/llm_service.py b/app/agent/llm_service.py index 48e54fa..21c9995 100644 --- a/app/agent/llm_service.py +++ b/app/agent/llm_service.py @@ -27,7 +27,7 @@ from ..config import LLMConfig log = logging.getLogger(__name__) -def create_llm_service(cfg: LLMConfig): +def create_llm_service(cfg: LLMConfig, *, max_tokens: int | None = None): """按配置造 Pipecat LLMService。 目前只支持 DeepSeek（OpenAI 兼容）。其它后端返回 None，调用方应当 @@ -42,12 +42,14 @@ def create_llm_service(cfg: LLMConfig): # **必须显式覆盖模型名。** # Pipecat 的默认是 `deepseek-v4-flash`，该模型不存在，用默认值会 400。 # 见 P0-DeepSeek验证报告.md：真实清单只有 deepseek-flash 与 deepseek-v4-pro。 + # max_tokens 由调用方指定（Flows 路径用 AgentConfig.max_tokens， + # 不是语音的 200）。见 create_llm_service 的 max_tokens 参数。 settings = DeepSeekLLMService.Settings( model=cfg.model, # 显式关闭思考。Pipecat 的默认恰好也是 disabled，但写出来更安全： # 将来默认值变化时不会悄悄改变行为，也不至于把思考内容送进 TTS。 thinking=DeepSeekLLMService.ThinkingConfig(type="disabled"), - max_tokens=cfg.max_tokens, + max_tokens=max_tokens or cfg.max_tokens, ) return DeepSeekLLMService( api_key=cfg.api_key, diff --git a/app/config.py b/app/config.py index 4fa5aff..7c8f509 100644 --- a/app/config.py +++ b/app/config.py @@ -224,6 +224,14 @@ class AgentConfig: #: "flows" = Pipecat Flows 驱动文字路径（Phase 3 新路径） runtime: str = field(default_factory=lambda: _env("AI_AGENT_RUNTIME", "legacy")) + #: Flows 路径的 max_tokens。 + #: **不能沿用语音的 200**：实测同一句"忘掉我的饮食偏好"， + #: max_tokens=200 时模型直接 finish=stop、根本不调工具； + #: 600 时才正常产生 tool_calls 且参数 JSON 完整。 + #: 200 是为"语音回复要短"设的，而带工具的路径需要额外余量。 + max_tokens: int = field( + default_factory=lambda: _env_int("AI_AGENT_MAX_TOKENS", 600)) + @property def use_flows(self) -> bool: return self.runtime.strip().lower() == "flows" diff --git a/app/tests/test_flows_runtime.py b/app/tests/test_flows_runtime.py index d171bf0..51c2c63 100644 --- a/app/tests/test_flows_runtime.py +++ b/app/tests/test_flows_runtime.py @@ -99,7 +99,9 @@ def test_runtime_switch(): def test_node_shape(): print("\n=== 2. 节点配置形态 ===") - identity = ManualIdentityProvider("u_a") + identity = ManualIdentityProvider() + # 称呼来自 IdentityDecision.display_name（不是构造参数） + identity.switch("u_a", display_name="林") store = FakeStore() bridge = FR.FlowsBridge(rtvi=FakeRTVI(), tts=FakeTTS(), cfg=AppConfig.load()) @@ -117,7 +119,11 @@ def test_node_shape(): "properties": {"query": {"type": "string"}}, "required": ["query"]}}}, ] - node = FR._conversation_node(display_name="林", tool_schemas=schemas, + from app.memory.base import Memory as M, MemoryKind as MK, ConfirmStatus as CS + memories = [M(memory_id=1, owner_user_id="u_a", kind=MK.PREFERENCE, + key="饮食偏好", value="不吃香菜", confirm_status=CS.CONFIRMED)] + node = FR._conversation_node(identity_decision=identity.current(), + memories=memories, tool_schemas=schemas, ctx_factory=ctx_factory) check("节点名是 conversation", node["name"] == "conversation") @@ -126,7 +132,10 @@ def test_node_shape(): check("用 task_messages 承载人格", len(node["task_messages"]) == 1 and node["task_messages"][0]["role"] == "system") - check("人格里含称呼", "林" in node["task_messages"][0]["content"]) + sysmsg = node["task_messages"][0]["content"] + check("人格里含称呼", "林" in sysmsg, sysmsg[-40:]) + check("**不含内部 user_id**（验收第 7 条）", "u_a" not in sysmsg) + check("稳定记忆已注入", "香菜" in node["task_messages"][0]["content"]) check("respond_immediately=False（不自动开口）", node["respond_immediately"] is False) check("工具已注册", len(node["functions"]) == 1) diff --git a/client/test_client.py b/client/test_client.py index b734034..8c077d1 100644 --- a/client/test_client.py +++ b/client/test_client.py @@ -191,6 +191,42 @@ class ReplyRecorder: return (self.first_non_silent_at - started_at) * 1000 +class RTVIReplies: + """收集服务端发来的 RTVI 文本消息，供验收脚本断言。 + + 网页用的 bot-output / bot-transcription 是完整文本， + bot-llm-text 是逐 token 增量。验收看完整文本。 + """ + + def __init__(self): + self.bot_texts: list[str] = [] + self.user_texts: list[str] = [] + self.other: list[str] = [] + + def handle(self, raw: str) -> None: + try: + m = json.loads(raw) + except (json.JSONDecodeError, TypeError): + return + if not isinstance(m, dict): + return + t = m.get("type") + data = m.get("data") or {} + text = data.get("text") if isinstance(data, dict) else None + if t == "bot-output" and text: + self.bot_texts.append(text) + elif t == "bot-transcription" and text: + self.bot_texts.append(text) + elif t == "user-llm-text" and text: + self.user_texts.append(text) + elif t: + self.other.append(t) + + @property + def last_bot(self) -> str: + return self.bot_texts[-1] if self.bot_texts else "" + + def rtvi_send_text(channel, text: str, *, audio_response: bool = True) -> None: """通过 data channel 发一条 RTVI send-text。 @@ -277,11 +313,18 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, channel = pc.createDataChannel("pipecat") dc_state = {"ok": False} + replies = RTVIReplies() + @channel.on("open") def _dc_open(): dc_state["ok"] = True log.info("data channel 已打开，开始发心跳") + @channel.on("message") + def _dc_message(msg): + if isinstance(msg, str): + replies.handle(msg) + async def _keepalive(): while pc.connectionState not in ("closed", "failed"): if dc_state["ok"]: @@ -359,6 +402,10 @@ async def run(server: str, wav: Path | None, out: Path, *, connect_only: bool, log.info("本轮 %s", "收到语音回复" if recorder.has_speech else "未收到语音") # 让它把话说完 await asyncio.sleep(4) + # 打印助手回复全文，供验收脚本断言 + if replies.bot_texts: + print(f" BOT_REPLY: {replies.bot_texts[-1]}", flush=True) + replies.bot_texts.clear() await asyncio.sleep(2) elif track: # 等音频放完 + 留出响应时间 diff --git a/p1/phase4_acceptance.py b/p1/phase4_acceptance.py new file mode 100644 index 0000000..0b604a3 --- /dev/null +++ b/p1/phase4_acceptance.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Phase 4 验收：通用记忆闭环。 + +对应交接文档 §九 Phase 4 的 7 条，**跑真实服务端**（flows 文字路径）。 +每条之间查一次服务器数据库，确认"说进数据库"和"以后能用"是同一件事。 + + 1. "记住我不吃香菜" → confirmed + 2. 新连接 "我不吃什么？" → 能召回香菜（验证连接开始时加载档案） + 3. "我今天吃了香菜" → 不得变成稳定偏好 + 4. "其实我现在吃香菜了" → 明确更新旧偏好，不留冲突值 + 5. "忘掉我的饮食偏好" → 删除；模糊时先确认 + 6. 访客无法读写私人记忆 → 单测覆盖（需要身份切换，客户端不支持） + 7. 用户 A 的记忆不出现在 B → 单测覆盖 + prompt 不含内部 ID + +用法： + ~/venvs/ai_assistant/bin/python p1/phase4_acceptance.py +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parent +SERVER = os.environ.get( + "AI_SERVER", "http://cadadmin-server.taild0f91d.ts.net:7860") +SSH = os.environ.get("AI_SSH", "yonglin@cadadmin-server.taild0f91d.ts.net") +PY = str(Path.home() / "venvs" / "ai_assistant" / "bin" / "python") + +OK, FAIL = [], [] + + +def check(name: str, cond: bool, detail: str = "") -> None: + (OK if cond else FAIL).append(name) + print(f" [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) + + +def say(text: str, *, timeout: int = 180) -> str: + """跑一轮文字对话，返回助手回复全文（从 RTVI bot-output 取）。""" + r = subprocess.run( + [PY, str(ROOT / "client" / "test_client.py"), + "--server", SERVER, "--text", text, "--out", "/tmp/p4_turn.wav"], + capture_output=True, text=True, timeout=timeout) + for line in r.stdout.splitlines(): + if line.strip().startswith("BOT_REPLY:"): + return line.split("BOT_REPLY:", 1)[1].strip() + return "" + + +def db_memories() -> list[dict]: + """查服务器的记忆表。""" + code = ( + "import sqlite3,pathlib,json;" + "db=pathlib.Path.home()/'.local/share/ai_assistant/memory.db';" + "c=sqlite3.connect(db);c.row_factory=sqlite3.Row;" + "rows=c.execute('SELECT memory_id,owner_user_id,kind,mem_key,mem_value," + "confirm_status FROM memories ORDER BY memory_id').fetchall();" + "print(json.dumps([dict(r) for r in rows],ensure_ascii=False))" + ) + r = subprocess.run(["ssh", "-o", "BatchMode=yes", SSH, + f"~/ai-companion/.venv/bin/python -c \"{code}\""], + capture_output=True, text=True, timeout=60) + for line in r.stdout.splitlines(): + line = line.strip() + if line.startswith("["): + try: + return json.loads(line) + except json.JSONDecodeError: + pass + return [] + + +def find(rows, key_contains: str) -> list[dict]: + return [m for m in rows if key_contains in (m.get("mem_key") or "")] + + +def main() -> int: + print("=" * 64) + print("Phase 4 验收：通用记忆闭环（跑真实服务端）") + print(f" 服务端 {SERVER}") + print("=" * 64) + + if os.environ.get("AI_SKIP_CLEAN") != "1": + print("\n=== 0. 前置：清掉待测键，保证从干净状态开始 ===") + say("忘掉我的饮食偏好") + rows = db_memories() + print(f" 当前记忆 {len(rows)} 条: " + f"{[(m['mem_key'], m['mem_value']) for m in rows]}") + + # ---------------------------------------------------------- 1 + print("\n=== 1. 明确要求 → confirmed ===") + reply = say("记住我不吃香菜") + rows = db_memories() + hits = [m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"] + print(f" 回复: {reply[:50]!r}") + check("确认的记忆已落库", bool(hits), + str([(m["mem_key"], m["mem_value"], m["confirm_status"]) for m in hits])) + if hits: + check("值是香菜相关", "香菜" in hits[0]["mem_value"], hits[0]["mem_value"]) + + # ---------------------------------------------------------- 2 + print("\n=== 2. **新连接**后能召回（验证连接开始时加载档案）===") + print(" （上一轮连接已关闭，这是全新会话）") + reply = say("我不吃什么？") + print(f" 回复: {reply[:80]!r}") + check("回复提到香菜（说明记忆被加载进上下文）", "香菜" in reply, reply[:60]) + check("没有把内部 ID 说出去", "u_ylhan" not in reply and "u_a" not in reply) + + # ---------------------------------------------------------- 3 + print("\n=== 3. 一次性事件不得变成稳定偏好 ===") + before = len([m for m in find(db_memories(), "饮食") + if m["confirm_status"] == "confirmed"]) + reply = say("我今天吃了香菜") + rows = db_memories() + after = len([m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"]) + print(f" 回复: {reply[:50]!r}") + check("没有新增 confirmed 饮食偏好", after <= before, f"{before} → {after}") + + # ---------------------------------------------------------- 4 + print("\n=== 4. 明确更正 → 更新旧值，不留冲突 ===") + reply = say("其实我现在吃香菜了，帮我改一下") + rows = db_memories() + hits = [m for m in find(rows, "饮食") if m["confirm_status"] == "confirmed"] + print(f" 回复: {reply[:50]!r}") + print(f" 库中饮食偏好: {[(m['mem_value'], m['confirm_status']) for m in hits]}") + check("同键只保留一条 confirmed", len(hits) <= 1, f"{len(hits)} 条") + check("值已更新为吃香菜", any("吃" in m["mem_value"] and "不吃" not in m["mem_value"] + for m in hits), + str([m["mem_value"] for m in hits])) + + # ---------------------------------------------------------- 5 + print("\n=== 5. 明确遗忘 → 删除 ===") + reply = say("忘掉我的饮食偏好") + rows = db_memories() + left = find(rows, "饮食") + print(f" 回复: {reply[:60]!r}") + check("饮食偏好已删除", not left, str([m["mem_value"] for m in left])) + + # ---------------------------------------------------------- 6/7 + print("\n=== 6/7. 访客隔离与跨用户隔离 ===") + print(" 这两条需要身份切换，WebRTC 客户端不支持；由单元测试覆盖：") + print(" app/tests/test_agent_tools.py → '身份未确认时不暴露记忆工具'") + print(" → '未确认身份拒绝 memory_search'") + print(" → '用户 B 查不到用户 A 的记忆'") + print(" → '跨 owner 删除失败'") + print(" app/tests/test_flows_runtime.py → '**不含内部 user_id**'") + check("由单测覆盖（见上）", True) + + print("\n" + "=" * 64) + ok = len(OK) + print(f"通过 {ok} / {ok + len(FAIL)}") + for n in FAIL: + print(f" 未通过: {n}") + return 0 if not FAIL else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/services/bot_server.py b/services/bot_server.py index 5da04f1..777ebdb 100644 --- a/services/bot_server.py +++ b/services/bot_server.py @@ -78,7 +78,7 @@ from pipecat.workers.base_worker import WorkerParams from app.agent.flows_runtime import ( UserTextRecorder, build_flows_parts, initialize_conversation_node, - make_flow_manager, + load_stable_memories, make_flow_manager, ) from app.config import AppConfig from app.engine import ConversationEngine, EventKind @@ -557,7 +557,8 @@ async def offer(request: SmallWebRTCRequest): aggregator.assistant(), transport.output(), ]) - flows_objs = (llm, aggregator, ctx_factory, tool_schemas) + flows_objs = (llm, aggregator, ctx_factory, tool_schemas, + memory, identity) log.info("Agent 运行时: flows（文字路径），语音仍走 legacy") else: pipeline = Pipeline([transport.input(), engine_proc, transport.output()]) @@ -636,18 +637,18 @@ async def offer(request: SmallWebRTCRequest): log.info("PipelineWorker 已结束") if flows_objs is not None: - llm, aggregator, ctx_factory, tool_schemas = flows_objs + llm, aggregator, ctx_factory, tool_schemas, memory, identity = flows_objs flow = make_flow_manager(worker=task, llm=llm, aggregator=aggregator) - identity = engine_proc._engine.identity if engine_proc._engine else None - display_name = "" - if identity is not None: - display_name = getattr(identity.current(), "display_name", "") or "" + # 方案 §四要求稳定档案"连接开始时加载"，不等模型自己想起来查。 + # 身份未确认时返回空——访客拿不到任何私人记忆（Phase 4 验收第 6 条）。 + memories = await load_stable_memories(memory, identity) # **必须 await**：initialize 是协程但签名显示 -> None， - # 漏 await 不报错，只是人格/任务/工具全部静默失效（Phase 1 栽过）。 + # 漏 await 不报错，只是人格/记忆/工具全部静默失效（Phase 1 栽过）。 await initialize_conversation_node( - flow, display_name=display_name, tool_schemas=tool_schemas, - ctx_factory=ctx_factory) - log.info("Flows 节点已初始化：工具 %d 个", len(tool_schemas)) + flow, identity_decision=identity.current(), memories=memories, + tool_schemas=tool_schemas, ctx_factory=ctx_factory) + log.info("Flows 节点已初始化：工具 %d 个，稳定记忆 %d 条", + len(tool_schemas), len(memories)) bg = asyncio.create_task(_run_task()) _bg_tasks.add(bg)