用 LangGraph 实现多 Agent 协作调试复杂 Bug

在微服务架构下，支付流水丢失等复杂 Bug 往往横跨网关、业务层和数据库，日志分散且格式不一。单靠人工检索极易遗漏关联上下文，且假设验证效率低下。LangGraph 将 Agent 协作建模为状态图，使各职责单元通过共享状态迭代收敛，适合构建可控的调试流水线。这种结构不仅解决了传统链式调用中上下文污染的问题，还通过显式的边定义，让调试过程中的每一步都具备可追溯性。

一、 架构选型：为何用状态图替代链式调用

传统 ReAct 循环在复杂 Bug 排查中易陷入重复推演，导致 Token 浪费且难以回溯。LangGraph 显式定义节点（Agent）与边（转移条件），状态在节点间单向流动并持续累积。这种结构支持并行分支与条件回退，例如当代码检索失败时，可以精准回退到日志分析节点获取更细粒度的时间戳，而不是盲目重启整个流程。

# 定义基础调试状态
from typing import TypedDict, List, Optional

class DebugState(TypedDict):
    bug_desc: str
    logs: List[str]
    code_snippets: List[str]
    hypotheses: List[str]
    verified_hypotheses: List[str]
    current_focus: str
    iteration: int
    # 新增字段以支持更好的状态管理
    search_keywords: List[str]
    error_context: str

使用 LangGraph 实战：用状态图编排多智能体协作 的核心思路，将调试过程拆解为可追溯的节点。状态对象不仅包含日志、代码片段与假设列表，还引入了搜索关键词以指导后续检索。每次迭代后由路由函数决定下一步走向，确保状态机始终处于可控路径中。

二、 日志分析 Agent 设计与实现

日志 Agent 的职责是从原始日志流中提取关键时间戳与错误码。它不直接下结论，而是输出结构化摘要供后续节点消费。该节点采用滑动窗口策略，仅保留最近 20 行日志，避免单次输入超出 LLM 上下文限制。同时，它会将提取到的业务 ID 写入状态，作为代码检索的锚点。

from langchain_core.messages import SystemMessage, HumanMessage
from langchain_openai import ChatOpenAI
import json

llm = ChatOpenAI(model="gpt-4", temperature=0)

def log_analysis_node(state: DebugState) -> dict:
    """分析日志流，提取异常片段与时间关联"""
    raw_logs = state.get("logs", [])
    # 截断至最近 20 行，防止 token 溢出
    context = "\n".join(raw_logs[-20:])
    
    prompt = f"""你是日志分析专家。
    输入日志：
    {context}
    
    要求：
    1. 识别 ERROR/WARN 级别条目
    2. 提取关键业务 ID 与时间戳
    3. 输出 JSON 格式摘要，包含 anomaly_timestamps 和 related_ids
    4. 生成 2-3 个用于代码检索的关键词
    """
    
    response = llm.invoke([
        SystemMessage(content="严格输出 JSON，无其他文字"),
        HumanMessage(content=prompt)
    ])
    
    try:
        summary = json.loads(response.content)
        # 提取检索关键词，默认为空
        keywords = summary.get("search_keywords", [])
    except json.JSONDecodeError:
        summary = {"anomaly_timestamps": [], "related_ids": []}
        keywords = []
        
    return {
        "current_focus": "log_analysis_complete",
        "logs": [f"[SUMMARY] {json.dumps(summary, ensure_ascii=False)}"] + raw_logs[:5],
        "search_keywords": keywords,
        "error_context": context
    }

三、 代码检索节点与上下文管理

代码 Agent 接收日志摘要中的业务 ID 及搜索关键词，在本地代码库中检索相关文件。此处引入向量检索能力，将报错堆栈中的类名与函数名作为查询键。检索结果需附带文件路径与行号，便于后续验证。若代码库规模超过十万行，纯关键词匹配效率低。

3.1 检索策略选择

可参考 用 LangGraph 编排 RAG 动态检索工作流 中的动态路由思路，根据报错类型选择检索策略：堆栈追踪用精确匹配，业务逻辑异常用语义检索。以下代码演示了基于关键词的简单检索逻辑，实际生产中应替换为向量数据库查询。

import subprocess
import json

def code_retrieval_node(state: DebugState) -> dict:
    """基于日志摘要检索代码片段"""
    # 优先使用上一节点生成的搜索关键词
    keywords = state.get("search_keywords", [])
    related_ids = []
    
    # 从上一步的日志摘要中提取 ID 作为补充查询
    for log_line in state.get("logs", []):
        if "[SUMMARY]" in log_line:
            try:
                summary_part = log_line.split("[SUMMARY] ")[1]
                summary = json.loads(summary_part)
                related_ids = summary.get("related_ids", [])
            except:
                pass
    
    search_terms = keywords[:2] + related_ids[:1]  # 混合检索策略
    
    if not search_terms:
        return {"code_snippets": [], "current_focus": "no_search_terms_found"}
    
    snippets = []
    for term in search_terms[:3]:  # 限制检索数量
        # 实际生产中应使用向量数据库，此处用 grep 演示
        try:
            result = subprocess.run(
                ["grep", "-rn", term, "./src/"],
                capture_output=True, text=True, timeout=5
            )
            if result.returncode == 0:
                # 仅保留前5行匹配结果，避免上下文爆炸
                snippets.extend(result.stdout.split("\n")[:5])
        except Exception as e:
            pass
            
    return {
        "code_snippets": snippets,
        "current_focus": "code_retrieval_complete",
        "search_keywords": search_terms # 持久化本次使用的关键词
    }

四、 假设验证节点与反馈闭环

验证 Agent 的核心任务是判断当前假设是否成立。它综合日志摘要与代码片段，生成可执行的验证步骤。若验证失败，状态图需回退至检索节点并调整检索词，而非直接终止。节点函数必须保持幂等性，所有更新通过返回值传递，避免直接修改传入的状态对象。

def hypothesis_verification_node(state: DebugState) -> dict:
    """验证假设并更新状态"""
    snippets = state.get("code_snippets", [])
    log_summary = "\n".join(state.get("logs", [])[:2])
    
    prompt = f"""当前代码片段：
    {snippets[:5]}
    
    日志摘要：
    {log_summary}
    
    请基于上述信息生成一个最可能的故障假设及验证方案：
    1. 假设描述
    2. 验证命令或测试用例
    3. 预期结果
    输出 JSON: {{hypothesis, verify_cmd, expected}}
    """
    
    response = llm.invoke([
        SystemMessage(content="输出有效 JSON，不要包含其他文本"),
        HumanMessage(content=prompt)
    ])
    
    try:
        data = json.loads(response.content)
        hypothesis = data.get("hypothesis", "unknown")
    except:
        hypothesis = "解析失败，需人工介入"
    
    # 简单模拟验证逻辑，实际应调用测试框架或执行 verify_cmd
    # 这里假设如果 snippets 中包含 Exception 或 Error 且假设与上下文相关，则视为通过
    is_valid = "Exception" in str(snippets) and len(snippets) > 0
    verified = []
    if is_valid:
        verified.append(hypothesis)
        
    return {
        "hypotheses": [hypothesis],
        "verified_hypotheses": verified,
        "current_focus": "verification_complete",
        "iteration": state.get("iteration", 0) + 1
    }

五、 状态路由与终止条件

路由函数决定流转方向。当验证通过或迭代次数超过阈值时，终止流程。共享记忆库在此处体现为状态对象的持续更新，各节点仅读全局状态并返回增量更新。这种设计避免了全量覆盖导致的信息丢失，同时通过 `iteration` 计数器防止无限循环。

from langgraph.graph import StateGraph, END, START
from typing import Literal

def router_after_verification(state: DebugState) -> Literal["retrieval", "end", "analysis"]:
    """根据验证结果决定下一步"""
    iter_count = state.get("iteration", 0)
    
    # 迭代超过 5 次或假设已验证，则结束
    if iter_count >= 5 or state.get("verified_hypotheses"):
        return "end"
    
    # 若代码检索无结果，回退至日志分析以获取更细粒度信息
    if not state.get("code_snippets"):
        return "analysis"
    
    # 否则继续检索新的代码区域或重新验证
    return "retrieval"

def build_debug_graph():
    graph = StateGraph(DebugState)
    
    graph.add_node("log_analysis", log_analysis_node)
    graph.add_node("code_retrieval", code_retrieval_node)
    graph.add_node("verification", hypothesis_verification_node)
    
    graph.add_edge(START, "log_analysis")
    graph.add_edge("log_analysis", "code_retrieval")
    graph.add_edge("code_retrieval", "verification")
    graph.add_conditional_edges(
        "verification",
        router_after_verification,
        {
            "analysis": "log_analysis",
            "retrieval": "code_retrieval",
            "end": END
        }
    )
    
    return graph.compile()

六、 性能权衡与长期记忆

多 Agent 迭代导致 LLM 调用次数线性增长。为控制延迟，建议将高频查询结果缓存。若涉及跨会话的知识积累，可引入向量数据库存储历史 Bug 模式。关于 Pinecone 向量库在 Agent 长期记忆中的延迟优化 的实践，可降低重复检索的成本。此外，若需结合命令行 AI 进行辅助调试，Aider 实战：用命令行 AI 结对编程 提供的代码编辑接口可嵌入到验证环节中，自动执行验证命令。

七、 本地调试环境搭建

调试流水线本身需要可视化工具监控执行轨迹。若团队使用本地模型，可结合 Open WebUI 实战：自建本地大模型可视化平台 中的界面方案，将 LangGraph 的 state 快照推送至前端展示。对于资源受限环境，参考 本地大模型量化部署：GGUF 与 llama.cpp 调优实战 降低推理成本，确保调试节点在本地也能流畅运行。

八、 落地实践与注意事项

节点函数必须保持幂等性，避免副作用污染共享状态。所有 LLM 调用需设置超时与重试机制。路由逻辑应避免依赖 LLM 判断，改用确定性规则减少不确定性。若需快速验证界面原型，可用 Gradio 实战：用 Python 快速搭建 AI 演示界面 封装调试流程，实时展示状态变更。特别要注意，`hypothesis_verification_node` 中不得直接修改传入的 `state` 对象，所有新增的假设必须通过 `return` 字典返回，由 LangGraph 框架负责合并状态。

小结

LangGraph 将调试过程从线性脚本升级为有向状态图，通过职责分离的 Agent 与共享记忆实现迭代收敛。核心收益在于：上下文可控、回溯路径清晰、验证闭环自动。工程落地的关键在于节点幂等性与路由确定性，避免 LLM 幻觉导致的无限循环。该模式可复用于日志分析、性能瓶颈定位等需多源信息融合的运维场景，显著降低人工排查复杂系统 Bug 的时间成本。