为什么选择 LangGraph？

在人工智能的应用实践中，检索增强生成（RAG）系统已成为处理简单信息查询的成熟方案——它能通过检索外部知识库获取精准信息，快速生成与上下文高度匹配的回答，有效解决了大模型“知识过时”“幻觉生成”等基础问题。

然而，随着AI应用向复杂场景（如多步骤任务规划、动态流程协作、长周期决策支持等）深入，仅靠RAG的“单次检索-单次生成”模式已难以满足需求：我们迫切需要一种能实现多步逻辑推理、实时保持任务状态、动态调整决策路径的智能体系统，来应对更具挑战性的复杂任务。

而这，正是LangGraph的核心价值所在——它专为构建具备“持续交互与动态推理能力”的AI智能体而生，恰好填补了复杂场景下的技术空白：

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

✨ LangGraph 的核心优势

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

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

✅ 功能亮点：

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

🔨 项目目标示例

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

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

📌 应用场景：

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

🧱 开发步骤详解

🚦 第 1 步：导入必要的库

from
 langchain_core.
tools
 
import
 tool
from
 langchain_community.
tools
.
tavily_search
 importTavilySearchResults
from
 langchain_core.
prompts
 importChatPromptTemplate
from
 langchain_core.
runnables
 importRunnable
from
 langchain_aws importChatBedrock
import
 boto3
from
 typing importAnnotated
from
 typing_extensions importTypedDict
from
 langgraph.
graph
.
message
 importAnyMessage, add_messages
from
 langchain_core.
messages
 importToolMessage
from
 langchain_core.
runnables
 importRunnableLambda
from
 langgraph.
prebuilt
 importToolNode, tools_condition

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

@tool
def
 
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_id
return
{
"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.content
or
 
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)

✅ 总结

通过以上 8 个步骤，你成功构建了一个支持多轮交互、状态记忆、错误处理与工具调度的智能太阳能节能助手。

🎯 LangGraph 的价值体现在：

•高度模块化与逻辑清晰的图式结构•灵活调用工具，响应用户个性化需求•支持真实业务场景（如能源、客服、营销）部署

💡 不论是构建客服智能体、节能顾问，还是更复杂的 AI 应用，LangGraph 都为你提供强大的基础设施和开发范式。

📌 欢迎将其扩展应用到更多实际项目中！

更多信息

hread_id,
}
}

存储已打印事件（防止重复）

_printed =
set
()

模拟与助手的对话流

for
question
in
tutorial_questions:
events = graph.stream(
{
“messages”
😦
“user”
, question)}, config, stream_mode=
“values”
)
foreventin events:
_print_event(event, _printed)

✅ 总结

通过以上 8 个步骤，你成功构建了一个支持多轮交互、状态记忆、错误处理与工具调度的智能太阳能节能助手。

🎯 LangGraph 的价值体现在：

•高度模块化与逻辑清晰的图式结构•灵活调用工具，响应用户个性化需求•支持真实业务场景（如能源、客服、营销）部署

💡 不论是构建客服智能体、节能顾问，还是更复杂的 AI 应用，LangGraph 都为你提供强大的基础设施和开发范式。

📌 欢迎将其扩展应用到更多实际项目中！

读者福利大放送：如果你对大模型感兴趣，想更加深入的学习大模型**，那么这份精心整理的大模型学习资料，绝对能帮你少走弯路、快速入门**

如果你是零基础小白，别担心——大模型入门真的没那么难，你完全可以学得会！

👉 不用你懂任何算法和数学知识，公式推导、复杂原理这些都不用操心；
👉 也不挑电脑配置，普通家用电脑完全能 hold 住，不用额外花钱升级设备；
👉 更不用你提前学 Python 之类的编程语言，零基础照样能上手。

你要做的特别简单：跟着我的讲解走，照着教程里的步骤一步步操作就行。

包括：大模型学习线路汇总、学习阶段，大模型实战案例，大模型学习视频，人工智能、机器学习、大模型书籍PDF。带你从零基础系统性的学好大模型！

现在这份资料免费分享给大家，有需要的小伙伴，直接VX扫描下方二维码就能领取啦😝↓↓↓

为什么要学习大模型？

数据显示，2023 年我国大模型相关人才缺口已突破百万，这一数字直接暴露了人才培养体系的严重滞后与供给不足。而随着人工智能技术的飞速迭代，产业对专业人才的需求将呈爆发式增长，据预测，到 2025 年这一缺口将急剧扩大至 400 万！!

大模型学习路线汇总

整体的学习路线分成L1到L4四个阶段，一步步带你从入门到进阶，从理论到实战，跟着学习路线一步步打卡，小白也能轻松学会！

大模型实战项目&配套源码

光学理论可不够，这套学习资料还包含了丰富的实战案例，让你在实战中检验成果巩固所学知识

大模型学习必看书籍PDF

我精选了一系列大模型技术的书籍和学习文档(电子版)，它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

大模型超全面试题汇总

在面试过程中可能遇到的问题，我都给大家汇总好了，能让你们在面试中游刃有余

这些资料真的有用吗？

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

👉获取方式：

😝有需要的小伙伴，可以保存图片到VX扫描下方二维码免费领取【保证100%免费】

相信我，这套大模型系统教程将会是全网最齐全 最适合零基础的！！