1

pip install crewai crewai-tools

1
2
3
4
5
6
7
8

from crewai import Agent

researcher = Agent(
    role="Security Researcher",
    goal="Find and analyze security vulnerabilities",
    backstory="Expert in penetration testing with 10 years experience",
    verbose=True
)

1
2
3
4
5
6
7

from crewai import Task

research_task = Task(
    description="Research the latest CVE vulnerabilities in web applications",
    expected_output="A detailed report of top 5 critical CVEs",
    agent=researcher
)

1
2
3
4
5
6
7
8
9

from crewai import Crew

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_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
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

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

# 工具
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Agent 定義
researcher = Agent(
    role="Vulnerability Researcher",
    goal="Find the latest critical security vulnerabilities",
    backstory="""You are a senior security researcher
    specializing in web application vulnerabilities.""",
    tools=[search_tool, web_tool],
    verbose=True
)

analyst = Agent(
    role="Security Analyst",
    goal="Analyze vulnerabilities and assess their impact",
    backstory="""You are a security analyst with expertise
    in risk assessment and threat modeling.""",
    verbose=True
)

writer = Agent(
    role="Technical Writer",
    goal="Create clear security advisories",
    backstory="""You are a technical writer who creates
    actionable security documentation.""",
    verbose=True
)

# Task 定義
research_task = Task(
    description="""Search for the top 5 critical CVEs
    announced in the past month. Focus on web applications.""",
    expected_output="List of CVEs with descriptions",
    agent=researcher
)

analysis_task = Task(
    description="""Analyze the CVEs from the research
    and assess their risk level and impact.""",
    expected_output="Risk assessment for each CVE",
    agent=analyst,
    context=[research_task]
)

report_task = Task(
    description="""Create a security advisory report
    based on the analysis.""",
    expected_output="Markdown formatted security report",
    agent=writer,
    context=[analysis_task]
)

# 建立 Crew
security_crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,
    verbose=True
)

# 執行
result = security_crew.kickoff()
print(result)

1
2
3
4
5

crew = Crew(
    agents=[agent1, agent2, agent3],
    tasks=[task1, task2, task3],
    process=Process.sequential  # 依序執行
)

1
2
3
4
5
6

crew = Crew(
    agents=[manager, worker1, worker2],
    tasks=[task1, task2, task3],
    process=Process.hierarchical,  # 管理者分配任務
    manager_llm=ChatOpenAI(model="gpt-4o")
)

1
2
3
4
5
6
7

from crewai_tools import (
    SerperDevTool,      # 搜尋
    WebsiteSearchTool,  # 網站搜尋
    FileReadTool,       # 讀取檔案
    DirectoryReadTool,  # 讀取目錄
    CodeInterpreterTool # 執行程式
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

from crewai_tools import BaseTool
from pydantic import BaseModel, Field

class ScanInput(BaseModel):
    target: str = Field(description="Target URL to scan")

class VulnerabilityScanner(BaseTool):
    name: str = "Vulnerability Scanner"
    description: str = "Scans a target for security vulnerabilities"
    args_schema: type[BaseModel] = ScanInput

    def _run(self, target: str) -> str:
        # 實作掃描邏輯
        return f"Scan results for {target}"

# 使用
scanner = VulnerabilityScanner()
agent = Agent(
    role="Security Scanner",
    tools=[scanner]
)

1
2
3
4
5

task2 = Task(
    description="Analyze the research results",
    agent=analyst,
    context=[task1]  # 依賴 task1 的結果
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from pydantic import BaseModel

class VulnerabilityReport(BaseModel):
    cve_id: str
    severity: str
    description: str
    mitigation: str

task = Task(
    description="Create vulnerability report",
    expected_output="Structured vulnerability report",
    output_pydantic=VulnerabilityReport,
    agent=writer
)

1
2
3
4
5
6
7
8

def on_task_complete(task_output):
    print(f"Task completed: {task_output}")

task = Task(
    description="Research vulnerabilities",
    callback=on_task_complete,
    agent=researcher
)

1
2
3
4
5
6
7
8
9

crew = Crew(
    agents=[...],
    tasks=[...],
    memory=True,  # 啟用記憶
    embedder={
        "provider": "openai",
        "config": {"model": "text-embedding-3-small"}
    }
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

agent = Agent(
    role="Researcher",
    llm=ChatOpenAI(model="gpt-4o")
)

# 或使用 Anthropic
agent = Agent(
    role="Analyst",
    llm=ChatAnthropic(model="claude-3-5-sonnet-20241022")
)

1
2
3
4
5
6

from langchain_community.llms import Ollama

agent = Agent(
    role="Researcher",
    llm=Ollama(model="llama3.3")
)

1
2
3
4
5

agent = Agent(
    role="Security Penetration Tester",  # 具體角色
    goal="Identify and exploit vulnerabilities",  # 明確目標
    backstory="10 years of experience in red team operations"  # 背景故事
)

1
2
3
4
5

# ✓ 好的任務定義
Task(description="Find SQL injection vulnerabilities in login form")

# ✗ 太模糊
Task(description="Test security")