innercontext/backend/innercontext/api/product_llm_tools.py

133 lines
4.6 KiB
Python

from typing import Any
from google.genai import types as genai_types
from innercontext.models import Product
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 _extract_requested_product_ids(
args: dict[str, object], max_ids: int = 8
) -> list[str]:
raw_ids = args.get("product_ids")
if not isinstance(raw_ids, list):
return []
requested_ids: list[str] = []
seen: set[str] = set()
for raw_id in raw_ids:
if not isinstance(raw_id, str):
continue
if raw_id in seen:
continue
seen.add(raw_id)
requested_ids.append(raw_id)
if len(requested_ids) >= max_ids:
break
return requested_ids
def _build_compact_actives_payload(product: Product) -> list[dict[str, object]]:
payload: list[dict[str, object]] = []
for active in product.actives or []:
if isinstance(active, dict):
name = str(active.get("name") or "").strip()
if not name:
continue
item: dict[str, object] = {"name": name}
percent = active.get("percent")
if percent is not None:
item["percent"] = percent
functions = active.get("functions")
if isinstance(functions, list):
item["functions"] = [str(f) for f in functions[:4]]
strength_level = active.get("strength_level")
if strength_level is not None:
item["strength_level"] = str(strength_level)
payload.append(item)
continue
name = str(getattr(active, "name", "") or "").strip()
if not name:
continue
item = {"name": name}
percent = getattr(active, "percent", None)
if percent is not None:
item["percent"] = percent
functions = getattr(active, "functions", None)
if isinstance(functions, list):
item["functions"] = [_ev(f) for f in functions[:4]]
strength_level = getattr(active, "strength_level", None)
if strength_level is not None:
item["strength_level"] = _ev(strength_level)
payload.append(item)
return payload[:24]
def _map_product_details(product: Product, pid: str) -> dict[str, object]:
ctx = product.to_llm_context()
inci = product.inci or []
return {
"id": pid,
"name": product.name,
"brand": product.brand,
"category": ctx.get("category"),
"recommended_time": ctx.get("recommended_time"),
"leave_on": product.leave_on,
"targets": ctx.get("targets") or [],
"effect_profile": ctx.get("effect_profile") or {},
"inci": [str(i)[:120] for i in inci[:128]],
"actives": _build_compact_actives_payload(product),
"context_rules": ctx.get("context_rules") or {},
"safety": ctx.get("safety") or {},
"min_interval_hours": ctx.get("min_interval_hours"),
"max_frequency_per_week": ctx.get("max_frequency_per_week"),
}
def build_product_details_tool_handler(products: list[Product]):
available_by_id = {str(p.id): p for p in products}
def _handler(args: dict[str, Any]) -> dict[str, object]:
requested_ids = _extract_requested_product_ids(args)
products_payload = []
for pid in requested_ids:
product = available_by_id.get(pid)
if product is None:
continue
products_payload.append(_map_product_details(product, pid))
return {"products": products_payload}
return _handler
PRODUCT_DETAILS_FUNCTION_DECLARATION = genai_types.FunctionDeclaration(
name="get_product_details",
description=(
"Use this to fetch canonical product data before making clinical/safety decisions. "
"Call it when you need to verify ingredient conflicts, irritation risk, barrier compatibility, "
"or usage cadence. Returns per-product fields: id, name, brand, category, recommended_time, "
"leave_on, targets, effect_profile, inci, actives, context_rules, safety, "
"min_interval_hours, and max_frequency_per_week."
),
parameters=genai_types.Schema(
type=genai_types.Type.OBJECT,
properties={
"product_ids": genai_types.Schema(
type=genai_types.Type.ARRAY,
items=genai_types.Schema(type=genai_types.Type.STRING),
description="Product UUIDs from the provided product list.",
)
},
required=["product_ids"],
),
)