LLM代理开发框架对比

LLM代理（智能体）正处于一个发展阶段。随着多个新框架的出现和该领域的新投资，现代人工智能代理正在克服不稳定的起源，迅速取代 RAG 成为实施重点。那么，2024 年最终会成为自主人工智能系统接管我们写电子邮件、预订航班、处理数据或任何其他任务的一年吗？

也许吧，但要达到这一点还有很多工作要做。任何构建代理的开发人员不仅必须选择基础——使用哪种模型、用例和架构——还必须选择要利用哪个框架。你是选择长期存在的 LangGraph，还是新进入的 LlamaIndex Workflows？还是走传统路线，自己编写整个代码？

这篇文章旨在让这个选择变得更容易一些。在过去的几周里，我在主要框架中构建了同一个代理，以在技术层面上检查每个代理的一些优缺点。每个代理的所有代码都可以在这个 repo 中找到。

1、用于测试的代理的背景

用于测试的代理包括函数调用、多种工具或技能、与外部资源的连接以及共享状态或内存。

代理具有以下功能：

• 回答知识库中的问题
• 与数据对话：回答有关 LLM 应用程序遥测数据的问题
• 分析数据：分析检索到的遥测数据中的更高级别趋势和模式

为了实现这些，代理有三个起始技能：带有产品文档的 RAG、跟踪数据库上的 SQL 生成和数据分析。代理 UI 使用简单的 gradio 驱动界面，代理本身构造为聊天机器人。

2、基于代码的代理（无框架）

开发代理时的第一个选择是完全跳过框架并完全自己构建代理。在着手这个项目时，这是我开始采用的方法。

2.1 纯代码架构

下面的基于代码的代理由 OpenAI 驱动的路由器组成，该路由器使用函数调用来选择要使用的正确技能。该技能完成后，它会返回路由器以调用另一项技能或响应用户。

代理会保存一个持续的消息和响应列表，这些列表在每次调用时都会完全传递到路由器中，以在整个周期中保留上下文。

def router(messages):
    if not any(
        isinstance(message, dict) and message.get("role") == "system" for message in messages
    ):
        system_prompt = {"role": "system", "content": SYSTEM_PROMPT}
        messages.append(system_prompt)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=skill_map.get_combined_function_description_for_openai(),
    )

    messages.append(response.choices[0].message)
    tool_calls = response.choices[0].message.tool_calls
    if tool_calls:
        handle_tool_calls(tool_calls, messages)
        return router(messages)
    else:
        return response.choices[0].message.content

技能本身在自己的类（例如 GenerateSQLQuery）中定义，这些类共同保存在 SkillMap 中。路由器本身只与 SkillMap 交互，它使用 SkillMap 加载技能名称、描述和可调用函数。这种方法意味着向代理添加新技能就像将该技能写为其自己的类，然后将其添加到 SkillMap 中的技能列表中一样简单。这里的想法是让添加新技能变得容易，而不会干扰路由器代码。

class SkillMap:
    def __init__(self):
        skills = [AnalyzeData(), GenerateSQLQuery()]

        self.skill_map = {}
        for skill in skills:
            self.skill_map[skill.get_function_name()] = (
                skill.get_function_dict(),
                skill.get_function_callable(),
            )

    def get_function_callable_by_name(self, skill_name) -> Callable:
        return self.skill_map[skill_name][1]

    def get_combined_function_description_for_openai(self):
        combined_dict = []
        for _, (function_dict, _) in self.skill_map.items():
            combined_dict.append(function_dict)
        return combined_dict

    def get_function_list(self):
        return list(self.skill_map.keys())

    def get_list_of_function_callables(self):
        return [skill[1] for skill in self.skill_map.values()]

    def get_function_description_by_name(self, skill_name):
        return str(self.skill_map[skill_name][0]["function"])

总体而言，这种方法实施起来相当简单，但也面临一些挑战。

2.2 纯代码代理的挑战

第一个困难在于构造路由器系统提示。通常，上面示例中的路由器坚持自己生成 SQL，而不是将其委托给正确的技能。如果您曾经尝试让 LLM 不做某事，您就会知道这种体验有多么令人沮丧；找到一个有效的提示需要经过多轮调试。考虑每个步骤的不同输出格式也很棘手。由于我选择不使用结构化输出，因此我必须准备好应对路由器和技能中每个 LLM 调用的多种不同格式。

2.3 纯代码代理的好处

基于代码的方法提供了良好的基线和起点，提供了一种很好的方法来学习代理的工作原理，而无需依赖来自主流框架的固定代理教程。虽然说服 LLM 表现可能具有挑战性，但代码结构本身足够简单易用，可能对某些用例有意义（更多信息请参见下面的分析部分）。

3、LangGraph

LangGraph 是最古老的代理框架之一，于 2024 年 1 月首次发布。该框架旨在通过采用 Pregel 图结构来解决现有管道和链的非循环性质。LangGraph 通过添加节点、边和条件边的概念来遍历图，使在代理中定义循环变得更加容易。LangGraph 建立在 LangChain 之上，并使用该框架中的对象和类型。

3.1 LangGraph 架构

LangGraph 代理在纸面上看起来与基于代码的代理类似，但其背后的代码却截然不同。LangGraph 在技术上仍然使用“路由器”，因为它使用函数调用 OpenAI 并使用响应继续执行新步骤。然而，程序在技能之间移动的方式控制方式完全不同。

tools = [generate_and_run_sql_query, data_analyzer]
model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)

def create_agent_graph():
    workflow = StateGraph(MessagesState)

    tool_node = ToolNode(tools)
    workflow.add_node("agent", call_model)
    workflow.add_node("tools", tool_node)

    workflow.add_edge(START, "agent")
    workflow.add_conditional_edges(
        "agent",
        should_continue,
    )
    workflow.add_edge("tools", "agent")

    checkpointer = MemorySaver()
    app = workflow.compile(checkpointer=checkpointer)
    return app

这里定义的图表有一个用于初始 OpenAI 调用的节点，称为上面的“代理”，还有一个用于工具处理步骤的节点，称为“工具”。 LangGraph 有一个名为 ToolNode 的内置对象，它获取可调用工具的列表并根据 ChatMessage 响应触发它们，然后再次返回到“代理”节点。

def should_continue(state: MessagesState):
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

def call_model(state: MessagesState):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

每次调用“代理”节点（换句话说：基于代码的代理中的路由器）后，should_continue 边缘都会决定是否将响应返回给用户或传递给 ToolNode 来处理工具调用。

在每个节点中，“状态”存储来自 OpenAI 的消息和响应列表，类似于基于代码的代理的方法。

3.2 LangGraph 的挑战

示例中 LangGraph 的大部分困难源于需要使用 Langchain 对象才能顺利运行。

挑战 1：函数调用验证

为了使用 ToolNode 对象，我不得不重构大部分现有的 Skill 代码。ToolNode 采用可调用函数列表，这最初让我认为我可以使用现有的函数，但由于我的函数参数，事情变得一团糟。

技能被定义为具有可调用成员函数的类，这意味着它们的第一个参数是“self”。 GPT-4o 非常聪明，不会在生成的函数调用中包含“self”参数，但 LangGraph 将其解读为由于缺少参数而导致的验证错误。

这花了几个小时才弄清楚，因为错误消息反而将函数中的第三个参数（数据分析技能中的“args”）标记为缺少的参数：

pydantic.v1.error_wrappers.ValidationError: 1 validation error for data_analysis_toolSchema
args field required (type=value_error.missing)

值得一提的是，错误信息来自 Pydantic，而不是 LangGraph。

我最终下定决心，用 Langchain 的 @tool 装饰器将我的技能重新定义为基本方法，并让一切正常运转。

@tool
def generate_and_run_sql_query(query: str):
    """Generates and runs an SQL query based on the prompt.

    Args:
        query (str): A string containing the original user prompt.

    Returns:
        str: The result of the SQL query.
    """

挑战 #2：调试

如前所述，在框架中调试很困难。这主要归结于令人困惑的错误消息和抽象概念，这些使查看变量变得更加困难。

抽象概念主要出现在尝试调试代理周围发送的消息时。LangGraph 将这些消息存储在状态 [“messages”] 中。图中的某些节点会自动从这些消息中提取数据，这可能会使节点访问消息时难以理解消息的值。

3.3 LangGraph 的优势

LangGraph 的主要优势之一是易于使用。图形结构代码简洁易懂。特别是如果您有复杂的节点逻辑，拥有图形的单一视图可以更轻松地理解代理是如何连接在一起的。LangGraph 还可以轻松转换在 LangChain 中构建的现有应用程序。

3.4 总结

如果您使用框架中的所有内容，LangGraph 可以干净利落地运行；如果您不使用它，请准备好面对一些调试难题。

4、LlamaIndex Workflows

Workflows 是代理框架领域的新成员，于今年夏初首次亮相。与 LangGraph 一样，它旨在使循环代理更易于构建。Workflows 还特别注重异步运行。

Workflows 的一些元素似乎直接响应了 LangGraph，特别是它使用事件而不是边和条件边。Workflows 使用步骤（类似于 LangGraph 中的节点）来容纳逻辑，并使用发出和接收事件在步骤之间移动。

上面的结构看起来与 LangGraph 结构类似，除了一个附加项。我在 Workflow 中添加了一个设置步骤来准备代理上下文，下面将详细介绍。尽管结构相似，但支持它的代码却大不相同。

4.1 Workflows 架构

下面的代码定义了 Workflow 结构。与 LangGraph 类似，这是我准备状态并将技能附加到 LLM 对象的地方。

class AgentFlow(Workflow):
    def __init__(self, llm, timeout=300):
        super().__init__(timeout=timeout)
        self.llm = llm
        self.memory = ChatMemoryBuffer(token_limit=1000).from_defaults(llm=llm)
        self.tools = []
        for func in skill_map.get_function_list():
            self.tools.append(
                FunctionTool(
                    skill_map.get_function_callable_by_name(func),
                    metadata=ToolMetadata(
                        name=func, description=skill_map.get_function_description_by_name(func)
                    ),
                )
            )

    @step
    async def prepare_agent(self, ev: StartEvent) -> RouterInputEvent:
        user_input = ev.input
        user_msg = ChatMessage(role="user", content=user_input)
        self.memory.put(user_msg)

        chat_history = self.memory.get()
        return RouterInputEvent(input=chat_history)

这也是我定义额外步骤“prepare_agent”的地方。此步骤根据用户输入创建一个 ChatMessage 并将其添加到工作流内存中。将其拆分为单独的步骤意味着我们在代理循环执行步骤时会返回到它，从而避免反复将用户消息添加到内存中。

在 LangGraph 案例中，我使用位于图表之外的 run_agent 方法完成了同样的事情。这种变化主要是风格上的，但我认为像我们在这里所做的那样，将此逻辑与工作流和图表放在一起会更简洁。

设置好工作流后，我定义了路由代码：

@step
async def router(self, ev: RouterInputEvent) -> ToolCallEvent | StopEvent:
    messages = ev.input

    if not any(
        isinstance(message, dict) and message.get("role") == "system" for message in messages
    ):
        system_prompt = ChatMessage(role="system", content=SYSTEM_PROMPT)
        messages.insert(0, system_prompt)

    with using_prompt_template(template=SYSTEM_PROMPT, version="v0.1"):
        response = await self.llm.achat_with_tools(
            model="gpt-4o",
            messages=messages,
            tools=self.tools,
        )

    self.memory.put(response.message)

    tool_calls = self.llm.get_tool_calls_from_response(response, error_on_no_tool_call=False)
    if tool_calls:
        return ToolCallEvent(tool_calls=tool_calls)
    else:
        return StopEvent(result=response.message.content)

以及工具调用处理代码：

@step
async def tool_call_handler(self, ev: ToolCallEvent) -> RouterInputEvent:
    tool_calls = ev.tool_calls

    for tool_call in tool_calls:
        function_name = tool_call.tool_name
        arguments = tool_call.tool_kwargs
        if "input" in arguments:
            arguments["prompt"] = arguments.pop("input")

        try:
            function_callable = skill_map.get_function_callable_by_name(function_name)
        except KeyError:
            function_result = "Error: Unknown function call"

        function_result = function_callable(arguments)
        message = ChatMessage(
            role="tool",
            content=function_result,
            additional_kwargs={"tool_call_id": tool_call.tool_id},
        )

        self.memory.put(message)

    return RouterInputEvent(input=self.memory.get())

这两个看起来更像基于代码的代理而不是 LangGraph 代理。这主要是因为 Workflows 将条件路由逻辑保留在步骤中而不是条件边缘中 — 第 18-24 行是 LangGraph 中的条件边缘，而现在它们只是路由步骤的一部分 — 并且 LangGraph 有一个 ToolNode 对象，它几乎自动执行 tool_call_handler 方法中的所有操作。

经过路由步骤后，我很高兴看到一件事，那就是我可以将我的 SkillMap 和基于代码的代理中的现有技能与 Workflows 结合使用。这些不需要任何更改即可与 Workflows 配合使用，这让我的生活变得轻松多了。

4.2 Workflows 的挑战

挑战 #1：同步与异步

虽然异步执行对于实时代理来说是更好的选择，但调试同步代理要容易得多。Workflows 旨在异步工作，尝试强制同步执行非常困难。

我最初以为我只需删除“异步”方法标识，并从“achat_with_tools”切换到“chat_with_tools”即可。但是，由于 Workflow 类中的底层方法也被标记为异步，因此必须重新定义这些方法才能同步运行。我最终坚持使用异步方法，但这并没有使调试变得更加困难。

挑战 #2：Pydantic 验证错误

与 LangGraph 的困境如出一辙，类似的问题也出现在技能上令人困惑的 Pydantic 验证错误。幸运的是，这一次这些问题更容易解决，因为 Workflows 能够很好地处理成员函数。最终，我不得不更加规范地为我的技能创建 LlamaIndex FunctionTool 对象：

for func in skill_map.get_function_list(): 
            self.tools.append(FunctionTool(
                skill_map.get_function_callable_by_name(func), 
                metadata=ToolMetadata(name=func, description=skill_map.get_function_description_by_name(func))))

4.3 工作流的优势

与 LangGraph 代理相比，我构建工作流代理要容易得多，主要是因为工作流仍然需要我自己编写路由逻辑和工具处理代码，而不是提供内置函数。这也意味着我的工作流代理看起来与我的基于代码的代理极为相似。

最大的区别在于事件的使用。我使用两个自定义事件在代理中的步骤之间移动：

class ToolCallEvent(Event):
    tool_calls: list[ToolSelection]

class RouterInputEvent(Event):
    input: list[ChatMessage]

基于事件的发射器-接收器架构取代了直接调用我的代理中的某些方法，例如工具调用处理程序。

如果您拥有更复杂的系统，其中包含多个异步触发的步骤，并且可能会发出多个事件，那么这种架构对于干净地管理这些步骤非常有用。

工作流的其他好处包括它非常轻量级，不会强迫您使用太多结构（除了使用某些 LlamaIndex 对象），并且其基于事件的架构为直接函数调用提供了一种有用的替代方案——尤其是对于复杂的异步应用程序。

4.4 比较框架

纵观这三种方法，每种方法都有其优点。

无框架方法是最容易实现的。因为任何抽象都是由开发人员定义的（即上例中的 SkillMap 对象），所以保持各种类型和对象的连贯性很容易。然而，代码的可读性和可访问性完全取决于个人开发人员，很容易看出，如果没有一些强制结构，日益复杂的代理会变得多么混乱。

LangGraph 提供了相当多的结构，这使得代理的定义非常明确。如果一个更广泛的团队正在合作开发代理，这种结构将提供一种实施架构的有用方法。对于那些不太熟悉该结构的人来说，LangGraph 也可能是一个很好的代理起点。然而，这有一个权衡——由于 LangGraph 为您做了很多事情，如果您没有完全接受该框架，它可能会导致麻烦；代码可能非常干净，但您可能需要付出更多调试的代价。

工作流介于两者之间。基于事件的架构可能对某些项目非常有用，而且在使用 LlamaIndex 类型方面要求更少的事实为那些没有在其应用程序中完全使用框架的人提供了更大的灵活性。

最终，核心问题可能归结为“您是否已经在使用 LlamaIndex 或 LangChain 来编排您的应用程序？” LangGraph 和 Workflows 都与各自的底层框架紧密相连，因此每个代理特定框架的额外优势可能不会让您仅凭优点就切换。

纯代码方法可能始终是一个有吸引力的选择。如果您能够严格记录和执行任何创建的抽象，那么确保外部框架中的任何内容都不会减慢您的速度就很容易了。

5、帮助选择代理框架的关键问题

当然，“视情况而定”永远不是一个令人满意的答案。这三个问题应该可以帮助您决定在下一个代理项目中使用哪个框架。

您是否已经在项目的重要部分中使用 LlamaIndex 或 LangChain？

如果是，请先探索该选项。

您是否熟悉常见的代理结构，或者您想要一些东西来告诉您应该如何构建代理？

如果您属于后者，请尝试 Workflows。如果您真的属于后者，请尝试 LangGraph。

您的代理之前是否构建过？

该框架的一个优点是，每个框架都有许多教程和示例。可供构建的纯代码代理示例要少得多。

6、结束语

选择代理框架只是众多选择之一，这些选择将影响生成式 AI 系统生产的结果。与往常一样，拥有强大的护栏和 LLM 跟踪是值得的——并且随着新的代理框架、研究和模型颠覆既定技术，保持敏捷性。

原文链接：Choosing Between LLM Agent Frameworks

BimAnt翻译整理，转载请标明出处