crewai-developer

CrewAI 开发者指南

概述

CrewAI 是一个精简、高速的 Python 框架，用于构建协作式 AI Agent 团队和结构化工作流。它帮助开发者创建具有特定角色、工具和目标的自主 AI Agent，让它们协同处理复杂任务。本指南涵盖 Crews（自主协作）、Flows（结构化编排）、Agent、任务和企业级部署。

核心概念

Agent：专业团队成员

Agent 是具有特定角色、目标和能力的自主 AI 单元。

from crewai import Agent

# Create a research agent
researcher = Agent(
    role='Senior Research Analyst',
    goal='Uncover cutting-edge developments in AI and data science',
    backstory="""You are an expert at a leading tech think tank.
    Your expertise lies in identifying emerging trends and technologies in AI,
    data science, and machine learning.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, scrape_tool]
)

# Create a writer agent
writer = Agent(
    role='Tech Content Strategist',
    goal='Craft compelling content on tech advancements',
    backstory="""You are a renowned content strategist, known for
    your insightful and engaging articles on technology and innovation.
    You transform complex concepts into compelling narratives.""",
    verbose=True,
    allow_delegation=True,
    tools=[write_tool]
)

Agent 的关键属性

agent = Agent(
    role='Role Name',              # The agent's job title
    goal='Specific objective',     # What the agent aims to achieve
    backstory='Background story',  # Context and expertise
    verbose=True,                  # Enable detailed logging
    allow_delegation=False,        # Can delegate tasks to other agents
    tools=[tool1, tool2],         # Available tools
    llm=custom_llm,               # Custom LLM configuration
    max_iter=15,                  # Maximum iterations for task
    max_rpm=10,                   # Rate limit (requests per minute)
    memory=True,                  # Enable memory
    cache=True,                   # Enable response caching
    system_template="template",   # Custom system prompt template
    prompt_template="template",   # Custom prompt template
    response_template="template"  # Custom response template
)

任务：单项工作分配

任务定义 Agent 要完成的具体工作。

from crewai import Task

# Research task
research_task = Task(
    description="""Conduct a comprehensive analysis of the latest advancements in AI.
    Identify key trends, breakthrough technologies, and potential industry impacts.
    Compile your findings in a detailed report.""",
    expected_output='A comprehensive 3-paragraph report on AI advancements',
    agent=researcher,
    tools=[search_tool],
    output_file='research_report.md'
)

# Writing task
write_task = Task(
    description="""Using the research analyst's report, develop an engaging blog post
    highlighting the most significant AI advancements.
    Make it accessible and engaging for a general audience.""",
    expected_output='A 4-paragraph blog post about AI advancements',
    agent=writer,
    context=[research_task],  # Depends on research_task output
    output_file='blog_post.md'
)

任务的关键属性

task = Task(
    description='Detailed task description',
    expected_output='Clear output format',
    agent=agent_instance,
    tools=[tool1, tool2],           # Task-specific tools
    context=[previous_task],        # Dependencies
    async_execution=False,          # Run asynchronously
    output_json=OutputClass,        # Structured output (Pydantic)
    output_pydantic=OutputClass,    # Pydantic validation
    output_file='result.txt',       # Save output to file
    callback=callback_function,     # Callback on completion
    human_input=False              # Request human feedback
)

Crews：组织 Agent 团队

Crews 会编排 Agent 协同工作，以实现共同目标。

from crewai import Crew, Process

# Create a crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # or Process.hierarchical
    verbose=True,
    memory=True,
    cache=True,
    max_rpm=10,
    share_crew=False
)

# Kickoff the crew
result = crew.kickoff()
print(result)

# Kickoff with custom inputs
result = crew.kickoff(inputs={
    'topic': 'Artificial Intelligence',
    'audience': 'developers'
})

流程类型

# Sequential process (tasks run one after another)
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.sequential
)

# Hierarchical process (manager delegates to agents)
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    process=Process.hierarchical,
    manager_llm='gpt-4'  # Required for hierarchical
)

Flows：结构化工作流编排

Flows 提供事件驱动、确定性的执行路径控制。

from crewai.flow.flow import Flow, listen, start

class BlogPostFlow(Flow):

    @start()
    def fetch_topic(self):
        """Entry point - fetch the topic to write about"""
        print("Starting blog post generation")
        return "AI advancements in 2024"

    @listen(fetch_topic)
    def research_topic(self, topic):
        """Research the topic"""
        print(f"Researching: {topic}")
        # Integrate with Crew for autonomous research
        research_crew = Crew(
            agents=[researcher],
            tasks=[research_task]
        )
        result = research_crew.kickoff(inputs={'topic': topic})
        return result

    @listen(research_topic)
    def write_blog_post(self, research_data):
        """Write the blog post"""
        print("Writing blog post...")
        write_crew = Crew(
            agents=[writer],
            tasks=[write_task]
        )
        result = write_crew.kickoff(inputs={'research': research_data})
        return result

    @listen(write_blog_post)
    def finalize(self, blog_post):
        """Finalize and save"""
        print("Blog post completed!")
        return blog_post

# Execute flow
flow = BlogPostFlow()
result = flow.kickoff()

Flow 状态管理

from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel

class ArticleState(BaseModel):
    topic: str = ""
    research: str = ""
    draft: str = ""
    final: str = ""

class ArticleFlow(Flow[ArticleState]):

    @start()
    def set_topic(self):
        self.state.topic = "AI Ethics"
        return self.state.topic

    @listen(set_topic)
    def research(self, topic):
        # Research logic
        self.state.research = "Research findings..."
        return self.state.research

    @listen(research)
    def write_draft(self, research):
        self.state.draft = "Draft content..."
        return self.state.draft

# Access state
flow = ArticleFlow()
flow.kickoff()
print(flow.state.topic)
print(flow.state.research)

Router 模式

from crewai.flow.flow import Flow, listen, start, router

class ContentFlow(Flow):

    @start()
    def categorize_content(self):
        return "technical"  # or "marketing", "blog"

    @router(categorize_content)
    def route_content(self, category):
        if category == "technical":
            return "write_technical"
        elif category == "marketing":
            return "write_marketing"
        else:
            return "write_blog"

    @listen("write_technical")
    def write_technical_doc(self):
        return "Technical documentation..."

    @listen("write_marketing")
    def write_marketing_copy(self):
        return "Marketing content..."

    @listen("write_blog")
    def write_blog_post(self):
        return "Blog post..."

工具：扩展 Agent 能力

内置工具

from crewai_tools import (
    SerperDevTool,      # Google search
    ScrapeWebsiteTool,  # Web scraping
    FileReadTool,       # Read files
    DirectoryReadTool,  # Read directories
    CodeDocsSearchTool, # Search code documentation
    CSVSearchTool,      # Search CSV files
    JSONSearchTool,     # Search JSON files
    MDXSearchTool,      # Search MDX files
    PDFSearchTool,      # Search PDF files
    TXTSearchTool,      # Search text files
    WebsiteSearchTool,  # Search websites
    SeleniumScrapingTool, # Browser automation
    YoutubeChannelSearchTool, # YouTube search
    YoutubeVideoSearchTool   # YouTube video search
)

# Using tools
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()
file_tool = FileReadTool()

agent = Agent(
    role='Researcher',
    tools=[search_tool, scrape_tool, file_tool]
)

自定义工具

from crewai_tools import BaseTool

class MyCustomTool(BaseTool):
    name: str = "Custom Tool Name"
    description: str = "Clear description of what the tool does"

    def _run(self, argument: str) -> str:
        # Implementation
        result = perform_operation(argument)
        return result

# Using custom tool
custom_tool = MyCustomTool()
agent = Agent(
    role='Specialist',
    tools=[custom_tool]
)

将函数用作工具

from crewai import Agent

def calculate_sum(a: int, b: int) -> int:
    """Calculate the sum of two numbers"""
    return a + b

agent = Agent(
    role='Calculator',
    tools=[calculate_sum]  # Pass function directly
)

记忆：从过往交互中学习

from crewai import Crew, Agent, Task

# Enable crew memory
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True,  # Enable all memory types
    verbose=True
)

# Configure specific memory types
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    memory=True,
    memory_config={
        'short_term': True,   # Remember within single run
        'long_term': True,    # Remember across runs
        'entity': True        # Remember entities (people, places)
    }
)

知识：RAG 集成

from crewai import Agent, Crew, Task, knowledge

# Create knowledge source
docs_knowledge = knowledge.StringKnowledgeSource(
    content="Company policies and procedures...",
    metadata={"source": "policy_docs"}
)

# Using knowledge in agent
agent = Agent(
    role='Policy Expert',
    goal='Answer questions about company policies',
    backstory='Expert in company policies',
    knowledge_sources=[docs_knowledge]
)

# Load knowledge from files
pdf_knowledge = knowledge.PDFKnowledgeSource(
    file_path='./documents/handbook.pdf'
)

txt_knowledge = knowledge.TextKnowledgeSource(
    file_path='./documents/faq.txt'
)

agent = Agent(
    role='Support Agent',
    knowledge_sources=[pdf_knowledge, txt_knowledge]
)

使用 Pydantic 实现结构化输出

from pydantic import BaseModel
from crewai import Task, Agent

class BlogPost(BaseModel):
    title: str
    content: str
    tags: list[str]
    word_count: int

# Task with structured output
write_task = Task(
    description='Write a blog post about AI',
    expected_output='Blog post with title, content, tags, and word count',
    agent=writer,
    output_pydantic=BlogPost
)

# Execute and get structured output
result = crew.kickoff()
blog_post: BlogPost = write_task.output.pydantic
print(blog_post.title)
print(blog_post.tags)

训练：提升性能

from crewai import Crew

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2]
)

# Training loop
crew.train(
    n_iterations=10,
    inputs={'topic': 'AI'},
    filename='trained_crew.pkl'
)

# Load trained crew
trained_crew = Crew.load('trained_crew.pkl')

人机协同

from crewai import Task

# Task requiring human input
review_task = Task(
    description='Review the draft and provide feedback',
    expected_output='Approved draft or feedback for revision',
    agent=editor,
    human_input=True  # Will pause and ask for input
)

# Conditional human input
task = Task(
    description='Generate report',
    expected_output='Final report',
    agent=analyst,
    callback=lambda output: validate_output(output),
    human_input=True if needs_review else False
)

测试 Crews

from crewai import Crew
import pytest

def test_research_crew():
    # Setup
    crew = Crew(
        agents=[researcher],
        tasks=[research_task]
    )

    # Execute
    result = crew.kickoff(inputs={'topic': 'AI'})

    # Assert
    assert result is not None
    assert 'AI' in result
    assert len(result) > 100

def test_crew_with_mock():
    # Mock agent behavior for testing
    mock_agent = Agent(
        role='Mock Agent',
        goal='Return test data',
        backstory='Test agent'
    )

    mock_task = Task(
        description='Test task',
        expected_output='Test output',
        agent=mock_agent
    )

    crew = Crew(agents=[mock_agent], tasks=[mock_task])
    result = crew.kickoff()

    assert result == 'Test output'

自定义 LLM

from langchain_openai import ChatOpenAI
from crewai import Agent, Crew

# Using custom LLM
custom_llm = ChatOpenAI(
    model='gpt-4-turbo-preview',
    temperature=0.7,
    max_tokens=2000
)

agent = Agent(
    role='Writer',
    llm=custom_llm
)

# Crew-level LLM
crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2],
    manager_llm=custom_llm  # For hierarchical process
)

异步执行

from crewai import Crew

crew = Crew(
    agents=[agent1, agent2],
    tasks=[task1, task2]
)

# Async kickoff
async def run_crew():
    result = await crew.kickoff_async(inputs={'topic': 'AI'})
    return result

# Kickoff for each (parallel execution)
inputs_list = [
    {'topic': 'AI'},
    {'topic': 'ML'},
    {'topic': 'Data Science'}
]

results = crew.kickoff_for_each(inputs=inputs_list)

回调和事件监听器

from crewai import Task, Agent

def on_task_complete(output):
    print(f"Task completed with output: {output}")
    # Log, notify, or process output

def on_task_error(error):
    print(f"Task failed with error: {error}")
    # Handle error, retry, or notify

task = Task(
    description='Analyze data',
    expected_output='Analysis report',
    agent=analyst,
    callback=on_task_complete
)

# Agent-level callbacks
agent = Agent(
    role='Analyst',
    step_callback=lambda step: print(f"Agent step: {step}"),
    task_callback=on_task_complete
)

企业级部署

环境配置

import os

# API keys
os.environ['OPENAI_API_KEY'] = 'your-key'
os.environ['SERPER_API_KEY'] = 'your-key'

# CrewAI+ (Enterprise)
os.environ['CREWAI_API_KEY'] = 'your-enterprise-key'

# Observability
os.environ['LANGCHAIN_TRACING_V2'] = 'true'
os.environ['LANGCHAIN_API_KEY'] = 'your-langchain-key'

项目结构

my_crew_project/
├── src/
│   └── my_crew_project/
│       ├── __init__.py
│       ├── main.py
│       ├── crew.py
│       ├── config/
│       │   ├── agents.yaml
│       │   └── tasks.yaml
│       └── tools/
│           └── custom_tool.py
├── tests/
│   └── test_crew.py
├── pyproject.toml
└── README.md

YAML 配置

agents.yaml

researcher:
  role: >
    Senior Research Analyst
  goal: >
    Uncover cutting-edge developments in {topic}
  backstory: >
    You are an expert researcher with deep knowledge in {topic}

writer:
  role: >
    Content Writer
  goal: >
    Create engaging content about {topic}
  backstory: >
    You are a skilled writer who makes complex topics accessible

tasks.yaml

research_task:
  description: >
    Conduct comprehensive research on {topic}
  expected_output: >
    A detailed research report with key findings
  agent: researcher

writing_task:
  description: >
    Write an article based on the research
  expected_output: >
    A well-structured article
  agent: writer
  context:
    - research_task

最佳实践

Agent 设计

✅ 推荐做法：

• 为 Agent 指定清晰、具体的角色
• 提供详细的背景故事作为上下文
• 仅配置必要的工具
• 为管理者启用委派功能
• 在开发期间使用详细模式

❌ 避免：

• 模糊或重叠的角色
• 工具过多（会造成混乱）
• 缺少背景故事
• 目标过于复杂

任务设计

✅ 推荐做法：

• 编写清晰、可执行的描述
• 指定预期的输出格式
• 设置恰当的任务依赖关系
• 使用上下文串联任务
• 为关键决策启用人工输入

❌ 避免：

• 含糊不清的描述
• 未指定预期输出
• 循环依赖
• 单项任务过于复杂

Crew 组织

✅ 推荐做法：

• 从顺序流程开始
• 复杂协调时使用层级流程
• 启用记忆以保留上下文
• 设置合理的速率限制
• 先使用小型数据集进行测试

❌ 避免：

• Agent 过多（3-5 个为最佳）
• 未经测试的复杂层级结构
• 多步流程中禁用记忆
• 未设置速率限制

常见模式

研究与撰写流程

# 1. Research agent gathers information
# 2. Analyst agent processes data
# 3. Writer agent creates content
# 4. Editor agent reviews and refines

researcher = Agent(role='Researcher', ...)
analyst = Agent(role='Analyst', ...)
writer = Agent(role='Writer', ...)
editor = Agent(role='Editor', ...)

research = Task(agent=researcher, ...)
analysis = Task(agent=analyst, context=[research], ...)
draft = Task(agent=writer, context=[analysis], ...)
final = Task(agent=editor, context=[draft], ...)

crew = Crew(
    agents=[researcher, analyst, writer, editor],
    tasks=[research, analysis, draft, final],
    process=Process.sequential
)

多阶段审批流程

class ApprovalFlow(Flow):

    @start()
    def create_draft(self):
        # Generate initial draft
        return draft_content

    @listen(create_draft)
    def request_review(self, draft):
        # Send for review
        return review_request

    @router(request_review)
    def check_approval(self, review):
        if review.approved:
            return "finalize"
        else:
            return "revise"

    @listen("revise")
    def revise_draft(self):
        # Revise and loop back
        return revised_draft

    @listen("finalize")
    def finalize_content(self):
        return final_content

快速参考

安装

# Using uv (recommended)
uv pip install crewai crewai-tools

# Using pip
pip install crewai crewai-tools

# With all extras
pip install 'crewai[all]'

CLI 命令

# Create new project
crewai create crew my_project

# Create flow
crewai create flow my_flow

# Install dependencies
crewai install

# Run project
crewai run

# Train crew
crewai train

# Replay task
crewai replay <task_id>

# Test crew
crewai test

必要导入

from crewai import Agent, Task, Crew, Process
from crewai.flow.flow import Flow, listen, start, router
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from pydantic import BaseModel

资源

如需了解高级模式、集成示例和故障排查，请参阅：

• 官方文档：https://docs.crewai.com/
• API 参考：https://docs.crewai.com/api-reference
• GitHub：https://github.com/joaomdmoura/crewAI
• 社区论坛：https://community.crewai.com/

扩展参考

请参阅 references/advanced_patterns.md，了解：

• MCP（Model Context Protocol）集成
• 可观测性和追踪设置
• 生产环境部署策略
• 高级 Flow 模式
• 性能优化