用 AI Agent 把 Crash 分析从"人工排雷"变成"自动诊断"

预计阅读时间：13 分钟

一线开发者对这套流程再熟悉不过：Crash 看板飘红 → 打开堆栈 → 看一眼觉得像内存问题 → 找对应模块负责人 → 对方说"我看看"→ 半天后回复"是第三方 SDK 的锅"→ 再拉 SDK 方排查 → 一轮下来半天没了。如果 Crash 量级大、版本多，专家基本被钉死在排查上，新功能迭代自然减速。

这篇文章讲的是：把这套依赖人的排查流程，交给一个结构化的 AI Agent 来做——不是让大模型"猜原因"，而是用 Tool + Workflow 的框架，把原本散落在各处的原子能力串成自动诊断链路，当前已在 Crash 自动分析场景落地。

传统排查为什么慢

手动排查 Crash 有三个硬伤：

信息分散——堆栈在崩溃平台、设备信息在用户反馈、代码变更在 Git、历史修复在工单系统，排查者要在四五个系统间来回跳。
专家依赖——只有熟悉某模块的人才能快速判断"这行代码在什么条件下会空指针"，新人拿到堆栈往往只能逐行读。
重复劳动——同一类 Crash（比如某个 SDK 版本的已知问题）每次都要人重新走一遍判断流程。

本质上，排查是一个"信息检索 + 规则匹配 + 语义推理"的组合任务，每一步都有固定模式，但串起来需要人做调度。这正是 Agent 擅长的领域。

Agent 框架的核心设计：Tool 与 Workflow

框架把能力拆成两层：

Tool（原子能力）——每个 Tool 对应一个最小可复用的操作单元，比如"查堆栈详情"、"搜索 Git diff"、"检索历史相似 Crash"、"调用大模型做语义归纳"。Tool 只管输入输出，不关心场景。
Workflow（场景策略）——把多个 Tool 按业务逻辑编排成一条诊断链路。Workflow 决定"先查什么、再查什么、什么条件下跳过、什么条件下终止并输出结论"。

这种拆法的好处是：新增一个数据源只需加一个 Tool，不改动 Workflow；换一个分析场景只需写新 Workflow，Tool 全部复用。当前落地的 Crash 自动分析 Workflow 大致走这样的链路：

接收 Crash ID
  → Tool: 拉取堆栈与设备信息
  → Tool: 提取关键帧与异常类型
  → Tool: 搜索近 30 天相似 Crash
  → Tool: 查询相关代码最近 Git diff
  → Tool: 调用 LLM 归纳根因与修复建议
  → 输出结构化诊断报告

中间任何一步如果命中"已知问题"（相似 Crash 已有修复记录），Workflow 会短路跳过后续步骤，直接输出历史结论——这比让大模型从头推理高效得多。

实战：用 Python 搭一个最小 Crash 分析 Agent

下面是一个可运行的骨架实现，展示了 Tool 注册、Workflow 编排、短路逻辑的核心结构。你可以在此基础上接入真实的崩溃平台 API、Git 服务和大模型 SDK。

import json
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional

# ── Tool 定义 ──────────────────────────────────────────────

@dataclass
class Tool:
    name: str
    description: str
    fn: Callable

    def run(self, input_data: dict) -> dict:
        """执行原子能力，返回结构化结果"""
        print(f"  ▶ Tool [{self.name}] 执行中...")
        result = self.fn(input_data)
        print(f"  ✔ Tool [{self.name}] 完成")
        return result


# ── Workflow 定义 ──────────────────────────────────────────

@dataclass
class WorkflowStep:
    tool_name: str
    input_key: str          # 从上下文中取哪个 key 作为输入
    output_key: str         # 结果写入上下文的哪个 key
    short_circuit: Optional[Callable] = None  # 返回 True 则跳过后续步骤

@dataclass
class Workflow:
    name: str
    steps: List[WorkflowStep] = field(default_factory=list)

    def run(self, context: dict, tool_registry: Dict[str, Tool]) -> dict:
        print(f"\n🚀 Workflow [{self.name}] 启动，初始上下文: {list(context.keys())}")
        for step in self.steps:
            tool = tool_registry[step.tool_name]
            input_data = context.get(step.input_key, {})
            result = tool.run(input_data)
            context[step.output_key] = result

            # 短路判断：比如已知问题直接出结论
            if step.short_circuit and step.short_circuit(context):
                print(f"  ⚡ 短路触发于 step [{step.tool_name}]，跳过后续步骤")
                break

        print(f"🏁 Workflow [{self.name}] 完成\n")
        return context


# ── 注册具体 Tool ──────────────────────────────────────────

def fetch_crash_detail(input_data: dict) -> dict:
    """模拟：从崩溃平台拉堆栈与设备信息"""
    crash_id = input_data.get("crash_id", "unknown")
    # 实际实现：调用崩溃平台 SDK / HTTP API
    return {
        "crash_id": crash_id,
        "exception_type": "NullPointerException",
        "stack_key_frames": ["MainActivity.java:42", "UserManager.java:108"],
        "device_os": "Android 14",
        "app_version": "3.2.1",
    }

def search_similar_crashes(input_data: dict) -> dict:
    """模拟：检索历史相似 Crash"""
    exception = input_data.get("exception_type", "")
    frames = input_data.get("stack_key_frames", [])
    # 实际实现：向量检索 / ES 查询崩溃平台历史数据
    if "NullPointerException" in exception and "UserManager.java:108" in frames:
        return {
            "found_known_issue": True,
            "known_issue_id": "BUG-2047",
            "known_fix_summary": "UserManager.getUser() 在 userId 为空时未做防御",
            "known_fix_commit": "a1b2c3d",
        }
    return {"found_known_issue": False}

def query_git_diff(input_data: dict) -> dict:
    """模拟：查相关代码最近变更"""
    frames = input_data.get("stack_key_frames", [])
    # 实际实现：调用 GitLab/GitHub API 查 recent commits
    return {
        "relevant_commits": [
            {"sha": "e4f5g6h", "msg": "refactor: UserManager.getUser 增加缓存层", "date": "2024-11-08"},
        ],
        "files_changed": ["UserManager.java"],
    }

def llm_summarize(input_data: dict) -> dict:
    """模拟：调用大模型归纳根因与修复建议"""
    # 实际实现：调用 OpenAI / 内部模型 SDK，把堆栈 + git diff + 历史信息拼成 prompt
    if input_data.get("found_known_issue"):
        return {
            "root_cause": "已知问题 BUG-2047：UserManager.getUser() 缺少空值防御",
            "fix_suggestion": "补加 userId 非空判断，参考 commit a1b2c3d",
            "confidence": "high",
        }
    return {
        "root_cause": "疑似 UserManager 缓存层重构引入空指针路径",
        "fix_suggestion": "检查 UserManager.getUser 缓存失效分支，增加 null fallback",
        "confidence": "medium",
    }


# ── 组装 Tool 注册表 ──────────────────────────────────────

tool_registry = {
    "fetch_crash_detail": Tool(
        name="fetch_crash_detail",
        description="从崩溃平台拉取堆栈与设备信息",
        fn=fetch_crash_detail,
    ),
    "search_similar_crashes": Tool(
        name="search_similar_crashes",
        description="检索近 30 天相似 Crash 记录",
        fn=search_similar_crashes,
    ),
    "query_git_diff": Tool(
        name="query_git_diff",
        description="查询堆栈涉及文件的最近 Git 变更",
        fn=query_git_diff,
    ),
    "llm_summarize": Tool(
        name="llm_summarize",
        description="调用大模型归纳根因与修复建议",
        fn=llm_summarize,
    ),
}


# ── 定义 Crash 分析 Workflow ──────────────────────────────

def is_known_issue(context: dict) -> bool:
    """短路条件：已命中已知问题，无需继续查 git / 推理"""
    return context.get("similar_result", {}).get("found_known_issue", False)

crash_workflow = Workflow(
    name="crash_auto_analysis",
    steps=[
        WorkflowStep("fetch_crash_detail",  "init_input",    "crash_detail"),
        WorkflowStep("search_similar_crashes", "crash_detail", "similar_result",
                     short_circuit=is_known_issue),
        WorkflowStep("query_git_diff",       "crash_detail",  "git_result"),
        WorkflowStep("llm_summarize",        "crash_detail",  "llm_result"),
    ],
)


# ── 运行 ──────────────────────────────────────────────────

if __name__ == "__main__":
    # 场景 1：已知问题 —— Workflow 在 search_similar 后短路
    context_v1 = {"init_input": {"crash_id": "CRASH-9001"}}
    result_v1 = crash_workflow.run(context_v1, tool_registry)
    print("📋 诊断报告 v1:")
    print(json.dumps(result_v1.get("similar_result"), indent=2, ensure_ascii=False))

    # 场景 2：未知问题 —— 走完完整链路
    # 修改 fetch_crash_detail 返回不同堆栈即可模拟，此处省略

运行后你会看到：场景 1 在 search_similar_crashes 命中已知问题后触发短路，直接输出历史结论，不再调用 Git 和 LLM——这正是 Workflow 编排的价值，不是每条 Crash 都需要大模型从头推理。

改造提示：把四个模拟函数替换成真实 API 调用即可接入生产环境——崩溃平台 SDK、Git 服务 API、向量检索服务、大模型推理端点，每个 Tool 只需改 fn 实现，Workflow 编排零改动。

落地中的几个关键决策

Tool 的粒度怎么定？ 太粗（一个 Tool 干三件事）会丧失复用性，太细（每个 API 参数一个 Tool）会让 Workflow 编排爆炸。经验是：一个 Tool 对应"一个 API 调用 + 一层轻量后处理"，比如 search_similar_crashes 内部可能调了 ES 又做了去重，但对外只暴露"是否命中已知问题 + 命中了哪个"。

Workflow 的短路条件从哪来？ 短路不是硬编码的，而是从业务规则提炼。Crash 场景的核心规则是"已知问题直接复用结论"，其他场景（比如 ANR 分析）的短路条件可能是"主线程阻塞时长 < 阈值则标记为低优先级跳过深度分析"。每个 Workflow 自己定义短路逻辑，Tool 不需要知道。

大模型在链路中的位置？ 放在最后一步做归纳，而不是放在第一步做"万能诊断"。先让 Tool 把事实查清楚（堆栈、历史、代码变更），再让模型基于结构化事实做推理——这样输出的置信度更高，也更容易做结果校验。

上线前的检查清单

[ ] 每个 Tool 的输入输出是否已定义清晰的 schema（字段、类型、必填），避免 Workflow 间传数据时字段名对不上
[ ] 短路条件是否覆盖了"低价值 Crash 批量跳过"的场景，避免 Agent 把算力浪费在噪音上
[ ] LLM 调用是否做了超时与 fallback（模型不可用时降级为纯规则输出）
[ ] 诊断报告是否落库并和工单系统打通，形成闭环——Agent 输出结论，人确认后自动关单
[ ] 新增数据源时是否只需加 Tool、不改 Workflow——如果做不到，说明 Tool 抽象粒度有问题

把 Crash 排查交给 Agent 不是为了消灭专家，而是让专家从重复劳动中抽身，只处理 Agent 标记为"低置信度"或"新模式"的少数疑难 Case。当 Tool 和 Workflow 的基建到位后，ANR、卡顿、内存泄漏的分析场景可以按同样的框架快速铺开——这才是 Agent 重构稳定性治理的真正杠杆。