一、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());
}
}
}九、最佳实践
- 角色定义清晰：每个Agent的角色、目标、背景故事要具体明确
- 任务粒度适中：任务不宜过大或过小，保持单一职责
- 合理选择流程：简单任务用顺序，独立任务用并行，复杂任务用层级
- 工具精简：每个Agent只配置必要的工具，避免混淆
- 错误处理：为Agent设置max_iterations限制，防止无限循环
十、总结
CrewAI让多Agent协作变得简单直观。通过定义角色、任务和流程，可以快速构建自动化AI工作流。适用于：自动化报告生成、代码审查、内容创作、数据分析等场景。相比单Agent，多Agent系统分工明确、可扩展性强，是构建复杂AI应用的理想选择。
文章摘自：https://www.cnblogs.com/czlws/p/19858139/crewai-multi-agent-collaboration-tutorial
京ICP备14020293号-4