❤️ 如果你也关注 AI 的发展现状，且对 AI 应用开发感兴趣，我会每日分享大模型与 AI 领域的开源项目和应用，提供运行实例和实用教程，帮助你快速上手AI技术！

🥦 AI 在线答疑 -> 智能检索历史文章和开源项目 -> 丰富的 AI 工具库 -> 每日更新 -> 尽在微信公众号 -> 搜一搜：蚝油菜花 🥦

🤖 「客服系统集体失业？开源框架让AI代理自组战队，人类只需看戏」

大家好，我是蚝油菜花。你是否被这些问题困扰——

• 👉 客服机器人死板：咨询转接5次仍解决不了问题
• 👉 企业系统烟囱林立：财务/HR/IT系统各自为战
• 👉 开发多AI协作系统：光是协调逻辑就写了800行代码...

今天揭秘的 Multi-Agent Orchestrator 正在引发智能协作革命！这个开源编排框架：

• ✅ 智能体自动组队：LLM分类器秒级匹配最佳代理，像滴滴派单般精准
• ✅ 动态热插拔机制：业务系统迭代无需停机，新增代理即插即用
• ✅ 军工级会话管理：10万级并发对话不串线，金融级数据加密

已有银行用它打造智能客服集群，游戏公司用它构建NPC生态圈——你的业务系统，是时候进化成「AI交响乐团」了！

🚀 快速阅读

Multi-Agent Orchestrator 是亚马逊开源的多智能体框架，能够动态分配代理、维护上下文、支持多种代理类型。

1. 核心功能：动态代理分配、上下文维护、支持多种代理类型。
2. 技术原理：基于 Orchestrator（编排器）、Classifier（分类器）、Agents（代理）等核心组件，实现智能代理的协同工作。

Multi-Agent Orchestrator 是什么

Multi-Agent Orchestrator 是用于管理和协调多个智能代理（Agent）的框架。通过分类器识别用户输入的意图，将请求分配给最适合的代理进行处理，通过对话存储保持上下文连贯性。支持多种类型的代理，如基于大语言模型（LLM）的代理和基于规则的代理，具有高度的灵活性和可扩展性。

用户输入后，分类器分析并选择合适的代理，代理处理请求并生成响应。整个对话过程会被记录在对话存储中，方便在多轮对话中保持连贯性。检索器会提供相关上下文信息，增强代理的性能。

Multi-Agent Orchestrator 的主要功能

• 动态代理分配：根据用户输入的上下文和意图，自动选择最适合的代理来处理请求。
• 支持多种代理类型：可以集成基于大语言模型（LLM）的代理、基于规则的代理、API调用代理等多种类型的代理，满足不同场景的需求。
• 代理生命周期管理：支持代理的动态加载、更新和卸载，便于系统扩展和维护。
• 上下文维护：通过对话存储功能，记录用户的输入和代理的响应，确保多轮对话的连贯性和一致性。
• 会话管理：支持多用户会话，能够区分不同用户的对话状态，避免混淆。
• 流式响应处理：支持异步流式响应，能够实时处理用户输入并逐步返回结果，提升用户体验。
• 智能分类器：对用户输入进行分析，快速识别最合适的代理，提高系统响应效率。
• 上下文检索：通过检索器提供相关上下文信息，帮助代理更好地理解用户意图，生成更准确的响应。
• 高度可扩展性：可以轻松添加新的代理类型或扩展现有代理的功能，适应复杂多变的应用场景。
• 与其他系统集成：支持与其他系统（如数据库、API服务等）集成，获取更多数据支持，增强代理的处理能力。
• 详细日志记录：记录代理之间的交互、分类器的输出以及用户的输入和响应，方便开发者进行调试和优化。
• 性能监控：提供性能监控功能，帮助开发者了解系统的运行状态，及时发现并解决问题。
• 生产级设计：具备高可用性和容错能力，适合在生产环境中使用。
• 安全机制：支持身份验证、授权和数据加密等安全机制，保护用户数据和隐私。

Multi-Agent Orchestrator 的技术原理

• Orchestrator（编排器）：作为核心组件，负责协调所有模块，管理信息流，确保请求被正确路由和处理。
• Classifier（分类器）：使用大型语言模型（LLM）分析用户输入、代理描述、对话历史和上下文，动态选择最适合处理当前请求的代理。
• Agents（代理）：处理具体任务并生成响应。代理可以是基于 LLM 的模型、API 调用、本地脚本或其他服务，每个代理都有其特定的技能和描述。
• Conversation Storage（对话存储）：用于维护对话历史，确保多轮对话的连贯性。支持多种存储方式，包括内存存储和 DynamoDB。
• Retrievers（检索器）：提供上下文和相关信息，帮助代理更好地理解用户意图。
• 协调机制：Multi-Agent Orchestrator 支持多种协调机制：
  • 集中式协调：由一个中央编排器分配任务并监控进度。
  • 分布式协调：代理之间通过协商分配角色和任务。
  • 混合模型：结合集中式和分布式的特点，既保留集中式的优势，又赋予代理一定的自主性。

如何运行 Multi-Agent Orchestrator

TypeScript 版本

安装

npm install multi-agent-orchestrator

使用

以下示例展示了如何使用 Multi-Agent Orchestrator 与两种不同类型的代理：支持 Converse API 的 Bedrock LLM 代理和 Lex Bot 代理。

import {
    MultiAgentOrchestrator, BedrockLLMAgent, LexBotAgent } from "multi-agent-orchestrator";

const orchestrator = new MultiAgentOrchestrator();

// 添加一个支持 Converse API 的 Bedrock LLM 代理
orchestrator.addAgent(
  new BedrockLLMAgent({
   
      name: "Tech Agent",
      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.",
      streaming: true
  })
);

// 添加一个 Lex Bot 代理用于处理旅行相关查询
orchestrator.addAgent(
  new LexBotAgent({
   
    name: "Travel Agent",
    description: "Helps users book and manage their flight reservations",
    botId: process.env.LEX_BOT_ID,
    botAliasId: process.env.LEX_BOT_ALIAS_ID,
    localeId: "en_US",
  })
);

// 示例使用
const response = await orchestrator.routeRequest(
  "I want to book a flight",
  'user123',
  'session456'
);

// 处理响应（流式或非流式）
if (response.streaming == true) {
   
    console.log("\n** RESPONSE STREAMING ** \n");
    // 立即发送元数据
    console.log(`> Agent ID: ${
     response.metadata.agentId}`);
    console.log(`> Agent Name: ${
     response.metadata.agentName}`);
    console.log(`> User Input: ${
     response.metadata.userInput}`);
    console.log(`> User ID: ${
     response.metadata.userId}`);
    console.log(`> Session ID: ${
     response.metadata.sessionId}`);
    console.log(
      `> Additional Parameters:`,
      response.metadata.additionalParams
    );
    console.log(`\n> Response: `);

    // 流式传输内容
    for await (const chunk of response.output) {
   
      if (typeof chunk === "string") {
   
        process.stdout.write(chunk);
      } else {
   
        console.error("Received unexpected chunk type:", typeof chunk);
      }
    }

} else {
   
    // 处理非流式响应（AgentProcessingResult）
    console.log("\n** RESPONSE ** \n");
    console.log(`> Agent ID: ${
     response.metadata.agentId}`);
    console.log(`> Agent Name: ${
     response.metadata.agentName}`);
    console.log(`> User Input: ${
     response.metadata.userInput}`);
    console.log(`> User ID: ${
     response.metadata.userId}`);
    console.log(`> Session ID: ${
     response.metadata.sessionId}`);
    console.log(
      `> Additional Parameters:`,
      response.metadata.additionalParams
    );
    console.log(`\n> Response: ${
     response.output}`);
}

Python 版本

安装

# 可选：设置虚拟环境
python -m venv venv
source venv/bin/activate  # 在 Windows 上使用 `venv\Scripts\activate`
pip install multi-agent-orchestrator[aws]

默认使用

以下是一个等效的 Python 示例，展示了如何使用 Multi-Agent Orchestrator 与 Bedrock LLM 代理和 Lex Bot 代理：

import os
import asyncio
from multi_agent_orchestrator.orchestrator import MultiAgentOrchestrator
from multi_agent_orchestrator.agents import BedrockLLMAgent, LexBotAgent, BedrockLLMAgentOptions, LexBotAgentOptions, AgentCallbacks

orchestrator = MultiAgentOrchestrator()

class BedrockLLMAgentCallbacks(AgentCallbacks):
    def on_llm_new_token(self, token: str) -> None:
        # 在这里处理响应流
        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.",
  model_id="anthropic.claude-3-sonnet-20240229-v1:0",
  callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(tech_agent)


# 添加一个 Lex Bot 代理用于处理旅行相关查询
orchestrator.add_agent(
    LexBotAgent(LexBotAgentOptions(
        name="Travel Agent",
        description="Helps users book and manage their flight reservations",
        bot_id=os.environ.get('LEX_BOT_ID'),
        bot_alias_id=os.environ.get('LEX_BOT_ALIAS_ID'),
        locale_id="en_US",
    ))
)

async def main():
    # 示例使用
    response = await orchestrator.route_request(
        "I want to book a flight",
        'user123',
        'session456'
    )

    # 处理响应（流式或非流式）
    if response.streaming:
        print("\n** RESPONSE STREAMING ** \n")
        # 立即发送元数据
        print(f"> Agent ID: {response.metadata.agent_id}")
        print(f"> Agent Name: {response.metadata.agent_name}")
        print(f"> User Input: {response.metadata.user_input}")
        print(f"> User ID: {response.metadata.user_id}")
        print(f"> Session ID: {response.metadata.session_id}")
        print(f"> Additional Parameters: {response.metadata.additional_params}")
        print("\n> Response: ")

        # 流式传输内容
        async for chunk in response.output:
            if isinstance(chunk, str):
                print(chunk, end='', flush=True)
            else:
                print(f"Received unexpected chunk type: {type(chunk)}", file=sys.stderr)

    else:
        # 处理非流式响应（AgentProcessingResult）
        print("\n** RESPONSE ** \n")
        print(f"> Agent ID: {response.metadata.agent_id}")
        print(f"> Agent Name: {response.metadata.agent_name}")
        print(f"> User Input: {response.metadata.user_input}")
        print(f"> User ID: {response.metadata.user_id}")
        print(f"> Session ID: {response.metadata.session_id}")
        print(f"> Additional Parameters: {response.metadata.additional_params}")
        print(f"\n> Response: {response.output.content}")

if __name__ == "__main__":
    asyncio.run(main())

资源

• GitHub 仓库：https://github.com/awslabs/multi-agent-orchestrator

❤️ 如果你也关注 AI 的发展现状，且对 AI 应用开发感兴趣，我会每日分享大模型与 AI 领域的开源项目和应用，提供运行实例和实用教程，帮助你快速上手AI技术！

🥦 AI 在线答疑 -> 智能检索历史文章和开源项目 -> 丰富的 AI 工具库 -> 每日更新 -> 尽在微信公众号 -> 搜一搜：蚝油菜花 🥦