beer_data/src/flask/brew_advice.py

"""Brewing advice from an LLM, generated for each recorded observation.

The advice is a *comment on the reading*, not a replacement for the
brewer's own comment.  Every reading gets its own advice text, stored in
its own field, so nothing the brewer wrote is ever overwritten.

Design notes
------------
- **The LLM is never in the request path.**  Recording a reading must not
  block on a model that may be slow or down, so the advice is generated in
  a background thread and written when it is ready.  The page shows the
  reading immediately and the advice appears on a later visit.
- **Failures are recorded, not raised.**  If the model is unreachable the
  advice field gets a short note saying so.  A silent failure would look
  identical to "no advice yet", which is confusing when you are waiting
  for a tip that is never going to arrive.
- **The model is configurable**, with a default that is known to work on
  this homelab.  Several models on the Ollama instance have been retired
  and fail at call time while still appearing in the model list, so the
  default is verified rather than assumed.
"""

from __future__ import annotations

import json
import logging
import os
import urllib.error
import urllib.request

log = logging.getLogger(__name__)

#: The Ollama instance on the homelab.  Reachable from the pod.
DEFAULT_OLLAMA_URL = os.getenv("OLLAMA_URL", "http://192.168.178.45:11434")

#: The model to use.  Override with the ``OLLAMA_MODEL`` environment
#: variable.
#:
#: A model that does NOT reason is much better here.  The reasoning
#: models emit their thinking out of the same token allowance as the
#: answer, and several of them burned the whole budget on thinking and
#: returned an EMPTY reply - which looks like a broken model rather than
#: a budget problem.  With `think: false` the model still shows its
#: working, but inline in the content, so the answer can be taken from
#: the end of it.
#:
#: `deepseek-v4-flash:cloud`, `deepseek-v3.2:cloud` and `qwen3.5:cloud`
#: are all retired and fail at call time while still appearing in the
#: model list, so the default is one that was probed.
DEFAULT_MODEL = os.getenv("OLLAMA_MODEL", "mistral-large-4:cloud")

#: How long to wait for the model before giving up.  The advice is
#: generated in the background, so this only bounds the background thread.
DEFAULT_TIMEOUT = float(os.getenv("LLM_TIMEOUT", "120"))

#: The advice is advice, not instruction.  Keep it short - it has to be
#: readable on the brewing page next to the brewer's own comment.
MAX_ADVICE_CHARS = 1200

SYSTEM_PROMPT = """\
You are an experienced home brewer giving practical advice to a single \
brewer about their own beer. You know brewing science: specific gravity, \
attenuation, yeast health, temperature, timing, and what a stalled or \
struck-out fermentation looks like.

You will be given one brew's readings so far: its name, the original \
gravity, each reading's date and specific gravity, and any comment the \
brewer wrote.

Give the brewer useful, specific, practical advice about THIS reading and \
this brew. Consider things like:

- Is the fermentation progressing at a normal rate for the style and \
temperature?
- Has it stalled, and if so what should they do about it?
- Is the yeast likely stressed, unhealthy, or done?
- Is the gravity drop plausible for the original gravity?
- Should they rack, add more hops, raise the temperature, or wait?

Rules:
- Advise about the reading that is marked as the one to advise on.  It
  may not be the most recent one - the brewer may be correcting an older
  reading - and the advice must be about the reading they asked about.
- If everything looks normal, say so briefly and say what to expect next.
- If something looks wrong, say what it is and what to do about it.
- Be specific to the numbers given. Do not give generic advice.
- If a reading looks like a typo or a mis-entry, say so and suggest \
checking it, but do not assume it - a slow or long fermentation can be \
genuine.
- Never say a reading is impossible. Beer cannot be denser than water, so \
a gravity below 1.000 is always a mis-entry or a typo, but the rest of the \
run may still be real and worth understanding.
- No preamble, no headings, no bullet-point templates, no "Great question!". \
Write two to five sentences of plain prose.
"""


def _reading_lines(readings: list[dict], subject_id: str | None = None) -> str:
    """The readings, as the compact list the model reads.

    The reading being advised about is marked, so the model does not have
    to work out which one from the dates - the brewer may be correcting
    an older reading.
    """
    lines = []
    for r in readings:
        parts = [f"- {r.get('date')}: gravity {r.get('sg_decimal')} ({r.get('point_scale')})"]
        if r.get("comment"):
            parts.append(f"brewer's comment: {r['comment']}")
        if r.get("final"):
            parts.append("(marked as the FINAL reading)")
        if subject_id is not None and str(r.get("_id")) == str(subject_id):
            parts.append("<<< THIS IS THE READING TO ADVISE ON")
        lines.append(" ".join(parts))
    return "\n".join(lines)


def build_prompt(run: dict, reading: dict, history: list[dict],
                 corrected: bool = False) -> str:
    """The user-message content for one observation.

    *corrected* says the reading has just been edited, so the advice
    should take that into account.
    """
    name = run.get("canonical_name") or run.get("name") or "this brew"
    og = run.get("og_decimal")
    if og is None and run.get("og_points"):
        og = f"{run['og_points'] / 1000:.3f}"

    parts = [
        f"Beer: {name}",
    ]
    if og:
        parts.append(f"Original gravity: {og}")
    if run.get("tap_number"):
        parts.append(f"Fermenter tap: {run['tap_number']}")
    if run.get("notes"):
        parts.append(f"Brewer's notes: {run['notes']}")

    parts.append("")
    parts.append("Readings so far (oldest first):")
    parts.append(_reading_lines(history, subject_id=reading.get("_id")))

    parts.append("")
    # The reading being advised about is NAMED, not just described as
    # "the most recent one".  The brewer may be editing an older reading,
    # and the advice is about THAT reading - so the prompt has to say
    # which one, or the model advises about the newest instead.
    parts.append(
        f"Give advice about the reading dated {reading.get('date') or 'unknown'}"
        f" (gravity {reading.get('sg_decimal')}),"
        + (" which the brewer has just corrected." if corrected else "")
    )
    if corrected:
        parts.append(
            "It is the reading they have just corrected, so it may differ "
            "from what it said before."
        )
    return "\n".join(parts)


def call_ollama(prompt: str, *, url: str | None = None, model: str | None = None,
                timeout: float | None = None) -> str:
    """Ask the model for advice.  Raises on failure."""
    base = (url or DEFAULT_OLLAMA_URL).rstrip("/")
    chosen_model = model or DEFAULT_MODEL
    wait = timeout if timeout is not None else DEFAULT_TIMEOUT

    payload = json.dumps({
        "model": chosen_model,
        "messages": [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt},
        ],
        "temperature": 0.4,
        # The reasoning models emit their thinking first, and it comes out
        # of the SAME budget as the answer.  At 500 tokens the reasoning
        # used the whole allowance and the reply came back empty - which
        # looks like a broken model rather than a budget problem.
        #
        # `think: false` does NOT help: the models still think, they just
        # put the working in the content instead of a separate field.
        # So the budget has to cover the reasoning AND the advice.
        "max_tokens": 8000,
        "stream": False,
    }).encode("utf-8")

    request = urllib.request.Request(
        f"{base}/v1/chat/completions",
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=wait) as response:
        body = json.loads(response.read().decode("utf-8"))

    choices = body.get("choices") or []
    if not choices:
        raise RuntimeError("the model returned no choices")
    message = choices[0].get("message") or {}
    content = message.get("content") or ""
    content = content.strip()
    if not content:
        raise RuntimeError("the model returned an empty response")
    return _strip_reasoning(content)[:MAX_ADVICE_CHARS]


#: The reasoning models sometimes put their working IN the content, even
#: with thinking turned off - "Let me look at this..." for several
#: thousand characters before the actual advice.  The advice is the last
#: few paragraphs, so the preamble is cut off.
_PREAMBLE_MARKERS = (
    "let me look",
    "let me think",
    "let me work",
    "okay, let me",
    "first, let me",
    "hmm,",
    "wait,",
    "let me analyze",
    "let me consider",
)


def _strip_reasoning(text: str) -> str:
    """Cut a reasoning preamble off the front of a reply.

    The advice is the prose at the END - the working comes first, in
    several paragraphs.  So the preamble is removed by finding the last
    paragraph break before the advice starts, not the first.

    Only a preamble that looks like working is removed; a reply that
    starts straight into the advice is left alone.
    """
    lowered = text.lower()
    for marker in _PREAMBLE_MARKERS:
        index = lowered.find(marker)
        # index 0 is fine - a reply can START with "Let me look at this".
        if index < 0:
            continue

        tail = text[index:]
        # The advice is the LAST paragraph of the tail.  Split on the
        # blank lines and take the paragraphs from the end, until the
        # running text stops looking like working.
        paragraphs = [p.strip() for p in tail.split("\n\n") if p.strip()]
        if len(paragraphs) < 2:
            continue

        # Walk back from the end while the paragraph still reads like the
        # model's working rather than the brewer's advice.
        kept = [paragraphs[-1]]
        for paragraph in reversed(paragraphs[:-1]):
            if _looks_like_advice(paragraph):
                kept.append(paragraph)
            else:
                break
        kept.reverse()

        candidate = "\n\n".join(kept).strip()
        if len(candidate) >= 80:
            return candidate
    return text


#: Words that mark a paragraph as the ADVICE rather than the model's
#: working.  The advice talks about what the brewer should DO.
_ADVICE_MARKERS = (
    "you should",
    "you could",
    "you may want",
    "you might",
    "give it",
    "let it",
    "take another",
    "take the",
    "raise",
    "lower",
    "rack",
    "keep",
    "suggest",
    "recommend",
    "in your case",
    "here is",
    "here's",
    "don't",
    "do not",
)


def _looks_like_advice(paragraph: str) -> bool:
    """True when a paragraph reads like advice to the brewer.

    The model's working is full of hedging - "perhaps", "or maybe", "it
    could be" - and is about interpreting the numbers.  The advice is
    about what to do.
    """
    lowered = paragraph.lower()
    if any(marker in lowered for marker in _ADVICE_MARKERS):
        return True
    # A paragraph that reads like working is not advice.  The signs are
    # hedging ("perhaps", "could be"), and the model talking to itself
    # ("wait", "hmm", "so", "actually") rather than to the brewer.
    working_words = ("perhaps", "maybe", "possibly", "could be", "might be",
                     "i think", "i am not sure", "actually", "hmm", "wait",
                     "let me", "so on brew day", "there is a discrepancy",
                     "that is a discrepancy", "but the", "which means")
    if any(word in lowered for word in working_words):
        return False
    return True


def generate_advice(run: dict, reading: dict, history: list[dict], *,
                     corrected: bool = False,
                     url: str | None = None, model: str | None = None,
                     timeout: float | None = None) -> tuple[bool, str]:
    """Generate the advice for one reading.

    Returns ``(ok, text)``.  On failure ``ok`` is False and ``text`` is a
    short note explaining what went wrong, so the reading carries a
    visible reason instead of silently having no advice.
    """
    prompt = build_prompt(run, reading, history, corrected=corrected)
    try:
        return True, call_ollama(prompt, url=url, model=model, timeout=timeout)
    except urllib.error.HTTPError as exc:
        # HTTPError is a subclass of URLError, so it has to be caught
        # FIRST or it would be swallowed by the unreachable branch.
        detail = ""
        try:
            detail = exc.read().decode("utf-8", "replace")[:300]
        except Exception:
            pass
        log.warning("the model returned HTTP %s: %s", exc.code, detail)
        return False, (f"Advice could not be generated - the model returned "
                       f"HTTP {exc.code}. {detail}".strip())
    except urllib.error.URLError as exc:
        log.warning("the model was unreachable: %s", exc)
        return False, "Advice could not be generated - the model was unreachable."
    except TimeoutError:
        log.warning("the model timed out")
        return False, "Advice could not be generated - the model timed out."
    except Exception as exc:  # noqa: BLE001 - the caller must never crash
        log.warning("advice generation failed: %s", exc)
        return False, f"Advice could not be generated - {exc}"


def generate_advice_in_background(run: dict, reading: dict,
                                   history: list[dict], writer,
                                   corrected: bool = False) -> None:
    """Generate the advice and hand it to *writer* when it is ready.

    *writer* is called as ``writer(reading_id, ok, text)``.  It runs in a
    background thread, so it must not touch the request context - the
    caller is responsible for giving it a way to reach Mongo.
    """
    ok, text = generate_advice(run, reading, history, corrected=corrected)
    try:
        writer(reading.get("_id"), ok, text)
    except Exception:  # noqa: BLE001 - never let the thread die noisily
        log.exception("could not store the generated advice")