Files

T

lzf_0626 6516ccb385 feat: 第二版——接口契约对齐 docs/05，修复静默故障与数据库基线

相对第一版 46fc976 的完整变更。组员迁移对照表见 docs/20。

一、对外契约对齐 docs/05（破坏性，共 4 处，组员需按 docs/20 调整）
1) 配置发布端点改为文档规定的复数资源名：submit→validations、
   approve→reviews（需 body decision）、activate→activations、
   rollback→rollbacks；第一版这 4 个动词式路径 docs/05 从未定义过。
2) 错误码由 8 个笼统码改为 15 个具体语义码（FORBIDDEN→AGENT_PERMISSION_DENIED、
   UNAUTHORIZED→AUTHENTICATION_REQUIRED、CONFLICT→RESOURCE_VERSION_CONFLICT、
   RESOURCE_NOT_FOUND→RUN_NOT_FOUND/SESSION_NOT_FOUND 等），
   输入类错误状态码 400→422。
3) POST /api/v1/agent-runs 与 GET /api/v1/agent-runs/{run_id} 统一为
   {data, meta} 信封（data 内字段名与语义未变）。
4) 错误响应体统一为 {error:{code,message,retryable,field_errors}, meta:{trace_id}}，
   不再返回 FastAPI 默认的 {"detail": ...}。

二、数据库基线与约束
新增 39 张表的基线迁移（链根）与联合唯一键纠偏（4 张表、删 8 增 4，幂等收敛）；
撤下 config_release 的双人复核 CHECK（应用层已允许自审，审核节点保留，
自审如实写入 reviewer_id）；记忆 active key 生成列与唯一键；
activate 开始记录 supersedes_release_id 使版本链可追溯。
docs/00 基线未修改，未重命名或删除任何表与字段。

三、修复会静默出错或无报错的缺陷
- 跑完集成测试后平台会静默失去生效配置：清理只删自己创建的版本，却没有恢复被它
  顶成 superseded 的原生效版本，且审计一并删除因而完全无痕，表现为所有工具被拒
  但没有任何报错。已修清理逻辑并加恢复。
- Worker 单轮异常导致进程退出；记忆抽取调用方的“事务已开始”异常；
  召回缓存丢失 degraded 标记；连接时区未生效导致 created_at/updated_at 差 8 小时；
  .env 与 os.getenv 密钥来源分裂导致“没有可用的已批准模型端点”。
- 记忆信号识别漏判与跨键误命中；SSE 未带 Accept 的协商行为。

四、功能补齐
记忆链路 P1/P2/P3（抽取、受控词表、召回与缓存、生命周期级联及投影事件）、
fin_* 场内交易只读 ORM 层、agent_intent_config 状态流转并在运行期真正生效、
限流（Redis 固定窗口、故障一律放行）、游标校验、trace_id 中间件、
示例业务 Agent fund_query_demo 与一键端到端验证脚本，以及审计/指纹/迁移状态工具。

五、文档与验证
新增 docs/19（业务 Agent 接入实操）、docs/20（第一版迁移指南）与 docs/evidence 证据；
docs/01/02/06/08/09/17 同步实现现状。

验证结果：ruff 通过、mypy 103 文件无错、unit+contract 447 passed、
integration 29 passed、acceptance_check --production 7 PASS、
demo_agent_e2e 9/9 PASS（含失败关闭反证）。

2026-09-10 15:55:54 +08:00

10 KiB

Raw Blame History

Agent 组员入门易懂版说明

适用对象：第一次接入本项目的客服、投顾、风控和运营开发人员，以及协助编码的 AI。
阅读目标：看完后，能够在不破坏底座的前提下写出一个业务 Agent。

1. 先用一句话理解这个项目

你们要做的是“业务 Agent”，底座已经负责公共工作。

可以把系统想成一个机场：

Controller = 机场入口和检票口
AgentFactory = 登机调度台
BaseAgent = 统一登机流程
你的 Agent = 具体航班
ToolExecutor = 受控的机场服务柜台
Worker = 真正执行飞行任务的工作人员

你只需要设计“这个航班要去哪里、提供什么业务服务”，不需要重新建机场、重新做安检或自己买飞机。

2. 你真正需要写的代码

一个业务 Agent 通常只需要三部分：

定义 Agent 的名字、角色和意图；
写一个继承 BaseAgent 的类；
在公共工厂注册一行代码。

最小结构如下：

app/service/agent/implementations/your_agent.py   # 业务 Agent
tests/unit/service/test_your_agent.py             # 业务测试
app/service/agent/bootstrap.py                    # 注册一行

不要复制 BaseAgent，不要复制模型调用，不要复制工具权限代码。

3. 第一步：定义 Agent

from app.core.contracts import AgentDefinition


class AdvisorAgent(BaseAgent):
    definition = AgentDefinition(
        agent_type="advisor",
        version="1.0.0",
        allowed_roles=("customer", "advisor", "operator", "admin"),
        allowed_portals=("api",),
        allowed_tools=("query_fund_quote",),
        supported_intents=("fund_quote", "general"),
    )

每一项是什么意思

字段	简单理解	示例
`agent_type`	Agent 的唯一名字	`advisor`
`version`	业务 Agent 版本	`1.0.0`
`allowed_roles`	哪些角色可以使用	`customer`、`advisor`
`allowed_portals`	从哪里进入	`api`
`allowed_tools`	代码允许使用哪些工具	`query_fund_quote`
`supported_intents`	这个 Agent 能处理哪些问题类型	`fund_quote`、`general`

注意：allowed_tools 是最大权限。配置中心只能减少工具，不能增加工具。

4. 第二步：写 `handle()`

from app.core.contracts import AgentRequest, CoreResult, RequestContext
from app.service.agent.base import BaseAgent


class AdvisorAgent(BaseAgent):
    definition = AgentDefinition(
        agent_type="advisor",
        version="1.0.0",
        allowed_roles=("customer", "advisor", "operator", "admin"),
        allowed_portals=("api",),
        allowed_tools=("query_fund_quote",),
        supported_intents=("fund_quote", "general"),
    )

    async def handle(
        self,
        request: AgentRequest,
        context: RequestContext,
    ) -> CoreResult:
        return CoreResult(text=f"你刚才说的是：{request.message}")

`request` 和 `context` 的区别

request 是用户说的话，例如：

请查询基金 159511 的行情

context 是系统验证后的身份信息，例如：

user_id、roles、permissions、trace_id、customer_ids

身份、客户范围和权限必须相信 context，不要相信用户自己在消息里写的客户编号。

5. 第三步：注册 Agent

在 app/service/agent/bootstrap.py 中添加：

factory.register(
    AdvisorAgent.definition,
    lambda _context: AdvisorAgent(AdvisorAgent.definition),
)

这行代码的意思是：

当系统收到 agent_type=advisor 时，
请使用 AdvisorAgent.definition 检查权限，
然后创建一个 AdvisorAgent 实例。

为什么必须在这里注册？因为 HTTP 服务和 Worker 都从这里拿 AgentFactory。如果你在别的地方注册，可能出现“接口能找到，Worker 找不到”的问题。

6. 系统会自动做什么

你写完 handle() 后，系统会自动完成：

检查请求
→ 检查 JWT、角色和权限
→ 读取配置
→ 读取客户记忆
→ 判断用户意图
→ 执行你的 handle()
→ 检查违规内容和敏感信息
→ 保存结果、审计和事件

所以你不需要在 handle() 里重复写：

JWT 验证；
角色判断；
数据库 Session；
记忆读取；
模型密钥读取；
工具权限判断；
审计日志；
SSE 推送。

7. 如何使用基金行情工具

底座已经提供：

工具名：query_fund_quote
权限码：fund:quote:read

7.1 先在 AgentDefinition 声明

allowed_tools=("query_fund_quote",)
supported_intents=("fund_quote", "general")

7.2 在 `handle()` 中调用

async def handle(
    self,
    request: AgentRequest,
    context: RequestContext,
) -> CoreResult:
    quotes = await self.call_tool(
        "query_fund_quote",
        {
            "fund_codes": ["159511"],
            "limit": 20,
        },
        intent="fund_quote",
        context=context,
    )
    return CoreResult(text=f"行情查询结果：{quotes}")

7.3 为什么必须用 `call_tool()`

因为 call_tool() 会自动检查：

工具是否存在
→ 当前意图是否允许用它
→ 当前角色是否允许用它
→ 当前用户是否有权限
→ 参数是否正确
→ 是否超时
→ 是否写入审计

不要这样写：

import httpx
response = httpx.get("https://api.fund.eastmoney.com/...")

也不要这样写：

from hq import get_southern_fund_market

外部行情必须由底座统一处理。

7.4 如何理解行情结果

重点字段：

nav             基金净值
nav_date        净值日期
daily_change    日涨幅
quote_source    eastmoney / cache / degraded
is_intraday     是否盘中
degraded        是否降级

当 degraded=true 时，应该说“当前数据可能不是最新行情”，不能说“已经成交”或“保证收益”。

8. 如何使用适当性校验

投顾 Agent 不要自己写风险等级比较：

if customer_level >= product_level:
    allowed = True

必须使用公共工具：

decision = await self.call_tool(
    "check_suitability",
    {
        "customer_risk_level": 3,
        "product_risk_level": 3,
        "product_requires_disclosure": True,
    },
    intent="risk_check",
    context=context,
)

返回结果主要看：

allowed                 是否允许
reason_code             原因
required_disclosure     是否需要风险揭示
requires_confirmation   是否需要二次确认
requires_recording      是否需要双录

风险不匹配或测评过期时，必须停止高风险流程并给出安全提示。

9. 如何调用模型

大多数 Agent 不需要自己调用模型，因为底座已经自动完成意图分类。

只有业务确实需要生成一段说明文字时，才使用：

result = await self.generate_with_model(
    approved_endpoints,
    "请根据以下行情生成简短风险说明：...",
)

这里的 approved_endpoints 必须来自底座路由结果，不能自己写模型地址。

禁止：

自己读取 API Key；
自己创建 OpenAI 或其他供应商客户端；
把模型原始回答直接返回；
让模型决定是否下单或修改持仓。

10. 配置中心怎么配

工具白名单的配置格式：

namespace: agent_tools
config_key: advisor:fund_quote
value_json: {"allowed_tools": ["query_fund_quote"]}

配置只能小于等于代码中的 allowed_tools。例如代码没有声明 query_fund_quote，仅靠数据库配置不能让 Agent 获得它。

行情运行配置使用：

namespace: fund_market
config_key: default

配置缺失或格式错误时，底座会使用安全默认值，不要在业务代码里直接读取配置表。

11. 常见错误和解决办法

错误一：`AGENT_TYPE_NOT_FOUND`

原因：没有在 bootstrap.py 注册 Agent。

解决：补充 factory.register(...)，再运行注册表契约测试。

错误二：`FORBIDDEN`

原因可能是：角色不在 allowed_roles、入口不在 allowed_portals、缺少权限或工具没有配置到当前意图。

解决：检查代码声明和已发布配置，不要在 Agent 中绕过检查。

错误三：`RECOVERABLE_ERROR`

常见原因：模型、行情、Redis 等外部依赖暂时不可用。让 Worker 按策略重试，不要把供应商异常原文展示给用户。

错误四：工具找不到

确认三件事：

AgentDefinition.allowed_tools 是否声明
当前 supported_intents 是否包含调用意图
agent_tools/<agent_type>:<intent> 是否发布了工具白名单

错误五：引用校验失败

不要手写 SourceReference。只引用 self.memories 或本次 call_tool() 返回的数据。

12. 测试怎么写

最少要测试：

正常请求
未授权角色
未授权入口
低置信意图
模型输出错误
工具参数错误
工具超时
行情降级
适当性不通过
禁止表达和敏感信息

提交前执行：

python -m pytest -q -p no:cacheprovider
python -m ruff check app tests tools alembic
python -m mypy app
python tools/check_authoritative_docs.py
python tools/audit_schema.py

13. 哪些事情绝对不能做

不要覆盖 BaseAgent.execute()；
不要自己实现鉴权、意图分类、工具权限或审计；
不要直接访问 MySQL、Redis、Milvus、Neo4j；
不要直接调用东方财富接口；
不要代客下单、确认成交或修改持仓；
不要把场外运营数据写进场内交易表；
不要修改已有表名、字段名、类型和既有含义；
不要提交 .env、私钥、Token 或 API Key。

14. 开发完成后的交接模板

Agent 名称：
支持意图：
新增工具：
修改文件：
配置项：
测试命令：
测试结果：
数据库是否变化：
是否涉及场外流程：
已知限制：

如果数据库有需求，必须额外说明：新增了哪些表/字段、如何证明没有修改旧表和旧字段，以及迁移回滚方式。

15. 最后记住

你的业务 Agent 越简单越好：

声明能力
→ 接收 request/context
→ 调用公共工具
→ 组织业务结果
→ 返回 CoreResult

安全、权限、模型、记忆、审计、恢复和外部依赖由底座负责。不要为了“方便”绕过底座，否则代码无法通过契约测试，也会给其他组员和生产运行带来不一致行为。

10 KiB Raw Blame History Unescape Escape