多 Agent 可观测性与调试

多 Agent 系统的监控与问题排查

🧩 1. 多 Agent 系统的调试为什么比单 Agent 难？有哪些挑战？

如果把单 Agent 比作一辆车，调试它就是检查发动机的运转。而多 Agent 系统则像一个十字路口的多车协同，问题往往不出在某一辆车本身，而出在它们之间的交规、信号和视野盲区。具体来说，挑战来自四个维度：

挑战一：失控的上下文传递与记忆污染

在多 Agent 协作中，Agent A 的输出会成为 Agent B 的输入。如果 Agent A 在一个关键数字上出现了幻觉，Agent B 会基于这个错误信息继续推理，最终答案完全跑偏。在调试时，你要一层层回溯，去找到“哪一个 Agent 最先说错了一句话”。这种错误传播链比单 Agent 的幻觉更难定位。

挑战二：不可复现的涌现行为

多 Agent 通过对话驱动，一个微小的 Prompt 差异、一条消息的顺序改变，都可能导致整个群聊走向完全不同的方向。昨天还能完美运行的流程，今天可能因为 LLM 的一次随机采样而在某个路口拐错了弯。这种非确定性让“在本地复现线上问题”变得几乎不可能。

挑战三：工具调用的权限与依赖交织

Agent A 调用了工具 X 得到结果，Agent B 需要工具 Y，但工具 Y 的调用参数来自 Agent A 的结果。如果工具 X 超时，Agent A 可能返回了一个空结果，Agent B 拿着空参数去调工具 Y，导致一连串无意义的报错。调试时要同时看工具链的调用顺序、参数流转和超时设定，复杂度成倍上升。

挑战四：并发与死锁

在多 Agent 群聊里，一个 Agent 等待另一个 Agent 的回复，而对方又在等待它提供更多信息，形成逻辑死锁。或者，Supervisor 将任务分派给多个 Worker，Worker 之间产生竞争条件。这些并发问题在单 Agent 中根本不存在。

工程上的应对思路：

• 全链路 Trace：为每一次跨 Agent 的消息传递加上唯一的 Trace ID，让所有调用形成一个树状结构。

• 快照与重放：将每一轮对话的状态序列化保存，出现问题时能像“黑匣子”一样回放。

• 单元化 Agent：尽量让每个 Agent 的职责单一，输入输出可独立验证，减少依赖。

---

🔍 2. 如何为多 Agent 系统设计 Trace 和日志体系？

针对上面的挑战，Trace 和日志不只是“记录发生了什么”，更是我们理解多 Agent 思维过程的唯一窗口。我习惯用“一次对话，一个 Trace；每个 Agent，一个 Span；每次工具调用，一个子 Span”的模式。

整体架构图

设计方案：OpenTelemetry + 自定义日志上下文

我们使用 OpenTelemetry 作为 Trace 标准，并在每个 Agent 的“边界”手动创建 Span。同时，日志里必须携带 trace_id 和 span_id，这样在 ELK 或 Grafana 里能把一次对话的所有日志瞬间串起来。

示例代码：为 Agent 调用包装一个带 Trace 的上下文管理器

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import uuid, logging, json, time

# 初始化 Tracer
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
tracer = trace.get_tracer(__name__)

class TracedAgent:
    def __init__(self, name, llm, tools):
        self.name = name
        self.llm = llm
        self.tools = tools

    async def run(self, task_input, parent_context=None):
        # 为当前 Agent 创建一个 Span
        with tracer.start_as_current_span(
            f"Agent.{self.name}",
            context=parent_context,
            attributes={"agent_name": self.name}
        ) as span:
            # 将 trace_id 注入日志
            trace_id = span.get_span_context().trace_id
            logger = logging.LoggerAdapter(
                logging.getLogger(),
                {"trace_id": f"{trace_id:032x}", "agent": self.name}
            )
            logger.info(f"开始执行任务: {task_input[:100]}...")

            # 在这里执行 Agent 的逻辑，并给工具调用单独创建 Span
            with tracer.start_as_current_span("LLM_Call") as llm_span:
                start = time.time()
                response = await self.llm.generate(task_input)
                llm_span.set_attribute("model", response.model)
                llm_span.set_attribute("tokens", response.usage.total_tokens)
                llm_span.set_attribute("latency_ms", (time.time() - start) * 1000)
                logger.info(f"LLM 调用完成，消耗 Token: {response.usage.total_tokens}")

            # 模拟工具调用
            if "需要搜索" in response.content:
                with tracer.start_as_current_span("Tool.search") as tool_span:
                    result = await self.tools["search"](task_input)
                    tool_span.set_attribute("result_count", len(result))
                    logger.info(f"搜索完成，找到 {len(result)} 条结果")
                    # 返回结果给大脑...

为什么这样设计？

• 每个 Agent 是一个 Span：在 Jaeger 或 Tempo 里，你能看到整条 Trace 下，各个 Agent 是并行还是串行，哪个 Agent 耗时最长，一目了然。

• 每次 LLM 调用和工具调用都是子 Span：这样你能精确统计每个步骤的耗时和 Token 开销。

• 日志带 Trace ID：当用户投诉“第三轮回答错了”，你在 ELK 里输入 trace_id:abc123，这个请求的全部“思考过程”就完整呈现，不再需要靠猜。

日志级别与内容的特殊设计：

• INFO：记录每个 Agent 的启动、结束，关键决策（如“决定调用搜索工具”）。

• DEBUG：记录 Agent 内部完整的 Prompt 和 LLM 原始回复（要注意 PII 脱敏）。

• ERROR：工具调用失败、LLM 返回格式解析失败等。

• AUDIT：任何触及敏感数据或重要决策的操作，单独记录，不可删除。

调试多 Agent 的一个核心习惯：把每一次对话的 Trace 都保存下来，做成“可回放的录像带”。你可以在开发环境里把线上 Trace 的每一步重新喂给 Agent，精确复现当时的状态，这比任何日志都更接近真相。

---

📊 3. 多 Agent 系统上线后，如何做性能监控和异常告警？

多 Agent 的性能监控，不能只盯着 CPU 和内存，我们要同时看业务健康度、调用链成本、安全边界和外部依赖。我习惯建立五个维度的看板，每条告警规则都直接关联业务影响。

监控维度图谱

维度一：应用级健康（红色告警）

• 任务成功率：Agent 最终给出有效答案的比例。如果这个值从 95% 掉到 80%，马上响警报。

• 单次任务端到端延迟：P50/P95/P99 耗时。如果 P95 超过 10 秒，用户体验会显著下降。

• 消息队列积压：如果用异步任务，积压数代表系统正在过载。

维度二：成本与效率（橙色告警）

• 每任务平均 Token 消耗：按模型和类型（输入/输出）拆分。若某 Agent 突然开始消耗大量 Token（比如陷入死循环），要立即发现。

• 平均推理步数：一个任务 Agent 走了多少轮。步数飙升通常意味着 Agent 在兜圈子或无法解决某个子问题。

维度三：协作质量（黄色告警）

• 工具调用成功率：按工具名分面。如果某个关键 API 的失败率突增，说明外部依赖出问题了。

• Agent 间通信次数：Supervisor 模式的“群聊”里，有效决策前的发言次数。突然变多可能意味着 Agent 们在推诿或无法达成一致。

示例代码：用 Prometheus 暴露指标

from prometheus_client import start_http_server, Counter, Histogram, Gauge

task_counter = Counter('agent_task_total', 'Task outcomes', ['status'])
llm_tokens = Counter('agent_llm_tokens', 'Token usage', ['model', 'type'])
task_duration = Histogram('agent_task_duration_seconds', 'Task latency')
tool_calls = Counter('agent_tool_calls', 'Tool calls', ['tool', 'status'])
active_tasks = Gauge('agent_active_tasks', 'Running tasks')
interaction_steps = Histogram('agent_interaction_steps', 'Steps per task')

start_http_server(9090)

# 在任务完成时
task_counter.labels(status='success').inc()
task_duration.observe(elapsed)
interaction_steps.observe(step_count)

Grafana 告警规则示例（PromQL）：

• 任务成功率过低： sum(rate(agent_task_total{status="success"}[5m])) / sum(rate(agent_task_total[5m])) < 0.85 持续 5 分钟就触发 PagerDuty。

• 某 Agent 陷入死循环（步数过多）： histogram_quantile(0.95, rate(agent_interaction_steps_bucket[5m])) > 20 说明 95% 的任务在 20 步内无法结束，需要检查该 Agent 的 Prompt 或工具调用逻辑。

• 成本异常飙升： sum(increase(agent_llm_tokens[1h])) > 100000 一小时内 Token 超过 10 万，可能需要限流或检查循环。

异常告警的分级与处理：

• Critical（致命）：任务成功率低于阈值、LLM 全部返回 5xx、敏感词护栏被绕过。这些需要立刻人工介入。

• Warning（警告）：P95 延迟上升、Token 消耗异常、某工具可用率下降。可以先观察，如果持续则升级。

• Info（信息）：新模型上线后的效果对比、流量波动。用于日常复盘，不告警。

告警的最终目的是减少人工阅读日志的时间。一条好的告警应该直接告诉你：哪个 Agent、在什么环节、出了什么问题、影响了多少用户。比如：

“[Critical] Agentic RAG 系统任务成功率降至 76%（低于 85% 阈值）。过去 5 分钟失败 34 次。请查看 Grafana 面板或查询 trace ID: abc123 的失败案例。”

收束：多 Agent 的可观测性建设，本质上是把一群“黑箱”智能体之间的每一次握手、每一次等待、每一次误解，全部翻译成可量化的信号。当某个信号开始尖叫时，你能在用户察觉到问题之前，就顺着 Trace 这条藤蔓，摸到问题的根——是 Prompt 的歧义，还是工具的暂时失联，抑或是那群 Agent 在某个问题上陷入了永无止境的辩论。这就是工程化 AI 带来的真正掌控力。

---