lzf_0626
9490e5043b
## 新增：docs/演示用/记忆系统演示文档-2026-09-14.md 按"操作 → 看到什么 → 这体现什么"写，五个场景（探针看存储 / 记住一件事 / 谁能读谁读不到 / 画像版本与投影 / 停 Worker），每个场景配可复现命令与**实测输出**，另附建议顺序与时长、 六条问答话术、演示前自检清单、受控信号词表摘录。 配套两个新工具（都已实跑）： - `tools/memory_demo_chain.py`：一键走完"客户说 → 候选 → 客户确认 → 管理员批准 → 记忆与画像自动收敛"，打印演示前后对比（实测 2 秒收敛）。 - `tools/memory_recall_demo.py`：同一客户换六个身份召回，当场看出"客户只读自己 / 员工要权限码+归属 / 运营读不到 / 管理员有权限码但没归属也读不到"。 ## 修掉两处会让演示当场断链的问题（都是真机复现的） 1. **候选批准后画像字段不收敛** `CustomerProfileCandidateService._promote` 只写画像快照、**不投** `profile.rebuild_requested`，而 `user_facts` 与画像字段是 `ProfileAssemblyService` 的重建链路写的。后果：批准后记忆变 active、快照版本 +1，但**画像字段停在旧值** （客户说"三年以内"，画像里还是"十年以上"），要等一次无关的重建才收敛 —— 而"客户说完 → 批准 → 画像变了"正是演示主线。 现补一条重建事件（走事件而不是就地重建：本方法所在事务还没提交， 另开 session 看不到刚写入的记忆）。 2. **画像快照的唯一键从来没起作用，而且埋雷** `uk_profile_snapshot_current` 建在列 `current_customer_id` 上（不是 `is_current`）， 但 `ProfileAssemblyService._write_snapshot` 旧行只置 `is_current=False`（不清该列）、 新行**不写**该列（实测 13 行该列全 NULL）。 后果：只要客户**先被重建过一次**，旧 current 行仍占着 `current_customer_id=9001`， 下一次"批准画像候选"就会撞 `Duplicate entry '9001' for key uk_profile_snapshot_current` → **整次批准 500**（本次实测踩到）。 现按唯一键的真实语义写：清旧行的该列、新行显式写客户号。 （`ProfileGenerationService` / 候选路径本来就是这么写的，只有这一处没对齐。） ## 守卫 新增 `tests/integration/test_profile_snapshot_current_invariant_mysql.py`： 按真实顺序"先重建再批准候选"，断言 ① 只有一条 current 且它占着唯一键、 ② 历史版本已归还该列、③ 批准不再 500、④ 批准投出了重建事件。 修复前这条用例会在第 ② 步失败。 ## 验证 - `pytest tests/unit tests/contract` → 1490 passed, 2 skipped, 0 failed - 新增集成用例通过；真机实测演示主线：候选 → verified → active → **2 秒内** `user_facts` 与 `fin_customer_profile.investment_horizon` 都变成新值 - `mypy tools/memory_demo_chain.py tools/memory_recall_demo.py` → 0 错；ruff 全绿 ## 文档 `docs/44-演示流程.md`：配套文档清单与"记忆链路"备选场景都指向新演示文档。
259 lines
10 KiB
Python
259 lines
10 KiB
Python
"""记忆系统一键演示（写数据！）：客户说一句话 → 候选 → 两把钥匙 → 记忆与画像收敛。
|
||
|
||
**它做什么**（全程走真实 HTTP，不绕过权限与幂等）：
|
||
|
||
1. 打印"演示前"的记忆与画像快照；
|
||
2. 用客户账号在客服会话里说一句带**记忆信号**的话（默认：投资期限）；
|
||
3. 等 Worker 把它抽成**画像候选**（`memory_unit.status='candidate'`）；
|
||
4. 客户确认（第一把钥匙，`memory:candidate:confirm`）；
|
||
5. 管理员批准（第二把钥匙，`memory:candidate:review` + admin）；
|
||
6. 等 Worker 自动收敛（`profile.rebuild_requested`），打印"演示后"快照。
|
||
|
||
**前置**：API 与 **Worker 都要在跑**（`python -m app.worker`），否则第 3/6 步永远停在
|
||
`queued`。演示会**真的写入数据**（这正是它的意义），要用 `--customer` 换一个客户来隔离。
|
||
|
||
用法::
|
||
|
||
python tools/memory_demo_chain.py # 客户 9001，默认那句话
|
||
python tools/memory_demo_chain.py --customer 10001
|
||
python tools/memory_demo_chain.py --message "我的投资期限是十年以上"
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from typing import Any
|
||
|
||
import httpx
|
||
from sqlalchemy import text
|
||
|
||
from app.infrastructure.db import SessionFactory
|
||
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(errors="replace")
|
||
|
||
BASE = "http://127.0.0.1:8000"
|
||
DEFAULT_MESSAGE = "我的投资期限是五年以上，长期持有，不着急用钱。"
|
||
#: 演示账号（与演示数据一致；客户号 9001 对应 cust_t）。
|
||
CUSTOMER_ACCOUNT = ("cust_t", "123456")
|
||
ADMIN_ACCOUNT = ("admin_t", "88888888")
|
||
|
||
|
||
def _hdr(token: str, *, idempotent: bool = False) -> dict[str, str]:
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
if idempotent:
|
||
headers["Idempotency-Key"] = uuid.uuid4().hex
|
||
return headers
|
||
|
||
|
||
def _login(client: httpx.Client, account: tuple[str, str]) -> tuple[str, int]:
|
||
"""返回 `(令牌, 用户号)` —— 用户号从登录响应里取（`data.user_id`），别写死。"""
|
||
response = client.post(
|
||
"/api/v1/auth/tokens", json={"username": account[0], "password": account[1]}
|
||
)
|
||
if response.status_code != 200:
|
||
raise SystemExit(f"登录 {account[0]} 失败：HTTP {response.status_code} {response.text[:200]}")
|
||
data = response.json()["data"]
|
||
return str(data["access_token"]), int(data["user_id"])
|
||
|
||
|
||
async def _snapshot(customer_id: int) -> dict[str, Any]:
|
||
async with SessionFactory() as session:
|
||
memories = [
|
||
tuple(row)
|
||
for row in (
|
||
await session.execute(
|
||
text(
|
||
"SELECT memory_key, content, status, version FROM memory_unit "
|
||
"WHERE customer_id = :cid AND status = 'active' ORDER BY memory_key"
|
||
),
|
||
{"cid": customer_id},
|
||
)
|
||
).all()
|
||
]
|
||
profile = [
|
||
tuple(row)
|
||
for row in (
|
||
await session.execute(
|
||
text(
|
||
"SELECT investor_type, investment_horizon, LEFT(risk_tags, 90) "
|
||
"FROM fin_customer_profile WHERE customer_id = :cid"
|
||
),
|
||
{"cid": customer_id},
|
||
)
|
||
).all()
|
||
]
|
||
facts = [
|
||
tuple(row)
|
||
for row in (
|
||
await session.execute(
|
||
text(
|
||
"SELECT fact_key, fact_value FROM user_facts "
|
||
"WHERE customer_id = :cid ORDER BY fact_key"
|
||
),
|
||
{"cid": customer_id},
|
||
)
|
||
).all()
|
||
]
|
||
count, top = (
|
||
await session.execute(
|
||
text(
|
||
"SELECT COUNT(*), COALESCE(MAX(version), 0) FROM profile_snapshots "
|
||
"WHERE customer_id = :cid"
|
||
),
|
||
{"cid": customer_id},
|
||
)
|
||
).all()[0]
|
||
pending = (
|
||
await session.execute(
|
||
text(
|
||
"SELECT COUNT(*) FROM domain_event_outbox "
|
||
"WHERE event_type = 'profile.rebuild_requested' AND status = 'pending'"
|
||
)
|
||
)
|
||
).all()[0][0]
|
||
return {
|
||
"memories": memories,
|
||
"profile": profile,
|
||
"facts": facts,
|
||
"snapshots": f"共 {count} 版，最高 v{top}",
|
||
"pending_rebuild": pending,
|
||
}
|
||
|
||
|
||
def _dump(title: str, data: dict[str, Any]) -> None:
|
||
print(f"\n--- {title} ---")
|
||
memories = data["memories"]
|
||
print(" active 长期记忆:", memories or "（无）")
|
||
print(" 画像字段 :", data["profile"] or "（无画像行）")
|
||
print(" user_facts :", data["facts"] or "（无）")
|
||
print(" 画像快照 :", data["snapshots"])
|
||
print(" 待消费的重建事件:", data["pending_rebuild"])
|
||
|
||
|
||
def _wait_run(client: httpx.Client, token: str, run_id: str) -> str:
|
||
for _ in range(40):
|
||
time.sleep(1.5)
|
||
payload = client.get(f"/api/v1/agent-runs/{run_id}", headers=_hdr(token)).json()["data"]
|
||
if payload.get("status") in ("succeeded", "failed", "cancelled"):
|
||
return str(payload["status"])
|
||
return "timeout"
|
||
|
||
|
||
def run(message: str) -> int:
|
||
with httpx.Client(base_url=BASE, timeout=60) as client:
|
||
# 客户号从**登录响应**里取，不写死：否则"看的是 A、写的是 B"，演示会前后矛盾。
|
||
customer, customer_id = _login(client, CUSTOMER_ACCOUNT)
|
||
admin, _admin_id = _login(client, ADMIN_ACCOUNT)
|
||
existing_ids = {
|
||
row["candidate_id"]
|
||
for row in (
|
||
client.get(
|
||
"/api/v1/users/me/memory-candidates", headers=_hdr(customer)
|
||
).json().get("data")
|
||
or []
|
||
)
|
||
}
|
||
|
||
before = asyncio.run(_snapshot(customer_id))
|
||
_dump(f"演示前（客户 {customer_id}）", before)
|
||
|
||
print(f"\n【1】客户在客服会话里说：{message}")
|
||
conversation = client.post(
|
||
"/api/v1/conversations", headers=_hdr(customer, idempotent=True),
|
||
json={"agent_type": "customer_service"},
|
||
).json()["data"]
|
||
run_response = client.post(
|
||
"/api/v1/agent-runs", headers=_hdr(customer, idempotent=True),
|
||
json={
|
||
"agent_type": "customer_service",
|
||
"session_id": conversation["session_id"],
|
||
"message": message,
|
||
"idempotency_key": uuid.uuid4().hex,
|
||
},
|
||
)
|
||
run_id = run_response.json()["data"]["run_id"]
|
||
print(f" 受理 202，run_id={run_id}（等 Worker 处理…）")
|
||
status = _wait_run(client, customer, run_id)
|
||
print(f" 运行终态：{status}")
|
||
if status != "succeeded":
|
||
print(" ⚠️ 运行没有成功，后续演示会断在这里（先确认 Worker 在跑）")
|
||
return 1
|
||
|
||
print("\n【2】等它被抽成画像候选（memory_unit.status='candidate'）")
|
||
candidate = None
|
||
for _ in range(20):
|
||
rows = client.get(
|
||
"/api/v1/users/me/memory-candidates", headers=_hdr(customer)
|
||
).json().get("data") or []
|
||
# 只认**这次新产生**的候选：库里可能躺着以前演示留下的 candidate，
|
||
# 按 id 差集挑，避免"批准了一条旧候选"这种假演示。
|
||
fresh = [
|
||
row for row in rows
|
||
if row.get("status") == "candidate" and row["candidate_id"] not in existing_ids
|
||
]
|
||
if fresh:
|
||
candidate = fresh[0]
|
||
break
|
||
time.sleep(1.5)
|
||
if candidate is None:
|
||
print(" ⚠️ 没有产生候选：这句话没命中记忆信号词表（换一句试试）")
|
||
return 1
|
||
print(f" 候选 #{candidate['candidate_id']}：{candidate['memory_key']} = "
|
||
f"{candidate['value']}（置信度 {candidate['confidence']}）")
|
||
|
||
print("\n【3】第一把钥匙：客户本人确认")
|
||
decided = client.post(
|
||
f"/api/v1/users/me/memory-candidates/{candidate['candidate_id']}/decisions",
|
||
headers=_hdr(customer, idempotent=True), json={"decision": "confirmed"},
|
||
).json()["data"]
|
||
print(f" 状态：{decided['status']}（candidate → verified）")
|
||
|
||
print("\n【4】第二把钥匙：管理员批准（这一步才进正式记忆）")
|
||
review = client.post(
|
||
f"/api/v1/admin/customer-profile-candidates/{candidate['candidate_id']}/reviews",
|
||
headers=_hdr(admin, idempotent=True),
|
||
json={"decision": "approved", "comment": "演示：客户已确认"},
|
||
)
|
||
if review.status_code != 200:
|
||
print(f" ⚠️ 批准失败 HTTP {review.status_code}：{review.text[:200]}")
|
||
return 1
|
||
print(f" 状态：{review.json()['data']['status']}（verified → active）")
|
||
|
||
print("\n【5】等 Worker 自动收敛（记忆 → 事实 → 画像字段 → 投影）")
|
||
converged = False
|
||
for attempt in range(15):
|
||
time.sleep(2)
|
||
current = asyncio.run(_snapshot(customer_id))
|
||
# 比**内容**而不是条数：同一个记忆键的值变了（约三年 → 十年以上）条数不变，
|
||
# 只比条数会误判成"没收敛"（踩过）。
|
||
if current["facts"] != before["facts"]:
|
||
print(f" 第 {attempt + 1} 次轮询（{(attempt + 1) * 2}s）：user_facts 已更新")
|
||
converged = True
|
||
break
|
||
if not converged:
|
||
print(" ⚠️ 没等到收敛：确认 Worker 在跑，或看 memory_sync_outbox.status")
|
||
|
||
_dump(f"演示后（客户 {customer_id}）", asyncio.run(_snapshot(customer_id)))
|
||
print(
|
||
"\n提示：想核对投影落没落库，跑 "
|
||
f"`python tools/probe_memory_state.py {customer_id}`；"
|
||
"想演「谁能读、谁读不到」跑 `python tools/memory_recall_demo.py`。"
|
||
)
|
||
return 0
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="记忆系统一键演示（会写数据）")
|
||
parser.add_argument("--message", default=DEFAULT_MESSAGE, help="客户说的那句话")
|
||
args = parser.parse_args()
|
||
return run(args.message)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|