d42fe18a52
- Updated `config.yaml` to change the advice storage table from `advice` to `advisories` and adjusted the `thinking_budget` from 11264 to 9216. - Modified `README.md` to reflect the new `advisories` table in the advisory generation process. - Refactored SQL queries in `batch.py`, `job.py`, and `advice_context.py` to align with the new table structure. - Enhanced the advice generation logic in `advice.py` to accommodate changes in the data model, including updates to the JSON structure for advice output. - Updated relevant functions and documentation to ensure consistency across the codebase regarding the new advisory schema.
394 lines
14 KiB
Python
394 lines
14 KiB
Python
"""Run the full single-field pipeline for one (field, crop, disease) job.
|
|
|
|
This is the per-job unit of work the batch orchestrator (`pipeline.batch`)
|
|
dispatches to its thread pool. It is the same sequence the original
|
|
single-field CLI ran (resolve disease model -> weather/phenology/treatments ->
|
|
query synthesis -> product prefilter -> vector search -> advice generation ->
|
|
persistence), restructured to take a `worklist.Job` (crop, organic flag,
|
|
station, and anmod_id already resolved by the worklist join) plus shared
|
|
`Resources`, instead of reading `field_id` / `disease_name` from config and
|
|
opening its own SQL/Weaviate connections.
|
|
|
|
`run_job` never raises: any failure — from either this module or anything it
|
|
calls — comes back as a `failed` JobResult, so one field's bad data or a
|
|
transient API error cannot stop the batch from processing every other field.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from pipeline.errors import PipelineError
|
|
from pipeline.observability import job_span
|
|
from pipeline.prompts import pair_slug
|
|
from pipeline.resources import Resources
|
|
from pipeline.retry import call_with_sql_retry
|
|
from pipeline.stages.advice import generate_advice, insert_advice
|
|
from pipeline.stages.advice_context import (
|
|
collapse_product_json,
|
|
enrich_advice_json,
|
|
load_last_advice,
|
|
load_previous_incubation_register,
|
|
)
|
|
from pipeline.stages.assemble import (
|
|
assemble_json_for_advice_generation,
|
|
build_json_for_query_synthesis,
|
|
)
|
|
from pipeline.stages.disease import resolve_disease_forecast_by_anmod, select_prefilter_rule
|
|
from pipeline.stages.incubations import (
|
|
covered_by_key,
|
|
first_future_disease_date,
|
|
forecast_has_risk,
|
|
incubation_start_covered,
|
|
parse_register,
|
|
reconcile_register,
|
|
register_payload,
|
|
)
|
|
from pipeline.stages.llm import synthesize_queries
|
|
from pipeline.stages.observation import load_observation
|
|
from pipeline.stages.phenology import load_phenology
|
|
from pipeline.stages.products import last_treatment_interval_open, prefilter_products
|
|
from pipeline.stages.treatments import load_treatment_history
|
|
from pipeline.stages.vector import RecommendedProduct
|
|
from pipeline.stages.weather import load_weather
|
|
from pipeline.worklist import Job
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class JobResult:
|
|
"""Outcome of one job. Constructed either by `run_job` or, for jobs the
|
|
batch orchestrator decides not to run at all, directly by `pipeline.batch`
|
|
(`skipped_existing`, `skipped_deadline`)."""
|
|
|
|
job: Job
|
|
status: str # "ok" | "no_risk" | "dry_run" | "failed" | "skipped_existing" | "skipped_deadline"
|
|
duration_s: float = 0.0
|
|
error_type: str | None = None
|
|
error_message: str | None = None
|
|
recommended_count: int = 0
|
|
allowed_products_empty: bool = False
|
|
message: str | None = None
|
|
payload: dict[str, Any] | None = None
|
|
|
|
|
|
def json_default(obj: Any) -> Any:
|
|
"""`json.dump(..., default=...)` hook for the `date`/`Decimal` values in a job payload."""
|
|
if isinstance(obj, date):
|
|
return obj.isoformat()
|
|
if isinstance(obj, Decimal):
|
|
return float(obj)
|
|
raise TypeError(f"Object of type {type(obj)!r} is not JSON serializable")
|
|
|
|
|
|
def output_stem(job: Job, disease_english: str) -> str:
|
|
"""File name stem (no extension) for this job's per-field output JSON."""
|
|
return f"field{job.field_id}__{pair_slug(job.crop_english, disease_english)}"
|
|
|
|
|
|
def run_job(resources: Resources, job: Job, as_of: date, dry_run: bool = False) -> JobResult:
|
|
"""
|
|
Run the full pipeline for one job.
|
|
|
|
Wrapped in `job_span`, the one root span per `(field, crop, disease,
|
|
as_of)` job (a no-op when observability is disabled/unreachable -- see
|
|
`pipeline/observability.py`). The span records the final `status` (and,
|
|
on failure, the error type) but never changes the fault-isolation
|
|
behaviour below: a failure always comes back as a `failed` JobResult,
|
|
tracing or no tracing.
|
|
"""
|
|
started = time.monotonic()
|
|
with job_span(job, as_of, dry_run) as span:
|
|
try:
|
|
payload = _run_job_inner(resources, job, as_of, dry_run)
|
|
except BaseException as exc: # noqa: BLE001 - fault isolation is the whole point of this wrapper
|
|
logger.exception(
|
|
"Job failed: field=%s crop=%s disease=%s",
|
|
job.field_id,
|
|
job.crop_english,
|
|
job.disease_english,
|
|
)
|
|
span.set_attribute("job.status", "failed")
|
|
span.set_attribute("job.error_type", type(exc).__name__)
|
|
return JobResult(
|
|
job=job,
|
|
status="failed",
|
|
duration_s=time.monotonic() - started,
|
|
error_type=type(exc).__name__,
|
|
error_message=str(exc),
|
|
)
|
|
|
|
status = payload["status"]
|
|
allowed_products_empty = False
|
|
if status in ("ok", "no_risk"):
|
|
allowed = (payload.get("json_for_advice_generation") or {}).get("allowed_products")
|
|
allowed_products_empty = not allowed
|
|
|
|
span.set_attribute("job.status", status)
|
|
span.set_attribute(
|
|
"job.recommended_count", len(payload.get("recommended_products") or [])
|
|
)
|
|
span.set_output(payload.get("message") or "")
|
|
|
|
return JobResult(
|
|
job=job,
|
|
status=status,
|
|
duration_s=time.monotonic() - started,
|
|
recommended_count=len(payload.get("recommended_products") or []),
|
|
allowed_products_empty=allowed_products_empty,
|
|
message=payload.get("message"),
|
|
payload=payload,
|
|
)
|
|
|
|
|
|
def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -> dict[str, Any]:
|
|
settings = resources.settings
|
|
conn = resources.sql_connection()
|
|
|
|
with resources.sql_sem:
|
|
disease = resolve_disease_forecast_by_anmod(
|
|
conn,
|
|
anmod_id=job.anmod_id,
|
|
disease_name=job.model_name,
|
|
disease_english=job.disease_english,
|
|
as_of=as_of,
|
|
fallback_station=job.station,
|
|
)
|
|
|
|
if disease.station is None:
|
|
raise PipelineError(
|
|
f"Field {job.field_id}: unable to determine weather station "
|
|
"(model_station / cmplay_station both missing)."
|
|
)
|
|
|
|
weather = load_weather(conn, disease.station, as_of)
|
|
phenology = load_phenology(conn, job.field_id, as_of)
|
|
treatment_history = load_treatment_history(conn, job.field_id, as_of)
|
|
observation_by_day = load_observation(conn, job.field_id, as_of)
|
|
product_index = resources.product_index
|
|
if product_index is None:
|
|
raise RuntimeError("load_product_index() must run before run_job().")
|
|
previous_register = parse_register(
|
|
load_previous_incubation_register(
|
|
conn, job.field_id, disease.disease_english, as_of
|
|
)
|
|
)
|
|
interval_open = last_treatment_interval_open(
|
|
conn,
|
|
product_index,
|
|
crop_english=job.crop_english,
|
|
disease_english=disease.disease_english,
|
|
latest_spray=treatment_history.latest(),
|
|
as_of=as_of,
|
|
)
|
|
|
|
def _covers(start: date):
|
|
return incubation_start_covered(
|
|
conn,
|
|
product_index,
|
|
crop_english=job.crop_english,
|
|
disease_english=disease.disease_english,
|
|
start_date=start,
|
|
treatment_history=treatment_history,
|
|
)
|
|
|
|
incubation_register = reconcile_register(
|
|
previous_register,
|
|
disease.forecasts_by_day,
|
|
as_of,
|
|
_covers,
|
|
)
|
|
|
|
# Risk is an uncovered incubation, a FASE2/FASE5, or a true observation.
|
|
# A window of only covered incubations is not risk. Advice is still generated.
|
|
has_risk = forecast_has_risk(disease.forecasts_by_day, incubation_register) or any(
|
|
observation_by_day.values()
|
|
)
|
|
future_disease = first_future_disease_date(
|
|
disease.forecasts_by_day, incubation_register, as_of
|
|
)
|
|
|
|
logger.info(
|
|
"Field %s (%s/%s): station=%s has_risk=%s first_future=%s",
|
|
job.field_id,
|
|
job.crop_english,
|
|
job.disease_english,
|
|
disease.station,
|
|
has_risk,
|
|
future_disease,
|
|
)
|
|
|
|
json_for_advice_generation = assemble_json_for_advice_generation(
|
|
as_of,
|
|
weather,
|
|
phenology,
|
|
disease.forecasts_by_day,
|
|
treatment_history.recent(),
|
|
observation_by_day,
|
|
register_payload(incubation_register),
|
|
)
|
|
json_for_query_synthesis = build_json_for_query_synthesis(
|
|
as_of,
|
|
job.crop_english,
|
|
disease.disease_english,
|
|
weather,
|
|
phenology,
|
|
future_disease,
|
|
)
|
|
|
|
payload: dict[str, Any] = {
|
|
"run": {
|
|
"as_of": as_of.isoformat(),
|
|
"field_id": job.field_id,
|
|
"crop": job.crop_english,
|
|
"crop_italian": job.crop_italian,
|
|
"disease": disease.disease_english,
|
|
"disease_raw": disease.disease_raw,
|
|
"organic": job.organic,
|
|
"station": disease.station,
|
|
"anmod_id": disease.anmod_id,
|
|
"has_risk": has_risk,
|
|
"first_future_disease_date": (
|
|
future_disease.isoformat() if future_disease else None
|
|
),
|
|
"prefilter_rule": None,
|
|
"llm_provider": settings.llm_query_synthesis.provider,
|
|
},
|
|
"json_for_advice_generation": json_for_advice_generation,
|
|
"json_for_query_synthesis": json_for_query_synthesis,
|
|
"generated_queries": [],
|
|
"candidate_count": 0,
|
|
"recommended_products": [],
|
|
"last_advice": None,
|
|
"advice": None,
|
|
"advice_input_json": None,
|
|
"advice_inserted": False,
|
|
"status": "ok",
|
|
"message": None,
|
|
"output_stem": output_stem(job, disease.disease_english),
|
|
}
|
|
|
|
if dry_run:
|
|
payload["status"] = "dry_run"
|
|
payload["message"] = (
|
|
"Dry run: skipped LLM, vector search, advice generation, and DB write."
|
|
)
|
|
return payload
|
|
|
|
recommended: list[RecommendedProduct] = []
|
|
if has_risk and interval_open:
|
|
logger.info(
|
|
"Field %s: last treatment interval still open; product search "
|
|
"skipped, advice generated without allowed products.",
|
|
job.field_id,
|
|
)
|
|
elif has_risk:
|
|
query_system_prompt = resources.prompts.query_prompt(job.crop_english, disease.disease_english)
|
|
queries = synthesize_queries(
|
|
settings,
|
|
query_system_prompt,
|
|
json_for_query_synthesis,
|
|
llm_call=resources.call_llm,
|
|
)
|
|
payload["generated_queries"] = queries
|
|
|
|
prefilter_rule = select_prefilter_rule(
|
|
disease.forecasts_by_day,
|
|
observation_by_day,
|
|
as_of,
|
|
covered_by_key(incubation_register),
|
|
)
|
|
payload["run"]["prefilter_rule"] = prefilter_rule
|
|
candidates = prefilter_products(
|
|
product_index,
|
|
crop_english=job.crop_english,
|
|
disease_english=disease.disease_english,
|
|
organic=job.organic,
|
|
rule=prefilter_rule,
|
|
)
|
|
payload["candidate_count"] = len(candidates)
|
|
|
|
recommended = resources.search_products(queries, candidates)
|
|
payload["recommended_products"] = [
|
|
{
|
|
"product_name": item.product_name,
|
|
"registration_number": item.registration_number,
|
|
"distance": item.distance,
|
|
"query_index": item.query_index,
|
|
"product_id": item.product_id,
|
|
}
|
|
for item in recommended
|
|
]
|
|
logger.info(
|
|
"Field %s: recommended products: %s",
|
|
job.field_id,
|
|
[p["product_name"] for p in payload["recommended_products"]],
|
|
)
|
|
else:
|
|
payload["status"] = "no_risk"
|
|
logger.warning(
|
|
"Field %s: no uncovered incubation, FASE2, or FASE5 in the 11-day "
|
|
"window, and no true observation in the past 5 days or today; "
|
|
"product search skipped, advice generated without allowed "
|
|
"products.",
|
|
job.field_id,
|
|
)
|
|
|
|
builder = resources.product_json_builder(job.crop_english, disease.disease_english)
|
|
with resources.sql_sem:
|
|
last_summary, last_treatments, last_date = load_last_advice(
|
|
conn, job.field_id, disease.disease_english, as_of
|
|
)
|
|
enrich_advice_json(
|
|
json_for_advice_generation,
|
|
conn=conn,
|
|
builder=builder,
|
|
recommended=recommended,
|
|
last_advice_summary=last_summary,
|
|
last_advice_treatments=last_treatments,
|
|
last_advice_date=last_date,
|
|
treatment_history=treatment_history,
|
|
)
|
|
payload["last_advice"] = json_for_advice_generation["last_advice"]
|
|
|
|
advice_system_prompt, advice_user_prompt = resources.prompts.advice_prompts(
|
|
job.crop_english, disease.disease_english
|
|
)
|
|
result = generate_advice(
|
|
settings,
|
|
advice_system_prompt,
|
|
advice_user_prompt,
|
|
json_for_advice_generation,
|
|
as_of,
|
|
llm_call=resources.call_llm,
|
|
)
|
|
advice_input_json = collapse_product_json(json_for_advice_generation)
|
|
|
|
def _insert() -> None:
|
|
with resources.sql_sem:
|
|
insert_advice(
|
|
conn,
|
|
issuance_date=as_of,
|
|
field_id=job.field_id,
|
|
disease_english=disease.disease_english,
|
|
advice_input_json=advice_input_json,
|
|
result=result,
|
|
)
|
|
|
|
call_with_sql_retry(settings.retry, _insert)
|
|
|
|
payload["advice"] = result.as_dict()
|
|
payload["advice_input_json"] = advice_input_json
|
|
payload["advice_inserted"] = True
|
|
payload["message"] = (
|
|
f"Returned {len(recommended)} recommended product(s); "
|
|
f"advice stored with is_treatment_recommended={int(result.is_treatment_recommended)}."
|
|
)
|
|
return payload
|