引言
在人工智能领域，检索增强生成（RAG）系统已成为处理简单查询并生成上下文相关响应的常见工具。然而，随着对更复杂人工智能应用需求的增长，我们需要超越这些检索能力的系统。于是，AI 智能体（AI Agent）应运而生——这些自主实体能够执行复杂的多步骤任务，在交互过程中保持状态，并动态适应新信息。LangGraph 作为 LangChain 库的强大扩展，旨在帮助开发者构建这些高级 AI 智能体，通过支持具有循环计算能力的有状态、多参与者应用来实现这一目标。
在本文中，我们将探讨LangGraph如何改变AI开发，并通过一个计算太阳能板节能效果的示例，逐步说明如何构建自己的AI智能体。此示例将展示LangGraph的独特功能如何创建智能、适应性强且适用于现实世界的AI系统。
什么是LangGraph？
LangGraph是一个构建于LangChain之上的高级库，旨在通过引入循环计算能力来增强您的大型语言模型（LLM）应用。虽然LangChain允许创建用于线性工作流的有向无环图（DAG），但LangGraph更进一步，支持添加循环，这对于开发复杂的、类似智能体的行为至关重要。这些行为使得LLM能够持续循环执行某个过程，根据不断变化的条件动态决定下一步要采取的行动。
LangGraph的核心概念是状态图：
状态：表示在计算过程中维护和更新的上下文或记忆。它确保图中的每一步都能访问之前步骤的相关信息，从而允许基于整个过程中积累的数据进行动态决策。 结点：作为图的基本构建块，代表单独的计算步骤或函数。每个结点执行特定任务，如处理输入、做出决策或与外部系统交互。结点可以自定义，以在工作流中执行各种操作。 边：连接图中的结点，定义从一个步骤到下一个步骤的计算流程。它们支持条件逻辑，允许执行路径根据当前状态发生变化，并促进数据和控制在结点之间的流动，从而实现复杂的多步骤工作流。
LangGraph通过无缝管理图结构、状态和协调，重新定义了AI开发，使创建复杂的多参与者应用成为可能。借助自动状态管理，LangGraph确保在交互过程中保留上下文，使AI能够智能地响应变化的输入。其简化的智能体协调保证了精确的执行和高效的信息交换，让开发者能够专注于创新工作流的设计，而不是技术细节。LangGraph的灵活性允许开发定制的高性能应用，而其可扩展性和容错性确保您的系统即使在企业级应用中也能保持稳健和可靠。
逐步指南
现在我们已经对LangGraph是什么以及它如何增强AI开发有了基本理解，接下来深入一个实际示例。在这个场景中，将构建一个AI智能体，旨在根据用户输入计算太阳能板的潜在节能效果。该智能体可以作为太阳能板销售商网站上的潜在客户生成工具，与潜在客户互动，提供个性化的节能估算。通过收集诸如每月电费成本等关键数据，该AI智能体帮助客户了解太阳能的经济效益，同时为销售团队的后续跟进筛选潜在客户。此示例展示了LangGraph在创建智能、动态系统方面的强大能力，这些系统可以自动化复杂任务并推动业务价值。
第一步：导入必要的库
首先导入项目所需的所有基本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函数：内部函数，用于执行具体的节能计算。假设值：代码中使用了一些假设值，如每度电的成本、每瓦太阳能板的成本等，这些值可以根据实际情况进行调整。 计算步骤：
首先计算用户的每月用电量（千瓦时）。 然后根据日照小时数计算所需的太阳能系统容量。 接着计算所需的太阳能板数量和安装成本。 最后计算年度节省金额和10年净节省金额。
返回结果：返回一个包含所需太阳能板数量、安装成本和10年净节省金额的字典。
该函数处理用户的每月电费成本，并返回太阳能板系统效益的详细估算，包括所需的太阳能板数量、安装成本以及十年内的净节省金额。为了简化计算，我们在其中做了一些假设，例如每度电的平均成本和平均日照小时数。然而，在更高级版本的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" # 指定此回退机制用于处理错误
)
代码说明：
handle_tool_error 函数：
该函数用于处理工具执行过程中发生的错误。 从状态中获取错误信息，并根据最后一次消息的工具调用生成相应的错误消息。 返回一个包含错误详情的 ToolMessage列表，每个消息与对应的工具调用ID关联。
create_tool_node_with_fallback 函数：
该函数用于创建一个带有错误回退处理的工具节点。 使用提供的工具列表创建 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 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'))
这种集成确保助手能够有效地解释并响应用户输入。
第六步：定义助手的工作流程
以上已经配置好了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方法，助手能够在对话中动态调用工具，实现更复杂的功能。工作流程整合：将提示模板与工具绑定后的LLM结合，形成完整的工作流程，确保助手能够智能地响应用户需求。
第七步：构建图结构
在这一步中，使用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 shutil # 未使用的库，已注释掉
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"表示只返回事件的值，而不是完整的事件对象。
_print_event ：自定义函数，用于打印事件内容，并确保不重复打印。
结论
通过上述步骤，已成功使用LangGraph创建了一个AI助手，该助手能够根据用户输入计算太阳能板的节能效果。本教程展示了LangGraph在管理复杂的多步骤流程中的强大能力，并强调了如何利用先进的AI工具高效解决现实世界的挑战。无论是为客户支持、能源管理还是其他应用开发AI智能体，LangGraph都提供了实现创意所需的灵活性、可扩展性和稳健性。
