| name | agent-collaboration-protocol |
| description | 多智能体协作规则框架，用于在没有原生 Agent Teams 功能的模型上实现协作。适用于 Kimi、DeepSeek、MiniMax、百川、零一万物等模型。触发词：多智能体协作、agent teams、协作规则、多agent编排。 |
Agent Collaboration Protocol
为无原生 Agent Teams 功能的模型提供多智能体协作框架。
适用模型
- Kimi (月之暗面)
- DeepSeek
- MiniMax (海螺 AI)
- 百川大模型
- 零一万物 Yi
- 其他无多智能体平台的 API 模型
协作模式
1. 角色分配模式 (Role-Based)
适用场景： 任务有明确的分工
┌─────────────┐
│Coordinator │ ← 任务分配、进度监控
└──────┬──────┘
│
┌──────┼──────┬──────────┐
│ │ │ │
▼ ▼ ▼ ▼
Agent1 Agent2 Agent3 Agent4
(研究) (分析) (执行) (验证)
协作规则：
- Coordinator 负责任务分解和结果汇总
- 每个 Agent 专注单一职责
- 结果需交叉验证
2. 竞争验证模式 (Competitive Verification)
适用场景： 需要高可靠性答案
┌─────────────┐
│ Question │
└──────┬──────┘
│
┌────────┼────────┐
▼ ▼ ▼
Agent1 Agent2 Agent3
(独立分析)
│ │ │
└────────┼────────┘
▼
┌─────────────┐
│Consensus │ ← 比较、投票、融合
└─────────────┘
协作规则：
- 多个 Agent 独立分析同一问题
- 比较结果差异
- 通过投票或融合得出最终答案
3. 流水线模式 (Pipeline)
适用场景： 任务有顺序依赖
Input → [Agent1] → [Agent2] → [Agent3] → Output
预处理 处理 后处理
协作规则：
- 每个 Agent 处理特定阶段
- 输出作为下一 Agent 输入
- 错误可回溯到具体阶段
4. 层级模式 (Hierarchical)
适用场景： 复杂任务需要多级决策
┌─────────────┐
│ Manager │ ← 战略决策
└──────┬──────┘
│
┌──────────┼──────────┐
▼ ▼ ▼
Supervisor Supervisor Supervisor
(战术决策) (战术决策) (战术决策)
│ │ │
┌───┴───┐ ┌───┴───┐ ┌───┴───┐
Worker Worker Worker Worker Worker Worker
(执行) (执行) (执行) (执行) (执行) (执行)
核心协作规则
1. 任务分解规则
def decompose_task(task):
"""任务分解"""
task_type = classify_task(task)
mode = select_mode(task_type)
roles = assign_roles(mode)
interfaces = define_interfaces(roles)
return {
"mode": mode,
"roles": roles,
"interfaces": interfaces
}
2. 通信协议
消息格式：
{
"from": "agent_id",
"to": "agent_id|broadcast",
"type": "request|response|notify|error",
"task_id": "unique_id",
"content": "...",
"timestamp": "ISO8601",
"metadata": {}
}
通信规则：
- 所有消息使用标准格式
- 重要消息需要确认 (ACK)
- 超时自动重试 (最多 3 次)
- 错误必须上报
3. 错误级联防御
错误检测：
def detect_error(result):
errors = []
if not validate_format(result):
errors.append("FORMAT_ERROR")
if not validate_logic(result):
errors.append("LOGIC_ERROR")
if not validate_bounds(result):
errors.append("BOUNDARY_ERROR")
return errors
错误隔离：
- 检测到错误的 Agent 立即停止
- 通知所有相关 Agent
- 记录错误上下文
- 启动恢复流程
4. 共识机制
投票规则：
def consensus(results):
"""达成共识"""
all_results = collect_results()
similarity_matrix = compute_similarity(all_results)
clusters = cluster_results(similarity_matrix)
largest_cluster = max(clusters, key=len)
final_result = merge_results(largest_cluster)
return final_result
Agent 角色定义
Coordinator (协调者)
职责：
能力要求：
Researcher (研究者)
职责：
能力要求：
Analyst (分析师)
职责：
能力要求：
Executor (执行者)
职责：
能力要求：
Validator (验证者)
职责：
能力要求：
实现示例
使用 LangChain 实现协作
from langchain.agents import AgentExecutor
from langchain.chat_models import ChatOpenAI
class AgentCollaboration:
def __init__(self, model_api, model_name):
self.llm = ChatOpenAI(
model=model_name,
openai_api_base=model_api,
openai_api_key="your_key"
)
self.agents = {}
def create_agent(self, role, system_prompt):
"""创建 Agent"""
self.agents[role] = {
"llm": self.llm,
"system_prompt": system_prompt,
"history": []
}
def collaborate(self, task, mode="role_based"):
"""执行协作"""
if mode == "role_based":
return self._role_based_collaboration(task)
elif mode == "competitive":
return self._competitive_collaboration(task)
elif mode == "pipeline":
return self._pipeline_collaboration(task)
最佳实践
1. Agent 数量控制
- 简单任务：1-2 个 Agent
- 中等任务：3-5 个 Agent
- 复杂任务：5-10 个 Agent
- 避免 Agent 过多导致协调开销
2. 任务粒度
- 每个 Agent 任务应明确且独立
- 任务间接口清晰定义
- 避免任务重叠
3. 错误处理
- 每个 Agent 应有错误检测
- 错误应快速传播
- 有恢复机制
4. 结果验证
- 关键结果需要多 Agent 验证
- 使用竞争模式提高可靠性
- 记录验证过程
参考资料
Activation Keywords
agent-collaboration-protocolagent-collaboration-protocolagent collaboration protocol
Tools Used
Instructions for Agents
- Read the task description carefully
- Follow the step-by-step process
- Use the appropriate tools
- Verify the results
Examples
Example 1: Basic Usage
User:
Agent:
Example 2: Advanced Usage
User:
Agent:
