任务 - CrewAI 框架

任务概述

在 CrewAI 框架中，一个 Task 是由一个 Agent 完成的特定任务。

任务提供执行所需的所有详细信息，例如描述、负责的智能体、所需工具等，从而实现广泛的行动复杂性。

CrewAI 中的任务可以是协作式的，需要多个智能体协同工作。这通过任务属性进行管理，并由队伍的流程协调，从而提高团队协作和效率。

CrewAI 企业版在 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。
配置（可选）	`config`	`Optional[Dict[str, Any]]`	任务特定的配置参数。
输出文件（可选）	`output_file`	`Optional[str]`	用于存储任务输出的文件路径。
输出 JSON （可选）	`output_json`	`Optional[Type[BaseModel]]`	用于构建 JSON 输出的 Pydantic 模型。
输出 Pydantic （可选）	`output_pydantic`	`Optional[Type[BaseModel]]`	用于任务输出的 Pydantic 模型。
回调（可选））	`callback`	`Optional[Any]`	任务完成后要执行的函数/对象。

创建任务

在 CrewAI 中创建任务有两种方式：使用 YAML 配置（推荐）或直接在代码中定义。

YAML 配置（推荐）

使用 YAML 配置提供了一种更清晰、更易于维护的方式来定义任务。我们强烈建议在您的 CrewAI 项目中使用此方法定义任务。

按照安装部分所述创建 CrewAI 项目后，导航到 src/latest_ai_development/config/tasks.yaml 文件并修改模板以符合您的特定任务要求。

YAML 文件中的变量（例如 {topic}）在运行队伍时将替换为您输入的值

代码
crew.kickoff(inputs={'topic': 'AI Agents'})

以下是使用 YAML 配置任务的示例

tasks.yaml
research_task:
  description: >
    Conduct a thorough research about {topic}
    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 {topic}
  agent: researcher

reporting_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.
    Formatted as markdown without '```'
  agent: reporting_analyst
  output_file: report.md

要在代码中使用此 YAML 配置，请创建一个继承自 CrewBase 的队伍类

crew.py
# src/latest_ai_development/crew.py

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

@CrewBase
class LatestAiDevelopmentCrew():
  """LatestAiDevelopment crew"""

  @agent
  def researcher(self) -> Agent:
    return Agent(
      config=self.agents_config['researcher'], # type: ignore[index]
      verbose=True,
      tools=[SerperDevTool()]
    )

  @agent
  def reporting_analyst(self) -> Agent:
    return Agent(
      config=self.agents_config['reporting_analyst'], # type: ignore[index]
      verbose=True
    )

  @task
  def research_task(self) -> Task:
    return Task(
      config=self.tasks_config['research_task'] # type: ignore[index]
    )

  @task
  def reporting_task(self) -> Task:
    return Task(
      config=self.tasks_config['reporting_task'] # type: ignore[index]
    )

  @crew
  def crew(self) -> Crew:
    return Crew(
      agents=[
        self.researcher(),
        self.reporting_analyst()
      ],
      tasks=[
        self.research_task(),
        self.reporting_task()
      ],
      process=Process.sequential
    )

您在 YAML 文件（agents.yaml 和 tasks.yaml）中使用的名称应与 Python 代码中的方法名匹配。

直接代码定义（备选）

或者，您可以直接在代码中定义任务，而无需使用 YAML 配置

task.py
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.
        Formatted as markdown without '```'
    """,
    agent=reporting_analyst,
    output_file="report.md"
)

直接指定一个 agent 进行分配，或者让 分层 CrewAI 流程根据角色、可用性等决定。

任务输出

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

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

默认情况下，TaskOutput 只包含 raw 输出。TaskOutput 仅在原始 Task 对象配置了 output_pydantic 或 output_json 时，才包含 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。

任务方法和属性

方法/属性	描述
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}")

任务依赖和上下文

任务可以使用 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
)

任务护栏

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

使用任务护栏

要向任务添加护栏，请通过 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.split())
        if word_count > 200:
            return (False, "Blog content exceeds 200 words")

        # Additional validation logic here
        return (True, result.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
)

护栏函数要求

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

LLMGuardrail

LLMGuardrail 类提供了一种强大的机制来验证任务输出。

错误处理最佳实践

结构化错误响应:

代码
from crewai import TaskOutput

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

错误类别:
- 使用特定错误码
- 包含相关上下文
- 提供可操作的反馈
验证链:

代码
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)
- 达到最大重试次数

带重试处理的示例

代码
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,
    max_retries=3  # Limit retry attempts
)

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

同样重要的是要注意，队伍中最后一个任务的输出将成为整个队伍的最终输出。

使用 `output_pydantic`

output_pydantic 属性允许您定义任务输出应符合的 Pydantic 模型。这确保了输出不仅具有结构，而且根据 Pydantic 模型进行了验证。

以下是演示如何使用 output_pydantic 的示例

代码
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 的示例

代码
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 工具中的工具，以增强任务性能和智能体交互。

创建带工具的任务

代码
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 属性实现

代码
# ...

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 属性来定义它应该等待异步任务的输出完成。

代码
#...

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
)

#...

回调机制

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

代码
# ...

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 属性访问特定任务的输出

代码
# ...
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 框架内任务执行的一致性和可靠性。

任务护栏

任务护栏提供了一种强大的方式，可以在任务输出传递给下一个任务之前对其进行验证、转换或过滤。护栏是可选函数，在下一个任务开始之前执行，使您可以确保任务输出满足特定要求或格式。

基本用法

定义您自己的验证逻辑

代码
from typing import Tuple, Union
from crewai import Task

def validate_json_output(result: str) -> Tuple[bool, Union[dict, str]]:
    """Validate that the output is valid JSON."""
    try:
        json_data = json.loads(result)
        return (True, json_data)
    except json.JSONDecodeError:
        return (False, "Output must be valid JSON")

task = Task(
    description="Generate JSON data",
    expected_output="Valid JSON object",
    guardrail=validate_json_output
)

利用无代码方法进行验证

代码
from crewai import Task

task = Task(
    description="Generate JSON data",
    expected_output="Valid JSON object",
    guardrail="Ensure the response is a valid JSON object"
)

使用 YAML

research_task:
  ...
  guardrail: make sure each bullet contains a minimum of 100 words
  ...

代码
@CrewBase
class InternalCrew:
    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    ...
    @task
    def research_task(self):
        return Task(config=self.tasks_config["research_task"])  # type: ignore[index]
    ...

使用自定义模型生成代码

代码
from crewai import Task
from crewai.llm import LLM

task = Task(
    description="Generate JSON data",
    expected_output="Valid JSON object",
    guardrail=LLMGuardrail(
        description="Ensure the response is a valid JSON object",
        llm=LLM(model="gpt-4o-mini"),
    )
)

护栏工作原理

可选属性：护栏是任务级别的可选属性，允许您仅在需要时添加验证。
执行时机：护栏函数在下一个任务开始之前执行，确保任务之间的数据流有效。
返回格式：护栏必须返回一个 (success, data) 的元组
- 如果 success 为 True，则 data 是经过验证/转换的结果
- 如果 success 为 False，则 data 是错误消息
结果路由:
- 成功时（True），结果会自动传递给下一个任务
- 失败时（False），错误信息发送回智能体以生成新的答案

常见用例

数据格式验证

代码
def validate_email_format(result: str) -> Tuple[bool, Union[str, str]]:
    """Ensure the output contains a valid email address."""
    import re
    email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    if re.match(email_pattern, result.strip()):
        return (True, result.strip())
    return (False, "Output must be a valid email address")

内容过滤

代码
def filter_sensitive_info(result: str) -> Tuple[bool, Union[str, str]]:
    """Remove or validate sensitive information."""
    sensitive_patterns = ['SSN:', 'password:', 'secret:']
    for pattern in sensitive_patterns:
        if pattern.lower() in result.lower():
            return (False, f"Output contains sensitive information ({pattern})")
    return (True, result)

数据转换

代码
def normalize_phone_number(result: str) -> Tuple[bool, Union[str, str]]:
    """Ensure phone numbers are in a consistent format."""
    import re
    digits = re.sub(r'\D', '', result)
    if len(digits) == 10:
        formatted = f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"
        return (True, formatted)
    return (False, "Output must be a 10-digit phone number")

高级功能

链接多个验证

代码
def chain_validations(*validators):
    """Chain multiple validators together."""
    def combined_validator(result):
        for validator in validators:
            success, data = validator(result)
            if not success:
                return (False, data)
            result = data
        return (True, result)
    return combined_validator

# Usage
task = Task(
    description="Get user contact info",
    expected_output="Email and phone",
    guardrail=chain_validations(
        validate_email_format,
        filter_sensitive_info
    )
)

自定义重试逻辑

代码
task = Task(
    description="Generate data",
    expected_output="Valid data",
    guardrail=validate_data,
    max_retries=5  # Override default retry limit
)

保存文件时创建目录

您现在可以指定任务在将输出保存到文件时是否应创建目录。这对于组织输出和确保文件路径结构正确特别有用。

代码
# ...

save_output_task = Task(
    description='Save the summarized AI news to a file',
    expected_output='File saved successfully',
    agent=research_agent,
    tools=[file_save_tool],
    output_file='outputs/ai_news_summary.txt',
    create_directory=True
)

#...

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

结论

任务是 CrewAI 中智能体行动的驱动力。通过正确定义任务及其结果，您可以为您的 AI 智能体有效工作（无论是独立工作还是作为协作单元）奠定基础。为任务配备适当的工具、理解执行过程并遵循稳健的验证实践，对于最大化 CrewAI 的潜力至关重要，确保智能体有效准备好执行其分配的任务，并确保任务按预期执行。

智能体队伍

本页内容

任务概述
任务执行流程
任务属性
创建任务
YAML 配置（推荐）
直接代码定义（备选）
任务输出
任务输出属性
任务方法和属性
访问任务输出
示例
任务依赖和上下文
任务护栏
使用任务护栏
护栏函数要求
LLMGuardrail
错误处理最佳实践
处理护栏结果
从任务中获取结构化且一致的输出
使用 output_pydantic
访问输出的说明
使用 output_json
访问输出的说明
将工具与任务集成
创建带工具的任务
引用其他任务
异步执行
回调机制
访问特定任务输出
工具覆盖机制
错误处理和验证机制
任务护栏
基本用法
定义您自己的验证逻辑
利用无代码方法进行验证
使用 YAML
使用自定义模型生成代码
护栏工作原理
常见用例
数据格式验证
内容过滤
数据转换
高级功能
链接多个验证
自定义重试逻辑
保存文件时创建目录
结论

开始

指南

核心概念

工具

智能体监控与可观察性

学习

遥测

​任务概述

​任务执行流程

​任务属性

​创建任务

​YAML 配置（推荐）

​直接代码定义（备选）

​任务输出

​任务输出属性

​任务方法和属性

​访问任务输出

​示例

​任务依赖和上下文

​任务护栏

​使用任务护栏

​护栏函数要求

​LLMGuardrail

​错误处理最佳实践

​处理护栏结果

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

​使用 output_pydantic

​访问输出的说明

​使用 output_json

​访问输出的说明

​将工具与任务集成

​创建带工具的任务

​引用其他任务

​异步执行

​回调机制

​访问特定任务输出

​工具覆盖机制

​错误处理和验证机制

​任务护栏

​基本用法

​定义您自己的验证逻辑

​利用无代码方法进行验证

​使用 YAML

​使用自定义模型生成代码

​护栏工作原理

​常见用例

​数据格式验证

​内容过滤

​数据转换

​高级功能

​链接多个验证

​自定义重试逻辑

​保存文件时创建目录

​结论

任务概述

任务执行流程

任务属性

创建任务

YAML 配置（推荐）

直接代码定义（备选）

任务输出

任务输出属性

任务方法和属性

访问任务输出

示例

任务依赖和上下文

任务护栏

使用任务护栏

护栏函数要求

LLMGuardrail

错误处理最佳实践

处理护栏结果

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

使用 `output_pydantic`

访问输出的说明

使用 `output_json`

访问输出的说明

将工具与任务集成

创建带工具的任务

引用其他任务

异步执行

回调机制

访问特定任务输出

工具覆盖机制

错误处理和验证机制

任务护栏

基本用法

定义您自己的验证逻辑

利用无代码方法进行验证

使用 YAML

使用自定义模型生成代码

护栏工作原理

常见用例

数据格式验证

内容过滤

数据转换

高级功能

链接多个验证

自定义重试逻辑

保存文件时创建目录

结论