From 8a6d48e5a1414236c225178d7357e02d9ec452e4 Mon Sep 17 00:00:00 2001 From: Arsham Mirehvandi <130353205+arshammi@users.noreply.github.com> Date: Tue, 8 Sep 2026 09:40:34 +0200 Subject: [PATCH] Enhance advice generation and context handling - Updated README.md to reflect changes in the `json_for_advice_generation` structure, including the addition of `date_of_today` and clarification of `last_advice` fields. - Modified job.py to load and pass the issuance date of the last advice. - Enhanced advice_context.py to include the issuance date in the last advice retrieval. - Updated assemble.py to include `date_of_today` in the JSON assembly for advice generation. - Improved advice.py to sanitize the `advice_summary` by removing relative date references and ensuring compliance with new guidelines. --- README.md | 10 ++- pipeline/job.py | 4 +- pipeline/stages/advice.py | 15 +++- pipeline/stages/advice_context.py | 25 ++++-- pipeline/stages/advice_summary.py | 78 +++++++++++++++++ pipeline/stages/assemble.py | 3 +- prompts/_output_format.md | 5 +- tests/test_advice_summary.py | 138 ++++++++++++++++++++++++++++++ 8 files changed, 265 insertions(+), 13 deletions(-) create mode 100644 pipeline/stages/advice_summary.py create mode 100644 tests/test_advice_summary.py diff --git a/README.md b/README.md index 3f99fe1..afb6e95 100644 --- a/README.md +++ b/README.md @@ -182,18 +182,22 @@ 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 `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` and `last_advice`; 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 | | `recommended_products` | Up to 6 items with `product_name` + `registration_number` | -| `last_advice` | `advice_summary` + `suggested_products` of the most recent previous advisory | +| `last_advice` | `date` (previous advisory's issuance date) + `advice_summary` + `suggested_products` of the most recent previous advisory. `advice_summary` is machine context for the next run: it is stored without relative day words, attention prefixes, or follow-the-label closers | | `advice` | Parsed LLM output: `full_advice`, `advice_summary`, `apply_treatment`, `treatments`, `dosage`, `apply_date` | | `sent_information` | The advice payload with every embedded product label collapsed back to a product name; mirrors `advice.sent_information` | | `advice_inserted` | Whether the `advice` row was written | | `status` | `ok`, `no_risk`, or `dry_run` (see `_run_report.json` for `failed` / `skipped_*`, which have no per-job file) | | `message` | Human-readable summary of the run outcome | +`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. + 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 and `allowed_products` is empty, but advice is still generated and stored; that @@ -365,6 +369,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` from the `advice` table +11. **Context enrichment** — `allowed_products`, `applied_treatment` expansion, `last_advice` (with issuance `date`) from the `advice` table 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 2f1ec4f..d2a880b 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -286,7 +286,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - builder = resources.product_json_builder(job.crop_english, disease.disease_english) with resources.sql_sem: - last_summary, last_treatments = load_last_advice( + last_summary, last_treatments, last_date = load_last_advice( conn, job.field_id, disease.disease_english, as_of ) enrich_advice_json( @@ -296,6 +296,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - recommended=recommended, last_advice_summary=last_summary, last_advice_treatments=last_treatments, + last_advice_date=last_date, ) payload["last_advice"] = json_for_advice_generation["last_advice"] @@ -307,6 +308,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) - advice_system_prompt, advice_user_prompt, json_for_advice_generation, + as_of, llm_call=resources.call_llm, ) sent_information = collapse_product_json(json_for_advice_generation) diff --git a/pipeline/stages/advice.py b/pipeline/stages/advice.py index ea376ac..9fd7e4c 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_summary import sanitize_advice_summary from pipeline.stages.llm import call_llm logger = logging.getLogger(__name__) @@ -71,6 +72,7 @@ def generate_advice( system_prompt: str, user_prompt: str, advice_json: dict[str, Any], + as_of: date, llm_call: Callable[..., str] = None, ) -> AdviceResult: """ @@ -93,15 +95,24 @@ def generate_advice( user_content, json_output=True, ) - return parse_advice(raw) + return parse_advice(raw, as_of) -def parse_advice(raw: str) -> AdviceResult: +def parse_advice(raw: str, as_of: date) -> AdviceResult: """Parse and validate the advice JSON returned by the LLM.""" payload = _load_json_object(raw) full_advice = _require_text(payload, "full_advice") summary = _require_text(payload, "advice_summary") + cleaned = sanitize_advice_summary(summary, as_of) + if not cleaned: + raise PipelineError("advice_summary is empty after sanitization.") + if cleaned != summary: + logger.warning( + "advice_summary was rewritten to remove relative dates, prefixes, " + "or boilerplate." + ) + summary = cleaned if len(summary) > SUMMARY_MAX_CHARS: logger.warning( "advice_summary is %d characters; truncating to %d.", diff --git a/pipeline/stages/advice_context.py b/pipeline/stages/advice_context.py index 11f130d..4136a77 100644 --- a/pipeline/stages/advice_context.py +++ b/pipeline/stages/advice_context.py @@ -4,7 +4,7 @@ from __future__ import annotations import copy import logging -from datetime import date +from datetime import date, datetime from typing import Any, Iterable import pyodbc @@ -13,6 +13,7 @@ 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.vector import RecommendedProduct +from pipeline.window import format_date logger = logging.getLogger(__name__) @@ -20,6 +21,7 @@ _PRODUCT_LIST_KEYS = ("allowed_products", "applied_treatment", "suggested_produc _LAST_ADVICE_SQL = """ SELECT TOP 1 + [date] AS issued_on, advice_summary, treatments FROM advice @@ -30,13 +32,23 @@ ORDER BY [date] DESC """ +def _row_date(value: Any) -> date | None: + if value is None: + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + return None + + def load_last_advice( conn: pyodbc.Connection, field_id: int, disease_english: str, as_of: date, -) -> tuple[str, list[str]]: - """Return (advice_summary, treatment names) of the most recent advice before as_of.""" +) -> tuple[str, list[str], date | None]: + """Return (advice_summary, treatment names, issuance date) of the most recent advice before as_of.""" row = fetch_one(conn, _LAST_ADVICE_SQL, (field_id, disease_english, as_of)) if row is None: logger.info( @@ -45,12 +57,13 @@ def load_last_advice( disease_english, as_of.isoformat(), ) - return "", [] + return "", [], None summary = (row.advice_summary or "").strip() treatments = [name.strip() for name in parse_json_list(row.treatments) if name.strip()] + issued_on = _row_date(row.issued_on) logger.info("Loaded previous advice with %d suggested product(s).", len(treatments)) - return summary, treatments + return summary, treatments, issued_on def enrich_advice_json( @@ -61,6 +74,7 @@ def enrich_advice_json( recommended: Iterable[RecommendedProduct], last_advice_summary: str, last_advice_treatments: Iterable[str], + last_advice_date: date | None = None, ) -> None: """ Add allowed_products and last_advice, and expand applied_treatment in place. @@ -90,6 +104,7 @@ def enrich_advice_json( known_names = vector_names | applied_names advice_json["last_advice"] = { + "date": format_date(last_advice_date) if last_advice_date else None, "advice_summary": last_advice_summary, "suggested_products": [ name if normalize_name(name) in known_names else builder.build(conn, name) diff --git a/pipeline/stages/advice_summary.py b/pipeline/stages/advice_summary.py new file mode 100644 index 0000000..76d30f1 --- /dev/null +++ b/pipeline/stages/advice_summary.py @@ -0,0 +1,78 @@ +"""Rewrite advice_summary so it is safe to feed back as last_advice. + +The field is machine context for the next run, not farmer-facing prose. +Relative day words, attention prefixes, and follow-the-label closers are +stripped here as a backstop for the output-format prompt. +""" + +from __future__ import annotations + +import re +from datetime import date, datetime, timedelta + +from pipeline.window import format_date + +_DEICTIC = r"(?:oggi|ieri|domani|dopodomani|today|yesterday|tomorrow)" + +_PREFIX = re.compile( + r"^\s*(?:attenzione|nota|importante|avviso|warning|attention)\s*[:!]\s*", + re.IGNORECASE, +) + +# "oggi, 13 aprile 2026" / "today, Apr 13th" — drop the deictic + comma, keep the date. +_DEICTIC_BEFORE_DATE = re.compile(rf"\b{_DEICTIC}\s*,\s*", re.IGNORECASE) + +_STANDALONE_DEICTIC = re.compile(rf"\b{_DEICTIC}\b", re.IGNORECASE) + +# Whole sentence that is only a follow-the-label reminder. Requires +# istruzioni/indicazioni so a real label constraint (season cap, PHI) survives. +_ETICHETTA_CLOSER = re.compile( + r"(?:^|(?<=[.!?]))\s*" + r"(?:Seguire|Rispettare|Attenersi(?:\s+a)?)" + r".{0,80}?" + r"(?:istruzioni|indicazioni)" + r".{0,40}?" + r"etichetta\s*\.", + re.IGNORECASE, +) + +_MULTI_SPACE = re.compile(r"[ \t]{2,}") + +_DEICTIC_OFFSETS = { + "oggi": 0, + "today": 0, + "domani": 1, + "tomorrow": 1, + "ieri": -1, + "yesterday": -1, + "dopodomani": 2, +} + + +def _as_date(issued_on: date) -> date: + if isinstance(issued_on, datetime): + return issued_on.date() + return issued_on + + +def _replace_standalone(match: re.Match[str], issued_on: date) -> str: + offset = _DEICTIC_OFFSETS[match.group(0).casefold()] + return format_date(issued_on + timedelta(days=offset)) + + +def sanitize_advice_summary(text: str, issued_on: date) -> str: + """Strip prefixes, relative day words, and label-instruction closers.""" + issued_on = _as_date(issued_on) + cleaned = (text or "").strip() + if not cleaned: + return "" + + cleaned = _PREFIX.sub("", cleaned, count=1) + cleaned = _DEICTIC_BEFORE_DATE.sub("", cleaned) + cleaned = _STANDALONE_DEICTIC.sub( + lambda match: _replace_standalone(match, issued_on), + cleaned, + ) + cleaned = _ETICHETTA_CLOSER.sub("", cleaned) + cleaned = _MULTI_SPACE.sub(" ", cleaned) + return cleaned.strip() diff --git a/pipeline/stages/assemble.py b/pipeline/stages/assemble.py index 2192df9..89ec8ff 100644 --- a/pipeline/stages/assemble.py +++ b/pipeline/stages/assemble.py @@ -66,7 +66,7 @@ def assemble_json_for_advice_generation( applied_treatments: dict[date, list[str]] | None = None, observation: dict[date, bool] | None = None, ) -> dict[str, Any]: - """Build meteorological_data (as_of-5..as_of) and weather_forecasts (as_of+1..as_of+5).""" + """Build meteorological_data (as_of-5..as_of), weather_forecasts (as_of+1..as_of+5), and date_of_today.""" window = build_window(as_of) applied = applied_treatments or {} meteorological = [ @@ -80,6 +80,7 @@ def assemble_json_for_advice_generation( if day > as_of ] return { + "date_of_today": format_date(as_of), "meteorological_data": meteorological, "weather_forecasts": forecasts_section, } diff --git a/prompts/_output_format.md b/prompts/_output_format.md index b918138..d5d5635 100644 --- a/prompts/_output_format.md +++ b/prompts/_output_format.md @@ -6,7 +6,7 @@ introduction, no commentary outside the JSON. | Key | Type | Rules | |---|---|---| | `full_advice` | string | Your complete advisory text, in Italian. | -| `advice_summary` | string | Italian summary of `full_advice`, **at most 1010 characters**. It must state the recommendation: whether a treatment should be administered or not; if yes, for what reason, on what date (and why that date), and with which product; if no, for what reason. | +| `advice_summary` | string | Italian summary of `full_advice`, **at most 1010 characters**. It must state the recommendation: whether a treatment should be administered or not; if yes, for what reason, on what date (and why that date), and with which product; if no, for what reason. Use calendar dates only (for example `13 aprile 2026` or `13-04-2026`). Never use relative day words such as oggi, ieri, domani, dopodomani, today, yesterday, or tomorrow — even when a calendar date follows them. Do not start with labels such as `Attenzione:`, `Nota:`, or `Importante:`. Do not close with a reminder to follow the product label; that belongs in `full_advice`. | | `apply_treatment` | integer | `1` if you recommend applying a product, `0` otherwise. | | `treatments` | array of strings | Names of the recommended product(s), exactly as written in the input payload. Empty array `[]` when you recommend no product. | | `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. | @@ -15,6 +15,9 @@ introduction, no commentary outside the JSON. 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. +`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. + ### Example shape { diff --git a/tests/test_advice_summary.py b/tests/test_advice_summary.py new file mode 100644 index 0000000..ebeca43 --- /dev/null +++ b/tests/test_advice_summary.py @@ -0,0 +1,138 @@ +"""Unit tests for advice_summary sanitization and parse_advice wiring.""" + +from __future__ import annotations + +import json +import unittest +from datetime import date, datetime + +from pipeline.errors import PipelineError +from pipeline.stages.advice import SUMMARY_MAX_CHARS, parse_advice +from pipeline.stages.advice_summary import sanitize_advice_summary + +_APR_13 = date(2026, 4, 13) + +_APR13_SUMMARY = ( + "Attenzione: il modello segnala un'infezione primaria di peronospora " + "(FASE5) per oggi, 13 aprile 2026. La vite si trova in una fase fenologica " + "di elevata suscettibilità (BBCH 15) e non risulta protetta. È " + "indispensabile intervenire per bloccare l'infezione in corso. Si raccomanda " + "di effettuare un trattamento con azione curativa domani, 14 aprile 2026. " + "Si consiglia l'impiego del prodotto PROFILER, dotato di attività sistemica " + "e curativa, alla dose di 3.0 kg/ha. Questo intervento è fondamentale per " + "impedire che l'infezione si stabilisca e per prevenire l'inizio " + "dell'epidemia stagionale. Seguire attentamente le istruzioni in etichetta." +) + + +def _advice_json(*, summary: str, full_advice: str = "Testo completo.") -> str: + return json.dumps( + { + "full_advice": full_advice, + "advice_summary": summary, + "apply_treatment": 1, + "treatments": ["PROFILER"], + "dosage": ["3.0 kg/ha"], + "apply_date": "2026-04-14", + }, + ensure_ascii=False, + ) + + +class SanitizeAdviceSummaryTests(unittest.TestCase): + def test_apr13_summary_strips_prefix_deictics_and_etichetta(self) -> None: + cleaned = sanitize_advice_summary(_APR13_SUMMARY, _APR_13) + self.assertFalse(cleaned.lower().startswith("attenzione")) + self.assertNotIn("oggi", cleaned.casefold()) + self.assertNotIn("domani", cleaned.casefold()) + self.assertNotIn("etichetta", cleaned.casefold()) + self.assertIn("13 aprile 2026", cleaned) + self.assertIn("14 aprile 2026", cleaned) + self.assertIn("PROFILER", cleaned) + + def test_english_today_tomorrow_before_date(self) -> None: + text = "Spray today, Apr 13th and again tomorrow, Apr 15th." + cleaned = sanitize_advice_summary(text, _APR_13) + self.assertEqual(cleaned, "Spray Apr 13th and again Apr 15th.") + self.assertNotIn("today", cleaned.casefold()) + self.assertNotIn("tomorrow", cleaned.casefold()) + + def test_standalone_oggi_and_domani(self) -> None: + text = "Le condizioni di oggi restano favorevoli e domani peggiorano." + cleaned = sanitize_advice_summary(text, _APR_13) + self.assertEqual( + cleaned, + "Le condizioni di 13-04-2026 restano favorevoli e 14-04-2026 peggiorano.", + ) + + def test_standalone_ieri(self) -> None: + cleaned = sanitize_advice_summary("Pioggia ieri.", _APR_13) + self.assertEqual(cleaned, "Pioggia 12-04-2026.") + + def test_etichetta_closer_variants(self) -> None: + variants = ( + "Intervenire subito. Seguire attentamente le istruzioni in etichetta.", + "Intervenire subito. Seguire attentamente le istruzioni riportate in etichetta.", + "Intervenire subito. Rispettare scrupolosamente le indicazioni riportate in etichetta.", + ) + for text in variants: + with self.subTest(text=text): + cleaned = sanitize_advice_summary(text, _APR_13) + self.assertEqual(cleaned, "Intervenire subito.") + self.assertNotIn("etichetta", cleaned.casefold()) + + def test_etichetta_with_real_content_is_kept(self) -> None: + text = ( + "Non superare i 2 trattamenti per stagione previsti in etichetta " + "per questo formulato." + ) + cleaned = sanitize_advice_summary(text, _APR_13) + self.assertEqual(cleaned, text) + + def test_prefix_only(self) -> None: + cleaned = sanitize_advice_summary( + "Attenzione: trattare il 13 aprile 2026.", + _APR_13, + ) + self.assertEqual(cleaned, "trattare il 13 aprile 2026.") + + def test_already_clean_is_idempotent(self) -> None: + text = "Trattare il 13 aprile 2026 con PROFILER a 3.0 kg/ha." + once = sanitize_advice_summary(text, _APR_13) + twice = sanitize_advice_summary(once, _APR_13) + self.assertEqual(once, text) + self.assertEqual(twice, text) + + def test_apr13_summary_is_idempotent(self) -> None: + once = sanitize_advice_summary(_APR13_SUMMARY, _APR_13) + twice = sanitize_advice_summary(once, _APR_13) + self.assertEqual(once, twice) + + def test_datetime_issued_on(self) -> None: + cleaned = sanitize_advice_summary("Trattare oggi.", datetime(2026, 4, 13, 8, 0)) + self.assertEqual(cleaned, "Trattare 13-04-2026.") + + +class ParseAdviceSanitizeTests(unittest.TestCase): + def test_parse_advice_sanitizes_summary(self) -> None: + result = parse_advice(_advice_json(summary=_APR13_SUMMARY), as_of=_APR_13) + self.assertNotIn("oggi", result.advice_summary.casefold()) + self.assertFalse(result.advice_summary.lower().startswith("attenzione")) + self.assertIn("13 aprile 2026", result.advice_summary) + + def test_empty_after_sanitize_raises(self) -> None: + summary = "Attenzione: Seguire attentamente le istruzioni in etichetta." + with self.assertRaises(PipelineError) as ctx: + parse_advice(_advice_json(summary=summary), as_of=_APR_13) + self.assertIn("empty after sanitization", str(ctx.exception)) + + def test_truncation_after_sanitize(self) -> None: + body = "Trattare il 13 aprile 2026. " * 80 + self.assertGreater(len(body), SUMMARY_MAX_CHARS) + result = parse_advice(_advice_json(summary=body), as_of=_APR_13) + self.assertEqual(len(result.advice_summary), SUMMARY_MAX_CHARS) + self.assertTrue(result.advice_summary.startswith("Trattare il 13 aprile 2026.")) + + +if __name__ == "__main__": + unittest.main()