CrewAI如何实现任务分工与协作？

Question

在使用CrewAI构建多智能体系统时，一个常见技术问题是：当多个Agent并行执行任务时，如何确保任务合理分工且避免重复工作？例如，若未明确角色权限与目标边界，不同Agent可能对同一子任务进行冗余处理，导致资源浪费与结果冲突。此外，在任务依赖链中，前序任务的延迟或输出格式不一致，可能阻塞后续Agent协作。开发者常需借助显式任务编排、共享上下文队列与状态同步机制来解决此类问题。如何设计高效的任务分配策略与通信协议，成为CrewAI实践中关键挑战。

Answer 1

让【宝码香车】来帮您解答，本回答参考DeepSeek编写，并整理提供。

如果答案能帮助您，请采纳，非常感谢！
.
每日一句正能量，让它带给您好心情：把日子过得有滋有味，就是对生活最好的回馈。
.

CrewAI 任务分工与协作实现方案

核心挑战分析

在CrewAI多智能体系统中，主要面临以下协作挑战：

• 角色边界模糊导致的重复工作
• 任务依赖管理不当引发的阻塞
• 上下文同步不及时造成的结果冲突

解决方案

1. 明确角色定义与权限划分

from crewai import Agent, Task, Crew


# 定义具有明确边界的角色
research_agent = Agent(
    role='市场研究员',
    goal='收集和分析市场数据，提供调研报告',
    backstory='专注于数据收集和分析的专业研究员',
    allow_delegation=False  # 禁止委托，避免越界
)


analysis_agent = Agent(
    role='数据分析师',
    goal='对调研数据进行深度分析，生成洞察',
    backstory='擅长数据挖掘和模式识别',
    allow_delegation=False
)

2. 任务依赖与执行顺序控制

# 定义有明确依赖关系的任务链
research_task = Task(
    description='收集2024年AI市场数据，输出标准化JSON格式',
    agent=research_agent,
    expected_output='{"market_size": "", "growth_rate": "", "key_players": []}'
)


analysis_task = Task(
    description='基于市场数据生成分析报告',
    agent=analysis_agent,
    context=[research_task],  # 显式依赖前序任务
    expected_output='包含趋势分析和建议的Markdown报告'
)


# 创建有序执行的团队
crew = Crew(
    agents=[research_agent, analysis_agent],
    tasks=[research_task, analysis_task],
    verbose=True
)

3. 共享上下文与状态同步

from crewai import Crew, Agent, Task, Process
from typing import Dict, Any
import json


class SharedContext:
    def __init__(self):
        self.data = {}
        self.lock = threading.Lock()
    
    def update(self, key: str, value: Any):
        with self.lock:
            self.data[key] = value
    
    def get(self, key: str, default=None):
        return self.data.get(key, default)


# 使用共享上下文
shared_context = SharedContext()


def research_callback(output: str):
    """研究任务完成后的回调函数"""
    try:
        data = json.loads(output)
        shared_context.update('market_data', data)
    except json.JSONDecodeError:
        # 处理格式错误，确保后续任务正常执行
        shared_context.update('market_data', {'error': '格式解析失败'})


research_task = Task(
    description='收集市场数据',
    agent=research_agent,
    callback=research_callback
)

4. 高效任务分配策略

from enum import Enum
from dataclasses import dataclass
from typing import List


class TaskPriority(Enum):
    HIGH = 3
    MEDIUM = 2
    LOW = 1


@dataclass
class TaskAssignment:
    task_id: str
    agent_role: str
    priority: TaskPriority
    dependencies: List[str]
    timeout: int = 300


class TaskOrchestrator:
    def __init__(self):
        self.pending_tasks = []
        self.completed_tasks = set()
        self.agent_capabilities = {
            'researcher': ['data_collection', 'market_analysis'],
            'analyst': ['data_analysis', 'insight_generation'],
            'writer': ['report_writing', 'content_creation']
        }
    
    def assign_task(self, task: TaskAssignment):
        """基于能力和依赖关系分配任务"""
        # 检查依赖是否满足
        if not all(dep in self.completed_tasks for dep in task.dependencies):
            self.pending_tasks.append(task)
            return False
        
        # 根据角色能力分配
        suitable_agents = [
            role for role, capabilities in self.agent_capabilities.items()
            if any(cap in capabilities for cap in task.required_capabilities)
        ]
        
        return suitable_agents[0] if suitable_agents else None

5. 通信协议与结果验证

import asyncio
from datetime import datetime
from pydantic import BaseModel


class TaskMessage(BaseModel):
    task_id: str
    sender: str
    receiver: str
    message_type: str  # 'request', 'response', 'update'
    content: dict
    timestamp: datetime
    signature: str  # 用于验证消息完整性


class CommunicationProtocol:
    def __init__(self):
        self.message_queue = asyncio.Queue()
        self.handlers = {}
    
    async def send_message(self, message: TaskMessage):
        """发送结构化消息"""
        await self.message_queue.put(message)
    
    async def process_messages(self):
        """处理消息队列"""
        while True:
            message = await self.message_queue.get()
            handler = self.handlers.get(message.message_type)
            if handler:
                await handler(message)

6. 完整的最佳实践示例

from crewai import Crew, Agent, Task, Process
import asyncio


async def main():
    # 1. 定义专业化的智能体
    researcher = Agent(
        role="高级研究员",
        goal="提供准确、全面的市场数据",
        backstory="数据驱动的研究专家",
        allow_delegation=False,
        max_iter=5  # 限制迭代次数避免无限循环
    )
    
    analyst = Agent(
        role="战略分析师",
        goal="基于数据提供可行的商业洞察",
        backstory="具有深度分析能力的战略专家",
        allow_delegation=False
    )
    
    # 2. 创建有明确输入输出的任务
    research_task = Task(
        description="收集Q3季度AI技术市场数据，包括市场规模、主要玩家和增长趋势",
        agent=researcher,
        expected_output="标准化的JSON数据，包含market_size, key_players, growth_rates字段",
        async_execution=True  # 允许异步执行
    )
    
    analysis_task = Task(
        description="分析市场数据，识别关键机会和风险",
        agent=analyst,
        context=[research_task],
        expected_output="包含SWOT分析和建议的战略报告",
        output_file="analysis_report.md"  # 明确输出位置
    )
    
    # 3. 配置并行处理流程
    crew = Crew(
        agents=[researcher, analyst],
        tasks=[research_task, analysis_task],
        process=Process.sequential,  # 或 Process.hierarchical 用于复杂工作流
        memory=True,  # 启用共享记忆
        cache=True,   # 缓存结果提高效率
        verbose=2
    )
    
    # 4. 执行并监控
    result = await crew.kickoff()
    return result


# 运行系统
if __name__ == "__main__":
    asyncio.run(main())

关键优化策略

1. 角色隔离：通过allow_delegation=False防止任务越界
2. 依赖管理：使用context参数明确任务依赖关系
3. 输出标准化：定义清晰的expected_output格式
4. 异步执行：对独立任务启用异步提高并行效率
5. 状态监控：实现任务执行状态的实时跟踪
6. 错误处理：建立健壮的重试和容错机制

这种设计确保了CrewAI系统中各智能体既能高效并行工作，又能保持必要的协作同步，有效避免了重复劳动和资源冲突。