CrewAI：让 AI 像团队一样工作 CrewAI 的核心理念简单而强大：把 Agent 组织成"团队"，每个 Agent 扮演一个角色，分工协作完成任务。
核心概念 Crew（团队） ├── Agent 1: 研究员（搜索资料） ├── Agent 2: 分析师（分析数据） ├── Agent 3: 写作者（生成报告） ├── Task 1: 搜索相关资料 ├── Task 2: 分析搜索结果 ├── Task 3: 撰写总结报告 └── Process: 顺序执行 / 层级管理 快速上手 from crewai import Agent, Task, Crew, Process # 1. 定义 Agent researcher = Agent( role="资深研究员", goal="找到关于 {topic} 的最新、最准确的信息", backstory="你是一位有 20 年经验的研究员，擅长从海量信息中提取关键洞察。", verbose=True, llm="qwen3-72b", tools=[search_tool, scrape_tool], ) analyst = Agent( role="数据分析师", goal="深入分析研究结果，发现趋势和模式", backstory="你是前麦肯锡分析师，擅长数据驱动的战略分析。", verbose=True, llm="qwen3-72b", ) writer = Agent( role="技术写作专家", goal="将分析结果转化为清晰、有洞见的报告", backstory="你是前 Wired 杂志编辑，擅长把复杂技术概念讲得通俗易懂。", verbose=True, llm="qwen3-72b", ) # 2. 定义 Task research_task = Task( description="研究 {topic} 的最新进展，包括技术突破、市场动态和未来趋势。", expected_output="一份包含 10 个关键发现的研究简报", agent=researcher, ) analysis_task = Task( description="基于研究结果，分析 {topic} 对行业的影响和机会。", expected_output="一份包含 SWOT 分析和 3 个战略建议的分析报告", agent=analyst, context=[research_task], # 依赖研究任务的输出 ) writing_task = Task( description="将分析和研究整合为一份面向高管的执行摘要。", expected_output="一份 2000 字的执行摘要，包含图表建议", agent=writer, context=[research_task, analysis_task], ) # 3. 组建 Crew crew = Crew( agents=[researcher, analyst, writer], tasks=[research_task, analysis_task, writing_task], process=Process.sequential, # 顺序执行 verbose=True, ) # 4. 执行 result = crew.kickoff(inputs={"topic": "AI Agent 框架"}) 执行模式 顺序模式 # 任务按顺序执行，前者输出作为后者输入 crew = Crew( agents=[agent_a, agent_b, agent_c], tasks=[task_1, task_2, task_3], process=Process.sequential, ) # task_1 → task_2 → task_3 层级模式 # 经理 Agent 负责分配任务 manager = Agent( role="项目经理", goal="协调团队高效完成任务", backstory="你是经验丰富的 PM，擅长任务分解和团队管理。", allow_delegation=True, ) crew = Crew( agents=[researcher, analyst, writer, manager], tasks=[complex_task], process=Process.hierarchical, manager_llm="qwen3-72b", ) # manager 分配子任务给其他 Agent 自定义流程 from crewai import Pipeline # 自定义流程：研究→分析→如果需要→再研究 pipeline = Pipeline( stages=[ research_crew, # 第一阶段：研究 review_crew, # 第二阶段：审查 revision_crew, # 第三阶段：修订（如果审查不通过） ] ) Agent 间通信 # Agent 可以委派任务给其他 Agent researcher = Agent( role="研究员", goal="研究给定主题", allow_delegation=False, # 不允许委派 ) manager = Agent( role="项目经理", goal="管理项目进度", allow_delegation=True, # 允许委派 ) # 委派示例 # manager: "我需要数据分析，请 analyst 处理" # analyst: "收到，开始分析..." # analyst: "分析完成，结果如下..." # manager: "收到，请 writer 基于分析结果写报告" 工具集成 from crewai_tools import ( SerperDevTool, # Google 搜索 ScrapeWebsiteTool, # 网页抓取 PDFSearchTool, # PDF 搜索 CSVSearchTool, # CSV 搜索 DirectoryReadTool, # 目录读取 FileReadTool, # 文件读取 ) # 预置工具 search = SerperDevTool() scraper = ScrapeWebsiteTool() # 自定义工具 from crewai_tools import BaseTool class DatabaseQueryTool(BaseTool): name: str = "数据库查询" description: str = "查询业务数据库，返回销售数据" def _run(self, query: str) -> str: results = db.execute(query) return json.dumps(results) db_tool = DatabaseQueryTool() researcher = Agent( role="数据研究员", tools=[search, scraper, db_tool], ) 记忆系统 # CrewAI 内置记忆系统 crew = Crew( agents=[agent_a, agent_b], tasks=[task_1, task_2], memory=True, # 启用记忆 embedder={ # 配置嵌入模型 "provider": "ollama", "config": {"model": "nomic-embed-text"}, }, ) # 记忆类型： # 1. 短期记忆：当前任务上下文 # 2. 长期记忆：跨任务的知识存储 # 3. 实体记忆：记住关键实体（人名、公司等） 真实案例：技术调研 # 案例：调研某个技术方向 # Agent 定义 tech_researcher = Agent( role="技术研究员", goal="深入调研 {tech} 的技术原理、优缺点和适用场景", backstory="你是前 Google 研究员，擅长技术深度分析。", tools=[search, scraper, github_tool], ) market_analyst = Agent( role="市场分析师", goal="分析 {tech} 的市场采用情况、竞品对比和投资机会", backstory="你是 a16z 的技术投资人，关注新兴技术趋势。", tools=[search, crunchbase_tool], ) architect = Agent( role="解决方案架构师", goal="设计基于 {tech} 的技术方案和实施路径", backstory="你是 AWS 首席架构师，擅长将技术转化为方案。", ) # 任务定义 tasks = [ Task( description="调研 {tech} 的核心技术原理、开源实现和性能基准", agent=tech_researcher, expected_output="技术调研报告（含代码示例和性能对比）", ), Task( description="分析 {tech} 的市场格局、主要玩家和投资趋势", agent=market_analyst, expected_output="市场分析报告（含竞品矩阵和投资建议）", context=[tasks[0]], ), Task( description="基于调研结果，设计一个 POC 技术方案", agent=architect, expected_output="技术方案文档（含架构图和实施计划）", context=[tasks[0], tasks[1]], ), ] crew = Crew(agents=[tech_researcher, market_analyst, architect], tasks=tasks) result = crew.kickoff(inputs={"tech": "MCP 协议"}) 与 LangGraph 对比 维度 CrewAI LangGraph 范式 角色扮演 图式编排 上手难度 低 高 灵活性 中 高 多 Agent 原生支持 需要构建 状态管理 自动 手动定义 循环控制 有限 完全支持 适合 内容创作、调研 复杂工作流 生产就绪 中 高 最佳实践 best_practices = { "角色定义": "backstory 越具体，Agent 表现越好", "任务粒度": "每个任务应该有明确的输入和输出", "工具选择": "给 Agent 它需要的工具，不要给多余的", "LLM 选择": "规划用大模型（72B），执行用小模型（7B）", "成本控制": "设置 max_iter 和 max_rpm 限制", "结果验证": "添加 review_task 让另一个 Agent 验证结果", } 结论 CrewAI 的优势在于"角色扮演"范式——用人类团队的思维组织 Agent：
...
