AI_Agro_Support/README.md

# AI Agro Support — Daily Pipeline

Daily agronomic advisory pipeline. For every field whose disease model is
enabled for advice, it builds an 11-day weather / phenology / disease-forecast
JSON, asks an LLM to synthesise one or two vector-search queries, prefilters
chemical products in SQL Server, and retrieves up to four matching products from
a local Weaviate `ProductProfile` collection. It then enriches that JSON with
full product label information and the previous advisory, asks a second LLM to
write the farmer-facing advice, and stores the result in the `advisories` table.

Each morning's `batch` run discovers `(field, disease, anmod_id)` jobs from the
database. A disease can generate advice when
`AI_agrosupport_an_pathogen_models.pat_mod_AI_model` is 1; its name is
`pat_mod_desc`. A field model is included when
`AI_agrosupport_agro_models.anmod_AI_advice` is 1 and `anmod_model` matches that
name. Several mathematical models for the same field and disease stay separate
jobs, one per `anmod_id`. The cultivated crop is read from the field layer. The
run then executes the sequence above for each job — in parallel, with one
field's failure never blocking another's.

## Prerequisites

- Python 3.12+
- Microsoft SQL Server (`DBMeteo`, and `SQL_APP_DATABASE` on the same instance) reachable with ODBC Driver 17 (or 18)
- Weaviate running locally (see `docker-compose.yml`) with a populated `ProductProfile` collection
- API key for the chosen LLM provider (Gemini by default)

## Setup

```powershell
cd d:\Uni\Thesis\AI-Agro-Support
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
copy .env.example .env
# Edit .env with SQL, Weaviate, and LLM credentials
```

Start Weaviate if it is not already running:

```powershell
docker compose up -d
```

## Configuration

- [`config.yaml`](config.yaml) — concurrency/rate limits, retry policy, the 09:00
  readiness target, LLM provider/model, vocab paths, and an optional field allowlist.
  Advice targets come from the database (see below)
- [`.env`](.env) — SQL Server, Weaviate, and API keys
- [`vocab/crops.yaml`](vocab/crops.yaml) / [`vocab/diseases.yaml`](vocab/diseases.yaml) — Italian → English maps
- [`prompts/`](prompts) — the prompt registry (see below)

### SQL databases

The pipeline opens one connection to `SQL_DATABASE`. Field, weather, and disease-model tables (`AI_agrosupport_*`) stay in that database (`DBMeteo`). `advisories`, `products`, `product_uses`, and `label_chunks` are read through the same connection from `SQL_APP_DATABASE` when that names a different database on the same instance. Leave `SQL_APP_DATABASE` unset, or set it equal to `SQL_DATABASE`, when those four tables live in the connected database. `SQL_SERVER` is `host` or `host,port`; the ODBC driver uses a comma before the port.

### Advice targets

`batch` does not read a crop or disease list from `config.yaml`. It loads
diseases with `pat_mod_AI_model = 1` from `AI_agrosupport_an_pathogen_models`
(`pat_mod_desc`), then fields from `AI_agrosupport_agro_models` where
`anmod_AI_advice = 1` and `anmod_model` matches one of those names. The field
id is `anmod_cmplay`. The crop is `cmplay_colture` on `AI_agrosupport_cmp_layers`,
joined to `AI_agrosupport_an_colture` and mapped through
[`vocab/crops.yaml`](vocab/crops.yaml) (`Vite` → `grapevine`). The English
disease name is mapped through [`vocab/diseases.yaml`](vocab/diseases.yaml)
(`PERONOSPORA` → `downy mildew`). Those two English names select the prompt
and the product filter. One agro-model row is one job. Its `anmod_id` is
written to `advisories.pathogen_model_id`. The same id is `model_model` in
`AI_agrosupport_agro_model_tmp2`, and the forecast uses `model_description`
from that model only. A field with three advice-enabled models for one disease
produces three advisories.

To enable another disease, set both flags in SQL. Add a `pat_mod_desc` mapping
to `vocab/diseases.yaml` if it is not already there (`peronospora` already is),
and optionally add a prompt directory (see below). An unmapped disease or crop
name is skipped and logged. A pair without a dedicated prompt uses the generic
fallback and is logged at startup.

### `worklist:`, `concurrency:`, `retry:`, `schedule:`

```yaml
worklist:
  field_allowlist: []           # restrict to specific field IDs while testing

concurrency:
  workers: 12                   # ThreadPoolExecutor size
  limits: { sql: 6, gemini: 12 } # see "Concurrency model" below
  gemini_requests_per_minute: 60

retry:
  attempts: 3
  initial_backoff_seconds: 2
  max_backoff_seconds: 30

schedule:
  deadline: "09:00"             # readiness target; jobs still run after this time
```

### Prompt registry (`prompts/`)

Each crop-disease pair needs its own advice prompt (its phytopathological
model's phase semantics, phenology notes, etc.). Prompts are plain Markdown
files resolved by `pipeline/prompts.py::PromptRegistry`, with a shared fallback
so a newly enabled pair never fails outright — it just logs a warning and
uses a generic prompt until a dedicated one is written.

```
prompts/
  _output_format.md                          shared JSON contract, appended to every advice system prompt
  query_synthesis/_default.md                generic query-synthesis system prompt
  query_synthesis/<crop>__<disease>.md        optional per-pair override
  advice/_default/{system.md,user.md}         fallback advice prompts
  advice/<crop>__<disease>/{system.md,user.md} per-pair advice prompts
```

The `<crop>__<disease>` slug is built from the canonical English crop and disease
names (lower-cased, non-alphanumeric runs collapsed to `_`) — e.g. `grapevine` +
`downy mildew` → `grapevine__downy_mildew`. To add a prompt for a new pair, create
`prompts/advice/<slug>/system.md` and `user.md`; `batch` logs every discovered
pair still falling back to `_default` at startup.

## Run

```powershell
.\.venv\Scripts\Activate.ps1

# Daily entry point: every advice-enabled field, disease, and model in the database
python -m pipeline batch

# Historical / test dates (useful when the live window has no FASE2/FASE5/INCUBAZPRIMARIA/observation)
python -m pipeline batch --as-of 2026-06-29

# Narrow to one crop, disease, and/or field while testing
python -m pipeline batch --crop grapevine --disease "downy mildew" --field-id 4085

# Re-run a partially failed morning without redoing already-written advice
python -m pipeline batch --skip-existing-advice

# Build every job's JSON but skip LLM calls, vector search, and the DB write
python -m pipeline batch --dry-run

# Single-field debugging. Without --anmod-id, every advice-enabled model for
# that field and disease is run.
python -m pipeline one --field-id 4085 --disease PERONOSPORA --dry-run
python -m pipeline one --field-id 4085 --disease PERONOSPORA --anmod-id 11 --dry-run
```

`batch` accepts `--deadline HH:MM` and `--workers N` to override
`schedule.deadline` / `concurrency.workers` from `config.yaml` for one run.
Jobs still run after `--deadline`; it is the time advice should be ready, not a cutoff.

## Concurrency model

One `ThreadPoolExecutor` (`concurrency.workers`) runs every discovered job.
Because dozens to a couple hundred jobs can run in the same morning, several
things that used to be "open it, use it, close it" per field are now shared
across the whole run (see `pipeline/resources.py`):

- **SQL**: each worker thread gets its own lazily-created `pyodbc` connection
  (connections cannot be shared across threads), kept open for the batch's
  lifetime. `concurrency.limits.sql` bounds how many jobs may run their
  SQL-heavy phase (forecast/weather/phenology/treatments lookup, or the
  product-JSON/advice-insert phase) at the same instant, independent of how
  many connections are open.
- **Weaviate**: one client is opened for the whole batch instead of one per
  field.
- **Gemini**: `near_text` search embeds through the same `GEMINI_API_KEY` as
  both LLM calls (`ProductProfile` is vectorised with text2vec-palm /
  gemini-embedding-001), so `concurrency.limits.gemini` and
  `gemini_requests_per_minute` are a single shared budget covering query
  synthesis, advice generation, *and* vector search — not two independent
  ones.
- **Products**: the whole `products` table is loaded into memory once per
  batch (`pipeline/stages/products.py::ProductIndex`) instead of re-scanned
  and re-parsed once per field, and `ProductJsonBuilder` is shared per
  `(crop, disease)` pair instead of per field.

Each of the three external boundaries — LLM, Weaviate, SQL — has its own
`tenacity` retry policy in `pipeline/retry.py` (exponential backoff with
jitter, `retry.attempts` tries), applied automatically inside
`Resources.call_llm` / `Resources.search_products` and around the
`insert_advice` transaction.

## Output

Each run writes to `output/<YYYY-MM-DD>/`:

- `field<field_id>__<crop>__<disease slug>__model<anmod_id>.json` — one file per job (see fields below)
- `_run_report.json` — the operational artifact for the morning: totals by
  status, `jobs_started_after_deadline` (jobs that began after the 09:00
  target; they are still written), plus a per-job breakdown (`field_id`,
  `disease`, `anmod_id`, `status`, `duration_s`, `recommended_count`,
  `allowed_products_empty`, `message`, and, on failure, `error_type` /
  `error_message`)

Per-job payload fields:

| Field | Description |
|---|---|
| `run` | Metadata (as_of, field, crop, disease, anmod_id, organic, station, …) |
| `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 |
| `recommended_products` | Up to 4 items with `product_name` + `registration_number` |
| `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`, `is_treatment_recommended`, `recommended_product_names`, `recommended_doses`, `recommended_application_date` |
| `advice_input_json` | The advice payload with every embedded product label collapsed back to a product name; mirrors `advisories.advice_input_json` |
| `advice_inserted` | Whether the `advisories` 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. `last_applied_treatment` is omitted unless
no spray appears in the meteorological window (`as_of-5 .. as_of`) and a spray
exists in `as_of-25 .. as_of-6`, including when `allowed_products` is empty;
when present it is `{ "date": "<spray day>", "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. The advisory explains that the model shows no
infection risk, that the last spray still covers the crop (or that its cover
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
job is labelled `no_risk`. A covered incubation does not count as risk. The run report also counts `ok` jobs whose
`allowed_products` ended up empty anyway (for example an organic field with no
organic-certified product in the catalog yet) — watch that number as the
`products` table fills in, since a sudden jump usually means the prefilter or
the Weaviate allowlist filter is misbehaving rather than that the catalog is
genuinely empty for that combination.

The process exit code is `0` only if every job in the report is `ok`, `no_risk`,
`dry_run`, or a deliberate `skipped_*`; it is `1` if any job's status is `failed`.

## Database writes

Each non-dry-run job inserts one row into `advisories` for `(issuance_date,
pathogen_model_id)`. `pathogen_model_id` stores that job's `anmod_id`. That pair
is the table's primary key, so any existing row for the same pair is deleted
first — re-running a day replaces that model's advisory and leaves the other
models for the same day in place. `field_id` and `disease` stay on the row:
`disease` is the English name that selects the prompt and the product filter.
The delete-then-insert pair runs inside one transaction (`pipeline/db.py::transaction`)
so a crash between the two statements cannot leave that model's advisory
missing for the day, and deadlocks are retried per `retry:` in config.yaml.

The primary key's clustered index on `(issuance_date, pathogen_model_id)` also
lets that DELETE seek straight to the row it needs instead of scanning the
table, so concurrent workers don't block each other on unrelated rows.

## Daily scheduling

Provide a CLI entry point only. Schedule it yourself on the deployment machine.
Advice should be ready by the `schedule.deadline` (09:00 by default), and weather
data should stay fresh, so start the run no earlier than ~07:00 — for example
via Windows Task Scheduler, daily at 07:00:

```
d:\Uni\Thesis\AI-Agro-Support\.venv\Scripts\python.exe -m pipeline batch --skip-existing-advice
```

with working directory `d:\Uni\Thesis\AI-Agro-Support`. Jobs still waiting at
09:00 are run and written. `_run_report.json` records how many of them started
late in `jobs_started_after_deadline`. `--skip-existing-advice`
makes a re-trigger (e.g. a scheduled retry later that morning) resumable at no
extra cost: it does one bulk check against `advisories` up front and only runs jobs
that don't have a row yet for that pathogen model.

## Observability (optional)

The pipeline can send OpenTelemetry/OpenInference traces to a self-hosted
[Arize Phoenix](https://github.com/Arize-ai/phoenix) instance for debugging a
bad advisory, comparing prompts/models later, and seeing per-job token usage
and USD cost for the two LLM stages. It is **off by default**, **fail-open**
(a down or missing Phoenix never fails `batch` or `one`), and **not** a
runtime dependency: nothing about the daily 07:00–09:00 run depends on it.

### Start Phoenix

```powershell
docker compose up -d
```

This brings up Weaviate and Phoenix together. Phoenix serves its UI (and the
OTLP/HTTP trace collector) at <http://localhost:6006>. SQLite on the
`phoenix_data` Docker volume is enough for this single-machine setup — see
"Why SQLite / why not Langfuse" below.

### Turn tracing on

Tracing is controlled by `observability:` in `config.yaml`:

```yaml
observability:
  enabled: false                                  # default: off
  project_name: ai-agro-support
  endpoint: http://localhost:6006/v1/traces
  hide_prompts: true                               # keep prompt/payload text out of spans
  max_attribute_chars: 4096                        # cap every exported string attribute
```

Any of these can be overridden per-environment without editing the file:

| Env var | Overrides |
|---|---|
| `PHOENIX_TRACING_ENABLED` | `observability.enabled` |
| `PHOENIX_COLLECTOR_ENDPOINT` | `observability.endpoint` |
| `PHOENIX_PROJECT_NAME` | `observability.project_name` |

With tracing on, `python -m pipeline one` or `batch` registers the tracer
once at startup (`pipeline/observability.py::setup_tracing`, called from
`pipeline/__main__.py` right after `load_settings()`), then instruments
`google-genai` (and `openai` / `anthropic`, if those `llm:` profiles are ever
enabled) via
[OpenInference](https://github.com/Arize-ai/openinference) auto-instrumentors.
Spans are batched and exported on a background thread, so a slow or down
collector cannot block a worker or push the run past `schedule.deadline`.

### What you get per job

Every `(field, crop, disease, anmod_id, as_of)` job produces one trace, rooted at a
`run_job` span, with:

- a `query_synthesis` span (gemini-3.8-flash by default) and an
  `advice_generation` span (gemini-3.1-pro-preview), each wrapping the provider's own
  auto-instrumented LLM span with `llm.model_name`, `llm.provider`, and
  token counts (`llm.token_count.prompt` / `completion` / `total`, plus
  `completion_details.reasoning` for Gemini thinking tokens);
- a `search_products` retriever span with the synthesized queries and the
  retrieved product IDs/distances (never product names or label content);
- `dry_run` jobs get only the `run_job` span — no LLM or retrieval children,
  since `--dry-run` skips those calls entirely.

Phoenix computes cost per span from its built-in model pricing table. If a
configured model (`gemini-3.8-flash` or `gemini-3.1-pro-preview`) is not
recognised, add it under **Settings → Models** (regex `Name Pattern`,
provider `google`, per-1M-token prices).

Trace and span inputs are hidden by default (`observability.hide_prompts`):
the full `json_for_advice_generation` payload (weather, phenology, product
labels) never leaves the process as span data. Outputs — the generated
advice text — stay visible, since that is what you actually want to read
when debugging a bad advisory; `output/<date>/*.json` and
`_run_report.json` remain the authoritative operational artifacts, not
Phoenix.

### Known gap: embedding cost

`search_products`'s Weaviate `near_text` calls are embedded **inside the
Weaviate container** (`text2vec-google`), using the same `GEMINI_API_KEY` as
the two LLM calls, but that request never passes through this process's
Python `google-genai` client — so no OpenInference instrumentor can see it.
Phoenix's per-job cost therefore covers the two LLM calls only and slightly
undercounts total Gemini spend (1–2 `gemini-embedding-001` calls per job, a
small fraction of a cent at this volume). The `search_products` span still
records `query_count`, so this gap can be estimated later if it matters; it
is not worth a custom cost pipeline at this scale.

### Why SQLite / why not Langfuse or Phoenix Cloud

- **SQLite, not Postgres:** this is a single-machine, single-writer
  deployment; Phoenix officially supports SQLite on a mounted volume for
  exactly this case. Postgres would add a second container and volume for no
  benefit here (switch later via `PHOENIX_SQL_DATABASE_URL` if that changes).
- **Self-hosted, not Phoenix Cloud:** farm/field data stays in the
  deployment by design — nothing here talks to a hosted endpoint.
- **Phoenix, not Langfuse:** Langfuse's self-hosted stack needs Postgres
  *and* ClickHouse; Phoenix needs one container and ingests plain
  OTLP + OpenInference, so this pipeline isn't locked into either vendor.

## Pipeline stages

Run once per `(field, disease, anmod_id)` job by `pipeline/job.py::run_job`:

1. **Worklist** — supported diseases (`AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1`) → advice-enabled field models (`AI_agrosupport_agro_models.anmod_AI_advice = 1`) → crop from `AI_agrosupport_cmp_layers` + `AI_agrosupport_an_colture`; organic flag is `cmplay_imp == 3` (`pipeline/worklist.py`, run once per batch, not per job)
2. **Disease model** — per day over as_of±5, from the worklist's known `anmod_id`. Rows come from `AI_agrosupport_agro_model_tmp2` where `model_model` equals that id. Only those rows' `model_description` values and `model_value` are used. `disease_forecast` is a list or null. Each `INCUBAZPRIMARIA` index is kept (`INCUBAZPRIMARIA2_80.7`, or `INCUBAZPRIMARIA_40.3` when the description has no trailing index), using that index's highest `model_value`. `FASE5` is appended when present; `FASE2` is appended only when `FASE5` is absent. `predicted_incubations_management` (loaded from the previous `advisories.advice_input_json` for the same `pathogen_model_id`, then written back with this run) records each index's `start_date` and `infection_covered`. Today counts as provisional: a start locks only once the index appears on a day before `as_of`. A covered incubation alone is not risk.
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** — one query loads crop-protection sprays over `as_of-25 .. as_of` from `AI_agrosupport_agro_ril_operations*` with `rilop_operation = 10`. Days in `as_of-5 .. as_of` fill meteorological `applied_treatment`. If the latest spray in the full 25-day window still has a non-NULL `product_uses.treatment_interval_min_days` strictly greater than days elapsed (any tank-mix partner), query synthesis, prefilter, and vector search are skipped and `allowed_products` stays empty. `last_applied_treatment` still uses only `as_of-25 .. as_of-6`, and only when the recent window is empty.
7. **LLM** — synthesise 1–2 anonymised search queries from the subset
8. **Product prefilter** — legal catalog state, then 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). A product stays eligible only when `registration_number` matches `AI_agrosupport_products_catalog.prodgov_registr` and `prodgov_state` is a non-blank value other than `Revocato`; revoked rows remain in the index for interval checks and applied-treatment labels:

   | Rule | Predicates |
   |---|---|
   | `observation` true on `as_of` or the past 5 days | `systemicity` in `{systemic, locally systemic, mixed}` and `has_eradicant_action = 1` |
   | uncovered incubation on a day before `as_of` (`as_of-5` .. `as_of-1`) | `systemicity` in `{systemic, locally systemic, mixed}` and `has_curative_action = 1` and `has_eradicant_action = 0` |
   | `FASE2` anywhere and no `FASE5` anywhere | `systemicity` in `{contact, translaminar}` and `has_preventive_action = 1` and `has_curative_action = 0` and `has_eradicant_action = 0` |
   | `FASE5` in the past and on `as_of` or later | `systemicity == mixed` and `has_preventive_action = 1` and `has_curative_action = 1` and `has_eradicant_action = 0` |
   | `FASE5` on `as_of` or later only | same predicates as the `FASE2`-only rule |
   | `FASE5` in the past only | `systemicity` in `{contact, translaminar, mixed}` and `has_preventive_action = 1` and `has_curative_action = 0` and `has_eradicant_action = 0` |
   | none of the above | no FASE-specific cut; legal-state / organic / crop / disease gates only |

   This stage is skipped (along with query synthesis and vector search) when the latest spray's min treatment interval is still open; see stage 6.

9. **Vector search** — Weaviate `near_text` on `ProductProfile`, always filtered with `contains_any` on the SQL-allowlisted `product_id`s (`product_id` must use `tokenization: field`), then reranked locally with `reranker-transformers` (`ms-marco-MiniLM-L-6-v2`) on `situation_fit_summary`. One synthesized query retrieves 10 and keeps the top 4 in rerank order; two queries retrieve 6 each and keep the top 2 unique each (a product already kept for the first query is skipped so the second can fill its two slots from the rest of its reranked six)
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 `advisories` row for the same `pathogen_model_id`, and optional `last_applied_treatment` (spray date + products) when the recent window is empty but a spray exists in `as_of-25 .. as_of-6`, including when `allowed_products` is empty. Enrichment reuses the job's 25-day treatment history; it does not query operations again.
12. **Advice generation** — second LLM profile returns the structured advisory
13. **Persistence** — the advisory plus its collapsed input are written to `advisories` with `pathogen_model_id` = `anmod_id`, inside one transaction