Update product retrieval limits in README and vector search logic

- Adjusted the maximum number of recommended products from 6 to 4 in README.md to reflect changes in the product retrieval process.
- Modified vector.py to limit the number of products retrieved per query to 4 for single queries and 2 for multiple queries, ensuring better deduplication and efficiency in product searches.

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

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,

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