refactor(api): remove shopping assistant logic from mcp_server
This commit is contained in:
parent
067e460dd2
commit
78df7322a9
3 changed files with 28 additions and 170 deletions
|
|
@ -147,15 +147,15 @@ class ProductParseResponse(SQLModel):
|
|||
|
||||
class AIActiveIngredient(ActiveIngredient):
|
||||
# Gemini API rejects int-enum values in response_schema; override with plain int.
|
||||
strength_level: Optional[int] = None
|
||||
irritation_potential: Optional[int] = None
|
||||
strength_level: Optional[int] = None # type: ignore[assignment]
|
||||
irritation_potential: Optional[int] = None # type: ignore[assignment]
|
||||
|
||||
|
||||
class ProductParseLLMResponse(ProductParseResponse):
|
||||
# Gemini response schema currently requires enum values to be strings.
|
||||
# Strength fields are numeric in our domain (1-3), so keep them as ints here
|
||||
# and convert via ProductParseResponse validation afterward.
|
||||
actives: Optional[list[AIActiveIngredient]] = None
|
||||
actives: Optional[list[AIActiveIngredient]] = None # type: ignore[assignment]
|
||||
|
||||
|
||||
class InventoryCreate(SQLModel):
|
||||
|
|
@ -571,9 +571,30 @@ def _build_shopping_context(session: Session) -> str:
|
|||
active_inv = [i for i in inv_by_product.get(p.id, []) if i.finished_at is None]
|
||||
has_stock = len(active_inv) > 0 # any unfinished inventory = in stock
|
||||
stock = "✓" if has_stock else "✗"
|
||||
|
||||
actives = []
|
||||
for a in p.actives or []:
|
||||
name = a.get("name") if isinstance(a, dict) else getattr(a, "name", None)
|
||||
if name:
|
||||
actives.append(name)
|
||||
actives_str = f", actives: {actives}" if actives else ""
|
||||
|
||||
ep = p.product_effect_profile
|
||||
if isinstance(ep, dict):
|
||||
effects = {k.replace("_strength", ""): v for k, v in ep.items() if v >= 3}
|
||||
else:
|
||||
effects = {
|
||||
k.replace("_strength", ""): v
|
||||
for k, v in ep.model_dump().items()
|
||||
if v >= 3
|
||||
}
|
||||
effects_str = f", effects: {effects}" if effects else ""
|
||||
|
||||
targets = [_ev(t) for t in (p.targets or [])]
|
||||
|
||||
products_lines.append(
|
||||
f" [{stock}] {p.name} ({p.brand or ''}) - {_ev(p.category)}, "
|
||||
f"targets: {p.targets or []}"
|
||||
f"targets: {targets}{actives_str}{effects_str}"
|
||||
)
|
||||
|
||||
return "\n".join(skin_lines) + "\n\n" + "\n".join(products_lines)
|
||||
|
|
@ -596,7 +617,8 @@ ZASADY:
|
|||
5. Sugeruj realistyczną częstotliwość użycia (dzienna, 2-3x tygodniowo, etc.)
|
||||
6. Zachowaj kolejność warstw: cleanse → toner → serum → moisturizer → SPF
|
||||
7. Jeśli użytkownik ma uszkodzoną barierę, unikaj silnych eksfoliantów i retinoidów
|
||||
8. Odpowiadaj w języku polskim
|
||||
8. Zwracaj uwagę na ewentualne konflikty polecanych składników z tymi, które użytkownik już posiada (np. nie polecaj peptydów miedziowych jeśli użytkownik nadużywa kwasów)
|
||||
9. Odpowiadaj w języku polskim
|
||||
|
||||
Format odpowiedzi - zwróć wyłącznie JSON zgodny z podanym schematem."""
|
||||
|
||||
|
|
|
|||
|
|
@ -167,9 +167,7 @@ async def analyze_skin_photos(
|
|||
)
|
||||
)
|
||||
|
||||
image_summary = (
|
||||
f"{len(photos)} image(s): {', '.join((p.content_type or 'unknown') for p in photos)}"
|
||||
)
|
||||
image_summary = f"{len(photos)} image(s): {', '.join((p.content_type or 'unknown') for p in photos)}"
|
||||
response = call_gemini(
|
||||
endpoint="skincare/analyze-photos",
|
||||
contents=parts,
|
||||
|
|
|
|||
|
|
@ -1,20 +1,16 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import date, datetime, timedelta
|
||||
from typing import Optional
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
from fastapi import HTTPException
|
||||
from fastmcp import FastMCP
|
||||
from google.genai import types as genai_types
|
||||
from pydantic import BaseModel
|
||||
from sqlmodel import Session, col, select
|
||||
|
||||
from db import engine
|
||||
from innercontext.llm import call_gemini
|
||||
from innercontext.models import (
|
||||
GroomingSchedule,
|
||||
LabResult,
|
||||
|
|
@ -32,21 +28,6 @@ from innercontext.models import (
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class ProductSuggestionOut(BaseModel):
|
||||
category: str
|
||||
product_type: str
|
||||
key_ingredients: list[str]
|
||||
target_concerns: list[str]
|
||||
why_needed: str
|
||||
recommended_time: str
|
||||
frequency: str
|
||||
|
||||
|
||||
class ShoppingSuggestionsOut(BaseModel):
|
||||
suggestions: list[ProductSuggestionOut]
|
||||
reasoning: str
|
||||
|
||||
|
||||
class MarketProduct(BaseModel):
|
||||
name: str
|
||||
brand: str
|
||||
|
|
@ -477,149 +458,6 @@ def get_lab_results_for_test(test_code: str) -> list[dict]:
|
|||
# ── Shopping assistant ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _ev(v: object) -> str:
|
||||
if v is None:
|
||||
return ""
|
||||
value = getattr(v, "value", None)
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
return str(v)
|
||||
|
||||
|
||||
def _build_shopping_context(session: Session) -> tuple[str, list[dict]]:
|
||||
snapshot = session.exec(
|
||||
select(SkinConditionSnapshot).order_by(
|
||||
col(SkinConditionSnapshot.snapshot_date).desc()
|
||||
)
|
||||
).first()
|
||||
|
||||
skin_lines = ["STAN SKÓRY:"]
|
||||
if snapshot:
|
||||
skin_lines.append(f" Data: {snapshot.snapshot_date}")
|
||||
skin_lines.append(f" Ogólny stan: {_ev(snapshot.overall_state)}")
|
||||
skin_lines.append(f" Typ skóry: {_ev(snapshot.skin_type)}")
|
||||
skin_lines.append(f" Nawilżenie: {snapshot.hydration_level}/5")
|
||||
skin_lines.append(f" Wrażliwość: {snapshot.sensitivity_level}/5")
|
||||
skin_lines.append(f" Bariera: {_ev(snapshot.barrier_state)}")
|
||||
concerns = ", ".join(_ev(c) for c in (snapshot.active_concerns or []))
|
||||
skin_lines.append(f" Aktywne problemy: {concerns or 'brak'}")
|
||||
if snapshot.priorities:
|
||||
skin_lines.append(f" Priorytety: {', '.join(snapshot.priorities)}")
|
||||
else:
|
||||
skin_lines.append(" (brak danych)")
|
||||
|
||||
stmt = select(Product).where(col(Product.is_tool).is_(False))
|
||||
products = session.exec(stmt).all()
|
||||
|
||||
product_ids = [p.id for p in products]
|
||||
inventory_rows = (
|
||||
session.exec(
|
||||
select(ProductInventory).where(
|
||||
col(ProductInventory.product_id).in_(product_ids)
|
||||
)
|
||||
).all()
|
||||
if product_ids
|
||||
else []
|
||||
)
|
||||
inv_by_product: dict[UUID, list[ProductInventory]] = {}
|
||||
for inv in inventory_rows:
|
||||
inv_by_product.setdefault(inv.product_id, []).append(inv)
|
||||
|
||||
products_data = []
|
||||
for p in products:
|
||||
if p.is_medication:
|
||||
continue
|
||||
active_inv = [i for i in inv_by_product.get(p.id, []) if i.finished_at is None]
|
||||
has_stock = any(i.is_opened and i.finished_at is None for i in active_inv)
|
||||
products_data.append(
|
||||
{
|
||||
"id": str(p.id),
|
||||
"name": p.name,
|
||||
"brand": p.brand or "",
|
||||
"category": _ev(p.category),
|
||||
"targets": p.targets or [],
|
||||
"actives": [
|
||||
{
|
||||
"name": a.get("name") if isinstance(a, dict) else a.name,
|
||||
"percent": a.get("percent") if isinstance(a, dict) else None,
|
||||
}
|
||||
for a in (p.actives or [])
|
||||
],
|
||||
"has_inventory": has_stock,
|
||||
}
|
||||
)
|
||||
|
||||
products_lines = ["POSIADANE PRODUKTY:"]
|
||||
for p in products_data:
|
||||
stock = "✓" if p["has_inventory"] else "✗"
|
||||
products_lines.append(
|
||||
f" [{stock}] {p['name']} ({p['brand']}) - {p['category']}, "
|
||||
f"targets: {p['targets']}"
|
||||
)
|
||||
|
||||
return "\n".join(skin_lines) + "\n\n" + "\n".join(products_lines), products_data
|
||||
|
||||
|
||||
_SHOPPING_SYSTEM_PROMPT = """Jesteś asystentem zakupowym w dziedzinie pielęgnacji skóry.
|
||||
Twoim zadaniem jest przeanalizować stan skóry użytkownika oraz produkty, które już posiada,
|
||||
a następnie zasugerować TYPY produktów (bez marek), które mogłyby uzupełnić ich rutynę.
|
||||
|
||||
ZASADY:
|
||||
0. Sugeruj tylko wtedy, gdy jest realna potrzeba - nie zwracaj stałej liczby produktów
|
||||
1. Sugeruj TYLKO typy produktów, NIGDY konkretne marki (np. "Salicylic Acid 2% Masque", nie "La Roche-Posay")
|
||||
2. Koncentruj się na produktach, których użytkownik NIE posiada w swoim inventarzu
|
||||
3. Bierz pod uwagę aktywne problemy skóry (acne, hyperpigmentacja, aging, etc.)
|
||||
4. Sugeruj realistyczną częstotliwość użycia (dzienna, 2-3x tygodniowo, etc.)
|
||||
5. Zachowaj kolejność warstw: cleanse → toner → serum → moisturizer → SPF
|
||||
6. Jeśli użytkownik ma uszkodzoną barierę, unikaj silnych eksfoliantów i retinoidów
|
||||
7. Odpowiadaj w języku polskim
|
||||
|
||||
Format odpowiedzi - zwróć wyłącznie JSON zgodny z podanym schematem."""
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def get_shopping_suggestions() -> dict:
|
||||
"""Analyze skin condition and inventory to suggest product types that could fill gaps.
|
||||
Returns generic product suggestions (e.g., 'Salicylic Acid 2% Masque'), not specific brands.
|
||||
"""
|
||||
with Session(engine) as session:
|
||||
context, products = _build_shopping_context(session)
|
||||
|
||||
prompt = (
|
||||
f"Na podstawie poniższych danych przeanalizuj, jakie TYPY produktów "
|
||||
f"mogłyby uzupełnić rutynę pielęgnacyjną użytkownika.\n\n"
|
||||
f"{context}\n\n"
|
||||
f"Zwróć wyłącznie JSON zgodny ze schematem."
|
||||
)
|
||||
|
||||
response = call_gemini(
|
||||
endpoint="shopping/suggest",
|
||||
contents=prompt,
|
||||
config=genai_types.GenerateContentConfig(
|
||||
system_instruction=_SHOPPING_SYSTEM_PROMPT,
|
||||
response_mime_type="application/json",
|
||||
response_schema=ShoppingSuggestionsOut,
|
||||
max_output_tokens=4096,
|
||||
temperature=0.4,
|
||||
),
|
||||
user_input=prompt,
|
||||
)
|
||||
|
||||
raw = response.text
|
||||
if not raw:
|
||||
raise HTTPException(status_code=502, detail="LLM returned empty response")
|
||||
|
||||
try:
|
||||
parsed = json.loads(raw)
|
||||
except json.JSONDecodeError as e:
|
||||
raise HTTPException(status_code=502, detail=f"Invalid JSON from LLM: {e}")
|
||||
|
||||
return {
|
||||
"suggestions": parsed.get("suggestions", []),
|
||||
"reasoning": parsed.get("reasoning", ""),
|
||||
}
|
||||
|
||||
|
||||
def _call_market_service(
|
||||
query: str, stores: Optional[list[str]] = None
|
||||
) -> list[MarketProduct]:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue