菜单
智能任务分解和多 Agent 编排系统（增强版 v2.1）。支持两种模式：(1) 并行分析模式 - 多维度代码审查；(2) 层级执行模式 - 主 Agent 统筹，子 Agent 实际干活。预期收益：重复分析速度 5-10x，成本降低 30-60%，成功率提升 20%。适用于：复杂项目分析、多维度代码审查、大规模任务执行、综合性研究任务
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
| name | agent-swarm |
| description | 智能任务分解和多 Agent 编排系统（增强版 v2.1）。支持两种模式：(1) 并行分析模式 - 多维度代码审查；(2) 层级执行模式 - 主 Agent 统筹，子 Agent 实际干活。预期收益：重复分析速度 5-10x，成本降低 30-60%，成功率提升 20%。适用于：复杂项目分析、多维度代码审查、大规模任务执行、综合性研究任务 |
| version | 2.1.0 |
| created | "2025-11-14T16:00:00.000Z" |
| last_updated | "2026-01-31T07:00:00.000Z" |
| maintainer | claude-evolution |
| status | active |
层级结构：
用户大任务
↓
主 Agent（项目经理）
├─ 分析需求
├─ 制定计划
├─ 拆解 TODO
└─ 分配任务
↓
子 Agent 1 子 Agent 2 子 Agent 3
(后端开发) (前端开发) (测试)
├─ 写代码 ├─ 写组件 ├─ 写测试
├─ 修改文件 ├─ 写样式 ├─ 运行测试
└─ 报告进度 └─ 报告进度 └─ 报告进度
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 重复分析速度 | 基准 | 5-10x | 缓存命中 |
| PR review 速度 | 基准 | 3-5x | 增量分析 |
| 成本 | 100% | 40-70% | 智能选择 |
| 成功率 | 基准 | +20% | 自动重试 |
| 角色覆盖 | 16 个 | 40+ 个 | 扩展库 |
当面对复杂任务时，自动分解为多个子任务，为每个子任务创建专门的 Agent（带有角色名称），并行执行以提升效率。
当用户请求包含以下特征时自动激活：
分析用户请求，识别：
根据任务类型，自动生成 Sub-Agent 配置：
代码分析任务：
数据分析任务：
项目研究任务：
新增场景支持：
使用 Task 工具并行启动所有 Sub-Agents：
在单个消息中发送多个 Task 调用
每个 Task 包含：
- description: Agent 角色名称
- prompt: 具体任务描述
- subagent_type: Explore/Plan/Bash 等
收集所有 Agent 的输出，生成综合报告：
用户请求：
请全面审查这个项目的代码
Agent Swarm 响应：
我将启动 4 个专门的 Agent 并行分析：
1. Code Quality Analyst - 分析代码质量和可维护性
2. Security Auditor - 检查安全漏洞和风险
3. Performance Engineer - 分析性能瓶颈
4. Documentation Reviewer - 检查文档完整性
[并行启动 4 个 Task Agents]
用户请求：
分析这个数据集的所有维度
Agent Swarm 响应：
我将创建 3 个专门的 Agent：
1. Data Explorer - 探索数据结构和分布
2. Statistical Analyst - 进行统计分析
3. Quality Checker - 检查数据质量问题
[并行启动 3 个 Task Agents]
最大 Agent 数量
Agent 类型选择
超时设置
智能缓存
use_cache=True（默认）.agent_swarm_cache增量分析
use_incremental=True（默认）动态角色生成
use_dynamic_roles=True（可选）智能模型选择
自动重试
# 基础使用
orchestrator = AgentOrchestrator(
max_agents=8,
use_cache=True,
use_incremental=True
)
# 启用所有增强功能
orchestrator = AgentOrchestrator(
max_agents=10,
use_cache=True,
use_incremental=True,
use_dynamic_roles=True
)
# 分析任务
result = orchestrator.analyze_task(
user_request="全面审查代码安全性",
task_context={
"files": ["src/**/*.py"],
"depth": "deep",
"budget": "medium"
}
)
## Agent Swarm 执行报告
### 参与的 Agents
1. [角色名称] - [任务描述]
2. [角色名称] - [任务描述]
...
### 执行结果
#### [Agent 1 角色名称]
[Agent 1 的发现和分析]
#### [Agent 2 角色名称]
[Agent 2 的发现和分析]
...
### 综合分析
[跨 Agent 的综合发现]
### 优先级建议
1. [高优先级问题]
2. [中优先级问题]
3. [低优先级问题]
系统自动收集以下指标：
Agent 统计
成本统计
性能统计
分析结果
# 📊 Agent Swarm 执行报告
**会话 ID**: abc123
**开始时间**: 2026-01-31 10:00:00
**结束时间**: 2026-01-31 10:05:30
**总耗时**: 330.5 秒
## Agent 统计
- **总数**: 6
- **完成**: 5 ✅
- **失败**: 1 ❌
- **运行中**: 0 🔄
- **平均执行时间**: 55.2 秒
## 成本统计
- **总成本**: $0.0234
- Haiku: $0.0089
- Sonnet: $0.0145
## 性能统计
- **缓存命中率**: 75.0%
- 命中: 3
- 未命中: 1
- **重试次数**: 2
- 成功: 1
- 成功率: 50.0%
## 分析结果
- **发现问题总数**: 23
- 🔴 严重: 2
- 🟠 高: 5
- 🟡 中: 10
- 🟢 低: 6
# 获取 Swarm 状态
status = coordinator.get_swarm_status()
# 获取重试统计
retry_stats = coordinator.get_retry_stats()
# 获取完整指标
metrics = coordinator.get_metrics()
# 生成报告
from metrics import MetricsReporter
reporter = MetricsReporter(metrics)
summary = reporter.generate_summary()
detailed = reporter.generate_detailed_report()
cost_breakdown = reporter.generate_cost_breakdown()
智能缓存
增量分析
并行优化
成本优化
| 场景 | 首次执行 | 缓存命中 | 增量分析 |
|---|---|---|---|
| 小项目 (< 1000 行) | 30s | 3s | 10s |
| 中型项目 (1000-5000 行) | 120s | 15s | 40s |
| 大型项目 (> 5000 行) | 300s | 30s | 90s |
适用场景
不适用场景
| 特性 | Kimi K2.5 | 本实现 | 状态 |
|---|---|---|---|
| 动态角色生成 | AI 自动生成 | LLM 分析生成 | ✅ 已实现 |
| 个性化名称 | 自动命名 | 预定义名称库 | ✅ 已实现 |
| 智能任务分解 | PARL 训练 | 规则 + LLM | ✅ 已实现 |
| Agent 间通信 | ✅ | 通信中心 | ✅ 已实现 |
| 动态调整 | ✅ | 动态生成/终止 | ✅ 已实现 |
| 智能缓存 | ❌ | 文件+结果缓存 | ✅ 增强 |
| 自动重试 | ❌ | 失败分析+重试 | ✅ 增强 |
| 增量分析 | ❌ | Git diff 驱动 | ✅ 增强 |
| 智能模型选择 | ❌ | 基于任务特征 | ✅ 增强 |
| 监控指标 | ❌ | 完整指标系统 | ✅ 增强 |
| 角色库规模 | 未知 | 40+ 角色 | ✅ 扩展 |
相比 Kimi AgentSwarm：
相比传统单 Agent：
orchestrator.py)# 自动分析任务并生成 Agent 配置
orchestrator = AgentOrchestrator(max_agents=10)
config = orchestrator.analyze_task(user_request)
# 每个 Agent 都有：
# - 角色名称（如 "Code Quality Analyst"）
# - 个性化名字（如 "质量守护者 Alex"）
# - 专门的任务描述
# - 合适的工具集
特点：
coordinator.py)# 通信中心
comm_hub = AgentCommunicationHub()
# Agent 可以：
# - 广播消息给所有 Agent
# - 发送私信给特定 Agent
# - 请求帮助
# - 报告进度
特点：
coordinator.py)# 动态调整器
adjuster = DynamicAgentAdjuster(comm_hub)
# 自动：
# - 检测是否需要新增 Agent
# - 终止已完成的 Agent
# - 重新平衡任务负载
触发条件：
claude_integration.py)# 完整的集成层
swarm = ClaudeCodeAgentSwarm(max_agents=10)
result = swarm.process_user_request(user_request)
# 自动生成：
# - 执行计划
# - Task 工具调用配置
# - 详细的 Agent prompts
# - 结果聚合报告
请用 agent swarm 全面分析这个项目
自动执行：
用 agent swarm 分析代码，重点关注安全和性能
智能调整：
# 在 Python 中直接使用
from claude_integration import ClaudeCodeAgentSwarm
swarm = ClaudeCodeAgentSwarm(max_agents=8)
result = swarm.process_user_request("分析数据质量")
# 获取 Task 调用配置
for task in result['task_calls']:
print(f"Agent: {task['description']}")
print(f"Model: {task['model']}")
代码分析（8 个角色）：
数据分析（4 个角色）：
安全审计（4 个角色）：
| 复杂度 | 默认模型 | 关键角色模型 |
|---|---|---|
| Low | haiku | haiku |
| Medium | haiku | sonnet |
| High | sonnet | sonnet |
关键角色（总是用 sonnet）：
Agent 数量：最多 10 个（vs Kimi 的 100 个）
角色生成：基于规则（vs Kimi 的 AI 学习）
任务分解：启发式（vs Kimi 的强化学习）
用户请求
↓
AgentOrchestrator (任务分析)
↓
生成 Agent 配置
↓
AgentCoordinator (协调器)
↓
┌─────────────────────────────────┐
│ AgentCommunicationHub (通信) │
│ DynamicAgentAdjuster (调整) │
└─────────────────────────────────┘
↓
并行启动 Agents (Task 工具)
↓
实时监控和调整
↓
结果聚合
↓
综合报告
这个实现通过以下方式接近 Kimi AgentSwarm 的效果：
虽然受限于资源无法达到 100 个 Agents，但在 10 个 Agents 的规模下，已经能够提供类似的多角度分析和并行执行能力。
并行分析模式（默认）- 用于代码审查和分析 层级执行模式 - 用于实际开发和任务执行
# 全面代码审查
/agent-swarm 全面审查这个项目的代码质量和安全性
# 安全审计
/agent-swarm 进行全面的安全审计
# PR Review（自动增量分析）
/agent-swarm 审查本次提交的代码变更
from hierarchical_system import HierarchicalAgentSystem
# 创建层级化系统
system = HierarchicalAgentSystem(max_workers=5)
# 执行大任务
result = system.execute_big_task("""
开发一个用户管理系统，包括：
1. 用户注册和登录功能
2. 用户信息管理（增删改查）
3. 权限控制
4. 完整的单元测试
5. API 文档
""")
# 查看进度
progress = system.get_progress()
print(progress)
执行流程：
1. 主 Agent 分析任务 → 制定计划
├─ 子任务 1: 设计数据库模型
├─ 子任务 2: 实现后端 API
├─ 子任务 3: 实现前端界面
├─ 子任务 4: 编写测试
└─ 子任务 5: 生成文档
2. 创建 Worker Agents
├─ Backend Developer
├─ Frontend Developer
├─ QA Tester
└─ Technical Writer
3. 分配任务并执行
├─ 批次 1: 设计数据库模型（串行）
├─ 批次 2: 后端 + 前端（并行）
├─ 批次 3: 测试（串行）
└─ 批次 4: 文档（串行）
4. 主 Agent 验证和整合结果
示例 1: 功能开发
system = HierarchicalAgentSystem()
result = system.execute_big_task("""
实现一个文件上传功能：
- 支持多文件上传
- 文件大小限制 10MB
- 支持图片预览
- 进度条显示
""")
示例 2: Bug 修复
result = system.execute_big_task("""
修复登录页面的 bug：
- 用户名验证失败
- 密码加密不正确
- 记住我功能无效
""")
示例 3: 代码重构
result = system.execute_big_task("""
重构用户服务模块：
- 提取公共逻辑
- 优化数据库查询
- 改进错误处理
- 添加单元测试
""")
# 安全审计
/agent-swarm 进行全面的安全审计
# 性能分析
/agent-swarm 分析性能瓶颈和优化机会
# 业务逻辑审查
/agent-swarm 审查业务逻辑实现
# 可观测性检查
/agent-swarm 检查日志、监控和追踪实现
# 自动检测 Git 变更
/agent-swarm 审查本次提交的代码变更
执行完成后，系统会自动生成：
# 删除缓存目录
rm -rf .agent_swarm_cache
# 设置预算级别
orchestrator.analyze_task(
user_request="分析代码",
task_context={"budget": "low"} # low/medium/high
)
# 限制 Agent 数量
orchestrator = AgentOrchestrator(max_agents=3)
可能原因：
解决方案：
系统会自动重试 3 次，如果仍然失败：
建议：
系统会根据复杂度自动调整。
自动启用条件：
手动控制：
orchestrator = AgentOrchestrator(use_incremental=False)
# 获取指标
metrics = coordinator.get_metrics()
# 生成报告
from metrics import MetricsReporter
reporter = MetricsReporter(metrics)
print(reporter.generate_detailed_report())
原因：任务过于复杂或文件过多
解决：
原因：过多使用 sonnet 模型
解决：
budget="low"原因：长期积累的缓存文件
解决：
新增功能：
改进：
欢迎贡献新的角色、优化策略或功能改进！
添加新角色：
orchestrator.py 的 _get_agent_templates() 中添加_generate_agent_name() 中添加名称映射优化建议：
MIT License
获客分析智能体 - 专注于营销获客效果分析。支持渠道效果、用户质量、ROI、回传、RTA、LTV等分析。触发方式：命令 /acquisition、/获客分析，或包含关键词（获客、渠道、ROI、CPS、转化率、回传、RTA、LTV）时自动触发。适用场景：(1) 渠道效果分析，(2) 用户质量分析，(3) ROI分析，(4) 回传分析，(5) RTA策略分析，(6) LTV分析，(7) 版位分析。
API 数据分析智能体 - 从 REST API 获取数据并进行分析。支持 API 调用、数据转换、批量获取、性能监控和对比分析。触发方式：命令 /api-analyze、/api分析，或包含关键词（API、接口、性能监控、数据对比）时自动触发。适用场景：(1) API 数据监控，(2) 第三方数据集成，(3) 实时数据分析，(4) API 性能测试，(5) 数据同步验证。
Dataphin 数据探索智能体 - 快速探索表结构、数据质量和数据分布。支持表查找、字段分析、数据采样、质量检查。触发方式：命令 /dp-explore、/数据探索，或包含关键词（表结构、字段、数据质量、数据分布）时自动触发。适用场景：(1) 快速了解新表，(2) 数据质量检查，(3) 字段分析，(4) 数据分布探索，(5) 表关系梳理。
GitHub 开源项目自动化分析器 - 自动克隆项目、测试功能、探索使用场景、评估有效性并生成完整的 MD 分析报告。触发方式：命令 /github-analyze、/项目分析，或包含关键词（分析项目、评估项目、GitHub 项目）时自动触发。适用场景：(1) 技术选型评估，(2) 开源项目调研，(3) 竞品分析，(4) 学习研究。
自我修复技能 - 检测技能执行失败，自动诊断根因并修复技能定义
反射技能 - 分析任务执行历史，提取成功模式和失败教训，自动更新相关技能定义