生产环境中的 AI 可观测性：使用 MLflow 洞察您的多智能体系统

构建智能体应用是容易的部分。难的是如何确保它不会意外清空生产数据库，或者将敏感数据泄露到公共 Slack 频道中。

我们都经历过这种情况：你在 Playground 中微调新的设置，感觉一切顺利，突然间你的智能体开始在公司实时信息流中“幻觉”出一些无意义的内容。直到同事发来一个困惑的表情符号，你才注意到它。虽然这在开发阶段只是个有趣的段子，但在生产环境中却是一场灾难。这证明了你既不知道你的智能体在做什么，也绝对不知道为什么会这样。

当你从简单的聊天机器人转向真正触及业务逻辑的自主系统时，零散的追踪和凭空猜测将成为严重的隐患。要想安全地将智能体部署到实际环境中，你需要掌握三件事：

• “是什么”：识别多智能体系统出现故障的各种离奇、新颖的方式。
• “怎么做”：构建所需的可观测性，以实时查看系统状态。
• “为什么”：精准定位关键指标，以便在事情偏离轨道之前引导系统。

MLflow 可以帮助你弥合这一差距，将神秘的智能体工作流黑盒转化为透明的、任务关键型的部署。

多智能体系统是如何失效的：为什么仅有监控还不够

在深入探讨多智能体部署之前，让我们先来看看单个智能体是如何失效的。任何发布过大语言模型（LLM）应用的人，对这些失效模式都不会感到陌生：

• 幻觉：模型凭空捏造数据，特别是在工具返回空结果或模糊结果时，模型会自行填补空白而不是上报错误。
• Schema 脆弱性：API 响应的微小变化（例如重命名了字段、新增了嵌套对象）就会在无形中破坏智能体解析输出并执行操作的能力。
• 上下文衰减：埋藏在长提示词深处的指令实际上会被忽略。模型只关注最近的上下文，从而丢掉了之前约束条件的线索。
• Token 消耗失控：智能体陷入重试循环——一次又一次地重新运行已经失败的同一个工具调用——白白烧掉 Token 和预算，却毫无结果。

现在，上述所有情况同样适用于多智能体系统中的单个智能体。但协同层引入的故障在本质上是完全不同的。

最常见的是级联错误。想象一下，您的编排器将一项财务计算委托给一个专业智能体。该专业智能体完成了任务，但误解了业务约束条件——例如，本应使用年度收入，它却使用了季度收入。接着，三个下游智能体将这个数字纳入了它们各自的分析中。当最终输出呈现给人类时，错误已经通过四个独立的推理步骤进行了累积。其结果看起来似乎合情合理，没人会立刻察觉到问题。

这之所以难以调试，并不是因为错误本身，而是因为故障起源于三步之前一个你并未关注的 Span。当问题完全出在另一个智能体的输入端时，你却在输出端耗费了数小时。

一个相关的问题是共享内存污染。当智能体将中间结果写入共享上下文时，一个智能体的幻觉就会变成后续智能体进行推理的“事实”。这种衰退是渐进的，而非突发性的，因此特别难以捕捉——你会注意到质量在下降，但无法准确指出它是什么时候开始的以及为什么会这样。

第三种值得了解的失效模式是智能体陷入等待循环。编排器在等待专业智能体的响应；而该专业智能体又在等待一个已经默默超时的工具。双方都没有抛出明显的错误。你的可观测性基础设施显示延迟增加，但没有任何迹象表明系统实际上已经停滞。

大语言模型（LLM）的非确定性使得这一切比传统软件中的同等故障更加棘手。将同一个提示词运行一百次，你会得到略有不同的输出。再加上每次重试的成本，这意味着多智能体系统中的故障不仅更难复现，而且发现成本更高。

从监控到可观测性：使用 MLflow 对 CrewAI 进行插桩

对于单个智能体，监控输入和输出通常就足够了。你知道输入了什么，也知道输出了什么，如果有什么不对劲，你只需阅读提示词和响应即可。调试是线性的。

但对于多智能体系统，这已经远远不够了。你不仅需要了解每个智能体返回了什么，还需要了解编排器为什么做出这样的委托决定、输出是如何在智能体之间流转的，以及链条中的哪个环节开始出现问题。监控单个输入和输出只能为你提供数据点；而跨完整 Trace（追踪）的可观测性才能为你呈现完整的故事。

分析单元从提示词转移到了状态转换。要在生产环境中理解多智能体系统，你需要对以下内容具备可见性：

• 编排与路由决策：为什么编排器会将这项任务分配给这个智能体，而不是另一个？考虑到任务要求，这种委派是否正确？
• 智能体间的数据流：智能体之间究竟传递了什么，是否完好无损地送达？这就是级联错误和内存污染变得肉眼可见的地方。
• 每一步的延迟：哪个智能体或工具是瓶颈？系统在哪些不该耗时的地方浪费了时间？

为了解决这些问题，我们首先需要看清系统内部正在发生什么。这意味着我们需要实现不仅能捕获单个工具调用，还能捕获完整的、嵌套且相互关联的智能体交互图的追踪（Tracing）。

图 1. MLflow UI 中的嵌套追踪。

在标准的 LLM 调用中，有起点和终点。但在多智能体工作流中，会有嵌套的 Span 和分支逻辑。MLflow 追踪（Tracing）允许我们通过捕获编排器（Orchestrator）与工作智能体（Workers）之间的父子关系来重建该图形。这能将黑盒执行转化为一个可导航的状态转换图。

开箱即用与自定义追踪

在 MLflow 中，只需一条命令即可启用全面的追踪。

mlflow.<framework>.autolog()

由于 MLflow 开箱即用支持多种类型，因此对 CrewAI 应用程序的监控将如下所示：

import mlflow

# Turn on auto tracing by calling mlflow.crewai.autolog()
mlflow.crewai.autolog()

# Optional: Set a tracking URI and an experiment
mlflow.set_tracking_uri("https://:5000")
mlflow.set_experiment("CrewAI")

然后是实际的 CrewAI 代码：

from crewai import Agent, Crew, Task

# Tasks and Agents definitions, see full example here
# https://mlflow.org.cn/docs/latest/genai/tracing/integrations/listing/crewai/#example-usage
...
crew = Crew(
        agents=[orchestrator, macro_data, researcher, results_lead],
        tasks=[plan_task, stats_task, research_task, synthesis_task],
        verbose=True,
    )
result = crew.kickoff()

Autolog（自动记录）功能可捕获每次运行的追踪，并将其聚合到易于导航的 DAG（有向无环图）中。但为了全面了解系统行为，我们建议追踪 MLflow autolog 不支持的自定义函数和工具。

为此，您可以使用自定义装饰器：

@mlflow.trace(span_type="TOOL", attributes={"key": "value"})
def func():
    ...

在我们的示例中：

@mlflow.trace(
    name="numeric_report_validation",
    attributes={"service": "heuristic numeric cross-check (report vs sources)"},
)
def validate_report_numbers_against_sources(
    final_report: str,
    *source_texts: str,
) -> str:
    # Function definition
    ...

@mlflow.trace() 装饰器允许你为任何函数创建 Span。这种简单的方法使我们能够捕获函数之间独特的关系，记录异常或自定义参数，并捕获上文提到的级联财务计算错误。

捕获相关指标

追踪告诉你发生了什么。指标则告诉你这是否可以接受。通过将针对 LLM 的 MLflow autolog()（例如 mlflow.openai.autolog()）与自定义属性结合使用，你可以将简单的追踪转变为可验证的审计追踪。一旦你建立了完整的追踪体系，下一步就是决定哪些数字才是真正关键的——因为在多智能体系统中，一个成功的响应可能会掩盖极其昂贵的执行成本。系统可能会在经历 15 次递归调用、对不稳定 API 进行 10 次重试以及智能体等待了整整两分钟（一个几乎不会有回应的请求）之后，才返回正确答案。

图 2. 在 MLflow UI 中探索指标。

我们可以将必要的指标归纳为三个关键支柱：

1. 编排与路由逻辑

每次编排器委派任务时，它都会做出一个决策。这个决策可能是正确的、冗余的，或者仅仅是缓慢的——如果不进行插桩，你将无法得知属于哪种情况。捕获路由行为有助于回答最关键的问题：监督者是否高效地将任务分配到了正确的地方？

以下是相关指标的一些示例：

• 委派成功率：监督者是否选择了正确的智能体或工具，还是被不准确的描述搞糊涂了？
• 委派延迟：监督者决定调用哪个智能体/工具所花费的时间。
• 冗余与循环检测：监督者是否选择了高效路径。

在 CrewAI 中，step_callback 是捕获路由行为的天然钩子——它在每个智能体步骤之后触发，为你提供智能体名称、输出和耗时。

import mlflow
import time

delegation_counts: dict[str, int] = {}
def track_routing(step_output) -> None:
      agent_name = step_output.agent
      delegation_counts[agent_name] = delegation_counts.get(agent_name, 0) + 1
      span = mlflow.get_current_active_span()

      if span:
          span.set_attributes({
              f"routing.{agent_name}.call_count": delegation_counts[agent_name],
              "routing.total_delegations": sum(delegation_counts.values()),
              # step_callback counts steps per agent, not supervisor delegations;
              # > 3 gives room for a normal reasoning + tool + retry cycle
              "routing.loop_detected": any(v > 3 for v in delegation_counts.values()),
          })
      # Log state handoff after each agent completes
      next_input = crew_state.get_state("current_input") or ""
      log_state_handoff(
          from_agent=agent_name,
          to_agent="next",  # CrewAI doesn't expose next agent here, use state
          output=str(step_output.result),
          next_input=next_input,
)

2. 状态一致性与内存污染

正如我们在级联错误示例中看到的那样，多智能体系统最大的风险不在于单个智能体的失效，而在于单个智能体输出的轻微错误被所有下游智能体奉为“事实真理（Ground Truth）”。跟踪状态一致性可以帮助你在这种偏差累积扩散之前检测到它。

以下指标有助于评估一致性和内存状态：

• Grounding 准确率（依据性准确率）：通过将智能体的输入与全局状态进行对比的评分，以了解该智能体的幻觉是否源自上一个智能体。
• 并发度：在同一执行窗口内尝试访问或更改同一元素的智能体数量。
• 上下文传递效率：智能体是在共享整个上下文，还是仅共享核心部分？

提高上下文传递效率的一个实用切入点是：计算一个智能体输出与下一个智能体接收内容之间的简单 Token 重合度评分。它不需要额外调用 LLM，并且可以在智能体传递过多或过少上下文时为你提供信号：

def handoff_efficiency_score(output_text: str, next_input_text: str) -> float:
    """Token overlap between agent output and the next agent's input.
    Score of 1.0 means the next agent received everything the previous one produced.
    Score close to 0.0 means most of the output was dropped before handoff.
    """
    output_tokens = set(output_text.lower().split())
    input_tokens = set(next_input_text.lower().split())
    if not output_tokens:
        return 0.0
    return len(output_tokens & input_tokens) / len(output_tokens)

@mlflow.trace(name="state_handoff", span_type="CHAIN")
def log_state_handoff(
    from_agent: str, to_agent: str, output: str, next_input: str
) -> None:
    span = mlflow.get_current_active_span()
    if span:
        span.set_attributes({
            "handoff.from_agent": from_agent,
            "handoff.to_agent": to_agent,
            "handoff.output_chars": len(output),
            "handoff.input_chars": len(next_input),
            "handoff.efficiency_score": handoff_efficiency_score(output, next_input),
        })

然后从每个任务回调（Task Callback）中调用它，在回调中你已经拥有了当前输出，并且可以从共享状态中读取上一个智能体的输出：

def _on_research_complete(output: object) -> None:
    text = crew_state.task_output_to_text(output)
    crew_state.set_state("research_brief", text)

# How much of the macro data specialist's output made it into the research brief?
stats = crew_state.get_state("macro_stats_snapshot") or ""
log_state_handoff(
    from_agent="macro_data_specialist",
    to_agent="research_analyst",
    output=stats,
    next_input=text,
)

3. 运维遥测与成本

在多智能体系统中，一个成功的响应可能会掩盖灾难性的运行故障。系统虽然给出了正确的答案，但却使用了多次递归调用和重试，并在等待循环中耗费了大量时间。

为了防止这些资源消耗和效率低下的情况，你必须捕获：

• Token 归因（Attribution）：每个节点、调用、路径或任务的成本；
• 单节点延迟和瓶颈检测：Span 持续时间与排队时间等指标可以检测瓶颈以及优化不佳的工具/数据库调用；
• 速率限制与吞吐量波动：API 调用次数、特定错误（例如 429）的频率以识别热点；
• 任务深度：用以识别智能体是否陷入了无限循环。

为了捕获自定义多智能体指标，你可以通过以下方式包装你的多智能体系统执行：

@mlflow.trace(name="Crew.kickoff", span_type="CHAIN")
def run_crew_with_metrics(crew: Crew) -> CrewOutput:
    """Run ``crew.kickoff()`` inside a traced span that collects summary metrics."""
    delegation_counts.clear()  # reset between runs to avoid false loop detection
    _crew_metrics.clear()
    t0 = time.perf_counter()
    result = crew.kickoff()
    duration_s = round(time.perf_counter() - t0, 3)

    span = mlflow.get_current_active_span()
    if span is not None:
        charter = crew_state.get_state("engagement_charter") or ""
        stats = crew_state.get_state("macro_stats_snapshot") or ""
        research = crew_state.get_state("research_brief") or ""
        synthesis = crew_state.get_state("synthesis_result") or ""

        span.set_attributes({
            "crew.total_duration_s": duration_s,
            "crew.task_count": len(crew.tasks),
            "crew.agent_count": len(crew.agents),
            "orchestration_lead.output_chars": len(charter),
            "macro_data_specialist.output_chars": len(stats),
            "research_analyst.output_chars": len(research),
            "synthesist.output_chars": len(synthesis),
            "validation.report_number_count": len(extract_numeric_tokens(synthesis)),
            "validation.source_number_count": len(extract_numeric_tokens(research)),
            **_crew_metrics,
        })
    return result

多服务可观测性

在某些情况下，多智能体系统可能会跨越通过 HTTP 请求连接的多个服务。理想情况下，我们希望对该系统拥有统一的追踪视图，而不是必须独立监控两端并在 Notebook 中将追踪拼接在一起。MLflow 支持 OTEL（OpenTelemetry），它能对拆分在多个服务中的应用程序进行插桩和监控。有关更多详细信息，请参阅此处。

虽然 MLflow 为大模型特有的 Span 提供了专门的视图，但多智能体系统并非孤立存在。它们依赖数据库、身份验证服务和第三方 API。通过利用 MLflow 的 OpenTelemetry (OTEL) 支持，你可以确保智能体追踪不会孤立在信息孤岛中：

• 统一上下文：将你的 MLflow 追踪导出到 Grafana、Datadog 或 Honeycomb 等企业级后端。这使得网站可靠性工程师（SRE）能够在同一个时间线上，将用户的系统前端点击与导致错误的特定智能体幻觉关联起来。
• 生产标准：现代基础设施团队已经普遍采用 OTEL。这意味着你的智能体追踪将与技术栈的其余部分显示在相同的仪表盘、告警和轮值指南中——无需独立的工具，也无需特殊处理。
• 跨服务关联：如果由于下游微服务超时导致智能体失败，OTEL 允许你在单个统一的追踪中，将智能体的重试逻辑与后端的 503 错误关联起来。

融会贯通

您可以在此代码仓库中找到使用 CrewAI 构建并由 MLflow 插桩的多智能体系统的端到端示例。

多智能体系统引入了在智能体之间开始交流之前不会浮现的故障模式——级联错误、内存污染，以及智能体在默默等待永远不会到来的响应时陷入停滞。要想超越“在我的电脑上运行正常”的局限，你的可观测性技术栈需要捕获全局全貌：不仅要了解每个智能体返回了什么，还要了解智能体之间是如何相互影响的，以及系统在哪些地方消耗了时间和资金。

MLflow 为您提供了实现这一目标的工具：针对基础需求的 autolog 自动记录，针对核心要害的自定义装饰器，以及在智能体仅为大型生产系统一部分时的 OTEL 集成支持。

看清正在发生的事情只是成功的一半。在本系列的第二部分中，我们将超越被动观察，探讨治理（Governance）与主动引导（Active Steering）——这些是你在问题触及生产环境之前，执行安全保障、管理成本并预防问题发生所需的机制。

接下来尝试什么？

• 在启用 autolog() 的情况下运行你现有的多智能体工作流，并寻找最慢或最昂贵的 Span。
• 在您的验证函数中添加 @mlflow.trace 装饰器，并开始度量上下文传递效率。
• 如果您的智能体跨越了多个服务，请配置 OTEL 导出，并首次在统一的视图中查看您的追踪。

如果您觉得 MLflow 有所帮助，请在 GitHub 上为我们点个 Star