构建动态多智能体工作流：利用LangChain和LangGraph进行AI协作

本文利用LangChain和LangGraph创建了一个简单的多智能体系统。这些智能体协同工作以完成任务。第一个智能体生成一串随机数，而第二个智能体将这些数乘以10。每个智能体使用OpenAI的GPT-4o API来执行这些任务。本文采用基于工作流的架构，智能体根据分配的任务进行交互。在这篇文章中，我们将分解每个脚本部分及其对整体流程的贡献。前置条件在深入代码之前，确保已安装以下内容：Python

程序员维他命

874人浏览 · 2024-12-09 11:33:09

程序员维他命 · 2024-12-09 11:33:09 发布

本文利用LangChain和LangGraph创建了一个简单的多智能体系统。这些智能体协同工作以完成任务。第一个智能体生成一串随机数，而第二个智能体将这些数乘以10。每个智能体使用OpenAI的GPT-4o API来执行这些任务。

本文采用基于工作流的架构，智能体根据分配的任务进行交互。在这篇文章中，我们将分解每个脚本部分及其对整体流程的贡献。

前置条件
在深入代码之前，确保已安装以下内容：

Python 3.7 或更高版本
OpenAI API 访问权限（你需要一个 API 密钥）
安装了 LangChain 和 LangGraph 库。你可以通过 pip 进行安装：

pip install langchain langgraph

设置环境
在脚本中，您必须将您的OpenAI API密钥设置为环境变量。这确保了智能体可以与GPT-4模型进行交互。您可以在终端中设置API密钥：

import osos.environ["OPENAI_API_KEY"] = "your_openai_api_key"

创建AI智能体
create_agent函数负责使用LangChain的ChatPromptTemplate来设置一个智能体。每个智能体都通过一个系统消息进行初始化，该消息指定了它将执行的任务。具体工作原理如下：

def create_agent(llm, system_message: str):    """Create an agent."""    prompt = ChatPromptTemplate.from_messages(        [            (                "system",                "You are a helpful AI assistant, collaborating with other assistants. "                "Work on the assigned task and provide useful outputs. "                "Prefix your response with FINAL ANSWER if you have completed your task."                " Here is the task: {system_message}",            ),            MessagesPlaceholder(variable_name="messages"),        ]    )    prompt = prompt.partial(system_message=system_message)    return prompt | llm

系统消息解释了智能体的角色及其应如何表现。例如，一个智能体被指示生成随机数，而另一个智能体被要求将这些数相乘。

智能体状态
为了跟踪智能体之间交换的消息，脚本使用TypedDict定义了智能体状态的结构。这有助于管理消息并识别最后一个消息的发送者：

class AgentState(TypedDict):    messages: Sequence[BaseMessage]    sender: str

每个智能体发送和接收消息，并且状态跟踪当前负责下一步动作的智能体。

定义工作流
工作流使用LangGraph的StateGraph实现。在这里，智能体作为工作流中的节点被添加，并且它们之间的转换基于路由器逻辑定义。

路由器函数有助于控制智能体之间消息的流动：

def router(state):    messages = state["messages"]    last_message = messages[-1]    if "FINAL ANSWER" in last_message.content:        if state["sender"] == "Agent_1":            return "Agent_2"        return END    return "continue"

工作流定义了智能体之间的交互方式以及控制权从一个智能体转移到另一个智能体的条件。

向工作流中添加智能体
可以使用workflow.add_node方法将智能体作为节点添加到工作流中。例如，Agent_1 负责生成随机数：

workflow.add_node("Agent_1", agent_1_node)workflow.add_node("Agent_2", agent_2_node)

条件边根据路由器逻辑添加，以将流程从一个智能体转移到另一个智能体。

主执行
脚本的主要部分负责初始化工作流，并根据用户的初始输入执行它。输入消息指示系统生成随机数并将它们乘以10：

if __name__ == "__main__":    initial_state = {        "messages": [            HumanMessage(content="Generate 10 random numbers and multiply each by 10.")        ],        "sender": "Agent_1",    }    events = graph.stream(initial_state, {"recursion_limit": 150})    for event in events:        print(event)        print("----")

在这里，工作流通过初始消息开始执行，并且系统将事件流经每个智能体。递归限制确保工作流不会无限运行。