From 6b523820f6a16dfc0905a5a80c7064174e26a28a Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Tue, 13 Oct 2026 10:25:42 +0200 Subject: [PATCH] Enhance advice generation logic with validation and retry mechanism - Updated `advice.py` to implement a validation check for advice responses, ensuring compliance with defined rules before storing. - Introduced a retry mechanism that provides feedback on rejected responses, allowing for corrective actions in subsequent attempts. - Modified the `parse_advice` function to handle validation errors more effectively, raising appropriate exceptions when violations occur. - Enhanced documentation in `README.md` to reflect the new validation rules and their implications for advice generation. --- README.md | 11 + pipeline/stages/advice.py | 73 ++++-- pipeline/stages/advice_guardrails.py | 318 ++++++++++++++++++++++++ tests/test_advice_guardrails.py | 356 +++++++++++++++++++++++++++ 4 files changed, 739 insertions(+), 19 deletions(-) create mode 100644 pipeline/stages/advice_guardrails.py create mode 100644 tests/test_advice_guardrails.py diff --git a/README.md b/README.md index 8f54725..439c88f 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,17 @@ is ending), or that a new treatment cannot be applied yet because the minimum interval since the last treatment has not elapsed; it does not say that the product list was empty. +Before a reply is stored, the pipeline checks that those rules hold: an empty +`allowed_products` list produces no prescription; every recommended product is +in `allowed_products` or `last_advice.suggested_products`; the dose sits inside +a registered `dose_min`–`dose_max` or concentration range for that product and +crop-disease pair (a product whose every use row has both dose bounds null is +accepted as written); the structured fields agree; and `full_advice` / +`advice_summary` do not contain model codes such as `FASE2` or +`INCUBAZPRIMARIA`. A failing reply is sent back to the model with the +violations, up to `retry.attempts`. If it still fails, that field is `failed` +and no `advisories` row is written. + If a field's 11-day window has no `FASE2`, no `FASE5`, no uncovered incubation, and no true `observation` day (`as_of-5 .. as_of`), the product search is skipped and `allowed_products` is empty, but advice is still generated and stored; that diff --git a/pipeline/stages/advice.py b/pipeline/stages/advice.py index 12a1015..e5895dd 100644 --- a/pipeline/stages/advice.py +++ b/pipeline/stages/advice.py @@ -14,6 +14,7 @@ import pyodbc from pipeline.config import Settings from pipeline.db import execute, transaction from pipeline.errors import PipelineError +from pipeline.stages.advice_guardrails import AdviceRejected, validate_advice from pipeline.stages.advice_summary import sanitize_advice_summary from pipeline.stages.llm import call_llm @@ -82,24 +83,51 @@ def generate_advice( """ Send json_for_advice_generation to the advice LLM and parse the structured reply. + A reply that fails `parse_advice` or `validate_advice` is sent back with + those violations appended, up to `settings.retry.attempts`. The loop stays + outside `llm_call`: that wrapper retries transport failures with the same + prompt, and advice temperature is 0, so a guardrail retry must change the + prompt. The last failure is raised and the caller does not store the row. + `llm_call` defaults to the plain, unguarded `call_llm` (fine for the single-field `one` CLI path). The batch orchestrator passes `resources.call_llm` instead, which adds the shared Gemini semaphore, rate limiting, and retry policy around the same call signature. """ llm_call = llm_call or call_llm - user_content = ( + base_user = ( f"{user_prompt}\n\n{json.dumps(advice_json, ensure_ascii=False, indent=2, default=_json_default)}" ) + attempts = max(1, settings.retry.attempts) + feedback = "" - raw = llm_call( - settings, - settings.llm_advice_generation, - system_prompt, - user_content, - json_output=True, - ) - return parse_advice(raw, as_of) + for attempt in range(1, attempts + 1): + user_content = base_user if not feedback else f"{base_user}\n\n{feedback}" + raw = llm_call( + settings, + settings.llm_advice_generation, + system_prompt, + user_content, + json_output=True, + ) + try: + result = parse_advice(raw, as_of) + violations = validate_advice(result, advice_json) + if violations: + raise AdviceRejected(violations) + return result + except PipelineError as exc: + logger.warning( + "Advice attempt %d/%d rejected: %s", + attempt, + attempts, + exc, + ) + if attempt == attempts: + raise + feedback = _retry_feedback(raw, exc) + + raise PipelineError("Advice LLM retries were exhausted.") def parse_advice(raw: str, as_of: date) -> AdviceResult: @@ -130,14 +158,6 @@ def parse_advice(raw: str, as_of: date) -> AdviceResult: is_treatment_recommended = _to_bool(payload.get("is_treatment_recommended")) recommended_application_date = _to_date(payload.get("recommended_application_date")) - if is_treatment_recommended and not recommended_product_names: - logger.warning("is_treatment_recommended is set but recommended_product_names is empty.") - if not is_treatment_recommended and recommended_product_names: - logger.warning( - "is_treatment_recommended is not set but %d treatment(s) were returned.", - len(recommended_product_names), - ) - return AdviceResult( full_advice=full_advice, advice_summary=summary, @@ -221,6 +241,22 @@ def _json_column(values: list[str], column: str, max_chars: int) -> str | None: return encoded +def _retry_feedback(raw: str, exc: PipelineError) -> str: + """Tell the next attempt exactly what to fix, plus the rejected reply.""" + if isinstance(exc, AdviceRejected): + reasons = "\n".join(f"- {item}" for item in exc.violations) + else: + reasons = f"- {exc}" + return ( + "The previous answer was rejected:\n" + f"{reasons}\n\n" + "Previous response:\n" + f"{raw}\n\n" + "Respond again with a single JSON object that fixes every rejection. " + "Keep every rule in the output format." + ) + + def _json_default(obj: Any) -> Any: if isinstance(obj, date): return obj.isoformat() @@ -293,5 +329,4 @@ def _to_date(value: Any) -> date | None: return datetime.strptime(text, fmt).date() except ValueError: continue - logger.warning("Could not parse recommended_application_date '%s'; storing NULL.", text) - return None + raise PipelineError(f"Could not parse recommended_application_date '{text}'.") diff --git a/pipeline/stages/advice_guardrails.py b/pipeline/stages/advice_guardrails.py new file mode 100644 index 0000000..ce86254 --- /dev/null +++ b/pipeline/stages/advice_guardrails.py @@ -0,0 +1,318 @@ +"""Reject advice replies that break the product, dose, or wording rules. + +`validate_advice` inspects an already parsed reply against the payload that +was sent to the model. It returns every violation it finds. `generate_advice` +turns a non-empty list into `AdviceRejected` and retries with that text. +""" + +from __future__ import annotations + +import re +from datetime import date +from decimal import Decimal, InvalidOperation +from typing import Any, Protocol + +from pipeline.errors import PipelineError +from pipeline.jsonutils import normalize_name + +_MODEL_CODE = re.compile(r"\bFASE\d*\b|INCUBAZPRIMARIA") +_DOSE = re.compile( + r"^\s*(?P\d+(?:[.,]\d+)?)" + r"(?:\s*[-–—]\s*(?P\d+(?:[.,]\d+)?))?" + r"\s*(?P\S.*?)\s*$" +) + + +class AdviceRejected(PipelineError): + """The advice reply failed one or more guardrails.""" + + def __init__(self, violations: list[str]) -> None: + self.violations = list(violations) + detail = "\n".join(f"- {item}" for item in self.violations) + super().__init__(f"Advice rejected:\n{detail}") + + +class _AdviceView(Protocol): + full_advice: str + advice_summary: str + is_treatment_recommended: bool + recommended_product_names: list[str] + recommended_doses: list[str] + recommended_application_date: date | None + + +def validate_advice(result: _AdviceView, advice_json: dict[str, Any]) -> list[str]: + """Return every guardrail violation in the parsed advice. Empty means accept.""" + violations: list[str] = [] + violations.extend(_empty_allowlist(result, advice_json)) + violations.extend(_allowlist(result, advice_json)) + # With an empty allowlist the only real fault is the prescription itself; + # dose findings would just add noise to the retry feedback. + if advice_json.get("allowed_products"): + violations.extend(_doses(result, advice_json)) + violations.extend(_structured(result)) + violations.extend(_model_codes(result)) + return violations + + +def _empty_allowlist(result: _AdviceView, advice_json: dict[str, Any]) -> list[str]: + if advice_json.get("allowed_products"): + return [] + if ( + result.recommended_product_names + or result.recommended_doses + or result.is_treatment_recommended + or result.recommended_application_date is not None + ): + return ["allowed_products is empty, so no product may be recommended."] + return [] + + +def _allowlist(result: _AdviceView, advice_json: dict[str, Any]) -> list[str]: + allowed = advice_json.get("allowed_products") or [] + if not allowed: + return [] + suggested = (advice_json.get("last_advice") or {}).get("suggested_products") or [] + known = _names(allowed) | _names(suggested) + violations: list[str] = [] + for name in result.recommended_product_names: + if normalize_name(name) not in known: + violations.append( + f"recommended product '{name}' is not in allowed_products " + "or last_advice.suggested_products." + ) + return violations + + +def _doses(result: _AdviceView, advice_json: dict[str, Any]) -> list[str]: + names = result.recommended_product_names + doses = result.recommended_doses + if not names or len(names) != len(doses): + return [] + violations: list[str] = [] + for name, dose_text in zip(names, doses): + violations.extend(_one_dose(name, dose_text, _find_uses(advice_json, name))) + return violations + + +def _one_dose(name: str, dose_text: str, uses: list[Any] | None) -> list[str]: + if not uses: + return [f"{name}: no registered uses in the payload."] + # No registered hectare ceiling on any row for this crop-disease pair. + if all(_dose_bounds_absent(use) for use in uses): + return [] + + parsed = _parse_dose(dose_text) + if parsed is None: + return [f"{name}: unparseable dose '{dose_text}'."] + low, high, unit = parsed + + if _any_unit(uses, "dose_unit", unit): + return _against_bounds( + name, + dose_text, + low, + high, + uses, + unit, + unit_key="dose_unit", + min_key="dose_min", + max_key="dose_max", + kind="dose", + ) + if _any_unit(uses, "concentration_unit", unit): + return _against_bounds( + name, + dose_text, + low, + high, + uses, + unit, + unit_key="concentration_unit", + min_key="concentration_min", + max_key="concentration_max", + kind="concentration", + ) + return [f"{name}: unknown unit '{unit}'."] + + +def _against_bounds( + name: str, + dose_text: str, + low: Decimal, + high: Decimal, + uses: list[Any], + unit: str, + *, + unit_key: str, + min_key: str, + max_key: str, + kind: str, +) -> list[str]: + matched = False + for use in uses: + if not isinstance(use, dict) or not _unit_matches(use.get(unit_key), unit): + continue + lower = _decimal(use.get(min_key)) + upper = _decimal(use.get(max_key)) + if lower is None or upper is None: + continue + matched = True + if lower <= low and high <= upper: + return [] + if not matched: + return [f"{name}: unit '{unit}' has no complete {kind} bounds."] + return [ + f"{name}: prescribed {dose_text} is outside the registered {kind} bounds for unit '{unit}'." + ] + + +def _structured(result: _AdviceView) -> list[str]: + names = result.recommended_product_names + doses = result.recommended_doses + violations: list[str] = [] + if result.is_treatment_recommended: + if not names: + violations.append( + "is_treatment_recommended is set but recommended_product_names is empty." + ) + elif len(doses) != len(names): + violations.append( + "recommended_doses must have one entry for each recommended product." + ) + if result.recommended_application_date is None: + violations.append( + "is_treatment_recommended is set but recommended_application_date is missing." + ) + return violations + + if names: + violations.append( + "is_treatment_recommended is not set but recommended_product_names is not empty." + ) + if doses: + violations.append( + "is_treatment_recommended is not set but recommended_doses is not empty." + ) + if result.recommended_application_date is not None: + violations.append( + "is_treatment_recommended is not set but recommended_application_date is set." + ) + return violations + + +def _model_codes(result: _AdviceView) -> list[str]: + violations: list[str] = [] + for field, text in ( + ("full_advice", result.full_advice), + ("advice_summary", result.advice_summary), + ): + seen: set[str] = set() + for match in _MODEL_CODE.finditer(text or ""): + token = match.group(0) + if token in seen: + continue + seen.add(token) + violations.append(f"{field} contains model code '{token}'.") + return violations + + +def _names(entries: list[Any]) -> set[str]: + names: set[str] = set() + for entry in entries: + if isinstance(entry, dict) and len(entry) == 1: + raw = str(next(iter(entry))) + else: + raw = str(entry) + cleaned = raw.strip() + if cleaned: + names.add(normalize_name(cleaned)) + return names + + +def _find_uses(advice_json: dict[str, Any], product_name: str) -> list[Any] | None: + key = normalize_name(product_name) + for entry in _product_entries(advice_json): + uses = _uses_of(entry, key) + if uses is not None: + return uses + return None + + +def _product_entries(advice_json: dict[str, Any]) -> list[Any]: + entries: list[Any] = [] + entries.extend(advice_json.get("allowed_products") or []) + last_advice = advice_json.get("last_advice") or {} + if isinstance(last_advice, dict): + entries.extend(last_advice.get("suggested_products") or []) + for day in advice_json.get("meteorological_data") or []: + if isinstance(day, dict): + entries.extend(day.get("applied_treatment") or []) + last_applied = advice_json.get("last_applied_treatment") or {} + if isinstance(last_applied, dict): + entries.extend(last_applied.get("products") or []) + return entries + + +def _uses_of(entry: Any, name_key: str) -> list[Any] | None: + if not isinstance(entry, dict) or len(entry) != 1: + return None + raw_name, body = next(iter(entry.items())) + if normalize_name(str(raw_name)) != name_key: + return None + if not isinstance(body, dict): + return None + uses = body.get("uses") + if not isinstance(uses, list): + return None + return uses + + +def _dose_bounds_absent(use: Any) -> bool: + if not isinstance(use, dict): + return False + return use.get("dose_min") is None and use.get("dose_max") is None + + +def _parse_dose(text: str) -> tuple[Decimal, Decimal, str] | None: + match = _DOSE.match(text or "") + if match is None: + return None + low = _decimal(match.group("low")) + if low is None: + return None + high_text = match.group("high") + high = low if high_text is None else _decimal(high_text) + if high is None or high < low: + return None + unit = match.group("unit").strip() + if not unit: + return None + return low, high, unit + + +def _any_unit(uses: list[Any], key: str, unit: str) -> bool: + return any(isinstance(use, dict) and _unit_matches(use.get(key), unit) for use in uses) + + +def _unit_matches(row_unit: Any, prescribed: str) -> bool: + normalized = _normalize_unit(row_unit) + return bool(normalized) and normalized == _normalize_unit(prescribed) + + +def _normalize_unit(value: Any) -> str: + if value is None: + return "" + return "".join(str(value).split()).casefold() + + +def _decimal(value: Any) -> Decimal | None: + if value is None: + return None + text = str(value).strip().replace(",", ".") + if not text: + return None + try: + return Decimal(text) + except InvalidOperation: + return None diff --git a/tests/test_advice_guardrails.py b/tests/test_advice_guardrails.py new file mode 100644 index 0000000..8f79785 --- /dev/null +++ b/tests/test_advice_guardrails.py @@ -0,0 +1,356 @@ +"""Guardrails on the advice reply, and the retry that sends violations back.""" + +from __future__ import annotations + +import json +import unittest +from datetime import date +from types import SimpleNamespace + +from pipeline.errors import PipelineError +from pipeline.stages.advice import AdviceResult, generate_advice, parse_advice +from pipeline.stages.advice_guardrails import AdviceRejected, validate_advice + +_WHEN = date(2026, 4, 14) + +_MELODY_USES = [ + { + "dose_min": 0.9, + "dose_max": 0.9, + "dose_unit": "kg/ha", + "concentration_min": 180, + "concentration_max": 450, + "concentration_unit": "g/hl", + }, + { + "dose_min": 1.8, + "dose_max": 1.8, + "dose_unit": "kg/ha", + "concentration_min": 180, + "concentration_max": 900, + "concentration_unit": "g/hl", + }, +] + + +def _use(**overrides: object) -> dict[str, object]: + row: dict[str, object] = { + "dose_min": 0.9, + "dose_max": 0.9, + "dose_unit": "kg/ha", + "concentration_min": None, + "concentration_max": None, + "concentration_unit": None, + } + row.update(overrides) + return row + + +def _product(name: str, uses: list[dict[str, object]]) -> dict[str, dict[str, object]]: + return {name: {"uses": uses}} + + +def _result( + *, + full_advice: str = "La fase fenologica segnalata consente un trattamento.", + advice_summary: str = "Trattare il 14 aprile 2026 con MELODY FLEX.", + is_treatment_recommended: bool = True, + recommended_product_names: list[str] | None = None, + recommended_doses: list[str] | None = None, + recommended_application_date: date | None = _WHEN, +) -> AdviceResult: + return AdviceResult( + full_advice=full_advice, + advice_summary=advice_summary, + is_treatment_recommended=is_treatment_recommended, + recommended_product_names=( + ["MELODY FLEX"] if recommended_product_names is None else recommended_product_names + ), + recommended_doses=["0.90 kg/ha"] if recommended_doses is None else recommended_doses, + recommended_application_date=recommended_application_date, + ) + + +def _no_treatment(*, recommended_application_date: date | None = None) -> AdviceResult: + return _result( + full_advice="Non e necessario un trattamento.", + advice_summary="Non trattare: il modello non indica rischio.", + is_treatment_recommended=False, + recommended_product_names=[], + recommended_doses=[], + recommended_application_date=recommended_application_date, + ) + + +def _payload( + allowed: list[object], + *, + suggested: list[object] | None = None, + applied: list[object] | None = None, + last_applied: list[object] | None = None, +) -> dict[str, object]: + payload: dict[str, object] = { + "allowed_products": allowed, + "last_advice": {"suggested_products": suggested or []}, + } + if applied is not None: + payload["meteorological_data"] = [{"applied_treatment": applied}] + if last_applied is not None: + payload["last_applied_treatment"] = {"date": "09-04-2026", "products": last_applied} + return payload + + +def _reply(**overrides: object) -> str: + body: dict[str, object] = { + "full_advice": "La fase fenologica segnalata consente un trattamento.", + "advice_summary": "Trattare il 14 aprile 2026 con MELODY FLEX.", + "is_treatment_recommended": 1, + "recommended_product_names": ["MELODY FLEX"], + "recommended_doses": ["0.90 kg/ha"], + "recommended_application_date": "2026-04-14", + } + body.update(overrides) + return json.dumps(body, ensure_ascii=False) + + +class ValidateAdviceTests(unittest.TestCase): + def test_empty_allowlist_with_no_prescription_passes(self) -> None: + self.assertEqual(validate_advice(_no_treatment(), _payload([])), []) + + def test_empty_allowlist_rejects_a_last_advice_product(self) -> None: + payload = _payload([], suggested=[_product("MELODY FLEX", _MELODY_USES)]) + violations = validate_advice(_result(), payload) + self.assertEqual( + violations, + ["allowed_products is empty, so no product may be recommended."], + ) + + def test_missing_allowlist_key_is_treated_as_empty(self) -> None: + violations = validate_advice(_result(), {}) + self.assertEqual( + violations, + ["allowed_products is empty, so no product may be recommended."], + ) + + def test_allowed_product_and_dose_pass(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + self.assertEqual(validate_advice(_result(), payload), []) + + def test_unknown_product_is_rejected(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _result(recommended_product_names=["OTHER"], recommended_doses=["1.0 kg/ha"]) + violations = validate_advice(result, payload) + self.assertIn( + "recommended product 'OTHER' is not in allowed_products " + "or last_advice.suggested_products.", + violations, + ) + + def test_last_advice_product_absent_from_allowed_passes(self) -> None: + payload = _payload( + [_product("OTHER", [_use()])], + suggested=[_product("MELODY FLEX", _MELODY_USES)], + ) + self.assertEqual(validate_advice(_result(), payload), []) + + def test_name_match_ignores_case_and_whitespace(self) -> None: + payload = _payload([_product(" melody flex ", _MELODY_USES)]) + self.assertEqual(validate_advice(_result(), payload), []) + + def test_dose_inside_either_registered_row_passes(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + low = _result() + high = _result(recommended_doses=["1.80 kg/ha"]) + cased = _result(recommended_doses=["0.90 KG/HA"]) + self.assertEqual(validate_advice(low, payload), []) + self.assertEqual(validate_advice(high, payload), []) + self.assertEqual(validate_advice(cased, payload), []) + + def test_dose_above_every_row_fails(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _result(recommended_doses=["2.00 kg/ha"]) + violations = validate_advice(result, payload) + self.assertTrue(any("outside the registered dose bounds" in item for item in violations)) + + def test_dose_below_dose_min_fails(self) -> None: + payload = _payload([_product("MELODY FLEX", [_use(dose_min=1.5, dose_max=2.0)])]) + result = _result(recommended_doses=["1,00 kg/ha"]) + violations = validate_advice(result, payload) + self.assertTrue(any("outside the registered dose bounds" in item for item in violations)) + + def test_range_must_lie_inside_one_row(self) -> None: + payload = _payload([_product("MELODY FLEX", [_use(dose_min=1.0, dose_max=2.0)])]) + inside = _result(recommended_doses=["1.2-1.8 kg/ha"]) + spanning = _result(recommended_doses=["0.5-1.8 kg/ha"]) + self.assertEqual(validate_advice(inside, payload), []) + self.assertTrue( + any("outside the registered dose bounds" in item for item in validate_advice(spanning, payload)) + ) + + def test_all_null_dose_bounds_trust_the_model(self) -> None: + uses = [_use(dose_min=None, dose_max=None, concentration_min=180, concentration_max=450)] + payload = _payload([_product("MELODY FLEX", uses)]) + result = _result(recommended_doses=["vedere etichetta"]) + self.assertEqual(validate_advice(result, payload), []) + + def test_null_dose_row_is_ignored_when_another_row_has_bounds(self) -> None: + uses = [ + _use(dose_min=None, dose_max=None), + _use(dose_min=0.9, dose_max=0.9), + ] + payload = _payload([_product("MELODY FLEX", uses)]) + self.assertEqual(validate_advice(_result(), payload), []) + too_high = _result(recommended_doses=["2.00 kg/ha"]) + self.assertTrue(any("outside the registered dose bounds" in item for item in validate_advice(too_high, payload))) + + def test_partial_dose_bounds_are_not_a_reference(self) -> None: + payload = _payload([_product("MELODY FLEX", [_use(dose_min=None, dose_max=1.8)])]) + violations = validate_advice(_result(), payload) + self.assertTrue(any("no complete dose bounds" in item for item in violations)) + + def test_concentration_unit_uses_concentration_bounds(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _result(recommended_doses=["200 g/hl"]) + self.assertEqual(validate_advice(result, payload), []) + + def test_unknown_unit_fails(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _result(recommended_doses=["2.0 bananas"]) + violations = validate_advice(result, payload) + self.assertTrue(any("unknown unit 'bananas'" in item for item in violations)) + + def test_unparseable_dose_fails_when_bounds_exist(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _result(recommended_doses=["vedere etichetta"]) + violations = validate_advice(result, payload) + self.assertTrue(any("unparseable dose" in item for item in violations)) + + def test_bare_name_without_uses_fails(self) -> None: + payload = _payload(["MELODY FLEX"]) + violations = validate_advice(_result(), payload) + self.assertTrue(any("no registered uses" in item for item in violations)) + + def test_uses_on_applied_treatment_when_suggestion_is_a_bare_name(self) -> None: + payload = _payload( + [_product("OTHER", [_use()])], + suggested=["MELODY FLEX"], + applied=[_product("MELODY FLEX", _MELODY_USES)], + ) + self.assertEqual(validate_advice(_result(), payload), []) + + def test_dose_list_must_match_product_list(self) -> None: + payload = _payload( + [ + _product("MELODY FLEX", _MELODY_USES), + _product("OTHER", [_use()]), + ] + ) + result = _result( + recommended_product_names=["MELODY FLEX", "OTHER"], + recommended_doses=["0.90 kg/ha"], + ) + violations = validate_advice(result, payload) + self.assertIn( + "recommended_doses must have one entry for each recommended product.", + violations, + ) + + def test_recommendation_requires_a_product(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _result( + recommended_product_names=[], + recommended_doses=[], + recommended_application_date=_WHEN, + ) + violations = validate_advice(result, payload) + self.assertIn( + "is_treatment_recommended is set but recommended_product_names is empty.", + violations, + ) + + def test_no_recommendation_rejects_a_date(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + result = _no_treatment(recommended_application_date=_WHEN) + violations = validate_advice(result, payload) + self.assertIn( + "is_treatment_recommended is not set but recommended_application_date is set.", + violations, + ) + + def test_model_codes_are_rejected_and_italian_fase_is_kept(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + self.assertEqual(validate_advice(_result(), payload), []) + self.assertIn("fase", _result().full_advice) + + coded = _result( + full_advice="Il modello indica FASE5.", + advice_summary="Incubazione INCUBAZPRIMARIA2_80.7 in corso.", + ) + violations = validate_advice(coded, payload) + self.assertIn("full_advice contains model code 'FASE5'.", violations) + self.assertIn("advice_summary contains model code 'INCUBAZPRIMARIA'.", violations) + + +class GenerateAdviceRetryTests(unittest.TestCase): + def _settings(self, attempts: int = 3) -> SimpleNamespace: + return SimpleNamespace( + retry=SimpleNamespace(attempts=attempts), + llm_advice_generation=object(), + ) + + def test_second_attempt_receives_the_violation_and_stops(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + replies = [ + _reply(recommended_doses=["9.0 kg/ha"]), + _reply(), + ] + calls: list[str] = [] + + def llm_call(_settings, _cfg, _system, user_content, json_output=False) -> str: + self.assertTrue(json_output) + calls.append(user_content) + return replies.pop(0) + + result = generate_advice( + self._settings(), + "system", + "user", + payload, + _WHEN, + llm_call=llm_call, + ) + self.assertEqual(result.recommended_doses, ["0.90 kg/ha"]) + self.assertEqual(len(calls), 2) + self.assertIn("outside the registered dose bounds", calls[1]) + self.assertIn("9.0 kg/ha", calls[1]) + + def test_exhausted_attempts_raise_and_do_not_continue(self) -> None: + payload = _payload([_product("MELODY FLEX", _MELODY_USES)]) + calls: list[str] = [] + + def llm_call(_settings, _cfg, _system, user_content, json_output=False) -> str: + calls.append(user_content) + return _reply(recommended_doses=["9.0 kg/ha"]) + + with self.assertRaises(AdviceRejected): + generate_advice( + self._settings(attempts=3), + "system", + "user", + payload, + _WHEN, + llm_call=llm_call, + ) + self.assertEqual(len(calls), 3) + + +class ParseAdviceDateTests(unittest.TestCase): + def test_unparseable_application_date_raises(self) -> None: + raw = _reply(recommended_application_date="not-a-date") + with self.assertRaises(PipelineError) as caught: + parse_advice(raw, _WHEN) + self.assertIn("Could not parse recommended_application_date", str(caught.exception)) + + +if __name__ == "__main__": + unittest.main()