ai
CrewAI Multi-Agent Automation Script
CrewAI Multi-Agent Automation Script คืออะไร
CrewAI เป็น Python framework สำหรับสร้าง multi-agent AI systems ที่ agents หลายตัวทำงานร่วมกันเพื่อทำ tasks ที่ซับซ้อน แต่ละ agent มี role, goal และ backstory เฉพาะ ใช้ LLMs เช่น GPT-4, Claude, Llama เป็น brain Automation Script คือการเขียน scripts ที่ทำงานอัตโนมัติ เมื่อรวมกับ CrewAI จะได้ระบบ AI agents ที่ทำงานซ้ำๆ ได้โดยไม่ต้องมนุษย์ดูแล เหมาะสำหรับ content creation, data analysis, research, code review และ customer support
CrewAI Fundamentals
# crewai_basics.py — CrewAI fundamentals
from crewai import Agent, Task, Crew, Process
# 1. Define Agents
researcher = Agent(
role="Senior Research Analyst",
goal="ค้นคว้าข้อมูลเชิงลึกเกี่ยวกับหัวข้อที่กำหนด",
backstory="""คุณเป็นนักวิจัยอาวุโสที่มีประสบการณ์ 10 ปี
เชี่ยวชาญในการค้นหาข้อมูลจากหลายแหล่ง
วิเคราะห์ trends และสรุปผลอย่างมีหลักการ""",
verbose=True,
allow_delegation=True,
)
writer = Agent(
role="Content Writer",
goal="เขียนบทความคุณภาพสูงจากข้อมูลวิจัย",
backstory="""คุณเป็นนักเขียนมืออาชีพที่เชี่ยวชาญ technical writing
สามารถอธิบายเรื่องซับซ้อนให้เข้าใจง่าย
เขียนได้ทั้งภาษาไทยและอังกฤษ""",
verbose=True,
)
editor = Agent(
role="Editor & QA",
goal="ตรวจสอบคุณภาพบทความ แก้ไขข้อผิดพลาด",
backstory="""คุณเป็น editor ที่เข้มงวดด้านคุณภาพ
ตรวจสอบ grammar, facts, structure และ SEO
ไม่ยอมให้ content คุณภาพต่ำผ่าน""",
verbose=True,
)
# 2. Define Tasks
research_task = Task(
description="""ค้นคว้าข้อมูลเกี่ยวกับ {topic} อย่างละเอียด:
- ข้อมูลพื้นฐานและ concepts สำคัญ
- Best practices และ real-world examples
- Tools และ technologies ที่เกี่ยวข้อง
- Statistics และ trends ล่าสุด""",
expected_output="รายงานวิจัยที่ครอบคลุม พร้อม sources",
agent=researcher,
)
writing_task = Task(
description="""เขียนบทความจากข้อมูลวิจัย:
- ความยาวอย่างน้อย 1500 คำ
- มี headings, subheadings ที่ชัดเจน
- มี code examples (Python)
- มี FAQ section""",
expected_output="บทความที่พร้อม publish",
agent=writer,
context=[research_task],
)
editing_task = Task(
description="""ตรวจสอบและแก้ไขบทความ:
- ตรวจ grammar และ spelling
- ตรวจความถูกต้องของข้อมูล
- ตรวจ SEO (keyword, meta description)
- ตรวจ readability""",
expected_output="บทความที่ผ่าน QA พร้อม feedback",
agent=editor,
context=[writing_task],
)
# 3. Create Crew
crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, writing_task, editing_task],
process=Process.sequential,
verbose=True,
)
# 4. Execute
# result = crew.kickoff(inputs={"topic": "Kubernetes Security Best Practices"})
# print(result)Automation Script Patterns
# automation_patterns.py — Common automation patterns
import json
class AutomationPatterns:
PATTERNS = {
"content_pipeline": {
"name": "Content Creation Pipeline",
"agents": ["Researcher", "Writer", "Editor", "SEO Specialist"],
"flow": "Research → Write → Edit → SEO Optimize → Publish",
"use_case": "Blog posts, documentation, reports",
},
"data_analysis": {
"name": "Data Analysis Pipeline",
"agents": ["Data Collector", "Analyst", "Visualizer", "Report Writer"],
"flow": "Collect Data → Analyze → Visualize → Generate Report",
"use_case": "Market research, competitive analysis, KPI reporting",
},
"code_review": {
"name": "Code Review Pipeline",
"agents": ["Code Reviewer", "Security Auditor", "Performance Analyst"],
"flow": "Review Code → Check Security → Analyze Performance → Summary",
"use_case": "PR reviews, security audits, code quality",
},
"customer_support": {
"name": "Customer Support Pipeline",
"agents": ["Classifier", "Knowledge Base Agent", "Response Writer", "QA Agent"],
"flow": "Classify Ticket → Search KB → Draft Response → QA Check",
"use_case": "Email support, chat support, ticket resolution",
},
"monitoring": {
"name": "System Monitoring Pipeline",
"agents": ["Log Analyzer", "Anomaly Detector", "Root Cause Analyst", "Remediation Agent"],
"flow": "Analyze Logs → Detect Anomaly → Find Root Cause → Suggest Fix",
"use_case": "Infrastructure monitoring, incident response",
},
}
def show_patterns(self):
print("=== Automation Patterns ===\n")
for key, pattern in self.PATTERNS.items():
print(f"[{pattern['name']}]")
print(f" Agents: {', '.join(pattern['agents'])}")
print(f" Flow: {pattern['flow']}")
print(f" Use case: {pattern['use_case']}")
print()
patterns = AutomationPatterns()
patterns.show_patterns()Advanced Crew Configuration
# advanced_crew.py — Advanced CrewAI configuration
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool
from langchain_openai import ChatOpenAI
# Custom LLM configuration
llm_gpt4 = ChatOpenAI(model="gpt-4o", temperature=0.7)
llm_fast = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
# Tools
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()
# Advanced Agent with tools and custom LLM
senior_researcher = Agent(
role="Senior Technical Researcher",
goal="ค้นคว้าข้อมูลเทคนิคเชิงลึก พร้อม code examples",
backstory="นักวิจัยเทคโนโลยีอาวุโส เชี่ยวชาญ cloud, DevOps, AI",
tools=[search_tool, web_tool],
llm=llm_gpt4,
max_iter=5,
max_rpm=10,
verbose=True,
)
# Agent with memory and delegation
project_manager = Agent(
role="Project Manager",
goal="ประสานงานระหว่าง agents ให้งานเสร็จตามเวลา",
backstory="PM ที่มีประสบการณ์จัดการ AI projects",
llm=llm_fast,
allow_delegation=True,
memory=True,
)
# Task with callbacks
def on_task_complete(output):
print(f"Task completed: {output.description[:50]}...")
# Save to file, send notification, etc.
research_task = Task(
description="ค้นคว้าเรื่อง {topic} ให้ครอบคลุม",
expected_output="รายงานวิจัย 500+ คำ พร้อม sources",
agent=senior_researcher,
callback=on_task_complete,
)
# Crew with advanced config
crew = Crew(
agents=[project_manager, senior_researcher],
tasks=[research_task],
process=Process.sequential,
memory=True,
cache=True,
max_rpm=20,
share_crew=False,
verbose=True,
)
print("Advanced crew configured")Batch Automation Script
# batch_automation.py — Batch processing with CrewAI
import json
class BatchAutomation:
CODE = """
# batch_crew.py — Process multiple tasks automatically
from crewai import Agent, Task, Crew, Process
import json
import time
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class BatchCrewRunner:
def __init__(self, config_path="crew_config.json"):
with open(config_path) as f:
self.config = json.load(f)
self.agents = self._create_agents()
self.results = []
def _create_agents(self):
agents = {}
for agent_config in self.config.get("agents", []):
agents[agent_config["id"]] = Agent(
role=agent_config["role"],
goal=agent_config["goal"],
backstory=agent_config["backstory"],
verbose=agent_config.get("verbose", True),
)
return agents
def process_batch(self, items, task_template):
'''Process a batch of items through the crew'''
for i, item in enumerate(items):
logger.info(f"Processing item {i+1}/{len(items)}: {item.get('title', 'unknown')}")
try:
# Create tasks for this item
tasks = []
for task_config in task_template:
task = Task(
description=task_config["description"].format(**item),
expected_output=task_config["expected_output"],
agent=self.agents[task_config["agent_id"]],
)
tasks.append(task)
# Create and run crew
crew = Crew(
agents=list(self.agents.values()),
tasks=tasks,
process=Process.sequential,
verbose=False,
)
result = crew.kickoff()
self.results.append({
"item": item,
"result": str(result),
"status": "success",
"timestamp": datetime.utcnow().isoformat(),
})
logger.info(f" Completed: {item.get('title')}")
except Exception as e:
logger.error(f" Failed: {e}")
self.results.append({
"item": item,
"error": str(e),
"status": "failed",
})
# Rate limiting
time.sleep(2)
return self.results
def save_results(self, output_path="batch_results.json"):
with open(output_path, 'w') as f:
json.dump(self.results, f, indent=2, ensure_ascii=False)
success = sum(1 for r in self.results if r["status"] == "success")
logger.info(f"Batch complete: {success}/{len(self.results)} succeeded")
# Usage
# runner = BatchCrewRunner("config.json")
# items = [
# {"title": "Kubernetes Security", "topic": "k8s security best practices"},
# {"title": "Docker Optimization", "topic": "docker image optimization"},
# ]
# results = runner.process_batch(items, task_template)
# runner.save_results()
"""
def show_code(self):
print("=== Batch Automation ===")
print(self.CODE[:600])
batch = BatchAutomation()
batch.show_code()Scheduling & Monitoring
# scheduling.py — Schedule and monitor CrewAI tasks
import json
import random
class SchedulingMonitoring:
SCHEDULER = """
# scheduler.py — Schedule CrewAI crews with APScheduler
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from crewai import Agent, Task, Crew, Process
import logging
logger = logging.getLogger(__name__)
scheduler = BackgroundScheduler()
def daily_content_crew():
'''Run content creation crew daily'''
researcher = Agent(role="Researcher", goal="Research trending topics", backstory="...")
writer = Agent(role="Writer", goal="Write blog post", backstory="...")
task = Task(
description="Research and write about today's trending tech topic",
expected_output="Blog post ready to publish",
agent=writer,
)
crew = Crew(agents=[researcher, writer], tasks=[task], process=Process.sequential)
result = crew.kickoff()
logger.info(f"Daily content created: {str(result)[:100]}")
def weekly_analysis_crew():
'''Run analysis crew weekly'''
analyst = Agent(role="Data Analyst", goal="Analyze weekly metrics", backstory="...")
task = Task(
description="Analyze this week's performance metrics and generate report",
expected_output="Weekly performance report",
agent=analyst,
)
crew = Crew(agents=[analyst], tasks=[task])
result = crew.kickoff()
logger.info(f"Weekly analysis done: {str(result)[:100]}")
# Schedule jobs
scheduler.add_job(daily_content_crew, CronTrigger(hour=6, minute=0)) # Daily 6 AM
scheduler.add_job(weekly_analysis_crew, CronTrigger(day_of_week='mon', hour=8)) # Monday 8 AM
scheduler.start()
"""
def show_scheduler(self):
print("=== Scheduler ===")
print(self.SCHEDULER[:500])
def dashboard(self):
print(f"\n=== CrewAI Dashboard ===")
print(f" Crews executed (24h): {random.randint(5, 20)}")
print(f" Tasks completed: {random.randint(15, 60)}")
print(f" Success rate: {random.uniform(90, 100):.1f}%")
print(f" Avg execution time: {random.uniform(30, 180):.0f}s")
print(f" Token usage (24h): {random.randint(50000, 200000):,}")
print(f" Estimated cost: ")
print(f" Active agents: {random.randint(3, 8)}")
sched = SchedulingMonitoring()
sched.show_scheduler()
sched.dashboard()FAQ - คำถามที่พบบ่อย
Q: CrewAI กับ LangChain Agents ต่างกันอย่างไร?
A: CrewAI: focus ที่ multi-agent collaboration — agents มี roles, goals, ทำงานร่วมกัน LangChain Agents: focus ที่ single agent + tools — ReAct pattern, function calling CrewAI ดีกว่า: tasks ที่ต้อง agents หลายตัว collaborate LangChain ดีกว่า: single agent + complex tool chains ใช้ร่วมกันได้: CrewAI agents ใช้ LangChain tools
เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง Multus CNI RBAC ABAC Policy
Q: ค่าใช้จ่าย API สำหรับ multi-agent เยอะไหม?
แนะนำเพิ่มเติม — สัญญาณเทรดรายวัน XM Signal
A: ค่อนข้างสูง — agents คุยกันหลาย rounds: GPT-4o: ~$5-15/1M tokens, CrewAI ใช้ ~5K-50K tokens/crew run ประหยัด: ใช้ GPT-4o-mini สำหรับ simple tasks ($0.15/1M tokens) ใช้ local LLM (Ollama + Llama 3) สำหรับ development/testing จำกัด max_iter, ใช้ cache, batch similar tasks
เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: LlamaIndex RAG Citizen Developer
Q: CrewAI ใช้กับ local LLM ได้ไหม?
A: ได้ — ผ่าน Ollama: llm = ChatOllama(model="llama3") รองรับ: Llama 3, Mistral, Mixtral, Phi-3 ข้อจำกัด: local models อาจ follow instructions ไม่ดีเท่า GPT-4 แนะนำ: develop/test กับ local → deploy กับ GPT-4o สำหรับ production
แนะนำเพิ่มเติม — เรียนเทรดกับ iCafeForex
เนื้อหาเกี่ยวข้อง — ทำความเข้าใจ Prometheus Alertmanager Data Pipeline ETL
Q: เริ่มเรียน CrewAI อย่างไร?
A: Step 1: pip install crewai crewai-tools Step 2: สร้าง crew ง่ายๆ 2 agents (researcher + writer) Step 3: เพิ่ม tools (search, web scraping) Step 4: ทดลอง batch processing Step 5: เพิ่ม scheduling + monitoring Resources: docs.crewai.com, YouTube tutorials, GitHub examples
เนื้อหาเกี่ยวข้อง — แนะนำให้อ่าน financial freedom — ข้อมูลครบถ้วน 2026
