基于LangGraph的多Agent告警诊断系统设计与实践

1. 项目概述

在当今复杂的分布式系统环境中，服务告警诊断一直是个耗时且容易出错的过程。传统方式需要运维人员手动查看各种监控系统、日志平台，再结合经验进行问题定位。这不仅效率低下，而且在面对海量数据时容易遗漏关键线索。

最近半年，我一直在探索如何利用大模型和Agent技术来优化这个流程。经过多次迭代，最终基于LangGraph框架实现了一个可落地的多Agent告警诊断工作流。这个系统能够自动完成从告警触发到根因分析的全流程，将原本需要30分钟的诊断过程缩短到2分钟内完成。

2. 核心设计思路

2.1 为什么选择多Agent架构

单Agent方案在处理复杂诊断任务时存在明显局限：

• 单一Agent难以同时具备专业监控知识、日志分析能力和系统架构理解
• 长上下文会导致LLM性能下降和成本上升
• 缺乏明确的职责划分，难以进行针对性优化

我们的多Agent设计将诊断过程解耦为三个专业角色：

1. Planner：相当于"诊断指挥官"，负责制定检查计划
2. Worker：相当于"数据采集员"，负责执行具体的工具调用
3. Evaluator：相当于"分析专家"，负责综合研判给出结论

2.2 状态管理的艺术

在多Agent系统中，状态管理是关键难点。我们采用TypedDict定义了一个强类型的全局状态结构：

python复制class AgentState(TypedDict):
    # 基础标识
    session_id: str
    trace_id: str
    
    # 各Agent子状态
    planner_state: Dict[str, Any]
    worker_state: Dict[str, Any]
    evaluator_state: Dict[str, Any]
    
    # 工具相关
    tool_results: Dict[str, Any]
    tool_cache: Dict[str, Any]
    
    # 诊断结果
    root_cause: str
    recommended_actions: List[Dict[str, Any]]

这种设计带来了几个好处：

• 类型提示让开发更安全
• 状态结构一目了然
• 方便进行持久化和回放

3. 核心组件实现

3.1 工具层抽象

我们定义了标准的工具接口规范：

python复制class ToolSpec(TypedDict):
    name: str
    description: str
    input_schema: Dict[str, Any] 
    action_type: str  # READ/WRITE/EXECUTE

以获取指标的工具为例：

python复制class GetMetricsTool:
    name = "get_service_metrics"
    
    def run(self, service_name: str, window: int = 5) -> Dict[str, Any]:
        # 实际会调用Prometheus API
        return {
            "latency_p99": 1200,
            "error_rate": 0.03
        }

工具层的标准化使得Agent可以无需修改代码就能接入新的数据源。

3.2 Planner实现

Planner的核心职责是根据告警类型生成检查计划：

python复制def planner_node(state: AgentState) -> Dict[str, Any]:
    # 根据告警类型定制不同plan
    if "high latency" in state["goal"]:
        plan = {
            "steps": [
                {"action": "fetch_metrics", "tool": "get_service_metrics"},
                {"action": "fetch_logs", "tool": "get_service_logs"}
            ]
        }
    # 更新状态
    return {
        "planner_state": {"plan": plan},
        "next_node": "worker"
    }

在实际项目中，这里的逻辑可以替换为LLM调用，让模型根据自然语言描述生成检查计划。

3.3 Worker实现

Worker负责执行Planner生成的计划，并加入了简单的缓存机制：

python复制def worker_node(state: AgentState) -> Dict[str, Any]:
    plan = state["planner_state"]["plan"]
    results = {}
    
    for step in plan["steps"]:
        # 检查缓存
        cache_key = f"{step['tool']}:{json.dumps(step['args'])}"
        if cache_key in state["tool_cache"]:
            results[step['tool']] = state["tool_cache"][cache_key]
            continue
            
        # 执行工具调用
        tool = get_tool(step["tool"])
        result = tool.run(**step["args"])
        
        # 更新缓存
        state["tool_cache"][cache_key] = result
        results[step['tool']] = result
    
    return {
        "tool_results": results,
        "next_node": "evaluator"
    }

3.4 Evaluator实现

Evaluator综合各类数据进行分析：

python复制def evaluator_node(state: AgentState) -> Dict[str, Any]:
    metrics = state["tool_results"]["get_service_metrics"]
    logs = state["tool_results"]["get_service_logs"]
    
    if metrics["error_rate"] > 0.02:
        return {
            "root_cause": "高错误率",
            "recommended_actions": [{
                "action": "扩容",
                "priority": 1
            }],
            "next_node": "__end__"
        }

在实际项目中，这里的分析逻辑可以接入大模型进行更复杂的推理。

4. 工作流编排

使用LangGraph将各个节点连接成完整工作流：

python复制from langgraph.graph import StateGraph

builder = StateGraph(AgentState)

# 添加节点
builder.add_node("planner", planner_node)
builder.add_node("worker", worker_node) 
builder.add_node("evaluator", evaluator_node)

# 设置边关系
builder.set_entry_point("planner")
builder.add_edge("planner", "worker")
builder.add_edge("worker", "evaluator")

# 编译
workflow = builder.compile()

执行工作流：

python复制initial_state = {
    "goal": "诊断order-service高延迟问题",
    "constraints": ["只读操作"]
}

final_state = workflow.invoke(initial_state)
print(final_state["root_cause"])

5. 生产环境考量

5.1 性能优化

我们在生产环境中加入了以下优化措施：

• 工具调用并行化
• 智能缓存策略
• 超时和重试机制

优化前后的性能对比：

5.2 监控指标

完善的监控是生产可用的关键。我们跟踪的核心指标包括：

• 各节点耗时
• 工具调用成功率
• 诊断准确率
• Token使用量

示例监控面板：

code复制[Planner] 平均延迟: 320ms (±45ms)
[Worker] 平均工具调用耗时: 1.2s (±0.3s)
[Evaluator] 诊断准确率: 87%

6. 扩展与定制

这套框架可以灵活扩展：

6.1 支持新的告警类型

只需在Planner中添加对应的检查逻辑：

python复制if "数据库连接失败" in alert_message:
    plan = {
        "steps": [
            {"tool": "check_db_connection"},
            {"tool": "verify_db_load"} 
        ]
    }

6.2 自定义分析逻辑

Evaluator可以接入不同的分析模型：

python复制def evaluator_node(state):
    if use_llm:
        return llm_analyze(state)
    else:
        return rule_based_analyze(state)

7. 踩坑经验

在实际落地过程中，我们总结了以下经验教训：

1. 工具超时处理：初期没有设置工具调用超时，导致整个工作流卡死。现在所有工具都强制设置timeout参数。

2. 状态版本控制：当状态结构变更时，旧的回放数据会解析失败。现在我们加入了version字段，并保持向后兼容。

3. LLM稳定性：直接让LLM生成plan有时会输出不合法的工具调用。现在我们会在Planner中加入校验逻辑，并准备fallback方案。

4. 成本控制：初期没有监控token使用量，导致意外的高成本。现在我们会实时计算并限制单次诊断的max_tokens。

8. 典型问题排查

在实际运维中，我们遇到了以下典型问题及解决方案：

9. 项目演进方向

当前系统已经稳定运行3个月，平均每天处理1200+次告警诊断。接下来的优化方向包括：

1. 智能降级机制：在系统高负载时自动降级检查深度
2. 主动预防：在问题发生前进行预测性检查
3. 知识沉淀：将诊断经验反哺到知识库中

这套多Agent诊断框架不仅适用于运维场景，经过适当改造后也可以应用于：

• 客户支持问题排查
• 财务异常检测
• 安全事件分析

在实际部署中，最关键的是要建立完善的测试体系。我们构建了一个包含200+种模拟场景的测试套件，每次代码变更都会运行完整的回归测试，确保核心诊断逻辑的稳定性。