全面比较：选择最佳ai代理框架langgraph Crewai与openai Swarm的实用指南

使用3种不同框架构建的相同代理金融应用的代理编排深入比较。

由聊天OpenAI生成的图像。提示：多代理协作系统。

我们将涵盖的内容

• 什么是代理？ 深入了解我们如何定义代理以及它们与AI管道和独立大语言模型的区别。
• 使用3个流行的代理框架构建的实践示例： LangGraph、CrewAI和OpenAI Swarm (完整代码)。
• 我们关于何时使用哪个框架的建议。
• 接下来可以期待什么： 第二部分的预览，我们将探讨这些框架中的可调试性和可观察性。

介绍

由大语言模型驱动的自主代理经历了起伏。从2023年AutoGPT和BabyAGI的病毒式演示到今天更为精细的框架，能够自主执行端到端任务的代理-大语言模型概念引发了人们的想象与怀疑。

为什么会重新引起兴趣？在过去的9个月中，大语言模型经历了显著升级：更长的上下文窗口、结构化输出、更好的推理能力以及简单的工具集成。这些进展使得构建可靠的代理应用程序比以往任何时候都更可行。

在本博客中，我们将探讨构建代理应用程序的三种流行框架：LangGraph、CrewAI和OpenAI Swarm。通过一个代理金融助手的实际示例，我们将突出每个框架的优势、劣势和实际用例。

代理是什么？

代理是由大语言模型（LLMs）驱动的先进系统，能够独立与环境互动并实时做出决策。与传统的LLM应用程序不同，后者结构为固定的预定义管道（例如，A → B → C），代理工作流程引入了一种动态和自适应的方法。代理利用工具 - 使其能够与环境互动的函数或API - 根据上下文和目标决定下一步。这种灵活性使代理能够偏离固定的顺序，从而实现更自主和高效的工作流程，适应复杂和不断变化的任务。

然而，这种灵活性也带来了自身的一系列挑战：

• 在任务之间管理状态和内存
• 编排多个子代理及其通信架构
• 确保工具调用的可靠性，并在出现复杂错误情况时进行处理
• 处理大规模的推理和决策。

我们为什么需要代理框架

从头开始构建代理并非易事。像 LangGraph、CrewAI 和 OpenAI Swarm 这样的框架简化了这一过程，使开发人员能够专注于他们的应用逻辑，而不是为了状态管理、编排和工具集成而重新发明轮子。

在其核心，代理框架提供了：

• 定义代理和工具的简单方法
• 编排机制
• 状态管理
• 额外的工具以支持更复杂的应用，如：持久层（内存）、中断等

我们将在接下来的部分中逐一介绍这些内容。‍

介绍代理框架

我们选择 LangGraph、CrewAI 和 OpenAI Swarm，因为它们代表了代理开发领域最新的思想流派。以下是快速概述：

• LangGraph： 顾名思义，LangGraph 把图形架构作为定义和编排代理工作流的最佳方式。与早期版本的 LangChain 不同，LangGraph 是一个设计良好的框架，具有许多强大且可定制的功能，适用于生产。然而，对于某些用例，它有时比必要的更复杂，并可能产生额外的开销。
• CrewAI： 相比之下，CrewAI 更容易入门。它提供了直观的抽象，帮助您专注于任务设计，而不是编写复杂的编排和状态管理逻辑。然而，权衡在于它是一个高度主观的框架，后续定制化更为困难。
• OpenAI Swarm： 一个轻量级、极简的框架，被 OpenAI 描述为“教育性”的，而不是“生产就绪的”。OpenAI Swarm 几乎代表了一种“反框架”——将许多功能留给开发者自行实现或由强大的 LLMs 自行解决。我们相信，对于今天有简单用例的用户，或者想要将灵活的代理工作流集成到现有 LLM 管道中的用户，它可能是完美的选择。

截至2024年11月的Github星标历史比较。

其他显著的框架

• LlamaIndex 工作流： 一个事件驱动的框架，概念上非常适合许多代理工作流。然而，现阶段我们发现，它仍然要求开发者编写大量的样板代码才能使其工作。LlamaIndex 团队正在积极改进工作流框架，我们希望他们能尽快创建更多高级抽象。

• AutoGen： 由微软开发的多代理对话编排框架，AutoGen 已被用于多种代理用例。AutoGen 团队从早期版本的错误和反馈中吸取教训，正在进行全面重写（从 v0.2 到 v0.4），转变为一个事件驱动的编排框架。

构建代理金融助手

为了基准测试这些框架，我们使用每个框架构建了相同的 代理金融助手。构建的应用程序的完整代码可以在这里找到：Relari Agent Examples。

我们希望代理能够处理复杂的查询，例如：

• Spirit Airline的财务健康状况与其竞争对手相比如何？
• 从财务角度来看，Apple表现最好的产品线是什么？他们在网站上营销什么？
• 帮我找一些市值低于50亿美元、年收入增长超过20%的消费股

以下是我们希望助手提供的全面响应的示例片段：

为了实现这些目标，我们通过 FMP API 为代理系统提供对金融数据库的访问，并提供互联网访问以研究互联网内容。

在构建代理AI应用程序时，我们需要做出的第一个选择是架构。有几种架构，每种都有其优缺点。在下面的图像中，有一些由 LangGraph 总结的流行架构（您可以在这里阅读有关架构选择的更多信息：多代理架构）。

我们为这个应用程序选择了监督架构作为教育目的。因此，我们将创建一个监督代理，其任务是决定将任务委派给哪个子代理，以及三个具有工具访问权限的子代理：一个金融数据代理，一个网络研究代理和一个总结代理。

让我们探索每个框架如何处理代理创建、工具集成、编排、内存和人机交互。

1. 定义代理和工具

我们首先来看如何定义一个常规代理，例如金融数据代理、网络研究代理和输出总结代理，并在每个框架中声明其相关工具。监督代理是一个特殊的代理，扮演编排角色，因此我们将在编排部分进行讨论。

LangGraph

创建一个简单的工具调用代理的最简单方法是使用预构建的 create_react_agent 函数，如下所示，我们可以提供希望该代理使用的工具和提示。

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

@tool
def get_stock_price(symbol: str) -> dict:
    """Fetch the current stock price for a given symbol.
    Args:
        symbol (str): The stock ticker symbol (e.g., "AAPL" for Apple Inc.).
    Returns:
        dict: A dictionary containing the stock price or an error message.
    """
    base_url = "https://financialmodelingprep.com/api/v3/quote-short"
    params = {"symbol": symbol, "apikey": os.getenv("FMP_API_KEY")}

    response = requests.get(base_url, params=params)
    if response.status_code == 200:
        data = response.json()
        if data:
            return {"price": data[0]["price"]}
    return {"error": "Unable to fetch stock price."}

financial_data_agent = create_react_agent(
    ChatOpenAI(model="gpt-4o-mini"),
    tools=[get_stock_price, get_company_profile, ...],
    state_modifier="You are a financial data agent responsible for retrieving financial data using the provided API tools ...",
)

在 LangGraph 中，一切都被结构化为图。实用函数 create_react_agent 创建一个简单的可执行图，其中包含代理节点和工具节点。

代理充当决策者，动态确定调用哪些工具，并评估是否有足够的信息过渡到 __end__ 状态。

在图中，实线表示确定性边缘（工具节点必须始终返回到代理），而虚线表示条件边缘，LLM 驱动的代理正在决定下一步去哪里。

节点和边缘是图的基础构建块。我们将在后面的编排部分看到，这个图可以作为一个更大、更复杂图中的一个节点来表示。

CrewAI

CrewAI的代理定义围绕代理与任务之间的关系展开（代理应该完成的内容）。

对于每个代理，我们必须定义其role、goal和backstory，并指定其可以访问的工具。

from crewai import Agent, Task

financial_data_agent = Agent(
    role="金融数据代理",
    goal="使用FMP API检索全面的金融数据，以提供回答用户查询所需的数据",
    backstory="""你是一位经验丰富的金融数据收集者，拥有广泛的金融信息收集经验。以你的精准性和
    使用FMP API查找最相关的金融数据点的能力而闻名，该API提供有关美国上市公司的金融数据""",
    tools=[
        StockPriceTool(),
        CompanyProfileTool(),
        ...
    ]
)

然后，我们必须创建需要代理执行的任务。任务必须包含description和expected_output。

gather_financial_data = Task(
    description=("进行全面的金融研究，以收集相关的金融数据，以帮助 "
    "回答用户查询：{query}。使用可用的金融工具获取准确 "
    "和最新的信息。重点寻找相关的股票价格、公司资料、 "
    "财务比率和其他相关的金融指标，以回答用户查询：{query}。"),
    expected_output="一组全面的金融数据点，直接回答查询：{query}。",
    agent=financial_data_agent,
)

这种构建LLMs提示的结构化方法提供了一个清晰且一致的框架，确保代理和任务得到良好定义。虽然这种方法有助于保持关注和连贯性，但在重复定义角色、目标、背景故事和任务描述时，有时可能会感觉僵硬或重复。

工具可以使用@tool装饰器进行集成，类似于LangGraph中的方法。值得一提的是，另外，我们可以扩展BaseTool类，这将是强制工具输入模式的更稳健方法，得益于使用Pydantic模型（这种方法也得到了LangGraph的支持）。

class StockPriceInput(BaseModel):
    """股票价格查询的输入模式。"""
    symbol: str = Field(..., description="股票代码")

class StockPriceTool(BaseTool):
    name: str = "获取股票价格"
    description: str = "获取给定股票代码的当前股票价格"
    args_schema: Type[BaseModel] = StockPriceInput

def _run(self, symbol: str) -> dict:

OpenAI Swarm

Swarm采取了一种不同的方法：OpenAI建议将推理流程结构化为系统提示中的“例程”（在instructions中），即代理为完成任务而遵循的预定义步骤或指令集合。这种做法来自OpenAI是可以理解的，因为他们希望开发者更多地依赖模型遵循指令的能力，而不是在代码中定义自定义逻辑集合。我们发现这种方法在使用更强大的大语言模型时简单有效，因为它能够跟踪例程中的推理。

对于工具，我们可以直接引入作为工具。

from swarm import Agent

financial_data_agent = Agent(
    name="Financial Data Agent",
    instructions="""You are a financial data specialist responsible for retrieving financial data using the provided API tools.
    Your tasks:
    Step 1. Given a user query, use the appropriate tool to fetch relevant financial data
    Step 2. Read the data and make sure they can answer the user query. If not, modify the tool input or use different tools to get more information.
    Step 3. Once you have gathered enough information, return only the raw data obtained from the tool. Do not add commentary or explanations""",
    functions=[
        get_stock_price,
        get_company_profile,
        ...
    ]
)

2. 编排

我们现在来看每个框架的核心部分，了解它们如何将多个子代理聚集在一起。

LangGraph

LangGraph的核心是基于图的编排。我们首先创建监督代理，它充当路由器，唯一的任务是分析情况并决定下一个调用哪个代理。执行代理本身只能将结果传回监督代理。

LangGraph要求明确的状态定义。AgentState类有助于在不同代理之间定义一个通用的状态模式。

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

对于每个代理，我们通过将状态包装在一个节点中与之交互，该节点将代理输出转换为一致的消息模式。

async def financial_data_node(state):
    result = await financial_data_agent.ainvoke(state)
    return {
        "messages": [
            AIMessage(
                content=result["messages"][-1].content, name="Financial_Data_Agent"
            )
        ]
    }

我们现在准备定义代理本身。

class RouteResponse(BaseModel):
    next: Literal[OPTIONS]

def supervisor_agent(state):
    prompt = ChatPromptTemplate.from_messages([
        ("system", ORCHESTRATOR_SYSTEM_PROMPT),
        MessagesPlaceholder(variable_name="messages"),
        (
            "system",
            "Given the conversation above, who should act next?"
            " Or should we FINISH? Select one of: {options}",
        ),
    ]).partial(options=str(OPTIONS), members=", ".join(MEMBERS))

    supervisor_chain = prompt | LLM.with_structured_output(RouteResponse)
    return supervisor_chain.invoke(state)

在定义了监督代理后，我们通过将每个代理作为节点添加并将所有执行逻辑作为边来定义代理工作流作为图。

在定义边时，我们有两种可能性：常规或条件边。当我们希望进行确定性转换时，使用常规边。例如，金融数据代理应始终将结果返回给监督代理，以决定下一步。

条件边则在我们希望大语言模型选择路径时使用（例如，监督代理决定是否有足够的数据直接发送给输出总结代理，或返回数据和网络代理以获取更多信息）。

from langgraph.graph import END, START, StateGraph

def build_workflow() -> StateGraph:
    """Construct the state graph for the workflow."""
    workflow = StateGraph(AgentState)

    workflow.add_node("Supervisor_Agent", supervisor_agent)
    workflow.add_node("Financial_Data_Agent", financial_data_node)
    workflow.add_node("Web_Research_Agent", web_research_node)
    workflow.add_node("Output_Summarizing_Agent", output_summarizing_node)

    workflow.add_edge("Financial_Data_Agent", "Supervisor_Agent")
    workflow.add_edge("Web_Research_Agent", "Supervisor_Agent")

    conditional_map = {
        "Financial_Data_Agent": "Financial_Data_Agent",
        "Web_Research_Agent": "Web_Research_Agent",
        "Output_Summarizing_Agent": "Output_Summarizing_Agent",
        "FINISH": "Output_Summarizing_Agent",
    }
    workflow.add_conditional_edges(
        "Supervisor_Agent", lambda x: x["next"], conditional_map
    )

    workflow.add_edge("Output_Summarizing_Agent", END)
    workflow.add_edge(START, "Supervisor_Agent")

    return workflow

这里是生成的图。

CrewAI

与 LangGraph 相比，CrewAI 抽象了大部分编排任务。

supervisor_agent = Agent(
    role="金融助理经理",
    goal="利用同事的技能来回答用户的查询：{query}。",
    backstory="""您是一位负责金融助理工作流程的经理，擅长监督具有不同技能的复杂工作人员，并确保能够在同事的帮助下回答用户的查询。
    您总是尝试先使用金融数据代理和/或网络爬虫代理收集数据。
    收集数据后，您必须委派给输出总结代理以创建全面的报告，而不是直接回答用户的查询。""",
    verbose=True,
    llm=ChatOpenAI(model="gpt-4o", temperature=0.5),
    allow_delegation=True,
)

与 LangGraph 类似，我们首先创建监督代理。注意 allow_delegation 标志，它允许代理将任务传递给其他代理。

接下来，我们使用 Crew 将代理聚集在一起。在这里选择 Process.hierarchical 是很重要的，以允许监督代理委派任务。在后台，监督代理接受用户查询并将其转换为任务，然后找到相关的代理来执行这些任务。如果我们想要创建一个更确定性的过程，其中任务将按顺序执行，则可以选择不使用管理代理并使用 Process.sequential。

finance_crew = Crew(
    agents=[
        financial_data_agent,
        web_scraping_agent,
        output_summarizing_agent
    ],
    tasks=[
        gather_financial_data,
        gather_website_information,
        summarize_findings
    ],
    process=Process.hierarchical,
    manager_agent=supervisor_agent,
)

OpenAI Swarm

Swarm orchestration uses a very simple strategy — handoffs. The core idea is to create a transfer function where it uses another agent as a tool.

This is undoubtedly the cleanest approach. The relationships are implicit in the transfer functions.

def transfer_to_summarizer():
    return summarizing_agent

def transfer_to_web_researcher():
    return web_researcher_agent

def transfer_to_financial_data_agent():
    return financial_data_agent

supervisor_agent = Agent(
    name="监督",
    instructions="""您是一个监督代理，负责协调金融数据代理、网络研究员代理和总结代理。
    您的任务：
    1. 根据用户查询，确定将任务委托给哪个代理
    2. 如果用户的查询需要金融数据，委托给金融数据代理
    3. 如果用户的查询需要网络研究，委托给网络研究员代理
    4. 如果已经有足够的信息来回答用户的查询，委托给总结代理以获得最终输出。
    永远不要自己总结数据。始终委托给总结代理以提供最终输出。
    """,
    functions=[ 
        transfer_to_financial_data_agent,
        transfer_to_web_researcher,
        transfer_to_summarizer
    ]
)

A drawback of this approach is that as the application grows the dependencies between agents are harder to keep track of.

3. 内存

内存是一个有状态的代理系统的关键组成部分。我们可以区分两层内存：

• 短期 记忆允许代理维持多轮 / 多步执行，
• 长期 记忆允许代理在会话之间学习和记住偏好。

这个主题可能会变得非常复杂，但让我们看一下每个框架中可用的最简单的内存编排。

LangGraph

LangGraph 在内线程（单个对话线程内的内存）和跨线程（跨对话的内存）之间区分内存。

为了保存内线程内存，LangGraph 提供了 MemorySaver() 类，该类将图的状态或对话历史保存到 checkpointer 中。

from langgraph.checkpoint.memory import MemorySaver

def build_app():
    """Build and compile the workflow."""
    memory = MemorySaver()
    workflow = build_workflow()
    return workflow.compile(checkpointer=memory)

要将代理执行与内存线程关联，请传递一个包含 thread_id 的配置。这告诉代理使用哪个线程的内存检查点。例如：

config = {"configurable": {"thread_id": "1"}}
app = build_app()
await run(app, input, config)

为了保存跨线程内存，LangGraph 允许我们将内存保存到 JSON 文档存储中。

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()  
user_id = "user_0"
store.put(
    user_id,
    "current_portfolio",
    {
        "portfolio": ["TSLA", "AAPL", "GOOG"],
    }
)

CrewAI

毫不奇怪，CrewAI采取了一种更简单但更严格的方法。在开发者方面，只需将内存设置为true。

finance_crew = Crew(
    agents=[financial_data_agent, web_researcher_agent, summarizing_agent],
    tasks=[gather_financial_data, gather_website_information, summarize_findings],
    process=Process.hierarchical,
    manager_agent=supervisor_agent,
    memory=True, 
    verbose=True, 
)

它在后台所做的事情非常复杂，因为它创建了几种不同的内存存储：

• 短期记忆：它创建了一个使用OpenAI Embeddings的ChromaDB向量存储，用于存储代理执行历史。
• 最近记忆：SQLite3数据库用于存储最近的任务执行结果。
• 长期记忆：SQLite3数据库用于存储任务结果，请注意，任务描述必须完全匹配（相当严格），才能检索到长期记忆。
• 实体记忆：提取关键实体并将实体关系存储到另一个ChromaDB向量存储中。

图片来自Relari。CrewAI自动生成的不同内存存储数据库。

OpenAI Swarm

Swarm使用简单的无状态设计，并没有任何内置的内存功能。关于OpenAI如何看待内存的一个参考可以在其有状态的Assistant API中看到。每个对话都有一个thread_id用于短期记忆，而每个助手都有一个assistant_id可以与长期记忆关联。

还可以集成第三方内存层提供者，如mem0或实现我们自己的短期和长期记忆。

4. 人类在循环中

虽然我们希望代理能够自主运行，但许多代理被设计为与人类互动。

例如，客户支持代理可以在执行链的整个过程中向用户询问信息。人类还可以充当审计员或指导者，以实现更无缝的人类与代理的协作。

LangGraph

LangGraph 允许我们在图中设置断点，如下所示，如果我们想在总结器构建最终输出之前添加一个人工检查点。

workflow.compile(checkpointer=checkpointer, interrupt_before=["Output_Summarizing_Agent"])

图将执行直到达到断点。然后我们可以实现一个步骤，以在继续图之前获取用户输入。

for event in graph.stream(initial_input, thread, stream_mode="values"):
    print(event)

try:

    user_approval = input("您想要继续到输出总结器吗？ (yes/no): ")
except:
    user_approval = "yes"

if user_approval.lower() == “yes”:

    for event in graph.stream(None, thread, stream_mode="values"):
        print(event)

else:

    print("操作已被用户取消。")

CrewAI

CrewAI 允许人类通过在代理初始化时设置 human_input=True 标志来向代理提供反馈。

代理执行后将暂停，并询问用户对其操作和结果输入自然语言反馈（见下文）。

然而，它不支持更多自定义的人机交互。

OpenAI Swarm

Swarm没有任何内置的人机协作功能。然而，在执行过程中添加人类输入的最简单方法是将人类添加为工具或作为AI代理可以转移到的代理。

功能差异总结

为了总结我们在三个框架中构建相同应用程序的发现。

图片来源：Relari。功能比较总结。

我们的建议

我们在流程图中总结了我们的建议，以帮助您决定从哪个框架开始。

图片来自 Relari。代理框架决策树。

接下来是什么？

本博客专注于构建代理的第一个版本。这些代理绝不是高效或可靠的。

下一个博客将专注于改进和迭代这些代理（通常是AI系统生产的大部分工作）。

• 定义成功标准和指标
• 基准测试使用不同框架构建的代理的性能
• 深入研究质量和可靠性
• 针对提示、工具、推理、架构进行有针对性的改进

我们将使用一些最新的代理评估、模拟和可观察性工具，深入探讨如何将这些代理转变为生产就绪的系统。

最初发布于 https://www.relari.ai.