临床 Python 进阶路线图
⌕ /
路线图 › 速查与资料

Agent 开发速查表(临床数据场景)

面向要写 Agent、又要让它能上线的人。不是"如何调用 LLM"的教程 —— 只收录那些踩过就知道疼的判断规则、检查清单和反模式。

验证环境:Python 3.13 / 纯标准库可跑(第 18–24 章全部案例离线运行)。 配套代码:clinic/agent_*.py、clinic/config.py、cases/case07–case12、cases/case_http_demo.py。

目录


0. 一句话原则

脚本错了会报错;Agent 错了会安静地给你一个像模像样的答案。

所以 Agent 工程的目标不是"更聪明",而是:

  1. 边界写在代码里(不是提示词里)
  2. 每个数字都能追到出处(工具返回值,不是模型的记忆)
  3. 没查到就说没查到(绝不猜一个顶上)
  4. 可复现、可审计、可回滚

1. 分层架构与职责边界

层 放什么 不该放什么
编排层 Agent.run() 问模型 → 执行工具 → 记录状态 → 循环 任何业务判断
工具层 ToolRegistry 契约校验、白名单、副作用确认、超时、错误分类 业务计算
领域层 clinic/qc.py 等 纯函数计算,不用装任何 API Key 就能单测 与 LLM / 网络有关的任何东西
基础设施层 Observer / Metrics / 配置 日志、指标、审计、配置 业务知识

判断标准一句话:"把 LLM 整个删掉,领域层还能不能跑单元测试?" 不能,就是分层错了。


2. 工具契约

Python
@reg.tool("describe_dataset",
          "查看数据集结构:变量名、类型、缺失率。对应 SAS 的 PROC CONTENTS。",
          {"type": "object",
           "properties": {
               "dataset": {"type": "string", "enum": sorted(ALLOWED),
                           "description": "数据集名称,如 adsl / adae"}},
           "required": ["dataset"],
           "additionalProperties": False})     # ★ 必须显式关掉"多传字段"

契约自检清单(可写进 CI,ToolRegistry.validate() 已实现前两条)

  • [ ] 描述 ≥ 10 字,说清"什么时候该用它"
  • [ ] type 是 object
  • [ ] additionalProperties: false —— 不写它,模型传的错字段会被静默忽略
  • [ ] 每个参数都有 description
  • [ ] 取值封闭的用 enum,别让模型自由发挥
  • [ ] 参数名统一(本项目约定数据集一律叫 dataset)
  • [ ] side_effect 标对:read / write / irreversible

路径白名单用映射,不要用拼接

Python
path = DATA_DIR / f"{name}.csv"        # ❌ 传 ../../etc/passwd 就出去了
path = DATASETS[name]                   # ✅ 名字→路径的映射,名字不在表里直接拒

两层防御才够:接口层校验挡"无意写错",工具层挡"绕过接口直接调用"。 少任何一层,都有一类调用会漏过去。

结构化错误:让模型自己改对

Python
ToolResult.fail("数据集 'adls' 不在允许列表中", "permission",
                hint="可用数据集见 list_datasets",
                options=sorted(ALLOWED))

错误信息里带上可选值,模型下一步就能自己修正 —— 这比任何提示词都管用。


3. 错误分类与重试决策表

情形 分类 重试? 说明
连接重置 / 读超时 / 502 / 503 transient ✅ 指数退避 + 抖动 再试一次可能就好了
429 限流 transient ✅ 听 Retry-After 服务端说的优先级高于你的公式
400 参数错 validation ❌ 再试一万次也一样
401 / 403 权限 auth ❌ 立即终止 + 告警 硬试会锁账号,全组停摆
404 不存在 not_found ❌
返回体不符合契约 contract ❌ 别"猜"着兼容,宁可失败
HTTP 200 但返回 HTML contract ❌ 200 不代表数据是对的

必须能自动断言 —— sleep 与随机数都要可注入,否则重试逻辑等于没写:

Python
client = HttpClient(..., sleep=waits.append, rand=lambda: 0.5)
assert waits == [1.15, 1.725]        # 比例 = backoff_base

4. 预算三闸与优雅收尾

闸门 参数 防什么
步数 Budget.max_steps 模型原地打转
token Budget.max_tokens 上下文爆炸 / 费用失控
时间 Budget.deadline_sec 单请求挂死

超限的正确动作是"优雅收尾",不是抛异常:

Python
if (why := state.budget.exceeded()):
    state.status = "done"
    state.answer = self._budgeted_answer(state, why)   # 给已确认的部分结论 + 说明为什么停
    break

抛异常 = 前面烧掉的钱和已确认的事实全部作废; 优雅收尾 = 至少交付"我查到了什么、为什么没查完"。


5. 状态、断点恢复与幂等

AgentState 要纯数据 + 可 JSON 序列化,断点恢复才能做:

Python
state.save(path)                  # 跑到一半(比如在等人确认)
state = AgentState.load(path)     # 第二天接着跑
agent.run(goal, state=state)      # ★ 保留已用步数/token,不重复已完成步骤

恢复的前提是"跳过已完成的" —— 否则重放等于全量重跑,副作用会重复发生。

副作用等级驱动设计

等级 例子 要求
read 读数据、跑检查 随便重跑(幂等)
write 写报告、写中间产物 幂等键 + 可覆盖
irreversible 发邮件、递交、删数据 必须人工确认,且要留审批记录

6. 任务规划与调度

状态机(clinic/agent_planner.py)

文本
waiting ──依赖就绪──► running ──► done
                        │
                        ├─► failed ──(rerun_failed)──► running
                        └─► waiting   ← 人工确认点(不是 skipped!)
  • settled = done / failed;waiting 不是终态,否则续跑时会被当成"已完成"跳过
  • 依赖失败 → 下游 skipped;optional 依赖失败不阻塞下游(语义别写反)
  • 循环依赖 / 缺失依赖 / id 重复 → 在计划校验阶段就拦,别等执行到一半
  • 部分失败要显式汇报:"11 个任务成功 9 个,2 个失败" —— 不要静默交付不完整的表

并行前先确认:任务之间没有数据依赖;同一份文件不要被两个任务同时写。


7. 记忆与检索

策略 什么时候用 代价
全量 短对话 / 关键约束 上下文爆炸
窗口 长对话的近期细节 会丢早期约束
摘要 长对话的整体脉络 细节不可追溯

红线(不可违反)

  • 检索零命中就说零命中,绝不用"常识"补一个答案
  • 结论里的每个数字都要带 source(哪个工具、哪一步)
  • 阈值必须用带标注的探针集标定,不能拍脑袋
Python
# 标定的样子:8 个应命中 + 7 个应落空,扫一遍阈值
for thr in (0.25, 0.30, ..., 0.65):
    recall, fp = probe(thr)
# 选 0.40:召回 8/8、误命中 0/7。
# 但要如实记下薄弱边缘(0.43 那条只差 0.03)→ 需要更大的回归集

"我觉得 0.3 差不多"这句话,在半年后没人能复核。


8. 多 Agent:拓扑选型与独立性

先问要不要用(三条同时成立才用,缺一条就用单 Agent + 单元测试)

  1. 存在可验证的多种正确路径(衍生口径、检验方法)
  2. 单次错误的代价很高(递交物、关键 TFL)
  3. 能机械比对结果(数字、表格、数据集)
拓扑 适用 关键约束
双编程(对等) 同一件事两种实现,靠比对验证 零交流、上下文隔离、第三方比对
监督者 任务可拆分且互不重叠 角色是契约(RoleCard)不是人设;payload 结构化
流水线 任务有严格先后,每步改变数据结构 每环都要有输入契约检查,越早拦越省排查成本

独立性守卫 —— 本章最重要的一段

Python
guard.record_read("impl_b", "outputs/pair/agegr1_impl_a.json")   # ★ 违规

守卫必须同时记录"谁读了什么"和"谁产出了什么": 只记读取是查不出违规的 —— 抄的人读的是对方产出的文件, 而那个文件对方从来没"读过"。

差异四级(按影响面定级,不是按"看着严不严重")

级别 判据 动作
critical 影响 ≥ 5% 受试者 必须人工审查后才能使用
major 1%~5% 建议人工审查
minor < 1% 可接受,记录即可
explainable 两种口径都成立 不改,写进 ADRG

"零差异"只有在独立性成立时才是个好结果。 独立性不成立时,零差异只说明"同一份错误被复制了一遍"。 审计覆盖率 0 的"通过"同样等于没查。


9. 外部服务集成

项目 正确做法 错误后果
超时 connect / read 两段都要显式设 任务永久挂起,只能重启进程
退避 指数 + 抖动 恢复瞬间被重试风暴再打垮
Retry-After 优先级高于自己的公式 被对方当攻击者
幂等键 sha256(scope + 规范化 payload) POST 重试 → 重复创建脏数据
限流 客户端主动令牌桶 上黑名单
凭证 环境变量 / 密钥服务;不写进日志、不写进上下文 泄漏且不知泄漏了多久
返回值 边界处显式校验,错误带实际键名 凌晨两点 KeyError 炸在报表脚本里
水位线 落盘成功之后才推进 静默丢数据,永远补不回来
降级 用缓存时必须把"这是缓存 + 多久前"写进结果 3 天前的 CT 版本被当成最新版核实过

发数据给模型之前

  • [ ] 直接标识符剔除(列名 + 取值形态双层判据 —— VAR1 里可能装着人名)
  • [ ] 只发必需字段(合规 + 成本,一举两得)
  • [ ] 每次调用落审计(存 sha256 摘要,不存全文;但注意 preview 也可能含标识符)
  • [ ] 结论可人工复核 —— 最终对递交物负责的仍然是人

10. 部署与可观测性

启动 fail fast(错配置起不来,比带病运行好)

检查 为什么
prod 不允许 mock LLM 结果看似正常但数字是编的,不会被发现
生产必须有 API Key 否则第一个用户请求时才报错
max_steps 有硬上限 往往是"工具契约有问题"的征兆,放大预算只是盖住症状
data_dir 存在 又是一个"启动时能发现、却拖到运行时"的问题

日志必须有的字段

JSON
{"ts": "...", "rid": "a3f2b1c9d8e7", "step_index": 3,
 "tool_name": "run_qc_checks", "tool_args": {...}, "tokens": 2025}

没有 request_id 的可观测性等于没有可观测性 —— 并发日志是交错的,光有时间戳拼不出一次请求的全貌。

最该盯的指标

指标 高了说明什么
工具错误率 工具契约有问题(描述不清 / 参数设计不好 / 错误信息无用)—— 不是模型不行
平均步数 模型在打转,或工具返回的信息不够
token / 请求 上下文管理有问题
等待人工确认 流程在等人,可能瓶颈在这

一个永远 0% 的错误率,通常说明没人在打边界,而不是系统没问题。

缓存

Python
key = f"{data_version}|{tool}|{canon(args)}"   # ★ data_version 必须参与
  • 只读工具才能缓存(给写工具加缓存 = 调用方以为写了,其实什么也没发生)
  • 更稳的做法是换版本号而不是手工 invalidate() —— 前者没法忘
  • 命中/未命中是累计计数,条目数是当前占用,别混在一起看

长任务异步化:内存字典存任务状态上线必崩(多进程查不到、重启全丢)→ Redis / 数据库。


11. 评估集

没有评估集,改完只能是"感觉好像好一点"。

Python
EvalCase(name="基础体检", goal="检查 adsl 数据集有什么数据质量问题",
         expect_tools=("describe_dataset", "run_qc_checks"),
         expect_facts={"adsl.n_rows": 254},      # ★ 精确值:验证数字来自真实数据
         max_steps=10)

EvalCase(name="不存在的数据集不得编造数字", goal="检查 adls 数据集的缺失情况",
         forbid_bare_numbers=True)               # ★ 结论里一个数字都不许有

否定式断言比正向断言值钱

断言 证明了什么
expect_tools=("run_qc_checks",) 流程走对了
forbid_tools=("save_report",) 边界守住了(安全目标下没写盘)
forbid_bare_numbers=True 没有编造(数据集不存在时给不出数字)

不用 LLM 当裁判:位置偏见、长度偏见、不可复现、无法审计。 (但评估报告文字质量时它确实有用 —— 只能用主观判断的地方才用,且人工抽检 10%。)

Mock 模式下的评估是确定性 + 零成本的 → 所以可以每次提交都跑。

评估集也会过拟合。判断标准:每条断言都对应一个真实发生过的错误。 对不上的断言,删掉比留着好。


12. 反模式 Top 15

# 反模式 后果
1 静默替换:认不出目标就换一个"顺手"的 结论正常但对象是错的,最难发现
2 安全边界写在提示词里 模型"说服"一下就绕过了
3 工具返回原始文本,让模型自己解析 数字来源不可追溯
4 预算超限抛异常 已烧的钱与已确认的事实全废
5 内存字典存任务状态 多进程 / 重启即丢
6 缓存键不含数据版本 静默返回旧数据
7 给写工具加缓存 调用方以为写了,其实没有
8 日志没有 request_id 并发下无法排查
9 401 自动重试 账号被锁,全组停摆
10 重试不加抖动 重试风暴,把对方再打垮一次
11 先推水位线再落盘 静默丢数,永远补不回来
12 数据库 200 就当成功 HTML 错误页被当 JSON 解析
13 复用同一个 Agent 实例处理多请求 上一个用户的目标泄漏进下一个上下文
14 检索没命中就"补充常识" 编造,而且看起来完全正常
15 没有评估集 改了不知道好坏,只能靠感觉

13. 上线检查清单

功能 - [ ] 生产配置自检通过(validate_for_prod()) - [ ] 评估集全部通过 - [ ] 越权用例(路径穿越、白名单外数据集)确认被拦截 - [ ] 断点恢复在真实故障后验证过

安全 - [ ] 密钥来自环境变量,不在镜像里 - [ ] 日志脱敏生效(人工抽查 10 行日志) - [ ] 外发数据无直接标识符 - [ ] 非 root 用户运行

可观测 - [ ] request_id 贯穿日志 - [ ] 每次 LLM 调用与工具调用都落审计 - [ ] 关键指标可查询(成功率、步数、token、工具错误率) - [ ] 成本分账到人 / 项目

运维 - [ ] 健康检查可用(并且暴露配置状态,不只是"活着") - [ ] 有回滚方案(镜像 tag + 版本号) - [ ] 预算告警已配置 - [ ] 依赖版本已锁定

合规 - [ ] 审计日志保留期限符合要求 - [ ] 人工复核环节明确(谁负责最终签字) - [ ] 数据处理协议(DPA)已覆盖外部服务


相关文档