解锁 llamaindex agentworkflow 的潜力：有效多代理编排的 5 个关键策略

推荐语

深入探索LlamaIndex AgentWorkflow，掌握高效多智能体编排策略。

核心内容：
1. LlamaIndex AgentWorkflow框架的介绍与应用案例
2. AgentWorkflow与Workflow的区别及优势
3. 多智能体编排的最佳实践和代码示例

杨芳贤

53A创始人/腾讯云(TVP)最具价值专家

本文介绍进一步优化

深入研究 LlamaIndex AgentWorkflow：一个近乎完美的多智能体编排解决方案。 图片由 DALL-E-3 提供

本文向您介绍了 LlamaIndex 最新的 AgentWorkflow 多智能体编排框架，通过一个项目演示了它的应用，突出了它的缺点，并解释了我如何解决这些问题。

通过阅读本文，您将学习如何使用 LlamaIndex AgentWorkflow 简化多智能体编排并提高开发效率。

本文讨论的项目源代码可在文章末尾找到，供您审查和修改，无需我的许可。

简介

最近，我不得不为工作审查 LlamaIndex 的官方文档，并对发生的巨大变化感到惊讶：LlamaIndex 已从 RAG 框架重新定位为集成了数据和工作流程的多智能体框架。 整个文档 现在围绕 AgentWorkflow 构建。

多智能体编排并非新鲜事物。

对于企业级应用，我们不使用一个独立的智能体来执行一系列任务。相反，我们更喜欢一个能够编排多个智能体协同完成复杂业务场景的框架。

说到多智能体编排框架，您可能听说过 LangGraph、CrewAI 和 AutoGen。然而，曾经与 LangChain 一样受欢迎的框架 LlamaIndex，在过去六个月里似乎在多智能体领域保持沉默。

考虑到 LlamaIndex 的高度成熟和社区参与度，LlamaIndex AgentWorkflow 的发布引起了我们的注意。因此，我的团队和我研究了一个月，发现对于实际应用而言，AgentWorkflow 是一个近乎完美的多智能体编排解决方案。

您可能很聪明，您可能会问，既然 LlamaIndex Workflow 已经发布了半年，Workflow 和 AgentWorkflow 有什么区别？要回答这个问题，我们首先必须看看如何使用 LlamaIndex Workflow 进行多智能体设置。

什么是 Workflow？

我之前写过一篇文章，详细介绍了 LlamaIndex Workflow 是什么以及如何使用它：

简单来说，Workflow 是一个基于事件驱动的框架，使用 Python asyncio 并发调用 API 到大型语言模型和各种工具。

我还写过关于使用 Workflow 实现类似于 OpenAI Swarm 的智能体移交的多智能体编排：

然而，Workflow 是一个相对低级别的框架，并且与其他 LlamaIndex 模块的连接相当松散，在实现复杂的多智能体逻辑时，需要频繁学习和调用 LlamaIndex 的底层 API。

如果您阅读过我的文章，您会注意到我严重依赖 LlamaIndex 的低级 API 跨 Workflow 的 step 方法进行函数调用和流程控制，这导致了工作流程和特定于智能体的代码之间的紧密耦合。对于我们这些想早点完成工作并在家享受晚餐的人来说，这并不理想。

也许 LlamaIndex 听到了开发者的呼声，从而诞生了 AgentWorkflow。

AgentWorkflow 如何工作？

AgentWorkflow 由一个 AgentWorkflow 模块和一个 Agent 模块组成。与现有的 LlamaIndex 模块不同，这两个模块都是专门为最近的多智能体目标量身定制的。在这里，我们首先讨论 Agent 模块：

Agent 模块

Agent 模块主要包含两个类：FunctionAgent 和 ReActAgent，它们都继承自 BaseWorkflowAgent，因此与之前的 Agent 类不兼容。

如果您的语言模型支持函数调用，请使用 FunctionAgent；如果不支持，请使用 ReActAgent。在本文中，我们使用函数调用来完成特定任务，因此我们将重点关注 FunctionAgent：

FunctionAgent 主要有三个方法：take_step、handle_tool_call_results 和 finalize。

FunctionAgent 中各种方法的插图。 图片由作者提供

take_step 方法接收当前的聊天历史记录 llm_input，以及智能体可用的工具。它使用 astream_chat_with_tools 和 get_tool_calls_from_response 来获取要执行的下一个工具，并将工具调用参数存储在 Context 中。

此外，take_step 输出当前轮的智能体参数和结果，以流的形式输出，方便调试和逐步查看中间智能体执行结果。

handle_tool_call_results 方法不直接执行工具——工具是在 AgentWorkflow 中并发调用的。它只是将工具执行结果保存在 Context 中。

finalize 方法接受一个 AgentOutput 参数，但不会更改它。相反，它从 Context 中提取工具调用堆栈，并将它们保存为 ChatMemory 中的聊天历史记录。

您可以继承和覆盖 FunctionAgent 方法来实现您的业务逻辑，我将在即将到来的项目实践中演示这一点。

Agentworkflow 模块

在介绍了 Agent 模块之后，让我们深入探讨 AgentWorkflow 模块。

在之前的项目中，我基于 Workflow 实现了一个编排流程。这是当时的流程图：

前一篇文章中实现的 workflow 的流程图。图片来自作者

由于我的代码参考了 LlamaIndex 的官方示例，AgentWorkflow 类似于我的实现，但进行了简化，因为它提取了 handoff 和函数调用逻辑。以下是 AgentWorkflow 的架构：

AgentWorkflow 的架构图。图片来自作者

入口点是 init_run 方法，它初始化 Context 和 ChatMemory。

接下来，setup_agent 识别值班 agent，提取其 system_prompt 并将其与当前的 ChatHistory 合并。

然后，run_agent_step 调用 agent 的 take_step 以获取所需的工具进行调用，同时将大型语言模型调用结果写入输出流。在即将到来的项目实践中，我将重写 take_step 以进行特定于项目的执行。

值得注意的是，handoff 作为工具被合并，集成到 run_agent_step 中的 agent 可执行工具中。如果值班 agent 决定将控制权转移给另一个 agent，则 handoff 方法在 Context 中定义 next_agent，并使用 DEFAULT_HANDOFF_OUTPUT_PROMPT 通知后续 agent 继续处理用户请求。

如果一个 agent 发现它无法处理用户的请求，它将使用 handoff 方法来转移控制权。图片来自作者

parse_agent_output 解释可执行工具；如果不再有工具，则 workflow 返回最终结果。否则，它启动并发执行。

call_tool 查找并执行特定工具的代码，将结果写入 ToolCallResult 并将其复制到输出流中。

aggregate_tool_results 整合工具调用结果，并检查是否执行了 handoff – 如果是，则切换到下一个值班 agent，重新启动该过程。否则，如果没有 handoff 或工具的 return_redirect 为 False，则重新启动。其他情况结束 Workflow，同时调用 agent 的 handle_tool_call_results 和 finalize 允许调整语言模型的结果。

除了标准的 Workflow 步骤方法外，AgentWorkflow 还包括一个 from_tools_or_functions 方法，便于理解名称。当使用 AgentWorkflow 作为独立 Agent 时，这将启动调用 FunctionAgent 或 ReActAgent，执行它们。这里有一个例子：

from tavily import AsyncTavilyClient

async def search_web(query: str) -> str:
    """Useful for using the web to answer questions"""
    client = AsyncTavilyClient()
    return str(await client.search(query))

workflow = AgentWorkflow.from_tools_or_functions(
    [search_web],
    system_prompt="You are a helpful assistant that can search the web for information."
)

在介绍了 AgentWorkflow 的基础知识后，我们现在将进入项目实践。为了提供直接的比较，该项目再次使用了之前文章中的客户服务示例，展示了 AgentWorkflow 开发的简单性。

基于 Agentworkflow 的客户服务项目实践

在之前的文章中，我演示了使用客户服务项目来展示 LlamaIndex Workflow 类似于 OpenAI Swarm 的多 agent 编排能力。

今天的项目使用 AgentWorkflow 来展示其开发的简易性，并使用相同的客户服务项目以获得清晰的理解。

最终效果

这是最终的项目展示：

该项目的最终效果。图片来自作者

如图所示，当用户提出请求时，系统会根据意图自动将其移交给相应的 agent。

接下来是核心代码。由于篇幅限制，这里只展示重要的代码；有关详细信息，请访问文章末尾的代码存储库。

定义 Agent

在多智能体客服项目中，我将创建一个新的 src_v2 文件夹，并修改 app.py 中的 sys.path 以重用之前创建的数据模型。

在之前的项目中，客户需求响应逻辑被写入 Workflow，这使得 workflow.py 变得笨重且难以维护。这次，ConciergeAgent、PreSalesAgent 和 PostSalesAgent 将真正处理客户服务，使用 AgentWorkflow 框架代码，不添加业务逻辑。

因此，一个新的 agents.py 文件定义了 concierge_agent、pre_sales_agent 和 post_sales_agent 智能体实例。

每个智能体都需要一个 name 和 description，它们至关重要，因为 AgentWorkflow 通过它们作为键值对来组织智能体，用于 handoff 引用，从而确定下一个智能体转换。

从 concierge_agent 开始，它检查用户是否注册了姓名——如果没有，它执行 login 工具进行注册；否则，根据意图，它决定是否将控制权转移给另外两个智能体。

concierge_agent = FunctionAgent(
    name="ConciergeAgent",
    description="An agent to register user information, used to check if the user has already registered their title.",
    system_prompt=(
        "You are an assistant responsible for recording user information."
        "You check from the state whether the user has provided their title or not."
        "If they haven't, you should ask the user to provide it."
        "You cannot make up the user's title."
        "If the user has already provided their information, you should use the login tool to record this information."
    ),
    tools=[login],
    can_handoff_to=["PreSalesAgent", "PostSalesAgent"]
)

然后是 pre_sales_agent，负责售前咨询。在收到请求后，它会审查聊天记录，根据查询结果查询 VectorIndex，并严格按照文档进行回复。如果用户不是询问售前问题，它会将控制权转移给另外两个智能体。

pre_sales_agent = FunctionAgent(
    name="PreSalesAgent",
    description="A pre-sales assistant helps answer customer questions about products and assists them in making purchasing decisions.",
    system_prompt=(
        "You are an assistant designed to answer users' questions about product information to help them make the right decision before purchasing."
        "You must use the query_sku_info tool to get the necessary information to answer the user and cannot make up information that doesn't exist."
        "If the user is not asking pre-purchase questions, you should transfer control to the ConciergeAgent or PostSalesAgent."
    ),
    tools=[query_sku_info],
    can_handoff_to=["ConciergeAgent", "PostSalesAgent"]
)

最后，post_sales_agent 处理有关产品使用的售后问题和政策。与 pre_sales_agent 一样，它只能基于现有文档进行回复，从而最大限度地减少大语言模型的误解。

post_sales_agent = FunctionAgent(
    name="PostSalesAgent",
    description="After-sales agent, used to answer user inquiries about product after-sales information, including product usage Q&A and after-sales policies.",
    system_prompt=(
        "You are an assistant responsible for answering users' questions about product after-sales information, including product usage Q&A and after-sales policies."
        "You must use the query_terms_info tool to get the necessary information to answer the user and cannot make up information that doesn't exist."
        "If the user is not asking after-sales or product usage-related questions, you should transfer control to the ConciergeAgent or PreSalesAgent."
    ),
    tools=[query_terms_info],
    can_handoff_to=["ConciergeAgent", "PreSalesAgent"]
)

使用 Chainlit 进行 UI 开发

由于不再需要 Workflow 逻辑，在开发完所有智能体后，可以直接开始 UI 开发，再次使用 Chainlit。

在 ready_my_workflow 中，初始化 AgentWorkflow 和 Context，同时在启动方法中将工作流和上下文实例存储在 user_session 中：

def ready_my_workflow() -> tuple[AgentWorkflow, Context]:
    workflow = AgentWorkflow(
        agents=[concierge_agent, pre_sales_agent, post_sales_agent],
        root_agent=concierge_agent.name,
        initial_state={
            "username": None
        }
    )
    ctx = Context(workflow=workflow)
    return workflow, ctx

@cl.on_chat_start
async def start():
    workflow, ctx = ready_my_workflow()
    cl.user_session.set("workflow", workflow)
    cl.user_session.set("context", ctx)

await cl.Message(
        author="assistant", content=GREETINGS
    ).send()

接下来，在 main 方法中，获取用户消息并调用工作流进行响应。提供了额外的代码来演示监视 AgentInput 和 AgentOutput 消息流；根据需要进行调整：

@cl.on_message
asyncdefmain(message: cl.Message):
    workflow: AgentWorkflow = cl.user_session.get("workflow")
    context: Context = cl.user_session.get("context")

handler = workflow.run(
        user_msg=message.content,
        ctx=context
    )
    stream_msg = cl.Message(content="")
    asyncfor event in handler.stream_events():
        ifisinstance(event, AgentInput):
            print(f"========{event.current_agent_name}:=========>")
            print(event.input)
            print("=================<")
        ifisinstance(event, AgentOutput) and event.response.content:
            print("<================>")
            print(f"{event.current_agent_name}: {event.response.content}")
            print("<================>")
        ifisinstance(event, AgentStream):
            await stream_msg.stream_token(event.delta)
    await stream_msg.send()

至此，我们的项目代码就完成了。AgentWorkflow 很好地封装了多智能体编排逻辑，使我们的 v2 版本更加专注，只需编写好智能体即可。

改进 FunctionAgent

运行我的项目代码时，您可能会注意到一些奇怪的地方：

该 agent 无法及时回复用户的请求，需要询问两次。图片来源：作者

系统正确识别了用户意图，并将其传递给下一个 agent，但后者没有立即响应，需要用户重复。

经过一系列调试，我找到了问题所在：接管的 agent 无法很好地追溯聊天记录以找到用户的请求。

因此，我尝试扩展 FunctionAgent 并修改了一些代码。经过一些调整后，agent 现在在收到交接后会立即响应，证明是有效的：

售后 agent 接管了用户的请求并立即回复。图片来源：作者

接下来，让我解释一下我是如何做的：

在原始的 FunctionAgent 方法中，每个 tool_call 请求和 tool_call 执行结果都作为一条消息保存在对话历史记录中。

例如，值班 agent 将控制权移交给下一个 agent 的操作需要将 handoff 方法的执行和 handoff 方法的执行结果保存为两条消息。

这导致下一个 agent 接管，而原始用户的请求消息已经在消息列表的前面很远，由于 token 限制很容易被踢出队列。

用户的请求由于 token 限制而被踢出。图片来源：作者

了解了这个问题，解决方案很简单：在 agent 转换后，将用户之前的请求重新定位到聊天历史记录的末尾。

当 AgentWorkflow 的 run_agent_step 开始时，FunctionAgent 中的 take_step 被调用。因此，在此处插入调整逻辑是理想的：

class MyFunctionAgent(FunctionAgent):
    @override
    async def take_step(
            self,
            ctx: Context,
            llm_input: List[ChatMessage],
            tools: Sequence[AsyncBaseTool],
            memory: BaseMemory,
    ) -> AgentOutput:
        last_msg = llm_input[-1] and llm_input[-1].content
        state = await ctx.get("state", None)
        print(f">>>>>>>>>>>{state}")
        if"handoff_result"inlast_msg:
            for message in llm_input[::-1]:
                if message.role == MessageRole.USER:
                    last_user_msg = message
                    llm_input.append(last_user_msg)
                    break

returnawaitsuper().take_step(ctx, llm_input, tools, memory)

如图所示，我反向迭代 chat_history，直到找到最近的用户请求消息，将其附加到 chat_history 的末尾。

可能出现的一个挑战是：如何仅在 agent 转换期间应用此操作，而绕过常规步骤？

之前，我们注意到 AgentWorkflow 在执行 handoff 方法后返回 handoff_output_prompt。后继 agent 的最新消息是这个 handoff_output_prompt。

因此，在 AgentWorkflow 初始化期间，我传入一个自定义的 handoff_output_prompt，类似于默认的，但预先用 "handoff_result" 标记：

def ready_my_workflow() -> tuple[AgentWorkflow, Context]:
    workflow = AgentWorkflow(
        agents=[concierge_agent, pre_sales_agent, post_sales_agent],
        root_agent=concierge_agent.name,
        handoff_output_prompt=(
            "handoff_result: Due to {reason}, the user's request has been passed to {to_agent}."
            "Please review the conversation history immediately and continue responding to the user's request."
        ),
        initial_state={
            "username": None
        }
    )
    ctx = Context(workflow=workflow)
    return workflow, ctx

因此，在 take_step 中，仅当消息包含 handoff_result 标签时，才会发生用户消息重新定位，从而有效地解决了问题。

结论

在当今日益丰富的多 agent 编排场景中，LlamaIndex 及时调整了其定位，并于上个月推出了 AgentWorkflow 框架，极大地简化了基于 LlamaIndex Workflow 的 agent 编排的开发。

在今天的文章中，我彻底解释了 AgentWorkflow 的原理，并通过客户服务项目实践说明了与仅使用 Workflow 相比，开发得到了简化。

虽然我相信 AgentWorkflow 使 LlamaIndex 的多 agent 解决方案接近完美，但该框架的最新发布意味着特定场景仍需要改进。我期待进一步的实践来增强它。继续前进，LlamaIndex！