CrewAI Multi-Agent สำหรับมือใหม่ Step by Step

CrewAI Multi-Agent สำหรับมือใหม่ Step by Step

CrewAI เป็น Python framework สำหรับสร้าง multi-agent AI systems ที่ทำงานร่วมกันเป็นทีม แต่ละ Agent มีบทบาท (Role), เป้าหมาย (Goal) และเครื่องมือ (Tools) เฉพาะตัว Agents สามารถแบ่งงาน สื่อสาร และส่งต่อผลลัพธ์ระหว่างกันได้อัตโนมัติ CrewAI เหมาะสำหรับงาน research, content creation, data analysis และ automation ที่ต้องการหลาย AI agents ทำงานประสานกัน บทความนี้เป็นคู่มือ step-by-step สำหรับมือใหม่ตั้งแต่ติดตั้งจนถึงสร้าง crew แรก

ติดตั้งและเริ่มต้น

# setup.py — CrewAI installation and setup
import json

class CrewAISetup:
    INSTALLATION = """
# 1. ติดตั้ง CrewAI
pip install crewai crewai-tools

# 2. ตั้ง API Key (ใช้ OpenAI หรือ Ollama)
# .env file
OPENAI_API_KEY=sk-your-key-here

# หรือใช้ Ollama (local, ฟรี)
# OPENAI_API_BASE=http://localhost:11434/v1
# OPENAI_MODEL_NAME=llama3
# OPENAI_API_KEY=ollama

# 3. สร้าง project
crewai create crew my-first-crew
cd my-first-crew

# 4. Project structure
my-first-crew/
├── src/
│   └── my_first_crew/
│       ├── config/
│       │   ├── agents.yaml    # Agent definitions
│       │   └── tasks.yaml     # Task definitions
│       ├── crew.py            # Crew orchestration
│       ├── main.py            # Entry point
│       └── tools/
│           └── custom_tool.py # Custom tools
├── .env
├── pyproject.toml
└── README.md
"""

    CONCEPTS = {
        "agent": {
            "name": "Agent (ตัวแทน AI)",
            "description": "AI ที่มี role, goal, backstory เฉพาะ — เหมือนพนักงานในทีม",
            "example": "Researcher Agent, Writer Agent, Editor Agent",
        },
        "task": {
            "name": "Task (งาน)",
            "description": "งานที่มอบหมายให้ Agent — มี description, expected output",
            "example": "Research topic, Write article, Review content",
        },
        "crew": {
            "name": "Crew (ทีม)",
            "description": "กลุ่ม Agents + Tasks ที่ทำงานร่วมกัน",
            "example": "Content Creation Crew, Research Crew",
        },
        "tool": {
            "name": "Tool (เครื่องมือ)",
            "description": "เครื่องมือที่ Agent ใช้ได้ เช่น search, file read, API call",
            "example": "SerperDevTool, FileReadTool, WebsiteSearchTool",
        },
    }

    def show_installation(self):
        print("=== Installation ===")
        print(self.INSTALLATION[:500])

    def show_concepts(self):
        print(f"\n=== Core Concepts ===")
        for key, concept in self.CONCEPTS.items():
            print(f"  [{concept['name']}] {concept['description']}")

setup = CrewAISetup()
setup.show_installation()
setup.show_concepts()

สร้าง Agents

# agents.py — Creating CrewAI agents
from crewai import Agent
from crewai_tools import SerperDevTool, WebsiteSearchTool

# Tool สำหรับค้นหาข้อมูล
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Agent 1: Researcher — ค้นหาข้อมูล
researcher = Agent(
    role="Senior Research Analyst",
    goal="ค้นหาและวิเคราะห์ข้อมูลล่าสุดอย่างละเอียดและแม่นยำ",
    backstory="""คุณเป็นนักวิจัยอาวุโสที่มีประสบการณ์ 15 ปี
    ในการค้นหาและวิเคราะห์ข้อมูลเชิงลึก คุณมีทักษะในการ
    แยกแยะข้อมูลที่น่าเชื่อถือจากข้อมูลที่ไม่น่าเชื่อถือ""",
    tools=[search_tool, web_tool],
    verbose=True,
    allow_delegation=True,  # ส่งต่องานให้ agent อื่นได้
    max_iter=5,             # จำกัดรอบการทำงาน
)

# Agent 2: Writer — เขียนเนื้อหา
writer = Agent(
    role="Content Writer",
    goal="เขียนบทความที่น่าสนใจ อ่านง่าย และให้ข้อมูลครบถ้วน",
    backstory="""คุณเป็นนักเขียนมืออาชีพที่เชี่ยวชาญการเขียน
    บทความเทคนิคเป็นภาษาไทย สามารถอธิบายเรื่องยากให้
    เข้าใจง่ายได้ มีสไตล์การเขียนที่ชวนอ่าน""",
    verbose=True,
    allow_delegation=False,
)

# Agent 3: Editor — ตรวจสอบคุณภาพ
editor = Agent(
    role="Senior Editor",
    goal="ตรวจสอบความถูกต้อง คุณภาพ และ SEO ของเนื้อหา",
    backstory="""คุณเป็นบรรณาธิการที่มีประสบการณ์ด้าน
    content quality, SEO optimization และ fact-checking
    คุณมีมาตรฐานสูงในเรื่องความถูกต้องของข้อมูล""",
    verbose=True,
    allow_delegation=True,
)

print("Agents created successfully!")
print(f"  1. {researcher.role}")
print(f"  2. {writer.role}")
print(f"  3. {editor.role}")

สร้าง Tasks และ Crew

# crew.py — Creating tasks and crew
from crewai import Task, Crew, Process

# Task 1: Research
research_task = Task(
    description="""ค้นหาข้อมูลเกี่ยวกับ {topic} อย่างละเอียด
    ครอบคลุม:
    - ข้อมูลพื้นฐานและคำจำกัดความ
    - เทคโนโลยีและเครื่องมือที่เกี่ยวข้อง
    - Use cases และตัวอย่างการใช้งานจริง
    - ข้อดี ข้อเสีย และข้อควรระวัง
    - แนวโน้มและอนาคต""",
    expected_output="รายงานวิจัยฉบับสมบูรณ์ พร้อม sources อ้างอิง",
    agent=researcher,
)

# Task 2: Write Article
writing_task = Task(
    description="""เขียนบทความภาษาไทยเกี่ยวกับ {topic} 
    จากข้อมูลวิจัยที่ได้ โดย:
    - เนื้อหา 1,500-2,000 คำ
    - มี code examples (Python)
    - แบ่งเป็น sections ชัดเจน
    - มี FAQ 4-5 ข้อ
    - ภาษาเข้าใจง่าย เป็นกันเอง""",
    expected_output="บทความ HTML พร้อม headings, code blocks, และ FAQ",
    agent=writer,
    context=[research_task],  # ใช้ผลลัพธ์จาก research_task
)

# Task 3: Edit
editing_task = Task(
    description="""ตรวจสอบบทความ:
    - ความถูกต้องของข้อมูลเทคนิค
    - คุณภาพภาษาไทย
    - SEO optimization (keywords, headings)
    - Code examples ถูกต้อง
    - แก้ไขและส่ง final version""",
    expected_output="บทความฉบับสมบูรณ์ที่ผ่านการตรวจสอบแล้ว",
    agent=editor,
    context=[writing_task],
)

# สร้าง Crew
content_crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    process=Process.sequential,  # ทำทีละ task ตามลำดับ
    verbose=True,
)

# รัน Crew
result = content_crew.kickoff(
    inputs={"topic": "Kubernetes Cost Optimization"}
)

print("=== Final Output ===")
print(result)

Custom Tools

# custom_tools.py — Creating custom tools for CrewAI
from crewai_tools import tool
import requests
import json

# Custom Tool 1: API caller
@tool("API Fetcher")
def fetch_api(url: str) -> str:
    """Fetch data from a REST API endpoint. 
    Input should be a valid URL."""
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
        return json.dumps(data, indent=2)[:2000]
    except Exception as e:
        return f"Error fetching API: {str(e)}"

# Custom Tool 2: Database query
@tool("Database Query")
def query_database(sql: str) -> str:
    """Execute a read-only SQL query against the database.
    Input should be a valid SELECT SQL query."""
    import sqlite3
    try:
        conn = sqlite3.connect("data.db")
        cursor = conn.execute(sql)
        rows = cursor.fetchall()
        columns = [desc[0] for desc in cursor.description]
        result = [dict(zip(columns, row)) for row in rows]
        conn.close()
        return json.dumps(result, indent=2)[:2000]
    except Exception as e:
        return f"Error: {str(e)}"

# Custom Tool 3: File writer
@tool("File Writer")
def write_file(content: str) -> str:
    """Write content to a file. Input format: 'filename|||content'"""
    try:
        parts = content.split("|||", 1)
        filename = parts[0].strip()
        file_content = parts[1].strip()
        with open(filename, "w", encoding="utf-8") as f:
            f.write(file_content)
        return f"File written: {filename} ({len(file_content)} chars)"
    except Exception as e:
        return f"Error: {str(e)}"

# ใช้ tools กับ Agent
from crewai import Agent

data_analyst = Agent(
    role="Data Analyst",
    goal="วิเคราะห์ข้อมูลจาก API และ database",
    backstory="คุณเป็นนักวิเคราะห์ข้อมูลที่เชี่ยวชาญ SQL และ API",
    tools=[fetch_api, query_database, write_file],
    verbose=True,
)

print("Custom tools created!")
print(f"  1. {fetch_api.name}")
print(f"  2. {query_database.name}")
print(f"  3. {write_file.name}")

Advanced Patterns

# advanced.py — Advanced CrewAI patterns
from crewai import Crew, Process
import json

class AdvancedPatterns:
    PATTERNS = {
        "sequential": {
            "name": "Sequential Process",
            "description": "Tasks ทำทีละอัน ตามลำดับ — ผลลัพธ์ส่งต่อ task ถัดไป",
            "use_case": "Content pipeline: Research → Write → Edit",
            "code": "process=Process.sequential",
        },
        "hierarchical": {
            "name": "Hierarchical Process",
            "description": "มี Manager agent คอยมอบหมายงาน — เหมือนหัวหน้าทีม",
            "use_case": "Complex projects ที่ต้องการ coordination",
            "code": "process=Process.hierarchical, manager_llm=ChatOpenAI(model='gpt-4')",
        },
        "parallel_tasks": {
            "name": "Parallel Independent Tasks",
            "description": "แยก tasks ที่ไม่เกี่ยวข้อง → รันพร้อมกัน → รวมผล",
            "use_case": "Multi-source research, parallel data collection",
        },
        "memory": {
            "name": "Memory (Long-term)",
            "description": "Agents จำข้อมูลจากการรันก่อนหน้า",
            "code": "Crew(memory=True, embedder={'provider': 'openai'})",
        },
    }

    TIPS = [
        "เริ่มจาก 2-3 agents — อย่าสร้างเยอะเกินไป",
        "เขียน backstory ให้ละเอียด — ช่วยให้ agent เข้าใจบทบาท",
        "ใช้ context parameter เชื่อม tasks — ส่งต่อข้อมูลระหว่าง tasks",
        "ตั้ง max_iter จำกัดรอบ — ป้องกัน infinite loop",
        "ใช้ verbose=True ตอน debug — ปิดตอน production",
        "Ollama + local LLM ลดค่า API ได้มาก (ฟรี!)",
    ]

    def show_patterns(self):
        print("=== Advanced Patterns ===\n")
        for key, p in self.PATTERNS.items():
            print(f"[{p['name']}]")
            print(f"  {p['description']}")
            print(f"  Use: {p['use_case']}")
            print()

    def show_tips(self):
        print("=== Tips for Beginners ===")
        for tip in self.TIPS:
            print(f"  • {tip}")

adv = AdvancedPatterns()
adv.show_patterns()
adv.show_tips()

FAQ - คำถามที่พบบ่อย

Q: CrewAI กับ LangChain Agents ต่างกัน?

A: CrewAI: เน้น multi-agent collaboration — หลาย agents ทำงานเป็นทีม LangChain: เน้น single agent + tools chain — agent เดียวทำหลายขั้นตอน ใช้ CrewAI: งานที่ต้องการหลายมุมมอง (research + write + edit) ใช้ LangChain: งานที่ agent เดียวทำได้ (RAG, QA, tool calling) สามารถใช้ร่วมกันได้ — CrewAI agents ใช้ LangChain tools ได้

Q: ใช้ LLM อะไรดี?

A: GPT-4o: ดีที่สุด แต่แพง (~$5-15/1M tokens) GPT-3.5: ถูกกว่า ผลลัพธ์ OK สำหรับงานง่าย Claude 3: ดีสำหรับ long-form content Ollama (Llama 3, Mixtral): ฟรี! รันบนเครื่องตัวเอง แนะนำ: เริ่มจาก GPT-3.5 → upgrade เป็น GPT-4 สำหรับ production → ใช้ Ollama สำหรับ development/testing

Q: ค่าใช้จ่ายเท่าไหร่?

A: CrewAI framework: ฟรี (open source) LLM API: ขึ้นกับ provider — GPT-4o ~$5/1M input tokens, GPT-3.5 ~$0.50/1M Ollama (local): ฟรี ต้อง GPU (8GB+ VRAM) ค่าใช้จ่ายต่อ crew run: $0.01-0.50 (GPT-3.5) หรือ $0.10-5.00 (GPT-4) ลดค่าใช้จ่าย: ใช้ Ollama สำหรับ dev, จำกัด max_iter, cache results

Q: เหมาะกับงานอะไร?

A: Content Creation: research → write → edit → publish Data Analysis: collect → clean → analyze → report Customer Support: classify → route → respond → follow-up Code Review: analyze → security check → performance review → report ไม่เหมาะ: real-time processing, simple single-step tasks, ที่ต้องการ deterministic output