OllamaAdapter Implementation

What this adapter does

Translates Ollama's NDJSON streaming to the universal StreamChunk. Handles message content, tool calls, the final done frame with timing. Implements health check (any HTTP 200 from /api/version) and model listing (/api/tags).

Translation map

Ollama field	StreamChunk field
`message.content`	`content`
`message.tool_calls`	`tool_calls` (passed through; already parsed dicts)
`done: true`	`done = True` + `tokens_per_sec`
`message.thinking`	`thinking` — reasoning models fill this and leave `content` empty on those frames, so carry both
`{"error": …}` line	raises `OllamaStreamError` — a failure after the 200 headers arrives as a stream line, not a status code

Where to handle errors

Inside the adapter's stream(). Connection refused → OllamaUnavailable (caught by orchestrator and triggers fallback). Read timeout → ModelLoadingTimeout. Malformed JSON line → OllamaProtocolError. A {"error": …} line mid-stream → OllamaStreamError. A stream that ends without a done frame → OllamaProtocolError as well. Each one carries enough context to log meaningfully.

The last two matter more than they look. Ollama sends everything after the response headers as NDJSON lines, including its own failures, so a dead runner is an ordinary line under an HTTP 200 — and an adapter that finishes quietly on one is indistinguishable, to the orchestrator above it, from an adapter that finished the answer. Raising is what lets fallback happen at all. Note what raising cannot undo: chunks already yielded have left the building. That is the problem the fallback lesson picks up.

EV v3 · 2026-09-18: Two holes in this adapter, both there since publication. The translation map listed thinking as coming from (absent); message.thinking is a declared field in Ollama's API types, and a live think: true run on qwen3.8 put it in 26 of 36 chunks — the old code yielded those 26 as empty-content chunks and dropped the reasoning entirely. And the loop had no branch for a {"error": …} line or for a stream that stops without a done frame, both of which Ollama really produces: server/routes.go encodes an error as an NDJSON line once the 200 headers are already written. Reproduced against a mock server, the old shape yielded a partial answer and then returned normally, so the orchestrator logged success on a truncated reply and fallback never fired. The adapter now raises OllamaStreamError and OllamaProtocolError for those two cases, carries thinking, stamps model on every chunk rather than only the last, and maps a read timeout to the ModelLoadingTimeout the prose had been promising without implementing. That last one needed a second fix, caught in review: the client was constructed with timeout=None, which disables read timeouts outright, so the new ReadTimeout branch could never have fired. It now sets an explicit httpx.Timeout — the same shape cwkPippa's own Ollama adapter uses.

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 would disable read timeouts entirely, and the
        # ReadTimeout branch below could never fire. Generation is slow,
        # not infinite: give it a long read budget and a short connect one.
        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 reports failures that happen AFTER the 200 headers
                    # as a bare {"error": ...} line inside the stream. Without
                    # this branch the loop reads it as an empty content chunk
                    # and the caller never learns the answer was cut off.
                    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 models put their scratchpad in message.thinking
                    # and leave message.content empty for those frames. Carry
                    # both or the whole thinking phase arrives as blank chunks.
                    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

        # A stream that stops without a done frame was truncated — a killed
        # runner, a dropped connection. Ending quietly here would look to the
        # orchestrator exactly like a completed answer.
        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", [])]

What this adapter does

Translation map

Where to handle errors

Code

External links

Exercise

Progress

Comments 0