多智能体协作与Strands架构深度解析

本文深入探讨了基于AWS Bedrock的多智能体协作系统，详细介绍了Arbiter模式的架构设计与实现。通过语义能力匹配、动态代理生成和黑板模型协调机制，实现了智能体的自适应协作与动态扩展，为构建大规模自主系统提供了完整解决方案。

多智能体协作与Strands | AWS DevOps与开发者生产力博客

在自主系统不断发展的背景下，多智能体协作不仅变得可行，而且日益必要。随着智能体获得更多能力（如高级推理、适应性和工具使用），挑战从个体性能转向有效协调。问题不再是“智能体能否完成任务？”，而是“如何在多个智能智能体之间组织执行？”

我们之前在关于使用Amazon Bedrock创建异步AI智能体的文章中介绍的Supervisor模式，为回答这个问题迈出了基础性一步。Supervisor通过充当集中协调器，在结构化的无服务器工作流中监控和委派任务，解决了第一代协调挑战。它在松散耦合的智能体之间提供异步编排、回退处理和状态跟踪，为组织提供了从单智能体原型转向多智能体系统的可靠方法。

然而，随着智能体系统规模扩大和变得更加动态，静态监督的局限性变得明显。Supervisor模型假设相对稳定的智能体集和可预测的工作流；但现代系统面临不断变化的任务、新兴能力和自适应协调的需求。这就是Arbiter模式作为自然演进出现的地方：一种下一代监督模型，通过动态智能体生成、语义任务路由和基于黑板模型的协调来扩展Supervisor。通过解决大型演进智能体生态系统的不可预测性和流动性，Arbiter模式使系统不仅能够管理复杂性，还能在其中蓬勃发展。

Arbiter模式通过添加三个关键功能直接构建于此：

  • 语义能力匹配：Arbiter不仅将已知任务分配给已知智能体，还会推理任务应该由哪种智能体处理——即使该智能体尚不存在。
  • 委托智能体创建：如果找不到合适的智能体，Arbiter将请求升级到Fabricator智能体，该智能体按需动态生成特定任务的智能体。这超越了委派，实现了真正的自适应生成。
  • 任务规划+上下文记忆：基于Supervisor的任务协调能力，Arbiter将复杂输入分解为结构化任务计划，并使用上下文记忆来跟踪执行、重试逻辑和智能体性能。

简而言之，Arbiter将静态编排转变为自适应协调。

黑板模型再探

为了实现跨智能体的松散、可扩展协作，Arbiter模式融合了黑板模型的原则——这是分布式AI中的经典架构。在该模型中，智能体机会主义地贡献到共享数据空间（“黑板”），响应变化并集体解决问题。

参考：参见“The Blackboard Model of Control”（Hayes-Roth等人）和早期应用如Hearsay-II以了解基础研究。

在我们扩展的Arbiter模式中，黑板成为语义事件基底。包括Arbiter在内的智能体发布和消费任务相关状态，实现松散耦合、事件驱动的协作。

工作原理

当事件进入系统时，Arbiter承担监督角色，但通过更大的动态性和适应性扩展它。与Supervisor模式类似，它首先解释事件并识别所需目标和子任务。然后执行能力评估，使用本地索引或对等发布的清单，类似于Supervisor查询Agents配置表。

  • 解释：Arbiter使用基于LLM的推理来提取任务目标和子任务。
  • 能力评估：使用本地索引或对等能力清单评估哪些智能体可以处理每个子任务。
  • 委派或生成：
    • 如果存在合适的智能体，任务相应路由。
    • 如果不存在，Arbiter向fabricator智能体发送生成请求。
  • 黑板协调：所有涉及的智能体读取/写入共享语义黑板，根据观察到的任务状态按需贡献。
  • 反思和适应：记录性能数据并用于通知未来的智能体创建、适应或弃用。

Arbiter模式架构

与通过静态配置列表维护编排的Supervisor不同，Arbiter引入了共享语义黑板，允许所有参与的智能体基于演进的任务状态读取、写入和协调。该黑板作为动态协作空间，实现任务中适应和更丰富的多智能体协调。

以下图表1：作为代码示例实现的Agentic AI Arbiter模式可在此处下载

图表1：Agentic AI Arbiter模式

以下序列描述了Arbiter模式，根据图表1：Agentic AI Arbiter模式中的编号步骤

  1. 进入系统的事件触发Supervisor函数
  2. Supervisor查询Agents Config表以获取智能体能力
  3. Supervisor使用Agents配置列表作为上下文来规划任务编排

选项：新智能体： 如果找不到有能力的智能体，Arbiter比基本监督模式更进一步：它向fabricator智能体发出生成请求，该智能体合成新的工作代码，存储以供运行时访问，并更新能力注册表，以便智能体系统可以立即受益于新技能。

  • 任务无法完成，请求创建新能力
  • 制造请求触发Fabrication智能体实例
  • Fabrication智能体查询资源注册表以获取可用工具（能力）
  • Fabricator生成工作智能体代码
  • 工作智能体代码存储在存储桶中供运行时访问
  • 新工作程序添加到Agents配置列表，包含智能体能力描述
  • 制造结果发布到消息总线

重复步骤1、2和3 选项：编排工作流： 如果存在合适的智能体，Arbiter通过调用适当的工作智能体来编排工作流，如Supervisor模型中一样跟踪进度和状态。

  • 任务编排存储用于跟踪端到端过程
  • 按名称/ID调用工作智能体的请求。为智能体调用添加工作流状态。
  • 调用工作智能体的请求触发工作智能体包装器实例
  • 工作智能体包装器加载智能体代码
  • 工作智能体推理并采取行动
  • 工作智能体向消息总线发送响应
  • Supervisor智能体更新工作流状态并跟踪编排

Arbiter包含反思和适应循环：记录来自任务执行的性能数据，分析并反馈到fabricator和协调逻辑中。这确保不仅即时完成任务，而且系统持续适应、淘汰性能不佳的智能体，并向更高效率演进。

Arbiter智能体：事件编排引擎

Supervisor智能体（Arbiter智能体）作为中央协调器组件，通过智能任务委派管理复杂的事件驱动工作流。

事件处理工作流： Arbiter模式遵循结构化方法处理传入事件

  • 配置加载：通过load_config_from_dynamodb()从Amazon DynamoDB加载可用智能体配置
  • LLM调用：使用事件上下文和可用工具规范调用Amazon Bedrock LLM
  • 决策分析：LLM分析事件并返回带有参数的工具调用决策
  • 任务分派：对于每个指定的工具调用：
    • 提取工具名称、输入参数和工具使用ID
    • 通过process_tool_call()将消息分派到相应的Amazon Simple Queue Service（SQS）队列
    • 维护工具调用列表用于工作流跟踪

工作流状态管理： 系统在整个执行过程中维护全面的状态跟踪

  • 使用create_workflow_tracking_record()在DynamoDB中创建工作流跟踪记录
  • 将所有调用的智能体初始化为未完成
  • 将唯一请求ID与编排实例关联
  • 持久化编排状态，包括对话历史和请求映射

完成协调： Arbiter通过系统过程协调任务完成

  • 事件接收：通过Amazon EventBridge接收智能体完成事件
  • 状态更新：使用update_workflow_tracking()更新工作流跟踪
  • 完成检查：在所有跟踪的智能体上执行完成检查
  • 结果聚合：当所有智能体完成时：
    • 从DynamoDB数据字段聚合结果
    • 将工具结果作为用户消息附加到对话中
    • 使用更新的上下文重新调用编排
  • 继续：继续直到LLM提供没有工具调用的最终响应

Fabricator智能体：动态能力生成

Fabricator智能体使用Strands智能体框架实现即时智能体开发，当系统中不存在所需功能时创建新能力。

智能体开发架构： Fabricator作为专门的Strands智能体运行，具有特定特征

  • 实现为具有专门系统提示以进行代码生成的Strands智能体
  • 由来自Arbiter的“新工作智能体”事件触发
  • 通过使用智能体指令增强提示接收能力需求
  • 系统提示包括：
    • Strands智能体实现示例
    • 可用Strands工具的完整目录
    • 代码生成模式和约定
    • 标准化的handler()函数要求

代码生成过程： 智能体遵循结构化的开发工作流

  • 需求分析：LLM分析能力需求并生成Python实现
  • 工具选择：优先使用现有Strands工具而非自定义@tool实现
  • 代码结构：创建遵循标准化模式的智能体：
    • 使用models.BedrockModel()进行Bedrock模型初始化
    • 具有适当工具选择的智能体实例化
    • 标准化的handler()函数接口
    • 事件驱动的完成信号
  • 文件创建：将生成的代码写入/tmp/目录以供立即可用

能力注册管道： 新能力通过多步过程注册

  • 文件存储：通过upload_file_to_s3()工具将文件上传到Amazon Simple Storage Service（S3）
  • 元数据注册：通过store_agent_config_dynamo()在DynamoDB中注册：
    • toolId：唯一能力标识符
    • filename：S3对象引用
    • schema：用于LLM工具调用的OpenAPI规范
    • description：人类可读的能力文档
    • action：用于Generic Wrapper的SQS队列路由配置
  • 完成通知：通过complete_task()工具向Arbiter发布完成事件

测试考虑： 原始实现揭示了关于测试方法的重要见解

  • 先前方法：在Fabricator内进行智能体测试导致：
    • 非结构化测试导致假阴性
    • 对生成智能体的过度优化
  • 推荐：具有标准化测试工具的单独测试智能体以进行验证反馈

通用包装器：动态执行运行时

通用包装器实现热加载模式，无需基础设施扩展即可实现无限智能体创建，为Fabricator生成的智能体提供通用执行环境。

这种热加载方法至关重要，因为它将能力增长与基础设施扩展解耦。不是为每个新智能体配置和维护新的基础设施组件（可能是数十甚至数百个智能体），系统重用单个执行包装器，可以动态加载和执行任意智能体代码。

这不仅使智能体创建实际上无限，而且确保基础设施效率、成本优化和简化操作，允许Arbiter和Fabricator演进系统能力而无需操作瓶颈。

在AWS示例代码中，热加载处理程序实现为AWS Lambda函数，表示在以下代码片段中：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

def process_event(event, context):
    orchestration_id = event["orchestration_id"]
    tool_use_id = event["tool_use_id"]
    request = event["tool_input"]
    tool_name = event['node']

    # 基于事件中的工具，从DDB加载详细信息
    tool = load_config_from_dynamodb(tool_name)
    config = tool['config']

    if isinstance(config, str):
        config = json.loads(config)

    file_name = config['filename']

    load_file_from_s3_into_tmp(os.environ["AGENT_BUCKET_NAME"], file_name)

    # 从tmp目录热加载模块
    spec = importlib.util.spec_from_file_location("module.name", "/tmp/loaded_module.py")
    loaded_module = importlib.util.module_from_spec(spec)
    sys.modules["module.name"] = loaded_module
    spec.loader.exec_module(loaded_module)

    # 使用Arbiter传递的任何参数调用通用处理程序
    try:
        print("attempting to use module")
        response = loaded_module.handler(**request)
        print(f"response: {response}")
    except Exception as e:
        print(f"error running module: {e}")


    # 最后，向Arbiter报告。由包装器处理。避免Frabricator尝试自行编写此部分
    post_task_complete(response, tool_use_id, tool_name, orchestration_id)

尽管此示例通过lambda函数演示，但热加载代码可以在Amazon Bedrock AgentCore Runtime或AWS原生容器服务（如Amazon Elastic Container Service（ECS）或Amazon Elastic Kubernetes Service（EKS））中执行。

热加载架构： 包装器实现几个关键架构原则

  • 单个基础设施组件处理所有动态创建的智能体的执行
  • 消除每个智能体单独基础设施配置的需求
  • 实现从S3存储的运行时代码加载
  • 在非超低延迟环境中接受延迟权衡以换取基础设施效率

动态加载过程： 系统遵循精确的加载序列

  • 消息处理：从传入SQS消息中提取智能体标识符
  • 配置检索：通过load_config_from_dynamodb()查询DynamoDB以获取智能体配置
  • 代码下载：从S3下载智能体实现到/tmp/目录
  • 运行时加载：使用importlib.util进行模块加载：
    • spec_from_file_location()创建模块规范
    • module_from_spec()实例化模块对象
    • exec_module()执行实际代码加载和执行

执行管理： 包装器提供全面的执行监督

  • 使用提供的参数调用标准化的handler()函数
  • 捕获执行结果并优雅处理错误条件
  • 维护不同智能体调用之间的执行隔离
  • 在智能体执行完成后实现资源清理

标准化通信协议： 通信遵循严格标准化以确保系统可靠性，这在多智能体环境中至关重要，其中数十甚至数百个动态生成的智能体可能交互。没有一致的消息格式、路由规则和完成信号，编排将变得脆弱，错误将不可预测地传播，调试几乎不可能。标准化保证每个智能体，无论何时创建，都可以无缝互操作，使Arbiter能够跨整个系统维护端到端可见性、可追溯性和容错性。

事件处理原则：

  • 事件发布由通用包装器专门处理，而非单个智能体
  • 确保跨所有智能体的一致事件驱动通信模式

完成事件结构：

  • orchestration_id：工作流上下文链接
  • tool_use_id：LLM工具调用映射
  • node：用于跟踪的智能体标识符
  • data：执行结果或错误信息

可靠性措施：

  • 将完成事件发布到EventBridge以供Arbiter处理
  • 保证工作流跟踪接收完成信号，无论执行结果如何

可扩展性特征： 热加载方法提供显著的可扩展性优势

  • 实现智能体扩展创建而基础设施影响最小
  • S3下载延迟在整体系统性能配置文件内可接受
  • 单个包装器实例可以执行多个智能体类型
  • 内存和资源管理在容器级别处理

结论

Arbiter模式代表了超越Supervisor架构的重要演进，提供了真正自主智能体系统所需的灵活性。通过引入语义丰富、上下文感知的编排，它实现了动态可扩展性，其中智能体能力随任务需求增长。该架构具有弹性，在智能体失败时重新分配或重新生成任务，并通过让智能体通过语义有意义的事件而非刚性API交互来实现松散耦合。最重要的是，它通过Arbiter引导的反馈循环嵌入持续适应，允许系统随时间学习和演进。这标志着从预编程逻辑到生成性、基于黑板模型的协调的转变，为去中心化、智能系统铺平了道路，这些系统可以大规模有效学习、适应和协作。

该系统提供几个关键能力

  • 异步处理：基于SQS的消息传递，用于可扩展执行
  • 持久状态管理（短期记忆）：基于DynamoDB的工作流跟踪
  • 可扩展性：用于无限智能体创建的热加载架构
  • 智能编排：LLM驱动的任务分解和排序
  • 自扩展能力：基于Strands的按需智能体创建
  • 标准化通信：可靠的事件驱动协议

该架构通过动态创建必要的处理能力并通过LLM驱动的工作流编排协调其执行，同时通过热加载模式维护基础设施效率，实现任意事件类型的处理。

comments powered by Disqus
© 2020 - 2025 qife
使用 Hugo 构建
主题 Stack 由 Jimmy 设计
本站总访问量次 本站访客数人次