1. 理解LangGraph的核心概念在开始实战之前我们需要先搞清楚LangGraph的几个关键概念。LangGraph是LangChain团队推出的新一代Agent开发框架它的核心创新在于引入了图结构和状态管理这两个重要特性。想象一下传统的LangChain工作流就像是一条直线上的火车轨道列车只能按照固定路线前进。而LangGraph则更像是一个地铁网络列车可以根据实时情况选择不同路线甚至可以绕圈运行。这种灵活性正是现代AI应用开发最需要的。LangGraph中的图指的是有向图数据结构由节点Node和边Edge组成。每个节点代表一个处理单元可以是大模型调用工具函数执行条件判断数据处理操作边则定义了节点之间的流转关系。特别重要的是LangGraph支持循环图这意味着工作流可以反复执行某些节点直到满足特定条件。2. 搭建开发环境2.1 基础环境配置首先确保你已经安装了Python 3.8版本。我推荐使用conda创建虚拟环境conda create -n langgraph python3.10 conda activate langgraph然后安装LangGraph核心包和常用工具pip install langgraph langchain-core langchain-community如果你计划使用特定的大模型服务比如DeepSeek还需要安装对应的适配器pip install langchain-deepseek2.2 模型接入配置在项目根目录创建.env文件存放你的API密钥DEEPSEEK_API_KEYyour_api_key_here OPENWEATHER_API_KEYyour_weather_api_key然后在代码中加载这些配置from dotenv import load_dotenv load_dotenv() from langchain.chat_models import init_chat_model model init_chat_model(modeldeepseek-chat, model_providerdeepseek)3. 构建第一个有状态Agent3.1 定义工具函数让我们创建一个实用的天气查询工具。这个工具将被用作我们Agent的一个能力from langchain_core.tools import tool from pydantic import BaseModel, Field import requests, json class WeatherQuery(BaseModel): loc: str Field(description查询天气的城市名称) tool(args_schemaWeatherQuery) def get_weather(loc: str): 查询指定城市的实时天气 url https://api.openweathermap.org/data/2.5/weather params { q: loc, appid: os.getenv(OPENWEATHER_API_KEY), units: metric, lang: zh_cn } response requests.get(url, paramsparams) return json.dumps(response.json())3.2 创建基础图结构现在我们来构建一个简单的循环图Agent。这个Agent会持续询问用户是否需要查询天气直到用户明确表示不需要为止。from langgraph.graph import Graph from langgraph.prebuilt import ToolNode from langchain_core.messages import HumanMessage, AIMessage # 初始化图 graph Graph() # 添加工具节点 tools [get_weather] tool_node ToolNode(tools) # 定义状态结构 from typing import TypedDict, List, Annotated from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[List[BaseMessage], lambda x, y: x y] # 添加节点 graph.add_node(tool, tool_node) graph.add_node(model, model) # 定义边关系 def should_continue(state: AgentState): last_message state[messages][-1] if 不需要 in last_message.content: return end return continue graph.add_conditional_edges( model, should_continue, {continue: tool, end: end} ) graph.add_edge(tool, model) # 设置入口和出口 graph.set_entry_point(model) graph.set_finish_point(end) # 编译成可执行Agent agent graph.compile()3.3 测试Agent现在我们可以测试这个有状态的Agent了response agent.invoke({ messages: [HumanMessage(content我想查询天气)] }) for msg in response[messages]: print(f{msg.type}: {msg.content})这个Agent会持续询问你要查询哪个城市的天气直到你回复不需要为止。这就是一个简单的有状态循环图Agent的完整实现。4. 高级状态管理技巧4.1 自定义状态更新默认情况下LangGraph会将所有节点的输出追加到消息历史中。但有时我们需要更精细的状态控制from langgraph.graph import StateGraph def model_node(state: AgentState): messages state[messages] last_message messages[-1] if isinstance(last_message, HumanMessage): response model.invoke(messages) return {messages: [response]} return {messages: []} def tool_node(state: AgentState): messages state[messages] last_message messages[-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: results [] for tool_call in last_message.tool_calls: tool next(t for t in tools if t.name tool_call[name]) output tool.invoke(tool_call[args]) results.append(output) return {messages: [ToolMessage(contentstr(results), tool_call_idtool_call[id])]} return {messages: []} # 使用StateGraph替代Graph graph StateGraph(AgentState) graph.add_node(model, model_node) graph.add_node(tool, tool_node)4.2 持久化状态对于长时间运行的Agent我们需要将状态保存到外部存储from langgraph.checkpoint import FileSystemCheckpointer checkpointer FileSystemCheckpointer(base_dir./checkpoints) agent graph.compile( checkpointercheckpointer, interrupt_before[tool] ) # 现在Agent会自动保存状态 thread_id user_123 agent.invoke( {messages: [HumanMessage(content查询北京天气)]}, {configurable: {thread_id: thread_id}} ) # 可以从上次中断处恢复 agent.invoke( {messages: [HumanMessage(content继续)]}, {configurable: {thread_id: thread_id}} )5. 调试与优化5.1 使用LangSmith监控LangSmith是LangChain生态中的监控平台可以实时跟踪Agent执行import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_PROJECT] MyWeatherAgent5.2 性能优化技巧对于复杂的图结构可以考虑以下优化方法并行执行对于没有依赖关系的节点可以并行执行from langgraph.graph import END graph.add_edge(tool1, model) graph.add_edge(tool2, model)缓存机制对频繁调用的工具结果进行缓存from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())超时控制防止单个节点执行时间过长from langchain_community.tools import Tool safe_weather_tool Tool( namesafe_get_weather, funclambda x: get_weather.with_config({run_name: get_weather, max_execution_time: 30})(x), description查询天气30秒超时 )6. 实战构建旅游规划Agent让我们把这些知识综合起来构建一个更复杂的旅游规划Agent。这个Agent能够查询目的地天气推荐当地景点规划行程路线记忆用户偏好6.1 定义工具集首先准备几个必要的工具函数tool def get_attractions(city: str) - str: 获取城市热门景点 # 这里可以接入旅游API return f{city}的著名景点景点1, 景点2, 景点3 tool def plan_itinerary(cities: list[str], days: int) - str: 根据城市和天数生成行程计划 return f{days}天行程计划第一天游览{cities[0]}的景点1和景点2... tools [get_weather, get_attractions, plan_itinerary]6.2 构建状态图from typing import Literal class TourismState(TypedDict): messages: List[BaseMessage] preferences: dict current_step: Literal[weather, attractions, planning] graph StateGraph(TourismState) # 定义多个专业节点 def weather_node(state: TourismState): # 实现天气查询逻辑 pass def attractions_node(state: TourismState): # 实现景点查询逻辑 pass def planning_node(state: TourismState): # 实现行程规划逻辑 pass # 添加节点和边 graph.add_node(weather, weather_node) graph.add_node(attractions, attractions_node) graph.add_node(planning, planning_node) # 定义复杂流转逻辑 def router(state: TourismState): if 天气 in state[messages][-1].content: return weather elif 景点 in state[messages][-1].content: return attractions else: return planning graph.add_conditional_edges(model, router) graph.add_edge(weather, model) graph.add_edge(attractions, model) graph.add_edge(planning, model) # 编译最终Agent tourism_agent graph.compile()6.3 测试完整流程response tourism_agent.invoke({ messages: [HumanMessage(content我想规划一个北京3日游)], preferences: {}, current_step: start }) print(response[messages][-1].content)这个Agent会根据用户输入自动判断应该执行哪个功能模块并在各模块间传递状态信息形成一个完整的工作闭环。 32、LangGraph开发框架（二）-- 实战构建有状态循环图Agent
1. 理解LangGraph的核心概念在开始实战之前我们需要先搞清楚LangGraph的几个关键概念。LangGraph是LangChain团队推出的新一代Agent开发框架它的核心创新在于引入了图结构和状态管理这两个重要特性。想象一下传统的LangChain工作流就像是一条直线上的火车轨道列车只能按照固定路线前进。而LangGraph则更像是一个地铁网络列车可以根据实时情况选择不同路线甚至可以绕圈运行。这种灵活性正是现代AI应用开发最需要的。LangGraph中的图指的是有向图数据结构由节点Node和边Edge组成。每个节点代表一个处理单元可以是大模型调用工具函数执行条件判断数据处理操作边则定义了节点之间的流转关系。特别重要的是LangGraph支持循环图这意味着工作流可以反复执行某些节点直到满足特定条件。2. 搭建开发环境2.1 基础环境配置首先确保你已经安装了Python 3.8版本。我推荐使用conda创建虚拟环境conda create -n langgraph python3.10 conda activate langgraph然后安装LangGraph核心包和常用工具pip install langgraph langchain-core langchain-community如果你计划使用特定的大模型服务比如DeepSeek还需要安装对应的适配器pip install langchain-deepseek2.2 模型接入配置在项目根目录创建.env文件存放你的API密钥DEEPSEEK_API_KEYyour_api_key_here OPENWEATHER_API_KEYyour_weather_api_key然后在代码中加载这些配置from dotenv import load_dotenv load_dotenv() from langchain.chat_models import init_chat_model model init_chat_model(modeldeepseek-chat, model_providerdeepseek)3. 构建第一个有状态Agent3.1 定义工具函数让我们创建一个实用的天气查询工具。这个工具将被用作我们Agent的一个能力from langchain_core.tools import tool from pydantic import BaseModel, Field import requests, json class WeatherQuery(BaseModel): loc: str Field(description查询天气的城市名称) tool(args_schemaWeatherQuery) def get_weather(loc: str): 查询指定城市的实时天气 url https://api.openweathermap.org/data/2.5/weather params { q: loc, appid: os.getenv(OPENWEATHER_API_KEY), units: metric, lang: zh_cn } response requests.get(url, paramsparams) return json.dumps(response.json())3.2 创建基础图结构现在我们来构建一个简单的循环图Agent。这个Agent会持续询问用户是否需要查询天气直到用户明确表示不需要为止。from langgraph.graph import Graph from langgraph.prebuilt import ToolNode from langchain_core.messages import HumanMessage, AIMessage # 初始化图 graph Graph() # 添加工具节点 tools [get_weather] tool_node ToolNode(tools) # 定义状态结构 from typing import TypedDict, List, Annotated from langchain_core.messages import BaseMessage class AgentState(TypedDict): messages: Annotated[List[BaseMessage], lambda x, y: x y] # 添加节点 graph.add_node(tool, tool_node) graph.add_node(model, model) # 定义边关系 def should_continue(state: AgentState): last_message state[messages][-1] if 不需要 in last_message.content: return end return continue graph.add_conditional_edges( model, should_continue, {continue: tool, end: end} ) graph.add_edge(tool, model) # 设置入口和出口 graph.set_entry_point(model) graph.set_finish_point(end) # 编译成可执行Agent agent graph.compile()3.3 测试Agent现在我们可以测试这个有状态的Agent了response agent.invoke({ messages: [HumanMessage(content我想查询天气)] }) for msg in response[messages]: print(f{msg.type}: {msg.content})这个Agent会持续询问你要查询哪个城市的天气直到你回复不需要为止。这就是一个简单的有状态循环图Agent的完整实现。4. 高级状态管理技巧4.1 自定义状态更新默认情况下LangGraph会将所有节点的输出追加到消息历史中。但有时我们需要更精细的状态控制from langgraph.graph import StateGraph def model_node(state: AgentState): messages state[messages] last_message messages[-1] if isinstance(last_message, HumanMessage): response model.invoke(messages) return {messages: [response]} return {messages: []} def tool_node(state: AgentState): messages state[messages] last_message messages[-1] if isinstance(last_message, AIMessage) and last_message.tool_calls: results [] for tool_call in last_message.tool_calls: tool next(t for t in tools if t.name tool_call[name]) output tool.invoke(tool_call[args]) results.append(output) return {messages: [ToolMessage(contentstr(results), tool_call_idtool_call[id])]} return {messages: []} # 使用StateGraph替代Graph graph StateGraph(AgentState) graph.add_node(model, model_node) graph.add_node(tool, tool_node)4.2 持久化状态对于长时间运行的Agent我们需要将状态保存到外部存储from langgraph.checkpoint import FileSystemCheckpointer checkpointer FileSystemCheckpointer(base_dir./checkpoints) agent graph.compile( checkpointercheckpointer, interrupt_before[tool] ) # 现在Agent会自动保存状态 thread_id user_123 agent.invoke( {messages: [HumanMessage(content查询北京天气)]}, {configurable: {thread_id: thread_id}} ) # 可以从上次中断处恢复 agent.invoke( {messages: [HumanMessage(content继续)]}, {configurable: {thread_id: thread_id}} )5. 调试与优化5.1 使用LangSmith监控LangSmith是LangChain生态中的监控平台可以实时跟踪Agent执行import os os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_PROJECT] MyWeatherAgent5.2 性能优化技巧对于复杂的图结构可以考虑以下优化方法并行执行对于没有依赖关系的节点可以并行执行from langgraph.graph import END graph.add_edge(tool1, model) graph.add_edge(tool2, model)缓存机制对频繁调用的工具结果进行缓存from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())超时控制防止单个节点执行时间过长from langchain_community.tools import Tool safe_weather_tool Tool( namesafe_get_weather, funclambda x: get_weather.with_config({run_name: get_weather, max_execution_time: 30})(x), description查询天气30秒超时 )6. 实战构建旅游规划Agent让我们把这些知识综合起来构建一个更复杂的旅游规划Agent。这个Agent能够查询目的地天气推荐当地景点规划行程路线记忆用户偏好6.1 定义工具集首先准备几个必要的工具函数tool def get_attractions(city: str) - str: 获取城市热门景点 # 这里可以接入旅游API return f{city}的著名景点景点1, 景点2, 景点3 tool def plan_itinerary(cities: list[str], days: int) - str: 根据城市和天数生成行程计划 return f{days}天行程计划第一天游览{cities[0]}的景点1和景点2... tools [get_weather, get_attractions, plan_itinerary]6.2 构建状态图from typing import Literal class TourismState(TypedDict): messages: List[BaseMessage] preferences: dict current_step: Literal[weather, attractions, planning] graph StateGraph(TourismState) # 定义多个专业节点 def weather_node(state: TourismState): # 实现天气查询逻辑 pass def attractions_node(state: TourismState): # 实现景点查询逻辑 pass def planning_node(state: TourismState): # 实现行程规划逻辑 pass # 添加节点和边 graph.add_node(weather, weather_node) graph.add_node(attractions, attractions_node) graph.add_node(planning, planning_node) # 定义复杂流转逻辑 def router(state: TourismState): if 天气 in state[messages][-1].content: return weather elif 景点 in state[messages][-1].content: return attractions else: return planning graph.add_conditional_edges(model, router) graph.add_edge(weather, model) graph.add_edge(attractions, model) graph.add_edge(planning, model) # 编译最终Agent tourism_agent graph.compile()6.3 测试完整流程response tourism_agent.invoke({ messages: [HumanMessage(content我想规划一个北京3日游)], preferences: {}, current_step: start }) print(response[messages][-1].content)这个Agent会根据用户输入自动判断应该执行哪个功能模块并在各模块间传递状态信息形成一个完整的工作闭环。