added json

app.py

@@ -70,7 +70,7 @@ def format_candidates_for_llm(cands, budget_twd=None):
             "image_url": c.get("image_url"),
             "score": c.get("score"),
         })
-    return json.dumps(filtered, ensure_ascii=False, indent=2)
+    return json.dumps(filtered, ensure_ascii=False, indent=2), filtered
 
 DERMA_SAFETY = (
     "Safety notes: For broken/infected skin, pregnancy/lactation, infants, "
@@ -82,7 +82,7 @@ def recommend_products(query_text: str, budget_twd: int | None = None, k: int =
     cands = product_search(query_text, k=k)
 
     # 2) Build short grounded context
-    context = format_candidates_for_llm(cands, budget_twd=budget_twd)
+    context_json, _ = format_candidates_for_llm(cands, budget_twd=budget_twd)
 
     # 3) Ask your LLM to pick & explain (plug into your existing generation path)
     system = (
@@ -90,119 +90,34 @@ def recommend_products(query_text: str, budget_twd: int | None = None, k: int =
         "from the provided list. Include a one-line why-it-helps and a brief how-to-use. "
         "Respect budget and do not invent products."
     )
-    user = f"User need: {query_text}\nCandidate products (JSON array):\n{context}\n{DERMA_SAFETY}"
+    user = f"User need: {query_text}\nCandidate products (JSON array):\n{context_json}\n{DERMA_SAFETY}"
 
     # --- if you already have Qwen2-VL loaded as text generator, reuse it.
     # Example skeleton (pseudo—replace with your app’s generate() function):
     try:
         # Replace this with your actual text-generation helper:
-        answer = f"(LLM picks here)\n\nContext:\n{context}"
+        answer = f"(LLM picks here)\n\nContext:\n{context_json}"
     except Exception as e:
-        answer = f"❌ Generation error: {e}\n\nHere are candidates:\n{context}"
+        answer = f"❌ Generation error: {e}\n\nHere are candidates:\n{context_json}"
 
     return answer
 
 
-def _count_products_from_tagged_json(text: str) -> int | None:
-    start_tag = "<DERMACARE_PRODUCTS_JSON>"
-    end_tag = "</DERMACARE_PRODUCTS_JSON>"
-    start = text.find(start_tag)
-    end = text.find(end_tag)
-    if start == -1 or end == -1 or end <= start:
-        return None
-    json_str = text[start + len(start_tag):end]
-    try:
-        payload = json.loads(json_str)
-        products = payload.get("products", [])
-        if isinstance(products, list):
-            return len(products)
-    except Exception:
+def _parse_recommendation_json(raw: str):
+    if not raw:
         return None
-    return None
-
-
-def _has_valid_suggestions(raw_suggestions: str) -> bool:
-    cleaned = raw_suggestions.strip()
-    if not cleaned:
-        return False
-
-    lower = cleaned.lower()
-    normalized = " ".join(lower.split())
-    if "no relevant products" in normalized:
-        return False
-
-    count = _count_products_from_tagged_json(cleaned)
-    if count is not None:
-        return count > 0
-
-    return True
-
-# ---- JSON block helpers ----
-def _extract_products_json_block(text: str) -> str | None:
-    start_tag = "<DERMACARE_PRODUCTS_JSON>"
-    end_tag = "</DERMACARE_PRODUCTS_JSON>"
-    start = text.find(start_tag)
-    end = text.find(end_tag)
+    cleaned = raw.strip()
+    if cleaned.startswith("```"):
+        lines = [line for line in cleaned.splitlines() if not line.strip().startswith("```")]
+        cleaned = "\n".join(lines)
+    start = cleaned.find('{')
+    end = cleaned.rfind('}')
     if start == -1 or end == -1 or end <= start:
         return None
-    json_str = text[start + len(start_tag):end]
     try:
-        payload = json.loads(json_str)
-        products = payload.get("products", [])
-        if isinstance(products, list) and len(products) > 0:
-            return f"{start_tag}{json.dumps(payload, ensure_ascii=False)}{end_tag}"
+        return json.loads(cleaned[start:end + 1])
     except Exception:
         return None
-    return None
-
-
-def _filter_candidates_by_budget(cands: list[dict], budget_twd: int | None) -> list[dict]:
-    if not budget_twd:
-        return cands
-    filtered: list[dict] = []
-    for c in cands:
-        currency = c.get("price_currency")
-        value = c.get("price_value")
-        if currency == "TWD" and isinstance(value, (int, float)) and value is not None:
-            if value <= budget_twd:
-                filtered.append(c)
-        else:
-            filtered.append(c)
-    return filtered
-
-
-def _build_products_json_block_from_candidates(cands: list[dict], max_items: int = 3) -> str | None:
-    if not cands:
-        return None
-    ranked = sorted(cands, key=lambda x: x.get("_score", 0.0), reverse=True)
-    picked = ranked[:max_items]
-    products = []
-    for c in picked:
-        products.append({
-            "id": c.get("id"),
-            "brand": c.get("brand_en") or c.get("brand_zh"),
-            "name": c.get("product_name_en") or c.get("product_name_zh"),
-            "category": c.get("category_en") or c.get("category_zh"),
-            "price_value": c.get("price_value"),
-            "price_currency": c.get("price_currency"),
-            "why": None,
-            "how": None,
-            "url": c.get("source_url"),
-            "image_url": c.get("image_url"),
-        })
-    valid_count = sum(1 for p in products if p.get("name"))
-    if valid_count == 0:
-        return None
-    payload = {"version": 1, "products": products}
-    return f"<DERMACARE_PRODUCTS_JSON>{json.dumps(payload, ensure_ascii=False)}</DERMACARE_PRODUCTS_JSON>"
-
-
-def _ensure_products_json_block(suggestions: str, cands: list[dict], budget_twd: int | None) -> str | None:
-    existing = _extract_products_json_block(suggestions)
-    if existing:
-        return existing
-    filtered = _filter_candidates_by_budget(cands, budget_twd)
-    return _build_products_json_block_from_candidates(filtered)
 
 # ---- Inference on GPU (ZeroGPU pattern) ----
 @spaces.GPU(duration=120)
@@ -311,25 +226,98 @@ def pet_answer_with_recs(image, question, temperature=0.7, top_p=0.95, max_token
             top_p=0.95,
         )
     trimmed = [o[len(i):] for i, o in zip(inputs["input_ids"], out)]
-    suggestions = processor.batch_decode(trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False)[0]
+    raw_response = processor.batch_decode(
+        trimmed,
+        skip_special_tokens=True,
+        clean_up_tokenization_spaces=False,
+    )[0]
 
-    # Final combined message
-    safety = (
-        "Safety notes: For broken/infected skin, pregnancy/lactation, infants, "
-        "or if symptoms worsen—seek a qualified dermatologist. Patch-test first."
-    )
-    suggestions = suggestions.strip()
-    include_products = _has_valid_suggestions(suggestions)
+    rec_data = _parse_recommendation_json(raw_response)
 
     sections = [base.strip()]
-    if include_products:
-        sections.append(f"Suggested products:\n{suggestions}")
-        if "<DERMACARE_PRODUCTS_JSON>" not in suggestions:
-            json_block = _ensure_products_json_block(suggestions, cands, budget_twd)
-            if json_block:
-                sections.append(json_block)
-        sections.append(safety)
-    
+    suggestion_text = None
+    product_json_payload = None
+
+    if rec_data:
+        recommend_flag = rec_data.get("recommend")
+        if isinstance(recommend_flag, str):
+            recommend_flag = recommend_flag.strip().lower() in {"yes", "true", "1"}
+        elif isinstance(recommend_flag, (int, float)):
+            recommend_flag = bool(recommend_flag)
+
+        recs = []
+        for item in rec_data.get("recommendations", []):
+            if isinstance(item, dict) and item.get("id"):
+                recs.append(item)
+
+        if recommend_flag and recs:
+            suggestion_lines = ["### Suggested Products", ""]
+            products_payload = []
+
+            for idx, rec in enumerate(recs[:3], start=1):
+                pid = rec.get("id")
+                candidate = candidate_lookup.get(pid, {})
+
+                brand = (
+                    candidate.get("brand_en")
+                    or candidate.get("brand_zh")
+                    or rec.get("brand")
+                    or ""
+                )
+                name = (
+                    candidate.get("product_name_en")
+                    or candidate.get("product_name_zh")
+                    or rec.get("name")
+                    or f"Product {idx}"
+                )
+                category = (
+                    candidate.get("category_en")
+                    or candidate.get("category_zh")
+                    or rec.get("category")
+                    or None
+                )
+                price_value = candidate.get("price_value")
+                price_currency = candidate.get("price_currency")
+                why = rec.get("why") or "Supports the user’s concern."
+                how = rec.get("how") or "Use as directed on the product label."
+                url = candidate.get("source_url") or rec.get("url")
+                image_url = candidate.get("image_url") or rec.get("image_url")
+
+                suggestion_lines.extend([
+                    f"{idx}. **{name}**",
+                    f"- **Why it helps:** {why}",
+                    f"- **How to use:** {how}",
+                    "",
+                ])
+
+                products_payload.append({
+                    "id": pid,
+                    "brand": brand,
+                    "name": name,
+                    "category": category,
+                    "price_value": price_value,
+                    "price_currency": price_currency,
+                    "why": why,
+                    "how": how,
+                    "url": url,
+                    "image_url": image_url,
+                })
+
+            if products_payload:
+                suggestion_text = "\n".join(suggestion_lines).strip()
+                product_json_payload = json.dumps(
+                    {"version": 1, "products": products_payload},
+                    ensure_ascii=False,
+                )
+
+    if suggestion_text and product_json_payload:
+        sections.append(
+            "Suggested products:\n"
+            f"{suggestion_text}\n\n"
+            f"<DERMACARE_PRODUCTS_JSON>{product_json_payload}</DERMACARE_PRODUCTS_JSON>"
+        )
+        sections.append(DERMA_SAFETY)
+
     return "\n\n".join([s for s in sections if s])
 
 # ---- UI ----