为什么选择LangGraph？一篇就够了，程序员必学收藏

为什么选择 LangGraph？

在 AI 世界中，检索增强生成（RAG）系统已广泛用于处理简单查询，生成上下文相关回答。

但随着 AI 应用复杂度不断提升，我们迫切需要一种能执行多步推理、保持状态和具备动态决策能力的智能体系统。

这正是 LangGraph 登场的舞台：

💡LangGraph 是 LangChain 的高级扩展，支持具备状态管理与循环控制能力的多角色智能体。

✨ LangGraph 的核心优势

LangGraph 通过无缝管理图结构、状态和智能体协作，重新定义了 AI 应用的开发方式：

LangGraph 的结构概念图（包括节点、状态与边的关系）

✅ 功能亮点：

•状态自动管理：跨轮对话中持续保留上下文•多工具调度协调：多角色、多工具之间高效协作•循环计算能力：支持反复推理与修正•灵活+可扩展+稳定：适用于原型验证到企业部署

🔨 项目目标示例

我们将通过一个实战案例来构建智能体：

💬根据用户每月电费计算太阳能节省效果的 AI 助手。

📌 应用场景：

•可部署在太阳能销售网站上•与访客互动、收集数据、给出节省建议•同时辅助销售线索筛选，实现业务转化

🧱 开发步骤详解

🚦 第 1 步：导入必要的库

from langchain_core.tools import toolfrom langchain_community.tools.tavily_search importTavilySearchResultsfrom langchain_core.prompts importChatPromptTemplatefrom langchain_core.runnables importRunnablefrom langchain_aws importChatBedrockimport boto3from typing importAnnotatedfrom typing_extensions importTypedDictfrom langgraph.graph.message importAnyMessage, add_messagesfrom langchain_core.messages importToolMessagefrom langchain_core.runnables importRunnableLambdafrom langgraph.prebuilt importToolNode, tools_condition

🧮 第 2 步：定义节能计算工具

@tooldef compute_savings(monthly_cost:float)->float:"""工具功能：根据用户每月电费计算切换到太阳能后的潜在节省。 参数: monthly_cost (float):用户当前每月电费支出（单位：美元） 返回: dict:包含以下内容的字典：-'number_of_panels':预计所需太阳能板数量-'installation_cost':安装费用估算-'net_savings_10_years':安装后10年内的净节省金额"""def calculate_solar_savings(monthly_cost):# 计算的基础假设值 cost_per_kWh =0.28# 每度电（千瓦时）价格（美元） cost_per_watt =1.50# 每瓦安装成本（美元） sunlight_hours_per_day =3.5# 平均每日日照时长（小时） panel_wattage =350# 单块太阳能板功率（瓦） system_lifetime_years =10# 系统使用年限 # 用户每月用电量（kWh） monthly_consumption_kWh = monthly_cost / cost_per_kWh # 系统所需功率（千瓦） daily_energy_production = monthly_consumption_kWh /30 system_size_kW = daily_energy_production / sunlight_hours_per_day # 计算面板数量和安装成本 number_of_panels = system_size_kW *1000/ panel_wattage installation_cost = system_size_kW *1000* cost_per_watt # 年节省金额与10年净收益 annual_savings = monthly_cost *12 total_savings_10_years = annual_savings * system_lifetime_years net_savings = total_savings_10_years - installation_cost return{"number_of_panels": round(number_of_panels),"installation_cost": round(installation_cost,2),"net_savings_10_years": round(net_savings,2)} # 返回计算结果return calculate_solar_savings(monthly_cost)

📈 工具返回结果：

•所需太阳能板数量•安装成本估算•十年期净节省金额

🧩 第 3 步：状态管理与错误处理

处理工具执行异常

def handle_tool_error(state)-> dict:"""工具执行出错时的错误处理函数。 参数: state (dict):当前 AI 智能体的状态，包含消息内容和工具调用信息。 返回: dict:包含每个出错工具对应错误消息的字典。"""# 从状态中提取错误信息 error = state.get("error") # 获取最近一条消息中的工具调用记录 tool_calls = state["messages"][-1].tool_calls # 返回包含错误提示的 ToolMessage 列表，每个消息关联一个 tool_call_idreturn{"messages":[ToolMessage( content=f"Error: {repr(error)}\n 请修复错误。", tool_call_id=tc["id"],)for tc in tool_calls]}

创建带回退机制的工具节点

def create_tool_node_with_fallback(tools: list)-> dict:"""创建包含错误回退机制的工具节点。 参数: tools (list):要包含在节点中的工具列表。 返回: dict:一个支持错误处理回退的工具节点。"""returnToolNode(tools).with_fallbacks([RunnableLambda(handle_tool_error)], exception_key="error")

🧠 第 4 步：定义状态结构与助手类

状态结构State

classState(TypedDict): messages:Annotated[list[AnyMessage], add_messages]

助手类Assistant

classAssistant:def __init__(self, runnable:Runnable):# 初始化，接收定义好流程的 runnable 对象self.runnable = runnable def __call__(self, state:State):whileTrue:# 使用当前状态调用 runnable result =self.runnable.invoke(state) # 如果没有工具调用结果，或者内容为空，则重新请求有效输出ifnot result.tool_calls and(not result.contentor isinstance(result.content, list)andnot result.content[0].get("text")): messages = state["messages"]+[("user","请返回有效输出。")] state ={**state,"messages": messages}else:break# 成功获取有效结果，退出循环 # 返回最终消息状态return{"messages": result}

🧱 第 5 步：使用 AWS Bedrock 配置 LLM

def get_bedrock_client(region):return boto3.client("bedrock-runtime", region_name=region) def create_bedrock_llm(client):returnChatBedrock( model_id='anthropic.claude-3-sonnet-20240229-v1:0', client=client, model_kwargs={'temperature':0}, region_name='us-east-1') llm = create_bedrock_llm(get_bedrock_client(region='us-east-1'))

✅ 确保你已设置 AWS 凭证，否则助手无法访问 Bedrock 模型。

🧭 第 6 步：定义助手工作流程

对话提示模板

primary_assistant_prompt =ChatPromptTemplate.from_messages([("system",'''你是 Solar Panels Belgium 的客户支持助手。你需要向用户获取以下信息：-每月电费支出 如果无法识别，请请求用户澄清，切勿胡乱猜测。收集完毕后，请调用相应工具。'''),("placeholder","{messages}"),])

工具绑定与执行流定义

# 定义助手会使用的工具part_1_tools =[ compute_savings] # 将工具绑定到助手的工作流程中part_1_assistant_runnable = primary_assistant_prompt | llm.bind_tools(part_1_tools)

🔄 第 7 步：构建 LangGraph 图结构

在这一步中，我们使用LangGraph来构建 AI 助手的图结构，控制它如何处理用户输入、调用工具并在不同阶段之间流转。

builder =StateGraph(State)builder.add_node("assistant",Assistant(part_1_assistant_runnable))builder.add_node("tools", create_tool_node_with_fallback(part_1_tools)) builder.add_edge(START,"assistant")# 从助手开始builder.add_conditional_edges("assistant", tools_condition)# 根据输入跳转到工具builder.add_edge("tools","assistant")# 工具执行后返回助手 memory =MemorySaver()graph = builder.compile(checkpointer=memory)

▶️ 第 8 步：运行 AI 助手

import uuid # 示例用户问题列表tutorial_questions =['hey','can you calculate my energy saving',"my montly cost is $100, what will i save"] # 创建唯一线程 ID，用于会话追踪thread_id = str(uuid.uuid4()) # 配置上下文config ={"configurable":{"thread_id": thread_id,}} # 存储已打印事件（防止重复）_printed =set() # 模拟与助手的对话流for question in tutorial_questions: events = graph.stream({"messages":("user", question)}, config, stream_mode="values")foreventin events: _print_event(event, _printed)