你有没有遇到过这种情况：

智能体跑了几十秒，最后只返回一个词——"执行失败"。

没有报错，没有堆栈，也不知道是哪一步错了。

这不是个案，这是常态。

引子

上周排查一个电商数据分析智能体失败的问题。任务跑了 45 秒，最后返回"执行失败"。没有日志，没有堆栈，没有中间状态。

团队排查了很久才定位到根因：规划阶段 LLM 返回的 JSON 格式错误，解析失败导致后续全部跳过。

如果智能体有日志，这个问题很快就能定位。比如"分析本月销售数据"这个任务，是在查询销售数据库后数据为空？还是在生成报告时格式出错？还是在计算 GMV 指标时数值异常？没有检查点，就只能在黑盒里猜。

黑盒系统最大的问题不是"会失败"，而是"失败了不知道哪里失败"。传统软件有堆栈跟踪，智能体没有。智能体的执行过程是 LLM 的黑盒推理，看不到中间状态。

可测性设计不是"加几个 print"，是系统性地暴露智能体的内部状态。让测试能看到"发生了什么"，而不是只看到"成功了"或"失败了"。

这篇文章讲四件事：日志标准化、状态暴露、确定性控制、检查点机制。

可测性改造的四个维度

┌──────────────────────────────────────────────────────────────┐
│                   可测性改造的四个维度                         │
├──────────────┬──────────────┬──────────────┬──────────────────┤
│  ① 日志标准化  │  ② 状态暴露   │  ③ 确定性控制  │  ④ 检查点机制    │
│  结构化 JSON  │  get_state   │  temp/seed   │  6 个关键节点    │
│  机器可读     │  随时查进度   │  版本锁定     │  断点可验证      │
└──────────────┴──────────────┴──────────────┴──────────────────┘

维度一：日志标准化

日志不是"记一笔"，是结构化记录。每个阶段输出什么、怎么输出、输出给谁看，需要统一规范。

结构化日志的核心是：机器可读、人类可查、按阶段分组。

日志格式：

trace_id（追踪 ID）：就像快递单号——一次任务从头到尾共用同一个 ID，所有日志都能串成一条线，方便排查。

{
    "timestamp": "2026-06-15T10:30:00.123Z",
    "trace_id": "trace_20260615_103000_001",
    "phase": "planning",
    "level": "info",
    "message": "开始规划任务",
    "data": {
        "task": "生成6月销售报告",
        "model": "qwen-plus",
        "temperature": 0.3
    }
}

每个阶段输出的日志内容：

阶段	日志内容	关键字段
规划	任务描述、模型、temperature、子任务列表、依赖关系	task, model, subtasks, dependencies
执行	子任务 ID、工具名、参数、结果、状态	subtask_id, tool, input, output, status
反思	已完成子任务、反思结果（continue/replan/done）、调整建议	completed, reflection_status, adjustments
总结	最终答案、Token 消耗、耗时	final_answer, tokens, elapsed

日志级别：

级别	使用场景	示例
INFO	正常流程	"开始执行子任务 task_1"
WARN	可恢复异常	"工具调用失败，准备重试"
ERROR	不可恢复异常	"规划失败：LLM 返回格式错误"
DEBUG	调试信息	"子任务 task_1 执行耗时 2.3s"

维度二：状态暴露

智能体需要暴露当前状态，让测试能查询"你现在到哪一步了"。

状态暴露接口：

class CustomAgent:
    def get_state(self) -> Dict:
        """
        获取智能体当前状态

        Returns:
            {
                "current_phase": "executing",
                "completed_subtasks": ["task_1", "task_2"],
                "failed_subtasks": ["task_3"],
                "skipped_subtasks": ["task_4"],
                "pending_subtasks": ["task_5"],
                "memory": {"key": "value"},
                "token_count": 1234,
                "llm_calls": 5,
                "elapsed": 12.3,
                "replan_count": 1,
            }
        """
        if not hasattr(self, 'state') or not self.state:
            return {"current_phase": "idle"}

        plan = self.state.plan
        return {
            "current_phase": self.state.current_phase,
            "completed_subtasks": [s.id for s in plan.subtasks if s.status == "success"],
            "failed_subtasks": [s.id for s in plan.subtasks if s.status == "failed"],
            "skipped_subtasks": [s.id for s in plan.subtasks if s.status == "skipped"],
            "pending_subtasks": [s.id for s in plan.subtasks if s.status == "pending"],
            "memory": self.state.memory.copy(),
            "token_count": self.state.total_tokens,
            "llm_calls": self.state.total_llm_calls,
            "elapsed": time.time() - self.state.start_time,
            "replan_count": self.state.replan_count,
        }

状态暴露的意义：

1. 调试：失败时查询状态，定位卡在哪一步

2. 监控：实时查询智能体状态，发现异常

3. 断点验证：执行到一半时查询状态，验证中间结果

维度三：确定性控制

确定性控制：就像给智能体"固定考试条件"——同样的输入，尽量得到同样的输出，方便测试和对比。

智能体的输出受 temperature 和随机种子影响。测试时需要控制这些变量，提高可重复性。

确定性控制的手段：

手段	作用	实现方式
固定 temperature	减少输出随机性	temperature=0.3
固定 seed	提高可重复性	模型支持时设置 seed=42
锁定模型版本	避免行为漂移	使用具体版本 qwen-plus-0813
固定 Prompt	避免 Prompt 变化	Prompt 写入配置文件

温度对稳定性的影响（示例：数学计算类任务上的观测，不同模型/任务不可外推）：

temperature	成功率	输出一致性（示意）
0.3	高	相对稳定
0.7	高	与 0.3 接近或略漂
1.0	视任务而定	更容易漂

（注：请以你锁定的模型版本 + 代表任务集重跑，再写入基线；开放域、工具链、长上下文下温度影响通常远大于上表。）

测试场景建议 temperature ≤ 0.3。生产场景按业务需求定：客服用低温度（稳定），创意用高温度（多样）。

维度四：检查点机制

checkpoint（检查点）：就像给智能体装"行车记录仪"——每一步都留下证据，失败了能回放，而不是只能猜。

检查点是执行过程中的关键节点。到达检查点时输出记录，支持断点验证。

表 1：6 个检查点是什么

检查点	触发时机	输出内容
CP-01: 规划完成	规划阶段结束	子任务列表、依赖关系、并行组
CP-02: 子任务开始	每个子任务开始执行	子任务 ID、工具名、参数
CP-03: 子任务完成	每个子任务执行完成	子任务 ID、结果、耗时
CP-04: 反思完成	反思阶段结束	反思结果（continue/replan/done）
CP-05: 重规划	触发重新规划	原规划、新规划、差异
CP-06: 总结完成	总结阶段结束	最终答案、Token 消耗、耗时

表 2：电商场景怎么用

检查点	电商场景验证示例
CP-01	子任务包含"查询销售数据""计算GMV""生成报告"
CP-02	调用 search（经营类）时 tool_input 含正确月份/品类等检索词
CP-03	查询销售数据后检查返回数据是否包含销售额、订单量、客单价
CP-04	检查数据完整性，决定是否补充查询促销数据
CP-05	数据异常时增加"排查异常原因"子任务
CP-06	生成报告后检查是否包含 GMV 指标、同比环比、结论建议

检查点不是"加几个 print"，是结构化的验证点。每个检查点可以独立验证。

断言：检查点的具体验证规则

检查点是"在哪里验证"，断言是"验证什么"。断言是检查点的核心——它定义了数据或输出的正确性条件。

电商数据分析场景的断言类型：

数据完整性断言

• 查询销售数据后，返回结果包含必要字段：销售额、订单量、客单价

• 报告生成后，包含 GMV 指标、同比/环比数据

• 促销评估报告包含：活动名称、时间段、参与用户数、ROI

数据合理性断言

• 销售额 > 0（负数意味着数据异常）

• 客单价在合理范围（如 10 ~ 10000 元，超出需标记）

• 订单量环比变化不超过 ±80%（异常波动需预警）

• GMV 与销售额在同一统计口径下自洽（口径混用是常见根因，断言前先锁定定义）

输出安全性断言

• 报告不包含用户手机号、身份证等敏感信息

• 报告不包含原始用户明细数据（仅展示聚合结果）

• 报告中的店铺名称已脱敏处理

业务逻辑断言

• 促销 ROI = (促销收入 - 促销成本) / 促销成本，结果应 > 0

• 月度销售趋势与历史数据趋势方向一致（不出现方向性矛盾）

• 各品类销售额之和 ≈ 总销售额（允许 ±1% 的舍入误差）

断言的设计原则：

1. 可量化

   ：断言条件必须是可计算的，不能是模糊描述

2. 有阈值

   ：给出明确的边界值，如"客单价 > 10 且 < 10000"

3. 有分级

   ：区分 ERROR（阻断）和 WARN（标记），不是所有断言失败都需要终止流程

4. 有上下文

   ：断言失败时输出实际值、期望范围、相关数据，方便定位

代码：结构化日志与检查点验证

下面保留三个核心模块的完整实现，集成与演示部分用伪代码说明思路（完整可运行版本见 GitHub 仓库）。

1. StructuredLogger — 结构化日志

class StructuredLogger:
    """结构化日志处理器"""

    def __init__(self, log_file: str = None, trace_id: str = None):
        self.trace_id = trace_id or f"trace_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        self.log_file = log_file
        self.entries = []
        # ... 文件 handler 初始化 ...

    def log(self, phase: str, level: str, message: str, data: Dict = None):
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "trace_id": self.trace_id,
            "phase": phase,
            "level": level,
            "message": message,
            "data": data or {},
        }
        self.entries.append(entry)
        if self.log_file:
            self.logger.info(json.dumps(entry, ensure_ascii=False))

    def get_entries(self, phase: str = None, level: str = None) -> List[Dict]:
        """获取日志条目（支持按阶段/级别过滤）"""
        result = self.entries
        if phase:
            result = [e for e in result if e["phase"] == phase]
        if level:
            result = [e for e in result if e["level"] == level]
        return result

2. CheckpointManager — 检查点管理

class CheckpointManager:
  CHECKPOINTS = {
      "CP-01": "规划完成", "CP-02": "子任务开始", "CP-03": "子任务完成",
      "CP-04": "反思完成", "CP-05": "重规划",   "CP-06": "总结完成",
  }

  def record(self, checkpoint_id: str, data: Dict = None):
      """到达检查点时调用，记录 ID + 附加数据"""
      entry = {
          "timestamp": datetime.now(timezone.utc).isoformat(),
          "checkpoint_id": checkpoint_id,
          "checkpoint_name": self.CHECKPOINTS[checkpoint_id],
          "data": data or {},
      }
      self.checkpoints.append(entry)

  def verify(self, expected: List[str]) -> Dict:
      """验证实际经过的检查点是否包含期望列表"""
      actual_ids = [cp["checkpoint_id"] for cp in self.checkpoints]
      missing = set(expected) - set(actual_ids)
      return {"passed": len(missing) == 0, "missing": list(missing), "actual": actual_ids}

3. EcommerceAssertionValidator — 断言示例

class EcommerceAssertionValidator:
    """电商数据分析断言验证器"""

    def assert_sales_positive(self, sales_data: Dict) -> AssertionResult:
        """断言：销售额 > 0"""
        sales = sales_data.get("销售额", 0)
        passed = isinstance(sales, (int, float)) and sales > 0
        return AssertionResult(name="销售额为正数", passed=passed,
                               actual_value=sales, expected_range="> 0", severity="ERROR")

    def assert_avg_order_reasonable(self, sales_data: Dict,
                                    min_val=10, max_val=10000) -> AssertionResult:
        """断言：客单价在合理范围"""
        avg_price = sales_data.get("客单价", 0)
        passed = min_val <= avg_price <= max_val
        return AssertionResult(name="客单价合理性", passed=passed,
                               actual_value=avg_price,
                               expected_range=f"{min_val} ~ {max_val}", severity="WARN")

    def assert_report_contains_gmv(self, report: str) -> AssertionResult:
        """断言：报告包含 GMV 指标"""
        passed = "GMV" in report or "商品交易总额" in report
        return AssertionResult(name="报告包含GMV指标", passed=passed, severity="ERROR")

    def assert_no_sensitive_info(self, report: str) -> AssertionResult:
        """断言：报告不包含手机号、身份证等敏感信息"""
        # 正则匹配手机号 / 身份证 ...
        passed = not has_phone and not has_id
        return AssertionResult(name="报告无敏感信息", passed=passed, severity="ERROR")

    def run_all(self, sales_data: Dict, report: str) -> Dict:
        """依次运行全部断言，汇总 ERROR / WARN"""
        # assert_field_exists / assert_sales_positive / ... 依次调用
        return {"passed": len(errors) == 0, "errors": len(errors), "details": [...]}

4. 集成与演示（伪代码）

def enhance_agent_with_observability(agent):
    """给已有 Agent 挂上日志 + 检查点 + get_state，不改原有业务逻辑"""
    agent.structured_logger = StructuredLogger()
    agent.checkpoint_manager = CheckpointManager()
    agent.get_state = lambda: {
        "current_phase": agent.state.current_phase,
        "completed_subtasks": [...],
        "checkpoints": agent.checkpoint_manager.get_checkpoints(),
        # ... token_count / elapsed 等
    }
    return agent

# 使用方式
agent = enhance_agent_with_observability(CustomAgent(temperature=0.3))
result = agent.run("生成6月销售报告，查询销售额、订单量、客单价，计算GMV")
state = agent.get_state()                          # 查当前进度
logs = agent.structured_logger.get_entries()       # 查结构化日志
cp_ok = agent.checkpoint_manager.verify(["CP-01", "CP-03", "CP-06"])  # 验证检查点

数据：可测性改造前后对比

指标	改造前	改造后	差异
缺陷定位	靠猜	有日志可查	从盲猜变为可追溯
评测一致性	波动大	波动小	可重复性提升
失败原因可追溯率	0%	100%	从无到有
中间状态可查询	否	是	支持断点验证

改造后的核心收益：失败原因可追溯。没有日志时，失败就是黑盒；有日志后，每个阶段的状态都可查。

1. 日志标准化规范

字段	类型	说明	示例
timestamp	ISO8601	日志时间	2026-06-15T10:30:00.123Z
trace_id	string	追踪 ID	trace_20260615_103000_001
phase	string	阶段	planning/executing/reflecting/summarizing
level	string	级别	info/warn/error/debug
message	string	消息	开始执行子任务 task_1
data	object	附加数据	{subtask_id, tool, input, output}

2. 状态暴露接口定义

def get_state() -> Dict:
    """
    返回智能体当前状态

    字段说明：
    - current_phase: 当前阶段
    - completed_subtasks: 已完成的子任务 ID 列表
    - failed_subtasks: 失败的子任务 ID 列表
    - skipped_subtasks: 跳过的子任务 ID 列表
    - pending_subtasks: 待执行的子任务 ID 列表
    - memory: 记忆字典
    - token_count: 累计 Token 消耗
    - llm_calls: 累计 LLM 调用次数
    - elapsed: 累计耗时（秒）
    - replan_count: 重新规划次数
    - checkpoints: 检查点记录列表
    """

3. 检查点验证清单（电商场景）

验证规则

检查点	触发时机	验证内容	失败影响
CP-01	规划完成	子任务列表非空、依赖关系无环	无法执行
CP-02	子任务开始	工具名在注册表中	跳过该子任务
CP-03	子任务完成	结果非空、状态为 success	后续子任务可能跳过
CP-04	反思完成	反思结果为 continue/replan/done	可能无限循环
CP-05	重规划	新规划包含未完成的子任务	可能遗漏任务
CP-06	总结完成	最终答案非空	输出不完整

电商场景示例

检查点	示例
CP-01	子任务应包含"查询销售数据""计算GMV""生成报告"
CP-02	search  的 tool_input 含正确检索词（经营/报告语义）
CP-03	查询返回数据包含销售额、订单量、客单价
CP-04	检查数据完整性，决定是否补充查询
CP-05	数据异常时增加"排查异常原因"子任务
CP-06	报告包含 GMV 指标、同比环比、结论建议

4. 确定性控制配置模板

# determinism_config.yaml
temperature: 0.3          # 测试建议 ≤0.3
seed: 42                  # 可选，模型支持时设置
model_version: "qwen-plus-0813"  # 锁定版本
max_context_turns: 10     # 上下文窗口
enable_safety_check: true # 安全检测

5. 电商数据断言清单

断言类型	断言条件	严重级别	示例
字段完整性	返回数据包含必要字段	ERROR	销售额、订单量、客单价
数值合理性	销售额 > 0	ERROR	负数意味着数据异常
数值合理性	客单价在 10 ~ 10000 元	WARN	超出范围需标记
数值合理性	订单量环比变化 ≤ ±80%	WARN	异常波动需预警
输出完整性	报告包含 GMV 指标	ERROR	缺失则报告不完整
输出安全性	报告无手机号/身份证	ERROR	敏感信息需脱敏
业务逻辑	各品类销售额之和 ≈ 总销售额	WARN	允许 ±1% 舍入误差

总结

可测性设计不是"加几个 print"，是系统性地暴露智能体的内部状态。

四个维度：日志标准化（结构化 JSON）、状态暴露（get_state 接口）、确定性控制（temperature/seed/版本锁定）、检查点机制（6 个关键节点）。

以电商数据分析智能体为例：查询销售数据后检查返回字段是否完整（CP-03），生成报告后验证是否包含 GMV 指标（CP-06），断言销售额 > 0、客单价在合理范围、报告不包含敏感信息。检查点决定"在哪里验证"，断言决定"验证什么"，两者配合才能覆盖关键质量维度。

改造后，失败原因从不可查到可追溯，每个阶段的状态都有日志可查。

---

如果你也在做智能体，建议你现在就做一件事：给 Agent 加 3 个检查点。

不用重构，不用换框架，只要能回答这三个问题：

1. 它在干什么？
2. 干到哪一步了？
3. 为什么会失败？

这就已经胜过 90% 的黑盒系统。

---

下一篇讲评分机制设计。LLM 评分本身不可信，需要规则评分 + LLM 评分 + 人工抽检的三层校验。

---

面试题模块

Q1：什么是检查点（Checkpoint）？在智能体测试中怎么用？

A：在我实际测试中，检查点是 Agent 执行过程中的"记录点"——记录规划输出、工具选择、执行结果等中间状态。没有检查点，Agent 返回"执行失败"你完全不知道是规划错了还是工具调用错了。有了检查点，你能精确定位到"规划时把工具A选成了工具B"。

Q2：加检查点会影响 Agent 的性能吗？

A：会有影响，但可以忽略不计。检查点的核心是"记录"而不是"计算"——只是把已有的中间状态写到日志或缓存中，不会增加 LLM 调用次数。实测数据显示，以我实测的数据来看，加检查点后每个任务平均多耗时 0.2-0.5 秒（主要是序列化开销）。

Q3：生产中如何权衡"可测性"和"性能"？

A：分级控制。开发环境打开全部检查点（详细日志），测试环境打开关键检查点（规划+工具选择），生产环境只打开简化检查点（错误日志）。通过环境变量控制检查点等级，不需要修改代码。