Multi-Vessel Fallback

Why fallback is the point

The most useful local AI pattern is not local everywhere — it's local-first with cloud fallback. Local handles 80–95% of work for free, with privacy. Cloud handles the rest when local can't. Your orchestrator picks per-call, transparently.

The four-vessel pattern (Pippa)

cwkPippa runs four AI vessels behind one orchestrator: Claude (primary, frontier reasoning), Codex (alternative, ChatGPT Pro OAuth), Gemini (third, Cloud Code Assist), Ollama (local fallback). The orchestrator picks based on:

Explicit user choice ("use Codex for this turn").
Health check ("Claude is degraded right now → try Codex").
Cost / privacy ("this is a heartbeat job → local").
Capability ("vision needed → Gemini or local Gemma").

The orchestrator is dumb on purpose

It doesn't know about NDJSON or SSE or tool argument formats. It only knows: try vessel A; if A's health check is bad or A throws, try vessel B. The adapters absorb provider details so the orchestrator stays simple.

Code

Multi-vessel orchestrator with fallback·python

class AIOrchestrator:
    """Try vessels in priority order; fall back on failure."""

    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  # explicit override; None = use priority

    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}")

Per-task vessel selection·python

async def route_task(orchestrator: AIOrchestrator, task: dict, prompt: str):
    """Route a task to the right vessel based on its shape."""
    if task.get("private") or task.get("heartbeat"):
        orchestrator.active = "ollama"          # never leave the machine
    elif task.get("needs_vision"):
        orchestrator.active = "gemini"          # vision-strong cloud
    elif task.get("needs_thinking"):
        orchestrator.active = "claude"          # frontier reasoning
    else:
        orchestrator.active = None              # default priority

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

Why fallback is the point

The four-vessel pattern (Pippa)

The orchestrator is dumb on purpose

Code

External links

Exercise

Progress

Comments 0