61f2124b1c
Replace the char-count estimate (EST_OUTPUT_CHARS) with ContractProgress:
points are earned only when real output units complete — first token (5),
each closed contract key / array item (5-90), root closed (95), stored (100).
Reasoning/thinking streams no longer fake progress: the first thinking
delta fires the first-token milestone ('the model is alive') and holds.
Surface the job's tone through ProcessingJobOut/JobInfo so the bar can
label itself 'Summarizing dry wit…'. Test rewritten to pin the new
honest-thinking semantics.
134 lines
5.1 KiB
Python
134 lines
5.1 KiB
Python
"""Summaries via a local Ollama server (`/api/chat`, JSON mode)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
|
|
import httpx
|
|
|
|
from shonar.services.ai import ProviderConfigError, ProviderTransientError, SummaryResult
|
|
from shonar.services.ai._llm import (
|
|
SYSTEM_PROMPT,
|
|
build_user_message,
|
|
make_progress_ticker,
|
|
parse_summary,
|
|
)
|
|
|
|
|
|
class OllamaProvider:
|
|
name = "ollama"
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
model: str = "",
|
|
timeout_s: float = 900.0,
|
|
http_client: httpx.AsyncClient | None = None,
|
|
) -> None:
|
|
if not base_url.strip():
|
|
raise ProviderConfigError("ollama needs SHONAR_LLM_BASE_URL.")
|
|
if not model.strip():
|
|
raise ProviderConfigError("ollama needs SHONAR_LLM_MODEL.")
|
|
self.base_url = base_url.rstrip("/")
|
|
self.model = model
|
|
self.timeout_s = timeout_s
|
|
self.http_client = http_client
|
|
|
|
@asynccontextmanager
|
|
async def _client(self) -> AsyncIterator[httpx.AsyncClient]:
|
|
if self.http_client is not None:
|
|
yield self.http_client
|
|
else:
|
|
async with httpx.AsyncClient(timeout=self.timeout_s) as client:
|
|
yield client
|
|
|
|
async def summarize(self, transcript: str, *, title: str | None = None,
|
|
tone: str | None = None, on_progress=None) -> SummaryResult:
|
|
ticker = make_progress_ticker(on_progress)
|
|
payload = {
|
|
"model": self.model,
|
|
# Stream (NDJSON lines) so the UI shows real progress while the
|
|
# model generates; the reply is reassembled from the chunks.
|
|
"stream": True,
|
|
"format": "json",
|
|
# qwen3-family models "think" by default: a long reasoning
|
|
# chain before the JSON answer, brutally slow on CPU and it
|
|
# does not improve the summary. Ask for the answer directly
|
|
# (ignored by non-thinking models).
|
|
"think": False,
|
|
"options": {"num_ctx": 8192},
|
|
"messages": [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user",
|
|
"content": build_user_message(transcript, title, tone)},
|
|
],
|
|
}
|
|
try:
|
|
async with self._client() as client, client.stream(
|
|
"POST", f"{self.base_url}/api/chat", json=payload
|
|
) as resp:
|
|
if resp.status_code == 404:
|
|
# Missing model and missing route both 404 here; both
|
|
# are configuration, not weather.
|
|
raise ProviderConfigError(
|
|
"Ollama has no such model or route (HTTP 404).")
|
|
if resp.status_code != 200:
|
|
raise ProviderTransientError(
|
|
f"Summarization failed (HTTP {resp.status_code}).")
|
|
content = await _collect_reply(resp, ticker)
|
|
except (httpx.TimeoutException, httpx.TransportError) as e:
|
|
raise ProviderTransientError(f"Ollama unreachable: {type(e).__name__}") from e
|
|
import json as _json
|
|
|
|
try:
|
|
data = _json.loads(content)
|
|
except ValueError as e:
|
|
raise ProviderTransientError("Ollama reply was not JSON.") from e
|
|
return parse_summary(data, self.model)
|
|
|
|
|
|
async def _collect_reply(resp: httpx.Response, ticker) -> str:
|
|
"""Reassemble the assistant reply from Ollama's streamed chat response.
|
|
|
|
Streaming answers are NDJSON (one JSON object per line, ``done`` on the
|
|
last); a server honoring stream=false returns one JSON body — both work.
|
|
The accumulated content *prefix* feeds [ticker], which derives progress
|
|
from completed JSON contract units only (see ContractProgress)."""
|
|
import json as _json
|
|
|
|
ctype = resp.headers.get("content-type", "")
|
|
if "ndjson" not in ctype and "event-stream" not in ctype:
|
|
body = await resp.aread()
|
|
try:
|
|
content = _json.loads(body)["message"]["content"]
|
|
except (ValueError, KeyError, TypeError) as e:
|
|
raise ProviderTransientError("Ollama sent an unreadable reply.") from e
|
|
ticker(content or "")
|
|
return content or ""
|
|
|
|
parts: list[str] = []
|
|
async for line in resp.aiter_lines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
obj = _json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
piece = (obj.get("message") or {}).get("content") or ""
|
|
# Like the openai_compat adapter: a thinking channel means the
|
|
# model started (the ticker's first-token milestone) but is not
|
|
# contract output, so it moves nothing beyond that.
|
|
thinking = (obj.get("message") or {}).get("thinking") or ""
|
|
if piece:
|
|
parts.append(piece)
|
|
ticker("".join(parts))
|
|
elif thinking:
|
|
ticker("")
|
|
if obj.get("done"):
|
|
break
|
|
content = "".join(parts)
|
|
if not content:
|
|
raise ProviderTransientError("Ollama sent an unreadable reply.")
|
|
return content
|