Agent工作流的可观测性设计：Trace追踪、断点重试与异常熔断机制-尧图网络科技

Agent工作流的可观测性设计：Trace追踪、断点重试与异常熔断机制

一、Agent的黑盒困境：为什么它卡住了？在哪一步？

多Agent工作流在生产环境面临的核心困境是可观测性缺失：执行失败时不知道在哪个Agent卡住、哪个工具的调用超时、为什么循环了8次还没终止。在一个生活任务处理Agent的故障排查中，一条"帮我规划周末行程"的请求因为调用了天气API但未设置超时，导致整个工作流挂起12分钟——而日志只显示"执行中"。

可观测性不是锦上添花，是Agent工作流生产化的必要条件。以下设计围绕三个核心能力：分布式Trace追踪（定位瓶颈）、断点重试（从失败步骤恢复而非重头执行）、熔断降级（保护系统免受级联故障）。

二、Agent可观测性的三层指标

Trace层使用OpenTelemetry在Agent、工具和LLM调用三个粒度上创建Span，构建调用链的全局视图。Metric层实时统计每个Agent的延迟、成功率和Token消耗——这些都是可设定告警阈值的指标。Log层记录决策过程（"Agent选择了工具X，参数为Y，原因是Z"），用于事后故障复盘。

三、断点重试与熔断的关键实现

# agent_observability/circuit_breaker.py """Agent级熔断器 设计意图： 1. 半开状态的设计：熔断后定期探测服务恢复情况 2. 每个Agent独立熔断，不影响其他Agent的正常工作 3. 熔断期间的请求直接返回fallback结果，而非报错 """ import time from enum import Enum from dataclasses import dataclass from functools import wraps from typing import Callable, Any class CircuitState(Enum): CLOSED = "closed" # 正常 OPEN = "open" # 熔断中 HALF_OPEN = "half_open" # 探测恢复 @dataclass class CircuitConfig: failure_threshold: int = 5 # 连续失败N次后熔断 recovery_timeout: float = 30.0 # 熔断30秒后尝试恢复 half_open_probes: int = 2 # 半开状态允许2次探测 class AgentCircuitBreaker: """Agent熔断器""" def __init__(self, agent_name: str, config: CircuitConfig | None = None): self.agent_name = agent_name self.config = config or CircuitConfig() self.state = CircuitState.CLOSED self.failure_count = 0 self.last_failure_time = 0.0 self.probe_count = 0 def call(self, func: Callable, *args, **kwargs) -> Any: """执行Agent调用，熔断期间返回None""" if self.state == CircuitState.OPEN: if self._should_attempt_recovery(): self.state = CircuitState.HALF_OPEN self.probe_count = 0 else: # 熔断中，直接返回fallback return self._fallback_response() try: result = func(*args, **kwargs) self._on_success() return result except Exception as exc: self._on_failure() raise exc def _on_success(self): """调用成功：重置熔断器""" self.failure_count = 0 if self.state == CircuitState.HALF_OPEN: self.state = CircuitState.CLOSED def _on_failure(self): """调用失败：累加计数，必要时触发熔断""" self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.config.failure_threshold: self.state = CircuitState.OPEN def _should_attempt_recovery(self) -> bool: """判断是否到了尝试恢复的时间""" return (time.time() - self.last_failure_time > self.config.recovery_timeout) def _fallback_response(self) -> dict: """熔断期间的降级响应""" return { "status": "degraded", "message": f"Agent {self.agent_name} 暂时不可用，请稍后再试", } # agent_observability/retry.py """断点重试管理器 设计意图： 1. 工作流执行到步骤N失败时，从失败步骤重试而非从头开始 2. 每个步骤的执行结果保存为checkpoint 3. 重试只重新执行失败的步骤和后续步骤 """ from typing import Any import json import hashlib class CheckpointManager: """工作流断点管理器""" def __init__(self, workflow_id: str): self.workflow_id = workflow_id self.checkpoints: dict[int, dict[str, Any]] = {} self.completed_steps: set[int] = set() def save(self, step_index: int, input_data: dict, output_data: dict): """保存步骤检查点""" checkpoint = { "step": step_index, "input_hash": self._hash(input_data), "output": output_data, } self.checkpoints[step_index] = checkpoint self.completed_steps.add(step_index) def resume_from(self) -> int: """返回应该从哪个步骤恢复执行（已完成步骤的最大index+1）""" if not self.completed_steps: return 0 return max(self.completed_steps) + 1 def get_checkpoint(self, step_index: int) -> dict | None: """获取某个步骤的输出结果，用于后续步骤的输入""" cp = self.checkpoints.get(step_index) return cp["output"] if cp else None def _hash(self, data: dict) -> str: """对输入数据计算哈希，用于检测输入是否变化""" serialized = json.dumps(data, sort_keys=True, default=str) return hashlib.md5(serialized.encode()).hexdigest()[:8]

CheckpointManager在工作流的每个步骤完成后保存检查点。重试时跳过已完成的步骤，直接从失败位置继续执行——对于有5个步骤的工作流，步骤4失败只需重试步骤4和5，而非重新执行全部5个步骤。

四、可观测性的开销控制

OpenTelemetry的Span采样率在生产环境中需要控制。全量trace（100%采样）在高峰期会产生每分钟约12万条Span——Jaeger的后端存储压力巨大。通过Head-based Sampling设置30%的采样率，同时使用Tail-based Sampling确保错误trace 100%保留。

日志存储是另一个隐形成本。每个Agent每次调用产生的完整日志（包括Prompt和响应内容）约3KB，日调用10000次就是30MB的日增日志量。通过设置日志保留策略（Info级别保留7天，Debug级别保留24小时），平衡问题排查的需求和存储成本。