第 8 章:智能体生命周期:生成与裁决的分离
8.1 工程问题的本质:模型输出如何变成可信结果
任何智能体系统的第一性问题都是:模型生成的文本,如何变成可信的业务结果? 业务用户看到的是自然语言答案,平台看到的是模型调用,真正决定答案是否可信的事实,却早在"输出如何被处理"时就已注定。
如果模型输出被直接呈现、直接落库,那么链路中的任何一次幻觉都会成为业务事实:售价比成本线低、订单数量超库存、回复泄露他人成本。排查这类问题时,人们往往归因于"模型不够聪明",但模型只是把链路中缺少的确定性暴露了出来。
设计原理: 可信结果 = 生成(可能出错)× 裁决(必须正确)。将两者分离,是智能体生命周期设计的第一原则。模型负责生成候选内容,确定性机制负责裁决该内容能否呈现、能否落库。分离之后,模型的能力边界与规则的安全边界各自清晰、各自演进。
8.2 统一契约:层间数据的稳定形状
生成与裁决的分离需要一个前提:各层之间传递的数据具有稳定形状。如果层间传递裸字典,字段名、类型、缺省语义在不同实现间漂移,裁决逻辑就无法可靠读取"模型说了什么"。
统一响应契约承载两类信息:内容面(content / data,模型与业务数据的载体)与控制面(action / need_confirm / rules_violated,裁决结果的载体)。控制面的存在使得规则裁决、审批需求等非文本信息可以随响应一起传递,而不必靠解析文本猜测——解析文本是脆弱的行为,文本是给人看的,结构是给系统用的[39]。
class AgentResponse:
def __init__(self, content: str = "", data: Optional[Dict[str, Any]] = None,
action: str = "", need_confirm: bool = False,
rules_violated: Optional[List[str]] = None,
agent_name: str = "", metadata: Optional[Dict[str, Any]] = None):
...
原理延伸: 契约的稳定性不是"文档写得清楚",而是"结构上无法不一致"。AgentResponse 以类而非字典作为载体,将字段的缺省值、类型语义固化在定义处;to_sse_stream() 保持契约不变,仅改变传输方式——契约是数据的形状,SSE 是数据的传输方式。
8.3 规则校验后置:裁决节点为何在生成之后
生命周期将规则校验安排在 LLM 调用之后、返回之前,如图 8-1 所示:
图 8-1 生成与裁决分离的生命周期
graph LR
P1["构建提示词<br/>(注入身份/权限/历史/规范)"] --> P2["LLM 调用<br/>(生成候选内容)"]
P2 --> P3["规则校验(裁决:通过 / 警告 / 阻断)<br/>← 确定性节点"]
P3 --> P4["格式化响应<br/>(按裁决结果输出)"]
原理: 校验必须在生成之后,因为只有生成了内容,校验才有对象;校验又必须在呈现之前,因为一旦呈现给用户或写入系统,错误即成事实。两个时机缺一不可。这与"先斩后奏"或"先奏后斩"都不相同——这里是"拟稿—审稿—发布"的流程制度化。
规则结果的三态处理体现了裁决的粒度,见表 8-1:
表 8-1 规则裁决的三态处理
| 规则结果 | 响应行为 | 语义 |
|---|---|---|
blocked(硬规则阻断) |
覆盖 LLM 输出为阻断说明 | 业务裁决优先于语言生成 |
warn(警告) |
保留输出并追加提醒 | 可继续,但必须显式告知 |
requires_approval |
标记需确认并挂起 | 交给人机协同(第 13 章) |
把生命周期摊开为时序,可看清"裁决"如何嵌在生成之后、返回之前,且每一步都留下痕迹,如图 8-2 所示:
图 8-2 "生成 → 裁决 → 留痕"时序
sequenceDiagram
autonumber
participant U as 用户
participant C as Coordinator
participant A as Agent(BaseAgent)
participant L as LLM
participant R as 规则引擎
participant D as 审计链
U->>C: 自然语言输入
C->>A: 隔离上下文后分发
A->>L: 构建提示词(身份/权限/历史/规范)
L-->>A: 候选内容(生成段)
Note over A,R: 以下为裁决段——确定性节点
A->>R: _apply_rules(业务数据)
R-->>A: RuleResult(pass / warn / block)
alt block(硬规则阻断)
A->>A: 覆盖输出为阻断说明
else warn(警告)
A->>A: 保留输出并追加提醒
else requires_approval
A->>D: 挂起并留痕
end
A->>D: 写入审计(谁/何时/基于什么数据)
A-->>C: AgentResponse(统一契约)
C-->>U: 回复
原理: 时序图把 8.1~8.3 节的三条论断钉在同一张图上——① 生成段与裁决段参与者不同(LLM 与规则引擎各司其职),生成能力再强也无法替代裁决;② 裁决段的输入是业务数据(_apply_rules(data))而非模型文本,故其结论与模型措辞无关;③ 裁决与留痕都在返回之前发生,用户看到的每一条回复都已过确定性节点、都已在审计链上留下记录。
8.4 查询流程编排:链式可信度
除代码内固定的生命周期外,框架还支持数据驱动的一级链——查询流程。链的步骤定义存放在配置中,由执行器逐步骤执行:db 步骤做确定性查询(参数化 WHERE + 权限门禁),llm 步骤做聚焦生成(prompt 引用前步骤结果)。
原理: 链式编排的工程意义不在于"步骤多",而在于每一步骤的信度可单独控制。确定性查询步骤的结果是可信事实,生成步骤的结果是推断观点——将两者分步骤呈现,用户才能区分"系统查到的"与"模型推断的"。这是可信输出在结构上的体现:事实与观点的分离。
# 示例:质量综合分析流程的四步编排(步骤定义存配置,可训练)
步骤1 db :查询 qc_records(质检记录) → 别名 qc
步骤2 db :查询 orders(发货单) → 别名 ord
步骤3 kb :检索客户反馈 / 投诉 → 别名 kb
步骤4 llm :模型汇总(引用 ${qc} ${ord} ${kb}) → 别名 ans
8.5 注入与降级:最小可用实现
框架不直接依赖 LLM 与数据库,外部能力通过鸭子类型接口注入:LLM Provider 提供 call/generate/chat_completion,数据库层提供 query_one/query_many/insert/update,审核日志提供 create/get_by_chain/verify_hash。
未注入时,每一级能力都有确定的最小可用实现(内存、默认配置、兜底文本)。
原理: 降级不是"功能缺失时的将就",而是每一级能力都有确定的最小可信实现。这带来两个直接结果:其一,框架可以在无 LLM、无数据库的 CI 环境完整验证流程逻辑——先验证正确性,再接入真实依赖;其二,生产环境任一外部依赖失效时,系统降级到"仍可回答、不产生错误业务事实"的状态,而非静默失败。
8.6 失败模式与权衡
表 8-2 生命周期设计的失败模式与权衡
| 失败形态 | 根因 | 设计应对 |
|---|---|---|
| 幻觉被当作事实落库 | 模型输出直接写入业务表 | 规则校验后置 + 高风险操作走审批 |
| 层间契约漂移 | 裸字典传递,字段语义自由演化 | 统一响应契约类 + 兼容性基线(API_SPEC) |
| 生成与裁决耦合 | 提示词里"请务必不要违规" | 规则引擎独立裁决,提示词不承担约束责任 |
| 外部依赖失效导致全线不可用 | 关键路径强依赖外部服务 | 鸭子类型注入 + 内存/默认降级 |
权衡: 生成与裁决分离以"多一次确定性调用"为代价(规则校验、字段校验、权限校验逐层叠加),换取"输出可信"。对延迟敏感的场景,框架以规则预检(见 17.3 节)在语义层先行命中,绕过模型调用,将分离的代价压到最低。
8.7 契约不变量与错误处理分层
8.7.1 契约不变量
统一响应契约要真正"稳定",必须显式声明不变量(invariant)——哪些组合在结构上合法、哪些非法。专家的评审习惯是追问:这个契约允许什么不可能状态? 两个核心不变量:
blocked即终态:action="blocked"时,content必须是阻断说明而非业务内容——模型输出已被覆盖,不允许出现"content 是业务结果、action 是 blocked"的混合状态;
- 确认与动作一致:
need_confirm=True时,action必须可被确认动作推进(如require_approval对应审批推进),不允许"标记需确认却无对应动作"的悬空状态。
不变量由框架层的构造逻辑保证(_format_response 按规则结果构造),而非依赖调用方自觉——这与 8.2 节"结构上无法不一致"的原理一致。
8.7.2 错误处理分层
生命周期中的错误按来源分三层,处理策略不同,见表 8-3:
表 8-3 分层错误处理策略
| 错误层 | 示例 | 处理策略 |
|---|---|---|
| 框架错误 | 规则引擎异常、契约校验失败 | fail-closed:默认阻断(第 11 章),宁拒不放 |
| 业务错误 | 参数缺失、权限不足 | 结构化响应(request_info / blocked),可对话纠正 |
| 模型错误 | LLM 返回空串、输出非法格式 | 降级兜底文本,不静默透传错误内容 |
原理: 分层的意义在于"错误的可解释性"——框架错误应表现为明确的阻断原因,业务错误应表现为可对话的引导,模型错误应表现为诚实的兜底("当前无法生成回复")。三层都禁止的,是把错误当作成功返回:任何一层失败都不能以"看起来成功"的形态结束。
8.7.3 流式输出的切割正确性
to_sse_stream() 将内容按标点切割为块:meta → message×n → done。切割的工程约束是不得破坏内容的消费语义:
- 控制面(agent_name / need_confirm / rules_violated / action / metadata)在首个 meta 事件一次性下发,接收方不必等全文即可决定交互方式;
- 内容按句读切割,避免在语义中途截断;对结构化内容(HTML 单据、Markdown),业务侧需保证切割点不会撕裂标记语法——流式是传输优化,不能成为内容破坏者。
8.8 实现剖析:统一契约、规则后置与降级
本节从 runtime/base_agent.py(654 行)摘取三段,印证 8.2 节的统一契约、8.3 节的"规则校验后置"与 8.5 节的降级设计。
8.8.1 统一契约:AgentResponse 的字段即不变量
class AgentResponse:
"""Agent 统一响应对象(在 Coordinator / Agent / API 层之间传递,避免裸 dict)"""
def __init__(self, content="", data=None, action="", need_confirm=False,
rules_violated=None, agent_name="", metadata=None):
self.content = content or "" # 主回复内容
self.data = data or {} # 结构化业务数据
self.action = action or "" # 建议的下一步动作
self.need_confirm = bool(need_confirm) # ★ 高风险操作需二次确认
self.rules_violated = rules_violated or [] # ★ 命中违规的规则列表
self.agent_name = agent_name or ""
self.metadata = metadata or {}
def to_dict(self):
"""序列化为字典,供 API 层返回前端"""
return {
"content": self.content, "data": self.data, "action": self.action,
"need_confirm": self.need_confirm,
"rules_violated": self.rules_violated,
"agent_name": self.agent_name, "metadata": self.metadata,
}
这段代码印证 8.2 节:契约的稳定性体现在"字段即不变量"。无论哪个 Agent、哪条路径,返回给上层的一定是这七个字段;其中 need_confirm 与 rules_violated 是控制面字段——它们把"是否需人工确认""命中了哪些规则"从自由文本里抽出来,变成上层可编程判断的结构化事实。
to_dict() 保证序列化契约与内存契约一致,避免"接口说一套、实现传一套"。
8.8.2 规则校验后置:process() 的四步生命周期
def process(self, user_input, context):
"""Agent 主处理入口
生命周期:接收 -> _build_prompt -> _call_llm -> _apply_rules -> _format_response
"""
# 注入当前用户上下文(S1:线程本地存储,避免并发时用户身份串扰)
BaseAgent._user_local.current_user = context.get("user") or {}
prompt = self._build_prompt(user_input, context) # 1. 构建提示词
llm_output = self._call_llm(prompt) # 2. 调用 LLM(生成)
rule_result = self._apply_rules(context.get("data", {})) # 3. 规则校验(裁决)
response = self._format_response(llm_output, rule_result) # 4. 格式化响应
response.metadata["elapsed_ms"] = round((time.time() - start_time) * 1000, 2)
return response
这段代码印证 8.3 节:裁决节点的位置在代码里是显式的第 3 步——它既不在生成之前(那样模型无从施展),也不在呈现之后(那样违规内容已经出去了),而是卡在"生成"与"呈现"之间。
这正是 8.3 节"拟稿—审稿—发布"的制度化:_call_llm 产出的是"拟稿",_apply_rules 是"审稿",_format_response 才"发布"。用户上下文用线程本地存储注入,是因为规则引擎与审批角色都需要读当前用户,而 Agent 实例会被多线程共享。
8.8.3 规则后置的三种结局:通过、阻断、异常转审批
def _apply_rules(self, data):
"""执行适用的业务规则校验(硬约束,LLM 不可绕过)"""
if not self.applicable_rules:
return _PassRuleResult()
# ★ 安全铁律:这些规则永远走硬编码,不接受引擎覆盖
_HARDCODED_RULES = frozenset({"data_flow_rule"})
for rule_name in self.applicable_rules:
if rule_name in _HARDCODED_RULES: # 硬编码路径
result = registry.get_rule(rule_name).check(data)
if result and not result.passed:
violated.append(rule_name)
final_result = result
if result.blocked: # ★ 硬规则阻断立即终止
break
continue
rule = registry.get_rule(rule_name) # 引擎执行路径
engine_steps = rule.load_config_from_db().get("engine_steps") if rule else None
if engine_steps:
context = {"data": data, "user": current_user()}
result = engine.execute({"engine_steps": engine_steps},
context, params, rule_name)
if result and not result.passed:
violated.append(rule_name)
final_result = result
if result.blocked:
break
continue
# ... 回退硬编码规则(向后兼容)
# ★ 引擎不可用时放行,避免阻断业务流程
这段代码印证 8.3 与 8.5 节:"后置裁决"并非单一路径,而是三层降级——引擎配置存在时走 engine_steps(可训练),配置缺失时回退硬编码 check()(向后兼容),引擎整体不可用时放行(避免阻断业务)。
同时 _HARDCODED_RULES 划出例外:data_flow_rule 永远走硬编码,不接受引擎覆盖——这是"动作能力不可训练"在裁决层的体现(第 11 章)。异常路径统一转 warn(需人工复核)而非静默放行,是 fail-closed 在 Agent 层的落地。
8.9 本章小结
- 可信结果 = 生成 × 裁决,二者必须分离;
- 统一契约将内容面与控制面固化,层间数据形状稳定;
- 规则校验在生成之后、呈现之前,是"拟稿—审稿—发布"的制度化;
- 查询流程将事实与观点分步骤呈现,链式信度可控;
- 降级是每级能力都有最小可信实现的设计原则。