software

使用 LangGraph、LangChain 和 Watsonx.ai 调查错误日志

发布: (2025年12月11日 GMT+8 18:36)
10 min read
原文: Dev.to

Source: Dev.to

介绍

在处理生产系统时，可观测性起着关键作用。它是事件调查的核心组成部分，是监控和告警的基础，并且在验证新功能、改进或修复的发布时极其有用。应用日志是可观测性的重要组成部分。1 日志可以帮助我们以高度细粒度了解系统在特定时间点的行为。

然而，理解应用日志可能很困难。日志数量庞大，找到相关日志具有挑战性。对日志进行索引并使用搜索引擎查询可以帮助，但它无法告诉你哪些日志与正在调查的问题相关。当我在日志中看到与事件时间相吻合的错误时，通常会自问一系列后续问题：

  • 该错误是否与问题相关，甚至可能是根本原因？
  • 这个错误是已知问题吗？
  • 如果已知，是否已经报告给了相应的团队？
  • 如果已经报告，是否正在处理或已修复？
  • 如果已修复，修复是否已经部署到我正在调查的环境中？
  • 如果已经部署，错误为何仍然出现？是否出现了回归？
  • 如果出现回归，是否已经报告给了相应的团队？

在事件处理中回答这些问题会消耗宝贵的时间。但答案往往包含关键信息——比如 bug 单中的解决办法或可以立即应用的热修复。因此，在事件期间进行更深入的日志调查非常有价值，我也相信生成式 AI 能够快速帮助回答这些问题。

在本文中，我们将探讨如何使用 IBM Watsonx.ai、LangGraph 和 LangChain 在 Python 中调查错误日志。文章结构如下：

  1. 技术基础——介绍 LangGraph、LangChain 与 Watsonx.ai。
  2. 日志调查代理的设计与实现。
  3. 结果总结与未来工作展望。

LangGraph、LangChain 与 Watsonx.ai

LangGraph

LangGraph 是一个基于图的编排框架，用于构建有状态的 AI 工作流（例如代理）。它让你可以将 AI 应用建模为有向图，其中：

  • 节点 是函数（LLM 调用、工具调用或自定义逻辑），它们在共享状态上操作。
  • 边 定义控制流，包括条件分支和循环。
  • 状态 是显式的共享数据结构（例如 dict 或 TypedDict），所有节点都可以读取和更新，从而轻松构建长期运行的有状态代理。

LangChain

LangChain 常被用作 LangGraph 节点内部的构建块。它提供了将 LLM 与数据和工具连接的实用工具。通过结合 LangChain 与 LangGraph，你可以构建能够循环推理和行动的 AI 代理，并可选地加入人工环节。

Watsonx.ai

Watsonx.ai 是 IBM 的企业 AI 平台，提供托管的 LLM 等能力。我们将把这三种工具结合起来，构建一个调查日志的 AI 代理。所需的 Python 包包括：

简单示例

下面的代码演示了一个拥有天气查询工具的基础代理。它使用 create_agent 辅助函数（create_react_agent 的后继）构建一个预配置的图，并在全局状态中维护消息历史。

from ibm_watsonx_ai import ChatWatsonx
from langchain_ibm import create_agent
import os

llm = ChatWatsonx(
    model_id="meta-llama/llama-3-70b-instruct",
    url=os.getenv("WATSONX_URL"),
)

def get_weather(city: str) -> str:
    return f"Weather in {city}: 30°C and sunny."

agent = create_agent(llm, tools=[get_weather])

response = agent.invoke({
    "messages": [
        {"role": "user", "content": "What is the weather in Berlin?"}
    ]
})
print(response)

对于更复杂的应用，你可以使用 Graph API 自行构建图。

日志调查代理

范围

该代理将回答前文提出的高层次问题。本文聚焦于三个核心能力：

  1. 搜索相关工作/工单/对话——在 Jira 和 GitHub 中并行搜索（演示 LangGraph 的并行特性）。该设计可扩展至 Slack、事件跟踪工具、事后报告或公司级搜索引擎（如 Glean）。
  2. 获取运行时上下文——检索 pod/容器名称和部署版本，以评估找到的条目是否相关。
  3. 调查工单和对话——查找解决办法或修复信息。

该代理将被封装在一个轻量级的 Dash UI 中（如下图所示）。

日志调查 UI

定义状态

在 LangGraph 中，状态在所有节点之间共享，并沿边传递。当多个节点并发修改同一属性时，需要自定义合并函数。对于本案例，我们将所有中间结果存入状态，以便在图执行完毕后展示给用户，增强对代理推理过程的信任。

from pydantic import BaseModel
from typing import Optional

class LogInvestigationState(BaseModel):
    # 用户提供的原始日志文本
    log_text: Optional[str] = None
    # 后续会添加的字段（例如搜索查询、结果、上下文等）

定义图

高层图结构包括以下步骤：

  1. 检查日志并生成查询——第一个节点解析提供的日志，并为每个外部系统（Jira、GitHub）生成搜索查询。
  2. 并行搜索——节点并行调用相应系统的 API；此步骤不需要 LLM。
  3. 生成工单节点——使用 Send 功能，图为每个找到的工单/对话生成一个节点。
  4. 评估工单相关性——LLM 根据标题、描述、完整日志和运行时上下文对每个工单进行评分，以判断其相关性。
  5. 汇总发现——最终节点聚合最相关的工单，提取解决办法，并为用户生成简洁答案。

下面是图定义的骨架（省略了 API 包装器和合并逻辑的细节）。

from langgraph.graph import StateGraph, Send
from langchain_ibm import ChatWatsonx
from typing import List, Dict
import os

# 初始化 LLM
llm = ChatWatsonx(
    model_id="meta-llama/llama-3-70b-instruct",
    url=os.getenv("WATSONX_URL"),
)

def inspect_log(state: LogInvestigationState) -> Dict:
    """从原始日志生成搜索查询。"""
    queries = llm.invoke(
        f"Extract concise search terms from the following log:\n{state.log_text}"
    )
    return {"queries": queries.split(",")}

def search_jira(state: Dict) -> Dict:
    """使用生成的查询搜索 Jira。"""
    # 实际 Jira API 调用的占位实现
    results = []  # Ticket 字典列表
    return {"jira_results": results}

def search_github(state: Dict) -> Dict:
    """使用生成的查询搜索 GitHub issue/PR。"""
    results = []  # Issue 字典列表
    return {"github_results": results}

def grade_ticket(state: Dict, ticket: Dict) -> Dict:
    """使用 LLM 对单个工单进行相关性评分。"""
    prompt = f"""
Log: {state['log_text']}
Operational context: {state.get('context', '')}
Ticket title: {ticket['title']}
Ticket description: {ticket['description']}
Rate the relevance of this ticket to the log on a scale of 0‑10 and provide a short justification.
"""
    rating = llm.invoke(prompt)
    ticket["rating"] = rating
    return {"graded_ticket": ticket}

def summarize(state: Dict) -> Dict:
    """为用户生成最终摘要。"""
    relevant = [
        t for t in state.get("graded_tickets", [])
        if int(t["rating"].split()[0]) >= 7
    ]
    summary = llm.invoke(
        f"Summarize the following relevant tickets and any suggested workarounds:\n{relevant}"
    )
    return {"summary": summary}

# 构建图
graph = StateGraph(LogInvestigationState)

graph.add_node("inspect_log", inspect_log)
graph.add_node("search_jira", search_jira)
graph.add_node("search_github", search_github)
graph.add_node("grade_ticket", grade_ticket)
graph.add_node("summarize", summarize)

# 定义边
graph.add_edge("inspect_log", "search_jira")
graph.add_edge("inspect_log", "search_github")
graph.add_conditional_edges(
    "search_jira",
    lambda s: Send("grade_ticket", s["jira_results"]),
)
graph.add_conditional_edges(
    "search_github",
    lambda s: Send("grade_ticket", s["github_results"]),
)
graph.add_edge("grade_ticket", "summarize")
graph.set_entry_point("inspect_log")

log_investigation_app = graph.compile()

编译后的图（log_investigation_app）可以使用包含原始日志文本的字典进行调用。最终的 summary 字段即可在 Dash UI 中展示。

结论

通过将 LangGraph 用于并行搜索编排、LangChain 用于 LLM 推理、以及 Watsonx.ai 提供强大模型推断相结合，我们能够构建一个能够快速回答事件调查中关键问题的代理。模块化的图结构使得向代理添加额外数据源（如 Slack、内部知识库）或更丰富的运行时上下文变得十分简便。

未来工作可能包括：

  • 基于日志内容的 动态工具选择。
  • 反馈回路：让人工能够批准或拒绝建议的解决办法。
  • 缓存 搜索结果，以降低重复调查的延迟。

Footnotes

  1. 可观测性基础——参见任何标准的可观测性参考资料。 ↩

Back to Blog

相关文章

阅读更多 »