最佳实践 #

项目结构 #

推荐目录结构 #

text 
my_crewai_project/
├── src/
│   └── my_crew/
│       ├── config/
│       │   ├── agents.yaml
│       │   └── tasks.yaml
│       ├── tools/
│       │   ├── __init__.py
│       │   └── custom_tool.py
│       ├── knowledge/
│       │   └── company_info.txt
│       ├── crew.py
│       └── main.py
├── tests/
│   ├── test_agents.py
│   └── test_crew.py
├── pyproject.toml
├── .env
└── README.md

配置分离 #

yaml 
# config/agents.yaml
researcher:
  role: >
    {topic} 高级研究分析师
  goal: >
    发现 {topic} 的前沿趋势
  backstory: >
    你是一位经验丰富的研究分析师
python 
# crew.py
from crewai.project import CrewBase, agent, crew

@CrewBase
class ResearchCrew:
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'

    @agent
    def researcher(self) -> Agent:
        return Agent(config=self.agents_config['researcher'])

Agent 设计最佳实践 #

1. 角色定义清晰 #

python 
# 好的设计
researcher = Agent(
    role="医疗数据研究分析师",
    goal="分析医疗数据，发现疾病模式和趋势",
    backstory="""你是一位拥有 15 年经验的医疗数据分析师，
    专注于流行病学和疾病预测。你擅长从复杂的医疗数据中
    发现隐藏的模式，并能将技术发现转化为可操作的建议。"""
)

# 不好的设计
researcher = Agent(
    role="分析师",
    goal="分析数据",
    backstory="我是一个分析师"
)

2. 合理的工具配置 #

python 
# 根据任务需求选择工具
research_agent = Agent(
    role="研究员",
    goal="研究市场趋势",
    tools=[SerperDevTool(), WebsiteSearchTool()],
    verbose=True
)

file_agent = Agent(
    role="文档分析师",
    goal="分析文档内容",
    tools=[FileReadTool(), PDFSearchTool(), DirectoryReadTool()],
    verbose=True
)

code_agent = Agent(
    role="开发者",
    goal="编写和执行代码",
    tools=[CodeInterpreterTool()],
    allow_code_execution=True,
    code_execution_mode="safe",
    verbose=True
)

3. 性能优化配置 #

python 
# 高效的 Agent 配置
efficient_agent = Agent(
    role="分析师",
    goal="快速分析数据",
    max_iter=15,
    max_rpm=30,
    cache=True,
    verbose=False
)

# 使用更便宜的模型进行工具调用
agent = Agent(
    role="分析师",
    llm="gpt-4o",
    function_calling_llm="gpt-4o-mini"
)

Task 设计最佳实践 #

1. 描述清晰具体 #

python 
# 好的设计
research_task = Task(
    description="""
    研究 2025 年 AI 芯片市场：
    
    1. 识别主要厂商和产品
       - NVIDIA、AMD、Intel 等
       - 产品规格和性能
    
    2. 分析市场份额和趋势
       - 当前市场份额
       - 增长趋势
    
    3. 评估技术发展方向
       - 架构创新
       - 能效提升
    
    数据来源：公开报告、新闻、公司财报
    """,
    expected_output="""
    一份结构化的市场分析报告，包含：
    - 执行摘要（200 字以内）
    - 市场概览
    - 竞争格局分析
    - 技术趋势
    - 投资建议
    """
)

# 不好的设计
research_task = Task(
    description="分析市场",
    expected_output="报告"
)

2. 合理使用上下文 #

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

3. 异步任务优化 #

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
)

# 汇总任务
summary_task = Task(
    description="综合分析所有数据",
    agent=analyst,
    context=[data_task_1, data_task_2, data_task_3]
)

Crew 编排最佳实践 #

1. 选择合适的流程 #

python 
# 简单线性任务：Sequential
simple_crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential
)

# 复杂协作任务：Hierarchical
complex_crew = Crew(
    agents=[researcher, analyst, writer, editor],
    tasks=[research_task, analysis_task, write_task, edit_task],
    process=Process.hierarchical,
    manager_llm="gpt-4o"
)

2. 合理的团队规模 #

python 
# 小型团队（2-3 个 Agent）：适合简单任务
small_crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task]
)

# 中型团队（4-6 个 Agent）：适合复杂任务
medium_crew = Crew(
    agents=[researcher, analyst, writer, editor],
    tasks=[research_task, analysis_task, write_task, edit_task]
)

# 大型团队（7+ 个 Agent）：需要层级管理
large_crew = Crew(
    agents=[manager, *specialists],
    tasks=tasks,
    process=Process.hierarchical,
    manager_llm="gpt-4o"
)

3. 启用必要功能 #

python 
crew = Crew(
    agents=[agent],
    tasks=[task],
    memory=True,
    cache=True,
    verbose=True,
    output_log_file="logs/execution.json"
)

性能优化 #

1. 缓存策略 #

python 
# Agent 级别缓存
agent = Agent(
    role="分析师",
    cache=True
)

# Crew 级别缓存
crew = Crew(
    agents=[agent],
    tasks=[task],
    cache=True
)

# 自定义缓存逻辑
def cache_func(args, result):
    return "error" not in result.lower()

tool.cache_function = cache_func

2. 速率限制 #

python 
# Agent 级别
agent = Agent(
    role="分析师",
    max_rpm=30
)

# Crew 级别
crew = Crew(
    agents=[agent],
    tasks=[task],
    max_rpm=60
)

3. 执行时间控制 #

python 
agent = Agent(
    role="分析师",
    max_iter=20,
    max_execution_time=300,
    max_retry_limit=2
)

4. 模型选择 #

python 
# 复杂推理：使用高级模型
reasoning_agent = Agent(
    role="战略规划师",
    llm="gpt-4o",
    reasoning=True
)

# 简单任务：使用经济模型
simple_agent = Agent(
    role="数据收集员",
    llm="gpt-4o-mini"
)

# 工具调用：使用专用模型
agent = Agent(
    role="分析师",
    llm="gpt-4o",
    function_calling_llm="gpt-4o-mini"
)

错误处理 #

1. 基本错误处理 #

python 
from crewai import Crew

try:
    result = crew.kickoff()
    print(result.raw)
except Exception as e:
    print(f"执行失败: {e}")
    log_error(e)

2. 重试机制 #

python 
agent = Agent(
    role="分析师",
    max_retry_limit=3
)

3. 超时处理 #

python 
import asyncio
from crewai import Crew

async def run_with_timeout(crew, inputs, timeout=300):
    try:
        result = await asyncio.wait_for(
            crew.akickoff(inputs=inputs),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        print("执行超时")
        return None

4. 优雅降级 #

python 
def run_crew_with_fallback(crew, inputs):
    try:
        result = crew.kickoff(inputs=inputs)
        return result
    except Exception as e:
        print(f"主流程失败: {e}")
        return fallback_process(inputs)

安全考虑 #

1. API 密钥管理 #

python 
import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")

2. 代码执行安全 #

python 
# 生产环境：使用安全模式
agent = Agent(
    role="开发者",
    allow_code_execution=True,
    code_execution_mode="safe"
)

# 仅在受信任环境使用不安全模式
agent = Agent(
    role="开发者",
    allow_code_execution=True,
    code_execution_mode="unsafe"
)

3. 输入验证 #

python 
from pydantic import BaseModel, Field, validator

class SafeInput(BaseModel):
    query: str = Field(..., max_length=1000)
    
    @validator('query')
    def sanitize(cls, v):
        return v.strip()

def safe_kickoff(crew, inputs):
    validated = SafeInput(**inputs)
    return crew.kickoff(inputs=validated.dict())

4. 敏感信息处理 #

python 
import re

def sanitize_output(output):
    patterns = [
        r'\b\d{16}\b',  # 信用卡号
        r'\b\d{3}-\d{2}-\d{4}\b',  # SSN
        r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'  # 邮箱
    ]
    
    for pattern in patterns:
        output = re.sub(pattern, '[REDACTED]', output)
    
    return output

测试策略 #

1. 单元测试 #

python 
import pytest
from crewai import Agent, Task

def test_agent_creation():
    agent = Agent(
        role="测试角色",
        goal="测试目标",
        backstory="测试背景"
    )
    assert agent.role == "测试角色"

def test_task_creation():
    task = Task(
        description="测试任务",
        expected_output="测试输出"
    )
    assert task.description == "测试任务"

2. 集成测试 #

python 
def test_crew_execution():
    crew = Crew(
        agents=[test_agent],
        tasks=[test_task]
    )
    result = crew.kickoff()
    assert result.raw is not None

3. Mock 测试 #

python 
from unittest.mock import patch

@patch('crewai.Agent.execute')
def test_agent_with_mock(mock_execute):
    mock_execute.return_value = "模拟结果"
    agent = Agent(role="测试", goal="测试", backstory="测试")
    result = agent.execute("测试输入")
    assert result == "模拟结果"

部署建议 #

1. 环境配置 #

python 
import os

class Config:
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    SERPER_API_KEY = os.getenv("SERPER_API_KEY")
    LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
    MAX_RPM = int(os.getenv("MAX_RPM", "60"))

2. 日志配置 #

python 
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('crewai.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

3. 监控指标 #

python 
def track_metrics(crew, result):
    metrics = {
        "execution_time": result.execution_time,
        "token_usage": crew.usage_metrics,
        "task_count": len(crew.tasks),
        "agent_count": len(crew.agents)
    }
    save_metrics(metrics)

4. 容器化部署 #

dockerfile 
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["python", "main.py"]

成本优化 #

1. 模型选择 #

python 
# 根据任务复杂度选择模型
def get_agent_for_task(complexity):
    if complexity == "high":
        return Agent(llm="gpt-4o")
    elif complexity == "medium":
        return Agent(llm="gpt-4o-mini")
    else:
        return Agent(llm="gpt-3.5-turbo")

2. 缓存利用 #

python 
# 启用缓存减少重复调用
crew = Crew(
    agents=[agent],
    tasks=[task],
    cache=True
)

3. 批量处理 #

python 
# 批量处理减少 API 调用
inputs_list = [input1, input2, input3]
results = crew.kickoff_for_each(inputs=inputs_list)

常见陷阱 #

1. 避免过度复杂的 Agent #

python 
# 不好：一个 Agent 做所有事情
super_agent = Agent(
    role="全能助手",
    goal="完成所有任务",
    backstory="..."
)

# 好：专业分工
researcher = Agent(role="研究员", ...)
analyst = Agent(role="分析师", ...)
writer = Agent(role="作家", ...)

2. 避免模糊的任务描述 #

python 
# 不好
task = Task(description="分析数据")

# 好
task = Task(
    description="""
    分析销售数据：
    1. 月度趋势
    2. 产品表现
    3. 区域分布
    """,
    expected_output="结构化分析报告"
)

3. 避免忽略错误处理 #

python 
# 不好
result = crew.kickoff()

# 好
try:
    result = crew.kickoff()
    handle_result(result)
except Exception as e:
    handle_error(e)

总结 #

遵循这些最佳实践，可以帮助你：

  • 构建可维护、可扩展的 CrewAI 应用
  • 优化性能和成本
  • 确保安全性和可靠性
  • 提高开发和部署效率

持续学习和实践，你将能够熟练运用 CrewAI 构建生产级的多智能体协作系统！

最后更新：2026-04-04