上一篇聊了 AutoGen 的多 Agent 协作,今天换一个框架——CrewAI。
如果说 AutoGen 像是给你一堆零件让你自己组装,CrewAI 更像是一个开箱即用的团队管理工具。它用三个核心概念把多 Agent 协作简化了:Agent(成员)、Task(任务)、Crew(团队)。定义好这三样东西,剩下的交给框架处理。
CrewAI 的三个核心概念
Agent:团队成员。每个 Agent 有角色(role)、目标(goal)、背景故事(backstory),这些决定了它的行为方式。
Task:具体的任务。每个 Task 有描述、预期输出格式,并指定由哪个 Agent 负责。
Crew:团队。把 Agent 和 Task 组合在一起,定义执行模式(顺序执行还是层级管理)。
这三层抽象很直觉——跟真实团队管理差不多。
环境准备
1
| pip install crewai crewai-tools
|
确保 Hermes 通过 Ollama 跑着:
快速上手:一个最小的 Crew
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
| from crewai import Agent, Task, Crew, LLM
llm = LLM( model="ollama/hermes3:8b", base_url="http://localhost:11434" )
researcher = Agent( role="技术研究员", goal="深入研究指定技术主题,提供准确全面的信息", backstory="你是一个有10年经验的技术研究员,擅长快速掌握新技术并整理成结构化的知识。", llm=llm, verbose=True )
research_task = Task( description="研究 WebAssembly 在服务端的应用现状,包括主要运行时、使用场景和优缺点。", expected_output="一份结构化的研究报告,包含:概述、主要运行时列表、使用场景分析、优缺点对比。", agent=researcher )
crew = Crew( agents=[researcher], tasks=[research_task], verbose=True )
result = crew.kickoff() print(result)
|
跑起来后你会看到 Agent 在思考、搜索(如果配了工具)、生成报告的全过程。verbose=True 会打印详细的执行日志。
多角色团队协作
内容生产团队
一个写博客的典型团队:研究员收集素材,作者写文章,编辑润色。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98
| from crewai import Agent, Task, Crew, Process, LLM
llm = LLM( model="ollama/hermes3:8b", base_url="http://localhost:11434" )
researcher = Agent( role="内容研究员", goal="为文章收集准确的技术信息和数据", backstory="""你是一个技术内容研究员,熟悉各种编程语言和框架。 你的工作是为文章提供准确的素材,包括技术细节、代码示例和最新动态。 你特别注重信息的准确性,宁可说不知道也不编造。""", llm=llm, allow_delegation=False )
writer = Agent( role="技术博客作者", goal="写出有深度、好读的技术文章", backstory="""你是一个有着丰富写作经验的技术博主。 你的文章特点是技术准确但语言轻松,喜欢用类比和实例来解释复杂概念。 你写的文章结构清晰,有引入、有正文、有总结,代码示例都是经过验证的。""", llm=llm, allow_delegation=False )
editor = Agent( role="内容编辑", goal="确保文章质量达标:技术准确、逻辑连贯、表达流畅", backstory="""你是一个经验丰富的技术内容编辑。 你关注的重点: 1. 技术描述是否准确 2. 逻辑是否连贯 3. 是否有遗漏的重要信息 4. 表达是否简洁清晰 你会给出具体的修改建议而不是笼统的评价。""", llm=llm, allow_delegation=False )
research_task = Task( description="""研究以下主题:{topic} 需要收集的信息: 1. 技术原理和核心概念 2. 实际应用场景 3. 代码示例(可运行的) 4. 优缺点分析 5. 与竞品的对比""", expected_output="一份包含以上5个方面的研究笔记,每个方面至少200字", agent=researcher )
writing_task = Task( description="""基于研究笔记撰写一篇技术博客文章。 要求: - 2000字左右 - 有吸引人的开头 - 技术内容准确且有深度 - 包含可运行的代码示例 - 语言风格轻松但专业""", expected_output="一篇完整的 Markdown 格式技术博客文章", agent=writer, context=[research_task] )
editing_task = Task( description="""审校文章,检查以下方面: 1. 技术准确性 2. 逻辑连贯性 3. 代码示例的正确性 4. 排版和格式 给出修改建议和修改后的最终版本。""", expected_output="修改建议列表 + 最终版文章", agent=editor, context=[writing_task] )
crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, writing_task, editing_task], process=Process.sequential, verbose=True )
result = crew.kickoff(inputs={"topic": "Rust 在 Web 后端开发中的应用"}) print(result)
|
代码开发团队
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
| from crewai import Agent, Task, Crew, Process, LLM
llm = LLM( model="ollama/hermes3:8b", base_url="http://localhost:11434" )
architect = Agent( role="软件架构师", goal="设计清晰可扩展的系统架构", backstory="""你是一个有15年经验的架构师。 你的设计原则:简单优先、关注点分离、面向接口编程。 你会输出清晰的模块划分、接口定义和数据流设计。""", llm=llm )
developer = Agent( role="Python 开发者", goal="按照架构设计编写高质量代码", backstory="""你是一个 Python 高手,擅长写干净、高效的代码。 你遵循 SOLID 原则,代码有完善的类型注解和文档字符串。 你会为关键函数编写单元测试。""", llm=llm )
qa_engineer = Agent( role="QA 工程师", goal="确保代码质量和功能正确性", backstory="""你是一个严格的 QA 工程师。 你擅长设计边界测试用例,找到那些开发者容易忽略的 bug。 你会用 pytest 编写测试,覆盖正常路径和异常路径。""", llm=llm )
design_task = Task( description="""为以下需求设计系统架构:{requirement} 输出: 1. 模块划分和职责说明 2. 核心类和接口定义 3. 数据流说明 4. 关键技术选型""", expected_output="完整的架构设计文档", agent=architect )
coding_task = Task( description="""基于架构设计文档实现代码。 要求: - 完整可运行的 Python 代码 - 有类型注解 - 有文档字符串 - 遵循架构设计的模块划分""", expected_output="完整的 Python 源代码", agent=developer, context=[design_task] )
testing_task = Task( description="""为实现的代码编写测试。 要求: - 使用 pytest - 覆盖正常路径和异常路径 - 包含边界条件测试 - 给出测试覆盖率估计""", expected_output="完整的 pytest 测试代码 + 测试说明", agent=qa_engineer, context=[coding_task] )
crew = Crew( agents=[architect, developer, qa_engineer], tasks=[design_task, coding_task, testing_task], process=Process.sequential, verbose=True )
result = crew.kickoff(inputs={ "requirement": "一个命令行密码管理器,支持添加、查询、删除密码,数据用 AES 加密存储在本地" }) print(result)
|
给 Agent 配备工具
CrewAI 支持给 Agent 配备工具,让它们能与外部世界交互:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
| from crewai import Agent, Task, Crew, LLM from crewai_tools import SerperDevTool, FileReadTool, DirectoryReadTool
llm = LLM( model="ollama/hermes3:8b", base_url="http://localhost:11434" )
file_reader = FileReadTool() dir_reader = DirectoryReadTool()
code_analyst = Agent( role="代码分析师", goal="分析代码库的结构和质量", backstory="你是一个专业的代码审计师,擅长分析代码架构、发现潜在问题。", tools=[file_reader, dir_reader], llm=llm )
analysis_task = Task( description="分析 ./src 目录下的 Python 代码,给出代码质量报告", expected_output="代码质量报告,包含:架构评价、代码问题列表、改进建议", agent=code_analyst )
crew = Crew( agents=[code_analyst], tasks=[analysis_task], verbose=True )
result = crew.kickoff()
|
自定义工具
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39
| from crewai.tools import BaseTool from pydantic import BaseModel, Field import subprocess
class CodeExecutionInput(BaseModel): code: str = Field(description="要执行的 Python 代码")
class PythonExecutor(BaseTool): name: str = "Python 代码执行器" description: str = "执行 Python 代码并返回结果" args_schema: type[BaseModel] = CodeExecutionInput
def _run(self, code: str) -> str: try: result = subprocess.run( ["python", "-c", code], capture_output=True, text=True, timeout=30 ) output = result.stdout if result.returncode != 0: output = f"错误:\n{result.stderr}" return output if output else "执行成功,无输出" except subprocess.TimeoutExpired: return "执行超时" except Exception as e: return f"执行失败: {e}"
executor = PythonExecutor()
coder = Agent( role="编程助手", goal="写代码并验证其正确性", backstory="你是一个注重代码质量的开发者,写完代码一定会运行验证。", tools=[executor], llm=llm )
|
执行模式
Sequential(顺序执行)
任务按定义的顺序依次执行,前一个任务的输出作为后一个的输入:
1 2 3 4 5
| crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, writing_task, editing_task], process=Process.sequential )
|
适合流水线式的工作流,比如上面的内容生产流程。
Hierarchical(层级执行)
设一个”经理”Agent 来调度其他 Agent:
1 2 3 4 5 6
| crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, writing_task, editing_task], process=Process.hierarchical, manager_llm=llm )
|
层级模式下,经理会自动决定任务分配、验收和重新分配。更接近真实的团队管理,但也更消耗模型推理资源。
输出格式控制
让 Agent 输出结构化数据:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
| from pydantic import BaseModel from typing import List
class ArticleOutline(BaseModel): title: str sections: List[str] target_audience: str estimated_word_count: int
outline_task = Task( description="为主题 '{topic}' 创建一个文章大纲", expected_output="结构化的文章大纲", agent=researcher, output_pydantic=ArticleOutline )
|
这样任务的输出就是一个 Pydantic 对象,方便后续代码处理。
记忆功能
CrewAI 支持长期记忆,让 Agent 能记住之前的交互:
1 2 3 4 5 6
| crew = Crew( agents=[researcher, writer], tasks=[research_task, writing_task], memory=True, verbose=True )
|
启用后,Agent 在后续任务中可以参考之前的对话和发现。这对于需要多次迭代的工作流特别有用。
踩坑和经验
1. Hermes 的 system prompt 遵循度
CrewAI 会把 Agent 的 role、goal、backstory 组合成一个很长的 system prompt。Hermes 在这方面表现不错,但如果 prompt 太长,8B 模型可能会”忘记”一些指令。建议 backstory 不要太啰嗦,抓住重点。
2. 任务输出的传递
context=[previous_task] 会把前一个任务的完整输出传给当前任务。如果前一个任务输出了几千字的内容,加上当前任务的 prompt,可能会超出上下文窗口。可以在 Ollama 里把 num_ctx 调大。
3. 执行时间
多 Agent 协作天然就是多次模型调用。三个 Agent 做三个任务,至少 3-6 次推理。在本地 8B 模型上,一个完整的 Crew 执行可能需要 3-10 分钟,取决于任务复杂度和硬件配置。
4. 错误处理
Agent 有时会生成格式不对的输出导致下游任务失败。可以设置重试:
1 2 3 4 5 6
| task = Task( description="...", expected_output="...", agent=some_agent, max_retries=3 )
|
和 AutoGen 怎么选
在 cocoloop 社区里经常有人问这个问题,我的建议:
- 想快速搭原型 -> CrewAI
- 需要复杂的对话模式(嵌套、条件分支)-> AutoGen
- 团队里有非技术人员需要理解流程 -> CrewAI(概念更直觉)
- 需要精细控制每次交互 -> AutoGen
两个框架底层都可以用 Hermes 驱动,切换成本不高。
如果你想要更图形化的方式来编排 Agent 工作流,可以看看 Hermes + Dify 的零代码方案。
CrewAI 把多 Agent 协作的复杂度降到了一个很友好的水平。Agent、Task、Crew 三个概念,加上 Hermes 的角色扮演能力,你可以快速组建出各种专业的 AI 团队。从内容生产到代码开发到数据分析,想象空间很大。