diff --git a/README.md b/README.md index 515c6dd..8c63108 100644 --- a/README.md +++ b/README.md @@ -3,7 +3,7 @@ Daily agronomic advisory pipeline. For every field growing a configured crop, 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 six matching products from a local Weaviate +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 `advice` table. @@ -186,7 +186,7 @@ Per-job payload fields: | `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` | +| `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`, `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` | @@ -377,7 +377,7 @@ Run once per `(field, disease)` job by `pipeline/job.py::run_job`: | `FASE5` today or in the past only | `systemicity` in `{contact, mixed}` and `preventive_action = 1` and `curative_action = 0` and `eradicant_action = 0` | | none of the above | no FASE-specific cut; organic/crop/disease gates only | -9. **Vector search** — Weaviate `near_text` on `ProductProfile`, filtered server-side to the allowlisted `product_id`s +9. **Vector search** — Weaviate `near_text` on `ProductProfile`, filtered server-side to the allowlisted `product_id`s. One synthesized query retrieves the top 4; two queries retrieve the top 2 unique each (the second search over-fetches so overlaps can still fill two products) 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, 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 diff --git a/pipeline/stages/vector.py b/pipeline/stages/vector.py index cb4e109..4bcc16b 100644 --- a/pipeline/stages/vector.py +++ b/pipeline/stages/vector.py @@ -69,8 +69,10 @@ def search_products( Embed each query and retrieve the nearest ProductProfile objects, scoped server-side to the SQL-prefiltered allowlist. - One query → top 6 overall. - Two queries → top 3 per query (deduplicated preferring first occurrence). + One query → top 4 overall. + Two queries → top 2 unique per query (deduplicated preferring first + occurrence). Later queries over-fetch so duplicates of earlier hits + can be replaced from the same response without a second search. `client` is a shared, already-connected Weaviate client (see `pipeline.resources.Resources`); this function does not open or close a @@ -95,16 +97,16 @@ def search_products( from weaviate.classes.query import Filter, MetadataQuery allowlist = {c.product_id: c for c in candidates} - per_query_limit = 6 if len(queries) == 1 else 3 + per_query_limit = 4 if len(queries) == 1 else 2 query_filter = None - fetch_limit = per_query_limit + unfiltered_overfetch = False if len(allowlist) <= _MAX_FILTERED_IDS: query_filter = Filter.any_of( [Filter.by_property("product_id").equal(pid) for pid in allowlist] ) else: - fetch_limit = min(200, max(per_query_limit * 5, 15)) + unfiltered_overfetch = True logger.warning( "Allowlist has %d product(s), above the %d-id server-side filter " "threshold; falling back to a global over-fetch + Python filter " @@ -118,6 +120,9 @@ def search_products( seen_ids: set[str] = set() for query_index, query in enumerate(queries): + fetch_limit = per_query_limit * (query_index + 1) + if unfiltered_overfetch: + fetch_limit = min(200, max(fetch_limit, per_query_limit * 5, 15)) response = collection.query.near_text( query=query, limit=fetch_limit, diff --git a/tests/test_vector.py b/tests/test_vector.py new file mode 100644 index 0000000..077ad76 --- /dev/null +++ b/tests/test_vector.py @@ -0,0 +1,108 @@ +"""Unit tests for Weaviate product retrieval limits.""" + +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from typing import Any + +from pipeline.stages.products import ProductCandidate +from pipeline.stages.vector import search_products + + +def _candidate(product_id: str) -> ProductCandidate: + return ProductCandidate( + product_id=product_id, + product_name=product_id.upper(), + registration_number=f"reg-{product_id}", + ) + + +def _hit(product_id: str, distance: float = 0.1) -> SimpleNamespace: + return SimpleNamespace( + properties={"product_id": product_id, "product_name": product_id.upper()}, + metadata=SimpleNamespace(distance=distance), + ) + + +class FakeNearTextQuery: + def __init__(self, hits_by_query: dict[str, list[Any]]) -> None: + self.hits_by_query = hits_by_query + self.calls: list[dict[str, Any]] = [] + + def near_text( + self, + query: str, + limit: int, + filters: Any, + return_metadata: Any, + return_properties: list[str], + ) -> SimpleNamespace: + self.calls.append({"query": query, "limit": limit}) + hits = list(self.hits_by_query.get(query, [])) + return SimpleNamespace(objects=hits[:limit]) + + +class FakeClient: + def __init__(self, hits_by_query: dict[str, list[Any]]) -> None: + self.query = FakeNearTextQuery(hits_by_query) + self.collections = SimpleNamespace( + get=lambda _name: SimpleNamespace(query=self.query) + ) + + +class SearchProductsLimitTests(unittest.TestCase): + def test_one_query_keeps_top_four(self) -> None: + candidates = [_candidate(f"p{i}") for i in range(1, 7)] + hits = [_hit(f"p{i}", distance=0.1 * i) for i in range(1, 7)] + client = FakeClient({"q1": hits}) + + results = search_products(client, ["q1"], candidates) + + self.assertEqual([item.product_id for item in results], ["p1", "p2", "p3", "p4"]) + self.assertEqual([item.query_index for item in results], [0, 0, 0, 0]) + self.assertEqual(client.query.calls, [{"query": "q1", "limit": 4}]) + + def test_two_queries_keep_two_each(self) -> None: + candidates = [_candidate(f"p{i}") for i in range(1, 7)] + client = FakeClient( + { + "q1": [_hit("p1"), _hit("p2"), _hit("p3")], + "q2": [_hit("p4"), _hit("p5"), _hit("p6")], + } + ) + + results = search_products(client, ["q1", "q2"], candidates) + + self.assertEqual( + [(item.product_id, item.query_index) for item in results], + [("p1", 0), ("p2", 0), ("p4", 1), ("p5", 1)], + ) + self.assertEqual( + client.query.calls, + [{"query": "q1", "limit": 2}, {"query": "q2", "limit": 4}], + ) + + def test_overlapping_hits_keep_first_occurrence(self) -> None: + candidates = [_candidate(f"p{i}") for i in range(1, 6)] + client = FakeClient( + { + "q1": [_hit("p1"), _hit("p2"), _hit("p3")], + "q2": [_hit("p2"), _hit("p1"), _hit("p4"), _hit("p5")], + } + ) + + results = search_products(client, ["q1", "q2"], candidates) + + self.assertEqual( + [(item.product_id, item.query_index) for item in results], + [("p1", 0), ("p2", 0), ("p4", 1), ("p5", 1)], + ) + self.assertEqual( + client.query.calls, + [{"query": "q1", "limit": 2}, {"query": "q2", "limit": 4}], + ) + + +if __name__ == "__main__": + unittest.main()