בניית סוכני AI עמידים לתקלות עם גיבוי וחוסן

כאשר סוכן בינה מלאכותית מפסיק פתאום להגיב עקב תקלת OpenAI API או קריסת שרת הסקה מקומי, תהליכים עסקיים נעצרים. ראינו את זה פעמים רבות: במערכות ייצור עם SLA של 99.9%, השבתת LLM של 10 דקות יכולה לגרום לאובדן אלפי הזמנות. לאחרונה, GPT-4o של לקוח פינטק נפל למשך 15 דקות עקב תקלת ענן—הסוכנים החלופיים שלנו עברו ל-Claude 3.5 תוך 400 אלפיות שנייה, והתהליך מעולם לא נעצר. לכן אנו מתכננים חוסן LLM ופול-סטאק שיכול לספוג מכה: ספקים חלופיים, מפסקי זרם (circuit breakers), נקודות ביקורת (checkpoints), והסלמה אנושית. הכל מוכן לשימוש, עם אחריות ותיעוד.

איך להפוך סוכני בינה מלאכותית לחסיני תקלות?

המפתח הוא להבטיח שכאשר רכיב כלשהו נכשל (LLM API, כלי, מתזמן), הסוכן או עובר מיידית לגיבוי או מתדרדר בצורה חלקה מבלי לאבד התקדמות. אנו מיישמים ארבע רמות הגנה: יתירות LLM (ספקים חלופיים), מפסק זרם לכלים, שחזור נקודות ביקורת למשימות ארוכות, והסלמה אנושית לשגיאות בלתי ניתנות לשחזור. בפועל, גיבוי בעדיפות עם מפסק זרם אמין פי 2-3 יותר מאשר round-robin בעומס שיא. SLA של סוכן בינה מלאכותית מגיע ל-99.99% עם גישה זו.

למה גיבוי בעדיפות עם מפסק זרם אמין יותר?

התרחיש הראשון והנפוץ ביותר הוא שהמודל הראשי (GPT-4o) אינו זמין עקב מגבלות קצב או תקלה. אנו משתמשים בשרשרת גיבוי: OpenAI → Anthropic → מודל מקומי (LLaMA 3 8B דרך vLLM). המעבר מתרחש תוך פחות מ-500 אלפיות שנייה באמצעות מפסק זרם לכל ספק. בדיקות בריאות מאפשרות לנו לזהות תקלות מוקדם.

בואו נשווה אסטרטגיות גיבוי פופולריות:

אנו בוחרים בעדיפות עם מפסק זרם: היא מציעה את האיזון הטוב ביותר בין ביצועים וחוסן. סובלנות תקלות MLOps למערכות כאלה דורשת ניטור ושחזור אוטומטי.

יישום מעשי: קוד ותצורות

בואו נבחן רכיבי Python מרכזיים.

ספק LLM חלופי

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import anthropic
import openai

class ResilientLLMClient:
    def __init__(self):
        self.providers = [
            {"name": "openai", "client": openai.AsyncOpenAI(), "model": "gpt-4o"},
            {"name": "anthropic", "client": anthropic.AsyncAnthropic(), "model": "claude-3-5-sonnet-20241022"},
            {"name": "local_vllm", "client": openai.AsyncOpenAI(base_url="http://gpu1:8000/v1"), "model": "llama-3-8b"},
        ]
        self.circuit_breakers = {p["name"]: CircuitBreaker() for p in self.providers}

    async def generate(self, messages: list, **kwargs) -> str:
        last_error = None
        for provider in self.providers:
            cb = self.circuit_breakers[provider["name"]]
            if cb.is_open():
                continue  # провайдер отключён по circuit breaker
            try:
                result = await self._call_provider(provider, messages, **kwargs)
                cb.record_success()
                return result
            except Exception as e:
                cb.record_failure()
                last_error = e
                logger.warning(f"Provider {provider['name']} failed: {e}")
        raise RuntimeError(f"All LLM providers failed. Last error: {last_error}")

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=1, max=10),
        retry=retry_if_exception_type((openai.RateLimitError, openai.APIConnectionError))
    )
    async def _call_provider(self, provider: dict, messages: list, **kwargs) -> str:
        response = await provider["client"].chat.completions.create(
            model=provider["model"],
            messages=messages,
            **kwargs
        )
        return response.choices[0].message.content

תבנית מפסק זרם

מפסק הזרם שלנו עוקב אחרי תבנית מפסק הזרם הקלאסית.

from enum import Enum
import time

class CircuitState(Enum):
    CLOSED = "closed"  # нормальная работа
    OPEN = "open"  # провайдер отключён
    HALF_OPEN = "half_open"  # тестируем восстановление

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = CircuitState.CLOSED

    def is_open(self) -> bool:
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = CircuitState.HALF_OPEN
                return False
            return True
        return False

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            logger.error(f"Circuit breaker OPEN after {self.failure_count} failures")

    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            self.failure_count = 0
            logger.info("Circuit breaker CLOSED - provider recovered")

נקודת ביקורת ושחזור משימות

למשימות ארוכות (>1 דקה), אנו שומרים התקדמות ב-Redis:

@dataclass
class AgentCheckpoint:
    task_id: str
    step_index: int
    completed_steps: list[StepResult]
    state: dict  # произвольное состояние агента
    saved_at: datetime

class CheckpointManager:
    def __init__(self, storage: Redis):
        self.storage = storage

    async def save(self, checkpoint: AgentCheckpoint):
        key = f"checkpoint:{checkpoint.task_id}"
        await self.storage.setex(
            key,
            3600,  # 1 час TTL
            pickle.dumps(checkpoint)
        )

    async def load(self, task_id: str) -> AgentCheckpoint | None:
        key = f"checkpoint:{task_id}"
        data = await self.storage.get(key)
        return pickle.loads(data) if data else None

    async def resume_or_start(self, task: AgentTask) -> AgentCheckpoint:
        existing = await self.load(task.id)
        if existing:
            logger.info(f"Resuming task {task.id} from step {existing.step_index}")
            return existing
        return AgentCheckpoint(
            task_id=task.id,
            step_index=0,
            completed_steps=[],
            state={},
            saved_at=datetime.utcnow()
        )

הסלמה אנושית לשגיאות בלתי ניתנות לשחזור

class EscalationPolicy:
    MAX_RETRIES = 3
    MAX_FALLBACK_ATTEMPTS = 2

    async def handle_failure(
        self, task: AgentTask, error: Exception, retry_count: int
    ) -> EscalationDecision:
        if retry_count < self.MAX_RETRIES and self._is_retriable(error):
            return EscalationDecision(action="retry", delay_seconds=2 ** retry_count)

        if isinstance(error, ToolUnavailableError):
            return EscalationDecision(action="use_fallback_tool", tool=self._get_fallback_tool(error.tool))

        # Эскалация к человеку
        await self.notify_human(task, error, retry_count)
        return EscalationDecision(
            action="escalate",
            message=f"Task {task.id} requires human intervention after {retry_count} retries: {error}"
        )

שלבים ליישום סובלנות תקלות

אנו עוקבים אחרי תהליך מוכח:

1. ניתוח (1-2 ימים): ביקורת על הארכיטקטורה הנוכחית, זיהוי נקודות כשל בודדות, איסוף דרישות RTO/RPO. עבור יתירות מרובת סוכנים, יש להגדיר תרחישים קריטיים.
2. עיצוב (3-5 ימים): בחירת תבניות (ניסיון חוזר, מפסק זרם, נקודת ביקורת), אב-טיפוס של אינטגרציות ספקים.
3. יישום (2-4 שבועות): כתיבת קוד, הגדרת מפסקי זרם, פריסת אחסון נקודות ביקורת, הגדרת הסלמה.
4. בדיקות (שבוע): בדיקות כאוס—דימוי תקלות LLM API, קריסות כלים, עומסים. מדידת זמן השהיה p99 ושיעור שחזור מוצלח.
5. פריסה ותיעוד (2-3 ימים): פריסה הדרגתית (canary), ניטור כל השכבות, כתיבת runbooks.

אנו גם מבצעים בדיקות עומס עם תקלות מדומות בכל שכבה. זה חושף בעיות נסתרות כמו timeouts אסימטריים בין ספקים שונים.

השוואת שכבות סובלנות תקלות

לוחות זמנים ומה כלול

לוחות זמנים משוערים: 2 עד 6 שבועות תלוי במורכבות המערכת ומספר הספקים. התמחור נקבע לאחר ביקורת.

תוצרים:

• לקוח LLM חלופי פרוס עם מפסק זרם
• מערכת נקודות ביקורת על Redis (או אחסון אחר)
• מדיניות הסלמה אנושית משולבת עם מסנג'ר
• תרחיש בדיקות כאוס
• תיעוד (runbook, דיאגרמת ארכיטקטורה)
• הדרכת צוות (שעה)
• אחריות ל-3 חודשים על הרכיבים המיושמים

circuit_breaker: failure_threshold: 5 reset_timeout: 60 half_open_max_requests: 1 monitor_interval: 10