CrewAI 任务（Tasks）核心内容总结

该文档系统讲解了 CrewAI 框架中“任务（Task）”的定义、属性配置、创建方式、执行逻辑及高级功能，是实现智能体任务分配与流程管控的核心指南，内容可分为任务概述、核心属性、创建方法、执行与输出管理、高级功能、最佳实践六部分，具体如下：

一、任务概述（Overview of a Task）

在 CrewAI 中，任务（Task） 是智能体（Agent）需完成的具体指派，具备以下核心特征：

1. 信息完整性：包含执行所需的全部细节（描述、负责人、工具等），支持从简单查询到复杂协作的各类场景；
2. 协作性：支持多智能体协作完成，通过任务属性与团队（Crew）的执行流程（Process）实现分工与流程管控；
3. 企业级工具：CrewAI Enterprise 提供“可视化任务构建器（Visual Task Builder）”，支持无代码操作（拖拽创建、可视化依赖、实时测试），简化复杂任务流设计。

二、任务核心属性（Task Attributes）

任务的行为与输出由多个属性定义，分为必填属性和可选属性，下表整理关键属性（完整属性见文档表格）：

类别	属性（参数）	类型	核心作用	默认值/说明
必填	description	str	清晰描述任务内容（如“研究2025年AI最新进展”）	无（必须手动指定，需明确、无歧义）
	expected_output	str	详细定义任务完成的标准（如“10个要点的子弹列表”）	无（必须手动指定，避免智能体输出偏离预期）
可选	agent	Optional[BaseAgent]	指定负责执行任务的智能体	None（若未指定，由 hierarchical 流程自动分配）
	tools	List[BaseTool]	限制任务中智能体可使用的工具（优先级高于智能体默认工具）	None（默认使用智能体自身工具）
	context	Optional[List[Task]]	指定任务依赖的其他任务（需等待依赖任务完成，使用其输出作为上下文）	None（无依赖）
	async_execution	bool	启用异步执行（团队不等待该任务完成，继续执行后续任务）	False
	human_input	bool	需人工审核智能体的最终输出后，任务才完成	False
	markdown	bool	要求智能体以 Markdown 格式输出（自动添加格式指令）	False
	output_file	str	任务输出的文件保存路径（如“report.md”）	None（不保存到文件）
	output_pydantic	Type[BaseModel]	用 Pydantic 模型定义结构化输出（确保格式一致性与数据验证）	None（默认输出原始文本）
	output_json	Type[BaseModel]	用 Pydantic 模型定义 JSON 结构化输出（便于解析与跨任务传递）	None（默认输出原始文本）
	guardrail	函数/字符串/LLMGuardrail	任务输出的验证/转换逻辑（确保数据质量，如“验证JSON格式”）	None（无验证）
	callback	Optional[Any]	任务完成后执行的回调函数（如发送通知、后续处理）	None（无回调）

三、任务创建方法（Creating Tasks）

CrewAI 支持两种任务创建方式，推荐YAML 配置（更易维护、便于批量管理），也可直接用代码定义。

1. YAML 配置（推荐）

通过 YAML 文件统一管理任务配置，支持变量替换（如 {topic}），步骤如下：

步骤1：编写 YAML 配置文件

示例 src/latest_ai_development/config/tasks.yaml：

# 研究任务（依赖 {topic} 变量，指定执行智能体为 researcher）
research_task:
  description: >
    深入研究 {topic} 领域，结合2025年当前时间，挖掘相关且有趣的信息
  expected_output: >
    包含10个要点的子弹列表，总结 {topic} 领域最相关的信息
  agent: researcher  # 对应 YAML 中定义的智能体名

# 报告任务（Markdown 格式，输出到文件）
reporting_task:
  description: >
    基于研究结果，将每个要点扩展为报告的完整章节，确保信息详实
  expected_output: >
    结构完整的报告，每个主题对应一个章节，用 Markdown 格式（无需代码块标记）
  agent: reporting_analyst  # 对应 YAML 中定义的智能体名
  markdown: true  # 启用 Markdown 格式
  output_file: report.md  # 输出到 report.md 文件

步骤2：在代码中加载 YAML

通过 CrewBase 类加载配置，任务方法名需与 YAML 中任务名一致：

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool

@CrewBase
class LatestAiDevelopmentCrew():
    # 加载智能体与任务的 YAML 配置
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    # 定义智能体（对应 YAML 中的 researcher）
    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            verbose=True,
            tools=[SerperDevTool()]  # 为智能体添加搜索工具
        )

    # 定义智能体（对应 YAML 中的 reporting_analyst）
    @agent
    def reporting_analyst(self) -> Agent:
        return Agent(
            config=self.agents_config['reporting_analyst'],
            verbose=True
        )

    # 定义任务（对应 YAML 中的 research_task）
    @task
    def research_task(self) -> Task:
        return Task(config=self.tasks_config['research_task'])

    # 定义任务（对应 YAML 中的 reporting_task）
    @task
    def reporting_task(self) -> Task:
        return Task(config=self.tasks_config['reporting_task'])

    # 定义团队（指定执行流程）
    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=[self.researcher(), self.reporting_analyst()],
            tasks=[self.research_task(), self.reporting_task()],
            process=Process.sequential  # 顺序执行任务
        )

# 运行团队，传入变量 {topic} 的值
crew = LatestAiDevelopmentCrew()
crew.kickoff(inputs={'topic': 'AI Agents'})

2. 直接代码定义

通过实例化 Task 类直接创建，适用于简单场景或快速测试：

from crewai import Task

# 研究任务（指定智能体、工具）
research_task = Task(
    description="""
        深入研究 AI Agents 领域，结合2025年当前时间，挖掘相关且有趣的信息
    """,
    expected_output="""
        包含10个要点的子弹列表，总结 AI Agents 领域最相关的信息
    """,
    agent=researcher,  # 关联已定义的 researcher 智能体
    tools=[SerperDevTool()]  # 为该任务指定搜索工具
)

# 报告任务（Markdown 格式，输出到文件）
reporting_task = Task(
    description="""
        基于研究结果，将每个要点扩展为报告的完整章节，确保信息详实
    """,
    expected_output="""
        结构完整的报告，每个主题对应一个章节，用 Markdown 格式
    """,
    agent=reporting_analyst,  # 关联已定义的 reporting_analyst 智能体
    markdown=True,
    output_file="report.md"
)

四、任务执行与输出管理

1. 执行流程（Execution Flow）

任务的执行顺序由团队（Crew）的 process 参数控制，支持两种核心流程：

流程类型	特点	适用场景
Process.sequential（顺序）	任务按定义顺序执行，前一个任务完成后才开始下一个	任务有明确依赖（如“研究→报告”），需按步骤推进
Process.hierarchical（分层）	按智能体角色与专长自动分配任务，无需手动指定执行顺序	多智能体协作，任务无严格先后依赖（如“同时收集数据+分析数据”）

代码示例：

from crewai import Crew, Process

# 顺序执行流程
sequential_crew = Crew(
    agents=[researcher, analyst],
    tasks=[research_task, analysis_task],
    process=Process.sequential
)

# 分层执行流程（自动分配任务）
hierarchical_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[data_task, analysis_task, report_task],
    process=Process.hierarchical
)

2. 任务输出（Task Output）

任务执行结果封装在 TaskOutput 类中，支持多种输出格式，核心属性如下：

属性	类型	说明
raw	str	原始输出文本（默认格式，所有任务均包含）
pydantic	Optional[BaseModel]	结构化输出（仅当任务配置 output_pydantic 时存在，符合 Pydantic 模型）
json_dict	Optional[Dict]	JSON 格式输出（仅当任务配置 output_json 时存在，可直接解析）
agent	str	执行任务的智能体角色名
output_format	OutputFormat	输出格式（RAW/JSON/Pydantic，自动根据配置识别）

输出访问示例：

# 执行任务后，通过 task.output 访问结果
crew.kickoff()
task_output = research_task.output

# 访问原始输出
print("原始输出：", task_output.raw)

# 访问 Pydantic 结构化输出（若配置）
if task_output.pydantic:
    print("结构化标题：", task_output.pydantic.title)

# 访问 JSON 输出（若配置）
if task_output.json_dict:
    print("JSON 数据：", task_output.json_dict["key"])

3. 结构化输出配置

通过 output_pydantic 或 output_json 确保任务输出格式一致，便于跨任务传递与解析：

（1）使用 output_pydantic（推荐）

定义 Pydantic 模型约束输出结构，支持直接访问属性：

from pydantic import BaseModel
from crewai import Task

# 定义 Pydantic 模型
class AINews(BaseModel):
    title: str  # 新闻标题
    summary: str  # 新闻摘要
    source: str  # 信息来源

# 创建任务（指定结构化输出）
news_task = Task(
    description="收集2025年AI领域 top1 新闻",
    expected_output="包含标题、摘要、来源的结构化信息",
    agent=researcher,
    output_pydantic=AINews  # 绑定模型
)

# 执行后访问结构化输出
crew.kickoff()
print("新闻标题：", news_task.output.pydantic.title)
print("新闻来源：", news_task.output.pydantic.source)

（2）使用 output_json

通过 Pydantic 模型约束 JSON 格式，支持字典式访问：

# 复用 AINews 模型，配置 JSON 输出
json_task = Task(
    description="收集2025年AI领域 top1 新闻",
    expected_output="JSON 格式的新闻信息",
    agent=researcher,
    output_json=AINews  # 绑定模型
)

# 执行后访问 JSON 输出
crew.kickoff()
print("新闻标题：", json_task.output.json_dict["title"])
print("新闻摘要：", json_task.output.json_dict["summary"])

五、任务高级功能

1. 任务依赖与上下文（Context & Dependencies）

通过 context 属性定义任务依赖，确保任务仅在依赖任务完成后执行，支持多依赖配置：

# 依赖1：研究AI进展
research_task = Task(
    description="研究2025年AI最新进展",
    expected_output="AI进展要点列表",
    agent=researcher
)

# 依赖2：研究AI应用案例
case_task = Task(
    description="收集AI在企业中的应用案例",
    expected_output="应用案例列表",
    agent=researcher,
    async_execution=True  # 异步执行（不阻塞其他任务）
)

# 主任务：依赖前两个任务的输出，生成报告
report_task = Task(
    description="结合AI进展与应用案例，生成分析报告",
    expected_output="结构化分析报告",
    agent=writer,
    context=[research_task, case_task]  # 等待两个依赖任务完成
)

2. 任务护栏（Guardrails）

用于验证/转换任务输出，确保数据质量，支持三种配置方式：

（1）自定义验证函数

import json
from typing import Tuple

# 验证输出是否为合法JSON
def validate_json(result: str) -> Tuple[bool, str]:
    try:
        json.loads(result)
        return (True, result)  # 验证通过，返回原始结果
    except json.JSONDecodeError:
        return (False, "输出必须是合法的JSON格式")  # 验证失败，返回错误信息

# 绑定护栏到任务
json_task = Task(
    description="生成JSON格式的AI新闻",
    expected_output="合法JSON字符串",
    agent=analyst,
    guardrail=validate_json,
    max_retries=3  # 验证失败时，智能体最多重试3次
)

（2）无代码文本指令

直接用文本描述验证规则，适合简单场景：

task = Task(
    description="生成产品价格列表",
    expected_output="包含产品名与价格的列表",
    agent=analyst,
    guardrail="确保每个价格格式为数字，且不包含敏感信息"  # 文本指令
)

（3）LLM 护栏（LLMGuardrail）

用指定 LLM 进行复杂验证，适合需要语义理解的场景：

from crewai import LLMGuardrail
from crewai.llm import LLM

task = Task(
    description="生成AI伦理报告",
    expected_output="符合伦理规范的分析内容",
    agent=writer,
    guardrail=LLMGuardrail(
        description="确保报告不包含歧视性内容，且引用权威来源",
        llm=LLM(model="gpt-4o-mini")  # 指定用于验证的LLM
    )
)

3. 异步执行（Async Execution）

通过 async_execution=True 启用异步任务，团队不等待该任务完成，适用于耗时且非关键路径的任务：

# 异步任务：耗时的数据收集（不阻塞报告任务）
data_task = Task(
    description="收集过去5年AI市场规模数据",
    expected_output="年度市场规模列表",
    agent=researcher,
    async_execution=True
)

# 主任务：无需等待数据任务，先完成报告框架
framework_task = Task(
    description="生成AI市场分析报告框架",
    expected_output="报告章节结构",
    agent=writer
)

# 后续任务：等待数据任务完成，填充报告内容
fill_task = Task(
    description="用市场规模数据填充报告",
    expected_output="完整报告",
    agent=writer,
    context=[data_task, framework_task]  # 等待两个任务完成
)

4. 回调机制（Callback）

任务完成后自动执行回调函数，支持通知、日志记录等后续操作：

from crewai import TaskOutput

# 任务完成后发送通知
def task_callback(output: TaskOutput):
    print(f"""
        任务完成通知：
        任务描述：{output.description}
        执行智能体：{output.agent}
        原始输出：{output.raw[:100]}...  # 仅显示前100字符
    """)

# 绑定回调到任务
research_task = Task(
    description="研究AI最新趋势",
    expected_output="趋势要点列表",
    agent=researcher,
    callback=task_callback  # 任务完成后执行 callback 函数
)

六、最佳实践与故障排除

1. 核心最佳实践

（1）任务描述与输出明确化

• 避免模糊表述（如“研究AI”→“研究2025年AI在医疗领域的3个核心应用”）；
• 明确输出格式（如“Markdown 表格”“JSON 数组”），减少智能体理解偏差。

（2）合理配置依赖与异步

• 关键路径任务用 sequential 流程，非关键任务用 async_execution 提升效率；
• 依赖任务数量控制在3个以内，避免复杂依赖导致的执行延迟。

（3）输出结构化与验证

• 跨任务传递数据时，必选 output_pydantic 或 output_json，确保格式一致；
• 敏感场景（如金融、法律）添加 guardrail，验证输出准确性与安全性。

（4）工具精细化分配

• 任务级工具（tools 属性）优先于智能体默认工具，避免智能体使用无关工具；
• 复杂任务（如代码开发）明确指定工具（如 CodeExecutionTool），并启用 safe 模式。

2. 常见故障排除

（1）任务执行顺序异常

• 原因：流程类型与依赖配置不匹配（如 hierarchical 流程下强行指定顺序）；
• 解决方案：明确依赖用 sequential 流程，无依赖用 hierarchical 流程；或通过 context 强制指定依赖。

（2）输出格式不符合预期

• 原因：未配置 output_pydantic/output_json，或 expected_output 描述模糊；
• 解决方案：添加结构化输出配置，或在 expected_output 中明确格式（如“每个要点以 - 开头”）。

（3）护栏验证频繁失败

• 原因：验证规则过严，或智能体理解偏差；
• 解决方案：简化验证规则，或在 guardrail 中添加示例（如“正确格式：{“name”:“产品A”,“price”:99}”）。

（4）异步任务未被等待

• 原因：后续任务未通过 context 依赖异步任务；
• 解决方案：确保依赖异步任务的后续任务，在 context 中包含该异步任务。

总结

CrewAI 任务是智能体执行工作的核心单元，通过清晰的属性配置、灵活的执行流程与强大的输出管理，支持从简单查询到复杂多智能体协作的全场景需求。创建时推荐用 YAML 配置提升可维护性，实际使用中需根据任务类型（如研究、报告、代码开发）适配关键属性（如 context、output_pydantic、guardrail），并遵循依赖管理与数据验证的最佳实践，确保任务高效、准确执行。