2.1K+ Star！Multi-Agent Orchestrator：一个灵活且强大的多Agent调度框架

欢迎关注我，持续获取更多内容，感谢赞&在看~

Multi-Agent Orchestrator 简介

Multi-Agent Orchestrator^[1] 是一个灵活且强大的框架，用于管理多个AI代理并处理复杂对话。支持智能意图分类、双语言支持、灵活的代理响应、上下文管理、可扩展架构和通用部署。

2.1K+ Star！Multi-Agent Orchestrator：一个灵活且强大的多Agent调度框架

项目特点

主要特点

智能意图分类：基于上下文和内容动态地将查询路由到最合适的代理。
双语言支持：完全用Python和TypeScript实现。
灵活的代理响应：支持来自不同代理的流式和非流式响应。
上下文管理：跨多个代理维护和利用对话上下文以实现连贯的交互。
可扩展架构：轻松集成新代理或自定义现有代理以满足特定需求。
通用部署：可在任何地方运行——从AWS Lambda到本地环境或任何云平台。
预构建的代理和分类器：提供多种现成的代理和多种分类器实现。

使用场景

多智能体协调器适用于从简单的聊天机器人到复杂的AI系统，适应多样化的需求并有效扩展。

项目使用

高级架构流程图

流程从用户输入开始，由分类器分析。
分类器利用代理特征和代理对话历史来选择最合适的任务代理。
一旦选择了代理，它就处理用户输入。
协调器然后保存对话，更新代理对话历史，然后将响应返回给用户。

安装与使用

项目提供了TypeScript和Python版本的安装和使用示例，以下展示了如何添加和使用不同类型的代理

1. TypeScript 版本

安装

npm install multi-agent-orchestrator

使用示例

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

const orchestrator = new MultiAgentOrchestrator();

// Add a Bedrock LLM Agent with Converse API support
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
  })
);

// Add a Lex Bot Agent for handling travel-related queries
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",
  })
);

// Example usage
const response = await orchestrator.routeRequest(
  "I want to book a flight",
  'user123',
  'session456'
);

// Handle the response (streaming or non-streaming)
if (response.streaming == true) {
    console.log("n** RESPONSE STREAMING ** n");
    // Send metadata immediately
    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: `);

    // Stream the content
    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 {
    // Handle non-streaming response (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}`);
}

2. Python 版本

安装

# Optional: Set up a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows use `venvScriptsactivate`
pip install multi-agent-orchestrator

使用示例

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:
        # 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.",
  model_id="anthropic.claude-3-sonnet-20240229-v1:0",
  callbacks=BedrockLLMAgentCallbacks()
))
orchestrator.add_agent(tech_agent)


# Add a Lex Bot Agent for handling travel-related queries
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():
    # Example usage
    response = await orchestrator.route_request(
        "I want to book a flight",
        'user123',
        'session456'
    )

    # Handle the response (streaming or non-streaming)
    if response.streaming:
        print("n** RESPONSE STREAMING ** n")
        # Send metadata immediately
        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: ")

        # Stream the content
        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:
        # Handle non-streaming response (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())