AWS Multi-Agent Orchestrator：亚马逊开源的生产级多Agent编排框架

AI 应用正在从单 Agent 向多 Agent 架构演进，就如同人类组织一样，AI 面对的业务问题越复杂，专业分工是必然，这时，如何管理协调这些分工Agent就越重要。正是这一原因，各家公司都发布了自家的 multiAgent 框架来解决这一问题，如 OpenAI 的 Swarm，微软的 AutoGen 等。

延伸阅读：

Swarm：一个OpenAI开源的multiAgent框架，简单哲学的代表

一文探秘LLM应用开发(24)-Prompt(架构模式Agent)

CrewAI：一个集众家所长的MutiAgent框架（内有基于Ollama的本地模型应用CrewAI的视频教程）

Dify构建全AI雇员的BPO公司示例，揭示AI Native公司的创建方法

AutoGen 2.0:微软发布代码优先的Agent框架TaskWeaver

今天介绍的就是 AWS 近期推出的一个 MultiAgent 框架——AWS Multi-Agent Orchestrator^[1]，这是 AWS 在 Multi-Agent 领域的首次尝试，提供了一整套强大而优雅的多 Agent 系统编排解决方案，具备多项特色的功能，值得学习借鉴。

Multi-Agent Orchestrator 类似于一个复杂的交响乐队指挥。当用户发送请求时，分类器（指挥）使用 LLM 分析请求，考虑用户请求、可用Agent描述、完整对话历史和当前会话上下文。然后，将请求路由到最合适的Agent，Agent使用其对话上下文处理请求。响应返回后，对话历史自动保存。

架构上由以下模块构成：

• Orchestrator：协调所有组件，管理信息流。
• Classifier：分析用户输入，识别最合适的Agent。
• Agents：处理请求并生成响应。
• Conversation Storage：维护对话历史。
• Retrievers：提供上下文和相关信息，增强Agent性能。

与其他的框架相比，它有以下特点：

1. 灵活的 Agent 能力，它的每个 Agent 可由不同模型驱动，并可部署在任何地方。并且提供很多具有 AWS 特色的内置 Agent，这些 Agent 简化了开发者与 AWS 服务使用集成的成本，比如 Bedrock LLM Agent 支持带有内置护栏和工具使用能力的 Amazon Bedrock 模型 、Lex Bot Agent 可与与 Amazon Lex 机器人集成，Lambda Agent 能够支持对接其他 AWS 服务接口等。同时支持创建自定义 Agent，集成额外服务或系统。

2. 智能路由模式，它允许创建复杂的路由模式，比如成本优化路，他可以根据性能需求进行路由，比如，简单的路由到小模型，复杂的任务路由到复杂模型，以及专业任务路由到专业模型上。同时还无缝支持多语言。

3. 智能记忆管理，多 Agent 系统记忆管理是除路由外另一个重要问题就是记忆管理。该框架支持跨多次交互的连贯对话，并内置记忆存储实现，也支持自定义。

4. 强大的可观测性工具，对于一个 Agent 系统来讲，可观测性尤为重要，该框架包括全面的日志记录和监控分析功能，比如性能指标，执行时间等。同时，它提供了一个称作 Agent 重叠的分析工具（ Agent Overlap Analysis tool）旨在优化 MultiAgent 系统，能够专门帮助分析系统中冗余Agent、潜在路由冲突等多 Agent 系统特有的问题。

这是一个完整的例子：

import uuid
import asyncio
from typing import Optional, List, Dict, Any
import json
import sys
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator, OrchestratorConfig
from multi_agent_orchestrator.agents import (BedrockLLMAgent,
 BedrockLLMAgentOptions,
 AgentResponse,
 AgentCallbacks)
from multi_agent_orchestrator.types import ConversationMessage, ParticipantRole

orchestrator = MultiAgentOrchestrator(options=OrchestratorConfig(
  LOG_AGENT_CHAT=True,
  LOG_CLASSIFIER_CHAT=True,
  LOG_CLASSIFIER_RAW_OUTPUT=True,
  LOG_CLASSIFIER_OUTPUT=True,
  LOG_EXECUTION_TIMES=True,
  MAX_RETRIES=3,
  USE_DEFAULT_AGENT_IF_NONE_IDENTIFIED=True,
  MAX_MESSAGE_PAIRS_PER_AGENT=10
))

#添加Agent
class BedrockLLMAgentCallbacks(AgentCallbacks):
    def on_llm_new_token(self, token: str) -> None:
        # handle response streaming here
        print(token, end='', flush=True)

tech_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
  name="Tech Agent",
  streaming=True,
  description="Specializes in technology areas including software development, hardware, AI, \
  cybersecurity, blockchain, cloud computing, emerging tech innovations, and pricing/costs \
  related to technology products and services.",
  callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(tech_agent)

health_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
  name="Health Agent",
  description="Focuses on health and medical topics such as general wellness, nutrition, diseases, treatments, mental health, fitness, healthcare systems, and medical terminology or concepts.",
  callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(health_agent)

# 发送消息

async def handle_request(_orchestrator: MultiAgentOrchestrator, _user_input: str, _user_id: str, _session_id: str):
    response: AgentResponse = await _orchestrator.route_request(_user_input, _user_id, _session_id)
    # Print metadata
    print("\nMetadata:")
    print(f"Selected Agent: {response.metadata.agent_name}")
    if response.streaming:
        print('Response:', response.output.content[0]['text'])
    else:
        print('Response:', response.output.content[0]['text'])


if __name__ == "__main__":
    USER_ID = "user123"
    SESSION_ID = str(uuid.uuid4())
    print("Welcome to the interactive Multi-Agent system. Type 'quit' to exit.")
    while True:
        # Get user input
        user_input = input("\nYou: ").strip()
        if user_input.lower() == 'quit':
            print("Exiting the program. Goodbye!")
            sys.exit()
        # Run the async function
        asyncio.run(handle_request(orchestrator, user_input, USER_ID, SESSION_ID))

小结

AWS Multi-Agent Orchestrator 相较于 OpenAI 的 MultiAgent 框架，其完成度明显上了一个台阶，并且是面向生产设计的，适用于构建复杂且高效的 AI 系统。特别是对于本身就在 AWS 上部署应用的开发者来说，是一个非常值得考虑的选择。

参考资料