LangChain 멀티에이전트 무한루프 막는 종료조건·가드레일

from typing import Literal, Optional
from pydantic import BaseModel, Field

class SupervisorDecision(BaseModel):
    action: Literal["delegate", "finish", "clarify"]
    target_agent: Optional[str] = None
    reason: str = Field(..., min_length=1)
    final_answer: Optional[str] = None

# LLM 출력은 반드시 이 스키마로 파싱되도록 강제

import time

class Budget:
    def __init__(self, max_steps: int, timeout_seconds: int):
        self.max_steps = max_steps
        self.timeout_seconds = timeout_seconds
        self.started_at = time.time()
        self.steps = 0

    def tick(self):
        self.steps += 1
        if self.steps > self.max_steps:
            raise RuntimeError("budget_exceeded: max_steps")
        if time.time() - self.started_at > self.timeout_seconds:
            raise RuntimeError("budget_exceeded: timeout")

import random
import time

RETRYABLE = {"rate_limited", "timeout", "temporary_unavailable"}

def call_with_retry(fn, *, max_retries: int = 3):
    last_err = None
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except Exception as e:
            last_err = e
            code = getattr(e, "code", "unknown")
            if code not in RETRYABLE or attempt == max_retries:
                raise
            sleep = (2 ** attempt) + random.random()
            time.sleep(sleep)
    raise last_err

from enum import Enum

class Phase(str, Enum):
    PLAN = "plan"
    ACT = "act"
    VERIFY = "verify"
    FINISH = "finish"
    FAIL_SAFE = "fail_safe"

ALLOWED_TRANSITIONS = {
    Phase.PLAN: {Phase.ACT, Phase.FAIL_SAFE},
    Phase.ACT: {Phase.VERIFY, Phase.FAIL_SAFE},
    Phase.VERIFY: {Phase.ACT, Phase.FINISH, Phase.FAIL_SAFE},
    Phase.FINISH: set(),
    Phase.FAIL_SAFE: set(),
}

def transition(cur: Phase, nxt: Phase) -> Phase:
    if nxt not in ALLOWED_TRANSITIONS[cur]:
        raise RuntimeError(f"invalid_transition: {cur} to {nxt}")
    return nxt

import hashlib
import json

class LoopDetector:
    def __init__(self, max_same: int = 3):
        self.max_same = max_same
        self.last = None
        self.same_count = 0

    def check(self, signature_obj: dict):
        raw = json.dumps(signature_obj, sort_keys=True, ensure_ascii=False)
        sig = hashlib.sha256(raw.encode("utf-8")).hexdigest()

        if sig == self.last:
            self.same_count += 1
        else:
            self.last = sig
            self.same_count = 1

        if self.same_count >= self.max_same:
            raise RuntimeError("loop_detected: repeated_signature")

from pydantic import ValidationError

def parse_or_fail(model_cls, text: str, *, allow_retry: bool = True):
    try:
        return model_cls.model_validate_json(text)
    except ValidationError:
        if not allow_retry:
            raise
        # 재시도는 1회만 허용하는 식으로 제한
        raise RuntimeError("schema_parse_failed")

from dataclasses import dataclass
from typing import Optional

@dataclass
class RunState:
    phase: str
    clarify_count: int = 0
    last_result: Optional[str] = None

budget = Budget(max_steps=20, timeout_seconds=60)
loop_detector = LoopDetector(max_same=3)
state = RunState(phase="plan")

while True:
    budget.tick()

    # 1) Supervisor 결정 받기 (스키마 강제)
    decision = get_supervisor_decision(state)  # SupervisorDecision

    # 2) 반복 감지 시그니처
    loop_detector.check({
        "phase": state.phase,
        "action": decision.action,
        "target": decision.target_agent,
    })

    # 3) 상태 전이 및 실행
    if decision.action == "finish":
        state.phase = "finish"
        return decision.final_answer or state.last_result or ""

    if decision.action == "clarify":
        state.clarify_count += 1
        if state.clarify_count > 2:
            state.phase = "fail_safe"
            return "추가 정보 요청이 반복되어 중단합니다. 목표/제약을 구체화해 주세요."
        question = decision.reason
        return f"추가 확인이 필요합니다: {question}"

    if decision.action == "delegate":
        state.phase = "act"
        result = run_worker(decision.target_agent, state)
        state.last_result = result
        state.phase = "verify"

        # 4) 검증 단계에서 실패하면 제한적으로 재실행
        ok = verify_result(result)
        if ok:
            state.phase = "finish"
            return result
        else:
            # verify 실패가 반복되면 fail_safe로
            continue

class CriticVerdict(BaseModel):
    score: float = Field(..., ge=0.0, le=1.0)
    missing: list[str] = Field(default_factory=list)
    must_fix: bool = False