Files
group_fqcd_jr/app/service/agent/governance.py
T
lzf_0626 9eebf9627f 记忆召回:按 sys_customer_assignment 归属定范围,不再把员工号当客户号
## 修的是什么

`governance.recall()` 把 `int(context.user_id)` 当客户号用。后果有两个,
方向相反但都致命:

1. **员工身份(风控/投顾/运营/管理员/system)恒空** —— 员工不是客户,
   那是个不存在的客户号;日志只说 "empty",看不出是"设计如此"还是"记忆坏了"。
2. **越权陷阱** —— 员工号与客户号同号段(演示数据里客户 9001-9020、
   员工 9002/9020 并存)。`int(user_id)` 一旦与真实客户号重合,就会把
   **陌生客户的长期记忆读进来并注入提示词**,且不报错、看起来正常。

同一个问题在代码里还有另外两处**各自判断**、口径互不一致:
`BaseAgent.recall_memory()` 要求"每条记忆 customer_id == context.user_id"
(否则抛"越过客户范围"),`review_output()` 的引用校验只认同一条件。

## 怎么修的

新增 `app/core/memory_scope.py` 作为**唯一判定口径**,三处共用:

- 客户身份(customer / authenticated_user):**只读自己**,分配表里有别行也不读别人;
- 员工身份:**只读 `sys_customer_assignment` 分配给自己**的客户
  (`context.customer_ids`,由 `IdentityRepository.load_context()` 读入);
  归属未维护 ⇒ **失败关闭**,并在日志里点名"归属未维护",与"库里确实没有记忆"区分开;
- 访客:无(上游已拦)。

细节约定:
- 归属客户按客户号**升序**召回、单次上限 `MAX_RECALL_CUSTOMERS=10`
  —— 升序是为了确定性(同一身份每次取同一批,不随数据库返回顺序漂移),
  上限是为了别把成百上千条他人记忆塞进一个提示词;
- 跨客户合并后按置信度降序、`(客户号, uuid)` 兜底排序,最多 10 条;
- 员工同时持有多个归属客户的记忆时,`memory_context_text()` **逐行标注客户号**
  并把提示词改成"多个客户的长期事实" —— 否则模型会把 A 客户的事实当成 B 客户的。
  单一客户时保持原格式(客户身份的提示词与改动前逐字相同);
- 引用校验与范围守卫都改用同一口径:员工引用**归属客户**的记忆不再被判成伪造引用;
  引用**非归属客户**的记忆即便被塞进 memories 也照样拦下。

## 验证(真实身份链路 + 生产召回装配)

`IdentityRepository.load_context` → `PlatformGovernance.recall`(含 Milvus 语义通道):

- 身份展开:roles=('advisor',)、customer_ids=('9001',)(sys_customer_assignment
  里唯一那行 9020→9001)、可读范围 (9001,);
- **修复前** `recall(int(user_id)=9020)` → **0 条**;
- **修复后** `recall(按归属)` → **2 条**(客户9001:进取型 / 约三年);
- 边界:客户身份 9001 可读范围 (9001,);无归属员工 9002 = ()(失败关闭,
  且**没有**把 9002 当客户号);未分配时的 9020 = ()。

测试:`pytest tests/unit tests/contract` → **1445 passed, 2 skipped, 1 failed**
(1432 + 新增 13;唯一失败是组员正在改的投顾页面,与记忆链路无关)。
新增用例:`tests/unit/core/test_memory_scope.py`(8 条,含"员工号不得被当成客户号"
的反例断言)、`tests/unit/service/test_agent_governance.py`(+5 条:归属召回/
无归属失败关闭且不碰数据库/客户只读自己/引用校验/越界守卫)。

## 遗留(已在 AGENTS.md 与文档里写明,未自行实施)

风控扫描这条线**仍读不到记忆**:它是唯一消费召回内容的地方
(`risk_agent.py:224`),而扫描上下文是 user_id="0"/roles=("system",) 且无归属行。
根因是**顺序问题**:召回发生在 handle() 之前,上下文里没有"本次目标客户"这个概念。
出路有两条:① 给风控专员补 sys_customer_assignment 行(运维动作,立即可用);
② 在 RequestContext 加显式的 target_customer_id 并校验它落在归属集合内
(推荐,但属跨线协议改动,等确认)。

文档:docs/演示用/记忆召回恒空-根因与修复-2026-09-14.md 新增 §五(含 §5.4 遗留说明)、
AGENTS.md 新增"记忆可读范围只有一个判定口径"易错点,并按 2026-09-14 复测更新测试基线。
2026-09-14 21:33:16 +08:00

357 lines
21 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.
import logging
import re
from collections.abc import Callable
from typing import Protocol
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.compliance_context import first_violation as _first_violation
from app.core.contracts import (
AgentDefinition,
AgentResult,
RecalledMemory,
RequestContext,
ResolvedAgentConfig,
)
from app.core.errors import ForbiddenAgentError, RecoverableAgentError
from app.core.memory_scope import (
RECALL_ITEM_LIMIT,
customer_memory_scope,
is_customer_identity,
memory_customer_in_scope,
)
from app.infrastructure.db import SessionFactory
from app.model.configuration import ConfigRelease
from app.service.memory_recall_service import MemoryRecallService
from app.service.runtime_config_service import RuntimeConfigService
logger = logging.getLogger(__name__)
# 组装层注入的召回服务工厂:允许带 Redis 缓存与向量库,缺省退化为纯结构化召回。
RecallServiceFactory = Callable[[AsyncSession], MemoryRecallService]
#: 固定免责声明在 `agent_reply_template` 里的 `template_code`(Task 1 已种入并审核通过)。
#: 按 `template_code` 解析而不是按 `scene`:唯一键 `uk_reply_template_active_one` 保证的是
#: "同一 template_code 只有一条 active",`scene` 上没有唯一性——按 scene 取会在同场景多模板时
#: 取到哪条不确定,而免责声明是必须逐字固定的合规话术。
DISCLAIMER_TEMPLATE_CODE = "TPL_DISCLAIMER"
#: 平台对外客服热线:**公开业务号码**,必须原样出现在面向客户的回复里
#: (spec B §5 要求固定联系方式)。脱敏规则专门放行它,
#: 否则 `TPL_TRANSFER_HUMAN` 与安全路由的 P2 话术会被打成 "[手机号已脱敏]",
#: 客户拿不到联系方式(已实测复现)。与 `app/core/customer_service_rules.CONTACT_PHONE`
#: 是同一个号码,改一处必须同步另一处。
CUSTOMER_SERVICE_HOTLINE = "15936583816"
#: 代码兜底免责声明,逐字等于 `TPL_DISCLAIMER` 的已审文案。
#: 免责声明是安全关键路径(上线门禁 F5:面向客户输出 100% 附固定话术),所以**不允许**
#: "库里查不到就没有话术"这条路径——取数失败或话术缺失时用本常量,宁可多一次失败的查询,
#: 也不能少一句话术。文案本身不得命中任何禁用字面(`agent_negative_word` 规则与下方
#: `hard_patterns`):治理层是朴素子串匹配、没有否定式豁免,写过「非保本」的合规话术会被
#: 自己的规则拦下(Task 1 已修复过一次的"自绊")。
FALLBACK_DISCLAIMER = (
"本内容仅为投资分析参考,不构成任何直接投资建议,不构成对任何产品的收益承诺,"
"据此操作风险自负,请谨慎对待。"
)
#: **内部** Agent 清单:其输出不面向客户,因此不追加面向客户的固定免责声明。
#: 判据是"输出形态"而不是"重要性"——风控/投顾分析的输出是字段化摘要
#: (预警编号、级别、建议动作),追加一句面向投资者的免责声明会破坏其字段契约,
#: 下游解析与 `tests/contract/test_risk_agent_contract.py` 都会因此失败(已实测)。
INTERNAL_AGENT_TYPES = frozenset({"risk"})
#: **确认面向客户**的 Agent 清单:门禁 F5(面向客户输出 100% 附固定话术)只对这些生效。
#: 为什么用"确认式"而不是"未知即注入":`review_output` 是同步纯函数,它的调用方里既有
#: 生产装配的 `PlatformGovernance`(能反查发布版本的 `agent_type`),也有各测试的治理替身
#: (**刻意不连库**,因此无从得知 agent 类型)。若把"未知"当成面向客户,每个替身测试都会被
#: 塞进一句话术,等于用测试噪声换一个假的安全感;而这些测试恰恰是在断言 Agent 的结构化输出。
#: 生产路径下客服 Agent 必然带发布版本(`config_release` 有 agent_tools 白名单),
#: 所以 F5 的覆盖不受影响——**未发布配置的 Agent 本来就没有可用工具、也不接受验收**。
CUSTOMER_FACING_AGENT_TYPES = frozenset({"customer_service", "fund_query_demo"})
class AgentGovernance(Protocol):
async def resolve(
self, definition: AgentDefinition, context: RequestContext
) -> ResolvedAgentConfig: ...
async def recall(self, context: RequestContext) -> tuple[RecalledMemory, ...]: ...
async def review(
self, result: AgentResult, context: RequestContext, config: ResolvedAgentConfig,
memories: tuple[RecalledMemory, ...],
*,
agent_type: str = "",
) -> AgentResult: ...
class PlatformGovernance:
"""Request-scoped snapshots; no AsyncSession or customer state retained on the singleton."""
def __init__(self, recall_factory: RecallServiceFactory | None = None) -> None:
# 组装层可注入带缓存/向量库的召回服务;缺省退化为纯结构化召回,
# 依赖缺失或故障只降级,不让记忆功能整体不可用。
self._recall_factory = recall_factory
async def resolve(
self, definition: AgentDefinition, context: RequestContext
) -> ResolvedAgentConfig:
async with SessionFactory() as session:
release = await session.scalar(select(ConfigRelease).where(
ConfigRelease.status == "active"))
tools: dict[str, tuple[str, ...]] = {}
if release is not None:
service = RuntimeConfigService(session)
for intent in definition.supported_intents:
configured = await service.allowed_tools(
release.id, definition.agent_type, intent)
tools[intent] = tuple(sorted(set(configured) & set(definition.allowed_tools)))
rows = (await session.execute(text("""
SELECT match_type, word_pattern, applicable_agents FROM agent_negative_word
WHERE status='active' AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL
"""))).mappings().all()
rules: list[tuple[str, str]] = []
import json
for row in rows:
agents = row["applicable_agents"]
if isinstance(agents, str):
agents = json.loads(agents)
if not agents or definition.agent_type not in agents:
# 失败关闭:`applicable_agents` 为空数组/NULL 表示"不适用任何 Agent",
# 跳过该规则。
# 旧判据 `if agents and ...` 在空取值时整条 `continue` 不执行,规则因此**溢出到
# 所有 Agent**(实测 advisor 也加载到 11 条):配置为空不是失效,而是扩大适用
# 范围,与项目一贯的失败关闭原则相反。
continue
if row["match_type"] not in {"exact", "contains"}:
# Python re has no execution timeout: never run unchecked admin regex.
raise RecoverableAgentError("禁止表达正则规则需要受限匹配器")
rules.append((str(row["match_type"]), str(row["word_pattern"])))
return ResolvedAgentConfig(
config_version=release.release_no if release else f"code:{definition.version}",
release_id=release.id if release else None,
prompt_version="none", model_endpoint="", allowed_tools=(),
allowed_tools_by_intent=tools, negative_rules=tuple(rules),
)
async def recall(self, context: RequestContext) -> tuple[RecalledMemory, ...]:
"""按**身份可读范围**召回长期记忆(范围口径见 `app/core/memory_scope.py`)。
此前这里把 `int(context.user_id)` 直接当客户号,于是:
① 员工身份(风控/投顾/管理员/system)**恒空** —— 员工不是客户,那是个不存在的客户号;
② 更糟的是**越权陷阱** —— 员工号与客户号同号段时,会把陌生客户的长期记忆读进来
并注入提示词。现在改为:客户身份只读自己,员工身份只读
`sys_customer_assignment` 分配给自己的客户(最小可见),
归属未维护时**失败关闭**(读不到任何东西)并在日志里点名原因。
"""
scope = customer_memory_scope(context)
if not scope:
self._log_empty_scope(context)
return ()
async with SessionFactory() as session:
service = (
self._recall_factory(session)
if self._recall_factory is not None
else MemoryRecallService(session)
)
merged: dict[str, tuple[float, RecalledMemory]] = {}
reasons: list[str] = []
for customer_id in scope:
result = await service.recall(customer_id)
reasons.extend(result.degraded_reasons)
for item in result.items:
previous = merged.get(item.memory_uuid)
if previous is None or item.confidence > previous[0]:
merged[item.memory_uuid] = (
item.confidence,
RecalledMemory(
memory_uuid=item.memory_uuid,
customer_id=str(customer_id),
content=item.content,
),
)
# 排序:置信度降序,再按 (客户号, uuid) 兜底 —— 置信度相同的两条不能因为
# 字典/数据库返回顺序不同而每次换位,同一输入必须给出同一结果。
ordered = sorted(
merged.values(),
key=lambda pair: (-pair[0], int(pair[1].customer_id), pair[1].memory_uuid),
)
items = tuple(memory for _, memory in ordered[:RECALL_ITEM_LIMIT])
# 召回结果此前完全没有出口:即使召回到内容也无人消费,运维无法判断
# "库里没有记忆"与"召回了但被丢弃"。这里把条数、来源与摘要打出来。
logger.info(
"memory recall scope=%s customers=%s count=%s degraded=%s reasons=%s items=%s",
"self" if is_customer_identity(context) else "assigned",
list(scope), len(items), bool(reasons), ",".join(sorted(set(reasons))) or "-",
[f"c{memory.customer_id}:{memory.content[:40]}" for memory in items[:5]],
)
if reasons:
logger.warning("memory recall degraded customers=%s reasons=%s",
list(scope), ",".join(sorted(set(reasons))))
if not items:
# 范围非空却一条都没有:这才是"库里确实没有该客户的记忆",
# 与"没有授权范围"必须区分开(后者见 `_log_empty_scope`)。
logger.info(
"memory recall empty: 客户 %s 在 memory_unit 里没有 active 记忆",
list(scope),
)
return items
@staticmethod
def _log_empty_scope(context: RequestContext) -> None:
"""可读范围为空时**点名原因**,避免把"没有授权"误判成"记忆坏了"。"""
if is_customer_identity(context):
logger.info(
"memory recall skipped: 客户身份但 user_id 不可解析为客户号 user_id=%r",
context.user_id,
)
return
logger.warning(
"memory recall skipped: 员工身份 roles=%s 在 sys_customer_assignment 里"
"没有生效的归属客户 ⇒ 无可读客户(失败关闭)。这不是记忆故障;"
"要读指定客户请先维护分配关系(或走 query_customer_profile 工具)",
list(context.roles),
)
async def review(
self, result: AgentResult, context: RequestContext, config: ResolvedAgentConfig,
memories: tuple[RecalledMemory, ...],
*,
agent_type: str = "",
) -> AgentResult:
# 裁定 1:读库是异步的,由本方法(异步层)做;`review_output` 保持同步、不接触数据库,
# 只把拿到的文本追加到输出末尾。
#
# `agent_type` 由 `BaseAgent._execute_governed()` 从**定义**传入(不是查库):它决定
# 门禁 F5 是否适用(见 `CUSTOMER_FACING_AGENT_TYPES`)。给了默认值是为了让既有的
# 治理替身按旧签名调用时仍能工作——那种情况下按"未声明"处理,不注入话术。
disclaimer = await _load_template_text(DISCLAIMER_TEMPLATE_CODE)
return review_output(result, context, config, memories, disclaimer=disclaimer,
agent_type=agent_type)
async def _load_template_text(template_code: str) -> str | None:
"""按 `template_code` 读取生效话术;查不到或取数失败返回 None,由调用方退回代码兜底。
取数条件与 `resolve()` 的规则查询同口径(`status='active'` 且审核字段非空),
避免"库里改了状态但代码照用"的偏差。异常一律吞掉并记日志:免责声明必须 100% 注入,
数据库故障时"少一句话术"比"整条回复失败"更糟,退回 `FALLBACK_DISCLAIMER` 即可满足门禁。
"""
try:
async with SessionFactory() as session:
content: str | None = await session.scalar(
text("""
SELECT content_text FROM agent_reply_template
WHERE template_code = :code AND status='active'
AND reviewer_id IS NOT NULL AND reviewed_at IS NOT NULL
ORDER BY version DESC LIMIT 1
"""),
{"code": template_code},
)
except Exception:
logger.warning("免责声明话术查询失败,退回代码兜底 template_code=%s",
template_code, exc_info=True)
return None
if content is None or not str(content).strip():
logger.warning("免责声明话术不存在或为空,退回代码兜底 template_code=%s", template_code)
return None
return str(content)
def review_output(
result: AgentResult, context: RequestContext, config: ResolvedAgentConfig,
memories: tuple[RecalledMemory, ...],
*,
disclaimer: str | None = None,
agent_type: str = "",
) -> AgentResult:
"""同步治理:引用校验 → 负面词判定/替换 → 脱敏 → 追加固定免责声明。
`disclaimer` 由异步层(`PlatformGovernance.review`)注入库内话术;本函数是同步的、
不接触数据库。默认 `None` 表示"调用方未提供",此时用 `FALLBACK_DISCLAIMER` 兜底——
因此既有调用点不必改签名也能拿到固定话术(门禁 F5 的 100% 覆盖)。
`agent_type` 用于判定**是否面向客户**:门禁 F5 要求的是"客服答复 100% 附固定话术",
而内部 Agent(风控预警、投顾分析)的输出是结构化摘要,追加话术会破坏它的字段契约
(`test_risk_agent_contract` 实测因此失败)。空串按"调用方未声明"处理,**保守照旧追加**,
避免漏加;只有明确列入内部清单的 Agent 才跳过。
"""
content = result.result
# 门禁 F5 的适用范围:**已确认面向客户**的 Agent。判据来自 `agent_type`,它由
# `PlatformGovernance.review()` 从发布版本反查(`tests` 的治理替身不连库 → 空串 →
# 不注入,与它们的断言一致)。空串/未知一律**不注入**而不是注入,理由见
# `CUSTOMER_FACING_AGENT_TYPES` 的说明。
customer_facing = agent_type in CUSTOMER_FACING_AGENT_TYPES
issued_tools = {f"{context.trace_id}:{record.tool_name}" for record in content.tool_calls
if record.status == "succeeded"}
# "本轮已知记忆"必须用**与召回同一套范围判定**(`app/core/memory_scope.py`)。
# 此前只在 `customer_id == context.user_id` 时才算已知,于是员工身份下
# `known` 恒空:即使 `recall()` 按归属召回了客户记忆,风控 Agent 一引用
# 就被判成"引用未来自本次已授权召回结果",整条运行直接失败。
scope = customer_memory_scope(context)
known = {memory.memory_uuid for memory in memories
if memory_customer_in_scope(memory.customer_id, scope)}
for reference in content.source_references:
valid = ((reference.source_type == "memory" and reference.source_id in known)
or (reference.source_type == "tool" and reference.source_id in issued_tools))
if not valid:
raise ForbiddenAgentError("引用未来自本次已授权召回结果")
output = content.text
hard_patterns = ("保证收益", "稳赚不赔", "保本保收益", "已为您下单", "已替您交易")
# 语境豁免(详见 `app/core/compliance_context.py`):零容忍规则是朴素子串匹配,
# 无法区分"作出承诺"与"禁止承诺/谈论该表述"。政策类问答会被误伤 —— 实测
# 「基金销售有哪些合规要求」的答案因含"严禁承诺保本保收益""禁止使用『保证收益』"
# 被整条替换,答复从 1097 字掉到 83 字,合规问答反而答不出来。
# 判定:命中处附近出现否定/禁止/引用线索才豁免,否则照旧拦(宁可误拦)。
blocked = _first_violation(output, hard_patterns) is not None
for kind, pattern in config.negative_rules:
if kind == "exact":
blocked |= output == pattern
else:
blocked |= _first_violation(output, (pattern,)) is not None
if blocked:
content = content.model_copy(update={
"text": "该内容需要人工核实。基金投资存在风险,本系统不代客交易。",
"source_references": (), "transfer_required": True,
"transfer_reason": "compliance_review_required",
})
else:
# Apply to text and citation titles, not only to the displayed answer.
def redact(value: str) -> str:
def _mask_mobile(match: re.Match[str]) -> str:
# 平台对外的客服热线是**公开业务号码**,不是客户个人手机号:
# 它必须原样出现在面向客户的回复里(spec B §5 要求固定联系方式),
# 否则 `TPL_TRANSFER_HUMAN` / 安全路由的 P2 话术会被脱敏成
# "[手机号已脱敏]",客户拿不到联系方式(实测过)。
if match.group(0) == CUSTOMER_SERVICE_HOTLINE:
return match.group(0)
return "[手机号已脱敏]"
value = re.sub(r"(?<!\d)1[3-9]\d{9}(?!\d)", _mask_mobile, value)
return re.sub(r"(?<!\d)\d{15,19}[Xx]?(?!\d)", "[敏感号码已脱敏]", value)
content = content.model_copy(update={
"text": redact(output),
"source_references": tuple(ref.model_copy(update={
"title": redact(ref.title) if ref.title else None,
}) for ref in content.source_references),
})
# 门禁 F5:面向客户输出 100% 附固定话术,所以**拦截分支与正常分支都要走到这里**——
# 被负面词替换掉的回复同样是面向客户的输出。位置固定在最后:文案不参与上面的负面词
# 判定,也不会被替换逻辑吃掉。
# 幂等:只认**我们自己追加过的那个形状**("\n\n" + 话术),而不是"文本以话术结尾"。
# 比 `endswith(disclaimer_text)` 更紧的原因:后者在**退化文案**下仍会漏加——管理员把文案
# 发布成"。"、答案恰好是"您好。"时,`endswith("。")` 为真 → 不追加,客户一条话术都拿不到;
# 认形状后这种文本必然被追加。失败形态因此从"合规漏加"变成"极端情况下多追加一次",
# 这正是门禁 F5(面向客户输出 100% 附固定话术)要的方向。正文中间出现同样文字也不会被
# 误判成"已追加"——只有**末尾这个形状**才算。
# 空串不算话术:显式传空一律退回代码兜底,避免"传了但传空"成为绕过强制注入的后门。
disclaimer_text = (disclaimer or "").strip() or FALLBACK_DISCLAIMER
# 追加形状只在这里定义一次:判据与追加共用同一份,避免两处各写一个分隔符而漂移。
appended_shape = f"\n\n{disclaimer_text}"
if customer_facing and not content.text.endswith(appended_shape):
content = content.model_copy(update={"text": f"{content.text}{appended_shape}"})
return result.model_copy(update={"result": content})