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`
(없음)	`thinking` (emit하는 reasoning 모델만 사용)

에러 어디서 처리?

Adapter의 stream() 안에서. Connection refused → OllamaUnavailable (orchestrator가 잡아서 fallback trigger). Read timeout → ModelLoadingTimeout. 깨진 JSON 줄 → OllamaProtocolError. 셋 다 로그에 남길 만한 context를 달고 올라와.

Code

Full OllamaAdapter·python

import httpx, json
from typing import AsyncIterator, Any

class OllamaUnavailable(Exception): ...
class OllamaProtocolError(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
        self.client = httpx.AsyncClient(timeout=None)

    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

        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

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

                    if chunk.get("done"):
                        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"], raw=chunk)
                        continue

                    yield StreamChunk(content=msg.get("content", ""), raw=chunk)
        except httpx.ConnectError as e:
            raise OllamaUnavailable("Ollama daemon not reachable") from e

    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