一、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(modelgpt-4o, temperature0.7) # 定义工具 search_tool DuckDuckGoSearchRun() # 定义Agent researcher Agent( role高级研究员, goal深入研究技术主题收集全面信息, backstory你是一位经验丰富的技术研究员擅长快速学习新技术并撰写详细报告。, verboseTrue, allow_delegationFalse, tools[search_tool], llmllm ) writer Agent( role技术作家, goal将技术信息转化为易懂的文章, backstory你是一位资深技术作家擅长将复杂概念用通俗语言解释。, verboseTrue, allow_delegationFalse, llmllm ) # 定义任务 research_task Task( description研究{topic}的最新发展趋势包括1)核心概念 2)主要应用场景 3)优缺点对比 4)学习资源推荐, expected_output一份详细的研究报告包含上述4个部分, agentresearcher ) write_task Task( description基于研究报告撰写一篇面向初学者的入门指南要求1)结构清晰 2)包含代码示例 3)通俗易懂, expected_output一篇2000字以上的技术博客文章, agentwriter, context[research_task] # 依赖research_task的结果 ) # 创建Crew crew Crew( agents[researcher, writer], tasks[research_task, write_task], processProcess.sequential, # 顺序执行 verbose2 ) # 运行 result crew.kickoff(inputs{topic: CrewAI多Agent框架}) print(result)回到顶部五、进阶示例代码审查团队from crewai import Agent, Task, Crew # 代码审查Agent code_reviewer Agent( role代码审查专家, goal审查代码质量发现潜在问题, backstory你是一位有10年经验的代码审查专家擅长发现代码中的bug、性能问题和安全隐患。, verboseTrue ) security_expert Agent( role安全专家, goal识别代码中的安全漏洞, backstory你是一位网络安全专家专注于代码安全审计和漏洞挖掘。, verboseTrue ) performance_expert Agent( role性能优化专家, goal发现性能瓶颈并提供优化建议, backstory你是一位性能优化专家擅长分析和优化代码性能。, verboseTrue ) # 定义任务 review_task Task( description审查以下Python代码找出bug和代码质量问题\n{code}, expected_output代码审查报告列出发现的问题, agentcode_reviewer ) security_task Task( description分析代码的安全风险包括1)注入攻击 2)敏感信息泄露 3)权限控制, expected_output安全审计报告, agentsecurity_expert ) performance_task Task( description分析代码性能识别1)时间复杂度问题 2)内存泄漏 3)I/O瓶颈, expected_output性能优化建议报告, agentperformance_expert ) # 并行执行 crew Crew( agents[code_reviewer, security_expert, performance_expert], tasks[review_task, security_task, performance_task], processProcess.parallel, # 并行执行 verbose2 ) # 测试代码 test_code def get_user_data(user_id): query fSELECT * 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你是一位经验丰富的项目经理擅长任务分解和团队协调。, verboseTrue, allow_delegationTrue # 允许委托 ) # 开发团队成员 frontend_dev Agent( role前端开发, goal完成前端开发任务, backstoryReact/Vue专家, verboseTrue ) backend_dev Agent( role后端开发, goal完成后端API开发, backstoryJava/Python后端专家, verboseTrue ) devops_engineer Agent( roleDevOps工程师, goal完成部署和CI/CD配置, backstoryDocker/K8s专家, verboseTrue ) # 任务 develop_feature Task( description开发新功能用户登录模块, expected_output完整的登录功能实现方案, agentproject_manager ) # 层级流程项目经理可以委托给团队成员 crew Crew( agents[project_manager, frontend_dev, backend_dev, devops_engineer], tasks[develop_feature], processProcess.hierarchical, # 层级流程 verbose2 ) 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(defaultpython, 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()], verboseTrue )回到顶部八、与Spring Boot集成# 通过REST API调用CrewAI # Spring Boot Controller RestController RequestMapping(/api/crew) public class CrewAIController { PostMapping(/research) public ResponseEntityString research(RequestBody MapString, 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()); } } }回到顶部九、最佳实践角色定义清晰每个Agent的角色、目标、背景故事要具体明确任务粒度适中任务不宜过大或过小保持单一职责合理选择流程简单任务用顺序独立任务用并行复杂任务用层级工具精简每个Agent只配置必要的工具避免混淆