From d8273b5cd1fa07143da42ee9f11158459ce2e624 Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Wed, 9 Sep 2026 16:00:23 +0200 Subject: [PATCH] Enhance advice context and treatment handling - Updated README.md to clarify the conditions for including `last_applied_treatment` in the advice JSON structure. - Modified job.py to pass `as_of` and `field_id` to the advice generation function. - Enhanced advice_context.py to conditionally add `last_applied_treatment` based on recent treatment history and allowed products. - Introduced new functions in treatments.py to group treatments by day and load the last treatment before the recent window. - Updated _output_format.md to specify the conditions under which `last_applied_treatment` is included in the advisory. --- README.md | 18 ++- pipeline/job.py | 2 + pipeline/stages/advice_context.py | 73 ++++++++++ pipeline/stages/treatments.py | 75 ++++++++-- prompts/_output_format.md | 18 ++- tests/test_advice_context.py | 221 ++++++++++++++++++++++++++++++ 6 files changed, 387 insertions(+), 20 deletions(-) create mode 100644 tests/test_advice_context.py diff --git a/README.md b/README.md index afb6e95..515c6dd 100644 --- a/README.md +++ b/README.md @@ -182,7 +182,7 @@ Per-job payload fields: | Field | Description | |---|---| | `run` | Metadata (as_of, field, crop, disease, organic, station, …) | -| `json_for_advice_generation` | Full 11-day meteorological + forecast payload, plus `date_of_today`, `allowed_products`, `applied_treatment` and `last_advice`; this is what the advice LLM receives | +| `json_for_advice_generation` | Full 11-day meteorological + forecast payload, plus `date_of_today`, `allowed_products`, `applied_treatment`, `last_advice`, and (when present) `last_applied_treatment`; this is what the advice LLM receives | | `json_for_query_synthesis` | Subset of the above, sent to the LLM for search query generation | | `generated_queries` | One or two natural-language search queries | | `candidate_count` | Products remaining after the prefilter | @@ -196,7 +196,17 @@ Per-job payload fields: `json_for_advice_generation.date_of_today` is oggi for this advisory (the run's `as_of`, formatted `DD-MM-YYYY`). `last_advice.date` is the previous advisory's -issuance date, in the same format. +issuance date, in the same format. `last_applied_treatment` is omitted unless +`allowed_products` is non-empty, no spray appears in the meteorological window +(`as_of-5 .. as_of`), and a spray exists in `as_of-25 .. as_of-6`; when present +it is `{ "date": "", "products": [...] }` at the end of the payload. +A product already described under `allowed_products` or +`last_advice.suggested_products` is listed by name; any other product is the +full label object. When a treatment is recommended and `allowed_products` is +non-empty, the advice LLM reuses `last_advice.suggested_products` if those +products remain appropriate, even when they are not in `allowed_products`; +otherwise it is restricted to `allowed_products`. If `allowed_products` is +empty, nothing is prescribed. If a field's 11-day window has no `FASE2`, `FASE5`, or `INCUBAZPRIMARIA` day and no true `observation` day (`as_of-5 .. as_of`), the product search is skipped @@ -353,7 +363,7 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: 3. **Observation** — `AI_agrosupport_agro_ril_pathogen` (`rilpato_layer`/`rilpato_date`/`rilpato_diffusion`); `observation` is true for `as_of` or any of the past 5 days if a row exists there with `rilpato_diffusion` other than `32` (including `NULL`) 4. **Weather** — `TDatiMeteo_D` for past days, `TDatiMeteo_D_FRC` for today/future 5. **Phenology** — carry-forward of latest observation ≤ day; future days null; if today's phase is missing, use yesterday's -6. **Applied treatments** — products sprayed over `as_of-5 .. as_of`, from `AI_agrosupport_agro_ril_operations*` with `rilop_operation = 10` +6. **Applied treatments** — products sprayed over `as_of-5 .. as_of`, from `AI_agrosupport_agro_ril_operations*` with `rilop_operation = 10`. A separate lookback over `as_of-25 .. as_of-6` is used only for `last_applied_treatment` during context enrichment. 7. **LLM** — synthesise 1–2 anonymised search queries from the subset 8. **Product prefilter** — organic / crop / disease gates, then one prefilter rule against the batch-wide, in-memory `products` index (`pipeline/stages/disease.py::select_prefilter_rule`, first match wins): @@ -369,6 +379,6 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: 9. **Vector search** — Weaviate `near_text` on `ProductProfile`, filtered server-side to the allowlisted `product_id`s 10. **Product JSON** — label details per product from `products`, `product_uses`, `label_chunks`, cached per `(crop, disease)` -11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advice` table +11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advice` table, and optional `last_applied_treatment` (spray date + products) when the recent window is empty but a spray exists within 25 days 12. **Advice generation** — second LLM profile returns the structured advisory 13. **Persistence** — the advisory plus its collapsed input are written to `advice` inside one transaction diff --git a/pipeline/job.py b/pipeline/job.py index d2a880b..e30aaab 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -297,6 +297,8 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - last_advice_summary=last_summary, last_advice_treatments=last_treatments, last_advice_date=last_date, + as_of=as_of, + field_id=job.field_id, ) payload["last_advice"] = json_for_advice_generation["last_advice"] diff --git a/pipeline/stages/advice_context.py b/pipeline/stages/advice_context.py index 4136a77..bf4e43d 100644 --- a/pipeline/stages/advice_context.py +++ b/pipeline/stages/advice_context.py @@ -12,6 +12,7 @@ import pyodbc from pipeline.db import fetch_one from pipeline.jsonutils import parse_json_list from pipeline.stages.product_json import ProductJsonBuilder, normalize_name +from pipeline.stages.treatments import load_last_treatment_before_recent_window from pipeline.stages.vector import RecommendedProduct from pipeline.window import format_date @@ -66,6 +67,29 @@ def load_last_advice( return summary, treatments, issued_on +def _entry_name(entry: Any) -> str: + """Product name from a plain string or a one-key Product JSON object.""" + if isinstance(entry, dict) and len(entry) == 1: + return str(next(iter(entry))) + return str(entry) + + +def _names_from_list(items: Iterable[Any]) -> set[str]: + names: set[str] = set() + for item in items: + name = _entry_name(item).strip() + if name: + names.add(normalize_name(name)) + return names + + +def _has_recent_applied_treatment(advice_json: dict[str, Any]) -> bool: + for day in advice_json.get("meteorological_data", []): + if day.get("applied_treatment"): + return True + return False + + def enrich_advice_json( advice_json: dict[str, Any], *, @@ -74,6 +98,8 @@ def enrich_advice_json( recommended: Iterable[RecommendedProduct], last_advice_summary: str, last_advice_treatments: Iterable[str], + as_of: date, + field_id: int, last_advice_date: date | None = None, ) -> None: """ @@ -85,6 +111,10 @@ def enrich_advice_json( thread's own SQL connection; `builder` may be shared with other threads handling other fields for the same crop/disease pair (see `ProductJsonBuilder`), so it never stores a connection itself. + + When allowed_products is non-empty, the meteorological window has no sprays, + and a spray exists in as_of-25 .. as_of-6, append last_applied_treatment + as the last key (date + products). Otherwise that key is omitted. """ recommended = list(recommended) @@ -112,6 +142,45 @@ def enrich_advice_json( ], } + _maybe_add_last_applied_treatment( + advice_json, + conn=conn, + builder=builder, + as_of=as_of, + field_id=field_id, + ) + + +def _maybe_add_last_applied_treatment( + advice_json: dict[str, Any], + *, + conn: pyodbc.Connection, + builder: ProductJsonBuilder, + as_of: date, + field_id: int, +) -> None: + if not advice_json.get("allowed_products"): + return + if _has_recent_applied_treatment(advice_json): + return + + last = load_last_treatment_before_recent_window(conn, field_id, as_of) + if last is None: + return + + treatment_date, product_names = last + last_advice = advice_json.get("last_advice") or {} + described_names = _names_from_list(advice_json.get("allowed_products") or []) | _names_from_list( + last_advice.get("suggested_products") or [] + ) + advice_json["last_applied_treatment"] = { + "date": format_date(treatment_date), + "products": [ + name if normalize_name(name) in described_names else builder.build(conn, name) + for name in product_names + ], + } + def _collapse_entry(entry: Any) -> Any: """Reduce a Product JSON object to its product name; leave plain names untouched.""" @@ -137,6 +206,10 @@ def _collapse_in_place(node: Any) -> None: for key, value in node.items(): if key in _PRODUCT_LIST_KEYS and isinstance(value, list): node[key] = [_collapse_entry(item) for item in value] + elif key == "last_applied_treatment" and isinstance(value, dict): + products = value.get("products") + if isinstance(products, list): + value["products"] = [_collapse_entry(item) for item in products] else: _collapse_in_place(value) elif isinstance(node, list): diff --git a/pipeline/stages/treatments.py b/pipeline/stages/treatments.py index 4410431..7252cd9 100644 --- a/pipeline/stages/treatments.py +++ b/pipeline/stages/treatments.py @@ -4,7 +4,7 @@ from __future__ import annotations import logging from datetime import date, timedelta -from typing import Any +from typing import Any, Iterable import pyodbc @@ -14,6 +14,10 @@ logger = logging.getLogger(__name__) # rilop_operation == 10 identifies a crop protection treatment. _TREATMENT_OPERATION = 10 +# Meteorological applied_treatment covers as_of-RECENT .. as_of. +RECENT_TREATMENT_DAYS = 5 +# last_applied_treatment looks at sprays strictly before that window, back to as_of-25. +LAST_TREATMENT_MAX_AGE_DAYS = 25 _APPLIED_TREATMENTS_SQL = """ SELECT DISTINCT @@ -40,6 +44,21 @@ def _to_date(value: Any) -> date: return date(value.year, value.month, value.day) +def _group_treatments_by_day(rows: Iterable[Any]) -> dict[date, list[str]]: + by_day: dict[date, list[str]] = {} + for row in rows: + if row.rilop_date is None: + continue + day = _to_date(row.rilop_date) + name = str(row.prodgov_name).strip() + if not name: + continue + names = by_day.setdefault(day, []) + if name not in names: + names.append(name) + return by_day + + def load_applied_treatments( conn: pyodbc.Connection, field_id: int, @@ -52,7 +71,7 @@ def load_applied_treatments( no operation matched the field, the date window, or the treatment operation type, in which case every applied_treatment stays empty downstream. """ - start = as_of - timedelta(days=5) + start = as_of - timedelta(days=RECENT_TREATMENT_DAYS) rows = fetch_all( conn, _APPLIED_TREATMENTS_SQL, @@ -69,18 +88,7 @@ def load_applied_treatments( ) return {} - by_day: dict[date, list[str]] = {} - for row in rows: - if row.rilop_date is None: - continue - day = _to_date(row.rilop_date) - name = str(row.prodgov_name).strip() - if not name: - continue - names = by_day.setdefault(day, []) - if name not in names: - names.append(name) - + by_day = _group_treatments_by_day(rows) logger.info( "Applied treatments for field %s: %d day(s), %d product mention(s)", field_id, @@ -88,3 +96,42 @@ def load_applied_treatments( sum(len(v) for v in by_day.values()), ) return by_day + + +def load_last_treatment_before_recent_window( + conn: pyodbc.Connection, + field_id: int, + as_of: date, +) -> tuple[date, list[str]] | None: + """ + Return the most recent spray in as_of-25 .. as_of-6, or None. + + The meteorological window already covers as_of-5 .. as_of; this lookback + is only used for last_applied_treatment when that recent window is empty. + """ + start = as_of - timedelta(days=LAST_TREATMENT_MAX_AGE_DAYS) + end = as_of - timedelta(days=RECENT_TREATMENT_DAYS + 1) + rows = fetch_all( + conn, + _APPLIED_TREATMENTS_SQL, + (field_id, start, end, _TREATMENT_OPERATION), + ) + by_day = _group_treatments_by_day(rows) + if not by_day: + logger.info( + "No crop protection operations for field %s between %s and %s.", + field_id, + start.isoformat(), + end.isoformat(), + ) + return None + + latest = max(by_day) + names = by_day[latest] + logger.info( + "Last applied treatment for field %s on %s: %s", + field_id, + latest.isoformat(), + names, + ) + return latest, names diff --git a/prompts/_output_format.md b/prompts/_output_format.md index d5d5635..a140ec6 100644 --- a/prompts/_output_format.md +++ b/prompts/_output_format.md @@ -12,11 +12,25 @@ introduction, no commentary outside the JSON. | `dosage` | array of strings | Prescribed dose for each product in `treatments`, in the same order (for example `"1.5 l/ha"`). Empty array `[]` when you recommend no product. | | `apply_date` | string or null | Recommended application date as `YYYY-MM-DD`. `null` when you recommend no product. | -Only recommend products that appear in the `allowed_products` list of the input payload. -If `allowed_products` is empty, you must not prescribe any product. +If `allowed_products` is empty, you must not prescribe any product (`treatments` +must be `[]` and `apply_treatment` must be `0`), even if +`last_advice.suggested_products` names one. + +When you do recommend a treatment (`apply_treatment` = 1) and `allowed_products` +is non-empty: if `last_advice.suggested_products` names one or more products +that are still appropriate now (phenology, weather, disease-model risk, and +that product's own label constraints), `treatments` must be those products — +same names, and the same set if it was a tank mix — even when they are absent +from `allowed_products`. Do not prescribe a product solely because `last_advice` +did. Otherwise recommend only products that appear in `allowed_products`. Never +invent a name that appears in neither `last_advice.suggested_products` nor +`allowed_products`. `date_of_today` is oggi for this advisory. `last_advice` was issued on `last_advice.date`, which is an earlier day; do not read dates in that block as the current day. +`last_applied_treatment` appears only when no spray occurred in the past-5-day +meteorological window but one did occur within the past 25 days; its `date` is +that spray day, not oggi. ### Example shape diff --git a/tests/test_advice_context.py b/tests/test_advice_context.py new file mode 100644 index 0000000..8ea241b --- /dev/null +++ b/tests/test_advice_context.py @@ -0,0 +1,221 @@ +"""Unit tests for last_applied_treatment enrichment and product-JSON collapse.""" + +from __future__ import annotations + +import unittest +from datetime import date +from types import SimpleNamespace +from unittest.mock import ANY, MagicMock, patch + +from pipeline.stages.advice_context import collapse_product_json, enrich_advice_json +from pipeline.stages.treatments import load_last_treatment_before_recent_window +from pipeline.stages.vector import RecommendedProduct + +_AS_OF = date(2026, 4, 16) +_FIELD_ID = 7216 +_SPRAY_DATE = date(2026, 4, 10) + + +def _recommended(name: str) -> RecommendedProduct: + return RecommendedProduct( + product_name=name, + registration_number=None, + distance=None, + query_index=0, + product_id=name, + ) + + +class _StubBuilder: + def build(self, conn, name): # noqa: ANN001 - matches ProductJsonBuilder.build + return {name: {"uses": []}} + + +def _payload(*, applied: list[str] | None = None) -> dict: + return { + "date_of_today": "16-04-2026", + "meteorological_data": [ + {"date": "11-04-2026", "applied_treatment": list(applied or [])}, + ], + "weather_forecasts": [], + } + + +def _enrich( + advice_json: dict, + *, + recommended: list[RecommendedProduct] | None = None, + last_advice_treatments: list[str] | None = None, + last_advice_date: date | None = None, +) -> None: + enrich_advice_json( + advice_json, + conn=MagicMock(), + builder=_StubBuilder(), + recommended=recommended or [], + last_advice_summary="", + last_advice_treatments=last_advice_treatments or [], + as_of=_AS_OF, + field_id=_FIELD_ID, + last_advice_date=last_advice_date, + ) + + +class LastAppliedTreatmentTests(unittest.TestCase): + def test_omits_when_allowed_products_empty_and_does_not_query(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window" + ) as loader: + _enrich(advice_json, recommended=[]) + loader.assert_not_called() + self.assertNotIn("last_applied_treatment", advice_json) + self.assertEqual(advice_json["allowed_products"], []) + + def test_omits_when_recent_applied_treatment_present(self) -> None: + advice_json = _payload(applied=["LIETO"]) + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window" + ) as loader: + _enrich(advice_json, recommended=[_recommended("LIETO")]) + loader.assert_not_called() + self.assertNotIn("last_applied_treatment", advice_json) + + def test_omits_when_loader_returns_none(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=None, + ) as loader: + _enrich(advice_json, recommended=[_recommended("LIETO")]) + loader.assert_called_once_with(ANY, _FIELD_ID, _AS_OF) + self.assertNotIn("last_applied_treatment", advice_json) + + def test_includes_at_end_when_all_gates_pass(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=(_SPRAY_DATE, ["LIETO"]), + ): + _enrich(advice_json, recommended=[_recommended("LIETO")]) + self.assertEqual(list(advice_json)[-1], "last_applied_treatment") + self.assertEqual( + advice_json["last_applied_treatment"], + {"date": "10-04-2026", "products": ["LIETO"]}, + ) + + def test_name_only_when_product_in_allowed_products(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=(_SPRAY_DATE, ["LIETO"]), + ): + _enrich(advice_json, recommended=[_recommended("LIETO")]) + self.assertEqual(advice_json["last_applied_treatment"]["products"], ["LIETO"]) + + def test_name_only_when_product_in_suggested_products_as_string(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=(_SPRAY_DATE, ["LIETO"]), + ): + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + last_advice_treatments=["LIETO"], + ) + self.assertEqual(advice_json["last_advice"]["suggested_products"], ["LIETO"]) + self.assertEqual(advice_json["last_applied_treatment"]["products"], ["LIETO"]) + + def test_name_only_when_product_in_suggested_products_as_object(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=(_SPRAY_DATE, ["TWINGO"]), + ): + _enrich( + advice_json, + recommended=[_recommended("LIETO")], + last_advice_treatments=["TWINGO"], + ) + self.assertEqual( + advice_json["last_advice"]["suggested_products"], + [{"TWINGO": {"uses": []}}], + ) + self.assertEqual(advice_json["last_applied_treatment"]["products"], ["TWINGO"]) + + def test_full_json_when_product_in_neither_list(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=(_SPRAY_DATE, ["UNKNOWN"]), + ): + _enrich(advice_json, recommended=[_recommended("LIETO")]) + self.assertEqual( + advice_json["last_applied_treatment"]["products"], + [{"UNKNOWN": {"uses": []}}], + ) + + def test_tank_mix_mixes_name_and_full_json(self) -> None: + advice_json = _payload() + with patch( + "pipeline.stages.advice_context.load_last_treatment_before_recent_window", + return_value=(_SPRAY_DATE, ["LIETO", "UNKNOWN"]), + ): + _enrich(advice_json, recommended=[_recommended("LIETO")]) + self.assertEqual( + advice_json["last_applied_treatment"]["products"], + ["LIETO", {"UNKNOWN": {"uses": []}}], + ) + + +class CollapseProductJsonTests(unittest.TestCase): + def test_collapses_last_applied_treatment_products(self) -> None: + collapsed = collapse_product_json( + { + "allowed_products": [{"LIETO": {"uses": []}}], + "last_advice": { + "date": "15-04-2026", + "advice_summary": "", + "suggested_products": [{"TWINGO": {"uses": []}}], + }, + "last_applied_treatment": { + "date": "10-04-2026", + "products": [{"UNKNOWN": {"uses": []}}, "LIETO"], + }, + } + ) + self.assertEqual(collapsed["allowed_products"], ["LIETO"]) + self.assertEqual(collapsed["last_advice"]["suggested_products"], ["TWINGO"]) + self.assertEqual( + collapsed["last_applied_treatment"], + {"date": "10-04-2026", "products": ["UNKNOWN", "LIETO"]}, + ) + + +class LoadLastTreatmentBeforeRecentWindowTests(unittest.TestCase): + def test_returns_latest_day_products(self) -> None: + rows = [ + SimpleNamespace(rilop_date=date(2026, 3, 22), prodgov_name="OLD"), + SimpleNamespace(rilop_date=date(2026, 4, 10), prodgov_name="LIETO"), + SimpleNamespace(rilop_date=date(2026, 4, 10), prodgov_name="TWINGO"), + ] + with patch("pipeline.stages.treatments.fetch_all", return_value=rows) as fetch: + result = load_last_treatment_before_recent_window( + MagicMock(), _FIELD_ID, _AS_OF + ) + params = fetch.call_args[0][2] + self.assertEqual(params[1], date(2026, 3, 22)) # as_of-25 + self.assertEqual(params[2], date(2026, 4, 10)) # as_of-6 + self.assertEqual(result, (_SPRAY_DATE, ["LIETO", "TWINGO"])) + + def test_returns_none_when_no_rows(self) -> None: + with patch("pipeline.stages.treatments.fetch_all", return_value=[]): + result = load_last_treatment_before_recent_window( + MagicMock(), _FIELD_ID, _AS_OF + ) + self.assertIsNone(result) + + +if __name__ == "__main__": + unittest.main()