引言：像管理团队一样管理 AI CrewAI 的核心理念是：将 AI Agent 组织成一个"团队"（Crew），每个成员有明确的角色、目标和背景故事，通过结构化的流程协作完成任务。2026 年的 CrewAI 3.0 版本在稳定性、可观测性和企业级特性上有了长足进步。
CrewAI 核心概念 三要素模型 Crew = Agents + Tasks + Process 概念 说明 类比 Agent 有角色和能力的 AI 成员 团队成员 Task 需要完成的具体任务 工作任务 Crew Agent 和 Task 的编排单元 项目团队 Process 任务执行方式（顺序/层次） 工作流程 Tool Agent 可使用的工具 工具软件 快速构建第一个 Crew 安装与基础 # pip install crewai crewai-tools from crewai import Agent, Task, Crew, Process from crewai.tools import tool from langchain_openai import ChatOpenAI # 配置 LLM llm = ChatOpenAI(model="gpt-5", temperature=0.7) 定义 Agent # 定义一个内容创作团队 strategist = Agent( role='内容策略师', goal='制定内容策略，确保内容符合品牌调性和目标受众需求', backstory=""" 你是一位拥有 10 年经验的内容策略师，曾在多家科技公司 负责内容规划。你擅长分析受众画像、制定内容日历和优化 内容分发策略。 """, llm=llm, verbose=True, memory=True, # 启用记忆 allow_delegation=False, ) writer = Agent( role='技术写作专家', goal='根据策略师的要求创作高质量的技术内容', backstory=""" 你是一位资深技术写作专家，擅长将复杂的技术概念 转化为通俗易懂的文章。你精通 Markdown 格式， 善于使用代码示例和图表来增强理解。 """, llm=llm, verbose=True, memory=True, ) editor = Agent( role='内容编辑', goal='审核并优化内容，确保质量和准确性', backstory=""" 你是一位严格但公正的内容编辑，关注语法准确性、 逻辑连贯性和事实正确性。你会提出具体的修改建议， 而不是泛泛而谈。 """, llm=llm, verbose=True, ) 定义 Task strategy_task = Task( description=""" 为一篇关于「AI Agent 在企业中的应用」的文章制定内容策略。 要求： 1. 目标受众：技术决策者（CTO、技术总监） 2. 内容长度：3000-5000 字 3. 风格：专业但不晦涩 4. 需要覆盖的关键点： - AI Agent 的定义和价值 - 主流框架对比 - 企业落地案例 - ROI 分析 输出：详细的内容大纲和写作要点。 """, expected_output="一份详细的内容大纲，包含章节结构和每章的要点", agent=strategist, ) writing_task = Task( description=""" 根据策略师的大纲，撰写完整的技术文章。 要求： - 使用 Markdown 格式 - 包含代码示例 - 包含对比表格 - 适当的图表说明 """, expected_output="一篇完整的 Markdown 格式技术文章", agent=writer, context=[strategy_task], # 依赖前一个任务的输出 ) editing_task = Task( description=""" 审查文章并优化： 1. 检查事实准确性 2. 优化段落过渡 3. 修正语法错误 4. 确保风格一致性 5. 输出最终版本 """, expected_output="审查意见 + 最终优化版文章", agent=editor, context=[writing_task], output_file='output/final_article.md', # 保存到文件 ) 组建 Crew 并执行 # 创建 Crew content_crew = Crew( agents=[strategist, writer, editor], tasks=[strategy_task, writing_task, editing_task], process=Process.sequential, # 顺序执行 verbose=True, memory=True, # 团队级记忆 planning=True, # 执行前先规划 # 层次模式配置 # process=Process.hierarchical, # manager_llm=llm, ) # 执行 result = content_crew.kickoff() print(result.raw) 自定义工具集成 创建自定义工具 from crewai.tools import BaseTool from pydantic import BaseModel, Field import requests import json class WebSearchInput(BaseModel): query: str = Field(..., description="搜索关键词") max_results: int = Field(default=5, description="最大返回结果数") class WebSearchTool(BaseTool): name: str = "web_search" description: str = "搜索互联网获取最新信息" args_schema: type[BaseModel] = WebSearchInput def _run(self, query: str, max_results: int = 5) -> str: # 使用搜索 API response = requests.get( "https://api.search.brave.com/res/v1/web/search", headers={"X-Subscription-Token": "YOUR_API_KEY"}, params={"q": query, "count": max_results} ) results = response.json().get("web", {}).get("results", []) formatted = [] for r in results: formatted.append(f"标题: {r['title']}\n链接: {r['url']}\n摘要: {r['description']}\n") return "\n".join(formatted) if formatted else "未找到相关结果" class DatabaseQueryInput(BaseModel): sql: str = Field(..., description="SQL 查询语句") class DatabaseQueryTool(BaseTool): name: str = "database_query" description: str = "查询企业数据库获取业务数据" args_schema: type[BaseModel] = DatabaseQueryInput def _run(self, sql: str) -> str: import sqlite3 conn = sqlite3.connect("data/company.db") cursor = conn.execute(sql) columns = [d[0] for d in cursor.description] rows = cursor.fetchall() conn.close() # 格式化为表格 result = [", ".join(columns)] for row in rows[:20]: # 限制返回行数 result.append(", ".join(str(v) for v in row)) return "\n".join(result) # 将工具分配给 Agent researcher = Agent( role='市场研究员', goal='收集和分析市场数据', backstory='你是市场研究专家，擅长数据收集和分析。', llm=llm, tools=[WebSearchTool(), DatabaseQueryTool()], ) 层次化团队模式 # 层次模式：管理者 Agent 负责任务分配 from crewai import Process manager = Agent( role='项目经理', goal='协调团队成员，确保任务按时高质量完成', backstory='你是一位经验丰富的项目经理，擅长资源分配和进度管理。', llm=llm, allow_delegation=True, # 允许委派任务 ) hierarchical_crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, writing_task, editing_task], process=Process.hierarchical, manager_agent=manager, # 指定管理者 manager_llm=llm, verbose=True, ) 生产级配置 错误处理与重试 from crewai import Crew from crewai.utilities import Logger class ProductionCrew: def __init__(self, config: dict): self.config = config self.max_retries = config.get("max_retries", 3) self.logger = Logger(verbose=True) def run_with_retry(self, crew: Crew, inputs: dict) -> dict: """带重试机制的 Crew 执行""" for attempt in range(self.max_retries): try: result = crew.kickoff(inputs=inputs) return { "status": "success", "result": result.raw, "token_usage": result.token_usage, } except Exception as e: self.logger.log(f"尝试 {attempt+1} 失败: {e}") if attempt < self.max_retries - 1: # 调整重试策略 crew.llm.temperature = min(crew.llm.temperature + 0.1, 1.0) continue return { "status": "failed", "error": str(e), "attempt": attempt + 1, } 监控与可观测性 # 使用 LangSmith 或自定义回调进行监控 from langchain.callbacks import get_openai_callback def monitored_crew_run(crew: Crew, task: str): """带监控的 Crew 执行""" # Token 使用追踪 with get_openai_callback() as cb: result = crew.kickoff(task) metrics = { "total_tokens": cb.total_tokens, "prompt_tokens": cb.prompt_tokens, "completion_tokens": cb.completion_tokens, "total_cost": cb.total_cost, "execution_time": result.execution_time, } # 发送到监控系统 send_metrics_to_dashboard(metrics) # 输出执行报告 print(f""" ═══════════════════════════════════ Crew 执行报告 ═══════════════════════════════════ 任务: {task[:50]}... 状态: 成功 Token 使用: {metrics['total_tokens']:,} 成本: ${metrics['total_cost']:.4f} 耗时: {metrics['execution_time']:.1f}s ═══════════════════════════════════ """) return result, metrics 缓存与优化 from crewai.utilities import CacheHandler import hashlib import json import os class FileCacheHandler(CacheHandler): """文件系统缓存""" def __init__(self, cache_dir: str = "./.crew_cache"): super().__init__() self.cache_dir = cache_dir os.makedirs(cache_dir, exist_ok=True) def _hash_key(self, agent_name: str, input_text: str) -> str: return hashlib.md5(f"{agent_name}:{input_text}".encode()).hexdigest() def get(self, agent_name: str, input_text: str): key = self._hash_key(agent_name, input_text) path = os.path.join(self.cache_dir, f"{key}.json") if os.path.exists(path): with open(path) as f: return json.load(f) return None def set(self, agent_name: str, input_text: str, value): key = self._hash_key(agent_name, input_text) path = os.path.join(self.cache_dir, f"{key}.json") with open(path, "w") as f: json.dump(value, f, ensure_ascii=False) # 使用缓存 crew = Crew( agents=[...], tasks=[...], cache_handler=FileCacheHandler(), # 启用嵌入缓存 embedder={ "provider": "openai", "config": { "model": "text-embedding-3-small", } }, ) 成本优化策略 策略 效果 实现难度 任务分配合适的模型 节省 40-60% 低 启用缓存 节省 20-30% 低 减少冗余上下文 节省 15-25% 中 并行化独立任务 节省 30-50% 时间 中 流式输出 改善用户体验 低 # 模型分层策略 models = { "planning": "gpt-5", # 规划用强模型 "execution": "gpt-4o-mini", # 执行用快速模型 "review": "gpt-5", # 审查用强模型 "summary": "gpt-4o-mini", # 总结用快速模型 } # 为不同 Agent 分配不同模型 strategist = Agent(role='策略师', llm=ChatOpenAI(model=models["planning"])) writer = Agent(role='写作者', llm=ChatOpenAI(model=models["execution"])) editor = Agent(role='编辑', llm=ChatOpenAI(model=models["review"])) 竞品对比 特性 CrewAI AutoGen LangGraph 学习曲线 低 中 高 角色定义 直观 灵活 程序化 工作流模式 顺序/层次 多种 自定义图 内置工具 丰富 基础 需集成 记忆系统 ✅ ✅ 需实现 人类介入 ✅ ✅ ✅ 适合场景 内容/分析 研究/开发 复杂流程 生产成熟度 中高 高 中 最佳实践 角色设计要明确：每个 Agent 的角色、目标和背景故事要有清晰区分 任务粒度适中：太粗导致质量差，太细导致开销大 合理使用上下文依赖：通过 context 参数传递任务间依赖 监控成本：多 Agent 会成倍增加 token 消耗 测试与迭代：先小规模测试，逐步扩展团队 结语 CrewAI 以其直观的角色-任务模型，成为构建 AI 团队最易上手的框架之一。它特别适合内容创作、市场分析、研究报告等"软"任务场景。对于需要精细控制执行流程的场景，可能需要结合其他框架使用。
...
