lzf_0626
c0e5c80929
## 说明
**这批改动不是本次会话写的**，它们在会话开始前就已在工作区里、一直未提交。
我做的是**验证**它确实成立，然后按你的指示代为提交。
出处：`docs/演示用/记忆系统排查报告-2026-09-14.md` 与同目录
`记忆系统修复文档-2026-09-14.md`（两份都在本次一并入库）。
排查报告的结论是「记忆系统没有坏」——库里有真实数据、170 条抽取事件全部消费成功；
真正的问题是「观测不到」+「召回结果没人消费」。
## 改动内容（按两份文档的编号）
- **F1 `RecalledMemory.content` 断头路**：`base.py` 新增 `memory_context_text()`，
`risk_agent._agent_system_prompt` 接收并注入记忆段。无记忆时返回空串，
因此 prompt 逐字不变 —— 这也是它能安全接线的理由。
- **F3 `governance.recall` 员工身份恒空**：补一条明确的语义日志。
员工身份下召回的是"该用户自身作为客户"的记忆，恒为空属预期，
但此前没有任何提示，运维看到 `count=0` 只会以为记忆坏了。
- **F4 可观测性**：`GET /api/v1/users/me/memories`（`stored` / `recalled` /
`downstream` / `pending_events` 四段）+ 抽取与召回的 6 处日志 +
三个只读探针 `tools/probe_memory_state.py`、`probe_memory_detail.py`、
`probe_agent_types.py`。
**未实施**（文档明确留作待决，我也不代为决定）：F2 `known` 引用校验永不触发
（需架构确认 memory 类 `source_references` 由业务填还是底座统一附加）、
F5 客服是否读写长期记忆（涉脱敏与复核，需产品+合规）。
## 我做的验证（会话内实测，非照录文档）
- 新接口 `GET /users/me/memories` 以 `cust_t` 调用 -> **HTTP 200**：
stored: total=2, by_status={'active': 2}
recalled: count=2, degraded=False
两条记忆：preference:horizon='约三年'(0.95)、preference:risk_level='稳健型'(0.98)
与排查报告 §〇 列出的那两条**完全吻合**。
- `pytest tests/unit tests/contract` 全绿（这批改动没有破坏既有测试）。
## 未验证的部分
`memory_context_text()` 接进 prompt 后的**端到端效果没有实测** —— 文档自己说明了
原因：当前 `risk` Agent 的召回恒空（员工身份不是客户），所以接线后行为不变，
要用测试替身才能验证注入。我没有为此编造证据。
102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
"""记忆系统明细探针（只读）：回答「写入了什么 / 今天有没有活动 / 为什么 dead」。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.infrastructure.db import SessionFactory
|
|
|
|
|
|
async def q(session: Any, sql: str, **params: Any) -> list[Any]:
|
|
try:
|
|
return list((await session.execute(text(sql), params)).all())
|
|
except Exception as exc:
|
|
print(f" ERROR {type(exc).__name__}: {exc}")
|
|
return []
|
|
|
|
|
|
async def main() -> None:
|
|
async with s_factory() as session:
|
|
print("=" * 66)
|
|
print("A) memory_unit 全部内容（写入是否成功的直接证据）")
|
|
print("=" * 66)
|
|
rows = await q(session, """
|
|
SELECT id, customer_id, memory_key, content, memory_type, status,
|
|
confidence, source_type, evidence_count, created_at, updated_at
|
|
FROM memory_unit ORDER BY updated_at DESC LIMIT 20
|
|
""")
|
|
if not rows:
|
|
print(" (memory_unit 为空 —— 一条记忆都没写入)")
|
|
for r in rows:
|
|
print(f" id={r[0]} customer={r[1]} [{r[5]}]")
|
|
print(f" key = {r[2]}")
|
|
print(f" value = {r[3]!r}")
|
|
print(f" type={r[4]} conf={r[6]} src={r[7]} evidence={r[8]}")
|
|
print(f" created={r[9]} updated={r[10]}")
|
|
|
|
print()
|
|
print("=" * 66)
|
|
print("B) 今天(2026-09-14)有没有任何新活动")
|
|
print("=" * 66)
|
|
for label, sql in (
|
|
("agent_run 今天创建", "SELECT COUNT(*) FROM agent_run WHERE created_at >= '2026-09-14'"),
|
|
("agent_run 最新一条", "SELECT MAX(created_at) FROM agent_run"),
|
|
("conversation_message 最新", "SELECT MAX(created_at) FROM conversation_message"),
|
|
("domain_event_outbox 最新", "SELECT MAX(created_at) FROM domain_event_outbox"),
|
|
("outbox pending 总数", "SELECT COUNT(*) FROM domain_event_outbox WHERE status='pending'"),
|
|
("outbox failed 总数", "SELECT COUNT(*) FROM domain_event_outbox WHERE status='failed'"),
|
|
):
|
|
rows = await q(session, sql)
|
|
print(f" {label:<26} {rows[0][0] if rows else '?'}")
|
|
|
|
print()
|
|
print("=" * 66)
|
|
print("C) 最近 10 条 outbox 事件（看最新动向）")
|
|
print("=" * 66)
|
|
rows = await q(session, """
|
|
SELECT event_type, status, retry_count, last_error, occurred_at
|
|
FROM domain_event_outbox ORDER BY id DESC LIMIT 10
|
|
""")
|
|
for r in rows:
|
|
err = (str(r[3])[:70] + "...") if r[3] and len(str(r[3])) > 70 else r[3]
|
|
print(f" {r[4]} | {r[0]:<34} | {r[1]:<9} | retry={r[2]}")
|
|
if err:
|
|
print(f" last_error: {err}")
|
|
|
|
print()
|
|
print("=" * 66)
|
|
print("D) agent.run_requested 死信的原因分布（取 5 条样本）")
|
|
print("=" * 66)
|
|
rows = await q(session, """
|
|
SELECT retry_count, last_error, occurred_at
|
|
FROM domain_event_outbox
|
|
WHERE event_type='agent.run_requested' AND status='dead'
|
|
ORDER BY id DESC LIMIT 5
|
|
""")
|
|
for r in rows:
|
|
err = (str(r[1])[:150] + "...") if r[1] and len(str(r[1])) > 150 else r[1]
|
|
print(f" {r[2]} retry={r[0]}")
|
|
print(f" {err}")
|
|
|
|
print()
|
|
print("=" * 66)
|
|
print("E) 最近 5 次 agent_run 的状态")
|
|
print("=" * 66)
|
|
rows = await q(session, """
|
|
SELECT run_id, agent_type, status, error_code, created_at
|
|
FROM agent_run ORDER BY id DESC LIMIT 5
|
|
""")
|
|
for r in rows:
|
|
print(f" {r[4]} | {r[1]:<18} | {r[2]:<10} | err={r[3]} | {r[0][:8]}")
|
|
|
|
|
|
def s_factory():
|
|
return SessionFactory()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|