在之前，我讨论了AI Agent的作用，并展示了使用LangChain框架的实现。虽然它适合作为概念验证（POC），但在生产环境中却有所不足。在这篇文章中，我将提供一个更适合生产级产品的解决方案。你将学习如何创建一个可扩展、高效的系统，该系统更适合实际应用，为你提供构建更强大AI解决方案的工具。

LangChain Agent在生产环境中的挑战

LangChain Agent的主要挑战是它们通过一个大的提示将整个工作流程的控制权授予单个大型语言模型（LLM）。在生产环境中，通常需要更精细的控制，尤其是在任务不断发展时。

那么，我们如何最好地将过程分解为更小、更集中的提示，每个提示负责任务的特定部分呢？我在这篇文章中讨论的这种方法使得调试和微调流程中的每个单独组件变得更容易。此外，没有单个LLM在所有任务上都是完美的。通常最好将过程分解为更小的任务，并为每个任务分配最适合的LLM，以优化性能和成本。

LangGraph库如何帮助你构建生产就绪系统

这就是LangGraph库发挥作用的地方。它提供了对Agent流程和状态的细粒度控制，这对于构建健壮、生产就绪的系统至关重要。LangGraph通过启用具有循环图的状态化、多Actor应用程序来扩展LangChain，使得构建复杂、可靠的Agent运行时更加容易。

LangGraph的主要关键功能包括：

• 循环和分支。在你的应用程序中实现循环和条件。在LangGraph中，每个节点代表一个LLM Agent，边是这些Agent之间的通信通道。这种结构允许清晰且易于管理的工作流程，其中每个Agent执行特定任务，并根据需要将信息传递给其他Agent。
• 持久状态管理。在每个图步骤之后自动保存状态。在任何点暂停和恢复图的执行，以支持错误恢复、人工干预工作流程、时间旅行等。
• 人工干预。中断图的执行，并让用户有权批准或编辑Agent计划的下一个动作。
• 流式传输支持。流式传输每个节点产生的输出（包括令牌流式传输）。
• 与LangChain和LangSmith集成：LangGraph与LangChain和LangSmith无缝集成（但并不要求它们）。

我开发了什么？

我构建了一个应用程序，可以帮助你规划下一次度假或商务旅行。输入提示，该应用程序将获取实时航班和酒店选项，并在用户友好的网页上显示它们。如果需要，你还可以通过电子邮件发送这些信息。

AI旅行Agent使用两个主要工具：

• 一个与Google Flights API交互的工具。
• 一个与Google Hotels API交互的工具。

此外，该应用程序还使用SendGrid API发送电子邮件。

运行应用程序

例如，如果你输入以下提示：

“我想从马德里飞往阿姆斯特丹，时间是10月1日至7日，请为我查找航班和4星级酒店。”

该应用程序将根据实时数据提供相关的航班和酒店选项。

你将收到一个输出，其中包含标识和链接，便于参考。

还有一个选项是通过电子邮件发送所有旅行数据。我将在技术部分解释这是如何实现的（提示：这涉及到人工干预功能）。

哦，太好了，我收到了渲染成HTML格式的所有旅行数据的电子邮件：

代理是如何处理旅行请求的？

让我们将AI代理处理用户旅行请求的过程分解为几个简单的步骤。假设用户输入了以下提示：

“我想从马德里飞往阿姆斯特丹，时间是10月1日至7日，请为我查找航班和4星级酒店。”

接下来会发生以下情况：

1. 用户请求发送给AI（LLM）：请求被发送到一个支持工具使用的强大语言模型（LLM）。LLM识别出请求中的两个任务：查找航班和酒店。

2. 任务分解：LLM将请求分解为两个较小的任务：

• 查找航班：系统使用专门设计用于搜索航班的工具。
• 查找酒店：系统还调用一个工具，帮助在阿姆斯特丹查找4星级酒店。

3. 工具激活：代理调用这些工具，并提供正确的数据（如日期、位置等）作为参数。每个工具运行并以结构化格式输出航班和酒店选项。

4. 处理结果：再次调用LLM来总结结果并以易于阅读的格式呈现。

5. 电子邮件选项：然后向用户提供通过电子邮件发送此信息的选项。如果用户决定通过电子邮件发送，他们会在表单中输入详细信息（如收件人的电子邮件）。

6. 电子邮件发送，流程结束：使用之前收集的数据，代理从暂停的地方继续。它调用电子邮件发送功能，将信息发送到提供的电子邮件地址。

一旦电子邮件发送成功，代理即完成任务，整个流程结束。

实现细节：图表是如何呈现的？

使用以下代码生成Mermaid编辑器的输入：

print(self.graph.get_graph().draw_mermaid())

为了可视化图表，请将此代码的输出内容输入到在线的Mermaid编辑器中。（请注意，具体的可视化效果可能会根据你使用的LangGraph版本而略有不同）。

如何实现这种类型的代理

下面的代码片段定义了一个实现代理的类，该类负责管理任务并使用语言模型（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': 'invoke_tools', 'email_sender': '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 'email_sender'
        return 'more_tools'
    def email_sender(self, state: AgentState):
        print('Sending email')
        email_llm = ChatOpenAI(model='gpt-4o', temperature=0.1)  # Instantiate another LLM
        email_message = [SystemMessage(content=EMAILS_SYSTEM_PROMPT), HumanMessage(content=state['messages'][-1].content)]
        email_response = email_llm.invoke(email_message)
        print('Email content:', 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'Calling: {t}')
            if not t['name'] in self._tools:  # check for bad tool name from LLM
                print('\n ....bad tool name....')
                result = 'bad tool name, retry'  # instruct LLM to retry if bad
            else:
                result = self._tools[t['name']].invoke(t['args'])
            results.append(ToolMessage(tool_call_id=t['id'], name=t['name'], content=str(result)))
        print('Back to the model!')
        return {'messages': results}

实现代码的关键概念

让我们来分解所提供代码的关键概念：

工具设置：

• TOOLS 变量包含代理可以使用的工具列表，如 flights_finder 和 hotels_finder。这些工具代表代理可以执行以协助用户的动作。我在之前的帖子中详细阐述了如何实现这样的工具。

AgentState：

• AgentState 是一个状态，它以消息对象的形式保存代理的当前状态。这个列表存储了代理与用户或工具之间交换的所有通信（消息）。
• 每条消息都使用 operator.add 顺序添加，意味着随着对话的进行，新信息会被附加到现有消息之后。

Agent 类：

• Agent 类控制代理如何通过调用工具、处理用户输入和管理工作流程来操作。它设置了一个 StateGraph，这本质上是任务（或节点）如何连接和执行的蓝图。

初始化（init 方法）：

绑定工具：

• 工具被组织成一个字典，其中每个工具都可以通过名称访问。
• 代理还将这些工具绑定到一个语言模型（LLM），特别是 ChatOpenAI，以便模型可以根据用户输入决定调用哪个工具。

构建 StateGraph：

代理创建一个 StateGraph，它定义了代理将遵循的操作流程。在 LangGraph 术语中，这是一个图，其中代理过程中的每个步骤都由一个节点表示（例如，调用工具、激活它们或发送电子邮件）。

StateGraph 有三个节点：

• call_tools_llm：代理调用 LLM 以根据任务决定使用哪个工具。
• invoke_tools：代理激活所选工具以执行特定任务，如查找航班或酒店。
• email_sender：代理通过电子邮件向用户发送结果。

起始节点是 call_tools_llm。

条件边和转换：

• 图具有条件边，意味着它根据代理的当前状态（即消息）决定下一步该做什么。例如，如果需要更多工具，它会移动到 invoke_tools 节点。如果不再需要工具，它会转换到 email_sender 节点。

内存管理：

• 代理使用 Checkpointer（通过 MemorySaver）来保存其进度。这允许代理记住其状态，并在中断时恢复操作。

决策和动作方法：

exists_action 方法：

此方法检查状态中的最新消息，以确定下一个动作。如果不再需要调用工具（即消息中没有剩余的工具调用），代理会转换到发送电子邮件。否则，它会继续调用额外的工具。在 LangGraph 术语中，此方法作为决策函数，决定代理接下来应该走哪条路径（边）。

email_sender 方法：

此方法处理最后一步，其中代理发送包含结果的电子邮件。代理实例化另一个 LLM 来帮助根据状态中的最后一条消息起草电子邮件。它使用提示（模板）来指导 LLM 生成电子邮件内容。然后，使用 SendGridAPIClient 将电子邮件发送给用户。

call_tools_llm 方法：

在此方法中，代理向 LLM 发送一系列消息，其中包括系统消息（指示 LLM 该做什么的预定义提示）和之前的消息。LLM 然后返回一条新消息，其中包含关于接下来要调用哪个工具的指令。这条新消息被添加到状态中。

invoke_tools 方法：

此方法负责根据 LLM 的指令实际调用工具。代理检查状态中的最后一条消息以获取工具调用。它遍历这些工具请求，确保工具有效，然后调用相应的工具。工具的结果作为工具消息被附加到状态中。如果请求了无效的工具，代理会提示 LLM 重试。

LangGraph 的主要功能在我的应用程序中是如何实现的？

持久状态管理：

memory = MemorySaver()
self.graph = builder.compile(checkpointer=memory, interrupt_before=['email_sender'])compile(checkpointer=memory, interrupt_before=['email_sender'])

为代理设计的一个关键部分是它能够记住其上一个状态。这一功能由MemorySaver管理，它确保如果代理需要暂停和恢复（例如，在等待用户决定是否发送电子邮件时），它能从正确的位置继续，而不是重新开始。

在生产环境中，有多个选项可用于检查点存储，如Postgres、MongoDB和Redis。

在app.py（处理UI代码）中，为每个会话生成一个唯一的thread_id，以维持用户交互的上下文。这个thread_id随每次对代理的调用一起传递，确保内存状态在请求之间得到保持。

  # Create a new thread ID# Create a new thread ID
            thread_id = str(uuid.uuid4())
            st.session_state.thread_id = thread_id
            # Create a message from the user input
            messages = [HumanMessage(content=user_input)]
            config = {'configurable': {'thread_id': thread_id}}
            # Invoke the agent
            result = st.session_state.agent.graph.invoke({'messages': messages}, config=config)

人机协作

代理允许在关键决策点进行人为干预：决定是否发送包含收集到的旅行信息（航班、酒店等）的电子邮件。这是通过“人机协作”功能来管理的，该功能在电子邮件发送之前暂停代理的执行，允许用户审查结果并提供必要的输入。

通过在graph.compile()函数中传递interrupt_before=['email_sender']来实现。

一旦用户决定发送电子邮件并提交数据，UI中运行以下代码以完成该过程：

def send_email(sender_email, receiver_email, subject, thread_id):
    try:
        populate_envs(sender_email, receiver_email, subject)
        config = {'configurable': {'thread_id': thread_id}}
        st.session_state.agent.graph.invoke(None, config=config)
        st.success('Email sent successfully!')
        # Clear session state
        for key in ['travel_info', 'thread_id']:
            if key in st.session_state:
                del st.session_state[key]

基于代理状态使用多个大型语言模型（LLM）

LangGraph的一个关键优势是在工作流的不同阶段灵活使用多个LLM，而不是依赖一个带有大型提示的单一LLM。这允许代理根据当前任务选择最合适的LLM。例如，你可以使用一个专门用于工具调用和任务处理的LLM，而另一个则更适合生成和格式化HTML格式的电子邮件内容。

与LangChain和LangSmith的集成

将LangChain集成到你的应用程序中，可以访问其用于构建复杂、模块化工作流的全面工具套件，这些工作流涉及语言模型。通过利用LangChain的能力，你可以轻松地将不同的LLM、工具和工作流纳入代理的功能中。

要与LangSmith集成，只需将这些环境变量添加到你的.env文件中：

LANGCHAIN_API_KEY=<langchain_api_key>
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=<langchain_project_name>

总结

通过将任务分解为更小、更易管理的组件，LangGraph使得调试更加容易，工具和LLM（大型语言模型）集成更加灵活，并包含了人机协作交互。AI旅行代理示例，它帮助用户查找实时的航班和酒店选项，展示了LangGraph处理复杂工作流的能力，利用多个工具并提供电子邮件功能。它还突出了LangGraph的关键特性，如持久状态管理、多步骤工作流以及与LangChain和LangSmith的无缝集成，使其成为构建强大AI驱动应用程序的强大框架。