ความเสถียรของ Agent และการกู้คืนเมื่อพัง

1from dataclasses import dataclass, field
2
3class StepBudgetExceeded(Exception):
4    pass
5
6@dataclass
7class StepBudget:
8    max_steps: int = 25
9    used: int = 0
10
11    def tick(self) -> None:
12        self.used += 1
13        if self.used > self.max_steps:
14            raise StepBudgetExceeded(
15                f"เกินเพดาน {self.max_steps} ก้าว - น่าจะติด loop"
16            )

1import hashlib
2import json
3from collections import Counter
4
5class LoopDetected(Exception):
6    pass
7
8class LoopGuard:
9    def __init__(self, max_repeats: int = 3):
10        self.max_repeats = max_repeats
11        self.seen: Counter[str] = Counter()
12
13    def _signature(self, node: str, payload: dict) -> str:
14        # ลายเซ็น = node + อาร์กิวเมนต์ที่ normalize แล้ว
15        body = json.dumps(payload, sort_keys=True, ensure_ascii=False)
16        return hashlib.sha1(f"{node}:{body}".encode()).hexdigest()
17
18    def visit(self, node: str, payload: dict) -> None:
19        sig = self._signature(node, payload)
20        self.seen[sig] += 1
21        if self.seen[sig] > self.max_repeats:
22            raise LoopDetected(
23                f"node '{node}' ทำงานซ้ำด้วย input เดิม {self.seen[sig]} ครั้ง"
24            )

1from langgraph.graph import StateGraph, END
2from langgraph.errors import GraphRecursionError
3from .loop_detector import LoopGuard, LoopDetected
4
5def build_agent():
6    graph = StateGraph(AgentState)
7    graph.add_node("plan", plan_node)
8    graph.add_node("act", act_node)
9    graph.add_edge("plan", "act")
10    graph.add_conditional_edges("act", route, {"continue": "plan", "done": END})
11    graph.set_entry_point("plan")
12    return graph.compile()
13
14agent = build_agent()
15
16def run(task: str):
17    guard = LoopGuard(max_repeats=3)
18    try:
19        return agent.invoke(
20            {"task": task, "guard": guard},
21            # เพดานแข็งของ LangGraph เอง
22            config={"recursion_limit": 25},
23        )
24    except (GraphRecursionError, LoopDetected) as exc:
25        # ไม่ปล่อยให้ crash - คืน partial result + ธง ให้ระบบจัดการต่อ
26        return {"status": "halted", "reason": str(exc)}

1import logging
2from dataclasses import dataclass
3from typing import Callable
4
5log = logging.getLogger("llm")
6
7class RetriableError(Exception):
8    """5xx, overload, timeout - ลองเจ้าถัดไปได้"""
9
10class FatalError(Exception):
11    """4xx จาก input เรา - fallback ไปก็พังเหมือนเดิม อย่าลอง"""
12
13@dataclass
14class Provider:
15    name: str
16    call: Callable[[str], str]
17
18def complete(prompt: str, providers: list[Provider]) -> str:
19    last: Exception | None = None
20    for provider in providers:                  # Claude ก่อน แล้ว Gemini
21        try:
22            return retry_with_backoff(lambda: provider.call(prompt))
23        except FatalError:
24            raise                                # input เราผิด ไม่ต้อง fallback
25        except RetriableError as exc:
26            log.warning("provider %s ล้มเหลว fallback ต่อ: %s", provider.name, exc)
27            last = exc
28    raise RuntimeError(f"ผู้ให้บริการล้มเหลวทั้งหมด: {last}")

1import random
2import time
3from typing import Callable, TypeVar
4
5T = TypeVar("T")
6
7def retry_with_backoff(
8    fn: Callable[[], T],
9    *,
10    max_attempts: int = 3,
11    base_delay: float = 1.0,
12    max_delay: float = 20.0,
13) -> T:
14    for attempt in range(max_attempts):
15        try:
16            return fn()
17        except RetriableError:
18            if attempt == max_attempts - 1:
19                raise
20            # exponential: base * 2^attempt, cap ที่ max_delay
21            backoff = min(base_delay * (2 ** attempt), max_delay)
22            # full jitter: สุ่มในช่วง [0, backoff] กัน thundering herd
23            time.sleep(random.uniform(0, backoff))
24    raise RuntimeError("unreachable")