用 Amazon Bedrock AgentCore Observability 排查生产 Agent 故障

2026-06-30 25 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间:12 分钟

Agent 进入生产环境后,最难处理的不是“模型答错了”这种显性问题,而是它为什么反复调用同一个工具、为什么一次工具调用没有返回、为什么用户看到超时但后端日志看起来正常。Amazon Bedrock AgentCore Observability 的价值就在这里:把 Agent 的执行轨迹、工具调用和运行指标放到同一个可分析的上下文里,让排障从猜测变成复盘。

这篇文章聚焦生产故障调试。性能优化和记忆管理不展开,因为它们属于另一个阶段的问题:系统先要能解释失败,才谈得上稳定提速。

生产 Agent 的故障通常不是单点崩溃

传统 API 服务出错时,常见路径是请求进入服务、访问数据库、返回响应。Agent 的路径更像一条动态生成的执行链:模型判断意图,选择工具,解析工具返回,再决定下一步。如果其中任何一步的输入输出不符合预期,问题可能表现为完全不同的症状。

常见模式包括:

  • 无限循环:Agent 多次选择同一个工具,或者不断改写同一个问题却没有收敛。
  • 工具调用失败:参数缺失、schema 不匹配、权限不足、下游超时。
  • 错误恢复失败:工具返回错误后,Agent 没有切换策略,而是继续走同一条失败路径。
  • 用户侧超时:单个步骤未必失败,但多轮推理和工具调用累计超过业务超时阈值。

Observability 要解决的不是“有没有日志”,而是能不能回答三个问题:Agent 当时看到了什么、它为什么选择这个动作、这个动作产生了什么结果。

用 trace 看 Agent 是怎么走到失败的

排查生产 Agent 时,trace 比单行日志更有用。日志告诉你某一步发生了什么,trace 告诉你这些步骤之间的因果关系。

一次有效的 Agent trace 至少应该能串起这些事件:

  • 用户输入或上游任务输入。
  • Agent 的中间推理步骤或决策节点。
  • 每次工具调用的名称、参数、耗时和结果状态。
  • 模型输出、工具返回和最终响应之间的关系。
  • 错误、重试、超时和停止原因。

当你看到一个无限循环,trace 通常会呈现出很清楚的形状:同一个 tool name 在短时间内重复出现,参数变化很小,返回结果也没有提供新信息。这个时候不要只改 prompt。更直接的修复可能是增加最大步数、为工具结果增加明确状态、或者让 Agent 在连续相同失败后进入降级路径。

用 metrics 判断故障范围,而不是只盯一个请求

trace 适合解释单次失败,metrics 适合判断影响面。生产排障时,这两类信息要配合使用。

可以重点看这些指标维度:

  • Agent 请求成功率和失败率。
  • 工具调用错误率,按工具名称拆分。
  • 每次会话的平均工具调用次数。
  • Agent 执行时长分位数,例如 p50、p95、p99。
  • 超时数量和重试数量。

如果某个工具错误率突然升高,问题可能在下游服务、权限或参数结构。如果整体 p95 拉长但错误率没有明显变化,可能是 Agent 走了更多步骤,或者工具响应变慢。如果平均工具调用次数异常升高,就要优先检查循环和决策条件。

可以这样实践:用 CloudWatch Logs Insights 找循环和工具失败

下面的命令和查询是一个可改造的排障骨架。具体日志组名称、字段名和区域要按你的 AgentCore Observability 配置调整。这里假设 trace 或事件日志里包含 sessionIdtoolNameeventTypestatuslatencyMs 这类字段。

先设置环境变量:

export AWS_REGION="us-east-1"
export LOG_GROUP="/aws/bedrock-agentcore/your-agent-runtime"

查找同一会话里工具调用次数异常高的请求,可以这样跑:

aws logs start-query \
  --region "$AWS_REGION" \
  --log-group-name "$LOG_GROUP" \
  --start-time $(date -u -d '1 hour ago' +%s) \
  --end-time $(date -u +%s) \
  --query-string '
fields @timestamp, sessionId, toolName, eventType, status, latencyMs
| filter eventType = "tool_invocation"
| stats count(*) as tool_calls,
        count_distinct(toolName) as distinct_tools,
        max(latencyMs) as max_latency_ms
  by sessionId
| filter tool_calls > 10
| sort tool_calls desc
| limit 20
'

查找工具失败集中在哪些工具上:

aws logs start-query \
  --region "$AWS_REGION" \
  --log-group-name "$LOG_GROUP" \
  --start-time $(date -u -d '1 hour ago' +%s) \
  --end-time $(date -u +%s) \
  --query-string '
fields @timestamp, toolName, status, errorType, latencyMs
| filter eventType = "tool_invocation"
| filter status in ["ERROR", "TIMEOUT", "FAILED"]
| stats count(*) as failures,
        pct(latencyMs, 95) as p95_latency_ms
  by toolName, errorType
| sort failures desc
| limit 20
'

如果你的日志字段不是 JSON 顶层字段,可以先在 Logs Insights 里用 fields @message 看原始结构,再调整查询。关键不是照抄字段名,而是形成固定动作:先找异常会话,再按工具聚合,再回到单条 trace 复盘。

可以这样实践:给工具调用加上可观测的边界

AgentCore Observability 能展示 Agent 行为,但工具本身也要返回清晰、稳定、可诊断的结果。下面是一个最小 Python 示例,演示如何包装工具调用:记录耗时、状态和错误类型,并返回结构化结果。你可以把同样的模式改造成 Lambda、容器服务或内部 API 的工具入口。

运行前只需要本地 Python 3:

import json
import time
from typing import Any, Callable, Dict


def observed_tool(tool_name: str, fn: Callable[[Dict[str, Any]], Dict[str, Any]]):
    def wrapper(args: Dict[str, Any]) -> Dict[str, Any]:
        started = time.time()
        event = {
            "eventType": "tool_invocation",
            "toolName": tool_name,
            "status": "OK",
            "latencyMs": None,
            "errorType": None,
        }

        try:
            result = fn(args)
            return {
                "ok": True,
                "tool": tool_name,
                "result": result,
            }
        except TimeoutError as exc:
            event["status"] = "TIMEOUT"
            event["errorType"] = type(exc).__name__
            return {
                "ok": False,
                "tool": tool_name,
                "error": "downstream_timeout",
                "retryable": True,
            }
        except ValueError as exc:
            event["status"] = "FAILED"
            event["errorType"] = type(exc).__name__
            return {
                "ok": False,
                "tool": tool_name,
                "error": "invalid_arguments",
                "retryable": False,
            }
        finally:
            event["latencyMs"] = int((time.time() - started) * 1000)
            print(json.dumps(event, ensure_ascii=False))

    return wrapper


def lookup_order(args: Dict[str, Any]) -> Dict[str, Any]:
    order_id = args.get("order_id")
    if not order_id:
        raise ValueError("order_id is required")
    return {"order_id": order_id, "status": "SHIPPED"}


if __name__ == "__main__":
    tool = observed_tool("lookup_order", lookup_order)
    print(json.dumps(tool({"order_id": "A-1001"}), ensure_ascii=False))
    print(json.dumps(tool({}), ensure_ascii=False))

这个例子没有假设某个具体 AgentCore SDK API,它表达的是生产工具应该遵守的接口习惯:错误要分类,是否可重试要明确,返回体要让 Agent 有机会换一条路径,而不是只能继续猜。

排障工作流:从告警到修复

一套实用的生产流程可以压缩成五步:

  1. 从告警或用户报障拿到时间窗口、会话 ID、Agent 版本和请求类型。
  2. 用 metrics 判断是单点问题还是整体退化,例如某个工具失败率升高,还是所有请求都变慢。
  3. 打开对应 trace,检查工具调用顺序、重复次数、错误恢复路径和最终停止原因。
  4. 针对根因修复:schema 校验、工具超时、权限、prompt 约束、最大步数、降级回复或重试策略。
  5. 回放相似请求,确认 trace 形状已经改变,而不是只让当前样例通过。

对无限循环,优先加硬边界:最大工具调用次数、最大执行时长、连续相同工具调用上限。对工具失败,优先让错误变得可解释:参数错误不要重试,下游超时可以有限重试,权限错误应该直接暴露给运维或配置流程。

采用建议:先让失败可见,再让系统聪明

Agent 系统很容易把复杂性藏在“智能”这个词后面。生产环境里,智能不等于不可解释。AgentCore Observability 提供的是调试入口,但团队还需要配套工程纪律:稳定的工具 schema、清晰的错误分类、按版本记录 trace、对循环和超时设置硬限制。

上线前可以检查这几项:

  • 每个工具都有名称、参数 schema、超时和错误分类。
  • trace 能关联到用户请求、会话、Agent 版本和工具调用。
  • metrics 能按工具名称和错误类型聚合。
  • 无限循环有最大步数或最大耗时保护。
  • 修复后用 trace 验证行为路径,而不只看最终答案。

当这些基础设施到位,生产 Agent 的故障就不再是一团雾。你可以看到它每一步做了什么,也能更快判断该改 prompt、改工具、改权限,还是改运行边界。


相关推荐