가장 유용한 local AI 패턴은 어디서나 local 아니라 — local-first에 cloud fallback. Local이 80–95%의 작업을 공짜로, privacy를 지키면서 처리해. Cloud는 local이 못 하는 나머지를 맡고. 그 선택은 orchestrator가 호출마다 알아서 하고, 쓰는 사람은 눈치도 못 채.
Multi-vessel 패턴 (피파)
피파가 orchestrator 하나 뒤에 AI vessel 일곱을 돌려. Claude (primary, frontier reasoning), Codex (ChatGPT Pro OAuth), Gemini (Cloud Code Assist), 그리고 Grok, Kimi, GLM, 마지막으로 local fallback인 Ollama. 출발은 넷이었어. 나머지 셋이 2026년에 들어오는 동안 orchestrator 모양은 안 바뀌었고. 이 패턴을 쓰는 이유가 그 한 문장이야. 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가 첫 chunk가 나가기 전에 예외를 던지면 vessel B로 넘어간다. provider 디테일은 adapter가 다 삼켜주니까 orchestrator가 이렇게 단순할 수 있는 거야.
Fallback의 마감은 첫 토큰
저 기울인 구절이 계약 전부야. 이미 yield한 generator는 출력을 도로 가져올 수 없어. 그 chunk들은 벌써 그려졌거나, 소켓에 쓰였거나, 사람이 읽었거든. vessel A가 토큰 열 개 뱉고 죽었는데 orchestrator가 조용히 vessel B로 넘어가면, 호출자는 A의 토큰 열 개 뒤에 B의 완전한 답이 이어붙은 걸 받아. 그리고 stream 어디에도 이음매 표시가 없어. 그러니까 갈아타기는 아직 아무것도 안 나갔을 때만 허용이야. 그 뒤엔 orchestrator가 VesselFailedMidStream을 던지고, 처음부터 다른 vessel로 다시 할지는 호출자가 정해. 투명한 fallback이랑 부분 답은 같이 못 가져. 둘 중 뭘 주는 건지 정하고, 타입으로 말해.
Code
Fallback 가진 multi-vessel orchestrator·python
class AllVesselsFailed(RuntimeError): ...
class VesselFailedMidStream(RuntimeError):
"""첫 chunk가 호출자에게 나간 뒤에 vessel이 죽은 경우."""
class AIOrchestrator:
"""우선순위대로 vessel을 시도하고, 실패하면 다음으로 넘어가."""
def __init__(self, priority: list[str] | None = None):
self.vessels: dict[str, AIAdapter] = {
"ollama": OllamaAdapter(model="qwen2.5:32b"),
"claude": ClaudeAdapter(model="claude-opus-4-7"),
"codex": CodexAdapter(model="gpt-5-codex"),
"gemini": GeminiAdapter(model="gemini-3-pro"),
}
# 기본은 local-first. cloud를 먼저 쓰고 local을 장애 대비 그물로
# 남기려면 priority=["claude", ..., "ollama"]를 넘겨.
self.priority = priority or ["ollama", "claude", "codex", "gemini"]
self.active: str | None = None # 명시적 override, None이면 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
yielded = False
try:
if not await vessel.health_check():
continue
async for chunk in vessel.stream(messages, **kwargs):
yielded = True
yield chunk
return
except Exception as e:
# Fallback이 투명한 건 첫 chunk가 나가기 전까지야. 토큰이 한 번
# 호출자한테 닿은 뒤에 vessel을 갈아타면, 부분 답 뒤에 두 번째
# 완전한 답이 이어붙어 버려. 조용히 넘기지 말고 던져서,
# 처음부터 다시 할지는 호출자가 정하게 해.
if yielded:
raise VesselFailedMidStream(
f"{name} failed after streaming had started") from e
last_err = e
continue
raise AllVesselsFailed(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
Ollama + cloud adapter 하나로 (진짜 key 없으면 stub) 2-vessel orchestrator 구현. Ollama daemon 멈추고 orchestrator가 cloud로 자동 fallback 확인. Ollama 재시작하고 다음 호출에서 local로 우선순위 돌아오는지 확인. 그다음 어려운 쪽을 일부러 만들어봐. local vessel이 chunk 몇 개 나간 *뒤에* 실패하게 하고 (stream() 안에서 raise), 답 두 개가 붙어서 나오는 대신 VesselFailedMidStream이 뜨는지 확인해.
Progress
Progress is local-only — sign in to sync across devices.