第二部 原理篇:规则驱动的智能体设计

第 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)——哪些组合在结构上合法、哪些非法。专家的评审习惯是追问:这个契约允许什么不可能状态? 两个核心不变量:

不变量由框架层的构造逻辑保证(_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。切割的工程约束是不得破坏内容的消费语义:

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 本章小结