引言

CrewAI以其简洁的API和角色扮演式多智能体设计，在2026年获得了大量生产用户。与AutoGen相比，CrewAI更注重"团队协作"的自然性。本文将分享CrewAI在生产环境中的实践经验。

CrewAI核心概念

Crew（团队）

一个Crew由多个Agent组成，每个Agent有特定角色、目标和工具。

Agent（成员）

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role='市场研究员',
    goal='收集和分析市场数据',
    backstory='你是一位有10年经验的市场研究专家，擅长数据分析和趋势预测。',
    tools=[search_tool, analytics_tool],
    llm='gpt-5',
    verbose=True
)

writer = Agent(
    role='技术写作专家',
    goal='将研究结果转化为清晰的报告',
    backstory='你是一位资深技术写作专家，擅长将复杂数据转化为易懂的报告。',
    llm='claude-4-opus',
    verbose=True
)

editor = Agent(
    role='内容编辑',
    goal='确保报告质量和一致性',
    backstory='你是一位严谨的编辑，对细节和质量有极高要求。',
    llm='gpt-5',
    verbose=True
)

Task（任务）

research_task = Task(
    description='研究2026年AI市场趋势，重点关注LLM和Agent领域。',
    agent=researcher,
    expected_output='一份包含数据和分析的市场研究报告',
    context=[]
)

writing_task = Task(
    description='基于研究报告，撰写一篇2000字的行业分析文章。',
    agent=writer,
    expected_output='一篇2000字的文章',
    context=[research_task]  # 依赖研究任务的输出
)

editing_task = Task(
    description='审核并修改文章，确保准确性和可读性。',
    agent=editor,
    expected_output='最终版文章',
    context=[writing_task]
)

Crew（组建团队）

crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

result = crew.kickoff()

2026年新特性

1. 流程类型

# 顺序流程
crew = Crew(agents=agents, tasks=tasks, process=Process.sequential)

# 层级流程（有管理者）
crew = Crew(
    agents=agents, 
    tasks=tasks, 
    process=Process.hierarchical,
    manager_llm='gpt-5'
)

# 自定义流程
from crewai.process import CustomProcess

class MyProcess(CustomProcess):
    def run(self, crew, tasks):
        # 自定义执行逻辑
        pass

2. 工具集成

from crewai.tools import tool

@tool("搜索网络")
def search(query: str) -> str:
    """搜索互联网获取最新信息"""
    return web_search(query)

@tool("执行代码")
def execute_code(code: str) -> str:
    """执行Python代码并返回结果"""
    return exec_python(code)

@tool("读取文件")
def read_file(path: str) -> str:
    """读取本地文件"""
    with open(path) as f:
        return f.read()

3. 记忆系统

crew = Crew(
    agents=agents,
    tasks=tasks,
    memory=True,  # 启用记忆
    memory_config={
        "provider": "chroma",  # 向量数据库
        "embedder": "bge-large-zh",
        "long_term": True,
        "short_term": True
    }
)

4. 人机协作

from crewai import HumanInput

# 在关键步骤加入人工审核
task = Task(
    description='生成营销文案',
    agent=writer,
    human_input=HumanInput(
        enabled=True,
        check_every=1,  # 每步都检查
        prompt="请审核以上内容，输入修改意见或'approve'确认。"
    )
)

生产实践经验

实践一：角色设计

# 好的角色设计
good_agent = Agent(
    role='资深安全审计员',  # 具体角色
    goal='发现代码中的安全漏洞并提供修复建议',  # 明确目标
    backstory='''你是一位有15年经验的网络安全专家，
    曾在Google和腾讯安全团队工作，精通OWASP Top 10漏洞
    和安全编码最佳实践。''',  # 丰富背景
    tools=[code_analyzer, vulnerability_db],
    llm='gpt-5'
)

# 不好的角色设计
bad_agent = Agent(
    role='助手',  # 太模糊
    goal='帮忙',  # 不明确
    backstory='你是一个AI助手。'  # 太简单
)

实践二：任务分解

# 好的任务分解：颗粒度适中
tasks = [
    Task(description='分析需求文档，提取功能点', agent=analyst),
    Task(description='为每个功能点设计测试用例', agent=test_designer),
    Task(description='编写自动化测试脚本', agent=test_engineer),
    Task(description='执行测试并生成报告', agent=test_runner),
]

# 不好的任务分解：太粗
tasks = [
    Task(description='做测试', agent=tester),  # 太笼统
]

实践三：错误处理

from crewai import CrewError

try:
    result = crew.kickoff()
except CrewError as e:
    print(f"Crew执行失败：{e}")
    # 降级处理
    result = fallback_process()

# Agent级别错误处理
class SafeAgent(Agent):
    def execute_task(self, task):
        try:
            return super().execute_task(task)
        except Exception as e:
            return f"任务执行失败：{e}。请重试或调整策略。"

实践四：成本控制

# 根据任务复杂度选择模型
researcher = Agent(
    role='研究员',
    llm='deepseek-v4',  # 研究用便宜模型
    max_iter=5
)

writer = Agent(
    role='作家',
    llm='claude-4-opus',  # 写作用高质量模型
    max_iter=3
)

# 设置预算上限
crew = Crew(
    agents=[researcher, writer],
    tasks=tasks,
    max_cost=1.0,  # 最大花费$1
)

实践五：质量保证

# 添加质量检查Agent
quality_checker = Agent(
    role='质量检查员',
    goal='确保输出质量达到标准',
    backstory='你是一位严格的质量检查专家。',
    llm='gpt-5'
)

quality_task = Task(
    description='检查最终输出的质量，评分并给出改进建议。',
    agent=quality_checker,
    expected_output='质量评分报告'
)

# 在流程末尾加入质量检查
crew = Crew(
    agents=[researcher, writer, editor, quality_checker],
    tasks=[research_task, writing_task, editing_task, quality_task]
)

部署方案

API服务

from fastapi import FastAPI
from crewai import Crew

app = FastAPI()

@app.post("/analyze")
async def analyze(topic: str):
    crew = create_research_crew(topic)
    result = crew.kickoff()
    return {"result": result}

@app.post("/analyze/stream")
async def analyze_stream(topic: str):
    crew = create_research_crew(topic)
    async for chunk in crew.kickoff_stream():
        yield chunk

异步执行

import asyncio

async def run_crews_concurrently(topics):
    crews = [create_research_crew(topic) for topic in topics]
    results = await asyncio.gather(*[crew.kickoff_async() for crew in crews])
    return results

监控与调试

from crewai import CrewMonitor

monitor = CrewMonitor()

@monitor.trace
def run_crew(crew, input_data):
    result = crew.kickoff(inputs=input_data)
    return result

# 查看执行详情
monitor.print_summary()
# 包括：每个Agent的执行时间、token消耗、输出质量

应用场景

场景一：内容生产

# 内容生产团队
content_crew = Crew(
    agents=[
        Agent(role='选题策划', ...),
        Agent(role='资料收集', ...),
        Agent(role='内容撰写', ...),
        Agent(role='排版编辑', ...),
        Agent(role='SEO优化', ...),
    ],
    tasks=[...],
    process=Process.sequential
)

场景二：代码审查

# 代码审查团队
review_crew = Crew(
    agents=[
        Agent(role='代码审查员', tools=[read_file, code_analyzer]),
        Agent(role='安全审计员', tools=[vulnerability_scanner]),
        Agent(role='性能分析师', tools=[profiler]),
        Agent(role='报告生成者'),
    ],
    tasks=[...]
)

场景三：数据分析

# 数据分析团队
data_crew = Crew(
    agents=[
        Agent(role='数据工程师', tools=[sql_tool, python_tool]),
        Agent(role='数据分析师', tools=[statistical_tool]),
        Agent(role='可视化专家', tools=[chart_tool]),
        Agent(role='报告撰写者'),
    ],
    tasks=[...]
)

结语

CrewAI在2026年已经成为生产环境中最流行的多智能体框架之一。它的角色扮演式设计让AI协作变得自然直观，丰富的工具集成和记忆系统让它能胜任复杂的实际任务。

记住：好的Crew就像好的团队——角色明确、分工合理、流程清晰、质量可控。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。