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.
This commit is contained in:
Arsham Mirehvandi
parent
46cc17e5d5
commit
6b523820f6
11
README.md
11
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
|
||||
|
||||
@ -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,16 +83,26 @@ 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 = ""
|
||||
|
||||
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,
|
||||
@ -99,7 +110,24 @@ def generate_advice(
|
||||
user_content,
|
||||
json_output=True,
|
||||
)
|
||||
return parse_advice(raw, as_of)
|
||||
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}'.")
|
||||
|
||||
318
pipeline/stages/advice_guardrails.py
Normal file
318
pipeline/stages/advice_guardrails.py
Normal file
@ -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<low>\d+(?:[.,]\d+)?)"
|
||||
r"(?:\s*[-–—]\s*(?P<high>\d+(?:[.,]\d+)?))?"
|
||||
r"\s*(?P<unit>\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
|
||||
356
tests/test_advice_guardrails.py
Normal file
356
tests/test_advice_guardrails.py
Normal file
@ -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()
|
||||
Loading…
Reference in New Issue
Block a user