"""LLM client for Costco Route Optimizer.
Calls local Ollama endpoints for item extraction and zone classification.
"""
import json
import requests
from costco_route.config import LLM_URL, EMBED_URL, LLM_MODEL, EMBED_MODEL
def _call_llm(prompt: str, timeout: int = 60) -> str:
"""Send a prompt to Ollama and return the raw text response."""
payload = {
"model": LLM_MODEL,
"messages": [{"role": "user", "content": prompt}],
"stream": False,
"options": {"temperature": 0.1, "num_predict": 1024},
}
resp = requests.post(LLM_URL, json=payload, timeout=timeout)
resp.raise_for_status()
return resp.json()["message"]["content"].strip()
def _get_embedding(text: str, timeout: int = 30) -> list[float]:
"""Get an embedding vector from Ollama."""
payload = {"model": EMBED_MODEL, "prompt": text}
resp = requests.post(EMBED_URL, json=payload, timeout=timeout)
resp.raise_for_status()
return resp.json()["embedding"]
def extract_items(raw_input: str) -> list[str]:
"""Parse a stream-of-consciousness grocery list into individual items.
Uses the LLM to normalize and separate compound phrases.
Falls back to simple comma/space splitting on LLM failure.
"""
from costco_route.config import EXTRACT_PROMPT
prompt = EXTRACT_PROMPT.format(input=raw_input)
try:
raw = _call_llm(prompt)
# Strip markdown code fences if present
raw = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
items = json.loads(raw)
if isinstance(items, list):
return [str(i).strip() for i in items if str(i).strip()]
except (json.JSONDecodeError, requests.RequestException, KeyError):
pass
# Fallback: simple split on commas, newlines, and common separators
import re
parts = re.split(r"[,\n]+", raw_input)
return [p.strip() for p in parts if p.strip()]
def classify_items(items: list[str]) -> dict[str, list[str]]:
"""Classify grocery items into Costco warehouse zones.
Returns a dict of zone_id → list of items.
Items that can't be classified go to zone "04" (Pantry — the catch-all center).
"""
from costco_route.config import ZONE_CLASSIFY_PROMPT
items_str = "\n".join(f"- {item}" for item in items)
prompt = ZONE_CLASSIFY_PROMPT.format(items=items_str)
try:
raw = _call_llm(prompt)
raw = raw.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
classified = json.loads(raw)
# Validate and normalize
result = {}
for zone_id, zone_items in classified.items():
zone_id = str(zone_id).strip()
if zone_id in ("01","02","03","04","05","06","07","08","09","10"):
if isinstance(zone_items, list):
result[zone_id] = [str(i).strip() for i in zone_items if str(i).strip()]
# Hard sanity checks: override obvious LLM misclassifications
result = _apply_sanity_checks(result)
return result
except (json.JSONDecodeError, requests.RequestException, KeyError):
# On failure, put everything in pantry (safest fallback)
return {"04": items}
def _apply_sanity_checks(classified: dict[str, list[str]]) -> dict[str, list[str]]:
"""Override obviously wrong zone assignments.
The LLM sometimes hallucinates zones for common items.
These rules catch the most frequent mistakes.
"""
# Items that must NOT be in Zone 10 (checkout/food court)
NEVER_ZONE_10 = [
"toilet paper", "paper towels", "detergent", "trash bags",
"laundry", "cleaning", "dish soap",
]
# Items that belong in Zone 06 (dairy/cold) not Zone 04 (pantry)
DAIRY_KEYWORDS = [
"cheese", "cheddar", "mozzarella", "yogurt", "butter", "sour cream",
"cream cheese", "half and half", "heavy cream",
]
# Items that belong in Zone 07 (fresh/produce) not Zone 02 (seasonal)
PRODUCE_KEYWORDS = [
"fruit", "tomato", "lettuce", "romaine", "onion", "potato",
"avocado", "banana", "apple", "berry", "berries", "vegetable",
"pepper", "carrot", "broccoli", "spinach", "salad", "corn",
"celery", "cucumber", "mushroom", "zucchini", "squash", "kale",
]
# Items that belong in Zone 04 (pantry) not elsewhere
PANTRY_KEYWORDS = {
"stock": "04", "broth": "04", "soup": "04", "canned": "04",
}
# Bacon → cold room (Zone 06), not meat counter (Zone 07)
BACON_KEYWORDS = ["bacon"]
# Fix Zone 10 misclassifications → move to Zone 08
if "10" in classified:
misplaced = []
remaining = []
for item in classified["10"]:
item_lower = item.lower()
if any(kw in item_lower for kw in NEVER_ZONE_10):
misplaced.append(item)
else:
remaining.append(item)
if misplaced:
classified.setdefault("08", []).extend(misplaced)
if remaining:
classified["10"] = remaining
else:
del classified["10"]
# Fix Zone 04 dairy items → move to Zone 06
if "04" in classified:
misplaced = []
remaining = []
for item in classified["04"]:
item_lower = item.lower()
if any(kw in item_lower for kw in DAIRY_KEYWORDS):
misplaced.append(item)
else:
remaining.append(item)
if misplaced:
classified.setdefault("06", []).extend(misplaced)
if remaining:
classified["04"] = remaining
else:
del classified["04"]
# Fix Zone 02 produce items → move to Zone 07
if "02" in classified:
misplaced = []
remaining = []
for item in classified["02"]:
item_lower = item.lower()
if any(kw in item_lower for kw in PRODUCE_KEYWORDS):
misplaced.append(item)
else:
remaining.append(item)
if misplaced:
classified.setdefault("07", []).extend(misplaced)
if remaining:
classified["02"] = remaining
else:
del classified["02"]
# Fix bacon → Zone 06 (cold room), not Zone 07 (meat counter)
for zone in list(classified.keys()):
bacon_items = [item for item in classified[zone] if "bacon" in item.lower()]
if bacon_items and zone != "06":
classified[zone] = [item for item in classified[zone] if "bacon" not in item.lower()]
classified.setdefault("06", []).extend(bacon_items)
if not classified[zone]:
del classified[zone]
# Fix stock/broth → Zone 04 (pantry)
for zone in list(classified.keys()):
stock_items = [item for item in classified[zone] if any(kw in item.lower() for kw in PANTRY_KEYWORDS)]
if stock_items and zone != "04":
classified[zone] = [item for item in classified[zone] if not any(kw in item.lower() for kw in PANTRY_KEYWORDS)]
classified.setdefault("04", []).extend(stock_items)
if not classified[zone]:
del classified[zone]
return classified
def get_embedding(text: str) -> list[float]:
"""Public embedding interface."""
return _get_embedding(text)