跳到主要内容 LangGraph 构建 Agent 逐步指南 | 极客日志
Python AI 算法
LangGraph 构建 Agent 逐步指南 本文介绍了如何使用 LangGraph 构建具备状态管理和循环计算能力的 AI 智能体。通过太阳能板节能计算示例,演示了从环境配置、工具定义、状态管理到图结构构建的完整流程。内容涵盖 LangChain 集成、AWS Bedrock 模型调用、错误处理机制及对话持久化策略,帮助开发者掌握复杂多步任务编排技术,实现可交互、自适应的企业级 AI 应用开发。重点讲解了 StateGraph 的定义、条件边逻辑、MemorySaver 的使用以及生产环境的部署优化建议。
宁静 发布于 2025/2/7 更新于 2026/4/20 2 浏览
引言 在人工智能领域,检索增强生成(RAG)系统已成为处理简单查询并生成上下文相关响应的常见工具。然而,随着对更复杂人工智能应用需求的增长,我们需要超越这些检索能力的系统。于是,AI 智能体(AI Agent)应运而生——这些自主实体能够执行复杂的多步骤任务,在交互过程中保持状态,并动态适应新信息。
LangGraph 作为 LangChain 库的强大扩展,旨在帮助开发者构建这些高级 AI 智能体,通过支持具有循环计算能力的有状态、多参与者应用来实现这一目标。
在本文中,我们将探讨 LangGraph 如何改变 AI 开发,并通过一个计算太阳能板节能效果的示例,逐步说明如何构建自己的 AI 智能体。此示例将展示 LangGraph 的独特功能如何创建智能、适应性强且适用于现实世界的 AI 系统。
什么是 LangGraph? LangGraph 是一个构建于 LangChain 之上的高级库,旨在通过引入循环计算能力来增强您的大型语言模型(LLM)应用。虽然 LangChain 允许创建用于线性工作流的有向无环图(DAG),但 LangGraph 更进一步,支持添加循环,这对于开发复杂的、类似智能体的行为至关重要。这些行为使得 LLM 能够持续循环执行某个过程,根据不断变化的条件动态决定下一步要采取的行动。
核心概念:状态图 LangGraph 的核心概念是状态图 (State Graph):
状态(State) :表示在计算过程中维护和更新的上下文或记忆。它确保图中的每一步都能访问之前步骤的相关信息,从而允许基于整个过程中积累的数据进行动态决策。
结点(Node) :作为图的基本构建块,代表单独的计算步骤或函数。每个结点执行特定任务,如处理输入、做出决策或与外部系统交互。结点可以自定义,以在工作流中执行各种操作。
边(Edge) :连接图中的结点,定义从一个步骤到下一个步骤的计算流程。它们支持条件逻辑,允许执行路径根据当前状态发生变化,并促进数据和控制在结点之间的流动,从而实现复杂的多步骤工作流。
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
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
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
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)
@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
return {
"messages" : [
ToolMessage(
content=f"Error: {repr (error)} \n please fix your mistakes." ,
tool_call_id=tc["id" ],
)
for tc in tool_calls
]
}
def create_tool_node_with_fallback (tools: list ) -> dict :
"""
创建一个带有错误回退处理的工具节点的函数。
参数:
tools (list): 包含在节点中的工具列表。
返回:
dict: 一个带有错误回退行为的工具节点。
"""
return ToolNode(tools).with_fallbacks(
[RunnableLambda(handle_tool_error)],
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 ):
self .runnable = runnable
def __call__ (self, state: State ):
while True :
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
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' ))
这种集成确保助手能够有效地解释并响应用户输入。在实际生产环境中,建议将密钥管理外包给 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 方法,助手能够在对话中动态调用工具,实现更复杂的功能。
工作流程整合 :将提示模板与工具绑定后的 LLM 结合,形成完整的工作流程,确保助手能够智能地响应用户需求。
第七步:构建图结构 在这一步中,使用 LangGraph 为 AI 助手构建图结构 ,该结构控制助手如何处理用户输入、触发工具以及在各个阶段之间切换。图结构定义了核心动作的结点 (如调用助手和工具)以及边 ,这些边决定了结点之间的流程。
在 LangGraph 中,每个结点 代表一个操作步骤,例如与用户交互或执行工具。为这个 AI 助手定义了两个关键结点:
助手结点 :管理对话流程,向用户询问电费成本并处理响应。
工具结点 :执行工具(例如 compute_savings)以计算用户的太阳能板节能效果。
builder = StateGraph(State)
builder.add_node("assistant" , Assistant(part_1_assistant_runnable))
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 来确保图结构在不同步骤之间保留对话状态。这使得助手能够记住用户的输入,从而在多步骤交互中保持连续性。
memory = MemorySaver()
graph = builder.compile (checkpointer=memory)
MemorySaver() :创建一个 MemorySaver 实例,用于保存和恢复对话状态。确保在多步骤交互中,助手能够记住用户的输入和上下文。
builder.compile(checkpointer=memory) :编译图结构,使其可以执行。传入 memory 作为检查点管理器,确保对话状态能够在不同步骤之间持久化。
第八步:运行助手
import uuid
tutorial_questions = [
'hey' ,
'can you calculate my energy saving' ,
"my montly cost is $100, what will i save"
]
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"
)
for event in events:
_print_event(event, _printed)
tutorial_questions :模拟用户与助手的对话内容,包含用户打招呼、请求计算节能效果以及提供每月电费成本。
uuid.uuid4() :生成一个唯一的对话 ID,用于标识当前对话会话。
config :配置参数,包含对话 ID,确保对话状态能够被正确保存和恢复。
_printed :用于跟踪已打印的事件,避免在事件流中重复打印相同内容。
graph.stream :使用 graph.stream 方法处理用户输入,并返回事件流。stream_mode="values" 表示只返回事件的值,而不是完整的事件对象。
第九步:部署与优化建议 在完成基础功能的构建后,为了确保 AI 智能体在生产环境中的稳定性和安全性,建议关注以下几点:
并发控制 :在高并发场景下,需考虑线程安全及资源限制。LangGraph 支持异步执行,可结合 asyncio 提高吞吐量。
状态存储 :MemorySaver 仅适用于内存测试。生产环境建议使用数据库(如 PostgreSQL)配合 PostgresSaver 进行持久化存储,确保会话数据不丢失。
输入验证 :对用户输入进行严格校验,防止 Prompt Injection 攻击。特别是在调用外部工具前,应验证参数的合法性和范围。
监控与日志 :集成日志系统(如 ELK Stack),记录关键决策点和工具调用结果,便于问题排查和性能分析。
成本控制 :LLM 调用通常按 Token 计费。可通过缓存常用回答、优化提示词长度等方式降低运营成本。
结论 通过上述步骤,已成功使用 LangGraph 创建了一个 AI 助手,该助手能够根据用户输入计算太阳能板的节能效果。本教程展示了 LangGraph 在管理复杂的多步骤流程中的强大能力,并强调了如何利用先进的 AI 工具高效解决现实世界的挑战。无论是为客户支持、能源管理还是其他应用开发 AI 智能体,LangGraph 都提供了实现创意所需的灵活性、可扩展性和稳健性。
掌握 LangGraph 的核心概念,如状态图、结点和边,能够帮助开发者构建具备记忆能力和循环逻辑的高级智能体。随着大模型技术的不断发展,LangGraph 将成为构建下一代企业级 AI 应用的重要基石。
相关免费在线工具 加密/解密文本 使用加密算法(如AES、TripleDES、Rabbit或RC4)加密和解密文本明文。 在线工具,加密/解密文本在线工具,online
RSA密钥对生成器 生成新的随机RSA私钥和公钥pem证书。 在线工具,RSA密钥对生成器在线工具,online
Mermaid 预览与可视化编辑 基于 Mermaid.js 实时预览流程图、时序图等图表,支持源码编辑与即时渲染。 在线工具,Mermaid 预览与可视化编辑在线工具,online
curl 转代码 解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具,curl 转代码在线工具,online
Base64 字符串编码/解码 将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具,Base64 字符串编码/解码在线工具,online
Base64 文件转换器 将字符串、文件或图像转换为其 Base64 表示形式。 在线工具,Base64 文件转换器在线工具,online