【LangGraph】使用 LangGraph 的 Pregel 运行时日志

创想小编 · 发表于 12 小时前

作者：彬彬侠
使用 LangGraph 时，Pregel 作为其底层计算模型，驱动图的执行，而运行时日志（Runtime Logging）是调试和监控 LangGraph 工作流的重要工具。LangGraph 提供了灵活的日志机制，帮助开发者跟踪节点执行、状态更新、消息传递等信息。以下是详细介绍如何使用 LangGraph 的 Pregel 运行时日志，结合 Pregel 的执行机制，确保解答准确、清晰、全面，并以中文呈现。

1. LangGraph 和 Pregel 的运行时日志背景

LangGraph 基于 Pregel 模型运行图工作流，每个超步（Superstep）执行一组节点，更新状态，并通过边传递控制流。运行时日志可以帮助：

调试

监控

优化

LangGraph 的日志通常依赖 Python 的日志模块（logging）或 LangChain 生态的回调机制（Callbacks）。Pregel 的运行时日志记录了每个超步的节点执行、状态更新和条件跳转等信息。

2. 设置 LangGraph 运行时日志的步骤

以下是使用 LangGraph 的 Pregel 运行时日志的详细步骤，涵盖基本配置、自定义日志和高级用法。
步骤 1：配置 Python 日志模块

LangGraph 使用 Python 的 logging 模块记录运行时信息。你需要确保日志级别和输出格式正确配置。
示例代码：

import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,# 设置日志级别（DEBUG、INFO、WARNING、ERROR）format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler()# 输出到控制台# 可添加 logging.FileHandler("logfile.log") 输出到文件])# 获取 LangGraph 的日志器
logger = logging.getLogger("langgraph")

复制代码

说明：

步骤 2：构建 LangGraph 工作流

假设你有一个简单的 LangGraph 工作流，包含状态和节点。我们将为其添加日志。
示例代码：

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph
from pydantic import Field
# 定义状态classState(TypedDict):
user_input: Annotated[str, Field(description="User's input")]
response: Annotated[str, Field(description="Generated response")]
step_count:int# 节点：生成响应defgenerate_response(state: State)-> State:
logger.info(f"Processing user_input: {state['user_input']}")
response =f"Echo: {state['user_input']}"
logger.debug(f"Generated response: {response}")return{"response": response,"step_count": state["step_count"]+1}# 构建图
graph = StateGraph(State)
graph.add_node("generate", generate_response)
graph.set_entry_point("generate")
graph.set_finish_point("generate")# 编译图
app = graph.compile()

复制代码

说明：

步骤 3：运行并查看日志

运行工作流，观察 Pregel 的运行时日志。
示例代码：

# 执行工作流
initial_state ={"user_input":"Hello!","response":"","step_count":0}
result = app.invoke(initial_state)# 打印结果print(result)

复制代码

示例输出（日志）：

2025-05-14 21:21:45,123 - langgraph - INFO - Processing user_input: Hello!
2025-05-14 21:21:45,124 - langgraph - DEBUG - Generated response: Echo: Hello!
{'user_input': 'Hello!', 'response': 'Echo: Hello!', 'step_count': 1}

复制代码

说明：

步骤 4：使用 LangChain 回调机制

LangGraph 继承了 LangChain 的回调系统（Callbacks），可以通过自定义回调捕获更详细的运行时信息，如节点开始/结束、状态更新等。
示例代码：

from langchain_core.callbacks import BaseCallbackHandler
from langgraph.graph import StateGraph
# 自定义回调classLoggingCallback(BaseCallbackHandler):defon_chain_start(self, serialized, inputs,**kwargs):
logger.info(f"Starting graph execution with inputs: {inputs}")defon_chain_end(self, outputs,**kwargs):
logger.info(f"Graph execution completed with outputs: {outputs}")defon_node_start(self, serialized, inputs, node_name,**kwargs):
logger.debug(f"Node {node_name} started with inputs: {inputs}")defon_node_end(self, outputs, node_name,**kwargs):
logger.debug(f"Node {node_name} ended with outputs: {outputs}")# 构建图（同上）
graph = StateGraph(State)
graph.add_node("generate", generate_response)
graph.set_entry_point("generate")
graph.set_finish_point("generate")# 编译图并添加回调
app = graph.compile(callbacks=[LoggingCallback()])# 执行
result = app.invoke({"user_input":"Hello!","response":"","step_count":0})

复制代码

示例输出：

2025-05-14 21:21:45,123 - langgraph - INFO - Starting graph execution with inputs: {'user_input': 'Hello!', 'response': '', 'step_count': 0}
2025-05-14 21:21:45,124 - langgraph - DEBUG - Node generate started with inputs: {'user_input': 'Hello!', 'response': '', 'step_count': 0}
2025-05-14 21:21:45,125 - langgraph - INFO - Processing user_input: Hello!
2025-05-14 21:21:45,126 - langgraph - DEBUG - Generated response: Echo: Hello!
2025-05-14 21:21 Virgen,125 - langgraph - DEBUG - Node generate ended with outputs: {'response': 'Echo: Hello!', 'step_count': 1}
2025-05-14 21:21:45,127 - langgraph - INFO - Graph execution completed with outputs: {'user_input': 'Hello!', 'response': 'Echo: Hello!', 'step_count': 1}

复制代码

说明：

步骤 5：高级日志配置（可选）

对于复杂工作流，可以进一步优化日志：

输出到文件：
1. logging.basicConfig(
2. level=logging.DEBUG,
3. handlers=[
4. logging.FileHandler("langgraph.log"),# 保存到文件
5. logging.StreamHandler()# 同时输出到控制台])
复制代码
结构化日志：
使用 structlog 或 loguru 替代标准 logging，支持 JSON 格式日志，便于分析。
1. from loguru import logger
3. # 替换标准 logger
4. logger.add("langgraph.log",format="{time} {level} {message}", level="DEBUG")
复制代码
集成外部工具：

3. Pregel 日志的关键信息

在 LangGraph 的 Pregel 运行时日志中，你可以捕获以下关键信息：

超步信息：
节点执行：
状态更新：
条件跳转：
错误和异常：
消息传递：

示例：捕获条件跳转日志

from langgraph.graph import StateGraph, END
# 定义状态classState(TypedDict):
user_input:str
response:str
step_count:int# 节点：决定是否继续defdecide_continue(state: State)->str:
logger.info(f"Deciding with step_count: {state['step_count']}")if state["step_count"]>=3:return END
return"generate"# 节点：生成响应defgenerate_response(state: State)-> State:
logger.info(f"Generating response for: {state['user_input']}")return{"response":f"Echo: {state['user_input']}","step_count": state["step_count"]+1}# 构建图
graph = StateGraph(State)
graph.add_node("generate", generate_response)
graph.add_node("decide", decide_continue)
graph.add_edge("generate","decide")
graph.add_conditional_edges("decide", decide_continue)
graph.set_entry_point("generate")# 编译并添加回调
app = graph.compile(callbacks=[LoggingCallback()])# 执行
result = app.invoke({"user_input":"Hello!","response":"","step_count":0})

复制代码

示例输出：

2025-05-14 21:21:45,123 - langgraph - INFO - Starting graph execution with inputs: {'user_input': 'Hello!', 'response': '', 'step_count': 0}
2025-05-14 21:21:45,124 - langgraph - DEBUG - Node generate started with inputs: {'user_input': 'Hello!', 'response': '', 'step_count': 0}
2025-05-14 21:21:45,125 - langgraph - INFO - Generating response for: Hello!
2025-05-14 21:21:45,126 - langgraph - DEBUG - Node generate ended with outputs: {'response': 'Echo: Hello!', 'step_count': 1}
2025-05-14 21:21:45,127 - langgraph - DEBUG - Node decide started with inputs: {'user_input': 'Hello!', 'response': 'Echo: Hello!', 'step_count': 1}
2025-05-14 21:21:45,128 - langgraph - INFO - Deciding with step_count: 1
2025-05-14 21:21:45,129 - langgraph - DEBUG - Node decide ended with outputs: generate
...
2025-05-14 21:21:45,135 - langgraph - INFO - Graph execution completed with outputs: {'user_input': 'Hello!', 'response': 'Echo: Hello!', 'step_count': 3}

复制代码

说明：

4. 调试和优化技巧

调试节点逻辑

使用回调的 on_node_error 捕获节点异常：
1. defon_node_error(self, error, node_name,**kwargs):
2. logger.error(f"Node {node_name} failed with error: {error}")
复制代码

分析性能

记录每个超步的执行时间：
1. import time
3. defgenerate_response(state: State)-> State:
4. start_time = time.time()
5. response =f"Echo: {state['user_input']}"
6. logger.info(f"Node generate took {time.time()- start_time:.2f} seconds")return{"response": response,"step_count": state["step_count"]+1}
复制代码

过滤无关日志

设置特定模块的日志级别：
1. logging.getLogger("langgraph").setLevel(logging.DEBUG)
2. logging.getLogger("langchain").setLevel(logging.WARNING)# 降低其他模块日志
复制代码

可视化日志

5. 常见问题解答

如何捕获 Pregel 的超步详细信息？

日志量过大怎么办？

如何记录工具调用日志？

为工具函数添加日志：
1. from langchain.tools import tool
3. @tooldefsearch(query:str)->str:
4. logger.info(f"Tool search called with query: {query}")returnf"Results for {query}"
复制代码

分布式环境下的日志如何处理？

6. 总结

LangGraph 的 Pregel 日志

配置方法：
关键信息：超步执行、节点输入/输出、状态变化、条件跳转、错误。优化技巧：调试节点、分析性能、过滤日志、可视化执行路径。Pregel 的作用：驱动 LangGraph 的迭代执行，日志反映了超步的计算和状态传递。

原文地址：https://blog.csdn.net/u013172930/article/details/147980148

25位大厂高管转战生成式AI创业！吸金猛，不

【LangGraph】使用 LangGraph 的 Pregel 运行时日志

浏览过的版块

向量数据库和嵌入模型

关于我们

产品与服务

解决方案

产品与服务