Local AI fallback ship하기

이 마지막 lesson 모양

여기서 quest를 하나로 묶어서 진짜 물건을 하나 띄워. 작은 backend service인데, cloud provider를 골라서 user prompt에 답하고, cloud에 못 닿을 땐 조용히 Ollama로 넘어가. 앞의 모든 track에서 배운 패턴이 production 모양으로 돌아간다는 걸 보이는, 가장 작은 증거야.

Production check-list

Adapter 패턴. Cloud랑 Ollama 둘 다 stream() 구현.
Health check. 둘 다 빠른 timeout 가진 health_check().
Orchestrator. Cloud를 먼저 부르고, 실패하거나 health가 나쁘면 Ollama로 넘어가.
Streaming. 어느 vessel이 답하든 frontend는 끊김 없는 SSE stream 하나만 봐.
Pin한 모델 tag. Ollama 쪽도 cloud 쪽도 버전을 못 박아둬.
Observability. 매 request마다 어느 vessel이 답했는지 남겨. 나중에 어떤 답이 유독 다르게 느껴질 때 이 로그가 없으면 아무것도 못 해.
Mini mode. Local 모델이 작으면 system prompt를 확 덜어내.
Warm-up. Startup task에서 keep_alive: -1로 local 모델을 미리 올려둬. 그래야 첫 fallback이 견딜 만한 속도로 나와.

이게 동작할 때 만든 거

privacy도, 비용도, 장애도 아는 서비스야. Cloud가 죽어도 user는 답을 계속 받아. Cloud가 rate-limit을 걸어도 계속 받고. Cloud가 비싸지면 bulk batch job이 local로 흘러가. Orchestrator는 단순하고 adapter는 좁아. 그래서 잘 안 부러져.

Code

End-to-end skeleton (FastAPI + orchestrator)·python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json, asyncio

app = FastAPI()

# Startup에서 orchestrator 빌드; local fallback pre-warm
ORCHESTRATOR = AIOrchestrator()

@app.on_event("startup")
async def warm_local():
    # Local fallback 모델 pre-load해서 첫 cold-fallback이 30초 wait 안 되게
    try:
        await ORCHESTRATOR.vessels["ollama"].client.post(
            "http://localhost:11434/api/generate",
            json={"model": "qwen2.5:7b", "keep_alive": "-1"},
            timeout=120.0,
        )
    except Exception:
        pass  # local 안 깔려있어도 ok

@app.post("/chat/stream")
async def chat_stream(req: dict):
    """User에게 답 stream, cloud→local fallback 가지고."""
    chosen_vessel = {"name": None}

    async def gen():
        async for chunk in ORCHESTRATOR.stream(req["messages"]):
            if chosen_vessel["name"] is None:
                chosen_vessel["name"] = chunk.model
            payload = {"content": chunk.content, "done": chunk.done}
            yield f"data: {json.dumps(payload)}\n\n"
        # 어느 vessel이 답했는지 로그 (ops 용)
        print(f"[/chat/stream] vessel={chosen_vessel['name']}")

    return StreamingResponse(gen(), media_type="text/event-stream")

# 실행: uvicorn main:app --host 0.0.0.0 --port 9000

Exercise

FastAPI 서비스 빌드. 실행. 4 시나리오 test: (1) cloud 닿음, (2) cloud firewall에서 차단, (3) cloud 닿는데 느림, (4) Ollama daemon 멈춤. 네 경우 다 사용자 눈에 보이는 동작이 매끄러운지 확인해. 각 실패 모드에서 '매끄럽다'가 정확히 뭔지 글로 적어봐. 그게 네 진짜 ops 계약이야.