diff --git a/app/agent/__init__.py b/app/agent/__init__.py new file mode 100644 index 0000000..9390a90 --- /dev/null +++ b/app/agent/__init__.py @@ -0,0 +1,35 @@ +"""Agent 层：工具、记忆策略、Flows 装配。 + +Phase 2 起建立。**两条运行路径共用这一层**： + + - legacy：`ConversationEngine` 的手写工具循环 + - flows：Pipecat Flows（Phase 3 接入，用 `AI_AGENT_RUNTIME` 切换） + +分层意图： + + tools.py 工具 schema 与执行器（身份由 ToolContext 注入） + memory_policy.py 记忆写入/删除的裁决（模型只提议，策略决定） + flows_runtime.py FlowManager 装配（Phase 3） + +策略层是本层最重要的部分：交接文档要求 +"不要仅依赖'请不要误记'的 system prompt"， +所以这些判断必须是可测试的代码，而不是提示词里的请求。 +""" + +from .memory_policy import ( + MemoryDecision, + PolicyResult, + decide_forget, + decide_remember, +) +from .tools import ( + ToolContext, + ToolOutcome, + available_schemas, + execute, +) + +__all__ = [ + "ToolContext", "ToolOutcome", "execute", "available_schemas", + "MemoryDecision", "PolicyResult", "decide_remember", "decide_forget", +] diff --git a/app/agent/llm_service.py b/app/agent/llm_service.py new file mode 100644 index 0000000..48e54fa --- /dev/null +++ b/app/agent/llm_service.py @@ -0,0 +1,59 @@ +"""Pipecat LLMService 工厂。 + +Phase 1 spike 的选型结论固化在这里：**复用 Pipecat 自带的 `DeepSeekLLMService`**， +不自写 `FallbackProvider → LLMService` 适配器。 + +理由（详见笔记库 `P1-Flows兼容性Spike结论.md`）： + + 1. 它自带 `DeepSeekLLMAdapter`，处理"思考模式下必须把 reasoning_content + 原样回传"这个约束——**正是本项目 P0 阶段踩过的坑**，框架已经做了。 + 2. `thinking` 默认就是 disabled，与本项目既有结论一致。 + 3. 支持自定义 `base_url`，将来要接本地 vLLM 也能用。 + 4. Flows 硬要求 `LLMService` 实例，而 `pipecat.flows` 的 `LLMAdapter` + **只管摘要生成**，不是 provider 适配器——想保留 `FallbackProvider` + 必须自己写 `LLMService` 子类，成本不划算。 + +**代价**：失去 `FallbackProvider` 的本地 vLLM 降级。当前本机 vLLM 已停用， +fallback 实际没有可用目标，所以先只用 DeepSeek。将来若需双后端， +Pipecat 有 `LLMSwitcher` 可用，不必自写。 +""" + +from __future__ import annotations + +import logging + +from ..config import LLMConfig + +log = logging.getLogger(__name__) + + +def create_llm_service(cfg: LLMConfig): + """按配置造 Pipecat LLMService。 + + 目前只支持 DeepSeek（OpenAI 兼容）。其它后端返回 None，调用方应当 + 回退到 legacy 路径，而不是带着一个坏 service 继续。 + """ + if cfg.name != "deepseek": + log.warning("Flows 路径暂不支持后端 %r，请继续使用 legacy", cfg.name) + return None + + from pipecat.services.deepseek.llm import DeepSeekLLMService + + # **必须显式覆盖模型名。** + # Pipecat 的默认是 `deepseek-v4-flash`，该模型不存在，用默认值会 400。 + # 见 P0-DeepSeek验证报告.md：真实清单只有 deepseek-flash 与 deepseek-v4-pro。 + settings = DeepSeekLLMService.Settings( + model=cfg.model, + # 显式关闭思考。Pipecat 的默认恰好也是 disabled，但写出来更安全： + # 将来默认值变化时不会悄悄改变行为，也不至于把思考内容送进 TTS。 + thinking=DeepSeekLLMService.ThinkingConfig(type="disabled"), + max_tokens=cfg.max_tokens, + ) + return DeepSeekLLMService( + api_key=cfg.api_key, + base_url=cfg.base_url, + settings=settings, + ) + + +__all__ = ["create_llm_service"] diff --git a/app/agent/memory_policy.py b/app/agent/memory_policy.py new file mode 100644 index 0000000..a0216ec --- /dev/null +++ b/app/agent/memory_policy.py @@ -0,0 +1,269 @@ +"""记忆写入/删除的策略层。 + +**为什么需要独立的一层**：交接文档明确要求 +"不要仅依赖'请不要误记'的 system prompt"。 + +模型可以提议动作，**这一层拥有最终决定权**。理由： + + - system prompt 是软的，模型会漂移；策略是硬的，可测试、可回归 + - "我今天吃了香菜"不能被记成"用户不吃香菜"这种稳定偏好—— + 这不是措辞问题，是**语义等级**问题，必须在代码里判定 + - 身份未确认时不得碰任何私人记忆，这也不能交给模型自觉 + +输入：用户原话、工具请求参数、身份确认状态、已有同键记忆。 +输出：allow_confirmed / allow_candidate / needs_confirmation / reject。 + +**本模块不依赖 LLM、不依赖网络、不依赖真实存储**，可以纯单元测试。 +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from enum import Enum + +from ..memory.base import ConfirmStatus, Memory, MemoryKind + + +class MemoryDecision(str, Enum): + """策略层的裁决。""" + ALLOW_CONFIRMED = "allow_confirmed" # 用户明确要求 → 写 confirmed + ALLOW_CANDIDATE = "allow_candidate" # 只是观察/推断 → 写 candidate + NEEDS_CONFIRMATION = "needs_confirmation" # 目标不明确 → 转入确认节点 + REJECT = "reject" # 不允许（越权、身份未确认等） + + +@dataclass +class PolicyResult: + decision: MemoryDecision + reason: str + #: 规范化后的写入内容。策略层可以改写模型给的 kind/key/value。 + kind: MemoryKind = MemoryKind.FACT + key: str = "" + value: str = "" + #: NEEDS_CONFIRMATION 时，候选目标（供确认节点展示） + candidates: list[Memory] = field(default_factory=list) + #: REJECT/NEEDS_CONFIRMATION 时给模型看的话（不冒充已执行） + message: str = "" + + @property + def confirm_status(self) -> ConfirmStatus: + return (ConfirmStatus.CONFIRMED + if self.decision is MemoryDecision.ALLOW_CONFIRMED + else ConfirmStatus.CANDIDATE) + + +# ---------------------------------------------------------------- 显式意图 + +#: 用户明确要求"记住"的说法。命中才允许写 confirmed。 +#: 刻意保守：宁可多判成 candidate（不参与召回），也不要误记成稳定事实。 +EXPLICIT_REMEMBER_PATTERNS = ( + "记住", "记一下", "记下", "帮我记", "帮我记住", + "别忘了", "别忘记", "不要忘", + "以后都", "以后要", "以后别", "今后都", + "我的偏好是", "我习惯", "我通常", "我总是", +) + +#: 明确表示"这是这一次的事"的说法。命中则**降级**为 candidate， +#: 即使句中同时出现"记住"之类的词。 +#: 例："记住我今天吃了香菜"——语义是"记下今天这件事"，不是稳定偏好。 +#: 注意不要收太宽的词。比如"先"看着像临时，但"我习惯先喝汤"是稳定偏好， +#: 收进来会造成漏判（把偏好降级成候选）。宁可少收。 +EPISODIC_MARKERS = ( + "今天", "昨天", "刚才", "刚刚", "这次", "这一回", "暂时", "目前", + "这会儿", "眼下", +) + +#: 主语是"今天/这次"这类时间词时，更可能是事件而非偏好 +_TIME_LEAD_RE = re.compile(r"^(今天|昨天|刚才|刚刚|这次|这会|这回|现在|目前)") + + +def has_explicit_remember_intent(user_text: str) -> bool: + """用户是否明确要求记住（而不是模型自己觉得值得记）。 + + 注意：本函数调用下面定义的 `looks_episodic`（Python 晚绑定，没问题）。 + """ + t = (user_text or "").strip() + if not t: + return False + if not any(p in t for p in EXPLICIT_REMEMBER_PATTERNS): + return False + # 明确是"这一次"的事 → 不算稳定偏好意图。 + # **必须复用 looks_episodic**：早先这里只检查句首时间词， + # 而"记住我今天吃了香菜"的时间词在中间，导致漏判（测试逮到的）。 + if looks_episodic(t): + return False + return True + + +def looks_episodic(user_text: str) -> bool: + """是否是"这一次发生的事"而非稳定事实。""" + t = (user_text or "").strip() + if not t: + return False + if _TIME_LEAD_RE.search(t): + return True + return any(m in t for m in EPISODIC_MARKERS) + + +# ---------------------------------------------------------------- 越权检查 + +#: 模型**不得**传入的字段——身份与授权范围必须由服务端注入。 +FORBIDDEN_ARG_KEYS = ( + "owner_user_id", "user_id", "owner", "uid", + "scope", "confirm_status", "trust_level", + "session_id", "turn_id", +) + + +def find_forbidden_args(args: dict) -> list[str]: + """返回模型试图传入的越权字段名。""" + if not isinstance(args, dict): + return [] + lower = {str(k).lower() for k in args.keys()} + return [k for k in FORBIDDEN_ARG_KEYS if k.lower() in lower] + + +# ---------------------------------------------------------------- 写入裁决 + +def decide_remember( + *, + user_text: str, + args: dict, + identity_confirmed: bool, + model_claims_explicit: bool = False, +) -> PolicyResult: + """裁决一次 memory_remember 请求。 + + 参数： + user_text: 本轮用户原话（策略的主要依据） + args: 模型给的工具参数（含 kind/key/value） + identity_confirmed: 当前身份是否已确认 + model_claims_explicit: 模型自称"用户明确要求记住"。 + 仅作参考，**不采信**——必须由 user_text 判定。 + """ + # 1. 身份未确认：一律拒绝，不碰任何私人记忆 + if not identity_confirmed: + return PolicyResult( + decision=MemoryDecision.REJECT, + reason="身份未确认", + message="现在还不能记录私人信息。请先确认身份。", + ) + + # 2. 越权字段：拒绝（不静默忽略，避免模型以为生效了） + forbidden = find_forbidden_args(args) + if forbidden: + return PolicyResult( + decision=MemoryDecision.REJECT, + reason=f"模型传入越权字段 {forbidden}", + message=f"参数不合法：不接受 {', '.join(forbidden)}。", + ) + + value = str(args.get("value") or "").strip() + if not value: + return PolicyResult( + decision=MemoryDecision.REJECT, + reason="value 为空", + message="没有可记录的内容。", + ) + + kind = _coerce_kind(args.get("kind")) + key = str(args.get("key") or "").strip() + + # 3. 判定语义等级。 + # 注意：以 user_text 为准，不采信 model_claims_explicit。 + explicit = has_explicit_remember_intent(user_text) + episodic = looks_episodic(user_text) + + if explicit and not episodic: + return PolicyResult( + decision=MemoryDecision.ALLOW_CONFIRMED, + reason="用户明确要求记住", + kind=kind, key=key, value=value, + ) + + # 4. 其余一律 candidate：模型推断、总结、或"这一次"的事件 + reason = "属于一次性事件" if episodic else "非用户明确要求，仅作观察" + return PolicyResult( + decision=MemoryDecision.ALLOW_CANDIDATE, + reason=reason, + kind=kind, key=key, value=value, + ) + + +def _coerce_kind(raw) -> MemoryKind: + """把模型给的 kind 收敛到枚举；不认识的一律当 FACT。""" + try: + return MemoryKind(str(raw)) + except (ValueError, TypeError): + return MemoryKind.FACT + + +# ---------------------------------------------------------------- 删除裁决 + +#: 模糊删除的说法——命中则必须先确认，不得猜目标 +AMBIGUOUS_FORGET_MARKERS = ( + "所有", "全部", "都删", "清空", "关于我的", "我的资料", "我的信息", + "这些", "那些", "乱七八糟", +) + + +def decide_forget( + *, + user_text: str, + matches: list[Memory], + identity_confirmed: bool, +) -> PolicyResult: + """裁决一次 memory_forget 请求。 + + 规则（交接文档 §八）： + - 精确匹配一条 → 返回待删除摘要，可执行 + - 多条或模糊 → 进入确认节点，**不得猜测删除目标** + - 删除必须同时匹配 owner（由存储层保证，这里只判数量与措辞） + """ + if not identity_confirmed: + return PolicyResult( + decision=MemoryDecision.REJECT, + reason="身份未确认", + message="现在不能操作私人记忆。", + ) + + if not matches: + return PolicyResult( + decision=MemoryDecision.REJECT, + reason="没有匹配的记忆", + message="没有找到相关的记忆。", + ) + + t = (user_text or "").strip() + ambiguous_wording = any(m in t for m in AMBIGUOUS_FORGET_MARKERS) + + if len(matches) == 1 and not ambiguous_wording: + m = matches[0] + return PolicyResult( + decision=MemoryDecision.ALLOW_CONFIRMED, + reason="精确匹配一条", + kind=m.kind, key=m.key, value=m.value, + candidates=[m], + message=f"将删除「{m.text}」。", + ) + + # 多条 / 模糊 → 必须确认 + return PolicyResult( + decision=MemoryDecision.NEEDS_CONFIRMATION, + reason=f"匹配到 {len(matches)} 条" + ("，且措辞模糊" if ambiguous_wording else ""), + candidates=list(matches), + message=( + f"找到 {len(matches)} 条相关记忆，需要你确认删哪一条：\n" + + "\n".join(f" {i+1}. {m.text}" for i, m in enumerate(matches[:5])) + ), + ) + + +__all__ = [ + "MemoryDecision", "PolicyResult", + "decide_remember", "decide_forget", + "has_explicit_remember_intent", "looks_episodic", + "find_forbidden_args", + "EXPLICIT_REMEMBER_PATTERNS", "EPISODIC_MARKERS", "FORBIDDEN_ARG_KEYS", +] diff --git a/app/agent/tools.py b/app/agent/tools.py new file mode 100644 index 0000000..dd54c28 --- /dev/null +++ b/app/agent/tools.py @@ -0,0 +1,415 @@ +"""工具层。 + +**两条路径共用**：legacy 的 `ConversationEngine` 手写循环和 Flows 都用这里的 +同一套 schema 与执行器。交接文档 Phase 2 的完成标准就是这个： +"工具层可被 legacy 和 flows 两条路径复用"。 + +设计要点： + + * **身份由服务端注入，模型不得传。** `ToolContext.owner_user_id` 来自会话， + 不在工具参数里。模型若试图传 `owner_user_id`/`scope`，策略层直接拒绝。 + * **记忆写入必须过策略层**（`memory_policy`），策略拥有最终决定权。 + 模型只是提议者。 + * **工具失败不拖垮对话。** 任何异常都转成给模型的文本结果，本轮继续。 + * **不把"请求已收到"说成"已经做到"。** 删除类操作要如实返回 + 已删除 / 没有找到 / 需要确认。 + +依赖注入：`ToolContext` 带 `store` 与 `search`，所以单元测试可以传 fake store， +不需要真实 SQLite 或网络。 +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field +from typing import Any, Awaitable, Callable + +from ..memory.base import ConfirmStatus, Memory, MemoryKind, MemoryStore, Scope +from ..search import WEB_SEARCH_TOOL, SearchError, SearXNGSearchClient +from .memory_policy import ( + MemoryDecision, + decide_forget, + decide_remember, + find_forbidden_args, +) + +log = logging.getLogger(__name__) + + +# ---------------------------------------------------------------- 上下文 + +@dataclass +class ToolContext: + """一次工具调用所需的全部环境。**身份在这里，不在工具参数里。**""" + + owner_user_id: str + session_id: str + turn_id: str + identity_confirmed: bool + user_text: str = "" + store: MemoryStore | None = None + search: SearXNGSearchClient | None = None + #: 记忆检索的条数与超时上限（沿用全局配置） + memory_limit: int = 5 + memory_timeout_s: float = 0.3 + + +@dataclass +class ToolOutcome: + """工具执行结果。`content` 是回灌给模型的文本。""" + content: str + ok: bool = True + meta: dict[str, Any] = field(default_factory=dict) + + +# ---------------------------------------------------------------- 工具 schema + +MEMORY_SEARCH_TOOL = { + "type": "function", + "function": { + "name": "memory_search", + "description": ( + "在本用户已确认的长期记忆中检索。当用户问起自己的偏好、习惯或" + "过去说过的事，而你手上没有这些信息时使用。只能查到已确认的记忆。" + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "要查的内容"}, + }, + "required": ["query"], + "additionalProperties": False, + }, + }, +} + +MEMORY_REMEMBER_TOOL = { + "type": "function", + "function": { + "name": "memory_remember", + "description": ( + "长期记住一条关于用户的稳定信息（偏好、习惯、固定事实）。" + "**只在用户明确要求记住时调用**；你自行推断出的信息会被降级为" + "候选，不会立刻生效。" + ), + "parameters": { + "type": "object", + "properties": { + "kind": { + "type": "string", + "enum": ["preference", "fact", "task"], + "description": "preference=稳定偏好，fact=事实，task=临时事项", + }, + "key": { + "type": "string", + "description": "简短键名，如“饮食偏好”。同一键的新值会覆盖旧值", + }, + "value": {"type": "string", "description": "要记住的内容"}, + }, + "required": ["kind", "value"], + "additionalProperties": False, + }, + }, +} + +MEMORY_FORGET_TOOL = { + "type": "function", + "function": { + "name": "memory_forget", + "description": ( + "删除关于用户的记忆。目标不明确时会被要求先确认，" + "**不要猜测用户想删哪一条**。" + ), + "parameters": { + "type": "object", + "properties": { + "query": {"type": "string", "description": "要删除的记忆内容或关键词"}, + }, + "required": ["query"], + "additionalProperties": False, + }, + }, +} + +MEMORY_UPDATE_TOOL = { + "type": "function", + "function": { + "name": "memory_update", + "description": ( + "更正已记住的信息。用户说“其实……”“不对，是……”这类纠正时使用。" + "会保留来源与更新时间。" + ), + "parameters": { + "type": "object", + "properties": { + "key": {"type": "string", "description": "要更正的记忆键名"}, + "value": {"type": "string", "description": "更正后的内容"}, + }, + "required": ["key", "value"], + "additionalProperties": False, + }, + }, +} + +ALL_MEMORY_TOOLS = [ + MEMORY_SEARCH_TOOL, + MEMORY_REMEMBER_TOOL, + MEMORY_FORGET_TOOL, + MEMORY_UPDATE_TOOL, +] + + +def available_schemas(ctx: ToolContext) -> list[dict]: + """当前上下文下可用的工具。 + + 身份未确认时**不暴露任何记忆工具**——比暴露后再拒绝更干净： + 模型根本看不到这些能力，就不会尝试。 + """ + schemas: list[dict] = [] + if ctx.search is not None: + schemas.append(WEB_SEARCH_TOOL) + if ctx.identity_confirmed and ctx.store is not None: + schemas.extend(ALL_MEMORY_TOOLS) + return schemas + + +# ---------------------------------------------------------------- 执行入口 + +async def execute(name: str, arguments: dict, ctx: ToolContext) -> ToolOutcome: + """按名字分发。未知工具/参数错误都返回可读文本，不抛异常。""" + fn = _HANDLERS.get(name) + if fn is None: + return ToolOutcome( + content=f"没有名为 {name} 的工具。请不要声称已经执行。", + ok=False, + ) + if not isinstance(arguments, dict): + return ToolOutcome(content="工具参数不是合法的对象。", ok=False) + + # 越权字段在这里统一拦一次（记忆工具内部还会再过策略层） + forbidden = find_forbidden_args(arguments) + if forbidden: + return ToolOutcome( + content=f"参数不合法：不接受 {', '.join(forbidden)}。", + ok=False, + ) + + try: + return await fn(arguments, ctx) + except Exception as e: # noqa: BLE001 — 工具失败不能拖垮对话 + log.exception("工具 %s 执行异常", name) + return ToolOutcome(content=f"工具执行失败：{type(e).__name__}。", ok=False) + + +# ---------------------------------------------------------------- web_search + +async def _web_search(args: dict, ctx: ToolContext) -> ToolOutcome: + if ctx.search is None: + return ToolOutcome(content="搜索功能当前不可用。", ok=False) + query = args.get("query") + if not isinstance(query, str) or not query.strip(): + return ToolOutcome(content="搜索失败：缺少有效的 query 参数。", ok=False) + + t0 = time.perf_counter() + try: + results = await ctx.search.search(query) + except SearchError as exc: + log.warning("联网搜索失败：%s", exc) + return ToolOutcome( + content=f"搜索失败：{exc}。请明确告知用户暂时无法取得实时资料。", + ok=False, + ) + ms = (time.perf_counter() - t0) * 1000 + log.info("联网搜索 %r：%d 条，%.0fms", query[:60], len(results), ms) + return ToolOutcome( + content=ctx.search.for_model(query, results), + meta={"query": query, "count": len(results), "ms": ms, + "urls": [getattr(r, "url", "") for r in results]}, + ) + + +# ---------------------------------------------------------------- 记忆工具 + +async def _memory_search(args: dict, ctx: ToolContext) -> ToolOutcome: + if not ctx.identity_confirmed or ctx.store is None: + # 交接文档：不确认身份时返回不可用，不搜索任何私人记忆 + return ToolOutcome( + content="现在还不能查询私人记忆。身份未确认。", ok=False) + + query = str(args.get("query") or "").strip() + if not query: + return ToolOutcome(content="查询内容为空。", ok=False) + + try: + # 只返回 confirmed、未过期、owner 匹配的记忆（存储层保证后两项） + memories = await ctx.store.retrieve( + ctx.owner_user_id, query, + limit=ctx.memory_limit, + include_candidates=False, + scopes=[Scope.PRIVATE, Scope.SHARED], + ) + except Exception as e: # noqa: BLE001 + log.warning("记忆检索失败：%s", e) + return ToolOutcome(content="记忆检索暂时不可用。", ok=False) + + if not memories: + return ToolOutcome(content="没有找到相关记忆。", meta={"count": 0}) + lines = "\n".join(f"- {m.text}" for m in memories) + return ToolOutcome( + content=f"找到 {len(memories)} 条：\n{lines}", + meta={"count": len(memories)}, + ) + + +async def _memory_remember(args: dict, ctx: ToolContext) -> ToolOutcome: + if ctx.store is None: + return ToolOutcome(content="记忆功能当前不可用。", ok=False) + + result = decide_remember( + user_text=ctx.user_text, + args=args, + identity_confirmed=ctx.identity_confirmed, + ) + + if result.decision is MemoryDecision.REJECT: + return ToolOutcome(content=result.message or "不允许写入。", ok=False) + + mem = Memory( + owner_user_id=ctx.owner_user_id, # ← 服务端注入，非模型传入 + scope=Scope.PRIVATE, # ← 默认 private，模型不得扩大 + kind=result.kind, + key=result.key, + value=result.value, + confirm_status=result.confirm_status, + source_session_id=ctx.session_id, + source_turn_id=ctx.turn_id, + ) + try: + memory_id = await ctx.store.remember(mem) + except Exception as e: # noqa: BLE001 + log.warning("写入记忆失败：%s", e) + return ToolOutcome(content="记忆写入暂时失败。", ok=False) + + if result.decision is MemoryDecision.ALLOW_CONFIRMED: + return ToolOutcome( + content=f"已记住：{result.value}", + meta={"memory_id": memory_id, "status": "confirmed"}, + ) + # 候选：**如实告诉模型它没有立刻生效**，否则模型会对用户说"记住了" + return ToolOutcome( + content=( + f"已作为候选记录（{result.reason}），尚未生效。" + f"如果用户确实要求长期记住，请让用户明确说一次。" + ), + meta={"memory_id": memory_id, "status": "candidate"}, + ) + + +async def _memory_forget(args: dict, ctx: ToolContext) -> ToolOutcome: + if not ctx.identity_confirmed or ctx.store is None: + return ToolOutcome(content="现在不能操作私人记忆。", ok=False) + + query = str(args.get("query") or "").strip() + if not query: + return ToolOutcome(content="没有指定要删除的内容。", ok=False) + + try: + matches = await ctx.store.retrieve( + ctx.owner_user_id, query, limit=6, include_candidates=True, + scopes=[Scope.PRIVATE, Scope.SHARED]) + except Exception as e: # noqa: BLE001 + log.warning("删除前检索失败：%s", e) + return ToolOutcome(content="记忆检索暂时不可用，未执行删除。", ok=False) + + result = decide_forget( + user_text=ctx.user_text, matches=matches, + identity_confirmed=ctx.identity_confirmed) + + if result.decision is MemoryDecision.REJECT: + return ToolOutcome(content=result.message, ok=False) + + if result.decision is MemoryDecision.NEEDS_CONFIRMATION: + # 不猜目标，交给确认流程 + return ToolOutcome( + content=result.message, + meta={"needs_confirmation": True, + "candidates": [m.memory_id for m in result.candidates]}, + ) + + target = result.candidates[0] + if target.memory_id is None: + return ToolOutcome(content="无法定位要删除的记忆。", ok=False) + try: + deleted = await ctx.store.forget(ctx.owner_user_id, target.memory_id) + except Exception as e: # noqa: BLE001 + log.warning("删除记忆失败：%s", e) + return ToolOutcome(content="删除失败。", ok=False) + + # 如实反馈，不把"请求已收到"说成"已经删除" + return ToolOutcome( + content=f"已删除：{target.text}" if deleted else "没有找到要删除的记忆。", + meta={"deleted": bool(deleted), "memory_id": target.memory_id}, + ) + + +async def _memory_update(args: dict, ctx: ToolContext) -> ToolOutcome: + if not ctx.identity_confirmed or ctx.store is None: + return ToolOutcome(content="现在不能修改私人记忆。", ok=False) + + key = str(args.get("key") or "").strip() + value = str(args.get("value") or "").strip() + if not key or not value: + return ToolOutcome(content="更正需要同时给出 key 和 value。", ok=False) + + try: + existing = await ctx.store.retrieve( + ctx.owner_user_id, key, limit=5, include_candidates=True, + scopes=[Scope.PRIVATE, Scope.SHARED]) + except Exception as e: # noqa: BLE001 + log.warning("更正前检索失败：%s", e) + return ToolOutcome(content="记忆检索暂时不可用。", ok=False) + + target = next((m for m in existing if m.key == key), None) + if target is None: + return ToolOutcome( + content=f"没有找到键为「{key}」的记忆，无法更正。", ok=False) + + # 同 key 的 confirmed 写入是幂等的：存储层按 owner+kind+key 覆盖旧值 + mem = Memory( + owner_user_id=ctx.owner_user_id, + scope=target.scope, + kind=target.kind, + key=key, + value=value, + confirm_status=ConfirmStatus.CONFIRMED, + source_session_id=ctx.session_id, + source_turn_id=ctx.turn_id, + ) + try: + await ctx.store.remember(mem) + except Exception as e: # noqa: BLE001 + log.warning("更正记忆失败：%s", e) + return ToolOutcome(content="更正失败。", ok=False) + + return ToolOutcome( + content=f"已更新「{key}」：{target.value} → {value}", + meta={"updated": True, "key": key}, + ) + + +_HANDLERS: dict[str, Callable[[dict, ToolContext], Awaitable[ToolOutcome]]] = { + "web_search": _web_search, + "memory_search": _memory_search, + "memory_remember": _memory_remember, + "memory_forget": _memory_forget, + "memory_update": _memory_update, +} + + +__all__ = [ + "ToolContext", "ToolOutcome", "execute", "available_schemas", + "WEB_SEARCH_TOOL", "ALL_MEMORY_TOOLS", + "MEMORY_SEARCH_TOOL", "MEMORY_REMEMBER_TOOL", + "MEMORY_FORGET_TOOL", "MEMORY_UPDATE_TOOL", +] diff --git a/app/engine.py b/app/engine.py index db78b7b..a9a35da 100644 --- a/app/engine.py +++ b/app/engine.py @@ -37,6 +37,7 @@ import numpy as np from .config import AppConfig from .identity.base import IdentityDecision, IdentityProvider +from .agent.tools import ToolContext, execute as execute_tool from .llm.base import ChunkKind, LLMProvider, ToolCall from .memory.base import ( ConfirmStatus, Memory, MemoryKind, MemoryStore, Scope, Turn, @@ -45,7 +46,6 @@ from .prompts import build_system_prompt, trim_history from .search import ( WEB_SEARCH_PROMPT, WEB_SEARCH_TOOL, - SearchError, SearXNGSearchClient, ) from .voice.stt import StreamingSTT @@ -250,6 +250,9 @@ class ConversationEngine: buffer = "" speech_done = False speaks = 0 + # 策略层要拿用户原话来判定"明确要求记住"还是"模型推断"， + # 所以本轮原话要挂到实例上，供 _execute_tool 构造 ToolContext。 + self._current_user_text = user_text async for ev in self._stream_and_speak(messages, metrics): if ev.kind == EventKind.RESPONSE_DELTA: reply_parts.append(ev.text) @@ -462,27 +465,30 @@ class ConversationEngine: metrics.llm_backend = getattr(self.llm, "last_backend", self.llm.name) async def _execute_tool(self, call: ToolCall, metrics: TurnMetrics) -> str: - """执行只读搜索工具；错误作为工具结果回灌，不让整轮对话中断。""" - if call.name != "web_search" or self.search is None: - return "工具不可用。请不要声称已经执行。" - query = call.parsed_arguments().get("query") - if not isinstance(query, str) or not query.strip(): - return "搜索失败：缺少有效的 query 参数。" + """执行工具。**实现已抽到 `app/agent/tools.py`，两条路径共用。** + legacy 与 Flows 必须用同一套 schema 和执行器，否则容易两处行为漂移 + （交接文档 Phase 2 的完成标准就是"工具层可被两条路径复用"）。 + 这里只负责把引擎的状态转成 ToolContext。 + """ + identity = self.identity.current() + ctx = ToolContext( + owner_user_id=identity.user_id, # ← 服务端注入，模型不得传 + session_id=self._session_id, + turn_id=f"t_{uuid.uuid4().hex[:10]}", + identity_confirmed=identity.confirmed, + user_text=getattr(self, "_current_user_text", ""), + store=self.memory, + search=self.search, + memory_limit=self.cfg.context.max_memories, + memory_timeout_s=self.cfg.context.memory_timeout_s, + ) started = time.perf_counter() - metrics.searches += 1 - try: - results = await self.search.search(query) - log.info("联网搜索 %r：返回 %d 条", query[:80], len(results)) - return self.search.for_model(query, results) - except SearchError as exc: - log.warning("联网搜索失败：%s", exc) - return f"搜索失败：{exc}。请明确告知用户暂时无法取得实时资料。" - except Exception as exc: # noqa: BLE001 — 工具失败不能拖垮对话 - log.exception("联网搜索异常：%s", exc) - return "搜索服务发生异常。请明确告知用户暂时无法取得实时资料。" - finally: + outcome = await execute_tool(call.name, call.parsed_arguments(), ctx) + if call.name == "web_search": + metrics.searches += 1 metrics.search_ms += (time.perf_counter() - started) * 1000 + return outcome.content async def _speak(self, text: str, metrics: TurnMetrics) -> AsyncIterator[Event]: """合成一段并产出可播放音频。""" diff --git a/app/tests/test_agent_tools.py b/app/tests/test_agent_tools.py new file mode 100644 index 0000000..055e481 --- /dev/null +++ b/app/tests/test_agent_tools.py @@ -0,0 +1,347 @@ +#!/usr/bin/env python3 +"""Phase 2 工具层与记忆策略测试。 + +**完全不依赖 LLM、网络或真实 SQLite**——用 fake store。 +这是交接文档 Phase 2 的完成标准之一（"策略测试不依赖真实 LLM"）。 + +覆盖交接文档 §十 的单元测试清单： + + - 工具参数 schema 与非法参数 + - 模型不能传 owner_user_id/scope 越权 + - 未确认身份拒绝 private memory + - 明确"记住"写 confirmed + - 推断信息只写 candidate + - 同 key 更新幂等 + - 模糊忘记返回 needs_confirmation + - 精确忘记只删除当前 owner + +用法： + ~/venvs/ai_assistant/bin/python app/tests/test_agent_tools.py +""" + +from __future__ import annotations + +import asyncio +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent.parent +sys.path.insert(0, str(ROOT)) + +from app.agent.memory_policy import ( # noqa: E402 + MemoryDecision, decide_forget, decide_remember, + find_forbidden_args, has_explicit_remember_intent, looks_episodic, +) +from app.agent.tools import ( # noqa: E402 + ToolContext, available_schemas, execute, +) +from app.memory.base import ( # noqa: E402 + ConfirmStatus, Memory, MemoryKind, MemoryStore, Scope, +) + +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 "")) + + +# ---------------------------------------------------------------- fake store + +class FakeStore(MemoryStore): + """内存实现。只实现测试用得到的方法，其余 raise。""" + + def __init__(self): + self.rows: list[Memory] = [] + self._next = 1 + self.calls: list[tuple[str, tuple]] = [] + + async def open(self): pass + async def close(self): pass + async def ensure_user(self, user_id, display_name="", trust_level="normal"): pass + async def set_user_display_name(self, user_id, display_name): pass + async def get_user_display_name(self, user_id): return "" + async def start_session(self, session_id, user_id): pass + async def end_session(self, session_id): pass + async def add_turn(self, turn): pass + async def purge_user(self, owner_user_id): return 0 + + async def remember(self, mem: Memory) -> int: + self.calls.append(("remember", (mem.owner_user_id, mem.key, mem.value, + mem.confirm_status.value))) + # 复刻存储层的幂等语义：同 owner+kind+key 的 confirmed 覆盖旧值 + if mem.confirm_status is ConfirmStatus.CONFIRMED and mem.key: + for r in self.rows: + if (r.owner_user_id == mem.owner_user_id and r.kind == mem.kind + and r.key == mem.key + and r.confirm_status is ConfirmStatus.CONFIRMED): + r.value = mem.value + r.updated_at = time.time() + return r.memory_id or 0 + m = Memory(**{**mem.__dict__, "memory_id": self._next}) + self._next += 1 + self.rows.append(m) + return m.memory_id + + async def forget(self, owner_user_id: str, memory_id: int) -> bool: + self.calls.append(("forget", (owner_user_id, memory_id))) + before = len(self.rows) + # 必须同时匹配 owner——只凭 id 删是越权 + self.rows = [r for r in self.rows + if not (r.memory_id == memory_id + and r.owner_user_id == owner_user_id)] + return len(self.rows) < before + + async def retrieve(self, owner_user_id, query, *, limit=5, + include_candidates=False, scopes=None): + self.calls.append(("retrieve", (owner_user_id,))) + out = [] + for r in self.rows: + if r.owner_user_id != owner_user_id: + continue + if not include_candidates and r.confirm_status is not ConfirmStatus.CONFIRMED: + continue + if query and self.relevance(query, r.text) <= 0: + continue + out.append(r) + return out[:limit] + + async def list_memories(self, owner_user_id, *, include_candidates=True): + return [r for r in self.rows if r.owner_user_id == owner_user_id] + + +def ctx(store=None, *, confirmed=True, user_text="", uid="u_a"): + return ToolContext(owner_user_id=uid, session_id="s1", turn_id="t1", + identity_confirmed=confirmed, user_text=user_text, + store=store) + + +# ---------------------------------------------------------------- 策略：写入 + +def test_policy_remember(): + print("\n=== 1. 写入裁决（策略层，无 LLM）===") + + # 明确要求 → confirmed + r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=True, + args={"kind": "preference", "key": "饮食偏好", "value": "不吃香菜"}) + check("明确“记住” → confirmed", r.decision is MemoryDecision.ALLOW_CONFIRMED, r.reason) + + # 模型推断 → candidate + r = decide_remember(user_text="我今天点了份牛肉面", identity_confirmed=True, + args={"kind": "preference", "key": "饮食偏好", "value": "喜欢牛肉面"}) + check("无明确意图 → candidate", r.decision is MemoryDecision.ALLOW_CANDIDATE, r.reason) + + # 模型自称"用户明确要求"不采信 + r = decide_remember(user_text="我今天点了份牛肉面", identity_confirmed=True, + args={"kind": "fact", "value": "喜欢牛肉面"}, + model_claims_explicit=True) + check("不采信模型自称的显式意图", r.decision is MemoryDecision.ALLOW_CANDIDATE, r.reason) + + # 身份未确认 → reject + r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=False, + args={"kind": "preference", "value": "不吃香菜"}) + check("身份未确认 → reject", r.decision is MemoryDecision.REJECT, r.reason) + + # 越权字段 → reject + r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=True, + args={"kind": "preference", "value": "x", + "owner_user_id": "u_b"}) + check("模型传 owner_user_id → reject", r.decision is MemoryDecision.REJECT, r.reason) + r = decide_remember(user_text="记住我不吃香菜", identity_confirmed=True, + args={"kind": "preference", "value": "x", "scope": "shared"}) + check("模型传 scope → reject", r.decision is MemoryDecision.REJECT, r.reason) + + # 空 value + r = decide_remember(user_text="记住", identity_confirmed=True, + args={"kind": "fact", "value": " "}) + check("空 value → reject", r.decision is MemoryDecision.REJECT, r.reason) + + +def test_episodic(): + print("\n=== 2. 事件 vs 稳定偏好（交接文档的关键验收）===") + check("“记住我不吃香菜”是明确意图", + has_explicit_remember_intent("记住我不吃香菜")) + check("“记住我今天吃了香菜”不算稳定意图（时间词在前）", + not has_explicit_remember_intent("记住我今天吃了香菜")) + check("“我今天吃了香菜”判为事件", looks_episodic("我今天吃了香菜")) + check("“我不吃香菜”不是事件", not looks_episodic("我不吃香菜")) + + # 交接文档 Phase 4 验收第 3 条：不能把一次性事件变成稳定偏好 + r = decide_remember(user_text="记住我今天吃了香菜", identity_confirmed=True, + args={"kind": "preference", "key": "饮食偏好", + "value": "吃香菜"}) + check("一次性事件降级为 candidate", + r.decision is MemoryDecision.ALLOW_CANDIDATE, r.reason) + + +def test_forbidden_args(): + print("\n=== 3. 越权字段识别 ===") + check("识别 owner_user_id", find_forbidden_args({"owner_user_id": "x"}) == ["owner_user_id"]) + check("识别 scope", find_forbidden_args({"scope": "shared"}) == ["scope"]) + check("大小写不敏感", find_forbidden_args({"Owner_User_ID": "x"}) != []) + check("正常参数不误报", find_forbidden_args({"query": "x", "value": "y"}) == []) + + +# ---------------------------------------------------------------- 策略：删除 + +def test_policy_forget(): + print("\n=== 4. 删除裁决 ===") + m1 = Memory(memory_id=1, owner_user_id="u_a", kind=MemoryKind.PREFERENCE, + key="饮食偏好", value="不吃香菜", confirm_status=ConfirmStatus.CONFIRMED) + m2 = Memory(memory_id=2, owner_user_id="u_a", kind=MemoryKind.PREFERENCE, + key="饮品偏好", value="喜欢美式", confirm_status=ConfirmStatus.CONFIRMED) + + r = decide_forget(user_text="忘掉我的饮食偏好", matches=[m1], identity_confirmed=True) + check("精确匹配一条 → 可执行", + r.decision is MemoryDecision.ALLOW_CONFIRMED, r.reason) + + r = decide_forget(user_text="忘掉我的偏好", matches=[m1, m2], identity_confirmed=True) + check("多条 → needs_confirmation", + r.decision is MemoryDecision.NEEDS_CONFIRMATION, r.reason) + check("needs_confirmation 带候选", len(r.candidates) == 2) + + r = decide_forget(user_text="忘掉我所有的记忆", matches=[m1], identity_confirmed=True) + check("模糊措辞 → needs_confirmation", + r.decision is MemoryDecision.NEEDS_CONFIRMATION, r.reason) + + r = decide_forget(user_text="忘掉我的饮食偏好", matches=[], identity_confirmed=True) + check("无匹配 → reject", r.decision is MemoryDecision.REJECT, r.reason) + + r = decide_forget(user_text="忘掉我的饮食偏好", matches=[m1], identity_confirmed=False) + check("身份未确认 → reject", r.decision is MemoryDecision.REJECT, r.reason) + + +# ---------------------------------------------------------------- 工具执行 + +async def test_tools(): + print("\n=== 5. 工具执行（fake store）===") + store = FakeStore() + + # 可用工具随身份变化 + s_off = available_schemas(ctx(store, confirmed=False)) + names_off = {t["function"]["name"] for t in s_off} + check("身份未确认时不暴露记忆工具", "memory_search" not in names_off, str(names_off)) + s_on = available_schemas(ctx(store, confirmed=True)) + names_on = {t["function"]["name"] for t in s_on} + check("身份确认后暴露记忆工具", + {"memory_search", "memory_remember", "memory_forget", "memory_update"} + <= names_on, str(names_on)) + + # 未知工具 + out = await execute("no_such_tool", {}, ctx(store)) + check("未知工具返回可读文本", not out.ok and "没有名为" in out.content) + + # 越权参数在入口就被拦 + out = await execute("memory_remember", + {"kind": "fact", "value": "x", "owner_user_id": "u_b"}, + ctx(store)) + check("入口拦越权参数", not out.ok and "不合法" in out.content) + + # 身份未确认时记忆工具不可用 + out = await execute("memory_search", {"query": "香菜"}, + ctx(store, confirmed=False)) + check("未确认身份拒绝 memory_search", not out.ok and "身份未确认" in out.content) + + # 明确记住 → confirmed，且 owner 来自 ctx 而非模型 + out = await execute("memory_remember", + {"kind": "preference", "key": "饮食偏好", "value": "不吃香菜"}, + ctx(store, user_text="记住我不吃香菜", uid="u_a")) + check("明确记住写入成功", out.ok and out.meta.get("status") == "confirmed", out.content) + check("owner 用 ctx 的 u_a", + any(c[1][0] == "u_a" for c in store.calls if c[0] == "remember")) + + # 推断 → candidate，且如实告知模型未生效 + out = await execute("memory_remember", + {"kind": "fact", "value": "喜欢牛肉面"}, + ctx(store, user_text="我今天吃了牛肉面")) + check("推断写入为 candidate", out.meta.get("status") == "candidate", out.content) + check("如实告知未生效", "尚未生效" in out.content, out.content) + + # 同 key 幂等更新：不应产生第二条 confirmed + await execute("memory_remember", + {"kind": "preference", "key": "饮食偏好", "value": "现在吃香菜了"}, + ctx(store, user_text="记住我饮食偏好改了")) + confirmed = [r for r in store.rows + if r.owner_user_id == "u_a" and r.key == "饮食偏好" + and r.confirm_status is ConfirmStatus.CONFIRMED] + check("同 key 只保留一条 confirmed", len(confirmed) == 1, f"{len(confirmed)} 条") + check("值已更新", confirmed[0].value == "现在吃香菜了", confirmed[0].value) + + # 检索：candidate 默认不返回 + out = await execute("memory_search", {"query": "牛肉面"}, ctx(store)) + check("candidate 不出现在检索结果", "牛肉面" not in out.content, out.content) + + # 更新已有键 + out = await execute("memory_update", + {"key": "饮食偏好", "value": "其实我现在吃香菜了"}, + ctx(store, user_text="其实我现在吃香菜了")) + check("更正成功", out.ok and out.meta.get("updated"), out.content) + check("更正后值变了", + any(r.key == "饮食偏好" and r.value == "其实我现在吃香菜了" + for r in store.rows)) + + # 更正不存在的键 + out = await execute("memory_update", {"key": "不存在的键", "value": "x"}, ctx(store)) + check("更正不存在的键 → 拒绝", not out.ok, out.content) + + # 精确删除 + out = await execute("memory_forget", {"query": "饮食偏好"}, + ctx(store, user_text="忘掉我的饮食偏好")) + check("精确删除执行", out.ok and out.meta.get("deleted"), out.content) + check("删除话术如实（已删除）", "已删除" in out.content, out.content) + + # 删除不存在 + out = await execute("memory_forget", {"query": "完全不存在的东西"}, + ctx(store, user_text="忘掉完全不存在的东西")) + check("删除不存在的记忆 → 不谎报", not out.ok and "没有找到" in out.content, out.content) + + # 模糊删除 → 需要确认 + store2 = FakeStore() + for k, v in (("饮食偏好", "不吃香菜"), ("饮品偏好", "喜欢美式")): + await execute("memory_remember", + {"kind": "preference", "key": k, "value": v}, + ctx(store2, user_text=f"记住我的{k}是{v}")) + out = await execute("memory_forget", {"query": "偏好"}, + ctx(store2, user_text="忘掉我的偏好")) + check("模糊删除 → 不猜目标", out.meta.get("needs_confirmation") is True, out.content) + check("模糊删除不实际删除", len(store2.rows) == 2, f"{len(store2.rows)} 条") + + +async def test_owner_isolation(): + print("\n=== 6. 跨用户隔离（交接文档 Phase 4 验收第 7 条）===") + store = FakeStore() + await execute("memory_remember", + {"kind": "preference", "key": "饮食偏好", "value": "不吃香菜"}, + ctx(store, user_text="记住我不吃香菜", uid="u_a")) + + out = await execute("memory_search", {"query": "香菜"}, ctx(store, uid="u_b")) + check("用户 B 查不到用户 A 的记忆", "香菜" not in out.content, out.content) + + # B 试图删 A 的 id：存储层必须同时匹配 owner + a_id = store.rows[0].memory_id + deleted = await store.forget("u_b", a_id) + check("跨 owner 删除失败", deleted is False) + check("A 的记忆还在", len(store.rows) == 1) + + +# ---------------------------------------------------------------- 主流程 + +def main() -> int: + test_policy_remember() + test_episodic() + test_forbidden_args() + test_policy_forget() + asyncio.run(test_tools()) + asyncio.run(test_owner_isolation()) + + print("\n" + "=" * 62) + print(f"通过 {len(OK)} / {len(OK) + len(FAIL)}") + for n in FAIL: + print(f" 未通过: {n}") + return 0 if not FAIL else 1 + + +if __name__ == "__main__": + sys.exit(main())