智能体与任务（Agents and Tasks）

概述

在 CrewAI 中，Agent（智能体） 和 Task（任务） 是构建多智能体系统的两个核心抽象：

Agent：具有特定角色、目标和能力的 AI 实体
Task：需要智能体完成的具体工作单元

Crews

Agent（智能体）

基本定义

python

from crewai import Agent

agent = Agent(
    role="Senior Research Analyst",
    goal="Analyze market trends and provide insights",
    backstory="""You are an experienced research analyst with
    expertise in market analysis and trend forecasting.""",
    tools=[search_tool, analysis_tool],
    llm="gpt-4",
    verbose=True
)

完整参数列表

参数	类型	说明	默认值
`role`	str	智能体角色	必填
`goal`	str	智能体目标	必填
`backstory`	str	背景故事	必填
`tools`	List[Tool]	可用工具	`[]`
`llm`	str/LLM	使用的 LLM	OpenAI GPT
`verbose`	bool	详细输出	`False`
`allow_delegation`	bool	允许委派	`True`
`memory`	bool	启用记忆	`False`
`max_iter`	int	最大迭代次数	`25`
`max_rpm`	int	每分钟最大请求数	`None`
`step_callback`	Callable	步骤回调函数	`None`

YAML 配置方式

agents.yaml：

yaml

researcher:
  role: >
    {topic} Senior Research Analyst
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You are an expert researcher with deep knowledge in {topic}.
    Known for your ability to find hidden patterns and insights.

writer:
  role: >
    Content Writer Specialist
  goal: >
    Create compelling and accurate content about {topic}
  backstory: >
    You are a skilled writer who transforms complex research
    into engaging, accessible content.

reviewer:
  role: >
    Quality Assurance Specialist
  goal: >
    Ensure content accuracy and quality
  backstory: >
    You have a keen eye for detail and ensure all content
    meets the highest standards.

在代码中使用配置

python

from crewai import Agent
from crewai.project import CrewBase, agent

@CrewBase
class MyProjectCrew:
    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'],
            tools=[SerperDevTool()],
            verbose=True
        )

    @agent
    def writer(self) -> Agent:
        return Agent(
            config=self.agents_config['writer'],
            verbose=True
        )

委派机制

当 allow_delegation=True 时，智能体可以将任务委派给其他智能体：

python

manager = Agent(
    role="Project Manager",
    goal="Coordinate team and ensure project completion",
    backstory="Experienced project manager...",
    allow_delegation=True  # 允许委派任务
)

worker = Agent(
    role="Developer",
    goal="Write high-quality code",
    backstory="Senior developer...",
    allow_delegation=False  # 只执行，不委派
)

Task（任务）

基本定义

python

from crewai import Task

task = Task(
    description="Research the latest trends in {topic}",
    expected_output="A comprehensive report with key findings",
    agent=researcher,
    tools=[search_tool],
    output_file="report.md"
)

完整参数列表

参数	类型	说明	默认值
`description`	str	任务描述	必填
`expected_output`	str	预期输出	必填
`agent`	Agent	执行智能体	`None`
`tools`	List[Tool]	任务专用工具	`[]`
`output_file`	str	输出文件路径	`None`
`context`	List[Task]	上下文任务	`[]`
`async_execution`	bool	异步执行	`False`
`human_input`	bool	需要人工输入	`False`
`callback`	Callable	完成回调	`None`

YAML 配置方式

tasks.yaml：

yaml

research_task:
  description: >
    Conduct comprehensive research about {topic}.
    Focus on recent developments and emerging trends.
    Current year is 2025.
  expected_output: >
    A detailed research report with:
    - Key findings (at least 10 points)
    - Data sources cited
    - Trend analysis
  agent: researcher

writing_task:
  description: >
    Create an engaging article based on the research findings.
    Make it accessible to a general audience.
  expected_output: >
    A well-structured article (1500-2000 words) with:
    - Compelling introduction
    - Main body with subsections
    - Conclusion with key takeaways
  agent: writer
  output_file: article.md
  context:
    - research_task  # 依赖研究任务的输出

review_task:
  description: >
    Review the article for accuracy, clarity, and quality.
    Provide detailed feedback and suggestions.
  expected_output: >
    A review report with:
    - Quality assessment
    - Factual accuracy check
    - Improvement suggestions
  agent: reviewer
  context:
    - writing_task

任务依赖（Context）

任务可以依赖其他任务的输出：

python

research_task = Task(
    description="Research the topic",
    expected_output="Research findings",
    agent=researcher
)

analysis_task = Task(
    description="Analyze the research findings",
    expected_output="Analysis report",
    agent=analyst,
    context=[research_task]  # 依赖 research_task 的输出
)

writing_task = Task(
    description="Write based on analysis",
    expected_output="Final document",
    agent=writer,
    context=[research_task, analysis_task]  # 依赖多个任务
)

异步执行

python

# 可以并行执行的任务
task1 = Task(
    description="Task 1",
    expected_output="Output 1",
    agent=agent1,
    async_execution=True  # 异步执行
)

task2 = Task(
    description="Task 2",
    expected_output="Output 2",
    agent=agent2,
    async_execution=True  # 异步执行
)

# 依赖异步任务的同步任务
final_task = Task(
    description="Combine results",
    expected_output="Final output",
    agent=agent3,
    context=[task1, task2]  # 等待异步任务完成
)

Human-in-the-Loop

需要人工确认或输入时：

python

review_task = Task(
    description="Review the draft and provide feedback",
    expected_output="Approved content",
    agent=reviewer,
    human_input=True  # 需要人工输入
)

实战示例

示例 1：研究报告团队

python

from crewai import Agent, Task, Crew, Process

# 定义智能体
researcher = Agent(
    role="Research Analyst",
    goal="Find accurate and relevant information",
    backstory="Expert at gathering and analyzing information"
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear and engaging content",
    backstory="Skilled at explaining complex topics"
)

editor = Agent(
    role="Editor",
    goal="Ensure content quality and accuracy",
    backstory="Experienced editor with attention to detail"
)

# 定义任务
research_task = Task(
    description="Research {topic} thoroughly",
    expected_output="Comprehensive research notes",
    agent=researcher
)

writing_task = Task(
    description="Write a report based on the research",
    expected_output="Draft report",
    agent=writer,
    context=[research_task]
)

editing_task = Task(
    description="Edit and polish the report",
    expected_output="Final polished report",
    agent=editor,
    context=[writing_task],
    output_file="final_report.md"
)

# 创建团队
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,
    verbose=True
)

# 执行
result = crew.kickoff(inputs={"topic": "AI in Healthcare"})

示例 2：代码审查团队

python

from crewai import Agent, Task, Crew

# 智能体
developer = Agent(
    role="Senior Developer",
    goal="Write clean, efficient code",
    backstory="10+ years of software development experience"
)

reviewer = Agent(
    role="Code Reviewer",
    goal="Ensure code quality and best practices",
    backstory="Expert in code review and best practices"
)

tester = Agent(
    role="QA Engineer",
    goal="Ensure code correctness through testing",
    backstory="Experienced in test-driven development"
)

# 任务
coding_task = Task(
    description="Implement {feature} following best practices",
    expected_output="Implementation code",
    agent=developer
)

review_task = Task(
    description="Review the code for quality issues",
    expected_output="Review feedback",
    agent=reviewer,
    context=[coding_task]
)

testing_task = Task(
    description="Write tests for the implementation",
    expected_output="Test code",
    agent=tester,
    context=[coding_task]
)

# 创建团队（层级模式）
crew = Crew(
    agents=[developer, reviewer, tester],
    tasks=[coding_task, review_task, testing_task],
    process=Process.hierarchical,  # 自动分配管理者
    manager_llm="gpt-4"
)

示例 3：使用回调函数

python

def on_task_complete(output):
    print(f"Task completed with output: {output}")
    # 可以在这里发送通知、记录日志等

task = Task(
    description="Complete the analysis",
    expected_output="Analysis results",
    agent=analyst,
    callback=on_task_complete  # 任务完成时调用
)

最佳实践

1. 角色设计

python

# 好的角色定义 - 具体且有特色
agent = Agent(
    role="Senior Financial Analyst specializing in Tech Stocks",
    goal="Provide actionable investment insights for tech sector",
    backstory="Former Wall Street analyst with 15 years experience..."
)

# 避免 - 太过笼统
agent = Agent(
    role="Analyst",
    goal="Analyze things",
    backstory="Good at analysis"
)

2. 任务描述

python

# 好的任务描述 - 清晰、具体、有上下文
task = Task(
    description="""
    Analyze the quarterly earnings report for {company}.
    Focus on:
    1. Revenue growth trends
    2. Profit margin changes
    3. Key risk factors
    Current date: {date}
    """,
    expected_output="""
    A structured analysis report with:
    - Executive summary (3-5 sentences)
    - Detailed findings for each focus area
    - Data-backed conclusions
    - Investment recommendation
    """
)

# 避免 - 模糊不清
task = Task(
    description="Analyze the company",
    expected_output="A report"
)

3. 工具分配

python

# 按需分配工具
researcher = Agent(
    role="Researcher",
    tools=[search_tool, web_scraper]  # 需要搜索和爬取
)

writer = Agent(
    role="Writer",
    tools=[]  # 纯写作，不需要工具
)

coder = Agent(
    role="Developer",
    tools=[code_executor, file_tool]  # 需要执行代码
)

4. 错误处理

python

try:
    result = crew.kickoff(inputs={"topic": "AI"})
except Exception as e:
    print(f"Crew execution failed: {e}")
    # 处理错误，可能重试或通知

与其他框架对比

特性	CrewAI	MetaGPT	LangGraph
智能体定义	Agent 类	Role 类	Node 函数
任务分配	显式绑定	订阅机制	图边定义
委派机制	`allow_delegation`	`_watch`	条件边
配置方式	YAML + Python	Python	Python
执行控制	Process 枚举	SOP 流程	图结构

下一节：17.3 工具与集成

智能体与任务（Agents and Tasks） ​

概述 ​

Agent（智能体） ​

基本定义 ​

完整参数列表 ​

YAML 配置方式 ​

在代码中使用配置 ​

委派机制 ​

Task（任务） ​

基本定义 ​

完整参数列表 ​

YAML 配置方式 ​

任务依赖（Context） ​

异步执行 ​

Human-in-the-Loop ​

实战示例 ​

示例 1：研究报告团队 ​

示例 2：代码审查团队 ​

示例 3：使用回调函数 ​

最佳实践 ​

1. 角色设计 ​

2. 任务描述 ​

3. 工具分配 ​

4. 错误处理 ​

与其他框架对比 ​

智能体与任务（Agents and Tasks）

概述

Agent（智能体）

基本定义

完整参数列表

YAML 配置方式

在代码中使用配置

委派机制

Task（任务）

基本定义

完整参数列表

YAML 配置方式

任务依赖（Context）

异步执行

Human-in-the-Loop

实战示例

示例 1：研究报告团队

示例 2：代码审查团队

示例 3：使用回调函数

最佳实践

1. 角色设计

2. 任务描述

3. 工具分配

4. 错误处理

与其他框架对比