使用 LangGraph 构建代理应用程序: 第2部分 — 多代理协作

在Langgraph系列的第一部分中，我们使用Langgraph构建了一个表格OCR代理。它由一个单一的LLM代理组成，该代理可以访问OCR工具（tesseract OCR），并且该代理负责准确地读取表格图像中的文本并回答用户的问题。

多智能体协作

多智能体协作涉及多个人工智能代理共同工作，每个人工智能代理专注于任务的特定方面。这反映了人类团队合作，个体提供自己的专业知识来完成共同目标。例如，在软件开发中，各种代理可能负责编码、测试、文档和项目管理，合作交付完整的软件解决方案。

这种方法将复杂的过程（如软件创建）拆分为由专门角色完成的较小子任务，例如软件工程师、产品经理、设计师和质量保证工程师。代理可以通过指导一个或多个大型语言模型（LLMs）执行特定操作来开发。多代理方法提供了将复杂任务分解成可管理部分的结构，就像在商业环境中将大型项目拆分一样。每个代理都遵循自己的工作流程，维护自己的记忆，并可以从其他代理获取帮助，从而导致多个LLM之间的相互作用和消息交换。

范围

在本文中，我们将通过使用LangGraph构建一个博客写作代理来实现多智能体协作的概念。涉及到两个代理：一个研究员和一个写手。

研究员：研究员负责从互联网上搜索与某一主题相关的信息，基本上就是对提供的主题进行研究。

作者：这个代理商将获得研究员收集的信息，其任务是起草一个结构良好、简洁明了的博客。

让我们一起烹饪吧！

首先，让我们安装所需的Python软件包。

%pip install --quiet -U langgraph langchain \
langchain_openai langchainhub langchain-community

现在，我们正在初始化用于我们的LLM（OpenAI）和Tavily的API密钥，这些将用于设置搜索工具。通过在这里注册来获取您的Tavily API密钥。

import getpass
import os

os.environ["OPENAI_API_KEY"] = getpass.getpass("OpenAI API Key:")
os.environ["TAVILY_API_KEY"] = getpass.getpass("Tavily Key:")

工具定义

现在我们初始化搜索工具，它将用于运行网络搜索。我们将使用langchain的tavily搜索工具。

from typing import Annotated

from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool

tavily_tool = TavilySearchResults(max_results=5)
tools=[tavily_tool]

代理设置

现在，我们初始化我们的代理人，即研究员和作家。研究员将有权限使用Tavily搜索工具，并被要求收集有关主题的信息，然后将其传递给作家。作家将使用研究员收集的信息并撰写博客。GPT-4o是我们将为这两位代理人使用的LLM。

from langchain_core.messages import (
    BaseMessage,
    HumanMessage,
    ToolMessage,
)
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langgraph.graph import END, StateGraph, START
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

researcher_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You're are a helpful AI assistant tasked with doing research on the topic provided by the user."
            "Your goal is to gather information on the mentioned topic by using the tools provided and then summarizing the information concisely."
            "You have access to the following tools: {tool_names}."
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)
researcher_prompt = researcher_prompt.partial(tool_names=", ".join([tool.name for tool in tools]))
researcher_agent = researcher_prompt | llm.bind_tools(tools)

writer_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
           "You're are a helpful AI assistant tasked with writing meaningful blog articles."
           "Your content should be guided by the outline and context provided by the Content Researcher."
           "While following the main objectives and direction set in the outline, you should also offer balanced, objective insights supported by the information from the Content Researcher." 
           "Be sure to clearly distinguish between your opinions and factual statements throughout the piece."
        ),
        MessagesPlaceholder(variable_name="messages"),
    ]
)
writer_agent = writer_prompt | llm

定义图状态

我们现在定义图表的状态。这将只是一个消息列表，以及一个键来跟踪最近的发送者。

import operator
from typing import Annotated, Sequence, TypedDict

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    sender: str

定义代理节点

现在我们需要定义节点。我们将创建一个辅助函数，它将部分初始化代理实例，并用作图中的节点。

import functools

from langchain_core.messages import AIMessage


# Helper function to create a node for a given agent
def agent_node(state, agent, name):
    result = agent.invoke(state)
    if isinstance(result, ToolMessage):
        pass
    else:
        result = AIMessage(**result.dict(exclude={"type", "name"}), name=name)
    return {
        "messages": [result],
        "sender": name,
    }

research_node = functools.partial(agent_node, agent=researcher_agent, name="Researcher")
writer_node = functools.partial(agent_node, agent=writer_agent, name="writer_agent")

定义边缘逻辑

我们可以定义一些边缘逻辑，根据代理的结果决定要做什么。

from typing import Literal


def router(state) -> Literal["call_tool", "__end__", "continue"]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        # The previous agent is invoking a tool
        return "call_tool"
    return "continue"

定义图表

现在我们有了定义图形的所有组件。下面是我们如何组装最终的图形：

添加代理和工具节点
研究者的条件边缘: 它可以调用工具或将收集到的信息传递给作者。
为工具节点添加一个条件边缘，它只能将输出返回到研究节点。
在作家和END之间添加一个边。
将入口点添加到研究者节点

workflow = StateGraph(AgentState)

workflow.add_node("Researcher", research_node)
workflow.add_node("writer_agent", writer_node)
workflow.add_node("call_tool", tool_node)

workflow.add_conditional_edges(
    "Researcher",
    router,
    {"continue": "writer_agent", "call_tool": "call_tool",},
)
workflow.add_conditional_edges(
    "call_tool",
    lambda x: x["sender"],
    {
        "Researcher": "Researcher"
    },
)
workflow.add_edge(
    "writer_agent",
    END
)

workflow.add_edge(START, "Researcher")
graph = workflow.compile()

让我们显示图表，看看它是什么样子

from IPython.display import Image, display

try:
    display(Image(graph.get_graph(xray=True).draw_mermaid_png()))
except Exception:
    pass

调用

有了创建好的图表，让我们调用它并写一些博客吧！

events = graph.stream(
    {
        "messages": [
            HumanMessage(
                content="Write a blog on the topic 'Getting started with options trading'"
            )
        ],
    },
    {"recursion_limit": 150},
)
for s in events:
    print(s)
    print("----")