当前位置：首页 > news >正文

LangGraph入门教程：构建循环状态管理的LLM应用

news 2025/6/13 14:08:02

LangGraph入门教程：构建循环状态管理的LLM应用

在大型语言模型(LLM)应用开发中，我们常常需要处理复杂的工作流程，而不仅仅是简单的线性执行。LangGraph作为LangChain的高级库，为我们提供了构建具有循环能力的LLM应用的强大工具。本文将通过一个简单的天气查询示例，带你入门LangGraph的核心概念和基本使用方法。

LangGraph核心概念

LangGraph是LangChain的一个高级库，它最大的特点是为LLM应用带来了循环能力，而不仅仅是传统LangChain的线性工作流。这使得我们可以构建更复杂、更智能的应用程序。LangGraph的核心组件包括：

状态：维护过程上下文，实现累计数据的动态决策
节点：带边计算步骤，制定特定的任务，适应不同的工作流
边：连接节点，定义工作流的执行路径，实现复杂的工作流程控制

构建一个天气查询应用

下面，我们将通过构建一个简单的天气查询应用来展示LangGraph的基本用法。

1. 环境准备和导入依赖

首先，我们需要导入必要的库和设置环境：

import os
from typing import Literalfrom dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI# 导入检查点，用于持久化保存
from langgraph.checkpoint.memory import Checkpoint, MemorySaver# 导入图以及状态
from langgraph.graph import END, StateGraph, MessagesState
# 导入工具节点
from langgraph.prebuilt import ToolNode
load_dotenv()

2. 定义工具

接下来，我们定义一个用于获取天气的简单工具：

@tool
def get_weather(region: str):"""获取指定地区的天气:param region: 地区:return:"""return f"{region}晴天，25度"# 将工具放入工具列表中
tools = [get_weather]# 创建工具节点
tool_node = ToolNode(tools)

3. 初始化模型并绑定工具

我们使用ChatOpenAI模型，并将工具绑定到模型上：

model = ChatOpenAI(base_url="https://openrouter.ai/api/v1", api_key=os.getenv("OPENAI_KEY"),model="google/gemini-2.0-flash-001", temperature=0
).bind_tools(tools)

4. 定义决策函数和模型调用函数

这是LangGraph的关键部分，我们需要定义一个决策函数来控制工作流的流向，以及一个用于调用模型的函数：

def should_continue(state: MessagesState) -> Literal["tools", END]:print("messages:", state['messages'])messages = state['messages']last_message = messages[-1]# 如果llm调用了工具，则转到"tools"节点if last_message.tool_calls:return "tools"# 否则，停止return ENDdef call_model(state: MessagesState):messages = state['messages']# 调用模型ai_msg = model.invoke(messages)# 返回列表并将最新生成的消息添加到列表中return {"messages": [ai_msg]}

5. 构建状态图

现在，我们可以开始构建状态图了：

# 初始化图。定义新的状态图
workflow = StateGraph(MessagesState)
# 定义图节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)# 定义入口和图边
# 第一个被调用的节点是agent
workflow.set_entry_point('agent')# 添加条件边
workflow.add_conditional_edges(# 起始节点，使用agent"agent",# 传递下一个调用节点的数据should_continue)# 添加从tools到agent的普通边
# 这就意味着调用tools节点后，返回到agent节点继续处理
workflow.add_edge("tools", "agent")

6. 设置检查点和编译图

# 初始化内存，在图运行之间的持久化状态
checkpointer = MemorySaver()  # 也可在redis、MongoDB中存# 编译图
# 这将其编译成一个LangChain可运行的对象
app = workflow.compile(checkpointer)

7. 执行图并获取结果

最后，我们可以执行这个图并获取结果：

final_state = app.invoke({"messages": [HumanMessage(content="今天上海天气如何？")]},config={"configurable": {"thread_id": 42}},  # 使用检查点来保存状态
)# 从final_state获取最后一条消息的内容
result = final_state['messages'][-1].content
print(result)

8. 测试会话状态保持

LangGraph一个重要的特性是可以保持会话状态。我们可以继续提问，系统会记住之前的上下文：

final_state = app.invoke({"messages": [HumanMessage(content="我问的是哪个城市？")]},config={"configurable": {"thread_id": 42}},  # 使用检查点来保存状态
)# 从final_state获取最后一条消息的内容
result = final_state['messages'][-1].content
print(result)