随着多智能体系统成为对话式 AI 的新趋势，如何协调多个专精 AI 的分工协作，成为构建下一代智能客服、规划助手、企业 Copilot 的关键挑战。
最近在 GitHub 发现了一个轻量级高灵活性的多 AI 智能体协调框架：Agent-Squad。
它可以实现多智能体之间的智能协作、对话路由和上下文共享。
该框架支持 Python 和 TypeScript 两种语言，开发者可以根据喜好选择。它还能部署在AWS Lambda、本地环境甚至其他云平台上，灵活性拉满。
你可以把它想象成一个超级聪明的“任务分配器”，能根据你的输入，自动把问题交给最合适的AI智能体。
核心功能
-
• 智能意图分类：能根据用户输入的内容、上下文和智能体描述，动态路由到最合适的智能体。 -
• 双语言支持：同时支持 Python 和 TypeScript，满足不同开发者的偏好。无论你是后端大牛还是前端新手，都能快速上手。 -
• 灵活的代理响应：支持来自不同代理的流式和非流式响应。 -
• 上下文管理： 在多个代理之间维护和利用对话上下文，以实现连贯的交互。 -
• 可扩展架构：它的模块化设计可以轻松集成新智能体或定制现有功能。
工作机制
1、从用户输入开始，由分类器进行分析。
2、分类器利用代理的特征和代理的对话历史来选择最适合任务的代理。
3、一旦选择了代理，它就会处理用户输入。
4、协调器随后保存对话，更新代理的对话历史，然后将响应返回给用户。
Agent-Squad 现在包括一个强大的新监督代理，它能够实现多个专业代理之间的复杂团队协调。这个新组件实现了“代理即工具”的架构，允许主代理并行协调一组专业代理，保持上下文并给出连贯的响应。
快速使用
Agent-Squad 实际安装也非常友好。对于开发者来说，简直是So Easy！
如果你是前端工程师，建议使用 npm 命令直接安装：
npm install agent-squad以下示例展示了支持Converse API的Bedrock LLM代理和Lex Bot代理。
import { AgentSquad, BedrockLLMAgent, LexBotAgent } from "agent-squad";
const orchestrator = new AgentSquad();
// 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}`);
}如果你是后端工程师，也会用 Python，可以使用 pip 命令直接安装：
pip install agent-squad[aws]以下是一个等效的Python示例，演示了使用Bedrock LLM Agent和Lex Bot Agent的Agent Squad：
import sys
import asyncio
from agent_squad.orchestrator import AgentSquad
from agent_squad.agents import BedrockLLMAgent, BedrockLLMAgentOptions, AgentStreamResponse
orchestrator = AgentSquad()
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",
))
orchestrator.add_agent(tech_agent)
health_agent = BedrockLLMAgent(BedrockLLMAgentOptions(
name="Health Agent",
streaming=True,
description="Specializes in health and well being",
))
orchestrator.add_agent(health_agent)
async def main():
# Example usage
response = await orchestrator.route_request(
"What is AWS Lambda?",
'user123',
'session456',
{},
True
)
# 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:
async for chunk in response.output:
if isinstance(chunk, AgentStreamResponse):
print(chunk.text, 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())以上代码示例主要作用：
-
• 使用支持Converse API的Bedrock LLM代理，允许进行多轮对话。 -
• 集成Lex Bot代理以执行专业任务（例子中是与旅行相关的查询）。 -
• 协调者根据输入将请求路由到最合适的代理的能力。 -
• 处理来自不同类型代理的流式和非流式响应。
同时Agent-Squad也采用了模块化架构设计，允许仅安装所需的组件，同时确保始终获得核心功能。
# AWS集成（包括核心编排功能以及全面的AWS服务集成（BedrockLLMAgent、AmazonBedrockAgent、LambdaAgent等）
pip install "agent-squad[aws]"
# Anthropic集成
pip install "agent-squad[anthropic]"
# OpenAI集成
pip install "agent-squad[openai]"
# 完整安装
pip install "agent-squad[all]"写在最后
Agent-Squad 不只是一个框架，它是你构建下一代多智能体系统的 “中枢神经”。轻量灵活、开箱即用，非常适合 AI 原型开发、RAG 系统升级、多角色对话系统等项目。
而且它的高灵活性能无缝融入各种场景，智能客服、旅行规划、学习助手、职场帮手、创意开发等都能找到它的用武之地。
它让多个AI智能体像团队一样分工协作，精准高效地完成任务，让我们能把时间留给更重要的事！
如果你是个爱折腾的技术控，可以去部署试试！
GitHub 项目地址：https://github.com/awslabs/agent-squad
（文：开源星探）