Multi-vessel fallback

Fallback이 핵심인 이유

가장 유용한 local AI 패턴은 어디서나 local 아니라 — local-first에 cloud fallback. Local이 80–95% 작업을 공짜로, privacy 가지고 처리. Cloud가 local이 못 하는 나머지. Orchestrator가 호출별로 transparent하게 고름.

네 vessel 패턴 (피파)

피파가 orchestrator 하나 뒤에 네 AI vessel 돌려: Claude (primary, frontier reasoning), Codex (alternative, ChatGPT Pro OAuth), Gemini (third, Cloud Code Assist), Ollama (local fallback). Orchestrator가 다음 기준으로 선택:

명시적 user 선택 ("이 turn은 Codex 써").
Health check ("Claude 지금 degraded → Codex 시도").
비용 / privacy ("이건 heartbeat job → local").
Capability ("vision 필요 → Gemini나 local Gemma").

Orchestrator는 의도적으로 단순

NDJSON이나 SSE나 tool argument 형식 몰라. 아는 거: vessel A 시도; A의 health check 나쁘거나 A가 throw하면 vessel B 시도. Adapter가 provider 디테일 흡수해서 orchestrator를 단순하게 유지.

Code

Fallback 가진 multi-vessel orchestrator·python

class AIOrchestrator:
    """우선순위 순으로 vessel 시도; 실패하면 fallback."""

    def __init__(self):
        self.vessels: dict[str, AIAdapter] = {
            "ollama": OllamaAdapter(model="qwen2.5:32b"),
            "claude": ClaudeAdapter(model="claude-opus-4-7"),
            "openai": OpenAIAdapter(model="gpt-5"),
            "gemini": GeminiAdapter(model="gemini-3-pro"),
        }
        self.priority = ["ollama", "claude", "openai", "gemini"]
        self.active: str | None = None  # 명시적 override; None = 우선순위 사용

    async def stream(self, messages: list[dict], **kwargs):
        order = [self.active] if self.active else self.priority
        last_err: Exception | None = None
        for name in order:
            vessel = self.vessels.get(name)
            if vessel is None:
                continue
            try:
                if not await vessel.health_check():
                    continue
                async for chunk in vessel.stream(messages, **kwargs):
                    yield chunk
                return
            except Exception as e:
                last_err = e
                continue
        raise RuntimeError(f"all vessels failed; last error: {last_err}")

Task별 vessel selection·python

async def route_task(orchestrator: AIOrchestrator, task: dict, prompt: str):
    """Task 모양에 따라 옳은 vessel로 라우팅."""
    if task.get("private") or task.get("heartbeat"):
        orchestrator.active = "ollama"          # 머신 떠나면 안 됨
    elif task.get("needs_vision"):
        orchestrator.active = "gemini"          # vision-강한 cloud
    elif task.get("needs_thinking"):
        orchestrator.active = "claude"          # frontier reasoning
    else:
        orchestrator.active = None              # default 우선순위

    async for chunk in orchestrator.stream([{"role": "user", "content": prompt}]):
        if chunk.done:
            break
        yield chunk.content

Fallback이 핵심인 이유

네 vessel 패턴 (피파)

Orchestrator는 의도적으로 단순

Code

External links

Exercise

Progress

댓글 0