当前位置：首页 > news >正文

AgentWorkflow 实战：从单 Agent 到多 Agent 协作的完整方案

news 2025/10/7 8:56:06

引言：为什么要有 AgentWorkflow？

随着越来越多 AI 应用不再只是“单问单答”——它们需要跨多个子任务（检索、分析、写作、校验、推理、反馈等）协同工作——我们开始遇到一些共性挑战：

各子模块之间手动编排逻辑（handoff）容易出错、难扩展
上下文与状态管理变乱：子任务之间如何共享、中断、恢复
不透明执行：难以看到在背后发生了什么步骤
有些任务必须有人类介入（审校、确认、安全阈值判断等）

LlamaIndex 的 AgentWorkflow 就是为应对这些挑战而设计的。它基于其底层的 Workflows 架构，提供更高层级的抽象：把多个 agent（每个 agent 专注于某些职责）组织起来、管理状态、协调执行流程，并允许 human-in-the-loop 的中断与恢复。简而言之，它是“多 agent 协作 + 可控交互”的基础设施。

下文我会先从基础入门（单 agent + FunctionAgent）讲起，然后讲 multi-agent 架构与 human-in-the-loop 机制，最后讨论在工程实践中常见的注意事项与扩展思路。

基础篇：一个 AgentWorkflow 是什么样子？

Workflows 的基础回顾

在深入 AgentWorkflow 之前，要理解它的底层依赖：Workflows。在 LlamaIndex 中，Workflow 是一种事件驱动（event-driven）的抽象：你可以把一个复杂任务拆成多个 “步骤（steps）”，每个步骤对某些事件感兴趣、接收事件、处理逻辑、再发出新的事件。这样就能构建非线性、可中断、可并行、可回环的流程。

Workflows 本身支持：

状态（state）在步骤间传递
事件的监听与发射
可观察性（instrumentation, 日志、监控）
中断、暂停与恢复（用于 human-in-the-loop）

AgentWorkflow 是建立在这个基础上的一个“专用于 agent 协作”的封装：它把 agent、工具调用、子任务切换、状态传递等常见模式内建进去，让使用者更专注于 agent 的职责定义，而非流程调度本身。

单 AgentWorkflow 示例（FunctionAgent）

最简单的情形：用一个 agent 来完成任务。LlamaIndex 提供了 AgentWorkflow.from_tools_or_functions(...) 方法，让你可以快速用工具或函数包装成一个 agent workflow。文档示例中这样写：

from llama_index.core.agent.workflow import AgentWorkflow
# 假设有一个异步工具 search_web(query: str) -> strworkflow = AgentWorkflow.from_tools_or_functions(tools=[search_web],llm=llm,system_prompt="You are a helpful assistant that can search the web for information."
)response = await workflow.run(user_msg="What is the latest news about AI?")

背后发生什么？AgentWorkflow 会：

把输入包装成一个 AgentInput 事件
调用 agent 的 take_step，让 agent 决定是否调用工具或直接回答
若有工具调用，则发出 ToolCall 事件，由执行引擎触发工具并产生 ToolCallResult 事件
Agent 会根据工具输出继续（handle_tool_call_results）或最终生成回答
AgentWorkflow 汇集这些事件流，向上层呈现最终结果

这个流程你可以通过 .stream_events() 观察详细事件：AgentInput、AgentOutput、ToolCall、ToolCallResult 等等。

这个“基础 agent”用例适合一些不太复杂的任务或者作为起点。但它无法分隔职责（research / writing / review 等）。

深入篇：多 Agent 协作与 Handoff 机制

真正体现 AgentWorkflow 强大的，是它支持多个 agent 协同——也就是 multi-agent 模式。每个 agent 专注某类职责，系统根据任务上下文决定哪个 agent 接管。

多 Agent 架构概念

在 AgentWorkflow 中，你提供一个 agent 列表（agents）和一个 root_agent（初始接手 agent 名称）。从用户输入开始，root_agent 处理或直接 handoff 给其他 agent，直至某个 agent 给出最终回答。

每个 agent 必须定义：

name（字符串标识）
description（用途说明）
system_prompt 或其他提示，以便 LLM 知道本 agent 的职责
tools（可调用工具）
can_handoff_to 列表（哪些 agent 它可以将控制权交给）

示例（简化版）：

from llama_index.core.agent.workflow import FunctionAgent, AgentWorkflowresearch_agent = FunctionAgent(name="ResearchAgent",description="搜集与分析资料",system_prompt="你是一个研究专家，负责找信息和整理笔记",tools=[search_web, record_notes],can_handoff_to=["WriteAgent"]
)write_agent = FunctionAgent(name="WriteAgent",description="撰写报告",system_prompt="你是一个写作专家，把研究笔记写成完整报告",tools=[write_report],can_handoff_to=["ReviewAgent"]
)review_agent = FunctionAgent(name="ReviewAgent",description="校验与润色报告",system_prompt="你是校验专家，检查逻辑、事实和语言表达",tools=[proofread, fact_check],can_handoff_to=[]
)  # 最后一个 agent 通常不给出 handoffworkflow = AgentWorkflow(agents=[research_agent, write_agent, review_agent],root_agent=research_agent.name,initial_state={"research_notes": {},"report_content": "","review_notes": ""}
)handler = workflow.run(user_msg="请给我一份关于 XX 的报告。")
async for event in handler.stream_events():# 这里可以打印或处理每个事件：AgentInput／AgentOutput／ToolCall／ToolCallResult 等...

LlamaIndex 的官方 multi-agent 示例就是类似的架构，用于生成报告的任务。

事件流监控与可视化

在 multi-agent 模式下，事件流尤为重要，因为你希望看到各 agent 之间交替执行、工具调用、handoff、输出等细节。通过 handler.stream_events()，你可以在实时或前端 UI 中展示这些事件。上面的示例展示如何在控制台打印不同类型事件。

事件类型包括但不限于：

AgentInput：输入给当前 agent 的上下文
AgentOutput：agent 对输入生成的回应（可能包含下步 tool 调用建议）
ToolCall / ToolCallResult：工具调用与返回
AgentStream：LLM 生成的 token 流（增量输出）
Handoff（隐含在 AgentOutput 或 tool 调用中）：切换控制权

这种事件驱动视图让系统透明、可调试，也便于在前端做 UI 可视化（如进度条、日志流等）。

AgentWorkflow 内部关机机制（handoff 逻辑）

在某个 agent 的执行过程中，它可以（通过内部工具或逻辑）决定将控制权交给另一个 agent，这就是 handoff 机制。AgentWorkflow 内部会检测 “handoff 指令” 并自动做切换，不需要外部编排。

一般流程：

当前 agent 在 take_step 或 handle_tool_call_results 时，基于状态与上下文判断：是否需 handoff
若 handoff，agent 会发出相应事件（如调用某个 handoff 工具）
AgentWorkflow 看到 handoff，设置下一个 agent 为当前执行者，继继续处理

设计良好的 handoff 机制使模块边界清晰，每个 agent 只专注自己的职责，而无需感知整体调度逻辑。

高阶篇：Human-in-the-Loop（人类在环）机制

自动 agent 虽然强大，但在很多场景中，我们需要插入“人工确认 / 审校 / 补充输入”——这就是 human-in-the-loop（HITL）范式。AgentWorkflow / Workflows 本身原生支持这种机制。

基本思路：通过事件暂停 & 等待人类响应

在 LlamaIndex 的 Workflows 文档中，有一个 “Human in the loop” 模式介绍：你可以定义某个工具在执行时发出 InputRequiredEvent，然后暂停，等待 HumanResponseEvent 来恢复。

核心步骤是：

工具内部在某一点发出 wait_for_event(...)，等待 HumanResponseEvent
在这个暂停期间，Workflow（AgentWorkflow）将暂停下来，不继续下一个步骤
外部系统（UI、前端、人工接口）监听到 InputRequiredEvent，展示给人类以供回应
人类输入 / 选择后，通过 HumanResponseEvent 传回 Workflow，继续执行

例如，一个“危险操作确认”工具可能写成：

from llama_index.core.workflow import InputRequiredEvent, HumanResponseEvent, Contextasync def dangerous_tool(ctx: Context) -> str:question = "Are you sure you want to proceed (yes/no)?"# 发出中断信号，等待用户输入response = await ctx.wait_for_event(HumanResponseEvent,waiter_id=question,waiter_event=InputRequiredEvent(question))if response.user_response.lower().startswith("y"):return "OK"else:raise RuntimeError("Operation aborted by user")

当这个工具在 agent 的步骤中被调用时，系统会 pause，直到用户回答为止。

LlamaIndex 的 Human-in-the-loop 文档里就有这个机制的说明。

示例：故事生成 + 人工选择路径

LlamaIndex 文档有一个 “Choose Your Own Adventure” 的 Workflow 示例：AI 生成故事段落 + 多个选项，然后人类从中选一个继续下去。这个就是 human-in-the-loop 的典型用例。

在那例子里：

LLM 生成一个剧情片段 + 多个可选动作（选项）
Workflow 发出 InputRequiredEvent，等待人类从这些选项中选一条
人类选完之后给出 HumanResponseEvent，继续进入下一段生成

这在交互式创作、自动问答系统带反馈环节、决策辅助系统等场景中非常有用。

在 AgentWorkflow 中使用 HITL

因为 AgentWorkflow 基于 Workflows，所以同样可以在 agent 的工具里插入 “等待人类输入” 的逻辑。在 multi-agent 系统里，这能用于：

在 agent 决定 handoff 前征求人工确认
在 agent 输出最终回答前让人类审校
在流程中插入人工校验、增补

LlamaIndex 的 AgentWorkflow 架构并未阻止这种做法。你在定义工具或 agent 步骤时，就可以嵌入 wait_for_event(...) 逻辑来控制中断与恢复。

比如，在 write_agent 完成本次报告初稿之后，插入一步让人类校验是否满意：

async def confirm_draft(ctx: Context) -> bool:question = "请确认这份初稿是否可接受（yes/no）"resp = await ctx.wait_for_event(HumanResponseEvent,waiter_id=question,waiter_event=InputRequiredEvent(question))return resp.user_response.lower().startswith("y")

在 agent 的逻辑里就可以判断：如果确认通过，则继续；否则可能派回 research_agent 继续深度检索或让 write_agent 修改。

中断、持久化与跨请求恢复

在实际线上系统中，人类的输入可能不在同一个请求周期内（网络延迟、页面关闭、异步接口等）。因此必须把 Workflow 状态持久化，以便在用户输入后继续执行。LlamaIndex 支持这种“暂停 + 恢复”的机制。文档与相关教程里展示了如何在 REST 接口里保存 workflow 状态、序列化、再恢复。

举例流程：

接口 A 接受用户初次请求，启动 AgentWorkflow.run，执行到某个 InputRequiredEvent 停止
接口 A 序列化 workflow 状态（包括内部 event queue、上下文 state 等），存入数据库
前端展示给用户选择／填写的界面
用户提交回答后，请求到接口 B，取回之前保存的 workflow 状态，恢复执行，并注入 HumanResponseEvent
继续后续 agent 执行，直到 workflow 完成

通过这种方式，就可以在分布式系统、异步系统中安全地嵌入 human-in-the-loop。

工程思考、挑战与扩展方向

AgentWorkflow 虽强大，但在实际工程化中有一些注意点与扩展思路值得关注。

常见挑战与解决建议

handoff 时 agent 无法及时响应
有用户在社区反映：某个 agent 接收到 handoff 后似乎没能马上 “记住”上下文，反应迟缓。文章作者调试后指出，是因为历史对话没有妥善传递到新 agent 的 prompt 里。
对策：在 state 或上下文里显式维护对话历史，并在每个 agent 的 prompt 中注入必要上下文。
过度 agent 拆分导致管理复杂性
虽然拆分职责带来清晰性，但 agent 数量太多、交互边界多，就可能导致调试困难、handoff 混乱。建议：按职责粒度拆分，不要细化到每一个微操作都用 agent。
状态膨胀与冗余
state 对象要精简，避免把大量中间数据、冗余副本塞进去。对于大型数据（如全文文档、图像、嵌入向量），可只引用 ID 或摘要。
性能 / 异步并发
agent 执行与工具调用可能并发，注意使用 async/await。对于长任务（如大型检索、外部 API 请求），要合理做超时与重试。
可观测性与监控
事件流虽然透明，但在生产系统中要做好日志聚合、错误追踪、指标采集。推荐把关键事件（agent 切换、错误、工具失败）接入监控系统。
回滚 / 容错 / Undo 机制
在多 agent 协作中，如果某一步出现异常（如工具调用失败、人工拒绝、agent crash），要有恢复或重试机制。可在 Workflow 中用异常捕获、状态检查、重启逻辑等。

扩展方向与未来思考

动态代理生成 / discovery agent
在一些更灵活的系统里，并不是所有 agent 在初始化就固定，而是根据任务动态“发现”或生成 agent，这需要 AgentWorkflow 支持动态 agent 注入、卸载。
智能路由 / 自治决策 agent
用一个 router agent 或调度 agent 来判断哪个专职 agent 承担任务，而不是在 root_agent 中硬编码 handoff 规则。
混合 agent 类型
除了 FunctionAgent、ReActAgent，或许还有 MemoryAgent、PlannerAgent 等角色，它们在同一 workflow 中协作。
跨 workflow 协作 / 嵌套 workflow
一个 agent 内部可能启动子 workflow（子任务流水线），这需要 AgentWorkflow 支持“workflow 嵌入”或“子流程”机制。
可解释性 / 可审计的 agent 决策轨迹
利用事件流记录所有 agent 决策节点、工具调用、handoff 路径，生成决策树／可视化路径，便于审计与回溯。
人类反馈闭环 / 在线学习
人类在环反馈后，不仅继续流程，还能把审校结果用于改进 agent prompt、调整策略、更新模型参数等。
与其它 agent 框架（如 LangGraph）比较或交互
有一些文章讨论了 LlamaIndex Workflow 与 LangGraph 在 human-in-the-loop 的差别：LlamaIndex 更擅长事件驱动、可视化、暂停恢复；LangGraph 强在状态图、复杂分支逻辑。

总结：AgentWorkflow 是一种理想“中层架构”

AgentWorkflow 并不是解决所有问题的灵丹妙药，但它在“让你专注写 agent”与“系统化调度”之间找到了一个很好的折中。

它把 agent 协作、hand-off、状态管理、事件可视化、human-in-the-loop 等功能封装起来，降低开发复杂度
它建立在 Workflows 之上，保留了流程的灵活性、可中断性与可观察性
它的设计思想适合未来越来越复杂的 agent 系统：多个子 agent、人工协同、可扩展性