Task 详解 #
Task 概述 #
在 CrewAI 框架中，Task 是分配给 Agent 的具体工作单元。每个 Task 定义了：
- 具体的工作内容
- 预期的输出格式
- 负责执行的 Agent
- 执行所需的工具
Task 属性 #
核心属性 #
| 属性 | 类型 | 描述 |
|---|---|---|
| description | str | 任务的详细描述，说明要完成的工作 |
| expected_output | str | 预期的输出格式和内容 |
| agent | Agent | 负责执行任务的 Agent |
输出属性 #
| 属性 | 类型 | 描述 |
|---|---|---|
| output_file | str | 输出文件的路径 |
| output_json | Type[BaseModel] | JSON 格式输出的模型 |
| output_pydantic | Type[BaseModel] | Pydantic 模型输出 |
执行属性 #
| 属性 | 类型 | 默认值 | 描述 |
|---|---|---|---|
| tools | List[BaseTool] | [] | 任务特定的工具列表 |
| async_execution | bool | False | 是否异步执行 |
| context | List[Task] | [] | 提供上下文的任务列表 |
| callback | Callable | None | 任务完成时的回调函数 |
高级属性 #
| 属性 | 类型 | 描述 |
|---|---|---|
| config | Dict | 任务的额外配置 |
| create_directory | bool | 是否创建输出目录 |
创建 Task #
方式一：直接代码定义 #
python
from crewai import Task
task = Task(
description="分析 2025 年 AI 芯片市场的竞争格局",
expected_output="""
一份结构化的报告，包含：
- 主要厂商和产品
- 市场份额分析
- 技术发展趋势
- 投资建议
""",
agent=analyst_agent
)
方式二：YAML 配置（推荐） #
yaml
# config/tasks.yaml
research_task:
description: >
研究 {topic} 的最新趋势和发展。
重点关注：
1. 技术发展
2. 市场动态
3. 主要玩家
expected_output: >
一份包含 3 个主要趋势的详细报告，
每个趋势包括描述、关键技术和未来展望。
agent: researcher
write_task:
description: >
根据研究结果撰写一篇博客文章。
文章应该引人入胜、信息丰富。
expected_output: >
一篇 4 段落的 Markdown 格式博客文章。
agent: writer
output_file: blog_post.md
python
# crew.py
from crewai import Task
from crewai.project import CrewBase, task
@CrewBase
class MyCrew:
tasks_config = 'config/tasks.yaml'
@task
def research_task(self) -> Task:
return Task(
config=self.tasks_config['research_task']
)
@task
def write_task(self) -> Task:
return Task(
config=self.tasks_config['write_task']
)
Task 类型示例 #
研究任务 #
python
from crewai import Task
research_task = Task(
description="""
研究 2025 年 AI 行业的最新趋势。
重点领域：
1. 大语言模型的发展
2. 多模态 AI 的进展
3. AI Agent 的应用
为每个领域提供：
- 技术描述
- 主要公司
- 市场影响
""",
expected_output="""
一份详细的研究报告，包含：
- 执行摘要
- 三个主要趋势的详细分析
- 每个趋势的关键洞察
- 未来展望
""",
agent=researcher,
tools=[search_tool, web_rag_tool]
)
写作任务 #
python
from crewai import Task
write_task = Task(
description="""
根据研究结果撰写一篇关于 AI 趋势的博客文章。
要求：
- 标题引人注目
- 引言吸引读者
- 正文内容丰富
- 结论发人深省
""",
expected_output="""
一篇 4-5 段落的 Markdown 格式博客文章，
包含标题、引言、正文和结论。
""",
agent=writer,
output_file="output/blog_post.md"
)
分析任务 #
python
from crewai import Task
analysis_task = Task(
description="""
分析提供的销售数据，找出关键趋势和异常。
数据范围：2024 年全年销售数据
分析维度：
- 月度趋势
- 产品类别表现
- 区域分布
""",
expected_output="""
一份数据分析报告，包含：
- 关键发现（3-5 个）
- 数据可视化建议
- 行动建议
""",
agent=analyst,
tools=[csv_tool, analysis_tool]
)
代码任务 #
python
from crewai import Task
code_task = Task(
description="""
编写一个 Python 脚本来处理 CSV 文件。
功能需求：
1. 读取 CSV 文件
2. 数据清洗
3. 生成统计报告
4. 导出结果
""",
expected_output="""
一个完整的 Python 脚本，
包含注释和错误处理。
""",
agent=developer,
output_file="scripts/process_data.py"
)
Task 输出 #
文件输出 #
python
task = Task(
description="生成市场分析报告",
expected_output="Markdown 格式的报告",
agent=analyst,
output_file="reports/market_analysis.md"
)
JSON 输出 #
python
from pydantic import BaseModel
from typing import List
class MarketReport(BaseModel):
title: str
summary: str
trends: List[str]
recommendations: List[str]
task = Task(
description="生成市场分析报告",
agent=analyst,
output_pydantic=MarketReport
)
result = crew.kickoff()
report = result.pydantic
print(report.title)
print(report.trends)
结构化输出 #
python
from pydantic import BaseModel, Field
from typing import List, Optional
class TrendAnalysis(BaseModel):
trend_name: str = Field(description="趋势名称")
description: str = Field(description="趋势描述")
impact_score: int = Field(description="影响分数 1-10")
key_players: List[str] = Field(description="主要参与者")
class AnalysisReport(BaseModel):
executive_summary: str
trends: List[TrendAnalysis]
future_outlook: str
confidence_level: Optional[str]
task = Task(
description="分析 AI 市场趋势",
agent=analyst,
output_pydantic=AnalysisReport
)
Task 上下文 #
任务依赖 #
python
from crewai import Task
research_task = Task(
description="研究 AI 趋势",
expected_output="研究报告",
agent=researcher
)
analysis_task = Task(
description="分析研究结果",
expected_output="分析报告",
agent=analyst,
context=[research_task]
)
write_task = Task(
description="撰写文章",
expected_output="博客文章",
agent=writer,
context=[research_task, analysis_task]
)
上下文流程 #
text
┌─────────────────────────────────────────────────────────────┐
│ Task 上下文流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ research_task │
│ │ │
│ ├──────────────┐ │
│ ▼ ▼ │
│ analysis_task summary_task │
│ │ │ │
│ └──────┬───────┘ │
│ ▼ │
│ write_task │
│ │ │
│ ▼ │
│ 输出 │
│ │
└─────────────────────────────────────────────────────────────┘
异步执行 #
异步任务 #
python
from crewai import Task
async_task = Task(
description="获取实时市场数据",
expected_output="市场数据报告",
agent=data_collector,
async_execution=True
)
sync_task = Task(
description="分析市场数据",
expected_output="分析报告",
agent=analyst,
context=[async_task]
)
并行执行 #
python
# 多个异步任务并行执行
data_task_1 = Task(
description="获取股票数据",
agent=collector_1,
async_execution=True
)
data_task_2 = Task(
description="获取新闻数据",
agent=collector_2,
async_execution=True
)
data_task_3 = Task(
description="获取社交媒体数据",
agent=collector_3,
async_execution=True
)
analysis_task = Task(
description="综合分析所有数据",
agent=analyst,
context=[data_task_1, data_task_2, data_task_3]
)
Task 回调 #
完成回调 #
python
def on_task_complete(output):
print(f"任务完成: {output}")
send_notification(output)
task = Task(
description="生成报告",
expected_output="报告内容",
agent=writer,
callback=on_task_complete
)
错误处理 #
python
def on_task_error(error):
log_error(error)
send_alert(error)
task = Task(
description="执行任务",
agent=agent,
callback=on_task_complete
)
Task 工具 #
任务特定工具 #
python
from crewai import Task
from crewai_tools import SerperDevTool, PDFSearchTool
search_tool = SerperDevTool()
pdf_tool = PDFSearchTool()
task = Task(
description="研究并分析 PDF 文档",
expected_output="分析报告",
agent=researcher,
tools=[search_tool, pdf_tool]
)
工具覆盖 #
python
# Agent 有默认工具
agent = Agent(
role="分析师",
tools=[default_tool]
)
# Task 可以覆盖或添加工具
task = Task(
description="特殊任务",
agent=agent,
tools=[special_tool]
)
Task 输出访问 #
访问输出 #
python
result = crew.kickoff()
# 访问最终输出
print(result.raw)
# 访问所有任务输出
for task_output in result.tasks_output:
print(f"Task: {task_output.description}")
print(f"Agent: {task_output.agent}")
print(f"Output: {task_output.raw}")
print(f"JSON: {task_output.json_dict}")
输出属性 #
python
# TaskOutput 属性
task_output = result.tasks_output[0]
task_output.raw # 原始输出
task_output.pydantic # Pydantic 模型
task_output.json_dict # JSON 字典
task_output.agent # 执行的 Agent
task_output.summary # 输出摘要
Task 最佳实践 #
1. 描述清晰 #
python
# 好的描述
good_task = Task(
description="""
分析 2025 年 AI 芯片市场：
1. 识别主要厂商和产品
2. 分析市场份额和趋势
3. 评估技术发展方向
4. 提供投资建议
数据来源：公开报告、新闻、公司财报
""",
expected_output="结构化的市场分析报告"
)
# 不好的描述
bad_task = Task(
description="分析市场",
expected_output="报告"
)
2. 预期输出明确 #
python
# 好的预期输出
good_task = Task(
description="研究 AI 趋势",
expected_output="""
一份 Markdown 格式的报告，包含：
- 标题
- 执行摘要（100-150 字）
- 3 个主要趋势（每个 200-300 字）
- 结论和建议
"""
)
3. 合理使用上下文 #
python
# 合理的任务链
research = Task(description="研究", agent=researcher)
analysis = Task(description="分析", agent=analyst, context=[research])
write = Task(description="写作", agent=writer, context=[research, analysis])
edit = Task(description="编辑", agent=editor, context=[write])
4. 异步优化 #
python
# 并行执行独立任务
task_1 = Task(description="任务1", agent=agent1, async_execution=True)
task_2 = Task(description="任务2", agent=agent2, async_execution=True)
task_3 = Task(description="任务3", agent=agent3, async_execution=True)
# 汇总任务
summary = Task(
description="汇总",
agent=agent4,
context=[task_1, task_2, task_3]
)
常见问题 #
任务超时 #
python
# 设置 Agent 的执行时间限制
agent = Agent(
role="分析师",
max_execution_time=300
)
task = Task(
description="复杂分析",
agent=agent
)
输出格式问题 #
python
# 使用 Pydantic 确保格式
from pydantic import BaseModel
class Report(BaseModel):
title: str
content: str
summary: str
task = Task(
description="生成报告",
output_pydantic=Report
)
任务依赖问题 #
python
# 确保依赖任务先执行
crew = Crew(
agents=[researcher, analyst, writer],
tasks=[research_task, analysis_task, write_task],
process=Process.sequential
)
下一步 #
现在你已经深入了解了 Task，接下来学习 Crew 详解，掌握团队编排和协作模式！
最后更新：2026-04-04