学习如何使用 LangGraph 设计结合了推理、矢量搜索和工具集成的可扩缩自主 AI 智能体。
概括
- AI 智能体是一种超越简单的聊天机器人的自主系统，它结合了记忆库、上下文，并具有自动完成任务的主动性。
- LangGraph 是一种框架，它使我们能够利用具有内置状态管理的节点（任务）和边缘（连接），构建复杂的 AI 工作流。
- 本指南将指导您构建 AI 赋能的客户支持智能体，该智能体可以划分优先级，识别相关主题，并确定是上报还是自动回复。
那么，AI 智能体究竟是什么？
让我们直面它吧 —“AI 智能体”听起来就像可以接管会议室的机器人。 实际上，它们是您得力的助手，可以简化复杂的工作流，消除重复性任务。 您可以把它们看作是聊天机器人的下一个进化阶段：它们不只是简单地等待提示；它们可以发起行动，协调多个步骤，并随时进行调整。
过去，打造一个“智能”系统意味着兼顾语言理解、代码生成、数据查找等各种不同的模型，然后将它们粘合在一起。 您的一半时间花在了集成上，另一半时间则花在了调试上。
智能体彻底颠覆了这一切。 它们将上下文、主动性和适应性融合在一个精心编排的流程中。 它们不仅实现了自动化，更是肩负使命的智者。 借助 LangGraph 之类的框架，我相信，组建一支自己的智能体团队实际上会很有趣。
LangGraph 究竟是什么？
LangGraph 是一种创新型框架，它彻底改变了我们构建涉及大语言模型 (LLM) 的复杂应用程序的方式。
想象一下，您正在指挥一支管弦乐队：每种乐器（或“节点”）都需要知道何时演奏，声音有多大，顺序如何。 这种情况下，LangGraph 就是您的指挥棒，为您提供以下内容：
- 图结构：它采用具有节点和边缘的图结构，使开发者能够设计适应各种分支和循环的灵活非线性工作流。 它可以反映复杂的决策过程，类似于神经通路的运作方式。
- 状态管理：LangGraph 提供了各种内置工具，可以实现状态保持和错误恢复，简化了应用程序中各个阶段的上下文数据的维护。 借助 Zep 等工具，它可以在短期记忆和长期记忆之间高效切换，提高了交互质量。
- 工具集成：借助 LangGraph，LLM 智能体可以轻松与外部服务或数据库协作，获取真实的数据，从而改进应用程序的功能和响应性。
- 人机协同：除了可以实现自动化外，LangGraph 还可以适应工作流中的人为干预，这对于需要分析监督或伦理考虑的决策过程至关重要。
无论您是在构建具有真实记忆的聊天机器人、交互式故事引擎，还是能够处理复杂问题的智能体团队，LangGraph 都可以将令人头疼的管道工程转变成简单明了、直观的状态机。
开始
要开始使用 LangGraph，您需要进行基本的设置，通常包括安装 langgraph 和 langchain-openai 等必要的库。 然后，您可以定义图中的节点（任务）和边缘（连接），有效地实现短期记忆的检查点，并利用 Zep 满足更持久的记忆需求。
操作 LangGraph 时，请记住以下几点：
- 具有灵活性的设计：利用强大的图结构来解释并非严格线性的潜在工作流分支和交互。
- 以深思熟虑的方式与工具交互：利用外部工具增强 LLM 功能，而不是取而代之。 为每个工具提供全面的描述，以实现精确使用。
- 采用丰富的记忆解决方案：高效地使用记忆库，留意 LLM 的上下文窗口，并考虑集成外部解决方案，以实现自动事实管理。
现在，我们已经介绍 LangGraph 的基础知识，我们来看一个实例。 为此，我们将开发一个用于提供客户支持的 AI 智能体。
这个智能体将接收电子邮件请求，分析电子邮件正文中的问题描述，然后确定请求的优先级和适当的主题/类别/部门。
系好安全带，我们开始吧！
首先，我们需要定义什么是“工具”。 可以把它看作是智能体的专属“助理”，使其能够与外部功能进行交互。
@tool 装饰器在这里必不可少。 LangChain 简化了自定义工具的创建，这意味着首先，定义一个 Python 函数，然后应用 @tool 装饰器。
为了进行说明，我们来创建第一个工具。 这个工具将帮助智能体根据电子邮件内容划分 IT 支持工单的优先级：
from langchain_core.tools import tool
@tool
def classify_priority(email_body: str) -> str:
"""Classify the priority of an IT support ticket based on email content."""
prompt = ChatPromptTemplate.from_template(
"""Analyze this IT support email and classify its priority as High, Medium, or Low.
High: System outages, security breaches, critical business functions down
Medium: Non-critical issues affecting productivity, software problems
Low: General questions, requests, minor issues
Email: {email}
Respond with only: High, Medium, or Low"""
)
chain = prompt | llm
response = chain.invoke({"email": email_body})
return response.content.strip()
太棒了！ 现在，我们有一个提示，指示 AI 接收电子邮件正文，对其进行分析，并将其优先级分为“高”、“中”或“低”。
就是这样！ 您刚刚编写了一个智能体可以调用的工具！
接下来，我们创建一个类似的工具来识别支持请求的主要主题（或类别）：
@tool
def identify_topic(email_body: str) -> str:
"""Identify the main topic/category of the IT support request."""
prompt = ChatPromptTemplate.from_template(
"""Analyze this IT support email and identify the main topic category.
Categories: password_reset, vpn, software_request, hardware, email, network, printer, other
Email: {email}
Respond with only the category name (lowercase with underscores)."""
)
chain = prompt | llm
response = chain.invoke({"email": email_body})
return response.content.strip()
现在，我们需要创建一个状态，在 LangGraph 中，这一小部分非常重要。
把它想象成图的中枢神经系统。 这就是节点之间的通信方式，就像优等生在课堂上传递纸条一样。
文档中显示：
“状态是表示应用程序当前快照的共享数据结构。”
在实践中呢？ 状态是在节点之间移动的结构化消息。 它将一个步骤的输出作为下一个步骤的输入。 基本上，它是将整个工作流粘合在一起的粘合剂。
因此，在构建图之前，我们必须先定义我们的状态结构。 本例中，我们的状态包括以下内容：
- 用户请求（电子邮件正文）
- 指定的优先级
- 确定的主题（类别）
它简单明了，因此您可以像专业人士一样浏览图。
from typing import TypedDict
# Define the state structure
class TicketState(TypedDict):
email_body: str
priority: str
topic: str
# Initialize state
initial_state = TicketState(
email_body=email_body,
priority="",
topic=""
)
节点与 边缘：LangGraph 的关键组成部分
LangGraph 的基本要素包括节点和边缘。
- 节点：它们是图中的操作单元，执行实际工作。 节点通常由可以执行任何逻辑（从计算到与语言模型 (LLM) 或外部集成交互）的 Python 代码组成。 从本质上讲，节点就像传统编程中的个别函数或智能体。
- 边缘：边缘定义节点之间的执行流，决定接下来会发生什么。 它们充当连接器，允许状态根据预定义条件从一个节点转换到另一个节点。 在 LangGraph 中，边缘在协调节点之间的序列和决策流方面至关重要。
为了掌握边缘的功能，我们来看一个消息传递应用程序的简单类比：
- 节点类似于积极参与对话的用户（或他们的设备）。
- 边缘代表着用户之间能够促进沟通的聊天主题或连接。
当用户选择一个聊天主题来发送消息时，会有效地创建一个边缘，将他们与另一个用户连接在一起。 与 LangGraph 状态的结构化模式类似，每次交互（无论是发送文本、语音还是视频消息）都遵循预定义的顺序。 它确保了沿边缘传递的数据的一致性和可解释性。
不同于事件驱动型应用程序的动态特性，LangGraph 采用在整个执行过程中保持一致的静态模式。 它简化了节点之间的通信，使开发者可以依赖稳定的状态格式，从而确保无缝的边缘通信。
设计基本工作流
可以将 LangGraph 中的流工程理解为设计一个状态机。 在这个情境中，每个节点代表一个不同的状态或处理步骤，而边缘定义了这些状态之间的转换。 这种方式对想要在 AI 的确定性任务序列与动态决策能力之间取得平衡的开发者特别有用。 我们来使用前面定义的 TicketState 类初始化 StateGraph，开始构建流程。
from langgraph.graph import StateGraph, START, END
workflow = StateGraph(TicketState)
节点添加：节点是基本要素，用于执行划分工单优先级或识别其主题等特定任务。
每个节点函数均接收当前状态，执行其操作，并返回一个字典以更新状态：
def classify_priority_node(state: TicketState) -> TicketState:
"""Node to classify ticket priority."""
priority = classify_priority.invoke({"email_body": state["email_body"]})
return {"priority": priority}
def identify_topic_node(state: TicketState) -> TicketState:
"""Node to identify ticket topic."""
topic = identify_topic.invoke({"email_body": state["email_body"]})
return {"topic": topic}
workflow.add_node("classify_priority", classify_priority_node)
workflow.add_node("identify_topic", identify_topic_node)
classify_priority_node 和 identify_topic_node 方法将更改 TicketState 并发送参数输入。
边缘创建：定义连接节点的边缘：
workflow.add_edge(START, "classify_priority")
workflow.add_edge("classify_priority", "identify_topic")
workflow.add_edge("identify_topic", END)
classify_priority 确定起点，而 identify_topic 确定到目前为止工作流的终点。
编译与执行：配置节点和边缘后，编译并执行该工作流。
graph = workflow.compile()
result = graph.invoke(initial_state)
太好了！ 您还可以生成 LangGraph 流的可视化表示。
graph.get_graph().draw_mermaid_png(output_file_path="graph.png")
如果将代码运行到此点，您就会看到一个与下面类似的图：
这幅图直观地显示了一次顺序执行：开始，然后划分优先级，接着确定主题，最后结束。
LangGraph 最强大的一个方面是它的灵活性，这让我们可以创建更复杂的流程和应用程序。 例如，我们可以修改工作流，使用以下行将 START 中的边缘添加到两个节点：
workflow.add_edge(START, "classify_priority")
workflow.add_edge(START, "identify_topic")
这一更改意味着智能体将同时执行 classify_priority 和 identify_topic。
LangGraph 中另一个非常有用的功能是能够使用条件边缘。 它们允许工作流根据对当前状态的评估进行分支，实现任务的动态路由。
我们来增强工作流。 我们将创建一个新工具，分析请求的内容、优先级和主题，以确定它是否为需要上报（例如，为人工团队提交工单）的高优先级问题。 如果不需要，将为用户生成一个自动响应。
@tool
def make_escalation_decision(email_body: str, priority: str, topic: str) -> str:
"""Decide whether to auto-respond or escalate to IT team."""
prompt = ChatPromptTemplate.from_template(
"""Based on this IT support ticket, decide whether to:
- "auto_respond": Send an automated response for simple/common or medium priority issues
- "escalate": Escalate to the IT team for complex/urgent issues
Email: {email}
Priority: {priority}
Topic: {topic}
Consider: High priority items usually require escalation, while complex technical issues necessitate human review.
Respond with only: auto_respond or escalate"""
)
chain = prompt | llm
response = chain.invoke({
"email": email_body,
"priority": priority,
"topic": topic
})
return response.content.strip()
此外，如果请求被确定为低优先级或中等优先级（导致“auto_respond”决策），我们将执行矢量搜索来检索历史回答。 然后，将使用此信息来生成适当的自动回答。 不过，这需要两个额外的工具：
@tool
def retrieve_examples(email_body: str) -> str:
"""Retrieve relevant examples from past responses based on email_body."""
try:
examples = iris.cls(__name__).Retrieve(email_body)
return examples if examples else "No relevant examples found."
except:
return "No relevant examples found."
@tool
def generate_reply(email_body: str, topic: str, examples: str) -> str:
"""Generate a suggested reply based on the email, topic, and RAG examples."""
prompt = ChatPromptTemplate.from_template(
"""Generate a professional IT support response based on:
Original Email: {email}
Topic Category: {topic}
Example Response: {examples}
Create a helpful, professional response that addresses the user's concern.
Keep it concise and actionable."""
)
chain = prompt | llm
response = chain.invoke({
"email": email_body,
"topic": topic,
"examples": examples
})
return response.content.strip()
现在，我们为这些新工具定义相应的节点：
def decision_node(state: TicketState) -> TicketState:
"""Node to decide on escalation or auto-response."""
decision = make_escalation_decision.invoke({
"email_body": state["email_body"],
"priority": state["priority"],
"topic": state["topic"]
})
return {"decision": decision}
def rag_node(state: TicketState) -> TicketState:
"""Node to retrieve relevant examples using RAG."""
examples = retrieve_examples.invoke({"email_body": state["email_body"]})
return {"rag_examples": examples}
def generate_reply_node(state: TicketState) -> TicketState:
"""Node to generate suggested reply."""
reply = generate_reply.invoke({
"email_body": state["email_body"],
"topic": state["topic"],
"examples": state["rag_examples"]
})
return {"suggested_reply": reply}
def execute_action_node(state: TicketState) -> TicketState:
"""Node to execute final action based on decision."""
if state["decision"] == "escalate":
action = f"🚨 ESCALATED TO IT TEAM\nPriority: {state['priority']}\nTopic: {state['topic']}\nTicket created in system."
print(f"[SYSTEM] Escalating ticket to IT team - Priority: {state['priority']}, Topic: {state['topic']}")
else:
action = f"✅ AUTO-RESPONSE SENT\nReply: {state['suggested_reply']}\nTicket logged for tracking."
print(f"[SYSTEM] Auto-response sent to user - Topic: {state['topic']}")
return {"final_action": action}
workflow.add_node("make_decision", decision_node)
workflow.add_node("rag", rag_node)
workflow.add_node("generate_reply", generate_reply_node)
workflow.add_node("execute_action", execute_action_node)
然后，条件边缘将使用 make_decision 节点的输出来定向流：
workflow.add_conditional_edges(
"make_decision",
lambda x: x.get("decision"),
{
"auto_respond": "rag",
"escalate": "execute_action"
}
)
如果 make_escalation_decision 工具（通过 decision_node）产生“auto_respond”，工作流将继续通过 RAG 节点（检索示例），然后是 generate_reply 节点（设计回答），最后是 execute_action 节点（记录 auto-response）。
相反，如果决策是“escalate”，流程将绕过 RAG 和生成步骤，直接转到 execute_action 来处理上报。 要添加剩余的标准边缘来完成图，请执行以下操作：
workflow.add_edge("rag", "generate_reply")
workflow.add_edge("generate_reply", "execute_action")
workflow.add_edge("execute_action", END)
数据集注释：对于此项目，我们用于支持检索增强生成 (RAG) 的数据集来自 Hugging Face 上的 Customer Support Tickets 数据集。 对该数据集进行了筛选，以便只包含分类为技术支持并限制为英语的条目。 它确保 RAG 系统只为技术支持任务检索高度相关的特定领域的示例。
此时，我们的图应当与下图类似：
当您使用一封电子邮件来执行此图，并导致高优先级分类和“escalate”决策时，您将看到以下响应：
.png)
同时，被分类为低优先级并导致“auto_respond”决策的请求将触发与下面类似的回复：
那么… 这一路都很顺利吗？
并不完全是。 有一些问题需要注意：
- 数据隐私：小心敏感信息 — 这些智能体需要防护。
- 计算成本：某些高级设置需要大量资源。
- 幻觉：LLM 偶尔也会编造一些内容（不过仍然比大多数实习生聪明）。
- 非确定性：相同的输入可能会返回不同的输出，这对创造力来说是好事，但对严格的流程来说却很棘手。
不过，大多数缺点都可以通过良好的规划、合适的工具和一点点思考加以控制。
LangGraph 将 AI 智能体从流行语变成实实在在的有效解决方案。 无论您是要自动执行客户支持，处理 IT 工单，还是构建自主应用程序，这个框架都让操作变得可行，甚至有趣。
您有任何问题或反馈吗？ 请说出来。 AI 革命需要像您这样的建设者。
.png)
那我们想要定义自己的参数，该如何定义呢？根据面向对象设计原则之一：
.png)
让我们将字标记函数 
同样，可以为输出补充其他功能，例如，活动链接或工具提示。
将加载 CSP 应用程序以编辑存储类的实例，在此示例中： 
此示例还显示了编辑类实例 
