Agent 作为工具模式

2026-04-16 AI Agent 编程 3.4k 字 12 分钟阅读
📑 目录
  1. 概述
  2. Manager-Worker 架构的任务分片与结果聚合策略
  3. 正文
  4. 相关阅读
  5. 参考文档
  6. 完整实战示例：并行研究助手与结果聚合
  7. 常见问题与调试
  8. 与其他方案对比
  9. Manager-Worker 架构的观察者模式与背压控制
  10. 生产环境部署与性能优化
    1. 任务队列集成的实践要点
    2. Worker 健康状态的关键指标
    3. 弹性扩缩容的架构考量
    4. 运维团队的协作建议

概述

Manager-Worker 架构的任务分片与结果聚合策略

Agent-as-Tool 模式本质上是一种分层委托架构：顶层 Manager 负责将复杂任务分解为可管理的子任务，每个子任务由一个专门的 Worker Agent 处理，最后 Manager 将各 Worker 的结果聚合为最终答案。

这种架构的工程挑战在于**任务分片（Sharding）和结果聚合（Aggregation）**的质量：

  1. 任务分片：如果 Manager 将任务划分得过粗，Worker 可能无法处理；如果划分得过细，协调开销会超过并行带来的收益。理想的分片应该满足"独立性"（子任务之间尽可能少依赖）和"同质性"（相似复杂度的子任务应该分组处理）。

  2. 结果聚合：当多个 Worker 返回结果后，Manager 需要决定如何整合它们。常见的聚合策略包括：

    • 串行拼接：按固定顺序连接各结果（适合报告生成）。
    • 去重合并：识别并合并重复信息（适合多源信息检索）。
    • 冲突解决：当 Worker 结果矛盾时，由 Manager 仲裁（适合事实核查）。
    • 投票机制：多个 Worker 对同一问题独立回答，取多数意见（适合需要高可靠性的场景）。

从并发模型来看，as_tool() 的调用是同步阻塞的：Manager 调用 Worker，等待其返回后才能继续。这与真正的异步任务队列（如 Celery）不同——如果某个 Worker 执行很慢，Manager 会被阻塞。对于需要高度并行的场景，可以考虑在 Manager 层显式使用 asyncio.gather 同时调用多个 Worker。

另一个需要注意的设计细节是失败传播。如果某个 Worker 执行失败（如工具异常、LLM 调用失败），默认行为是将错误信息返回给 Manager。Manager 需要具备处理部分失败的能力——是忽略失败的 Worker、重试、还是将失败信息纳入最终输出。

Manager-Specialist 模式：使用 Agent.as_tool() 构建分层多智能体系统。

正文

相关阅读

参考文档

完整实战示例：并行研究助手与结果聚合

以下示例展示了如何使用 Agent-as-Tool 构建一个并行研究系统，Manager 同时分派多个研究任务，然后综合各专家的观点生成最终报告：

import asyncio
from agents import Agent, Runner


# Worker 1：技术趋势研究员
tech_researcher = Agent(
    name="TechResearcher",
    instructions="""
你是技术趋势研究专家。给定一个技术主题  #分析：
1. 当前主流实现方案
2. 关键技术挑战
3. 未来3年发展趋势
输出不超过200字  #使用 bullet points。
""".strip(),
    model="gpt-5-nano")

常见问题与调试

问题一：Worker 输出格式不一致导致聚合困难

当多个 Worker 使用不同的格式和风格输出时，Manager 可能难以有效整合。解决方案：

  1. 为所有 Worker 制定统一的输出模板（如强制 JSON 或 Markdown 结构）。
  2. 在 Manager 的 instructions 中提供各 Worker 输出的解析示例。
  3. 使用 custom_output_extractor 对 Worker 输出进行标准化清洗。

问题二：Worker 之间的信息冗余

不同 Worker 可能从各自角度重复提到相同的信息（如某项技术的主要厂商）。优化方法：

  1. 在 Manager 层实现去重逻辑，或使用 LLM 进行摘要去重。
  2. 为 Worker 划分更明确的职责边界，减少重叠。
  3. 使用共享上下文（如预先检索的公共信息）避免各 Worker 重复做相同的基础工作。

问题三：部分 Worker 失败导致整体质量下降

如果 3 个 Worker 中有 1 个失败，Manager 可能基于不完整的信息做出错误判断。应对措施：

  1. 在聚合输入中明确标注哪些来源缺失或失败。
  2. 实现降级策略：如果关键技术 Worker 失败，则拒绝生成报告而不是基于猜测输出。
  3. 对关键 Worker 实现重试机制，提高成功率。

与其他方案对比

维度Agents SDK as_tool()CrewAI ProcessLangGraph Map-Reduce
并行能力需手动 gather有限（按角色顺序）原生支持 Map-Reduce
结果聚合自定义逻辑通过任务回调内置 Reduce 节点
失败处理需自行实现基础支持节点级重试和降级
适用规模中小规模（<10 Worker）中小团队大规模流水线

LangGraph 的 Map-Reduce 模式是目前最成熟的并行任务处理方案，它内置了任务分发、结果收集和容错机制，适合构建复杂的数据处理流水线。CrewAI 的 Process 更侧重于"谁做什么"的角色分工，并行能力相对有限。Agents SDK 的 as_tool() 提供了最底层的控制，适合需要精细定制并行策略和聚合逻辑的场景。

Manager-Worker 架构的观察者模式与背压控制

在 Manager-Worker 分层系统中，Manager 不仅要派发任务，还需要实时感知各 Worker 的执行状态，以便在失败时及时降级或重试。这种需求天然适合观察者模式（Observer Pattern）：Worker Agent 作为被观察的主题（Subject），Manager 作为观察者（Observer），通过事件回调机制实现状态的异步通知和解耦。当 Worker 的执行状态发生变化时（开始执行、进度更新、执行完成、发生错误），所有注册的观察者都会收到通知，从而做出相应的处理决策。
mermaid
sequenceDiagram
participant M as Manager Agent
participant O as 任务编排器
participant W1 as Worker A
participant W2 as Worker B
participant W3 as Worker C

M ->> O: 提交并行任务
O ->> W1: 异步调用 as_tool()
O ->> W2: 异步调用 as_tool()
O ->> W3: 异步调用 as_tool()
W1 -->> O: 进度事件: 30%
W2 -->> O: 完成事件 + 结果
W3 -->> O: 错误事件: 超时
O -->> M: 聚合通知
M ->> O: 对 W3 发起重试
O ->> W3: 第二次调用
W3 -->> O: 完成事件 + 结果
O -->> M: 最终聚合结果


观察者模式的核心优势在于**松耦合**。Manager 不需要轮询每个 Worker 的状态，也不需要与 Worker 建立直接的依赖关系；两者通过事件总线或编排器间接通信。这种架构使得 Worker 的增删变得极为简单，新增一个 Worker 只需将其注册到事件系统中即可，Manager 的代码完全无需改动。同时，事件驱动的设计也为后续扩展提供了便利，例如可以很容易地接入日志系统、监控告警和审计追踪。

然而，单纯的并行调用可能导致**背压（Backpressure）**问题：当 Manager 持续高速派发任务，而某个 Worker 因网络延迟或模型限流响应缓慢时，待处理的任务队列会无限增长，最终耗尽内存甚至触发级联故障。解决背压需要在编排层引入**令牌桶限流（Token Bucket）**和**超时熔断（Circuit Breaker）**机制。令牌桶控制任务提交速率，防止瞬间流量洪峰冲垮下游；熔断器在错误率超过阈值时快速失败，给故障服务留出恢复时间。

以下是一个带背压控制的并行任务编排器实现：

```python
import asyncio
from dataclasses import dataclass
from typing import Callable, Any

@dataclass
class TaskEvent:
    task_id: str
    worker_name: str
    status: str  # "started", "progress", "completed", "failed"
    payload: Any = None

class BackpressureOrchestrator:
    """带背压控制的并行任务编排器，基于信号量限制并发数。"""
    
    def __init__(self, max_concurrent: int = 3, rate_limit: float = 10.0):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.rate_limit = rate_limit
        self._last_submit = 0.0
        self._lock = asyncio.Lock()
        self.observers: list[Callable[[TaskEvent], None]] = []
    
    def subscribe(self, callback: Callable[[TaskEvent], None]):
        """Manager 注册观察者回调。"""
        self.observers.append(callback)
    
    def _notify(self, event: TaskEvent):
        for cb in self.observers:
            cb(event)
    
    async def submit(self, worker: Callable, task_id: str, *args) -> Any:
        async with self._lock:
            now = asyncio.get_event_loop().time()
            elapsed = now - self._last_submit
            if elapsed < 1.0 / self.rate_limit:
                await asyncio.sleep(1.0 / self.rate_limit - elapsed)
            self._last_submit = now
        
        async with self.semaphore:
            self._notify(TaskEvent(task_id, worker.__name__, "started"))
            try:
                result = await worker(*args)
                self._notify(TaskEvent(task_id, worker.__name__, "completed", result))
                return result
            except Exception as e:
                self._notify(TaskEvent(task_id, worker.__name__, "failed", str(e)))
                raise

在设计大规模 Manager-Worker 系统时，还应遵循扇出-扇入模式（Fan-out/Fan-in）：Manager 将任务扇出到多个 Worker 并行处理，再通过扇入阶段统一收集和聚合结果。asyncio.gather 是实现扇入的最简单方式，但对于可能部分失败的场景，应使用 asyncio.gather(*tasks, return_exceptions=True) 避免单点失败导致整体崩溃。这种容错设计是生产系统稳定运行的关键保障。更进一步，可以为不同类型的 Worker 配置不同的超时策略和重试次数，对于关键路径上的 Worker 给予更多的容错机会，而对于非关键 Worker 则可以快速失败，优先保证整体响应速度。

在分布式链路追踪方面，建议为每个任务分配唯一的 trace_id，并在所有 Worker 调用中透传该标识。这样当系统出现问题时，可以通过链路追踪工具（如 Jaeger、Zipkin）快速定位瓶颈节点。同时，结合结构化日志（如 JSON 格式日志）和集中式日志收集（如 ELK 堆栈），可以构建完整的可观测性体系，让系统的每一个状态转换都留下可追溯的痕迹。这对于调试复杂的并行任务执行流程尤为关键，因为并发场景下的问题往往具有间歇性和难以复现的特点。

生产环境部署与性能优化

任务队列集成的实践要点

将本章节的技术应用到生产环境时，首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略：先在开发环境验证核心逻辑，再迁移到预发布环境进行压力测试，最后才全量上线。部署过程中应配置完善的日志收集和指标监控，确保任何问题都能被快速发现和定位。

具体来说，需要在基础设施层面做好以下准备：容器资源限制（CPU/内存）、网络策略配置（防火墙规则、服务网格）、持久化存储选型（SSD vs 标准盘）以及备份恢复方案。对于高可用要求严格的场景，建议部署多实例并配置负载均衡，避免单点故障导致服务中断。

Worker 健康状态的关键指标

监控是生产系统的生命线。针对本章节涉及的功能，建议重点跟踪以下指标：请求延迟（P50/P95/P99）、错误率（4xx/5xx/超时）、吞吐量（QPS/TPS）以及资源利用率（CPU/内存/磁盘/网络）。这些指标应接入统一的监控大盘，并设置合理的告警阈值。

除了基础指标，还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析，可以更准确地评估系统改进的实际价值，避免陷入"为了优化而优化"的陷阱。

弹性扩缩容的架构考量

随着业务规模增长，单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量，而非事后补救。水平扩展通常比垂直扩展更具成本效益，但也引入了分布式系统的复杂性（数据一致性、服务发现、负载均衡等）。

在扩展过程中，建议遵循"无状态优先"原则：将状态外置到独立的存储层（如 Redis、PostgreSQL），使计算层可以随时水平扩容。对于无法避免的状态（如会话、缓存），采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试，确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册（Runbook），涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时，通过定期的复盘会议，将线上事故转化为团队的学习素材，持续完善系统的健壮性。

在工具链方面，推荐将本章节的配置和脚本纳入版本控制（Git），并使用 Infrastructure as Code（IaC）工具（如 Terraform、Ansible）管理基础设施变更。这不仅能提高部署效率，还能确保环境一致性，减少"在我机器上能跑"的问题。

Manager-Worker 模式的关键在于任务分解的合理性。如果子任务粒度过大，Worker 可能长时间占用资源；如果粒度过小，协调开销会超过并行收益。建议通过实验找到最优分片大小。

结果聚合的质量决定了 Manager Agent 的最终输出。如果 Worker 返回的结果存在矛盾，Manager 需要具备冲突检测和仲裁能力。可以在 Manager 的提示词中明确给出仲裁规则（如以数据源 A 为准）。