CrewAI 核心概念 团队（Crews）篇
CrewAI 团队（Crews）核心内容总结
该文档系统讲解了 CrewAI 框架中“团队（Crew）”的定义、属性配置、创建方式、执行流程及输出管理，是实现多智能体协作完成复杂任务的核心指南，内容可分为团队概述、核心属性、创建方法、执行与输出管理、高级功能五部分，具体如下：
一、团队概述（Overview of a Crew）
在 CrewAI 中，团队（Crew） 是由多个智能体（Agent）组成的协作单元，核心作用是：
- 统筹任务执行：定义任务分配策略与工作流，协调智能体完成一组关联任务；
- 支持多智能体协作：通过指定执行流程（如顺序、分层），实现智能体间的分工与配合；
- 统一管理资源：集中配置内存、缓存、日志等资源，避免智能体重复配置；
- 企业级支持：配合 CrewAI Enterprise 的可视化工具，简化团队构建与流程调试。
二、团队核心属性（Crew Attributes）
团队的行为、资源与协作模式由多个属性定义，下表整理关键属性（完整属性见文档表格）：
| 类别 | 属性（参数） | 类型 | 核心作用 | 默认值/说明 |
|---|---|---|---|---|
| 必填 | tasks | List[Task] | 分配给团队的任务列表（需与智能体能力匹配） | 无（必须手动指定） |
agents | List[Agent] | 组成团队的智能体列表（需明确角色与专长） | 无（必须手动指定） | |
| 可选 | process | Process | 团队执行流程（顺序/分层），决定任务分配与执行顺序 | Process.sequential（顺序执行） |
verbose | bool | 启用详细日志（调试用，显示任务执行进度与智能体交互） | False | |
manager_llm | 多类型 | 分层流程（hierarchical）中“管理智能体”使用的LLM（必须配置分层流程） | None（分层流程下必填） | |
manager_agent | BaseAgent | 自定义管理智能体（替代默认管理逻辑，分层流程下可用） | None（默认自动生成管理智能体） | |
max_rpm | Optional[int] | 团队级每分钟最大请求数（限制API调用频率，优先级高于智能体自身配置） | None（不限制） | |
memory | 多类型 | 存储团队执行记忆（短期/长期/实体记忆），支持跨任务上下文复用 | None（需手动配置内存提供商） | |
cache | bool | 启用工具执行结果缓存（减少重复调用，提升效率） | True | |
output_log_file | Optional[Union[bool, str]] | 日志保存路径（True→保存为logs.txt，或指定自定义路径如“crew_logs.json”） | None（不保存日志） | |
planning | bool | 启用团队规划能力（执行前自动生成任务计划，添加到每个任务描述中） | False | |
knowledge_sources | List[BaseKnowledgeSource] | 团队级知识库（所有智能体可访问，替代单个智能体的知识库配置） | None（无共享知识库） | |
step_callback | Optional[Any] | 每个智能体执行步骤后触发的回调函数（如日志记录、进度监控） | None（无回调） | |
task_callback | Optional[Any] | 每个任务完成后触发的回调函数（如结果校验、通知发送） | None（无回调） |
三、团队创建方法（Creating Crews）
CrewAI 支持两种团队创建方式，推荐YAML 配置+装饰器（更易维护、适合大型项目），也可直接用代码定义。
1. YAML 配置+装饰器（推荐）
通过 YAML 文件统一管理智能体与任务配置，结合 CrewBase 类与装饰器自动收集资源，步骤如下：
步骤1：准备 YAML 配置文件
- 智能体配置：
config/agents.yaml（参考“智能体”文档，定义角色、目标、背景）； - 任务配置：
config/tasks.yaml（参考“任务”文档，定义描述、预期输出、关联智能体）。
步骤2：用装饰器定义团队类
通过 @CrewBase、@agent、@task、@crew 等装饰器，自动收集智能体与任务，无需手动维护列表：
from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff
from typing import List
from crewai.agents.agent_builder.base_agent import BaseAgent@CrewBase # 标记该类为团队基类
class MarketAnalysisCrew:"""市场分析团队：负责收集市场数据并生成报告"""agents: List[BaseAgent] # 自动收集的智能体列表tasks: List[Task] # 自动收集的任务列表# 指向 YAML 配置文件路径agents_config = "config/agents.yaml"tasks_config = "config/tasks.yaml"# （可选）团队启动前处理输入（如添加额外数据）@before_kickoffdef prepare_inputs(self, inputs):inputs["report_date"] = "2025-10" # 向输入添加报告日期return inputs# （可选）团队完成后处理输出（如补充报告说明）@after_kickoffdef process_output(self, output):output.raw += "\n本报告由 MarketAnalysisCrew 生成，数据截止2025-10"return output# 定义智能体1（对应 YAML 中"market_researcher"）@agentdef market_researcher(self) -> Agent:return Agent(config=self.agents_config["market_researcher"],verbose=True)# 定义智能体2（对应 YAML 中"report_writer"）@agentdef report_writer(self) -> Agent:return Agent(config=self.agents_config["report_writer"],verbose=True)# 定义任务1（对应 YAML 中"collect_market_data"）@taskdef collect_market_data(self) -> Task:return Task(config=self.tasks_config["collect_market_data"])# 定义任务2（对应 YAML 中"generate_market_report"）@taskdef generate_market_report(self) -> Task:return Task(config=self.tasks_config["generate_market_report"])# 定义团队（指定执行流程与全局配置）@crewdef crew(self) -> Crew:return Crew(agents=self.agents, # 自动获取@agent装饰器定义的智能体tasks=self.tasks, # 自动获取@task装饰器定义的任务process=Process.sequential, # 顺序执行任务verbose=True,output_log_file="market_crew_logs.json" # 日志保存为JSON文件)# 运行团队（传入输入参数）
result = MarketAnalysisCrew().crew().kickoff(inputs={"topic": "AI市场规模"})
print(result.raw)
2. 直接代码定义（适用于简单场景）
不依赖 YAML 配置，直接在代码中定义智能体、任务与团队，手动管理资源列表：
from crewai import Agent, Crew, Task, Process
from crewai_tools import SerperDevToolclass SimpleResearchCrew:# 定义智能体1：研究员def research_agent(self) -> Agent:return Agent(role="AI研究员",goal="收集2025年AI领域最新进展",backstory="10年科技领域研究经验，擅长挖掘前沿技术信息",tools=[SerperDevTool()], # 配置搜索工具verbose=True)# 定义智能体2：报告撰写者def writer_agent(self) -> Agent:return Agent(role="报告撰写者",goal="将研究结果整理为结构化报告",backstory="擅长将复杂信息转化为易懂的报告，熟悉Markdown格式",verbose=True)# 定义任务1：研究任务def research_task(self) -> Task:return Task(description="收集2025年AI领域top5最新进展，每个进展附来源",expected_output="包含5个要点的子弹列表，每个要点含进展描述与来源",agent=self.research_agent())# 定义任务2：报告任务def write_task(self) -> Task:return Task(description="基于研究结果，生成Markdown格式的分析报告",expected_output="含标题、章节、引用的完整报告，不少于1000字",agent=self.writer_agent(),context=[self.research_task()] # 依赖研究任务的输出)# 定义团队def crew(self) -> Crew:return Crew(agents=[self.research_agent(), self.writer_agent()], # 手动传入智能体tasks=[self.research_task(), self.write_task()], # 手动传入任务process=Process.sequential,verbose=True,max_rpm=15 # 限制团队每分钟最多15次API调用)# 运行团队
result = SimpleResearchCrew().crew().kickoff()
print(result.raw)
四、团队执行与输出管理
1. 核心执行流程（Process）
团队支持两种执行流程，适配不同协作场景：
| 流程类型 | 特点 | 适用场景 | 关键配置要求 |
|---|---|---|---|
Process.sequential（顺序） | 任务按定义顺序执行，前一个任务完成后才启动下一个 | 任务有明确依赖关系（如“研究→分析→报告”），需线性推进 | 无需额外配置，默认流程 |
Process.hierarchical（分层） | 由“管理智能体”统筹协调：分配任务给合适的智能体，验证结果后再推进 | 多智能体分工协作（如“数据收集→数据清洗→数据分析→可视化”），需动态分配任务 | 必须配置 manager_llm（管理智能体用的LLM）或 manager_agent（自定义管理智能体） |
分层流程示例：
from crewai import Crew, Process
from crewai.llm import LLM# 分层流程团队（需配置manager_llm）
hierarchical_crew = Crew(agents=[data_collector, data_cleaner, analyst, visualizer],tasks=[collect_task, clean_task, analyze_task, visualize_task],process=Process.hierarchical,manager_llm=LLM(model="gpt-4"), # 管理智能体用的LLMverbose=True
)
hierarchical_crew.kickoff()
2. 团队启动方法（Kickoff）
提供4种启动方法，支持同步/异步、单输入/多输入场景：
| 方法名 | 功能描述 | 适用场景 |
|---|---|---|
kickoff(inputs={}) | 同步启动团队，执行一次任务（传入单个输入参数） | 单任务场景（如生成一份报告） |
kickoff_for_each(inputs_array) | 同步启动团队，对输入数组中的每个元素执行一次任务（顺序执行） | 批量任务场景（如为多个主题生成报告） |
kickoff_async(inputs={}) | 异步启动团队，执行一次任务（需配合async/await使用） | 不阻塞主线程的场景（如同时运行多个团队） |
kickoff_for_each_async(inputs_array) | 异步启动团队，对输入数组中的每个元素并发执行任务 | 批量且需高效处理的场景（如同时为10个主题生成报告） |
示例：批量异步执行
import asyncio# 定义输入数组（多个主题）
inputs_array = [{"topic": "AI在医疗领域的应用"},{"topic": "AI在金融领域的应用"},{"topic": "AI在教育领域的应用"}
]# 异步批量执行
async def run_batch():crew = MarketAnalysisCrew().crew()results = await crew.kickoff_for_each_async(inputs=inputs_array)for i, result in enumerate(results):print(f"=== 结果 {i+1}：{inputs_array[i]['topic']} ===")print(result.raw)asyncio.run(run_batch())
3. 团队输出（Crew Output）
团队执行结果封装在 CrewOutput 类中，包含全局结果与各任务细节，核心属性如下：
| 属性 | 类型 | 说明 |
|---|---|---|
raw | str | 团队最终原始输出（默认取最后一个任务的输出） |
pydantic | Optional[BaseModel] | 结构化输出（仅当最后一个任务配置 output_pydantic 时存在） |
json_dict | Optional[Dict] | JSON格式输出（仅当最后一个任务配置 output_json 时存在） |
tasks_output | List[TaskOutput] | 所有任务的输出列表（每个元素为 TaskOutput 对象，含单个任务的结果） |
token_usage | Dict[str, Any] | 团队执行的总token消耗（含所有智能体的LLM调用与工具使用） |
输出访问示例：
# 执行团队并获取输出
crew = MarketAnalysisCrew().crew()
crew_output = crew.kickoff(inputs={"topic": "AI市场规模"})# 访问全局结果
print("团队最终报告：", crew_output.raw)
print("总Token消耗：", crew_output.token_usage)# 访问单个任务的输出（如第一个任务：数据收集）
collect_task_output = crew_output.tasks_output[0]
print("数据收集任务结果：", collect_task_output.raw)
print("数据收集任务智能体：", collect_task_output.agent)
4. 日志与 metrics 管理
（1）日志保存
通过 output_log_file 配置日志保存，支持TXT/JSON格式：
# 保存为默认TXT日志（logs.txt）
crew = Crew(..., output_log_file=True)# 保存为自定义JSON日志（crew_execution.json）
crew = Crew(..., output_log_file="crew_execution.json")
（2）执行 metrics 查看
团队执行后，通过 usage_metrics 属性查看详细消耗统计：
crew.kickoff()
# 查看每个任务的token消耗、执行时间等
print("团队执行详情：", crew.usage_metrics)
五、团队高级功能
1. 记忆与缓存（Memory & Cache）
（1）团队级记忆
配置共享记忆，支持智能体跨任务复用上下文（如长期记忆存储历史执行经验）：
from crewai.memory import ShortTermMemorycrew = Crew(agents=[agent1, agent2],tasks=[task1, task2],memory=ShortTermMemory(), # 配置短期记忆memory_config={"retention_limit": 10}, # 记忆保留10个步骤verbose=True
)
（2）工具缓存
默认启用 cache=True，自动缓存工具执行结果，避免重复调用（如相同搜索关键词不重复请求API）：
# 禁用缓存（适合实时性要求高的场景）
crew = Crew(..., cache=False)
2. 团队级规划（Planning）
启用 planning=True 后，团队执行前会自动生成任务计划，并将计划添加到每个任务描述中，提升执行连贯性：
from crewai.llm import LLMcrew = Crew(agents=[researcher, writer],tasks=[research_task, write_task],planning=True, # 启用规划能力planning_llm=LLM(model="gpt-4o-mini"), # 用于生成计划的LLMverbose=True
)
# 执行前会生成类似：“1. 先执行研究任务，收集3个核心数据；2. 基于数据生成3章节报告”的计划
3. 任务重放（Replay）
通过 CLI 命令重放指定任务（需先保存任务输出），支持调试与断点续跑：
# 1. 查看最新执行的任务ID
crewai log-tasks-outputs# 2. 重放指定任务（替换<Task_ID>为实际ID）
crewai replay -t <Task_ID>
总结
CrewAI 团队是多智能体协作的核心载体，通过灵活的流程配置（顺序/分层）、资源管理（记忆/缓存/日志）与启动方法，支持从简单线性任务到复杂分工协作的全场景需求。创建时推荐用 YAML+装饰器模式提升可维护性，实际使用中需根据任务依赖关系选择执行流程（顺序适合线性任务，分层适合动态分工），并利用日志、metrics 与重放功能优化执行效率与调试体验。