本文介绍了如何使用 LangGraph 构建具备状态管理和循环计算能力的 AI 智能体。通过太阳能板节能计算示例，演示了从环境配置、工具定义、状态管理到图结构构建的完整流程。内容涵盖 LangChain 集成、AWS Bedrock 模型调用、错误处理机制及对话持久化策略，帮助开发者掌握复杂多步任务编排技术，实现可交互、自适应的企业级 AI 应用开发。重点讲解了 StateGraph 的定义、条件边逻辑、MemorySaver 的使用以及生产环境的部署优化建议。
在人工智能领域，检索增强生成（RAG）系统已成为处理简单查询并生成上下文相关响应的常见工具。然而，随着对更复杂人工智能应用需求的增长，我们需要超越这些检索能力的系统。于是，AI 智能体（AI Agent）应运而生——这些自主实体能够执行复杂的多步骤任务，在交互过程中保持状态，并动态适应新信息。
LangGraph 作为 LangChain 库的强大扩展，旨在帮助开发者构建这些高级 AI 智能体，通过支持具有循环计算能力的有状态、多参与者应用来实现这一目标。
在本文中，我们将探讨 LangGraph 如何改变 AI 开发，并通过一个计算太阳能板节能效果的示例，逐步说明如何构建自己的 AI 智能体。此示例将展示 LangGraph 的独特功能如何创建智能、适应性强且适用于现实世界的 AI 系统。
LangGraph 是一个构建于 LangChain 之上的高级库，旨在通过引入循环计算能力来增强您的大型语言模型（LLM）应用。虽然 LangChain 允许创建用于线性工作流的有向无环图（DAG），但 LangGraph 更进一步，支持添加循环，这对于开发复杂的、类似智能体的行为至关重要。这些行为使得 LLM 能够持续循环执行某个过程，根据不断变化的条件动态决定下一步要采取的行动。
LangGraph 的核心概念是状态图（State Graph）：
LangGraph 通过无缝管理图结构、状态和协调，重新定义了 AI 开发，使创建复杂的多参与者应用成为可能。借助自动状态管理，LangGraph 确保在交互过程中保留上下文，使 AI 能够智能地响应变化的输入。其简化的智能体协调保证了精确的执行和高效的信息交换，让开发者能够专注于创新工作流的设计，而不是技术细节。LangGraph 的灵活性允许开发定制的高性能应用，而其可扩展性和容错性确保您的系统即使在企业级应用中也能保持稳健和可靠。
现在我们已经对 LangGraph 是什么以及它如何增强 AI 开发有了基本理解，接下来深入一个实际示例。在这个场景中，将构建一个 AI 智能体，旨在根据用户输入计算太阳能板的潜在节能效果。该智能体可以作为太阳能板销售商网站上的潜在客户生成工具，与潜在客户互动，提供个性化的节能估算。通过收集诸如每月电费成本等关键数据，该 AI 智能体帮助客户了解太阳能的经济效益，同时为销售团队的后续跟进筛选潜在客户。
首先导入项目所需的所有基本 Python 库和模块。
from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable
from langchain_aws import ChatBedrock
import boto3
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableLambda
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition
这些导入中包含了后续将要用到的 LangChain、LangGraph 以及 AWS 服务。
定义一个工具，该工具将根据用户提供的每月电费成本来计算节能效果。
@tool # 使用@tool 装饰器将该函数标记为一个工具，使其可以在 LangGraph 中被调用
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 = 1.00 # 每度电的成本。此数值根据需要进行修改，这里仅仅是示例
cost_per_watt = 1.50 # 每瓦太阳能板的成本。此数值根据需要进行修改，这里仅仅是示例
sunlight_hours_per_day = 3.5 # 每天的平均日照小时数
panel_wattage = 350 # 每块太阳能板的功率（瓦）
system_lifetime_years = 10 # 太阳能系统的使用寿命（年）
# 计算每月用电量（单位：千瓦时）
monthly_consumption_kWh = monthly_cost / cost_per_kWh
# 计算所需的系统容量（单位：千瓦）
daily_energy_production = monthly_consumption_kWh / 30 # 假设每月 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 # 10 年总节省金额
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)
代码说明：
@tool 装饰器：将该函数标记为一个工具，使其可以在 LangGraph 中被调用。calculate_solar_savings 函数：内部函数，用于执行具体的节能计算。该函数处理用户的每月电费成本，并返回太阳能板系统效益的详细估算。为了简化计算，我们在其中做了一些假设，例如每度电的平均成本和平均日照小时数。然而，在更高级版本的 AI 智能体中，我们可以直接从用户那里获取这些信息，从而更精确地根据用户的独特情况定制估算结果。
有效的状态管理和错误处理对于构建健壮的 AI 系统至关重要。在此步骤中，我们定义了一些工具来管理错误并维护对话的状态。
def handle_tool_error(state) -> dict:
"""
处理工具执行过程中发生的错误的函数。
参数:
state (dict): AI 智能体的当前状态，包括消息和工具调用详情。
返回:
dict: 包含每个遇到问题的工具的错误消息的字典。
"""
# 从当前状态中获取错误信息
error = state.get("error")
# 从状态的消息历史中获取最后一次消息的工具调用
tool_calls = state["messages"][-1].tool_calls
# 返回一个包含错误详情的 ToolMessage 列表，每个消息与对应的工具调用 ID 关联
return {
"messages": [
ToolMessage(
content=f"Error: {repr(error)}\n please fix your mistakes.", # 格式化错误消息以便用户理解
tool_call_id=tc["id"], # 将错误消息与对应的工具调用 ID 关联
)
for tc in tool_calls # 遍历每个工具调用，生成单独的错误消息
]
}
def create_tool_node_with_fallback(tools: list) -> dict:
"""
创建一个带有错误回退处理的工具节点的函数。
参数:
tools (list): 包含在节点中的工具列表。
返回:
dict: 一个带有错误回退行为的工具节点。
"""
# 使用提供的工具创建 ToolNode，并附加一个回退机制
# 如果发生错误，将调用 handle_tool_error 函数来处理错误
return ToolNode(tools).with_fallbacks(
[RunnableLambda(handle_tool_error)], # 使用 lambda 函数包装错误处理函数
exception_key="error" # 指定此回退机制用于处理错误
)
代码说明：
ToolMessage 列表，每个消息与对应的工具调用 ID 关联。ToolNode，并附加一个回退机制。如果发生错误，将调用 handle_tool_error 函数来处理错误。handle_tool_error 函数，系统能够在工具执行失败时提供有意义的错误反馈。create_tool_node_with_fallback 函数确保在工具执行失败时，系统能够优雅地处理错误并继续运行。这些函数确保在工具执行过程中遇到的任何错误都能得到优雅处理，并为用户提供有用的反馈。
在这一步中，定义 AI 智能体如何管理其状态（对话的持续上下文），并确保它能够适应用户输入和工具输出做出响应。
为此，使用 Python 的 TypedDict 创建一个 State 类，用于定义传递的消息结构。状态将保存消息，包括来自用户的输入以及来自智能体或工具的输出。
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
接下来，创建 Assistant 类，该类负责运行 AI 智能体、与工具交互以及管理对话的流程。Assistant 调用工具，确保它们返回适当的结果，并处理执行过程中可能出现的重新提示或错误。其核心功能包括调用 Runnable，该 Runnable 定义了调用 LLM 和工具（如 compute_savings）的过程，然后监控结果。如果智能体未能返回有效响应，或者工具未提供有意义的数据，Assistant 会重新提示用户或请求澄清。它会持续循环执行 Runnable，直到获得有效输出，从而确保顺利执行和有效响应。
class Assistant:
def __init__(self, runnable: Runnable):
# 初始化时传入一个 runnable 对象，该对象定义了与工具交互的流程
self.runnable = runnable
def __call__(self, state: State):
while True:
# 使用当前状态（消息和上下文）调用 runnable
result = self.runnable.invoke(state)
# 如果工具未能返回有效输出，重新提示用户澄清或重试
if not result.tool_calls and (
not result.content
or isinstance(result.content, list)
and not result.content[0].get("text")
):
# 添加一条消息，要求用户提供有效响应
messages = state["messages"] + [("user", "Respond with a real output.")]
state = {**state, "messages": messages}
else:
# 当获得有效输出时，退出循环
break
# 返回处理 runnable 后的最终状态
return {"messages": result}
代码说明：
Assistant 类：该类负责管理 AI 智能体的执行流程，确保工具调用和用户交互的顺利进行。通过 runnable 对象定义与工具和 LLM 的交互逻辑。__call__ 方法：该方法用于执行智能体的主要逻辑。通过循环调用 runnable，直到获得有效输出。如果工具调用失败或返回无效结果，会重新提示用户提供有效输入。state 对象维护对话的上下文和消息历史。while True 循环，确保智能体在获得有效输出之前持续运行。这种设置对于维持对话的流畅性至关重要，并确保助手能够根据上下文做出适当的响应。
在这一步中，使用 AWS Bedrock 配置大型语言模型（LLM），当然也可以使用其他模型。关于 AWS 的有关应用，可以参考其官方文档。
def get_bedrock_client(region):
return boto3.client("bedrock-runtime", region_name=region)
def create_bedrock_llm(client):
return ChatBedrock(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 Secrets Manager 或环境变量，避免硬编码敏感信息。
以上已经配置好了 LLM 和工具，下一步是定义 AI 助手的工作流程。这包括创建对话模板，指定助手将使用的工具，并配置 AI 智能体如何响应用户输入以及触发不同的功能（如计算太阳能节能效果）。工作流程本质上是控制助手如何与用户交互、收集信息并调用工具以提供结果的逻辑。
工作流程的第一部分涉及创建提示模板，该模板定义了助手如何与用户沟通。提示帮助引导 AI 助手确定要询问用户的内容、如何根据输入进行响应以及何时触发像 compute_savings 这样的工具。
在这种情况下，助手需要询问用户的每月电费成本以计算太阳能板的节能效果。以下是定义对话的方式：
primary_assistant_prompt = ChatPromptTemplate.from_messages(
[
(
"system", # 系统消息，用于定义助手的行为和任务
'''你是一位乐于助人的太阳能板客户支持助手。
需要从用户那里获取以下信息：
- 每月电费成本
如果无法获取这些信息，请要求用户澄清！不要随意猜测。
获取所有信息后，调用相关工具
''',
),
("placeholder", "{messages}"), # 占位符，用于插入对话历史或用户输入
]
)
代码说明：
ChatPromptTemplate.from_messages：用于创建一个对话提示模板，定义助手的行为和对话流程。system 系统消息：定义了助手的基本任务和行为准则。助手需要从用户那里获取每月电费成本，并在无法获取时要求用户澄清。获取所有必要信息后，助手会调用相关工具（如 compute_savings）。{messages} 是一个占位符，用于插入对话历史或用户输入，确保对话的上下文能够动态更新。接下来，定义助手在交互过程中将使用的工具，其中主要工具是 compute_savings，它根据用户的每月电费成本计算潜在的节能效果。在列表中指定工具后，使用 llm.bind_tools() 方法将它们绑定到助手的工作流程中。此步骤确保 AI 助手能够在对话过程中根据需要访问并触发这些工具，从而在用户和助手之间创建无缝的交互体验。
# 定义助手将使用的工具
part_1_tools = [
compute_savings # 主要工具：计算太阳能节能效果
]
# 将工具绑定到助手的工作流程
part_1_assistant_runnable = primary_assistant_prompt | llm.bind_tools(part_1_tools)
代码说明：
part_1_tools：定义助手在交互过程中将使用的工具列表。当前只有一个工具 compute_savings，用于计算太阳能节能效果。llm.bind_tools(part_1_tools)：将工具列表绑定到 LLM（大型语言模型），使助手能够在对话过程中调用这些工具。这一步确保助手能够根据需要动态触发工具。primary_assistant_prompt | llm.bind_tools(part_1_tools)：使用管道操作符 | 将提示模板（primary_assistant_prompt）与绑定工具后的 LLM 结合起来。这样，助手的工作流程既包含了对话逻辑，也具备了调用工具的能力。bind_tools 方法，助手能够在对话中动态调用工具，实现更复杂的功能。在这一步中，使用 LangGraph 为 AI 助手构建图结构，该结构控制助手如何处理用户输入、触发工具以及在各个阶段之间切换。图结构定义了核心动作的结点（如调用助手和工具）以及边，这些边决定了结点之间的流程。
在 LangGraph 中，每个结点代表一个操作步骤，例如与用户交互或执行工具。为这个 AI 助手定义了两个关键结点：
compute_savings）以计算用户的太阳能板节能效果。# 创建一个 StateGraph 实例，用于定义图结构
builder = StateGraph(State)
# 添加助手节点，负责管理对话流程
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
# 添加工具节点，负责执行工具（如 compute_savings）并处理错误
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))
代码说明：
StateGraph(State)：创建一个 StateGraph 实例，用于定义图结构。State 是之前定义的状态类，用于维护对话的上下文和消息历史。add_node("assistant", Assistant(part_1_assistant_runnable))：添加一个名为 assistant 的节点，该节点由 Assistant 类实例化。part_1_assistant_runnable 是助手的工作流程，结合了提示模板和工具调用逻辑。add_node("tools", create_tool_node_with_fallback(part_1_tools))：添加一个名为 tools 的节点，该节点负责执行工具（如 compute_savings）。create_tool_node_with_fallback 函数确保在工具执行失败时能够优雅地处理错误。边定义了流程在结点之间的移动方式。在这里，助手启动对话，然后在收集到所需的输入后过渡到工具结点，并在工具执行完毕后返回到助手结点。
# 添加从起始节点到助手节点的边，表示对话从助手开始
builder.add_edge(START, "assistant") # 从助手开始
# 添加条件边，根据条件决定是否从助手节点转移到工具节点
builder.add_conditional_edges("assistant", tools_condition) # 在收集到输入后转移到工具节点
# 添加从工具节点返回到助手节点的边，表示工具执行完毕后继续由助手处理
builder.add_edge("tools", "assistant") # 工具执行后返回助手
代码说明：
add_edge(START, "assistant")：定义从起始节点（START）到助手节点（assistant）的边。表示对话流程从助手开始，助手首先与用户交互。add_conditional_edges("assistant", tools_condition)：添加条件边，根据 tools_condition 函数的返回值决定是否从助手节点转移到工具节点。tools_condition 函数用于判断是否需要调用工具（例如，是否已收集到用户的电费成本）。add_edge("tools", "assistant")：定义从工具节点（tools）返回到助手节点（assistant）的边。表示工具执行完毕后，流程返回到助手，助手继续与用户交互。用 MemorySaver 来确保图结构在不同步骤之间保留对话状态。这使得助手能够记住用户的输入，从而在多步骤交互中保持连续性。
# 创建一个 MemorySaver 实例，用于保存对话状态
memory = MemorySaver()
# 编译图结构，并传入 MemorySaver 作为检查点管理器
graph = builder.compile(checkpointer=memory)
代码说明：
MemorySaver()：创建一个 MemorySaver 实例，用于保存和恢复对话状态。确保在多步骤交互中，助手能够记住用户的输入和上下文。builder.compile(checkpointer=memory)：编译图结构，使其可以执行。传入 memory 作为检查点管理器，确保对话状态能够在不同步骤之间持久化。最后，通过初始化图结构并启动对话来运行助手。
# 导入必要的库
import uuid # 用于生成唯一 ID
# 创建一个示例对话，模拟用户与助手的交互
tutorial_questions = [
'hey', # 用户打招呼
'can you calculate my energy saving', # 用户请求计算节能效果
"my montly cost is $100, what will i save" # 用户提供每月电费成本并询问节省金额
]
# 生成一个唯一的对话 ID
thread_id = str(uuid.uuid4())
# 配置参数，包含对话 ID
config = {
"configurable": {
"thread_id": thread_id, # 将生成的唯一 ID 作为对话 ID
}
}
# 用于跟踪已打印的事件，避免重复打印
_printed = set()
# 遍历用户的问题，模拟对话流程
for question in tutorial_questions:
# 使用 graph.stream 方法处理用户输入，并获取事件流
events = graph.stream(
{"messages": ("user", question)}, config, stream_mode="values"
)
# 遍历事件流并打印事件内容
for event in events:
_print_event(event, _printed) # 打印事件内容，确保不重复打印
代码说明：
tutorial_questions：模拟用户与助手的对话内容，包含用户打招呼、请求计算节能效果以及提供每月电费成本。uuid.uuid4()：生成一个唯一的对话 ID，用于标识当前对话会话。config：配置参数，包含对话 ID，确保对话状态能够被正确保存和恢复。_printed：用于跟踪已打印的事件，避免在事件流中重复打印相同内容。graph.stream：使用 graph.stream 方法处理用户输入，并返回事件流。stream_mode="values" 表示只返回事件的值，而不是完整的事件对象。在完成基础功能的构建后，为了确保 AI 智能体在生产环境中的稳定性和安全性，建议关注以下几点：
asyncio 提高吞吐量。MemorySaver 仅适用于内存测试。生产环境建议使用数据库（如 PostgreSQL）配合 PostgresSaver 进行持久化存储，确保会话数据不丢失。通过上述步骤，已成功使用 LangGraph 创建了一个 AI 助手，该助手能够根据用户输入计算太阳能板的节能效果。本教程展示了 LangGraph 在管理复杂的多步骤流程中的强大能力，并强调了如何利用先进的 AI 工具高效解决现实世界的挑战。无论是为客户支持、能源管理还是其他应用开发 AI 智能体，LangGraph 都提供了实现创意所需的灵活性、可扩展性和稳健性。
掌握 LangGraph 的核心概念，如状态图、结点和边，能够帮助开发者构建具备记忆能力和循环逻辑的高级智能体。随着大模型技术的不断发展，LangGraph 将成为构建下一代企业级 AI 应用的重要基石。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online