构建用于 API 故障诊断的有界 LangGraph Agent

决策：自动诊断，审慎修复

API 故障 Agent 可以通过收集日志、健康检查和近期部署元数据来节省时间。但它不应在没有提示的情况下重启服务、回滚版本或更改配置。更安全的设计会将只读诊断与产生副作用的修复操作分开，限制 Agent 的工作范围，并向操作人员明确呈现失败或缺失的证据。

这是一个实用设计示例，并非经过测试的集成方案或可直接用于生产环境的服务。以下工具实现仅供说明：请替换为你自己的、经过身份验证且权限受限的可观测性客户端。

前置条件与工作流

你需要 Python、已在应用中配置的 LangGraph 兼容聊天模型，以及现有诊断系统对应的客户端函数。本示例假设这些客户端函数会返回数据或抛出异常，不依赖特定的模型提供商、端点或工具集成。

一次实用的故障处理运行应遵循以下受限循环：

1. 在受影响的服务和时间范围内，收集一小组预先定义的信号。
2. 将每项结果标记为成功、失败或不可用。
3. 让模型总结证据，并指出仍然未知的事项。
4. 在达到限定的图步骤数后停止；执行任何会改变状态的操作前，先请求人工审核。

状态字段的意义在于保障运维清晰度：日志查询失败时，不能让它看起来像是一次成功执行、但没有匹配事件的日志搜索。

定义带有明确故障信号的诊断工具

让工具契约保持精简。要求提供服务名称和有界时间范围；如果固定查询足以满足需求，就不要赋予模型任意查询权限。返回结构化结果，让后续节点和故障报告能够区分没有证据与工具故障。

from typing import Literal, TypedDict
from langchain.tools import tool

class Signal(TypedDict):
    status: Literal["ok", "error", "unavailable"]
    source: str
    data: str
    error: str | None


def query_logs(service: str, minutes: int) -> Signal:
    if not 1 <= minutes <= 60:
        return {
            "status": "error", "source": "logs", "data": "",
            "error": "minutes must be between 1 and 60",
        }
    try:
        # Replace with a read-only, scoped observability client call.
        rows = logs_client.search(service=service, minutes=minutes)
        return {
            "status": "ok", "source": "logs",
            "data": str(rows), "error": None,
        }
    except TimeoutError as exc:
        return {
            "status": "unavailable", "source": "logs", "data": "",
            "error": f"timeout: {exc}",
        }

@tool
def inspect_api(service: str, minutes: int = 15) -> dict:
    """Collect scoped, read-only API incident signals."""
    return query_logs(service, minutes)

为健康检查或部署元数据添加独立工具，并使用相同的结果结构。不要捕获所有异常后都返回空字符串：应保留足够的错误上下文，以便诊断集成故障，同时避免在日志和模型上下文中暴露密钥或敏感负载。

限制执行范围并明确标示未完成的运行

对于使用 LangGraph 的 create_agent 构建的 Agent，可在调用时设置图递归上限。该上限限制图步骤数；它不能保证得到有用的答案，也不能代替工具级超时。应在外围应用中处理达到上限的情况，并清楚地返回“未完成”状态，而不是将这次运行视为已完成的诊断。

from langchain.agents import create_agent

agent = create_agent(
    model=your_configured_chat_model,
    tools=[inspect_api],
    system_prompt=(
        "Diagnose API incidents using available evidence. "
        "Report each source as ok, error, or unavailable. "
        "Do not claim a cause without supporting evidence. "
        "Do not perform remediation."
    ),
)

try:
    result = agent.invoke(
        {"messages": [{"role": "user", "content":
            "Investigate elevated errors for payments-api over the last 15 minutes."}]},
        config={"recursion_limit": 12},
    )
except Exception as exc:
    # Map known graph-limit and provider/tool failures to your app's
    # incident-run status; do not expose credentials or raw secrets.
    result = {"status": "incomplete", "reason": type(exc).__name__}

这里的上限值仅为示例，并非适用于所有场景的建议。请根据允许调用的工具数量和模型轮次来选择，并在工具层和应用层添加超时与请求大小限制。记录运行 ID、所选服务和时间范围、尝试过的工具、工具状态及终止原因。借助这些信号，操作人员可以区分“未发现错误”和“日志后端超时”。

在修复前加入人工审核

如果之后添加了会改变生产环境状态的工具，应使用 LangChain 的 human-in-the-loop 中间件和检查点器为这些工具设置审批关卡。文档所述的中间件可以暂停符合条件的工具调用，以便进行审批、编辑、拒绝或直接响应。状态会被保存，因此可以在作出决定后恢复执行；生产环境中应使用持久化检查点器，而不是示例中用于原型开发的内存选项。请参阅 LangChain 人机协同文档。

一项实用策略是：只读诊断无需中断，而重启、回滚或配置更改工具必须经过审批。拒绝提议的操作时，应选择拒绝；文档提醒，“respond”用于以工具身份采取行动，而不是拒绝会产生副作用的调用。

请在自己的工具实现中验证经过编辑的参数：文档所述的中间件不会根据显示的 schema 验证编辑内容。

需要考虑的故障情况

• 结果为空： 应报告“未返回匹配事件”，而不是“服务运行正常”。
• 工具超时或授权失败： 保留失败来源和错误类别；不要无限静默重试。
• 重复调用工具： 限制图执行范围并约束工具输入。如果添加重试，应设置明确且次数较少的重试策略，并公开尝试次数。
• 中断后无法恢复： 确认已配置检查点器，并且恢复时使用相同的线程标识符。源文档要求人工审核时使用检查点。
• 误导性的最终摘要： 要求报告中的结论引用已收集的信号，并列出不可用的数据源和未解决的问题。

故障运行检查清单

部署前，请确认诊断工具均为只读且范围受限；时间范围和查询量均有限制；工具错误能够被观察到；图执行设置了上限；每项会产生副作用的操作都会暂停，等待人工决策。测试超时、结果为空、操作被拒绝以及中断后恢复等路径。优秀的 Agent 不只是给出一个看似合理的根因，还会展示实际获取了哪些证据，以及诊断到哪里为止。

来源

• LangChain：Human-in-the-loop 中间件 — 中断决策、条件式工具审核、检查点以及恢复暂停的执行。