ylhan/ai-companion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 2：抽出工具与策略层

Browse Source 
按 P1-Pipecat Flows Agent化与记忆改造交接.md 的 Phase 2 完成。
legacy 路径行为不变，只是改为复用新工具层。

新增 app/agent/
- memory_policy.py：记忆写入/删除的裁决层。模型只提议，策略决定。
  不依赖 LLM/网络/真实存储，可纯单元测试。
  - 身份未确认 → 一律 reject，不碰私人记忆
  - 模型传 owner_user_id/scope 等越权字段 → reject（不静默忽略）
  - 只有用户明确说"记住/以后/别忘了/我的偏好是"才写 confirmed
  - "记住我今天吃了香菜"这类一次性事件 → 降级 candidate
  - 不采信模型自称的"用户明确要求"
  - 模糊/多条删除 → needs_confirmation，不猜目标
- tools.py：工具 schema 与执行器，legacy 与 flows 共用。
  - 身份由 ToolContext 注入，**不在工具参数里**
  - 身份未确认时不暴露任何记忆工具（比暴露后拒绝更干净）
  - 写入结果如实回报：candidate 明确告知"尚未生效"，
    删除如实返回"已删除/没有找到/需要确认"，不把"请求已收到"说成"已做到"
  - 工具失败转成给模型的文本，不拖垮本轮对话
- llm_service.py：Pipecat LLMService 工厂，固化 Phase 1 spike 的选型结论
  （复用 DeepSeekLLMService；必须显式覆盖模型名为 deepseek-flash，
  因为 Pipecat 默认的 deepseek-v4-flash 不存在）

改造 app/engine.py
- _execute_tool 改为委托 app/agent/tools.execute()，删除重复实现
- 本轮用户原话挂到实例上，供策略层判定语义等级
- 移除不再使用的 SearchError 导入

新增 app/tests/test_agent_tools.py：45 项断言，**不依赖 LLM/网络/SQLite**
（用 fake store）。覆盖交接文档 §十 的清单：越权参数、未确认身份、
明确记住 vs 推断、同 key 幂等、模糊删除、精确删除、跨用户隔离。

回归：test_agent_tools 45/45、test_memory 30/30、test_identity 19/19、
test_search 12/12、test_transport 4/4、test_profile_memory 22/22，无退化。
This commit is contained in:
ylhan
2026-09-16 13:25:46 +08:00
parent c37a968548
commit 4fe2755fd6
6 changed files with 1150 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tests/test_agent_tools.py
+347
View File
@@ -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())