05.3 — 角色分配与编排
定位: 多 Agent 系统的"组织设计"——理解如何为 Agent 设计角色、分配任务、编排工作流 面试高频度: ⭐⭐⭐ 考查方式: 设计题（设计多 Agent 角色体系）、策略题（动态 vs 静态角色分配）
一、这是什么？为什么需要它？
是什么
角色分配 是为每个 Agent 定义职责、权限和行为的规范。任务编排 是将多个 Agent 的工作组织成有序工作流的过程。
为什么需要角色和编排？
❌ 没有角色的多 Agent：
所有 Agent 能力相同
不知道谁该做什么
重复工作和互相推诿并存
缺乏责任感
✅ 有角色的多 Agent：
每个 Agent 有明确的职责边界
知道自己的任务和依赖
专业化带来更高质量
可追溯（谁做的什么）核心洞察：角色设计就是多 Agent 系统的"组织架构设计"——好的角色设计让协作自然发生。
二、原理拆解
2.1 经典角色模式
| 角色 | 职责 | prompt 关键词 | 典型行为 |
|---|---|---|---|
| 管理者 (Manager) | 分解任务、分配、汇总 | 规划、分配、监督 | "你负责 X，完成后报告" |
| 执行者 (Worker) | 执行具体任务 | 执行、完成、实现 | 专注完成分配的任务 |
| 审查者 (Reviewer) | 检查质量、反馈 | 审查、验证、确认 | "第3行有问题，建议改为..." |
| 协调者 (Coordinator) | 解决冲突、同步进度 | 协调、沟通、仲裁 | "A 和 B 的方案可以结合..." |
| 专家 (Specialist) | 领域深度知识 | 专业、深入分析 | 提供某个领域的深入见解 |
2.2 角色系统设计
python
class AgentRole:
"""Agent 角色定义"""
def __init__(self, name: str, description: str,
responsibilities: list, permissions: list):
self.name = name
self.description = description
self.responsibilities = responsibilities
self.permissions = permissions
def build_system_prompt(self) -> str:
"""根据角色生成 system prompt"""
return f"""你是{self.name}。
{self.description}
你的职责：
{chr(10).join(f'- {r}' for r in self.responsibilities)}
你有以下权限：
{chr(10).join(f'- {p}' for p in self.permissions)}
请在你的职责范围内工作，不要越界。"""
# 定义常用角色
ROLES = {
"researcher": AgentRole(
"研究员",
"负责信息收集和数据分析",
["搜索和分析信息", "整理数据报告", "验证事实"],
["使用搜索工具", "访问知识库"]
),
"writer": AgentRole(
"写手",
"负责内容创作和文档编写",
["撰写文档", "优化表达", "保证可读性"],
["编辑文件", "使用格式化工具"]
),
"reviewer": AgentRole(
"审查员",
"负责质量审核和反馈",
["检查错误", "给出改进建议", "验证合规性"],
["阅读所有文件", "标注问题"]
),
"coordinator": AgentRole(
"协调员",
"负责团队协作和进度管理",
["分配任务", "跟踪进度", "解决冲突"],
["查看所有角色状态", "重新分配任务"]
)
}2.3 角色分配策略
| 策略 | 做法 | 优点 | 缺点 |
|---|---|---|---|
| 固定角色 | 每个 Agent 角色固定 | 简单、可预测 | 不够灵活 |
| 动态角色 | 根据任务动态分配角色 | 灵活、资源利用好 | 需要角色切换机制 |
| 混合角色 | 基础角色固定 + 临时角色 | 兼顾稳定性和灵活性 | 实现复杂 |
python
class DynamicRoleManager:
"""动态角色管理器"""
def __init__(self, agents: list):
self.agents = agents # 所有可用 Agent
self.role_assignments = {} # agent -> current role
def assign_role(self, agent, role: str, task: str):
"""为 Agent 分配角色"""
self.role_assignments[agent.name] = role
prompt = f"""你现在的角色是：{ROLES[role].name}
{ROLES[role].description}
当前任务：{task}
请在当前角色的职责范围内工作。"""
agent.set_system_prompt(prompt)
def switch_role(self, agent, new_role: str):
"""切换 Agent 角色"""
old_role = self.role_assignments.get(agent.name)
print(f"{agent.name}: {old_role} → {new_role}")
self.assign_role(agent, new_role, "")
def determine_role(self, task: str) -> str:
"""根据任务决定需要的角色"""
return llm.call(f"任务：{task}\n可选角色：{list(ROLES.keys())}\n最需要哪个角色？")2.4 任务编排模式
串行编排：
Agent A → Agent B → Agent C
适合：流水线式任务（收集 → 分析 → 报告）
并行编排：
Agent A ──→ 结果合并
Agent B ──→ 结果合并
Agent C ──→ 结果合并
适合：独立子任务（同时搜索多个数据源）
混合编排：
┌──→ Agent B ──→ ┐
A ──→ Agent C ──→ D
└──→ Agent E ──→ ┘
适合：复杂任务（多个分析 → 综合汇总）python
class TaskOrchestrator:
"""任务编排器"""
def orchestrate(self, workflow: list, agents: dict):
"""执行工作流"""
results = {}
for step in workflow:
task = step["task"]
assignee = step["assignee"]
depends_on = step.get("depends_on", [])
# 检查依赖是否满足
for dep in depends_on:
if dep not in results:
raise ValueError(f"依赖 {dep} 未完成")
# 执行任务
context = {dep: results[dep] for dep in depends_on}
agent = agents[assignee]
result = agent.run(task, context)
results[task["name"]] = result
return results
# 示例工作流
workflow = [
{"task": {"name": "收集数据", "desc": "搜索市场数据"},
"assignee": "研究员"},
{"task": {"name": "分析数据", "desc": "分析趋势"},
"assignee": "分析师", "depends_on": ["收集数据"]},
{"task": {"name": "生成报告", "desc": "写报告"},
"assignee": "写手", "depends_on": ["分析数据"]},
]
orchestrator = TaskOrchestrator()
results = orchestrator.orchestrate(workflow, agents)三、图解全景
┌──────────────────────────────────────────────────────────┐
│ 多 Agent 角色与编排全景 │
│ │
│ 角色定义层 │
│ ┌──────────────────────────────────────┐ │
│ │ Researcher │ Writer │ Reviewer │ │
│ │ Coordinator │ Specialist │ │
│ └──────────────────┬───────────────────┘ │
│ │ │
│ 编排层 │ │
│ ┌──────────────────▼───────────────────┐ │
│ │ │ │
│ │ 并行区 串行区 条件分支 │ │
│ │ ┌──┐ ┌──→ ┌──→ ┌──┐ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ │──→ 合并 ─→ ─→ ─→ │ │ │
│ │ │ │ │ │ │ │ │
│ │ └──┘ └─────────┘──┘ │ │
│ └──────────────────┬───────────────────┘ │
│ │ │
│ 执行层 │ │
│ ┌──────────────────▼───────────────────┐ │
│ │ Agent A (研究员) │ │
│ │ Agent B (分析员) │ │
│ │ Agent C (写手) │ │
│ │ Agent D (审查员) │ │
│ └──────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────┘四、实战验证
python
# 角色化多 Agent：内容生产流水线
# 定义 Agent
agents = {
"researcher": Agent(ROLES["researcher"]),
"writer": Agent(ROLES["writer"]),
"reviewer": Agent(ROLES["reviewer"]),
"coordinator": Agent(ROLES["coordinator"])
}
# 任务：生产一篇技术文章
pipeline = """
1. [研究员] 收集 AI Agent 最新趋势信息
2. [写手] 基于收集的信息撰写文章
3. [审查员] 审查文章质量
4. [协调员] 如果需要修改，协调修改流程
"""
# 执行
orchestrator = TaskOrchestrator()
results = orchestrator.orchestrate([
{"task": "收集 AI Agent 最新趋势", "assignee": "researcher"},
{"task": "基于收集的信息写文章", "assignee": "writer", "depends_on": ["researcher"]},
{"task": "审查文章", "assignee": "reviewer", "depends_on": ["writer"]},
], agents)
print("生产完成！")五、面试视角
| 追问 | 答案要点 |
|---|---|
| 角色设计的原则？ | 单一职责（每个 Agent 聚焦一个领域）、职责不重叠（不要两个 Agent 做同样的事）、权限最小化（只给需要的权限）、角色可替换（任何 Agent 都能承担某个角色） |
| 固定角色 vs 动态角色的选择？ | 固定角色适合流程稳定的场景（简单、可预测），动态角色适合任务多样的场景（灵活、资源高效）。生产环境常用混合模式 |
| 任务编排中如何处理 Agent 故障？ | 超时检测（Agent 超时无响应则重试）、角色冗余（关键角色有备选 Agent）、任务重分配（将失败任务重新分配）、降级模式（退化为串行执行） |
| 角色和工具的关系？ | 角色决定了 Agent "可以做什么"，工具决定了 Agent "能做什么"。角色是权限层（应该做什么），工具是能力层（能够做什么） |
📚 相关链接
- **多智能体架构** — 角色运行的架构环境
- **通信与协调** — 角色间的通信
- **CrewAI** — CrewAI 的角色系统
- **工具注册** — 角色的工具权限
- ← 返回 **多智能体系统索引**