Files
group_fqcd_jr/app/api/controllers/public_platform.py
T
lzf_0626 c0e5c80929 记忆系统:recall 结果接入 prompt + 可观测性(既有改动,代为提交)
## 说明

**这批改动不是本次会话写的**,它们在会话开始前就已在工作区里、一直未提交。
我做的是**验证**它确实成立,然后按你的指示代为提交。

出处:`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 的召回恒空(员工身份不是客户),所以接线后行为不变,
要用测试替身才能验证注入。我没有为此编造证据。
2026-09-14 20:35:46 +08:00

170 lines
7.0 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from typing import Any, Literal
from fastapi import APIRouter, Depends, Header, Query
from pydantic import Field
from app.api.dependencies.auth import build_request_context
from app.api.dependencies.rate_limit import enforce_rate_limit
from app.api.schemas.admin import StrictPayload
from app.api.schemas.conversations import HandoverRequest
from app.core.contracts import RequestContext
from app.service.public_platform_service import PublicPlatformService
from app.service.public_product_service import PublicProductService
router = APIRouter(prefix="/api/v1", tags=["public-platform"],
dependencies=[Depends(enforce_rate_limit)])
class SessionCreate(StrictPayload):
agent_type: str = Field(pattern=r"^[a-z][a-z0-9_]{1,31}$")
class Cancellation(StrictPayload):
reason: str = Field(default="user_cancelled", max_length=128)
class CandidateDecisionPayload(StrictPayload):
decision: Literal["confirmed", "rejected"]
@router.post("/conversations", status_code=201)
async def create_session(
payload: SessionCreate,
key: str | None = Header(default=None, alias="Idempotency-Key"),
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().write("create", "", context, key, payload.model_dump())
@router.get("/conversations/{session_id}")
async def get_session(
session_id: str, context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().session(session_id, context)
@router.post("/conversations/{session_id}/closures")
async def close_session(
session_id: str, key: str | None = Header(default=None, alias="Idempotency-Key"),
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().write("close", session_id, context, key, {})
@router.post("/agent-runs/{run_id}/cancellations", status_code=202)
async def cancel_run(
run_id: str, payload: Cancellation,
key: str | None = Header(default=None, alias="Idempotency-Key"),
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().write("cancel", run_id, context, key, payload.model_dump())
@router.get("/handover-requests/{handover_id}")
async def get_handover(
handover_id: str, context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().handover(handover_id, context)
# 转人工申请的唯一入口:必须走 PublicPlatformService.write("handover"),
# 它在同一事务内写工单 + Outbox 事件(conversation.transfer_requested)+ 审计。
# 历史上 conversations.py 另有一份只写审计的实现,因注册顺序覆盖了本入口,
# 导致转人工的异步链路(Outbox → Worker)永不触发,已合并删除。
@router.post("/conversations/{session_id}/handover-requests", status_code=202)
async def request_handover(
session_id: str,
payload: HandoverRequest,
key: str | None = Header(default=None, alias="Idempotency-Key"),
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().write(
"handover", session_id, context, key, payload.model_dump()
)
@router.get("/users/me/memory-profile")
async def my_memory(
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().memory(int(context.user_id), context)
@router.get("/customers/{customer_id}/memory-profile")
async def customer_memory(
customer_id: int, context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
return await PublicPlatformService().memory(customer_id, context)
@router.get("/users/me/memories")
async def my_memories_debug(
context: RequestContext = Depends(build_request_context), # noqa: B008
query: str | None = None,
limit: int = 10,
) -> dict[str, Any]:
"""记忆系统可观测端点:一次返回「库里有什么」「能不能召回到」「事件有没有被消费」。
这是排查"记忆到底有没有在工作"的唯一出口 —— 在此之前,`memory_unit` 原始行
没有任何读接口,`GET /users/me/memory-profile` 只返回画像快照(记忆的下游产物),
因此"写入成功但画像还没重建"与"根本没写入"在外部完全无法区分。
"""
return await PublicPlatformService().memories_debug(
int(context.user_id), context, query=query, limit=limit
)
@router.get("/users/me/memory-candidates")
async def my_memory_candidates(
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
"""返回当前用户可确认的画像候选,不返回证据原文。"""
from app.service.customer_profile_candidate_service import CustomerProfileCandidateService
return await CustomerProfileCandidateService().list_for_customer(context)
@router.post("/users/me/memory-candidates/{candidate_id}/decisions")
async def decide_memory_candidate(
candidate_id: int,
payload: CandidateDecisionPayload,
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
"""用户确认或拒绝自己的候选;确认后仍需管理员审核才能激活。"""
from app.service.customer_profile_candidate_service import CustomerProfileCandidateService
return await CustomerProfileCandidateService().decide_by_customer(
candidate_id, payload.decision, context
)
@router.get("/products")
async def list_products(
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
"""公开产品列表:在售场内基金 + 各自最新行情(访客令牌即可访问)。
这是访客三个页面(首页推荐 / 产品列表 / 产品详情)的数据源,
替代原先前端手写的 `common/mock-data.js`。
只要求**有效令牌**、不检查权限码:访客令牌的上下文只有 `roles=("visitor",)`
且不带权限,与 `/api/v1/conversations`、`/api/v1/agent-runs` 的访客口径一致。
"""
return await PublicProductService().list_products(context)
@router.get("/products/{product_code}/nav-history")
async def product_nav_history(
product_code: str,
days: int = Query(default=90, ge=1, le=365),
context: RequestContext = Depends(build_request_context), # noqa: B008
) -> dict[str, Any]:
"""产品历史净值序列(编号 `P002`):产品详情页净值走势图的数据源。
数据来自 `fin_nav_history`,由 `tools/sync_nav_history.py` 从东财净值接口同步。
鉴权口径与 P001 相同:**要求有效令牌但不校验权限码**(访客令牌可用)。
表为空时返回 `count=0` 与空数组,**不是错误** —— 前端据此显示"尚未接入"。
"""
return await PublicProductService().nav_history(product_code, context, days=days)