AI 时代的团队协同：深入探索 CrewAI 框架与多智能体实战

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
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool

# 1. 定义自定义工具 (Define Custom Tools)
@tool
def python_code_formatter(code_content: str) -> str:
    """Format python code string by removing unnecessary whitespaces and ensuring basic syntax rules."""
    return code_content.strip()

# 2. 定义具备进阶属性的智能体 (Define Agents)
product_manager = Agent(
    role="Product Manager",
    goal="Translate a rough user request into a clear and structured product requirements document (PRD).",
    backstory="You are an expert product manager who excels at understanding user needs and breaking them down into detailed technical requirements.",
    verbose=True,
    allow_delegation=False,
    max_iter=10
)

developer = Agent(
    role="Senior Software Engineer",
    goal="Design the software architecture and write clean, efficient python code based on the product requirements document.",
    backstory="You are an experienced software engineer who writes robust python code and implements designs correctly according to product specifications.",
    verbose=True,
    allow_delegation=True, # 允许 PM 或 QA 向开发提问
    tools=[python_code_formatter],
    max_iter=15
)

qa_engineer = Agent(
    role="QA Engineer",
    goal="Analyze the requirements and the generated code to design a comprehensive test suite and write test scripts.",
    backstory="You are a detail-oriented quality assurance engineer who ensures code quality by identifying edge cases and writing thorough tests.",
    verbose=True,
    allow_delegation=False,
    max_iter=10
)

# 3. 定义显式依赖与文件输出的任务 (Define Tasks)
user_idea = "A command-line tool that fetches daily weather information for a specific city and sends a summary via email if the temperature is below 0 degrees Celsius."

prd_task = Task(
    description=f"Analyze the following user idea: '{user_idea}'. Create a detailed and structured product requirements document (PRD) in Markdown format outlining the core features, input parameters, flow of execution, and edge cases.",
    expected_output="A structured product requirements document in Markdown format.",
    agent=product_manager,
    output_file="requirements.md" # 将 PM 的产出真实写入到本地磁盘文件
)

code_task = Task(
    description="Read the product requirements document (PRD) generated in the previous step. Write a fully functioning Python script. Ensure the code is modular, well-documented, and includes proper exception handling.",
    expected_output="A complete Python script implementing the weather alert command-line tool.",
    agent=developer,
    context=[prd_task], # 显式声明依赖于 prd_task 的上下文
    output_file="weather_alert.py" # 将 RD 的代码写入本地 Python 文件
)

test_task = Task(
    description="Analyze the product requirements document (PRD) and the generated Python script. Write a comprehensive test suite using Python unittest framework. Cover both happy paths and edge cases.",
    expected_output="A Python script containing unittest cases to verify the implementation.",
    agent=qa_engineer,
    context=[prd_task, code_task], # 显式将 PRD 和源码作为输入上下文
    output_file="test_weather_alert.py" # 将 QA 的测试脚本写入本地文件
)

# 4. 组装 Crew 并激活内存与缓存系统
dev_crew = Crew(
    agents=[product_manager, developer, qa_engineer],
    tasks=[prd_task, code_task, test_task],
    process=Process.sequential,
    memory=True, # 开启记忆系统
    verbose=True
)

# 5. 启动流水线
result = dev_crew.kickoff()

print("######################")
print("Collaboration Completed. Result saved to local files.")