1. 故障复盘为什么总是拖到三小时起步线上出故障最折磨人的往往不是修的那一刻而是事后复盘。我待过的几个团队几乎都经历过这样的场景凌晨两点被告警叫醒一顿操作把服务拉回来第二天下午开复盘会结果一屋子人对着几个日志文件翻来翻去谁也说不太清楚“第一个异常到底出现在哪台机器、哪个时间点”。三个小时过去会议纪要里写的还是“疑似某服务超时导致雪崩”下次同类问题照样复现。问题出在哪不是大家不认真而是日志调查这件事本身太依赖人肉检索。一个中等规模的微服务系统一次故障涉及的日志可能横跨十几个服务、几十个 Pod、上百万行文本。人眼去 grep效率低不说还特别容易漏掉关键线索——因为故障的因果链往往藏在时间戳的缝隙里而不是某一条显眼的 ERROR 里。我后来琢磨出一个思路把“日志调查”这件事本身做成一个 Agent让它像一个有经验的 SRE 一样先看告警、再定位时间窗口、然后逐个服务排查、最后给出带证据链的结论。这样复盘会就不用再翻日志了直接看 Agent 输出的调查报告就行。这篇文章就把我基于蓝耘元生代平台、用LangGraph编排、结合OpenTelemetry链路数据做出来的这个日志调查 Agent从头到尾拆一遍。整套东西跑下来原本三小时的复盘现在基本一杯咖啡没喝完报告就出来了。适合谁来参考如果你正在做可观测性建设、或者被故障复盘折磨过、又或者想找一个 LangGraph 的真实落地场景而不是只会写“hello world”的 demo那这篇应该对你有用。下面我会把设计思路、核心节点、实操步骤、踩过的坑全部摊开讲。2. 整体设计为什么是 Agent 而不是一条脚本2.1 从“脚本思维”到“Agent 思维”的转变一开始我也想过写个 Python 脚本不就行了输入时间范围grep 出所有 ERROR按服务分组输出个报告。但真做起来发现根本不够用。原因很简单日志调查是一个需要多轮决策的过程而不是一条直线。举个实际例子。告警说“订单服务 P99 延迟飙升”。脚本能做的只是把订单服务的 ERROR 日志捞出来。但真正的原因可能是订单服务本身没问题是它依赖的库存服务变慢了而库存服务变慢又是因为数据库连接池被一个慢查询占满。这条因果链脚本是发现不了的因为它需要“先看订单、发现调用下游超时、再去查下游、再往下查”这种动态的、根据中间结果决定下一步查哪里的行为。这正是 Agent 的用武之地。Agent 的核心能力是根据当前掌握的信息决定下一步该做什么。放到日志调查里就是先看告警定位入口再根据入口服务的依赖关系决定排查顺序每查完一个服务就更新对故障范围的判断直到收敛出根因。2.2 为什么选 LangGraph 来编排市面上做 Agent 编排的框架不少我最终选LangGraph主要是三个原因。第一它把 Agent 建模成图Graph而不是链Chain。日志调查天然有分支和循环——比如“查完一个服务如果发现还有下游依赖就回到排查节点继续查如果已经定位到根因就跳到报告节点”。这种带条件跳转和循环的结构用 LangGraph 的节点加边来表达非常自然。相比之下LangChain 的 Chain 更适合线性流程遇到需要回退和循环的场景就有点别扭。这也是为什么网上老有人问 langchain 和 langgraph 的区别核心就在这LangGraph 是状态机式的图编排LangChain 更偏线性的组件串联。第二状态管理是内置的。LangGraph 用一个共享的 State 对象在节点之间传递数据每个节点读取 State、处理后写回 State。日志调查过程中我们需要持续累积“已排查的服务”“发现的异常”“当前怀疑的根因”这些信息用 State 来承载再合适不过。第三它和 OpenAI 的模型调用集成得很顺。Agent 的“决策”环节需要 LLM 来判断LangGraph 里直接调 OpenAI 的接口就行不用自己再包一层。2.3 蓝耘元生代在架构里的位置蓝耘元生代在这个方案里承担的是模型服务底座的角色。简单说Agent 每次需要“思考”——比如判断某段日志是不是异常、决定下一步查哪个服务——都要调用一次大模型。蓝耘元生代提供了稳定的模型推理服务我们通过它的 API 来调用不用自己折腾模型部署和算力调度。对于团队里没有专门 MLOps 人力的场景这一点很省心把精力放在 Agent 逻辑上模型这块交给平台。整个架构的分工是这样的OpenTelemetry 负责采集链路和日志数据蓝耘元生代提供模型推理LangGraph 负责编排调查流程OpenAI 兼容接口作为模型调用的统一协议。四者各司其职拼成一个完整的日志调查 Agent。2.4 数据从哪来OpenTelemetry 的关键作用日志调查 Agent 要能工作前提是数据得齐。这里我强烈建议用OpenTelemetry做统一采集。原因很实际如果每个服务的日志格式都不一样Agent 光做格式解析就要累死。OpenTelemetry 的好处是它把Trace链路、Log日志、Metric指标三样东西用统一的语义约定串起来了。具体来说每条日志都带trace_id和span_id这样 Agent 就能顺着一条 trace 把跨服务的调用链还原出来。比如订单服务的一条超时日志它的trace_id能关联到库存服务的慢查询日志Agent 就能自动把这两条日志串成因果链。没有这个关联Agent 就只能靠时间戳猜准确率会掉一大截。我在实操里用的采集方案是应用侧接入 OpenTelemetry SDK日志通过 OTLP 协议发到 CollectorCollector 再统一转发到日志存储。这样不管后端服务用什么语言写的日志格式都是统一的Agent 处理起来省事很多。3. 核心节点拆解一个日志调查 Agent 该有哪些能力3.1 状态设计Agent 的“工作记忆”在写任何节点之前先把 State 设计好这是整个 Agent 的地基。我的 State 大概长这样from typing import TypedDict, Annotated import operator class InvestigationState(TypedDict): alert_info: dict # 告警原始信息 time_window: tuple # 调查时间窗口 entry_service: str # 入口服务 investigated_services: Annotated[list, operator.add] # 已排查服务 findings: Annotated[list, operator.add] # 发现的异常 suspected_root_cause: str # 当前怀疑的根因 next_action: str # 下一步动作 report: str # 最终报告这里有个细节值得说investigated_services和findings用了Annotated[list, operator.add]这是 LangGraph 的状态累加机制。意思是每个节点往这两个字段里写数据时是追加而不是覆盖。日志调查过程中每查一个服务就追加一条记录这样最后能完整还原调查路径。如果不用这个机制后一个节点会把前一个节点的结果覆盖掉调查过程就丢了。3.2 入口定位节点从告警到调查起点第一个节点负责把告警信息翻译成调查起点。输入是告警内容比如“订单服务 P99 延迟超过 2 秒”输出是入口服务和调查时间窗口。这个节点里我调了一次模型让 LLM 从告警文本里抽取结构化信息。为什么要用 LLM 而不是正则因为告警文本的写法太杂了有的写“订单服务”有的写“order-service”有的还带一堆指标名。用 LLM 抽取鲁棒性比写死正则好得多。def locate_entry(state: InvestigationState): prompt f从以下告警中提取入口服务和调查时间窗口 告警内容{state[alert_info][message]} 告警时间{state[alert_info][timestamp]} 请输出 JSON 格式包含 entry_service 和 time_window_minutes 两个字段。 response call_llm(prompt) # 调用蓝耘元生代模型服务 parsed parse_json(response) return { entry_service: parsed[entry_service], time_window: ( state[alert_info][timestamp] - parsed[time_window_minutes] * 60, state[alert_info][timestamp] 300 ) }时间窗口的设计有个经验往前多留一点往后少留一点。故障的诱因往往在告警之前就出现了所以往前留的时间要够我一般留 30 分钟而告警之后系统通常已经在恢复了往后留 5 分钟就够。这个不对称的窗口设计能显著提高找到根因的概率。3.3 日志检索节点精准捞取而不是全量拉取第二个节点根据入口服务和时间窗口去捞日志。这里最容易犯的错是一次性把时间窗口内所有日志都拉出来结果几百万行数据把上下文撑爆模型根本处理不了。我的做法是分层检索先只捞入口服务的 ERROR 和 WARN 级别日志如果发现异常再顺着 trace_id 去捞关联服务的日志。这样每次给模型的数据量都可控。def fetch_logs(state: InvestigationState): service state.get(current_service, state[entry_service]) start, end state[time_window] # 第一层捞当前服务的异常日志 logs log_client.query( serviceservice, startstart, endend, level[ERROR, WARN], limit200 ) # 提取 trace_id为下一层检索做准备 trace_ids list({log[trace_id] for log in logs if log.get(trace_id)}) return { current_logs: logs, related_trace_ids: trace_ids }注意limit这个参数一定要设。我一开始没设结果某次故障日志量太大直接把模型上下文撑爆Agent 报错退出。后来固定 200 条配合按时间倒序基本能覆盖关键信息。3.4 异常分析节点让模型做“有证据的判断”拿到日志后第三个节点让模型分析这批日志里有没有异常以及异常指向什么。这个节点的 prompt 设计是整个 Agent 里最关键的我改了好几版才稳定下来。核心原则是要求模型给出判断依据而不是只给结论。早期版本我只让它输出“有没有异常”结果它经常瞎猜。后来改成要求它引用具体的日志行作为证据准确率明显提升。def analyze_logs(state: InvestigationState): logs_text format_logs(state[current_logs]) prompt f你是一名资深 SRE正在排查线上故障。 当前服务{state.get(current_service, state[entry_service])} 日志内容 {logs_text} 请分析 1. 是否存在异常如果有异常的具体表现是什么 2. 每条异常判断必须引用具体日志行作为证据。 3. 如果日志中出现了对其他服务的调用失败或超时请指出下游服务名。 4. 输出 JSON{{has_anomaly: bool, anomalies: [...], downstream_services: [...]}} response call_llm(prompt) result parse_json(response) return { findings: [{service: state.get(current_service, state[entry_service]), **result}], downstream_services: result.get(downstream_services, []) }3.5 决策节点下一步查哪里这是 Agent 的“大脑”。它根据当前累积的 findings决定下一步是继续查下游服务、还是已经可以收敛出根因、还是需要扩大时间窗口重新查。def decide_next(state: InvestigationState): prompt f基于目前的调查结果决定下一步动作。 已排查服务{state[investigated_services]} 已发现异常{state[findings]} 待查下游服务{state.get(downstream_services, [])} 可选动作 - investigate_downstream继续排查下游服务 - conclude已能确定根因生成报告 - expand_window信息不足扩大时间窗口重查 输出 JSON{{action: ..., target_service: ..., reason: ...}} response call_llm(prompt) return {next_action: parse_json(response)}这个节点的 prompt 里我把可选动作明确列出来而不是让模型自由发挥。这是从踩坑里学来的给模型划定动作空间比让它自由决策稳定得多。早期版本我让它自己说下一步干嘛结果它经常输出一些我根本没实现的动作Agent 直接崩。3.6 报告生成节点带证据链的结论最后一个节点把所有 findings 汇总成一份调查报告。这份报告的结构我固定成四段故障概述、时间线、根因分析、证据链。其中证据链最重要每条结论后面都要附上对应的日志行这样复盘会上大家能直接核对不用再回去翻日志。def generate_report(state: InvestigationState): prompt f根据以下调查结果生成一份故障调查报告。 告警信息{state[alert_info]} 调查路径{state[investigated_services]} 发现的问题{state[findings]} 报告结构 1. 故障概述一段话 2. 时间线按时间顺序列出关键事件 3. 根因分析明确指出根因 4. 证据链每条结论附上对应日志行 要求所有结论必须有日志证据支撑不得臆测。 report call_llm(prompt) return {report: report}4. 用 LangGraph 把节点串成调查流程4.1 图的构建节点、边与条件跳转节点写好了接下来用 LangGraph 把它们串起来。核心是定义条件边——根据决策节点的输出决定走哪条路。from langgraph.graph import StateGraph, END workflow StateGraph(InvestigationState) # 添加节点 workflow.add_node(locate_entry, locate_entry) workflow.add_node(fetch_logs, fetch_logs) workflow.add_node(analyze_logs, analyze_logs) workflow.add_node(decide_next, decide_next) workflow.add_node(generate_report, generate_report) # 设置入口 workflow.set_entry_point(locate_entry) # 线性边 workflow.add_edge(locate_entry, fetch_logs) workflow.add_edge(fetch_logs, analyze_logs) workflow.add_edge(analyze_logs, decide_next) # 条件边根据决策结果跳转 def route_decision(state: InvestigationState): action state[next_action][action] if action investigate_downstream: return fetch_logs elif action expand_window: return fetch_logs else: return generate_report workflow.add_conditional_edges( decide_next, route_decision, { fetch_logs: fetch_logs, generate_report: generate_report } ) workflow.add_edge(generate_report, END) app workflow.compile()这段代码里最关键的是add_conditional_edges。它让decide_next节点可以根据状态动态决定下一步。注意investigate_downstream和expand_window都指向fetch_logs但进入fetch_logs时 State 里的current_service或time_window已经被决策节点更新过了所以实际查的是不同的东西。这就是 LangGraph 状态机的精髓同一个节点因为 State 不同行为就不同。4.2 循环控制防止 Agent 无限打转Agent 有个经典问题可能陷入死循环。比如决策节点一直说“继续查下游”但下游服务其实没有日志于是反复查同一个服务。我的解法是加一个最大迭代次数和已排查服务去重。在decide_next节点里如果发现待查服务已经在investigated_services里了就强制收敛。def decide_next(state: InvestigationState): # 强制收敛条件 if len(state[investigated_services]) 8: return {next_action: {action: conclude, reason: 已达最大排查深度}} downstream state.get(downstream_services, []) new_targets [s for s in downstream if s not in state[investigated_services]] if not new_targets and state[findings]: return {next_action: {action: conclude, reason: 无新下游可查已有足够发现}} # ... 正常决策逻辑提示最大迭代次数我设的是 8。这个数字是拍脑袋定的但实测下来够用——一般故障的因果链不会超过 5 层留点余量到 8 比较稳妥。设太大反而容易让 Agent 在无关服务上浪费时间。4.3 状态在节点间的流转实录为了让你直观感受 State 是怎么流转的我拿一次真实故障跑一遍。初始 State{ alert_info: {message: 订单服务 P99 延迟超过 2 秒, timestamp: 1700000000}, investigated_services: [], findings: [] }经过locate_entry后entry_service被设为order-servicetime_window设为告警前 30 分钟到后 5 分钟。经过fetch_logs和analyze_logs后findings里多了一条订单服务大量context deadline exceeded下游指向inventory-service。decide_next判断要查下游于是current_service被设为inventory-service回到fetch_logs。第二轮分析发现库存服务有慢查询日志trace_id指向数据库连接池耗尽。decide_next判断可以收敛跳到generate_report。最终报告里时间线、根因、证据链一应俱全。整个过程从触发到出报告实测 40 秒左右。5. 实操落地从零把 Agent 跑起来5.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.10 以上LangGraph 对低版本支持不太好。pip install langgraph langchain-openai opentelemetry-api opentelemetry-sdk这里langchain-openai是用来调 OpenAI 兼容接口的。因为我们用的是蓝耘元生代的模型服务它提供 OpenAI 兼容的 API所以直接复用这个库就行不用自己写 HTTP 请求。5.2 模型服务配置对接蓝耘元生代配置模型调用这块关键是base_url和api_key。蓝耘元生代的控制台里能拿到这两个值。from langchain_openai import ChatOpenAI llm ChatOpenAI( modelyour-model-name, base_urlhttps://your-lanyun-endpoint/v1, api_keyyour-api-key, temperature0.1 )temperature我设的 0.1因为日志分析需要稳定输出不需要创造性。设太高的话同样的日志两次分析结果可能不一样排查起来很头疼。注意如果你在本地调试时遇到model provider openai not found这类报错八成是base_url没配对。OpenAI 兼容接口的路径一定要带/v1少一段都不行。5.3 OpenTelemetry 数据接入实操数据接入这块我用的是 OpenTelemetry Collector 做中转。应用侧配置from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter provider TracerProvider() processor BatchSpanProcessor(OTLPSpanExporter(endpointhttp://collector:4317)) provider.add_span_processor(processor) trace.set_tracer_provider(provider)日志侧我建议在日志里显式带上trace_id。很多框架默认不带需要手动注入。这一步不做的话后面 Agent 就没法做跨服务关联效果会大打折扣。5.4 完整跑一遍一次真实故障的调查记录我拿之前一次真实的库存服务故障做了测试。告警是“订单服务 P99 延迟超过 2 秒”触发 Agent 后第 8 秒定位到入口服务order-service时间窗口确定第 15 秒捞到 47 条 ERROR 日志全是context deadline exceeded第 22 秒模型分析出下游指向inventory-service第 30 秒捞库存服务日志发现慢查询第 38 秒模型判断根因是数据库连接池耗尽第 45 秒报告生成完毕报告里明确写了根因是库存服务的某个查询在特定参数下走了全表扫描导致连接池被占满进而拖垮订单服务。证据链里附了三条关键日志。这份报告拿到复盘会上大家五分钟就对齐了结论。6. 踩坑记录与常见问题排查6.1 模型输出格式不稳定怎么办这是最常见的坑。模型有时候输出 JSON有时候输出带 markdown 代码块的 JSON有时候还夹带解释文字。我的解法是三层防护prompt 里明确要求“只输出 JSON不要任何其他文字”解析时先尝试直接json.loads失败就正则提取{...}再解析再失败就重试一次重试时把上次的错误输出也塞进 prompt 让它纠正。6.2 日志量太大导致上下文超限前面提过一定要设limit。另外我还会做日志去重——同样的错误信息出现几百次其实只需要保留几条样本加一个计数。这样能大幅压缩上下文。6.3 Agent 陷入循环的排查思路如果发现 Agent 反复查同一个服务先看investigated_services有没有正确累加。如果没累加多半是 State 的Annotated没配对。如果累加了但还在循环就是决策节点的收敛条件没写好加上最大迭代次数兜底。6.4 常见问题速查表问题现象可能原因解决方向模型调用报 401api_key 错误或过期检查蓝耘元生代控制台的 key报 provider not foundbase_url 路径不对确认带/v1后缀Agent 输出乱码模型返回非 JSON加强 prompt 约束加解析兜底调查结果不准日志缺 trace_id应用侧注入 trace_idAgent 跑很久不结束循环未收敛加最大迭代次数和去重报告没有证据链prompt 未强制要求明确要求引用日志行6.5 几个提升准确率的实操心得第一给模型喂日志时带上服务名和时间戳。别只给日志正文上下文信息越全模型判断越准。第二决策节点的动作空间要收窄。别让模型自由发挥把可选动作列清楚它就不会输出你没实现的动作。第三报告生成前先做一次 findings 去重。有时候同一个异常会被多个服务重复报告去重后报告更干净。第四时间窗口用不对称设计。前面说过往前多留往后少留这个细节对命中率影响很大。7. 后续可以怎么扩展这套 Agent 跑通之后我陆续加了几个扩展。一个是把 Metric 也接进来让 Agent 在分析日志的同时看指标曲线判断会更立体。另一个是把历史故障库接进来让 Agent 在决策时参考“上次类似故障是怎么解决的”相当于给它加了经验记忆。还有一个方向是把报告直接推到协作工具故障处理完自动生成报告并通知相关人省掉人工整理这一步。这块我还在试主要是格式对齐比较费劲但方向是明确的。如果你也在做类似的东西我的建议是先把最小闭环跑通——哪怕只查一个服务、只输出一段结论也比一上来就追求大而全要好。Agent 这东西跑起来之后迭代才有意义。