粤ICP备17114055号
微信扫码
添加专属顾问
支持私有云部署
我要投稿
LangGraph：AI智能体开发新突破，实现复杂任务自动化 核心内容： 1. LangGraph与AI智能体的结合，超越传统RAG系统 2. LangGraph核心概念：状态图、结点、状态和边 3. 通过太阳能板节能效果示例，逐步构建AI智能体
在人工智能领域，检索增强生成（RAG）系统已成为处理简单查询并生成上下文相关响应的常见工具。然而，随着对更复杂人工智能应用需求的增长，我们需要超越这些检索能力的系统。于是，AI 智能体（AI Agent）应运而生——这些自主实体能够执行复杂的多步骤任务，在交互过程中保持状态，并动态适应新信息。LangGraph 作为 LangChain 库的强大扩展，旨在帮助开发者构建这些高级 AI 智能体，通过支持具有循环计算能力的有状态、多参与者应用来实现这一目标。
在本文中，我们将探讨LangGraph如何改变AI开发，并通过一个计算太阳能板节能效果的示例，逐步说明如何构建自己的AI智能体。此示例将展示LangGraph的独特功能如何创建智能、适应性强且适用于现实世界的AI系统。
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 函数：内部函数，用于执行具体的节能计算。该函数处理用户的每月电费成本，并返回太阳能板系统效益的详细估算，包括所需的太阳能板数量、安装成本以及十年内的净节省金额。为了简化计算，我们在其中做了一些假设，例如每度电的平均成本和平均日照小时数。然而，在更高级版本的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 类：
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'))
这种集成确保助手能够有效地解释并响应用户输入。
以上已经配置好了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) ：
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 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都提供了实现创意所需的灵活性、可扩展性和稳健性。
53AI，企业落地大模型首选服务商
产品：场景落地咨询+大模型应用平台+行业解决方案
承诺：免费场景POC验证，效果验证后签署服务协议。零风险落地应用大模型，已交付160+中大型企业
2025-03-05
Agent 最全 Playbook：场景、记忆和交互创新
2025-03-05
基于LangGraph的智能文章生成Agent架构设计思路
2025-03-02
LangGraph全新4大预构建Agents框架登场
2025-03-01
极简LangChain智能体开发入门指南
2025-03-01
释放 Langflow 的力量：使用 Rag 构建多工具代理
2025-02-28
刚刚，LangGraph官方开源Ultra版本，多智能体能力提升10倍！
2025-02-25
AI Agent 开发框架巡礼：流行度、案例与理性选择
2025-02-22
手把手教你使用Qwen-Agent开发智能体应用实战教程
2024-10-10
2024-04-08
2024-07-13
2024-06-03
2024-09-04
2024-08-18
2024-04-08
2024-06-24
2024-07-10
2024-03-28
2025-02-05
2024-12-02
2024-11-25
2024-10-30
2024-10-11
2024-08-18
2024-08-16
2024-08-04
回到顶部