在我的最近一篇博客文章中，使用LangChain代理构建多模态聊天机器人的指南，我探讨了AI代理的角色，并演示了如何使用LangChain框架实现。虽然它适合用于概念验证（POC），但在实际生产环境中表现却不尽如人意。在这篇文章中，我将提供一个更适合产品级应用的解决方案。您将学会如何构建一个可扩展且高效的系统，更适合实际应用，并获得构建更强大AI解决方案所需工具。
在生产环境中使用LangChain代理的挑战LangChain代理面临的主要挑战是，通过一个大型提示，给予单一的大型语言模型对整个工作流程的完全控制。在实际生产环境中，更细粒度的控制往往被需要，尤其是在任务随时间变化时。
所以我们如何才能最有效地将过程分解为更小、更专注的任务提示，每个提示负责任务的一个特定部分？我在本文中提到的方法让调试和微调流程中的每个单独部分变得更简单。此外，并没有任何一个大型语言模型（LLM）适用于所有任务。通常，将过程拆分成更小的任务，并为每个任务选择最合适的语言模型，可以同时提高效率和节省成本。
LangGraph 如何帮助你构建准备好的生产系统这就是LangGraph库派上用场的时候了。它提供了对代理状态和流程的精细控制，这对于构建稳健且可投入生产的系统至关重要。LangGraph扩展了LangChain的功能，通过启用带有循环图的有状态多代理应用，从而更容易构建出复杂和可靠的代理运行环境。
LangGraph的主要特点包括：
- 循环和分支。在应用程序中实现循环和条件判断。
在 LangGraph 中，每个节点代表一个 LLM 代理，边则是这些代理之间的通信渠道。这种结构使得工作流既清晰又易于管理，每个代理执行特定任务，并在必要时将信息传递给其他代理。 - 持久状态管理。自动保存每个步骤后的状态。在任何一点暂停和恢复图执行，以支持错误恢复、人机协作工作流、时间旅行等功能。
- 人机协作。中断图的执行，并让用户选择批准或编辑代理计划的下一个动作。
- 流式支持。支持每个节点生成的输出（包括令牌流）的流式传输。
- 与 LangChain 和 LangSmith 紧密集成：LangGraph 无缝集成到 LangChain 和 LangSmith（但没有它们也可以使用）。
我建立了一个应用程序，可以帮助你规划下一次的假期或商务旅行。输入相关信息，该应用程序会获取实时航班和酒店选项，并在用户友好的网页上显示这些信息。如果你想，也可以通过电子邮件将这些信息发送出去。
你可以在那里找到该GitHub中的完整代码库。
AI旅行助手，使用两种主要工具。
- 一个可以与谷歌航班交互的工具。
- 一个可以与谷歌酒店交互的工具。
此外，该应用程序还使用SendGrid API（发送电子邮件的服务）来发送邮件信息。
运行应用比如说，当你输入这样的提示，
我想从10月1日至10月7日从马德里到阿姆斯特丹旅行，帮我找找航班和四星酒店。
该应用程序会根据实时信息为你提供相关的航班和酒店选择。
你将收到一个包含标志和链接的输出结果，便于查阅。
请注意，这些结果来自Google Flights和Google Hotels API接口，我们无意推广任何特定品牌或产品。
也有一个选项可以通过电子邮件发送所有旅行信息。技术部分会解释这个实现方法（提示：涉及人类在环特性）。
哦，不错哦，我已经把所有旅行数据转换成了HTML并通过邮件发送出去了。
让我们来分解一下。AI助手会通过几个简单的步骤来处理用户的旅行请求。比如，用户可能输入了这样的提示：
“我想从10月1日到10月7日从马德里去阿姆斯特丹，帮我找一下航班和四星级，找到合适的四星级酒店。”
接下来会这样：
-
用户请求发送给AI（LLM）：请求被发送给一个支持工具调用的强大语言模型（LLM）。这个AI（LLM）识别出这两个任务：找航班和订酒店。
- 任务分解：大型语言模型将其分解为两个较小的任务：
- 查找航班: 系统使用专门设计的工具来查找航班。
- 查找酒店: 系统还使用了一个工具来查找阿姆斯特丹的四星酒店。
3. 工具启用：代理激活这些工具，并提供相应的数据（如日期、地点等）作为参数。每个工具运行后会输出结构化的航班和酒店选项。
4. 处理结果如下：再次让LLM处理并总结结果，以易于理解的格式呈现。
5. 通过电子邮件发送的选项：用户可以选择通过电子邮件发送这些信息。如果用户决定通过电子邮件发送，他们需要填写相关信息（比如收件人的邮箱地址）。
7. 邮件已发送，流程至此结束：根据之前收集的数据，代理从上次停止的地方继续。它调用发送邮件的功能，将相关信息发送到提供的邮箱地址。
一旦邮件一发送出去，代理程序就完成了它的使命，流程就完成了。
如何实施这种类型的代理？ 这个图怎么样？Mermaid编辑器的输入是用以下代码生成的：
打印图形的Mermaid表示（self.graph.get_graph().draw_mermaid()）要将该图可视化，将此代码生成的输出粘贴到 Mermaid 的在线编辑器中 (https://mermaid.live/edit)。（请注意，具体结果可能因你使用的 LangGraph（一个语言分析工具）版本不同而略有差异）。
以下代码示例定义了一个类，该类实现了一个代理(agent)，负责管理任务，并利用语言模型（LLM）调用工具。
TOOLS = [flights_finder, hotels_finder]
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
class Agent:
def __init__(self):
self._tools = {t.name: t for t in TOOLS}
self._tools_llm = ChatOpenAI(model='gpt-4o').bind_tools(TOOLS)
builder = StateGraph(AgentState)
builder.add_node('call_tools_llm', self.call_tools_llm)
builder.add_node('invoke_tools', self.invoke_tools)
builder.add_node('email_sender', self.email_sender)
builder.set_entry_point('call_tools_llm')
builder.add_conditional_edges('call_tools_llm', Agent.exists_action, {'more_tools': '更多工具', 'email_sender': '发送邮件'})
builder.add_edge('invoke_tools', 'call_tools_llm')
builder.add_edge('email_sender', END)
memory = MemorySaver()
self.graph = builder.compile(checkpointer=memory, interrupt_before=['email_sender'])
print(self.graph.get_graph().draw_mermaid())
@staticmethod
def exists_action(state: AgentState):
result = state['messages'][-1]
if len(result.tool_calls) == 0:
return '发送邮件'
return '更多工具'
def email_sender(self, state: AgentState):
print('发送邮件：')
email_llm = ChatOpenAI(model='gpt-4o', temperature=0.1) # 实例化另一个LLM
email_message = [SystemMessage(content=EMAILS_SYSTEM_PROMPT), HumanMessage(content=state['messages'][-1].content)]
email_response = email_llm.invoke(email_message)
print('邮件内容:', email_response.content)
message = Mail(from_email=os.environ['FROM_EMAIL'], to_emails=os.environ['TO_EMAIL'], subject=os.environ['EMAIL_SUBJECT'],
html_content=email_response.content)
try:
sg = SendGridAPIClient(os.environ.get('SENDGRID_API_KEY'))
response = sg.send(message)
print(response.status_code)
print(response.body)
print(response.headers)
except Exception as e:
print(str(e))
def call_tools_llm(self, state: AgentState):
messages = state['messages']
messages = [SystemMessage(content=TOOLS_SYSTEM_PROMPT)] + messages
message = self._tools_llm.invoke(messages)
return {'messages': [message]}
def invoke_tools(self, state: AgentState):
tool_calls = state['messages'][-1].tool_calls
results = []
for t in tool_calls:
print(f'调用: {t}')
if not t['name'] in self._tools: # 检查LLM提供的无效工具名称
print('\n ....无效工具名称....')
result = '无效工具名称，指示LLM重新尝试' # 如果无效，指示LLM重新尝试
else:
result = self._tools[t['name']].invoke(t['args'])
results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
print('回到模型处理！')
return {'messages': results}让我们来拆解一下所提供的代码中的关键点：
一. 工具配置：- 变量
TOOLS包含了工具列表，例如flights_finder和hotels_finder，这些是代理可以使用的这些工具。这些工具代表了代理可以执行的动作，来帮助用户。我在之前的帖子中详细介绍了如何实现这些工具。
- AgentState 是一种状态，它表示代理的当前状态，以
messages对象的形式保存。这个列表存储了代理与用户或工具之间交换的所有消息。 - 每条消息都按顺序添加，例如通过
operator.add。这意味着随着对话的进展，新消息会被添加到消息列表的末尾。
- 代理类（Agent类）通过调用工具、处理用户输入并管理任务流程来控制代理的操作。它设置了一个状态图（StateGraph），状态图本质上是这些任务（或节点）如何连接和执行的具体方案。
- 工具被组织成一个字典，可以通过名称来访问每个工具。
- 代理还将这些工具与语言模型（LLM），比如ChatOpenAI，关联起来，以便模型可以根据用户的输入来选择调用哪个工具。
- 代理创建一个
状态图，该图定义了代理将遵循的操作步骤。在 LangGraph 的术语中，这实际上是一个图，其中每个步骤都是一个节点（例如，调用工具或功能，或发送电子邮件）。 - 该状态图包含三个节点：
- call_tools_llm : 代理调用大型语言模型来决定根据任务使用哪种工具。
- invoke_tools : 代理启动选定的工具来执行特定任务，如查找航班或酒店。
- email_sender : 代理通过电子邮件将结果发送给用户邮箱。
- 开始节点是
call_tools_llm
- 该图具有条件边，意味着它根据代理当前的状态（也就是接收到的消息）来决定下一步的动作。例如，如果需要更多工具，它将进入调用工具节点；如果不再需要工具，则转向发送邮件的节点。
- 代理通过MemorySaver中的Checkpointer来保存其进度。这使代理能够中断后恢复其状态并继续操作。
这种方法检查状态中的最新消息来决定下一步的行动。如果没有更多工具调用的需要（即消息中没有剩余的工具调用），代理将转而发送邮件。否则，将继续调用更多的工具。在LangGraph术语中，这种方法就像一个决策函数，来决定代理下一步应该走哪条路（边）。
发送邮件的方法如下：这种方法处理最后一步，即代理发送带有结果的电子邮件。代理实例化另一个 LLM 来根据状态中最后一条消息起草邮件。它使用提示（即模板）来指导 LLM 生成邮件内容。然后通过 SendGridAPIClient 将邮件发送给用户。
调用工具LLM方法:在这个方法中，代理方发送一个消息列表给大型语言模型（LLM），其中包括一个系统消息（一个预定义的提示，告知大型语言模型要执行的任务）和先前的消息记录。大型语言模型随后返回一个新的消息，其中包含调用下个工具的指示。这个新消息将被添加到状态中。
调用工具方法:注：如果“invoke_tools”在整篇文档中作为一个技术术语被使用，并且需要保持一致性，可以保持英文形式或使用文档中约定的常用中文翻译。此处假设整个文档倾向于使用中文术语，除非另有说明。
此方法负责实际根据大语言模型的指令调用工具。代理检查状态中的最后一条消息以查找工具调用请求。它遍历这些工具调用请求，确保工具有效且可用，然后调用相应的工具并执行。工具的结果将以工具消息的形式返回，并追加到状态中。如果请求了无效的工具，代理将提示大语言模型再次尝试。
LangGraph的主要功能在我的应用程序中是怎样的？ 1. 持久状态的管理 # 这里创建了一个名为MemorySaver的对象，它用于管理内存。
memory = MemorySaver()
# 使用builder的compile方法来构建图，其中checkpointer参数设置为memory，表示在构建过程中使用MemorySaver来保存检查点。
# interrupt_before参数设置为['email_sender']，表示在email_sender之前中断。
self.graph = builder.compile(checkpointer=memory, interrupt_before=['email_sender'])代理设计的一个关键部分是它记住上次状态的能力。这由一个MemorySaver（记忆保存器）管理，确保当代理需要暂停和恢复时，能够从正确的点继续，而不会从头开始。例如，在等待用户决定发送一封邮件时。
在实际应用中，检查点工具可以使用多种选项，比如的选项有Postgres、MongoDB和Redis。
在 app.py（处理 UI 代码的部分）中，为每个会话生成一个唯一的 thread_id 以保持用户交互的上下文。每次调用代理时都会传递这个 thread_id，确保内存状态在不同请求间保持一致。
# 生成一个新的线程ID
thread_id = str(uuid.uuid4())
st.session_state.thread_id = thread_id
# 根据用户输入生成一条消息
messages = [HumanMessage(content=user_input)]
config = {'configurable': {'thread_id': thread_id}}
# 调用代理来运行
result = st.session_state.agent.graph.invoke({'messages': messages}, config=config)代理允许在关键决策点进行人工干预：决定是否发送一封包含收集到的旅行信息（航班、酒店等）的邮件。这通过“人在回路”功能实现，在发送邮件之前暂停代理执行，让用户检查结果并提供必要的输入。
在调用 graph.compile() 时添加 _interrupt_before=['emailsender'] 参数。
当用户决定发送电子邮件并提交数据后，以下代码将在UI中运行来完成流程。
def 发送邮件(发件人邮箱, 收件人邮箱, 主题, 线程ID):
try:
设置环境变量(发件人邮箱, 收件人邮箱, 主题)
配置信息 = {'可配置的': {'线程ID': 线程ID}}
st.session_state.代理组件.graph调用(None, config=配置信息)
st.success('邮件发送成功！')
# 清除会话中的状态信息
for key in ['travel_info', 'thread_id']:
if key in st.session_state:
del st.session_state[key]根据代理状态使用多个大语言模型
LangGraph的一个关键优势是其灵活的特性，可以在工作流的不同阶段使用多个大型语言模型（LLM），而不是依赖于一个大型语言模型来完成所有任务。这使得代理可以根据当前任务选择最适合的模型。例如，你可以用一个擅长调用工具和处理任务的模型，再用一个更适合生成和格式化HTML邮件内容的模型。
4. 与LangChain和LangSmith集成:将 LangChain 集成到您的应用程序中，可以利用其丰富的工具集来构建使用语言模型的复杂且模块化的工作流程。通过利用 LangChain 的功能，您可以轻松地将不同的大型语言模型、工具和工作流程集成到代理的功能中。
要将LangSmith集成到您的项目中，只需将这些环境变量添加到您的.env文件里。
LANGCHAIN_API_KEY=<langchain_api_key>
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=<langchain_project_name>通过将任务分解为更小且更易于管理的组件，LangGraph 允许更容易调试、大型语言模型的灵活集成，并且可以包含人类参与的交互。AI 旅行助手示例，帮助用户找到实时航班和酒店选项，展示了 LangGraph 处理复杂流程的能力，利用多个工具并提供邮件功能。它还突显了 LangGraph 的关键功能，如持久状态管理、多步骤流程以及与 LangChain 和 LangSmith 的无缝集成，使其成为一个强大的平台，用于构建稳健的 AI 驱动应用。