CrewAI多Agent协作实战：构建自动化AI工作流

2026/7/2 5:07:59

一、CrewAI简介

CrewAI是一个用于编排多Agent协作的Python框架，灵感来自AutoGen但更加简洁易用。它允许开发者定义多个AI Agent，每个Agent有特定角色、目标和工具，Agent之间可以互相委托任务，形成工作流。相比单Agent，多Agent系统可以处理更复杂的任务，如研究报告生成、代码审查、数据分析等。

回到顶部

二、核心概念

Agent（智能体）：具有角色、目标、背景故事的AI实体 Task（任务）：Agent需要完成的具体工作 Crew（团队）：一组协作的Agent和任务 Process（流程）：任务执行方式（顺序/并行/层级） Tool（工具）：Agent可调用的外部功能

回到顶部

三、环境搭建

# 安装CrewAI pip install crewai # 安装LangChain工具（可选） pip install langchain langchain-openai # 安装搜索工具 pip install duckduckgo-search

回到顶部

四、基础示例：研究团队

from crewai import Agent, Task, Crew, Process from langchain_openai import ChatOpenAI from langchain.tools import DuckDuckGoSearchRun # 初始化LLM llm = ChatOpenAI(model="gpt-4o", temperature=0.7) # 定义工具 search_tool = DuckDuckGoSearchRun() # 定义Agent researcher = Agent( role="高级研究员", goal="深入研究技术主题，收集全面信息", backstory="你是一位经验丰富的技术研究员，擅长快速学习新技术并撰写详细报告。", verbose=True, allow_delegation=False, tools=[search_tool], llm=llm ) writer = Agent( role="技术作家", goal="将技术信息转化为易懂的文章", backstory="你是一位资深技术作家，擅长将复杂概念用通俗语言解释。", verbose=True, allow_delegation=False, llm=llm ) # 定义任务 research_task = Task( description="研究{topic}的最新发展趋势，包括：1)核心概念 2)主要应用场景 3)优缺点对比 4)学习资源推荐", expected_output="一份详细的研究报告，包含上述4个部分", agent=researcher ) write_task = Task( description="基于研究报告，撰写一篇面向初学者的入门指南，要求：1)结构清晰 2)包含代码示例 3)通俗易懂", expected_output="一篇2000字以上的技术博客文章", agent=writer, context=[research_task] # 依赖research_task的结果 ) # 创建Crew crew = Crew( agents=[researcher, writer], tasks=[research_task, write_task], process=Process.sequential, # 顺序执行 verbose=2 ) # 运行 result = crew.kickoff(inputs={"topic": "CrewAI多Agent框架"}) print(result)

回到顶部

五、进阶示例：代码审查团队

from crewai import Agent, Task, Crew # 代码审查Agent code_reviewer = Agent( role="代码审查专家", goal="审查代码质量，发现潜在问题", backstory="你是一位有10年经验的代码审查专家，擅长发现代码中的bug、性能问题和安全隐患。", verbose=True ) security_expert = Agent( role="安全专家", goal="识别代码中的安全漏洞", backstory="你是一位网络安全专家，专注于代码安全审计和漏洞挖掘。", verbose=True ) performance_expert = Agent( role="性能优化专家", goal="发现性能瓶颈并提供优化建议", backstory="你是一位性能优化专家，擅长分析和优化代码性能。", verbose=True ) # 定义任务 review_task = Task( description="审查以下Python代码，找出bug和代码质量问题：\n{code}", expected_output="代码审查报告，列出发现的问题", agent=code_reviewer ) security_task = Task( description="分析代码的安全风险，包括：1)注入攻击 2)敏感信息泄露 3)权限控制", expected_output="安全审计报告", agent=security_expert ) performance_task = Task( description="分析代码性能，识别：1)时间复杂度问题 2)内存泄漏 3)I/O瓶颈", expected_output="性能优化建议报告", agent=performance_expert ) # 并行执行 crew = Crew( agents=[code_reviewer, security_expert, performance_expert], tasks=[review_task, security_task, performance_task], process=Process.parallel, # 并行执行 verbose=2 ) # 测试代码 test_code = """ def get_user_data(user_id): query = f"SELECT * FROM users WHERE id = {user_id}" return db.execute(query) """ result = crew.kickoff(inputs={"code": test_code}) print(result)

回到顶部

六、层级流程示例：项目经理模式

from crewai import Agent, Task, Crew, Process # 项目经理Agent（可以委托任务给其他Agent） project_manager = Agent( role="项目经理", goal="协调团队成员完成任务", backstory="你是一位经验丰富的项目经理，擅长任务分解和团队协调。", verbose=True, allow_delegation=True # 允许委托 ) # 开发团队成员 frontend_dev = Agent( role="前端开发", goal="完成前端开发任务", backstory="React/Vue专家", verbose=True ) backend_dev = Agent( role="后端开发", goal="完成后端API开发", backstory="Java/Python后端专家", verbose=True ) devops_engineer = Agent( role="DevOps工程师", goal="完成部署和CI/CD配置", backstory="Docker/K8s专家", verbose=True ) # 任务 develop_feature = Task( description="开发新功能：用户登录模块", expected_output="完整的登录功能实现方案", agent=project_manager ) # 层级流程：项目经理可以委托给团队成员 crew = Crew( agents=[project_manager, frontend_dev, backend_dev, devops_engineer], tasks=[develop_feature], process=Process.hierarchical, # 层级流程 verbose=2 ) result = crew.kickoff() print(result)

回到顶部

七、自定义工具

from crewai.tools import BaseTool from pydantic import BaseModel, Field class CodeAnalysisInput(BaseModel): """代码分析工具输入参数""" code: str = Field(..., description="要分析的代码") language: str = Field(default="python", description="编程语言") class CodeAnalysisTool(BaseTool): name: str = "代码分析工具" description: str = "分析代码质量和潜在问题" args_schema: type[BaseModel] = CodeAnalysisInput def _run(self, code: str, language: str = "python") -> str: # 这里可以接入真实的代码分析服务 # 如SonarQube、CodeClimate等 lines = code.strip().split("\n") issues = [] for i, line in enumerate(lines, 1): if len(line) > 100: issues.append(f"行{i}: 代码过长({len(line)}字符)") if "TODO" in line: issues.append(f"行{i}: 发现TODO待办事项") return "\n".join(issues) if issues else "未发现明显问题" # 使用自定义工具 analyzer = Agent( role="代码分析师", goal="分析代码质量", tools=[CodeAnalysisTool()], verbose=True )

回到顶部

八、与Spring Boot集成

# 通过REST API调用CrewAI # Spring Boot Controller @RestController @RequestMapping("/api/crew") public class CrewAIController { @PostMapping("/research") public ResponseEntity<String> research(@RequestBody Map<String, String> body) { String topic = body.get("topic"); // 调用Python脚本 ProcessBuilder pb = new ProcessBuilder( "python3", "crew_research.py", topic ); pb.redirectErrorStream(true); try { Process process = pb.start(); BufferedReader reader = new BufferedReader( new InputStreamReader(process.getInputStream()) ); StringBuilder output = new StringBuilder(); String line; while ((line = reader.readLine()) != null) { output.append(line).append("\n"); } process.waitFor(); return ResponseEntity.ok(output.toString()); } catch (Exception e) { return ResponseEntity.status(500).body(e.getMessage()); } } }

回到顶部