feat(api): expand routines tool-calling to reduce prompt load

Keep the /routines/suggest base context lean by sending only active names and fetching detailed safety, actives, usage notes, and INCI on demand. Add a conservative fallback when tool roundtrip limits are hit to preserve safe outputs instead of failing the request.
This commit is contained in:
Piotr Oleszczyk 2026-03-04 11:52:07 +01:00
parent cfd2485b7e
commit 558708653c
3 changed files with 329 additions and 47 deletions

View file

@ -345,8 +345,9 @@ def _build_products_context(
f" leave_on={ctx.get('leave_on', '')}" f" leave_on={ctx.get('leave_on', '')}"
f" targets={ctx.get('targets', [])}" f" targets={ctx.get('targets', [])}"
) )
if "actives" in ctx: active_names = _extract_active_names(p)
entry += f" actives={ctx['actives']}" if active_names:
entry += f" actives={active_names}"
active_inventory = [inv for inv in p.inventory if inv.finished_at is None] active_inventory = [inv for inv in p.inventory if inv.finished_at is None]
open_inventory = [inv for inv in active_inventory if inv.is_opened] open_inventory = [inv for inv in active_inventory if inv.is_opened]
@ -394,12 +395,6 @@ def _build_products_context(
entry += f" max_frequency_per_week={ctx['max_frequency_per_week']}" entry += f" max_frequency_per_week={ctx['max_frequency_per_week']}"
usage_count = recent_usage_counts.get(p.id, 0) usage_count = recent_usage_counts.get(p.id, 0)
entry += f" used_in_last_7_days={usage_count}" entry += f" used_in_last_7_days={usage_count}"
usage_notes = ctx.get("usage_notes")
if usage_notes:
compact_notes = " ".join(str(usage_notes).split())
if len(compact_notes) > 260:
compact_notes = compact_notes[:257] + "..."
entry += f' usage_notes="{compact_notes}"'
lines.append(entry) lines.append(entry)
return "\n".join(lines) + "\n" return "\n".join(lines) + "\n"
@ -422,13 +417,126 @@ def _get_available_products(
def _build_inci_tool_handler( def _build_inci_tool_handler(
products: list[Product], products: list[Product],
):
def _mapper(product: Product, pid: str) -> dict[str, object]:
inci = product.inci or []
compact_inci = [str(i)[:120] for i in inci[:128]]
return {
"id": pid,
"name": product.name,
"inci": compact_inci,
}
return _build_product_details_tool_handler(products, mapper=_mapper)
def _build_actives_tool_handler(
products: list[Product],
):
def _mapper(product: Product, pid: str) -> dict[str, object]:
actives_payload = []
for a in product.actives or []:
if isinstance(a, dict):
active_name = str(a.get("name") or "").strip()
if not active_name:
continue
item = {"name": active_name}
percent = a.get("percent")
if percent is not None:
item["percent"] = percent
functions = a.get("functions")
if isinstance(functions, list):
item["functions"] = [str(f) for f in functions[:4]]
strength_level = a.get("strength_level")
if strength_level is not None:
item["strength_level"] = str(strength_level)
actives_payload.append(item)
continue
active_name = str(getattr(a, "name", "") or "").strip()
if not active_name:
continue
item = {"name": active_name}
percent = getattr(a, "percent", None)
if percent is not None:
item["percent"] = percent
functions = getattr(a, "functions", None)
if isinstance(functions, list):
item["functions"] = [_ev(f) for f in functions[:4]]
strength_level = getattr(a, "strength_level", None)
if strength_level is not None:
item["strength_level"] = _ev(strength_level)
actives_payload.append(item)
return {
"id": pid,
"name": product.name,
"actives": actives_payload[:24],
}
return _build_product_details_tool_handler(products, mapper=_mapper)
def _build_usage_notes_tool_handler(
products: list[Product],
):
def _mapper(product: Product, pid: str) -> dict[str, object]:
notes = " ".join(str(product.usage_notes or "").split())
if len(notes) > 500:
notes = notes[:497] + "..."
return {
"id": pid,
"name": product.name,
"usage_notes": notes,
}
return _build_product_details_tool_handler(products, mapper=_mapper)
def _build_safety_rules_tool_handler(
products: list[Product],
):
def _mapper(product: Product, pid: str) -> dict[str, object]:
ctx = product.to_llm_context()
return {
"id": pid,
"name": product.name,
"incompatible_with": (ctx.get("incompatible_with") or [])[:24],
"contraindications": (ctx.get("contraindications") or [])[:24],
"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"),
}
return _build_product_details_tool_handler(products, mapper=_mapper)
def _build_product_details_tool_handler(
products: list[Product],
mapper,
): ):
available_by_id = {str(p.id): p for p in products} available_by_id = {str(p.id): p for p in products}
def _handler(args: dict[str, object]) -> dict[str, object]: def _handler(args: dict[str, object]) -> 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(mapper(product, pid))
return {"products": products_payload}
return _handler
def _extract_requested_product_ids(
args: dict[str, object], max_ids: int = 8
) -> list[str]:
raw_ids = args.get("product_ids") raw_ids = args.get("product_ids")
if not isinstance(raw_ids, list): if not isinstance(raw_ids, list):
return {"products": []} return []
requested_ids: list[str] = [] requested_ids: list[str] = []
seen: set[str] = set() seen: set[str] = set()
@ -439,26 +547,26 @@ def _build_inci_tool_handler(
continue continue
seen.add(raw_id) seen.add(raw_id)
requested_ids.append(raw_id) requested_ids.append(raw_id)
if len(requested_ids) >= 8: if len(requested_ids) >= max_ids:
break break
return requested_ids
products_payload = []
for pid in requested_ids: def _extract_active_names(product: Product) -> list[str]:
product = available_by_id.get(pid) names: list[str] = []
if product is None: for a in product.actives or []:
if isinstance(a, dict):
name = str(a.get("name") or "").strip()
else:
name = str(getattr(a, "name", "") or "").strip()
if not name:
continue continue
inci = product.inci or [] if name in names:
compact_inci = [str(i)[:120] for i in inci[:128]] continue
products_payload.append( names.append(name)
{ if len(names) >= 12:
"id": pid, break
"name": product.name, return names
"inci": compact_inci,
}
)
return {"products": products_payload}
return _handler
_INCI_FUNCTION_DECLARATION = genai_types.FunctionDeclaration( _INCI_FUNCTION_DECLARATION = genai_types.FunctionDeclaration(
@ -480,6 +588,63 @@ _INCI_FUNCTION_DECLARATION = genai_types.FunctionDeclaration(
), ),
) )
_ACTIVES_FUNCTION_DECLARATION = genai_types.FunctionDeclaration(
name="get_product_actives",
description=(
"Return detailed active ingredients (name, strength, concentration, functions) "
"for selected product UUIDs."
),
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 AVAILABLE PRODUCTS.",
)
},
required=["product_ids"],
),
)
_USAGE_NOTES_FUNCTION_DECLARATION = genai_types.FunctionDeclaration(
name="get_product_usage_notes",
description=(
"Return compact usage notes for selected product UUIDs (application method, "
"timing, and cautions)."
),
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 AVAILABLE PRODUCTS.",
)
},
required=["product_ids"],
),
)
_SAFETY_RULES_FUNCTION_DECLARATION = genai_types.FunctionDeclaration(
name="get_product_safety_rules",
description=(
"Return safety and compatibility rules for selected product UUIDs: "
"incompatible_with, contraindications, context_rules and safety flags."
),
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 AVAILABLE PRODUCTS.",
)
},
required=["product_ids"],
),
)
def _build_objectives_context(include_minoxidil_beard: bool) -> str: def _build_objectives_context(include_minoxidil_beard: bool) -> str:
if include_minoxidil_beard: if include_minoxidil_beard:
@ -668,9 +833,10 @@ def suggest_routine(
"INPUT DATA:\n" "INPUT DATA:\n"
f"{skin_ctx}\n{grooming_ctx}\n{history_ctx}\n{day_ctx}\n{products_ctx}\n{objectives_ctx}" f"{skin_ctx}\n{grooming_ctx}\n{history_ctx}\n{day_ctx}\n{products_ctx}\n{objectives_ctx}"
"\nNARZEDZIA:\n" "\nNARZEDZIA:\n"
"- Masz dostep do funkcji get_product_inci(product_ids).\n" "- Masz dostep do funkcji: get_product_inci, get_product_safety_rules, get_product_actives, get_product_usage_notes.\n"
"- Uzyj jej tylko, gdy potrzebujesz dokladnego skladu INCI do oceny bezpieczenstwa, kompatybilnosci lub redundancji aktywnych.\n" "- Wywoluj narzedzia tylko, gdy potrzebujesz detali do decyzji klinicznej/bezpieczenstwa.\n"
"- Nie zgaduj INCI; jesli potrzebujesz skladu, wywolaj funkcje dla konkretnych UUID.\n" "- Staraj sie grupowac zapytania: podawaj wszystkie potrzebne UUID w jednym wywolaniu narzedzia.\n"
"- Nie zgaduj detali skladu i zasad bezpieczenstwa; jesli potrzebujesz szczegolow, wywolaj odpowiednie narzedzie.\n"
f"{notes_line}" f"{notes_line}"
f"{_ROUTINES_SINGLE_EXTRA}\n" f"{_ROUTINES_SINGLE_EXTRA}\n"
"Zwróć JSON zgodny ze schematem." "Zwróć JSON zgodny ze schematem."
@ -684,7 +850,12 @@ def suggest_routine(
update={ update={
"tools": [ "tools": [
genai_types.Tool( genai_types.Tool(
function_declarations=[_INCI_FUNCTION_DECLARATION], function_declarations=[
_INCI_FUNCTION_DECLARATION,
_SAFETY_RULES_FUNCTION_DECLARATION,
_ACTIVES_FUNCTION_DECLARATION,
_USAGE_NOTES_FUNCTION_DECLARATION,
],
) )
], ],
"tool_config": genai_types.ToolConfig( "tool_config": genai_types.ToolConfig(
@ -695,15 +866,53 @@ def suggest_routine(
} }
) )
function_handlers = {
"get_product_inci": _build_inci_tool_handler(available_products),
"get_product_safety_rules": _build_safety_rules_tool_handler(
available_products
),
"get_product_actives": _build_actives_tool_handler(available_products),
"get_product_usage_notes": _build_usage_notes_tool_handler(available_products),
}
try:
response = call_gemini_with_function_tools( response = call_gemini_with_function_tools(
endpoint="routines/suggest", endpoint="routines/suggest",
contents=prompt, contents=prompt,
config=config, config=config,
function_handlers={ function_handlers=function_handlers,
"get_product_inci": _build_inci_tool_handler(available_products)
},
user_input=prompt, user_input=prompt,
max_tool_roundtrips=2, max_tool_roundtrips=3,
)
except HTTPException as exc:
if (
exc.status_code != 502
or str(exc.detail) != "Gemini requested too many function calls"
):
raise
conservative_prompt = (
f"{prompt}\n\n"
"TRYB AWARYJNY (KONSERWATYWNY):\n"
"- Osiagnieto limit wywolan narzedzi.\n"
"- Nie wywoluj narzedzi ponownie.\n"
"- Zaproponuj maksymalnie konserwatywna, bezpieczna rutyne na podstawie dostepnych juz danych,"
" preferujac lagodne produkty wspierajace bariere i fotoprotekcje.\n"
"- Gdy masz watpliwosci, pomijaj ryzykowne aktywne kroki.\n"
)
response = call_gemini(
endpoint="routines/suggest",
contents=conservative_prompt,
config=get_creative_config(
system_instruction=_ROUTINES_SYSTEM_PROMPT,
response_schema=_SuggestionOut,
max_output_tokens=4096,
),
user_input=conservative_prompt,
tool_trace={
"mode": "fallback_conservative",
"reason": "max_tool_roundtrips_exceeded",
},
) )
raw = response.text raw = response.text

View file

@ -250,6 +250,9 @@ def test_suggest_routine(client, session):
kwargs = mock_gemini.call_args.kwargs kwargs = mock_gemini.call_args.kwargs
assert "function_handlers" in kwargs assert "function_handlers" in kwargs
assert "get_product_inci" in kwargs["function_handlers"] assert "get_product_inci" in kwargs["function_handlers"]
assert "get_product_safety_rules" in kwargs["function_handlers"]
assert "get_product_actives" in kwargs["function_handlers"]
assert "get_product_usage_notes" in kwargs["function_handlers"]
def test_suggest_batch(client, session): def test_suggest_batch(client, session):

View file

@ -4,15 +4,20 @@ from datetime import date, timedelta
from sqlmodel import Session from sqlmodel import Session
from innercontext.api.routines import ( from innercontext.api.routines import (
_build_actives_tool_handler,
_build_day_context, _build_day_context,
_build_grooming_context, _build_grooming_context,
_build_inci_tool_handler, _build_inci_tool_handler,
_build_objectives_context, _build_objectives_context,
_build_products_context, _build_products_context,
_build_recent_history, _build_recent_history,
_build_safety_rules_tool_handler,
_build_skin_context, _build_skin_context,
_build_usage_notes_tool_handler,
_contains_minoxidil_text, _contains_minoxidil_text,
_ev, _ev,
_extract_active_names,
_extract_requested_product_ids,
_get_available_products, _get_available_products,
_is_minoxidil_product, _is_minoxidil_product,
) )
@ -336,3 +341,68 @@ def test_build_inci_tool_handler_returns_only_available_ids(session: Session):
assert products[0]["id"] == str(available.id) assert products[0]["id"] == str(available.id)
assert products[0]["name"] == "Available" assert products[0]["name"] == "Available"
assert products[0]["inci"] == ["Water", "Niacinamide"] assert products[0]["inci"] == ["Water", "Niacinamide"]
def test_extract_requested_product_ids_dedupes_and_limits():
ids = _extract_requested_product_ids(
{
"product_ids": [
"id-1",
"id-2",
"id-1",
3,
"id-3",
"id-4",
]
},
max_ids=3,
)
assert ids == ["id-1", "id-2", "id-3"]
def test_extract_active_names_uses_compact_distinct_names(session: Session):
p = Product(
id=uuid.uuid4(),
name="Test",
category="serum",
brand="Test",
recommended_time="both",
leave_on=True,
actives=[
{"name": "Niacinamide", "percent": 10},
{"name": "Niacinamide", "percent": 5},
{"name": "Zinc PCA", "percent": 1},
],
product_effect_profile={},
)
names = _extract_active_names(p)
assert names == ["Niacinamide", "Zinc PCA"]
def test_additional_tool_handlers_return_product_payloads(session: Session):
p = Product(
id=uuid.uuid4(),
name="Detail Product",
category="serum",
brand="Test",
recommended_time="both",
leave_on=True,
usage_notes="Apply morning and evening.",
actives=[{"name": "Niacinamide", "percent": 5, "functions": ["niacinamide"]}],
incompatible_with=[{"target": "Retinol", "scope": "same_step"}],
context_rules={"safe_after_shaving": True},
product_effect_profile={},
)
ids_payload = {"product_ids": [str(p.id)]}
actives_out = _build_actives_tool_handler([p])(ids_payload)
assert actives_out["products"][0]["actives"][0]["name"] == "Niacinamide"
notes_out = _build_usage_notes_tool_handler([p])(ids_payload)
assert notes_out["products"][0]["usage_notes"] == "Apply morning and evening."
safety_out = _build_safety_rules_tool_handler([p])(ids_payload)
assert "incompatible_with" in safety_out["products"][0]
assert "context_rules" in safety_out["products"][0]