任务

概览

在 CrewAI 框架中，Task 是由 Agent 完成的一项具体分配。

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

CrewAI 中的任务可以是协作式的，需要多个智能体共同完成。这由任务属性和团队的流程共同编排，从而提升协作效率。

Note

CrewAI AMP 在 Crew Studio 中提供可视化任务构建器，用于简化复杂任务的创建和串联。你可以无需编写代码，就以可视化方式设计任务流并实时测试。

可视化任务构建器支持：

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

任务执行流程

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

• 顺序式：按照定义顺序执行任务
• 分层式：根据角色和专长将任务分配给智能体

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

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 示例：

{
  "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。使用带有 condition 字段的 "type": "ConditionalTask" 来定义条件任务。

经典 YAML 配置

使用 crewai create crew <name> --classic 创建的经典项目会使用 config/tasks.yaml 和 crew.py 中的 @CrewBase 类。该模式仍然受支持，适用于现有 YAML 项目或更偏好基于装饰器的 Python 组织方式的团队。

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

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

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 团队流程根据角色、可用性等因素来决定。

任务输出

理解任务输出对构建有效的 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，最后是原始输出。

访问任务输出

任务执行完成后，你可以通过 Task 对象的 output 属性访问其输出。TaskOutput 类提供了多种方式来交互和呈现这一输出。

示例

# 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 格式化

# 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 配置

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 属性依赖其他任务的输出。例如：

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 支持两种护栏类型：

1. 基于函数的护栏：带有自定义验证逻辑的 Python 函数，可让你完全控制验证过程，并确保结果可靠、可确定。

2. 基于 LLM 的护栏：使用智能体 LLM 根据自然语言标准验证输出的字符串描述。这类方式非常适合复杂或主观的验证需求。

基于函数的护栏

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

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）
• 提供清晰、明确的字符串，说明验证标准

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 护栏会：

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

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

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 参数会被忽略。

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
)

在这个示例中，护栏按以下顺序执行：

1. validate_word_count 检查字数
2. validate_no_profanity 检查是否存在不当语言（使用步骤 1 的输出）
3. format_output 格式化最终结果（使用步骤 2 的输出）

如果任何护栏失败，错误会返回给智能体，并且任务会最多根据 guardrail_max_retries 进行重试。

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

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

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 评估主观标准的灵活性。

护栏函数要求

1. 函数签名：

   • 必须只接受一个参数（任务输出）
   • 应返回 (bool, Any) 元组
   • 建议添加类型注解，但不是必须

2. 返回值：

   • 成功时：返回 (bool, Any) 元组。例如：(True, validated_result)
   • 失败时：返回 (bool, str) 元组。例如：(False, "Error message explain the failure")

错误处理最佳实践

1. 结构化错误响应：

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))

2. 错误类别：

   • 使用具体的错误代码
   • 提供可操作的反馈
   • 在可能的情况下保留原始上下文

目录创建行为

create_directory 参数控制当 output_file 指向的目录不存在时，CrewAI 如何处理。

默认行为（create_directory=True）

• 自动为输出文件创建必要的目录
• 适用于大多数用例
• 方便新项目和动态路径

禁用目录创建（create_directory=False）

• 需要预先存在的目录结构
• 如果目录缺失则会失败
• 更适合对文件系统有严格控制的场景

何时禁用目录创建：

• 生产环境下有严格目录结构控制
• 需要预配置目录的安全敏感应用
• 具有特定权限要求的系统
• 目录创建需要审计的合规环境

错误处理

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

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 中使用结构化输出：

结论

任务是 CrewAI 中驱动智能体动作的核心。 通过正确定义任务及其结果，你就为 AI 智能体高效工作打下了基础，无论它们是独立运行还是作为协作单元运行。 为任务配备合适的工具、理解执行流程并遵循稳健的验证实践，对于最大化 CrewAI 的潜力至关重要， 这能确保智能体为自己的分配做好充分准备，并使任务按预期执行。