OllamaAdapter 구현

이 adapter가 하는 일

Ollama의 NDJSON streaming을 universal StreamChunk로 번역해. Message content랑 tool call, 그리고 timing을 실은 final done frame까지 처리하고. Health check는 /api/version이 HTTP 200을 주는지로, model listing은 /api/tags로 구현해.

번역 map

Ollama 필드	StreamChunk 필드
`message.content`	`content`
`message.tool_calls`	`tool_calls` (그대로 통과, 이미 dict로 풀려 있어)
`done: true`	`done = True` + `tokens_per_sec`
`message.thinking`	`thinking` — reasoning 모델이 여길 채우고 그 frame의 `content`는 비워 두니까 둘 다 실어야 해
`{"error": …}` 줄	`OllamaStreamError`를 던져 — 200 헤더가 나간 뒤의 실패는 status code가 아니라 stream 줄로 와

에러 어디서 처리?

Adapter의 stream() 안에서. Connection refused → OllamaUnavailable (orchestrator가 잡아서 fallback trigger). Read timeout → ModelLoadingTimeout. 깨진 JSON 줄 → OllamaProtocolError. stream 중간의 {"error": …} 줄 → OllamaStreamError. done frame 없이 끝난 stream → 이것도 OllamaProtocolError. 전부 로그에 남길 만한 context를 달고 올라와.

뒤의 둘이 보기보다 중요해. Ollama는 응답 헤더 뒤로 나가는 걸 전부 NDJSON 줄로 보내. 자기 실패까지도. 그래서 죽은 runner는 HTTP 200 아래의 평범한 줄 하나야. 거기서 조용히 끝내는 adapter는 위에 있는 orchestrator 눈에 답을 다 끝낸 adapter랑 구분이 안 돼. 던져야 fallback이 일어날 수 있는 거지. 다만 던지는 걸로 되돌릴 수 없는 게 뭔지도 봐. 이미 yield한 chunk는 나가 버렸어. 그게 fallback 레슨이 이어받는 문제야.

EV v3 · 2026-09-18: 이 adapter에 구멍이 둘 있었어. 둘 다 발행일부터야. 번역 map이 thinking을 (없음)에서 온다고 적어 뒀는데, message.thinking은 Ollama API 타입에 선언돼 있는 필드고, qwen3.8로 think: true 스트림을 직접 돌려 보니 36 chunk 중 26개에 들어 있었어. 예전 코드는 그 26개를 빈 content chunk로 흘려보내고 reasoning을 통째로 버렸던 거야. 그리고 루프에 {"error": …} 줄 처리도, done frame 없이 끊긴 stream 처리도 없었어. 둘 다 Ollama가 실제로 만들어 내는 상황이야. server/routes.go가 200 헤더를 이미 쓴 뒤엔 에러를 NDJSON 줄로 인코딩하거든. mock 서버로 재현해 보니 예전 모양은 부분 답을 yield하고 그냥 정상 종료했어. orchestrator는 잘린 답에 성공 로그를 찍었고 fallback은 아예 안 돌았지. 이제 그 둘엔 OllamaStreamError랑 OllamaProtocolError를 던지고, thinking을 실어 나르고, model을 마지막 chunk만이 아니라 전부에 찍고, read timeout을 산문이 약속만 하고 구현은 안 해 뒀던 ModelLoadingTimeout에 연결했어. 마지막 건 리뷰에서 한 번 더 걸렸어. client를 timeout=None으로 만들고 있었는데 그러면 read timeout이 아예 꺼져서 새로 넣은 ReadTimeout 가지가 탈 수가 없었거든. 이제 httpx.Timeout을 명시해. 피파의 Ollama adapter가 쓰는 것과 같은 모양이야.

Code

Full OllamaAdapter·python

import httpx, json
from typing import AsyncIterator, Any

class OllamaUnavailable(Exception): ...
class OllamaProtocolError(Exception): ...
class OllamaStreamError(Exception): ...
class ModelLoadingTimeout(Exception): ...

class OllamaAdapter(AIAdapter):
    def __init__(self, base_url: str = "http://localhost:11434",
                 model: str = "qwen2.5:7b"):
        self.base_url = base_url.rstrip("/")
        self.model = model
        # timeout=None이면 read timeout이 통째로 꺼져서 아래 ReadTimeout
        # 가지가 영영 안 타. 생성이 느린 거지 무한한 게 아니니까,
        # read는 길게 connect는 짧게 줘.
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(connect=10.0, read=600.0, write=10.0, pool=10.0))

    async def stream(
        self,
        messages: list[dict],
        tools: list[dict] | None = None,
        **opts: Any,
    ) -> AsyncIterator[StreamChunk]:
        payload: dict = {"model": self.model, "messages": messages, "stream": True}
        if tools:
            payload["tools"] = tools
        if opts:
            payload["options"] = opts

        saw_done = False
        try:
            async with self.client.stream("POST", f"{self.base_url}/api/chat",
                                           json=payload) as r:
                r.raise_for_status()
                async for line in r.aiter_lines():
                    if not line:
                        continue
                    try:
                        chunk = json.loads(line)
                    except json.JSONDecodeError as e:
                        raise OllamaProtocolError(f"bad NDJSON: {e}") from e

                    # Ollama는 200 헤더가 나간 뒤에 생긴 실패를 stream 안의
                    # {"error": ...} 줄 하나로 알려 줘. 이 가지가 없으면
                    # 루프는 그걸 빈 content chunk로 읽고, 호출자는 답이
                    # 잘렸다는 걸 끝까지 모르게 돼.
                    if "error" in chunk:
                        raise OllamaStreamError(chunk["error"])

                    msg = chunk.get("message", {})

                    if chunk.get("done"):
                        saw_done = True
                        ec = chunk.get("eval_count", 0)
                        ed = chunk.get("eval_duration", 1) or 1
                        tps = ec / (ed / 1e9) if ed else 0.0
                        yield StreamChunk(done=True, tokens_per_sec=tps,
                                          model=self.model, raw=chunk)
                        return

                    if msg.get("tool_calls"):
                        yield StreamChunk(tool_calls=msg["tool_calls"],
                                          model=self.model, raw=chunk)
                        continue

                    # Reasoning 모델은 scratchpad를 message.thinking에 담고
                    # 그 frame의 message.content는 비워 둬. 둘 다 실어야지,
                    # 안 그러면 thinking 구간 전체가 빈 chunk로 도착해.
                    yield StreamChunk(content=msg.get("content", ""),
                                      thinking=msg.get("thinking", ""),
                                      model=self.model, raw=chunk)
        except httpx.ConnectError as e:
            raise OllamaUnavailable("Ollama daemon not reachable") from e
        except httpx.ReadTimeout as e:
            raise ModelLoadingTimeout(f"{self.model} did not respond in time") from e

        # done frame 없이 끝난 stream은 잘린 거야. runner가 죽었거나
        # 연결이 끊겼거나. 여기서 조용히 끝내면 orchestrator 눈에는
        # 완료된 답이랑 똑같이 보여.
        if not saw_done:
            raise OllamaProtocolError("stream ended without a done frame")

    async def health_check(self) -> bool:
        try:
            r = await self.client.get(f"{self.base_url}/api/version", timeout=2.0)
            return r.status_code == 200
        except Exception:
            return False

    async def list_models(self) -> list[str]:
        r = await self.client.get(f"{self.base_url}/api/tags")
        r.raise_for_status()
        return [m["name"] for m in r.json().get("models", [])]

이 adapter가 하는 일

번역 map

에러 어디서 처리?

Code

External links

Exercise

Progress

댓글 0