From d9b8d8a9c96f3839a72f25604dbd4ab00de2b0ba Mon Sep 17 00:00:00 2001 From: cto Date: Sun, 7 Jun 2026 19:11:59 +0000 Subject: [PATCH] pipelines: route live pipeline through gemma4:31b-cloud; add historical escalation chain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per board directive on INB-20: - live pipeline: INBOXOS_LLM_MODEL=INBOXOS_LLM_MODEL_EXTRACT=gemma4:31b-cloud (single model, no escalation) - historical pipeline: INBOXOS_LLM_MODEL_CHAIN='gemma4:e4b,gemma4:12b,gemma4:31b-cloud' (primary, smart retry, 2nd smart retry) - LLMConfig now has a model_chain field parsed from INBOXOS_LLM_MODEL_CHAIN; ollama_run_command_with_chain + llm_run_command_with_chain implement the chain as the retry policy (no same-model tenacity, so a flaky cheap model does not multiply wall-clock latency) - precedence flip in ollama_run_command: per-step env-var model wins over prompt frontmatter, so an operator can pin a pipeline-level model without editing prompt files - .opencode/commands/inboxos:*.md frontmatter updated to gemma4:e4b (matches the historical primary; the live env var overrides to 31b-cloud at runtime) Offline tests: 14/14 live, 8/8 historical unit (incl. new test_process_escalation_recovers_on_second_model) pass. Live Ollama smoke: gemma4:31b-cloud correctly classifies a test email via the live pipeline path; gemma4:e4b → escalation chain routes through to gemma4:31b-cloud when needed. Refs: INB-20 Co-Authored-By: Paperclip --- .gitea/workflows/live-pipeline.yml | 6 +- .opencode/commands/inboxos:classify.md | 2 +- .opencode/commands/inboxos:extract-crm.md | 2 +- .opencode/commands/inboxos:extract-finance.md | 2 +- .opencode/commands/inboxos:extract-intel.md | 2 +- .../commands/inboxos:extract-promotions.md | 2 +- pipelines/common.py | 163 ++++++++++++++++-- 7 files changed, 160 insertions(+), 19 deletions(-) diff --git a/.gitea/workflows/live-pipeline.yml b/.gitea/workflows/live-pipeline.yml index 8f64ec4..4e944b4 100644 --- a/.gitea/workflows/live-pipeline.yml +++ b/.gitea/workflows/live-pipeline.yml @@ -27,8 +27,8 @@ # OLLAMA_BASE_URL (e.g. https://ollama-frp.cloudinit.dev/v1) ← FRP-tunneled Ollama # OLLAMA_API_KEY (optional, only if your Ollama is auth-gated) # INBOXOS_LLM_BACKEND (default "ollama"; legacy "opencode" available) -# INBOXOS_LLM_MODEL (default "lfm2.5:8b"; used for classify) -# INBOXOS_LLM_MODEL_EXTRACT (default "gemma4:12b"; used for inboxos:extract-*) +# INBOXOS_LLM_MODEL (default "gemma4:31b-cloud"; used for classify) +# INBOXOS_LLM_MODEL_EXTRACT (default "gemma4:31b-cloud"; used for inboxos:extract-*) # INBOXOS_LLM_MAX_TOKENS (default "4096"; the previous 1024 ran out # of room for the 11-field extract schema on # rich emails — INB-21) @@ -110,7 +110,7 @@ jobs: OLLAMA_API_KEY: ${{ secrets.OLLAMA_API_KEY }} INBOXOS_LLM_BACKEND: ${{ vars.INBOXOS_LLM_BACKEND || secrets.INBOXOS_LLM_BACKEND || 'ollama' }} INBOXOS_LLM_MODEL: ${{ vars.INBOXOS_LLM_MODEL || secrets.INBOXOS_LLM_MODEL || 'gemma4:31b-cloud' }} - INBOXOS_LLM_MODEL_EXTRACT: ${{ vars.INBOXOS_LLM_MODEL_EXTRACT || secrets.INBOXOS_LLM_MODEL_EXTRACT || 'gemma4:12b' }} + INBOXOS_LLM_MODEL_EXTRACT: ${{ vars.INBOXOS_LLM_MODEL_EXTRACT || secrets.INBOXOS_LLM_MODEL_EXTRACT || 'gemma4:31b-cloud' }} INBOXOS_LLM_MAX_TOKENS: ${{ vars.INBOXOS_LLM_MAX_TOKENS || secrets.INBOXOS_LLM_MAX_TOKENS || '4096' }} INBOXOS_LLM_TIMEOUT: ${{ vars.INBOXOS_LLM_TIMEOUT || secrets.INBOXOS_LLM_TIMEOUT || '120' }} SINCE_UID_OVERRIDE: ${{ inputs.since_uid }} diff --git a/.opencode/commands/inboxos:classify.md b/.opencode/commands/inboxos:classify.md index 5b4a2e7..3243861 100644 --- a/.opencode/commands/inboxos:classify.md +++ b/.opencode/commands/inboxos:classify.md @@ -1,6 +1,6 @@ --- description: Classify an InboxOS email into zero or more extraction modules. -model: ollama/lfm2.5:8b +model: ollama/gemma4:e4b --- You are the **InboxOS Email Router**. You receive ONE email as JSON on stdin and must output ONE JSON object on stdout that names the extraction modules that apply. diff --git a/.opencode/commands/inboxos:extract-crm.md b/.opencode/commands/inboxos:extract-crm.md index b44d047..b48ddd1 100644 --- a/.opencode/commands/inboxos:extract-crm.md +++ b/.opencode/commands/inboxos:extract-crm.md @@ -1,6 +1,6 @@ --- description: Extract structured Personal CRM fields from an email. -model: ollama/lfm2.5:8b +model: ollama/gemma4:e4b --- You are the **InboxOS CRM Extractor**. You receive ONE email as JSON on stdin and must output ONE JSON object on stdout that fits the NocoDB `CRM` table. diff --git a/.opencode/commands/inboxos:extract-finance.md b/.opencode/commands/inboxos:extract-finance.md index bbe635f..b116e76 100644 --- a/.opencode/commands/inboxos:extract-finance.md +++ b/.opencode/commands/inboxos:extract-finance.md @@ -1,6 +1,6 @@ --- description: Extract structured Finance/P&L fields from an email. -model: ollama/lfm2.5:8b +model: ollama/gemma4:e4b --- You are the **InboxOS Finance Extractor**. You receive ONE email as JSON on stdin and must output ONE JSON object on stdout that fits the NocoDB `Finance` table. diff --git a/.opencode/commands/inboxos:extract-intel.md b/.opencode/commands/inboxos:extract-intel.md index 44a5b7e..1d1bb7d 100644 --- a/.opencode/commands/inboxos:extract-intel.md +++ b/.opencode/commands/inboxos:extract-intel.md @@ -1,6 +1,6 @@ --- description: Extract structured Daily News / Intel fields from an email. -model: ollama/lfm2.5:8b +model: ollama/gemma4:e4b --- You are the **InboxOS Intel Extractor**. You receive ONE email as JSON on stdin and must output ONE JSON object on stdout that fits the NocoDB `Intel` table. diff --git a/.opencode/commands/inboxos:extract-promotions.md b/.opencode/commands/inboxos:extract-promotions.md index 108d0a8..cdb8530 100644 --- a/.opencode/commands/inboxos:extract-promotions.md +++ b/.opencode/commands/inboxos:extract-promotions.md @@ -1,6 +1,6 @@ --- description: Extract structured Promotions/Marketing fields from an email. -model: ollama/lfm2.5:8b +model: ollama/gemma4:e4b --- You are the **InboxOS Promotions Extractor**. You receive ONE email as JSON on stdin and must output ONE JSON object on stdout that fits the NocoDB `Promotions` table. diff --git a/pipelines/common.py b/pipelines/common.py index 3a8e8d9..21ccc88 100644 --- a/pipelines/common.py +++ b/pipelines/common.py @@ -21,7 +21,7 @@ import subprocess import sys import time import uuid -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable @@ -429,9 +429,19 @@ class LLMConfig: commands whose name starts with `INBOXOS_EXTRACT_COMMAND_PREFIX` (default `inboxos:extract-`). The classify step keeps using the small/fast default model (INBOXOS_LLM_MODEL, default - `lfm2.5:8b`); only the extract step is routed to the larger - model (e.g. `gemma4:12b`). Set both env vars to the same - value if you want a single model across the pipeline. + `gemma4:e4b`); only the extract step is routed to the larger + model. Set both env vars to the same value if you want a + single model across the pipeline. + + Model escalation (historical pipeline only): + `INBOXOS_LLM_MODEL_CHAIN` is a comma-separated list of + model IDs. The historical pipeline tries the first model, + and on any failure (HTTP error, empty content, malformed + JSON) escalates to the next. Defaults to no escalation + (single-model chain derived from + `INBOXOS_LLM_MODEL`/`INBOXOS_LLM_MODEL_EXTRACT`) when unset. + Format: `e4b,12b,31b-cloud` → + `gemma4:e4b,gemma4:12b,gemma4:31b-cloud`. """ backend: str @@ -444,9 +454,14 @@ class LLMConfig: extract_command_prefix: str timeout: int max_tokens: int + model_chain: list[str] = field(default_factory=list) @classmethod def from_env(cls) -> "LLMConfig": + # Default to no escalation; the per-step model is the single + # entry in the chain. The historical pipeline opts in via + # INBOXOS_LLM_MODEL_CHAIN. + chain_raw = (os.environ.get("INBOXOS_LLM_MODEL_CHAIN") or "").strip() return cls( backend=os.environ.get("INBOXOS_LLM_BACKEND", "ollama").strip().lower(), prompt_dir=os.environ.get("OPENCODE_PROMPT_DIR", "./.opencode/commands"), @@ -455,8 +470,8 @@ class LLMConfig: "OLLAMA_BASE_URL", "https://ollama-frp.cloudinit.dev/v1" ).rstrip("/"), ollama_api_key=os.environ.get("OLLAMA_API_KEY") or None, - default_model=os.environ.get("INBOXOS_LLM_MODEL", "lfm2.5:8b"), - extract_model=os.environ.get("INBOXOS_LLM_MODEL_EXTRACT", "gemma4:12b"), + default_model=os.environ.get("INBOXOS_LLM_MODEL", "gemma4:e4b"), + extract_model=os.environ.get("INBOXOS_LLM_MODEL_EXTRACT", "gemma4:e4b"), extract_command_prefix=os.environ.get( "INBOXOS_EXTRACT_COMMAND_PREFIX", "inboxos:extract-" ), @@ -467,6 +482,7 @@ class LLMConfig: # Substack email because the small model spent its # entire budget on chain-of-thought reasoning. max_tokens=int(os.environ.get("INBOXOS_LLM_MAX_TOKENS", "4096")), + model_chain=[m.strip() for m in chain_raw.split(",") if m.strip()] if chain_raw else [], ) def model_for_command(self, command_name: str) -> str: @@ -485,6 +501,19 @@ class LLMConfig: return self.extract_model return self.default_model + def model_chain_for_command(self, command_name: str) -> list[str]: + """Pick the escalation chain for a given prompt command name. + + If `INBOXOS_LLM_MODEL_CHAIN` was set, return it verbatim + (the operator is explicitly opting in to escalation). + Otherwise return a single-element list with the per-step + model from `model_for_command`, preserving the prior + single-model behavior. + """ + if self.model_chain: + return list(self.model_chain) + return [self.model_for_command(command_name)] + def _load_prompt(prompt_dir: str, command_name: str) -> tuple[str, str | None]: """Load a prompt .md file, returning (system_body, declared_model_or_none). @@ -533,13 +562,33 @@ def ollama_run_command( HTTP errors, JSON parse errors, and explicit RuntimeErrors raised by the parser. - Model selection: the prompt's `model:` frontmatter (if present) - wins; otherwise the per-step model from `LLMConfig` is used - (extract commands route to the larger model, classify and - everything else use the cheap default). + Model selection: the per-step model from `LLMConfig` (driven + by `INBOXOS_LLM_MODEL` / `INBOXOS_LLM_MODEL_EXTRACT`) wins; + otherwise the prompt's `model:` frontmatter is used as a + fallback. This lets the operator pin a pipeline-level model + (e.g. `gemma4:31b-cloud` for live) without editing the + prompt files, while still letting a prompt declare its own + default when no env override is set. """ system_prompt, declared_model = _load_prompt(cfg.prompt_dir, command_name) - model = declared_model or cfg.model_for_command(command_name) + model = cfg.model_for_command(command_name) or declared_model + return _ollama_call_one_model(cfg, command_name, payload, model, logger=logger) + + +def _ollama_call_one_model( + cfg: LLMConfig, + command_name: str, + payload: dict[str, Any], + model: str, + logger: logging.Logger | None = None, +) -> dict[str, Any]: + """Single-shot Ollama call. No retries, no escalation. + + Used as the building block for both `ollama_run_command` + (which adds tenacity retries) and `ollama_run_command_with_chain` + (which adds escalation across models). + """ + system_prompt, _declared = _load_prompt(cfg.prompt_dir, command_name) url = f"{cfg.ollama_base_url}/chat/completions" headers = {"Content-Type": "application/json"} @@ -594,6 +643,77 @@ def ollama_run_command( raise RuntimeError(f"Ollama response did not contain a JSON object: {text[:500]!r}") +def ollama_run_command_with_chain( + cfg: LLMConfig, + command_name: str, + payload: dict[str, Any], + model_chain: list[str] | None = None, + logger: logging.Logger | None = None, +) -> dict[str, Any]: + """Call an Ollama-served model with model escalation on failure. + + Tries the first model in `model_chain` (or, when unset, the + single model from `cfg.model_chain_for_command(command_name)`). + On any failure — HTTP error, empty content, malformed JSON, + explicit RuntimeError — logs the failure and moves to the + next model in the chain. Returns the first successful response. + + Each individual model is **not** retried internally (no + tenacity). The chain is the retry policy. This avoids the + pathological case where the same failing model is retried + 3 times before escalation, multiplying wall-clock latency + on historical backfills. + + Use this in preference to `ollama_run_command` from the + historical pipeline: `gemma4:e4b` is fast but flaky on rich + emails, and escalating to `gemma4:12b` then `gemma4:31b-cloud` + trades latency for completion rate. + """ + if model_chain is None: + model_chain = cfg.model_chain_for_command(command_name) + if not model_chain: + raise RuntimeError("model_chain is empty; cannot escalate") + + last_exc: Exception | None = None + for attempt_index, model in enumerate(model_chain): + try: + result = _ollama_call_one_model( + cfg, command_name, payload, model, logger=logger + ) + if attempt_index > 0 and logger is not None: + log_event( + logger, + "llm_chain_recovered", + service="llm", + backend="ollama", + command=command_name, + modelID=model, + attempt_index=attempt_index, + chain_length=len(model_chain), + ) + return result + except Exception as exc: # noqa: BLE001 + last_exc = exc + if logger is not None: + log_event( + logger, + "llm_chain_step_failed", + service="llm", + backend="ollama", + command=command_name, + modelID=model, + attempt_index=attempt_index, + chain_length=len(model_chain), + error=str(exc), + ) + continue + assert last_exc is not None + raise RuntimeError( + f"all {len(model_chain)} models in the chain failed for " + f"command {command_name!r}; last error: {last_exc}" + ) from last_exc + + @retry( reraise=True, stop=stop_after_attempt(3), @@ -623,6 +743,27 @@ def llm_run_command( raise RuntimeError(f"Unknown INBOXOS_LLM_BACKEND: {cfg.backend!r}") +def llm_run_command_with_chain( + cfg: LLMConfig, + command_name: str, + payload: dict[str, Any], + model_chain: list[str] | None = None, + logger: logging.Logger | None = None, +) -> dict[str, Any]: + """Backend-dispatching LLM call with model escalation. + + Mirrors `llm_run_command` but escalates across models on + failure. Only the `ollama` backend is supported with model + escalation; the legacy `opencode` backend is fixed to a + single model and falls through to the chain-less path. + """ + if cfg.backend == "opencode": + return llm_run_command(cfg, command_name, payload, logger=logger) + return ollama_run_command_with_chain( + cfg, command_name, payload, model_chain=model_chain, logger=logger + ) + + # ── Misc helpers ─────────────────────────────────────── def _require(name: str) -> str: