Agent Squad：AWS Labs 开源多代理编排框架，智能处理复杂对话与团队协作

一、Agent Squad是什么？

Agent Squad是AWS Labs开发的一款灵活、轻量的开源框架，前身为Multi-Agent Orchestrator，专注于实现多个AI代理（Agents）的高效协作与任务编排，以处理复杂对话场景。它支持Python和TypeScript双语言开发，具备智能意图分类、动态代理路由、上下文维护、流式响应等核心能力，同时提供可扩展架构和预构建组件，适用于客户支持、电商服务、内容创作等多种场景，帮助开发者快速构建从简单聊天机器人到复杂AI协作系统的应用。

简单来说，Agent Squad是一个“AI代理的指挥中心”：它能接收用户输入，分析需求意图，自动分配给最合适的专业代理（如客服代理、技术支持代理、翻译代理等），并协调多个代理协作完成任务，同时记录对话历史以保证交互连贯性。无论是简单的单轮问答，还是需要多代理分工的复杂流程（如“用户咨询订单→转物流代理查询→返回结果→客服代理总结”），Agent Squad都能高效处理。

作为开源项目，Agent Squad的代码完全公开，开发者可自由修改、扩展或集成到现有系统中，且支持在本地环境、AWS Lambda、其他云平台等多种环境部署，兼顾灵活性与实用性。

二、功能特色

Agent Squad的核心价值在于“让多代理协作更简单、更智能”，其功能特色可概括为以下7点，通过表格可更清晰呈现核心模块的价值：

1. 智能意图分类：让“对的代理”做“对的事”

传统多代理系统中，用户查询往往需要手动指定代理（如“请转技术支持”），而Agent Squad的意图分类器能自动完成这一过程。它会结合两方面信息决策：一是代理特征（每个代理预先定义的擅长领域，如“订单查询代理”擅长处理物流、付款问题）；二是对话上下文（用户历史提问、已提供的信息）。例如：

• 用户输入“我的订单还没到”，分类器会识别“订单”“物流”关键词，将查询路由到“订单查询代理”；

• 若用户接着问“怎么申请退款”，分类器结合历史上下文（已知用户在处理订单问题），自动路由到“售后退款代理”。

2. 双语言支持：覆盖主流技术栈

Agent Squad同时提供Python和TypeScript两种语言的原生实现，且两者功能完全对齐，不存在“某语言功能缺失”的问题：

• Python版本：发布在PyPI（包名agent-squad），支持Python 3.11+，集成了LangChain等主流AI框架工具，适合数据科学、后端开发团队；

• TypeScript版本：发布在npm（包名agent-squad），提供完整TypeScript类型定义，支持ES6+语法，适合前端、全栈开发团队，可直接集成到React、Vue等前端项目中。

3. 上下文管理：让代理“记住”对话历史

在多代理协作中，“上下文断裂”是常见问题（如A代理处理后，B代理不知道之前的对话内容）。Agent Squad通过统一的上下文存储机制解决这一问题：

• 对话历史以结构化格式（如JSON）存储，包含用户输入、代理响应、时间戳等信息；

• 当查询从A代理流转到B代理时，编排器会自动将完整上下文传递给B代理，确保B代理“了解前情”；

• 支持上下文过期策略（如只保留最近10轮对话），避免数据冗余影响性能。

4. SupervisorAgent：实现“代理即工具”的团队协作

Agent Squad新增的SupervisorAgent是其核心增强功能，它扮演“团队管理者”的角色，支持“代理即工具”架构——即每个专业代理可被视为一个“工具”，由SupervisorAgent统一调度，实现更复杂的协作场景：

• 并行处理：同时调用多个代理处理不同子任务（如“分析用户反馈”时，同时调用情感分析代理和关键词提取代理）；

• 动态任务分配：根据任务进度调整代理分工（如“撰写产品文案”时，先让大纲代理生成结构，再让内容代理填充细节，最后让校对代理检查）；

• 层级系统构建：支持嵌套结构（如“客户支持团队”包含一级客服代理、二级技术代理、三级专家代理，由SupervisorAgent根据问题复杂度升级处理）。

三、技术细节

1. 整体架构

Agent Squad的架构采用“分层设计”，核心流程可分为“输入→分析→处理→输出”四步，具体如下：

![架构流程图（文字描述）]
用户输入 → 分类器（或SupervisorAgent）→ 选择目标代理 → 代理处理输入 → 编排器（保存上下文）→ 返回响应给用户

• 用户输入层：接收文本输入（未来计划支持语音、图像等多模态输入），格式为字符串或结构化数据（如包含用户ID、会话ID的JSON）；

• 分析层：由分类器或SupervisorAgent组成，负责决策“谁来处理”。分类器适用于简单路由场景，SupervisorAgent适用于复杂协作场景；

• 处理层：由多个专业代理组成，每个代理包含“能力定义”（擅长领域）和“处理逻辑”（调用AI模型、API等完成任务）；

• 存储层：负责保存对话上下文，支持内存存储（本地测试）、Redis（分布式场景）、AWS DynamoDB（云环境）等多种存储方式；

• 输出层：将代理处理结果格式化后返回，支持流式输出（通过SSE或WebSocket）和非流式输出（JSON/文本）。

2. 核心组件详解

（1）分类器（Classifier）

分类器是决定“查询路由”的核心组件，其工作原理基于“规则+模型”：

• 规则分类：通过关键词匹配、正则表达式等预定义规则路由（如含“退款”关键词的查询路由到售后代理）；

• 模型分类：当规则无法覆盖时，调用AI模型（如GPT-3.5/4、AWS Bedrock模型）分析意图，返回最匹配的代理ID。

开发者可通过配置文件自定义分类逻辑，例如：

# Python示例：分类器规则配置 
classifier_rules = [ 
  {"keywords": ["订单", "物流", "收货"], "agent_id": "order_agent"}, 
  {"keywords": ["退款", "退货", "发票"], "agent_id": "refund_agent"}, 
  {"default": True, "agent_id": "general_agent"} # 默认路由到通用代理 
]

（2）代理（Agent）

每个代理是一个独立的功能单元，需实现标准化接口（handle方法），接收“输入文本+上下文”，返回“处理结果”。代理可分为两类：

• 基础代理：直接调用AI模型处理任务（如客服代理调用LLM生成回复）；

• 工具代理：调用外部工具（API、数据库、函数）完成任务（如订单代理调用电商平台API查询订单状态）。

示例：一个简单的Python代理实现

from agent_squad import BaseAgent 

class OrderAgent(BaseAgent): 
  def handle(self, input_text: str, context: dict) -> str: 
    # 调用订单查询API 
    order_id = extract_order_id(input_text) # 从输入中提取订单号 
    order_status = order_api.query(order_id) # 调用外部API 
    return f"您的订单{order_id}当前状态：{order_status}"

（3）编排器（Orchestrator）

编排器是整个框架的“胶水”，负责串联分类器、代理和上下文存储：

• 接收用户输入并传递给分类器；

• 根据分类结果调用目标代理；

• 保存代理响应到上下文存储；

• 格式化响应并返回给用户。

其核心逻辑确保流程自动化，开发者无需手动干预代理间的协作。

（4）SupervisorAgent

作为高级组件，SupervisorAgent具备“任务规划”和“代理调度”能力，内部包含：

• 任务分解器：将复杂任务拆分为子任务（如“用户想了解产品A的价格和优惠”拆分为“查询价格”和“查询优惠”）；

• 代理选择器：为每个子任务匹配最合适的代理；

• 结果整合器：汇总子任务结果，生成最终响应。

例如，处理“分析某产品的用户评价并生成改进建议”时，SupervisorAgent的流程为：

1. 分解任务为“提取评价关键词”“分析情感倾向”“生成建议”；

2. 分别调用“关键词提取代理”“情感分析代理”“建议生成代理”；

3. 整合三个代理的结果，输出结构化报告。

3. 技术依赖与兼容性

Agent Squad的轻量特性体现在其低依赖设计上，核心依赖如下：

四、应用场景

Agent Squad的灵活性使其适用于从简单到复杂的多种AI交互场景，以下为典型案例：

1. 智能客户支持系统

场景需求：企业客户支持需处理订单查询、售后退款、技术故障等多类问题，传统单一客服机器人难以覆盖所有领域。
Agent Squad解决方案：

• 部署“订单代理”“退款代理”“技术支持代理”三个专业代理；

• 配置分类器：通过关键词（如“订单号”“退货”“无法连接”）路由查询；

• 当用户问题涉及多个领域（如“我的订单没收到，想退款”），SupervisorAgent协调“订单代理”查询状态后，自动转“退款代理”处理；

• 上下文管理确保用户无需重复提供订单号等信息。

2. 电商全流程服务

场景需求：电商平台需提供商品推荐、库存查询、物流跟踪、投诉处理等一站式服务。
Agent Squad解决方案：

• 构建代理矩阵：推荐代理（调用推荐算法）、库存代理（查询数据库）、物流代理（对接物流API）、投诉代理（生成工单）；

• 利用流式响应：推荐代理实时返回“为您推荐以下商品：1.xxx 2.xxx”，提升用户等待体验；

• 部署在AWS Lambda：通过云函数触发代理处理，降低服务器成本，支持高并发（如促销活动期间）。

3. 内容创作协作平台

场景需求：自媒体团队需要快速生成文章初稿、进行语法校对、翻译多语言版本。
Agent Squad解决方案：

• 由SupervisorAgent作为“主编”，协调三个代理：

• 初稿代理：根据主题生成文章框架和内容；

• 校对代理：检查语法错误和逻辑漏洞；

• 翻译代理：将校对后的内容翻译成英文、日文等；

• 支持“人工介入”：当代理处理结果不符合要求时，用户可手动调整，新内容会更新到上下文，供后续代理参考。

4. 企业内部智能助手

场景需求：企业员工需要查询IT设备申请流程、HR政策（如年假规则）、财务报销标准等内部信息。
Agent Squad解决方案：

• 定制“IT代理”“HR代理”“财务代理”，分别对接企业内部知识库；

• 分类器结合员工身份（如部门、职位）路由查询（如“财务代理”只响应财务部门员工的报销问题）；

• 部署在企业内网：上下文存储使用内部Redis，确保数据安全，避免敏感信息泄露。

五、使用方法

Agent Squad的使用流程简单直观，以下分别以Python和TypeScript为例，演示从安装到运行的完整步骤。

1. Python版本使用步骤

步骤1：安装依赖

通过PyPI安装agent-squad包：

pip install agent-squad

步骤2：定义代理

创建两个示例代理：通用代理（处理简单问题）和订单代理（处理订单查询）。

from agent_squad import BaseAgent, Orchestrator, RuleBasedClassifier 

# 通用代理：处理问候、感谢等简单交互 
class GeneralAgent(BaseAgent): 
  def handle(self, input_text: str, context: dict) -> str: 
    greetings = ["你好", "哈喽", "hi"] 
    if any(g in input_text for g in greetings): 
      return "您好！请问有什么可以帮您的？" 
    return "我会尽力为您解答，请详细说明您的问题。" 

# 订单代理：处理订单相关查询（模拟调用API） 
class OrderAgent(BaseAgent): 
  def handle(self, input_text: str, context: dict) -> str: 
    # 模拟从输入中提取订单号 
    order_id = "ORD12345" # 实际场景中可通过正则提取 
    # 模拟查询结果 
    return f"您的订单{order_id}已发货，预计3天后送达。"

步骤3：配置分类器

定义分类规则，让分类器知道“哪些问题该交给哪个代理”。

# 分类规则：含“订单”关键词的查询路由到OrderAgent，其余到GeneralAgent 
classifier = RuleBasedClassifier([ 
  {"keywords": ["订单", "物流", "收货"], "agent_id": "order_agent"}, 
  {"default": True, "agent_id": "general_agent"} 
])

步骤4：初始化编排器并运行

将代理和分类器注册到编排器，处理用户输入。

# 注册代理：key为agent_id，value为代理实例 
agents = { 
  "general_agent": GeneralAgent(), 
  "order_agent": OrderAgent() 
} 

# 创建编排器 
orchestrator = Orchestrator(agents=agents, classifier=classifier) 

# 处理用户输入 
user_input1 = "你好，我的订单什么时候到？" 
response1 = orchestrator.process(user_input1) 
print(response1) # 输出：您的订单ORD12345已发货，预计3天后送达。 

user_input2 = "谢谢！" 
response2 = orchestrator.process(user_input2) 
print(response2) # 输出：我会尽力为您解答，请详细说明您的问题。（通用代理响应）

2. TypeScript版本使用步骤

步骤1：安装依赖

通过npm安装agent-squad包：

npm install agent-squad

步骤2：定义代理

import { BaseAgent, Orchestrator, RuleBasedClassifier } from 'agent-squad'; 

// 通用代理 
class GeneralAgent extends BaseAgent { 
 handle(inputText: string, context: Record<string, any>): string { 
  const greetings = ["你好", "哈喽", "hi"]; 
  if (greetings.some(g => inputText.includes(g))) { 
   return "您好！请问有什么可以帮您的？"; 
  } 
  return "我会尽力为您解答，请详细说明您的问题。"; 
 } 
} 

// 订单代理 
class OrderAgent extends BaseAgent { 
 handle(inputText: string, context: Record<string, any>): string { 
  const orderId = "ORD12345"; 
  return `您的订单${orderId}已发货，预计3天后送达。`; 
 } 
}

步骤3：配置分类器与运行

// 分类规则 
const classifier = new RuleBasedClassifier([ 
 { keywords: ["订单", "物流", "收货"], agentId: "order_agent" }, 
 { default: true, agentId: "general_agent" } 
]); 

// 注册代理 
const agents = { 
 general_agent: new GeneralAgent(), 
 order_agent: new OrderAgent() 
}; 

// 创建编排器并处理输入 
const orchestrator = new Orchestrator({ agents, classifier }); 

const userInput1 = "你好，我的订单什么时候到？"; 
const response1 = orchestrator.process(userInput1); 
console.log(response1); // 输出：您的订单ORD12345已发货，预计3天后送达。

3. 进阶使用：SupervisorAgent示例

以下为Python示例，演示如何用SupervisorAgent协调多代理处理复杂任务：

from agent_squad import SupervisorAgent, BaseAgent 

# 子代理1：提取文本关键词 
class KeywordAgent(BaseAgent): 
  def handle(self, input_text: str, context: dict) -> str: 
    return f"关键词：{input_text.split()[:3]}" # 取前3个词作为关键词 

# 子代理2：分析文本情感 
class SentimentAgent(BaseAgent): 
  def handle(self, input_text: str, context: dict) -> str: 
    return "情感：正面" # 模拟分析结果 

# 创建SupervisorAgent，定义任务分解逻辑 
supervisor = SupervisorAgent( 
  agents={"keyword_agent": KeywordAgent(), "sentiment_agent": SentimentAgent()}, 
  task_decomposer=lambda input_text: [ # 分解任务为两个子任务 
    {"task": "提取关键词", "agent_id": "keyword_agent"}, 
    {"task": "分析情感", "agent_id": "sentiment_agent"} 
  ] 
) 

# 处理输入 
result = supervisor.handle("这款产品很好用，推荐购买！", {}) 
print(result) 
# 输出：整合结果：\n1. 关键词：['这款', '产品', '很好用']\n2. 情感：正面

六、常见问题解答（FAQ）

1. Agent Squad与其他多代理框架（如LangChain Agents、AutoGPT）有什么区别？

• 定位不同：LangChain Agents更侧重“单一代理调用工具”，AutoGPT侧重“自主代理完成长任务”，而Agent Squad专注“多代理协作与任务分配”；

• 轻量性：Agent Squad无强依赖，部署简单，适合快速集成到现有系统；

• 企业级支持：由AWS Labs维护，兼容性更好（如无缝对接AWS Lambda、Bedrock等服务）。

2. 支持哪些AI模型？是否只能用AWS的模型？

Agent Squad不绑定特定AI模型，支持所有主流LLM（如OpenAI GPT系列、Anthropic Claude、AWS Bedrock模型、开源模型如Llama 3等）。开发者只需在代理的handle方法中调用对应模型API即可，例如：

# 调用OpenAI模型的代理示例 
import openai 

class GPTAgent(BaseAgent): 
  def handle(self, input_text: str, context: dict) -> str: 
    response = openai.ChatCompletion.create( 
      model="gpt-3.5-turbo", 
      messages=[{"role": "user", "content": input_text}] 
    ) 
    return response.choices[0].message.content

3. 如何自定义分类器？除了规则和模型，还有其他方式吗？

支持三种分类器自定义方式：

• 规则分类器：通过关键词、正则表达式定义（适合简单场景）；

• 模型分类器：调用LLM生成代理ID（适合复杂意图）；

• 混合分类器：先尝试规则分类，失败后自动切换到模型分类（平衡效率与准确性）。

示例：模型分类器（Python）

from agent_squad import BaseClassifier 

class ModelBasedClassifier(BaseClassifier): 
  def classify(self, input_text: str, context: dict) -> str: 
    # 调用LLM生成最匹配的agent_id 
    prompt = f"用户输入：{input_text}，请从['order_agent', 'refund_agent']中选一个最合适的代理ID" 
    response = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}]) 
    return response.choices[0].message.content.strip()

4. 部署到AWS Lambda需要注意什么？

• 内存配置：建议至少512MB内存，避免处理大上下文时超时；

• 冷启动优化：将代理初始化逻辑放在Lambda的全局代码中（而非handler函数内），减少冷启动时间；

• 上下文存储：Lambda是无状态的，需使用外部存储（如DynamoDB）保存对话上下文，而非内存存储。

5. 如何处理代理间的冲突（如两个代理对同一问题给出不同答案）？

可通过以下方式解决：

• 在SupervisorAgent中添加“结果校验器”，对比多代理结果，选择置信度最高的响应；

• 配置“优先级代理”，当冲突时以高优先级代理的结果为准；

• 回退到人工处理，将冲突结果返回给用户，让用户选择正确答案。

七、相关链接

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

• 官方文档：https://awslabs.github.io/agent-squad/

八、总结

Agent Squad作为AWS Labs推出的开源多代理编排框架，以轻量、灵活、跨语言为核心优势，通过智能意图分类、上下文管理、SupervisorAgent等功能，实现了多个AI代理的高效协作，解决了单一代理难以处理复杂对话场景的问题。无论是客户支持、电商服务等企业级应用，还是内容创作、内部助手等轻量化场景，开发者都能通过其预构建组件快速上手，并基于可扩展架构进行定制化开发。其双语言支持和跨环境部署能力，进一步降低了集成门槛，使其成为连接多AI代理与实际业务需求的实用工具。