CrewAI การจัดการสถานะงานและโซลูชันการเก็บรักษาข้อมูลถาวร

ในการพัฒนาระบบ Multi-Agent ด้วย CrewAI นั้น ปัญหาที่พบบ่อยที่สุดคือการสูญเสียข้อมูลสถานะเมื่อระบบรีสตาร์ทหรือเกิดข้อผิดพลาด วันนี้ผมจะมาแบ่งปันประสบการณ์ตรงในการแก้ไขปัญหา AttributeError: 'NoneType' object has no attribute 'output' ที่เกิดขึ้นระหว่างการพัฒนาระบบ Pipeline ขนาดใหญ่ และวิธีการตั้งค่า Persistence Layer ที่เหมาะสม

ปัญหาจริงที่เจอใน Production

สถานการณ์เฉพาะ: ระบบ CrewAI ที่รันอยู่บน Kubernetes Pod ของลูกค้าสตาร์ทอัพแห่งหนึ่ง พบว่าหลังจาก Pod ถูก Restart เนื่องจาก Auto-scaling หรือ Node failure ตัว Agent ทั้งหมดจะเริ่มงานใหม่ตั้งแต่ต้น ทำให้ข้อมูลที่ประมวลผลไปแล้ว 40 นาทีหายไปทันที ส่งผลให้ต้อง Re-run ทั้ง Pipeline และสูญเสียค่าใช้จ่าย API อย่างไม่จำเป็น

สถาปัตยกรรมการจัดการสถานะใน CrewAI

CrewAI เองไม่ได้มี Built-in Persistence ในตัว ดังนั้นเราต้องตั้งค่าการเก็บรักษาข้อมูลเอง โดยมี 3 รูปแบบหลักที่ใช้กัน:

• In-Memory State - เก็บใน RAM ของ Process ปัจจุบัน
• Database Persistence - ใช้ PostgreSQL, MongoDB หรือ SQLite
• Checkpoint-based Recovery - บันทึกสถานะเป็นไฟล์หรือ Object Storage

การติดตั้ง CrewAI Persistence Library

pip install crewai[persistence] psycopg2-binary redis

# config.yaml
persistence:
  type: "postgres"
  connection_string: "postgresql://user:pass@host:5432/crewai_db"
  checkpoint_interval: 30  # วินาที
  retry_attempts: 3
  retry_delay: 5

redis_cache:
  enabled: true
  host: "10.112.2.4"
  port: 6379
  ttl: 3600

Implementation การเก็บ State ด้วย PostgreSQL

import json
import psycopg2
from datetime import datetime
from typing import Dict, Any, Optional
from crewai import Agent, Task, Crew

class CrewAIPersistence:
    def __init__(self, connection_string: str):
        self.conn = psycopg2.connect(connection_string)
        self._init_database()
    
    def _init_database(self):
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS crew_snapshots (
                    id SERIAL PRIMARY KEY,
                    crew_id VARCHAR(255) UNIQUE,
                    crew_name VARCHAR(255),
                    status VARCHAR(50),
                    agents_state JSONB,
                    tasks_state JSONB,
                    created_at TIMESTAMP DEFAULT NOW(),
                    updated_at TIMESTAMP DEFAULT NOW()
                )
            """)
            self.conn.commit()
    
    def save_checkpoint(self, crew: Crew, crew_id: str):
        agents_state = {}
        for agent in crew.agents:
            agents_state[agent.role] = {
                "status": getattr(agent, 'status', 'idle'),
                "memory": agent.memory if hasattr(agent, 'memory') else []
            }
        
        tasks_state = {}
        for task in crew.tasks:
            tasks_state[task.description] = {
                "status": task.status if hasattr(task, 'status') else 'pending',
                "output": str(task.output) if task.output else None
            }
        
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO crew_snapshots 
                (crew_id, crew_name, status, agents_state, tasks_state, updated_at)
                VALUES (%s, %s, %s, %s, %s, %s)
                ON CONFLICT (crew_id) DO UPDATE SET
                    status = EXCLUDED.status,
                    agents_state = EXCLUDED.agents_state,
                    tasks_state = EXCLUDED.tasks_state,
                    updated_at = EXCLUDED.updated_at
            """, (
                crew_id,
                crew.name,
                'running',
                json.dumps(agents_state),
                json.dumps(tasks_state),
                datetime.now()
            ))
            self.conn.commit()
    
    def load_checkpoint(self, crew_id: str) -> Optional[Dict[str, Any]]:
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT crew_name, status, agents_state, tasks_state
                FROM crew_snapshots WHERE crew_id = %s
            """, (crew_id,))
            result = cur.fetchone()
            
            if result:
                return {
                    "crew_name": result[0],
                    "status": result[1],
                    "agents_state": result[2],
                    "tasks_state": result[3]
                }
        return None

การใช้งานร่วมกับ CrewAI
persistence = CrewAIPersistence("postgresql://user:pass@localhost:5432/crewai")

crew = Crew(
    agents=[researcher, analyzer, writer],
    tasks=[task1, task2, task3],
    verbose=True
)

ตรวจสอบ checkpoint ก่อนเริ่มงาน
existing_state = persistence.load_checkpoint("production_pipeline_001")
if existing_state and existing_state["status"] == "running":
    print(f"พบงานที่ค้างอยู่ กำลังกู้คืนสถานะ...")
    # Restore logic here
else:
    crew.kickoff()
    persistence.save_checkpoint(crew, "production_pipeline_001")

การใช้งานร่วมกับ HolySheep AI

สำหรับการเรียกใช้ LLM API ใน CrewAI คุณสามารถใช้ HolySheep AI เป็น API Provider ที่มีความเร็วต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ OpenAI โดยตรง รองรับ DeepSeek V3.2, Claude Sonnet 4.5 และ Gemini 2.5 Flash

# crewai_config.py
import os
from crewai import LLM

ตั้งค่า HolySheep เป็น LLM Provider
os.environ["OPENAI_API_BASE"] = "https://api.holysheep.ai/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

llm = LLM(
    model="deepseek/deepseek-v3.2",
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)

สร้าง Agent พร้อม Custom LLM
researcher = Agent(
    role="Senior Research Analyst",
    goal="ค้นหาและวิเคราะห์ข้อมูลล่าสุดเกี่ยวกับ {topic}",
    backstory="ผู้เชี่ยวชาญด้านการวิจัยที่มีประสบการณ์ 10 ปี",
    verbose=True,
    llm=llm
)

หรือใช้ Ollama สำหรับ Local Model
local_llm = LLM(
    model="ollama/llama3.1",
    base_url="http://localhost:11434"
)

Checkpoint System สำหรับ Long-Running Tasks

import time
import threading
from functools import wraps

def checkpointable(func):
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        checkpoint_id = f"{func.__name__}_{int(time.time())}"
        
        # เริ่ม checkpoint monitor
        monitor_thread = threading.Thread(
            target=self._monitor_checkpoint,
            args=(checkpoint_id,),
            daemon=True
        )
        monitor_thread.start()
        
        try:
            result = func(self, *args, **kwargs)
            self.persistence.save_checkpoint(
                self.crew, 
                checkpoint_id
            )
            return result
        except Exception as e:
            print(f"เกิดข้อผิดพลาด: {e}")
            self._trigger_recovery(checkpoint_id)
            raise
    
    return wrapper

class CheckpointableCrew:
    def __init__(self, crew: Crew, persistence: CrewAIPersistence):
        self.crew = crew
        self.persistence = persistence
        self.checkpoint_interval = 60
    
    def _monitor_checkpoint(self, checkpoint_id: str):
        while True:
            time.sleep(self.checkpoint_interval)
            try:
                self.persistence.save_checkpoint(self.crew, checkpoint_id)
                print(f"Checkpoint saved: {checkpoint_id}")
            except Exception as e:
                print(f"Checkpoint failed: {e}")
    
    @checkpointable
    def run_task(self, task: Task):
        result = self.crew.execute_task(task)
        return result
    
    def _trigger_recovery(self, checkpoint_id: str):
        state = self.persistence.load_checkpoint(checkpoint_id)
        if state:
            print(f"กู้คืนจาก checkpoint: {checkpoint_id}")
            # Implement recovery logic
            for agent_name, agent_data in state["agents_state"].items():
                self._restore_agent_state(agent_name, agent_data)

การใช้งาน
checkpointable_crew = CheckpointableCrew(crew, persistence)
result = checkpointable_crew.run_task(research_task)

Redis Cache สำหรับ Session State

import redis
import json
from typing import Optional, Dict, Any

class RedisSessionManager:
    def __init__(self, host: str = "localhost", port: int = 6379, ttl: int = 3600):
        self.redis_client = redis.Redis(
            host=host,
            port=port,
            decode_responses=True
        )
        self.ttl = ttl
    
    def save_session(self, session_id: str, state: Dict[str, Any]):
        key = f"crewai:session:{session_id}"
        self.redis_client.setex(
            key,
            self.ttl,
            json.dumps(state)
        )
    
    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        key = f"crewai:session:{session_id}"
        data = self.redis_client.get(key)
        if data:
            return json.loads(data)
        return None
    
    def delete_session(self, session_id: str):
        key = f"crewai:session:{session_id}"
        self.redis_client.delete(key)
    
    def extend_ttl(self, session_id: str, additional_seconds: int):
        key = f"crewai:session:{session_id}"
        self.redis_client.expire(key, self.ttl + additional_seconds)

ใช้ร่วมกับ CrewAI Flow
session_manager = RedisSessionManager(host="10.112.2.4", port=6379)

ก่อนเริ่มงาน
current_session = session_manager.get_session("session_12345")
if current_session:
    print(f"กู้คืน session ที่มีอยู่: {len(current_session)} items")

หลังจากประมวลผลเสร็จ
session_manager.save_session("session_12345", {
    "last_task": "analysis",
    "completed_tasks": ["research", "analysis"],
    "pending_tasks": ["report_generation"],
    "context": {"query": "AI trends 2024"}
})

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

if task.status == 'completed' and task.output:
    result = task.output
else:
    print(f"Task {task.id} still processing")

from psycopg2.pool import ThreadedConnectionPool

pool = ThreadedConnectionPool(
    minconn=2, 
    maxconn=10,
    connection_string,
    connect_timeout=30
)

# ตรวจสอบ Redis
systemctl status redis
หรือ start Redis
redis-server --daemonize yes

def safe_load_json(data):
    try:
        return json.loads(data)
    except (json.JSONDecodeError, TypeError):
        return {"status": "unknown", "data": {}}

Best Practices สำหรับ Production

• Backup ก่อน Restart - ตั้งค่า Pre-stop Hook ใน Kubernetes เพื่อบันทึก State ก่อน Pod ปิด
• Idempotent Operations - ออกแบบ Tasks ให้รันซ้ำได้โดยไม่เกิดผลกระทบ
• State Validation - ตรวจสอบความถูกต้องของ State ก่อน Restore
• Cleanup Old Checkpoints - ลบ Checkpoints เก่าอัตโนมัติเพื่อประหยัด Storage
• Monitoring & Alerting - ตั้งค่า Alert เมื่อ Recovery จาก Checkpoint

Performance Benchmark

สรุป

การจัดการ State และ Persistence ใน CrewAI เป็นสิ่งจำเป็นสำหรับระบบ Production ที่ต้องการความเสถียรและสามารถกู้คืนจากความผิดพลาดได้ การใช้งานร่วมกับ HolySheep AI ช่วยลดค่าใช้จ่ายในการเรียก API ได้อย่างมีนัยสำคัญ พร้อมทั้งความเร็วที่ต่ำกว่า 50ms ทำให้ Pipeline ทำงานได้รวดเร็วและประหยัดค่าใช้จ่ายมากขึ้น