任务

关于在 CrewAI 框架内管理和创建任务的详细指南。

概述

在 CrewAI 框架中，Task 是由 Agent 完成的具体任务指派。

任务提供了执行所需的全部必要细节，例如描述、负责的智能体、所需工具等，从而支撑各种复杂度的操作。

CrewAI 中的任务可以具有协作性，需要多个智能体共同完成。这通过任务属性来管理，并由团队的流程进行编排，从而提升团队协作与效率。

NOTE

CrewAI AMP 在 Crew Studio 中提供了可视化任务构建器，可简化复杂任务的创建与串联。无需编写代码，即可可视化地设计任务流程并实时测试。

可视化任务构建器支持：

拖拽式任务创建
可视化任务依赖与流程
实时测试与验证
便捷的分享与协作

任务执行流程

任务可以通过两种方式执行：

顺序流程：任务按定义的顺序依次执行
层级流程：根据智能体的角色和专长将任务分配给它们

执行流程在创建团队时定义：

python

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential  # or Process.hierarchical
)

任务属性

属性	参数	类型	描述
描述	`description`	`str`	对任务内容的清晰简洁说明。
预期输出	`expected_output`	`str`	对任务完成形态的详细描述。
名称 (可选)	`name`	`Optional[str]`	任务的名称标识符。
智能体 (可选)	`agent`	`Optional[BaseAgent]`	负责执行该任务的智能体。
工具 (可选)	`tools`	`List[BaseTool]`	该智能体在此任务中限定使用的工具/资源。
上下文 (可选)	`context`	`Optional[List["Task"]]`	其他任务，其输出将作为该任务的上下文。
异步执行 (可选)	`async_execution`	`Optional[bool]`	任务是否应异步执行。默认为 False。
人工输入 (可选)	`human_input`	`Optional[bool]`	任务是否需要人工审核智能体的最终答案。默认为 False。
Markdown (可选)	`markdown`	`Optional[bool]`	任务是否应指示智能体以 Markdown 格式返回最终答案。默认为 False。
配置 (可选)	`config`	`Optional[Dict[str, Any]]`	任务特定的配置参数。
输出文件 (可选)	`output_file`	`Optional[str]`	用于存储任务输出的文件路径。
创建目录 (可选)	`create_directory`	`Optional[bool]`	若 `output\_file` 的目录不存在，是否创建该目录。默认为 True。
输出 JSON (可选)	`output_json`	`Optional[Type[BaseModel]]`	用于结构化 JSON 输出的 Pydantic 模型。
输出 Pydantic (可选)	`output_pydantic`	`Optional[Type[BaseModel]]`	用于任务输出的 Pydantic 模型。
回调 (可选)	`callback`	`Optional[Any]`	任务完成后执行的函数/对象。
护栏 (可选)	`guardrail`	`Optional[Callable]`	在进入下一个任务之前验证任务输出的函数。
护栏 (可选)	`guardrails`	`Optional[List[Callable]]`	在进入下一个任务之前验证任务输出的护栏列表。
护栏最大重试次数 (可选)	`guardrail_max_retries`	`Optional[int]`	护栏验证失败时的最大重试次数。默认为 3。

NOTE

任务属性 max_retries 已废弃，将在 v1.0.0 中移除。请改用 guardrail_max_retries 来控制护栏失败时的重试次数。

创建任务

在 CrewAI 中创建任务有两种常见方式：使用 JSONC 项目配置（推荐用于新团队），或直接在代码中定义任务。

JSONC 配置（推荐）

使用 crewai create crew <name> 创建的新项目在 crew.jsonc 中定义任务。agents 数组指向 agents/ 中的文件，tasks 数组定义团队应按顺序执行的工作。

按照安装部分所述创建 CrewAI 项目后，编辑生成的 crew.jsonc。

NOTE

在任务的 description、expected_output 和 output_file 中使用 {placeholder} 值。将默认值放在顶层的 inputs 对象中；crewai run 会提示补全缺失的值。

下面是一个包含两个有序任务的 crew.jsonc 示例：

jsonc

{
  "name": "Research Crew",
  "agents": ["researcher", "reporting_analyst"],
  "tasks": [
    {
      "name": "research_task",
      "description": "Conduct thorough research about {topic}. Include current and relevant information.",
      "expected_output": "A list of the most relevant information about {topic}.",
      "agent": "researcher"
    },
    {
      "name": "reporting_task",
      "description": "Review the research and expand it into a detailed report.",
      "expected_output": "A polished markdown report without fenced code blocks.",
      "agent": "reporting_analyst",
      "context": ["research_task"],
      "markdown": true,
      "output_file": "report.md"
    }
  ],
  "inputs": {
    "topic": "AI Agents"
  }
}

每个任务都必须包含 description 和 expected_output。agent 的值应与 agents 中列出的智能体名称一致。context 是先前任务名称的列表；不接受前向引用，以便顺序上下文保持明确。

任务条目支持任何公开的 Task 字段。常见字段包括 name、agent、context、output_file、tools、human_input、async_execution、guardrail、guardrails、guardrail_max_retries、markdown、input_files、output_json、output_pydantic、response_model 和 converter_cls。对于条件任务，请使用 "type": "ConditionalTask" 并配合 condition 字段。

经典 YAML 配置

使用 crewai create crew <name> --classic 创建的经典项目使用 config/tasks.yaml 以及 crew.py 中的 @CrewBase 类。对于现有的 YAML 项目或偏好基于装饰器的 Python 装配方式的团队，此方式仍受支持。

直接在代码中定义（替代方式）

或者，你也可以不使用 YAML 配置，直接在代码中定义任务：

python

from crewai import Task

research_task = Task(
    description="""
        Conduct a thorough research about AI Agents.
        Make sure you find any interesting and relevant information given
        the current year is 2025.
    """,
    expected_output="""
        A list with 10 bullet points of the most relevant information about AI Agents
    """,
    agent=researcher
)

reporting_task = Task(
    description="""
        Review the context you got and expand each topic into a full section for a report.
        Make sure the report is detailed and contains any and all relevant information.
    """,
    expected_output="""
        A fully fledge reports with the mains topics, each with a full section of information.
    """,
    agent=reporting_analyst,
    markdown=True,  # Enable markdown formatting for the final output
    output_file="report.md"
)

TIP

直接指定 agent 来分配任务，或者让 hierarchical 的 CrewAI 流程根据角色、可用性等来决定。

任务输出

理解任务输出对于构建高效的 AI 工作流至关重要。CrewAI 通过 TaskOutput 类提供了一种结构化的任务结果处理方式，它支持多种输出格式，并且可以轻松地在任务之间传递。

在 CrewAI 框架中，任务的输出被封装在 TaskOutput 类中。该类提供了一种结构化的方式来访问任务结果，包括原始输出、JSON 和 Pydantic 模型等多种格式。

默认情况下，TaskOutput 只包含 raw 输出。只有当原始 Task 对象分别配置了 output_pydantic 或 output_json 时，TaskOutput 才会包含 pydantic 或 json_dict 输出。

任务输出属性

属性	参数	类型	描述
描述	`description`	`str`	任务的描述。
摘要	`summary`	`Optional[str]`	任务摘要，由描述的前 10 个词自动生成。
原始	`raw`	`str`	任务的原始输出。这是输出的默认格式。
Pydantic	`pydantic`	`Optional[BaseModel]`	表示任务结构化输出的 Pydantic 模型对象。
JSON 字典	`json_dict`	`Optional[Dict[str, Any]]`	表示任务 JSON 输出的字典。
智能体	`agent`	`str`	执行该任务的智能体。
输出格式	`output_format`	`OutputFormat`	任务输出的格式，选项包括 RAW、JSON 和 Pydantic。默认是 RAW。
消息	`messages`	`list[LLMMessage]`	上次任务执行产生的消息。

任务方法与属性

方法/属性	描述
json	如果输出格式为 JSON，则返回任务输出的 JSON 字符串表示。
to_dict	将 JSON 和 Pydantic 输出转换为字典。
str	返回任务输出的字符串表示，优先级为 Pydantic、JSON、raw。

访问任务输出

任务执行后，可以通过 Task 对象的 output 属性访问其输出。TaskOutput 类提供了多种方式与该输出交互并呈现它。

示例

python

# Example task
task = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool]
)

# Execute the crew
crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=True
)

result = crew.kickoff()

# Accessing the task output
task_output = task.output

print(f"Task Description: {task_output.description}")
print(f"Task Summary: {task_output.summary}")
print(f"Raw Output: {task_output.raw}")
if task_output.json_dict:
    print(f"JSON Output: {json.dumps(task_output.json_dict, indent=2)}")
if task_output.pydantic:
    print(f"Pydantic Output: {task_output.pydantic}")

Markdown 输出格式

markdown 参数可为任务输出启用自动 Markdown 格式化。设置为 True 时，任务会指示智能体使用正确的 Markdown 语法格式化最终答案。

使用 Markdown 格式化

python

# Example task with markdown formatting enabled
formatted_task = Task(
    description="Create a comprehensive report on AI trends",
    expected_output="A well-structured report with headers, sections, and bullet points",
    agent=reporter_agent,
    markdown=True  # Enable automatic markdown formatting
)

当 markdown=True 时，智能体将收到额外指令，要求使用以下方式格式化输出：

# 用于标题
**text** 用于粗体文本
*text* 用于斜体文本
- 或 * 用于项目符号
`code` 用于行内代码
language ``` 用于代码块

使用 Markdown 的 YAML 配置

yaml

analysis_task:
  description: >
    Analyze the market data and create a detailed report
  expected_output: >
    A comprehensive analysis with charts and key findings
  agent: analyst
  markdown: true # Enable markdown formatting
  output_file: analysis.md

Markdown 输出的优势

一致的格式：确保所有输出遵循正确的 Markdown 约定
更好的可读性：使用标题、列表和强调的结构化内容
可直接用于文档：输出可直接用于文档系统
跨平台兼容性：Markdown 得到普遍支持

NOTE

当 markdown=True 时，Markdown 格式化指令会自动添加到任务提示词中，因此你无需在任务描述中指定格式要求。

任务依赖与上下文

任务可以使用 context 属性依赖其他任务的输出。例如：

python

research_task = Task(
    description="Research the latest developments in AI",
    expected_output="A list of recent AI developments",
    agent=researcher
)

analysis_task = Task(
    description="Analyze the research findings and identify key trends",
    expected_output="Analysis report of AI trends",
    agent=analyst,
    context=[research_task]  # This task will wait for research_task to complete
)

任务护栏

任务护栏提供了一种方式，在任务输出传递给下一个任务之前对其进行验证和转换。此功能有助于确保数据质量，并在智能体输出不符合特定标准时向智能体提供反馈。

CrewAI 支持两种类型的护栏：

基于函数的护栏：具有自定义验证逻辑的 Python 函数，让你完全控制验证过程，并确保结果可靠且确定。
基于 LLM 的护栏：使用智能体 LLM 根据自然语言标准验证输出的字符串描述。这些非常适合复杂或主观的验证要求。

基于函数的护栏

要向任务添加基于函数的护栏，请通过 guardrail 参数提供验证函数：

python

from typing import Tuple, Union, Dict, Any
from crewai import TaskOutput

def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    """Validate blog content meets requirements."""
    try:
        # Check word count
        word_count = len(result.raw.split())
        if word_count > 200:
            return (False, "Blog content exceeds 200 words")

        # Additional validation logic here
        return (True, result.raw.strip())
    except Exception as e:
        return (False, "Unexpected error during validation")

blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A blog post under 200 words",
    agent=blog_agent,
    guardrail=validate_blog_content  # Add the guardrail function
)

基于 LLM 的护栏（字符串描述）

无需编写自定义验证函数，你可以使用利用基于 LLM 验证的字符串描述。当你向 guardrail 或 guardrails 参数提供字符串时，CrewAI 会自动创建一个 LLMGuardrail，它使用智能体的 LLM 根据你的描述验证输出。

要求：

任务必须分配有 agent（护栏使用该智能体的 LLM）
提供清晰、描述性的字符串说明验证标准

python

from crewai import Task

# Single LLM-based guardrail
blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A blog post under 200 words",
    agent=blog_agent,
    guardrail="The blog post must be under 200 words and contain no technical jargon"
)

基于 LLM 的护栏特别适用于：

难以用程序表达的复杂验证逻辑
主观标准，如语气、风格或质量评估
更易于描述而非用代码表达的自然语言要求

LLM 护栏将：

根据你的描述分析任务输出
如果输出符合标准，则返回 (True, output)
如果验证失败，则返回带有具体反馈的 (False, feedback)

带有详细验证标准的示例：

python

research_task = Task(
    description="Research the latest developments in quantum computing",
    expected_output="A comprehensive research report",
    agent=researcher_agent,
    guardrail="""
    The research report must:
    - Be at least 1000 words long
    - Include at least 5 credible sources
    - Cover both technical and practical applications
    - Be written in a professional, academic tone
    - Avoid speculation or unverified claims
    """
)

多个护栏

你可以使用 guardrails 参数向任务应用多个护栏。多个护栏按顺序执行，每个护栏接收前一个护栏的输出。这允许你将验证和转换步骤串联起来。

guardrails 参数接受：

护栏函数或字符串描述的列表
单个护栏函数或字符串（与 guardrail 相同）

注意：如果提供了 guardrails，它将优先于 guardrail。设置 guardrails 时，guardrail 参数将被忽略。

python

from typing import Tuple, Any
from crewai import TaskOutput, Task

def validate_word_count(result: TaskOutput) -> Tuple[bool, Any]:
    """Validate word count is within limits."""
    word_count = len(result.raw.split())
    if word_count < 100:
        return (False, f"Content too short: {word_count} words. Need at least 100 words.")
    if word_count > 500:
        return (False, f"Content too long: {word_count} words. Maximum is 500 words.")
    return (True, result.raw)

def validate_no_profanity(result: TaskOutput) -> Tuple[bool, Any]:
    """Check for inappropriate language."""
    profanity_words = ["badword1", "badword2"]  # Example list
    content_lower = result.raw.lower()
    for word in profanity_words:
        if word in content_lower:
            return (False, f"Inappropriate language detected: {word}")
    return (True, result.raw)

def format_output(result: TaskOutput) -> Tuple[bool, Any]:
    """Format and clean the output."""
    formatted = result.raw.strip()
    # Capitalize first letter
    formatted = formatted[0].upper() + formatted[1:] if formatted else formatted
    return (True, formatted)

# Apply multiple guardrails sequentially
blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A well-formatted blog post between 100-500 words",
    agent=blog_agent,
    guardrails=[
        validate_word_count,      # First: validate length
        validate_no_profanity,    # Second: check content
        format_output             # Third: format the result
    ],
    guardrail_max_retries=3
)

在此示例中，护栏按顺序执行：

validate_word_count 检查字数
validate_no_profanity 检查不当语言（使用第 1 步的输出）
format_output 格式化最终结果（使用第 2 步的输出）

如果任何护栏失败，错误会发送回智能体，并且任务最多重试 guardrail_max_retries 次。

混合使用基于函数和基于 LLM 的护栏：

你可以在同一个列表中组合基于函数和基于字符串的护栏：

python

from typing import Tuple, Any
from crewai import TaskOutput, Task

def validate_word_count(result: TaskOutput) -> Tuple[bool, Any]:
    """Validate word count is within limits."""
    word_count = len(result.raw.split())
    if word_count < 100:
        return (False, f"Content too short: {word_count} words. Need at least 100 words.")
    if word_count > 500:
        return (False, f"Content too long: {word_count} words. Maximum is 500 words.")
    return (True, result.raw)

# Mix function-based and LLM-based guardrails
blog_task = Task(
    description="Write a blog post about AI",
    expected_output="A well-formatted blog post between 100-500 words",
    agent=blog_agent,
    guardrails=[
        validate_word_count,  # Function-based: precise word count check
        "The content must be engaging and suitable for a general audience",  # LLM-based: subjective quality check
        "The writing style should be clear, concise, and free of technical jargon"  # LLM-based: style validation
    ],
    guardrail_max_retries=3
)

这种方法将程序化验证的精确性与基于 LLM 评估主观标准的灵活性结合起来。

护栏函数要求

函数签名：
- 必须恰好接受一个参数（任务输出）
- 应返回 (bool, Any) 元组
- 建议使用类型提示，但它是可选的
返回值：
- 成功时：返回 (bool, Any) 元组。例如：(True, validated_result)
- 失败时：返回 (bool, str) 元组。例如：(False, "Error message explain the failure")

错误处理最佳实践

结构化错误响应：

python

from crewai import TaskOutput, LLMGuardrail

def validate_with_context(result: TaskOutput) -> Tuple[bool, Any]:
    try:
        # Main validation logic
        validated_data = perform_validation(result)
        return (True, validated_data)
    except ValidationError as e:
        return (False, f"VALIDATION_ERROR: {str(e)}")
    except Exception as e:
        return (False, str(e))

错误类别：
- 使用具体的错误代码
- 包含相关上下文
- 提供可操作的反馈
验证链：

python

from typing import Any, Dict, List, Tuple, Union
from crewai import TaskOutput

def complex_validation(result: TaskOutput) -> Tuple[bool, Any]:
    """Chain multiple validation steps."""
    # Step 1: Basic validation
    if not result:
        return (False, "Empty result")

    # Step 2: Content validation
    try:
        validated = validate_content(result)
        if not validated:
            return (False, "Invalid content")

        # Step 3: Format validation
        formatted = format_output(validated)
        return (True, formatted)
    except Exception as e:
        return (False, str(e))

处理护栏结果

当护栏返回 (False, error) 时：

错误会发送回智能体
智能体尝试修复问题
该过程重复，直到：
- 护栏返回 (True, result)
- 达到最大重试次数（guardrail_max_retries）

带重试处理的示例：

python

from typing import Optional, Tuple, Union
from crewai import TaskOutput, Task

def validate_json_output(result: TaskOutput) -> Tuple[bool, Any]:
    """Validate and parse JSON output."""
    try:
        # Try to parse as JSON
        data = json.loads(result)
        return (True, data)
    except json.JSONDecodeError as e:
        return (False, "Invalid JSON format")

task = Task(
    description="Generate a JSON report",
    expected_output="A valid JSON object",
    agent=analyst,
    guardrail=validate_json_output,
    guardrail_max_retries=3  # Limit retry attempts
)

从任务获取一致的结构化输出

NOTE

同样重要的是，团队最终任务的输出会成为实际团队本身的最终输出。

使用 `output_pydantic`

output_pydantic 属性允许你定义一个任务输出应遵循的 Pydantic 模型。这确保输出不仅是结构化的，而且还根据 Pydantic 模型进行验证。

以下示例演示如何使用 output_pydantic：

python

import json

from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel

class Blog(BaseModel):
    title: str
    content: str

blog_agent = Agent(
    role="Blog Content Generator Agent",
    goal="Generate a blog title and content",
    backstory="""You are an expert content creator, skilled in crafting engaging and informative blog posts.""",
    verbose=False,
    allow_delegation=False,
    llm="gpt-4o",
)

task1 = Task(
    description="""Create a blog title and content on a given topic. Make sure the content is under 200 words.""",
    expected_output="A compelling blog title and well-written content.",
    agent=blog_agent,
    output_pydantic=Blog,
)

# Instantiate your crew with a sequential process
crew = Crew(
    agents=[blog_agent],
    tasks=[task1],
    verbose=True,
    process=Process.sequential,
)

result = crew.kickoff()

# Option 1: Accessing Properties Using Dictionary-Style Indexing
print("Accessing Properties - Option 1")
title = result["title"]
content = result["content"]
print("Title:", title)
print("Content:", content)

# Option 2: Accessing Properties Directly from the Pydantic Model
print("Accessing Properties - Option 2")
title = result.pydantic.title
content = result.pydantic.content
print("Title:", title)
print("Content:", content)

# Option 3: Accessing Properties Using the to_dict() Method
print("Accessing Properties - Option 3")
output_dict = result.to_dict()
title = output_dict["title"]
content = output_dict["content"]
print("Title:", title)
print("Content:", content)

# Option 4: Printing the Entire Blog Object
print("Accessing Properties - Option 5")
print("Blog:", result)

在此示例中：

定义了一个包含 title 和 content 字段的 Pydantic 模型 Blog。
任务 task1 使用 output_pydantic 属性指定其输出应遵循 Blog 模型。
执行团队后，你可以通过多种方式访问结构化输出，如下所示。

访问输出的说明

字典式索引：你可以使用 result["field_name"] 直接访问字段。这之所以可行，是因为 CrewOutput 类实现了 getitem 方法。
直接从 Pydantic 模型访问：直接从 result.pydantic 对象访问属性。
使用 to_dict() 方法：将输出转换为字典并访问字段。
打印整个对象：只需打印 result 对象即可查看结构化输出。

使用 `output_json`

output_json 属性允许你以 JSON 格式定义预期输出。这确保任务输出是有效的 JSON 结构，便于在你的应用程序中解析和使用。

以下示例演示如何使用 output_json：

python

import json

from crewai import Agent, Crew, Process, Task
from pydantic import BaseModel

# Define the Pydantic model for the blog
class Blog(BaseModel):
    title: str
    content: str

# Define the agent
blog_agent = Agent(
    role="Blog Content Generator Agent",
    goal="Generate a blog title and content",
    backstory="""You are an expert content creator, skilled in crafting engaging and informative blog posts.""",
    verbose=False,
    allow_delegation=False,
    llm="gpt-4o",
)

# Define the task with output_json set to the Blog model
task1 = Task(
    description="""Create a blog title and content on a given topic. Make sure the content is under 200 words.""",
    expected_output="A JSON object with 'title' and 'content' fields.",
    agent=blog_agent,
    output_json=Blog,
)

# Instantiate the crew with a sequential process
crew = Crew(
    agents=[blog_agent],
    tasks=[task1],
    verbose=True,
    process=Process.sequential,
)

# Kickoff the crew to execute the task
result = crew.kickoff()

# Option 1: Accessing Properties Using Dictionary-Style Indexing
print("Accessing Properties - Option 1")
title = result["title"]
content = result["content"]
print("Title:", title)
print("Content:", content)

# Option 2: Printing the Entire Blog Object
print("Accessing Properties - Option 2")
print("Blog:", result)

在此示例中：

定义了一个包含 title 和 content 字段的 Pydantic 模型 Blog，用于指定 JSON 输出的结构。
任务 task1 使用 output_json 属性表示它期望符合 Blog 模型的 JSON 输出。
执行团队后，你可以通过两种方式访问结构化 JSON 输出，如下所示。

访问输出的说明

使用字典式索引访问属性：你可以使用 result["field_name"] 直接访问字段。这之所以可行，是因为 CrewOutput 类实现了 getitem 方法，允许你将输出视为字典。在此选项中，我们从 result 中检索 title 和 content。
打印整个 Blog 对象：通过打印 result，你会得到 CrewOutput 对象的字符串表示。由于 str 方法被实现为返回 JSON 输出，因此这会将整个输出显示为表示 Blog 对象的格式化字符串。

通过使用 output_pydantic 或 output_json，你可以确保任务以一致且结构化的格式生成输出，从而更轻松地在应用程序内或跨多个任务处理和利用数据。

将工具与任务集成

利用 CrewAI 工具包和 LangChain 工具中的工具，以增强任务性能和智能体交互。

使用工具创建任务

python

import os
os.environ["OPENAI_API_KEY"] = "Your Key"
os.environ["SERPER_API_KEY"] = "Your Key" # serper.dev API key

from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool

research_agent = Agent(
  role='Researcher',
  goal='Find and summarize the latest AI news',
  backstory="""You're a researcher at a large company.
  You're responsible for analyzing data and providing insights
  to the business.""",
  verbose=True
)

# to perform a semantic search for a specified query from a text's content across the internet
search_tool = SerperDevTool()

task = Task(
  description='Find and summarize the latest AI news',
  expected_output='A bullet list summary of the top 5 most important AI news',
  agent=research_agent,
  tools=[search_tool]
)

crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=True
)

result = crew.kickoff()
print(result)

这演示了带有特定工具的任务如何覆盖智能体的默认工具集，以实现量身定制的任务执行。

引用其他任务

在 CrewAI 中，一个任务的输出会自动传递到下一个任务，但你可以专门定义应将哪些任务（包括多个任务）的输出用作另一个任务的上下文。

当你有一个任务依赖于另一个并非紧接着执行的任务的输出时，这很有用。这通过任务的 context 属性完成：

python

# ...

research_ai_task = Task(
    description="Research the latest developments in AI",
    expected_output="A list of recent AI developments",
    async_execution=True,
    agent=research_agent,
    tools=[search_tool]
)

research_ops_task = Task(
    description="Research the latest developments in AI Ops",
    expected_output="A list of recent AI Ops developments",
    async_execution=True,
    agent=research_agent,
    tools=[search_tool]
)

write_blog_task = Task(
    description="Write a full blog post about the importance of AI and its latest news",
    expected_output="Full blog post that is 4 paragraphs long",
    agent=writer_agent,
    context=[research_ai_task, research_ops_task]
)

#...

异步执行

你可以将任务定义为异步执行。这意味着团队不会等待它完成后再继续执行下一个任务。这对于需要很长时间才能完成，或对接下来要执行的任务并不关键的任务很有用。

然后，你可以在后续任务中使用 context 属性定义它应等待异步任务的输出完成。

python

#...

list_ideas = Task(
    description="List of 5 interesting ideas to explore for an article about AI.",
    expected_output="Bullet point list of 5 ideas for an article.",
    agent=researcher,
    async_execution=True # Will be executed asynchronously
)

list_important_history = Task(
    description="Research the history of AI and give me the 5 most important events.",
    expected_output="Bullet point list of 5 important events.",
    agent=researcher,
    async_execution=True # Will be executed asynchronously
)

write_article = Task(
    description="Write an article about AI, its history, and interesting ideas.",
    expected_output="A 4 paragraph article about AI.",
    agent=writer,
    context=[list_ideas, list_important_history] # Will wait for the output of the two tasks to be completed
)

#...

回调机制

回调函数在任务完成后执行，允许根据任务结果触发操作或通知。

python

# ...

def callback_function(output: TaskOutput):
    # Do something after the task is completed
    # Example: Send an email to the manager
    print(f"""
        Task completed!
        Task: {output.description}
        Output: {output.raw}
    """)

research_task = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool],
    callback=callback_function
)

#...

访问特定任务的输出

团队运行完成后，你可以使用任务对象的 output 属性访问特定任务的输出：

python

# ...
task1 = Task(
    description='Find and summarize the latest AI news',
    expected_output='A bullet list summary of the top 5 most important AI news',
    agent=research_agent,
    tools=[search_tool]
)

#...

crew = Crew(
    agents=[research_agent],
    tasks=[task1, task2, task3],
    verbose=True
)

result = crew.kickoff()

# Returns a TaskOutput object with the description and results of the task
print(f"""
    Task completed!
    Task: {task1.output.description}
    Output: {task1.output.raw}
""")

工具覆盖机制

在任务中指定工具可以动态调整智能体能力，体现 CrewAI 的灵活性。

错误处理与验证机制

在创建和执行任务时，内置了某些验证机制，以确保任务属性的健壮性和可靠性。这些机制包括但不限于：

确保每个任务只设置一种输出类型，以保持清晰的输出预期。
防止手动分配 id 属性，以维护唯一标识符系统的完整性。

这些验证有助于维护 crewAI 框架内任务执行的一致性和可靠性。

保存文件时创建目录

create_directory 参数控制 CrewAI 在将任务输出保存到文件时是否应自动创建目录。该特性对于组织输出、确保文件路径结构正确特别有用，在处理复杂的项目层级结构时尤为如此。

默认行为

默认情况下 create_directory=True，这意味着 CrewAI 会自动创建输出文件路径中缺失的任何目录：

python

# Default behavior - directories are created automatically
report_task = Task(
    description='Generate a comprehensive market analysis report',
    expected_output='A detailed market analysis with charts and insights',
    agent=analyst_agent,
    output_file='reports/2025/market_analysis.md',  # Creates 'reports/2025/' if it doesn't exist
    markdown=True
)

禁用目录创建

如果你想阻止自动创建目录并确保目录已存在，请设置 create_directory=False：

python

# Strict mode - directory must already exist
strict_output_task = Task(
    description='Save critical data that requires existing infrastructure',
    expected_output='Data saved to pre-configured location',
    agent=data_agent,
    output_file='secure/vault/critical_data.json',
    create_directory=False  # Will raise RuntimeError if 'secure/vault/' doesn't exist
)

YAML 配置

你也可以在 YAML 任务定义中配置此行为：

yaml

analysis_task:
  description: >
    Generate quarterly financial analysis
  expected_output: >
    A comprehensive financial report with quarterly insights
  agent: financial_analyst
  output_file: reports/quarterly/q4_2024_analysis.pdf
  create_directory: true # Automatically create 'reports/quarterly/' directory

audit_task:
  description: >
    Perform compliance audit and save to existing audit directory
  expected_output: >
    A compliance audit report
  agent: auditor
  output_file: audit/compliance_report.md
  create_directory: false # Directory must already exist

用例

自动创建目录（create_directory=True）：

开发与原型验证环境
按日期文件夹生成动态报告
目录结构可能变化的自动化工作流
带有用户专属文件夹的多租户应用

手动管理目录（create_directory=False）：

具有严格文件系统控制的生产环境
目录必须预先配置的安全敏感型应用
有特定权限要求的系统
对目录创建进行审计的合规环境

错误处理

当 create_directory=False 且目录不存在时，CrewAI 会抛出 RuntimeError：

python

try:
    result = crew.kickoff()
except RuntimeError as e:
    # Handle missing directory error
    print(f"Directory creation failed: {e}")
    # Create directory manually or use fallback location

观看下方视频，了解如何在 CrewAI 中使用结构化输出：

结论

任务（Task）是 CrewAI 中智能体行动的驱动力。通过恰当定义任务及其预期结果，你为 AI 智能体高效工作（无论是独立工作还是作为协作单元）奠定了基础。为任务配备合适的工具、理解执行流程、遵循稳健的校验实践，对于最大化发挥 CrewAI 的潜力至关重要，可确保智能体为各自的任务做好充分准备，并使任务按预期执行。

任务

概述

任务执行流程

任务属性

创建任务

JSONC 配置（推荐）

经典 YAML 配置

直接在代码中定义（替代方式）

任务输出

任务输出属性

任务方法与属性

访问任务输出

示例

Markdown 输出格式

使用 Markdown 格式化

使用 Markdown 的 YAML 配置

Markdown 输出的优势

任务依赖与上下文

任务护栏

基于函数的护栏

基于 LLM 的护栏（字符串描述）

多个护栏

护栏函数要求

错误处理最佳实践

处理护栏结果

从任务获取一致的结构化输出

使用 output_pydantic

访问输出的说明

使用 output_json

访问输出的说明

将工具与任务集成

使用工具创建任务

引用其他任务

异步执行

回调机制

访问特定任务的输出

工具覆盖机制

错误处理与验证机制

保存文件时创建目录

默认行为

禁用目录创建

YAML 配置

用例

错误处理

结论

友情链接

使用 `output_pydantic`

使用 `output_json`