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

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:

docker compose up -d

Configuration

• 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 — SQL Server, Weaviate, and API keys
• vocab/crops.yaml / vocab/diseases.yaml — Italian → English maps
• 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 (Vite → grapevine). The English disease name is mapped through 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:

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

.\.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 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

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:

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 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_ids (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