From 7720f1def5c91dbccc05423ea40a1d670eda76fb Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Tue, 14 Apr 2026 11:43:19 -0700 Subject: [PATCH] feat: QueryTranslator with domain-aware system prompt and category hint injection --- app/llm/query_translator.py | 83 ++++++++++++++++++++++++++- tests/test_query_translator.py | 100 ++++++++++++++++++++++++++++++++- 2 files changed, 179 insertions(+), 4 deletions(-) diff --git a/app/llm/query_translator.py b/app/llm/query_translator.py index 5ee8a83..6b18c47 100644 --- a/app/llm/query_translator.py +++ b/app/llm/query_translator.py @@ -11,7 +11,10 @@ from __future__ import annotations import json import logging from dataclasses import dataclass -from typing import Optional +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from app.platforms.ebay.categories import EbayCategoryCache log = logging.getLogger(__name__) @@ -85,6 +88,80 @@ def _parse_response(raw: str) -> SearchParamsResponse: ) +# ── System prompt template ──────────────────────────────────────────────────── + +_SYSTEM_PROMPT_TEMPLATE = """\ +You are a search assistant for Snipe, an eBay listing intelligence tool. +Your job is to translate a natural-language description of what someone is looking for +into a structured eBay search configuration. + +Return ONLY a JSON object with these exact fields — no preamble, no markdown, no extra keys: + base_query (string) Primary search term, short — e.g. "RTX 3080", "vintage Leica" + must_include_mode (string) One of: "all" (AND), "any" (OR), "groups" (CNF: pipe=OR within group, comma=AND between groups) + must_include (string) Filter string per mode — leave blank if nothing to filter + must_exclude (string) Comma-separated terms to exclude — e.g. "mining,for parts,broken" + max_price (number|null) Maximum price in USD, or null + min_price (number|null) Minimum price in USD, or null + condition (array) Any of: "new", "used", "for_parts" — empty array means any condition + category_id (string|null) eBay category ID from the list below, or null if no match + explanation (string) One plain sentence summarizing what you built + +eBay "groups" mode syntax example: to find a GPU that is BOTH (nvidia OR amd) AND (16gb OR 8gb): + must_include_mode: "groups" + must_include: "nvidia|amd, 16gb|8gb" + +Phrase "like new", "open box", "refurbished" -> condition: ["used"] +Phrase "broken", "for parts", "not working" -> condition: ["for_parts"] +If unsure about condition, use an empty array. + +Available eBay categories (use category_id verbatim if one fits — otherwise omit): +{category_hints} + +If none match, omit category_id (set to null). Respond with valid JSON only. No commentary outside the JSON object. +""" + + +# ── QueryTranslator ─────────────────────────────────────────────────────────── + class QueryTranslator: - """Stub — implemented in Task 6.""" - pass + """Translates natural-language search descriptions into SearchParamsResponse. + + Args: + category_cache: An EbayCategoryCache instance (may have empty cache). + llm_router: An LLMRouter instance from circuitforge_core. + """ + + def __init__(self, category_cache: "EbayCategoryCache", llm_router: object) -> None: + self._cache = category_cache + self._llm_router = llm_router + + def translate(self, natural_language: str) -> SearchParamsResponse: + """Translate a natural-language query into a SearchParamsResponse. + + Raises QueryTranslatorError if the LLM fails or returns bad JSON. + """ + # Extract up to 10 keywords for category hint lookup + keywords = [w for w in natural_language.split()[:10] if len(w) > 2] + hints = self._cache.get_relevant(keywords, limit=30) + if not hints: + hints = self._cache.get_all_for_prompt(limit=40) + + if hints: + category_hints = "\n".join(f"{cid}: {path}" for cid, path in hints) + else: + category_hints = "(no categories cached — omit category_id)" + + system_prompt = _SYSTEM_PROMPT_TEMPLATE.format(category_hints=category_hints) + + try: + raw = self._llm_router.complete( + natural_language, + system=system_prompt, + max_tokens=512, + ) + except Exception as exc: + raise QueryTranslatorError( + f"LLM backend error: {exc}", raw="" + ) from exc + + return _parse_response(raw) diff --git a/tests/test_query_translator.py b/tests/test_query_translator.py index 53380c4..20f16eb 100644 --- a/tests/test_query_translator.py +++ b/tests/test_query_translator.py @@ -2,11 +2,12 @@ from __future__ import annotations import json +from pathlib import Path from unittest.mock import MagicMock, patch import pytest -from app.llm.query_translator import QueryTranslatorError, SearchParamsResponse, _parse_response +from app.llm.query_translator import QueryTranslator, QueryTranslatorError, SearchParamsResponse, _parse_response # ── _parse_response ─────────────────────────────────────────────────────────── @@ -70,3 +71,100 @@ def test_parse_response_missing_required_field(): }) with pytest.raises(QueryTranslatorError): _parse_response(raw) + + +# ── QueryTranslator (integration with mocked LLMRouter) ────────────────────── + +from app.platforms.ebay.categories import EbayCategoryCache +from circuitforge_core.db import get_connection, run_migrations + + +@pytest.fixture +def db_with_categories(tmp_path): + conn = get_connection(tmp_path / "test.db") + run_migrations(conn, Path("app/db/migrations")) + cache = EbayCategoryCache(conn) + cache._seed_bootstrap() + return conn + + +def _make_translator(db_conn, llm_response: str) -> QueryTranslator: + from app.platforms.ebay.categories import EbayCategoryCache + cache = EbayCategoryCache(db_conn) + mock_router = MagicMock() + mock_router.complete.return_value = llm_response + return QueryTranslator(category_cache=cache, llm_router=mock_router) + + +def test_translate_returns_search_params(db_with_categories): + llm_out = json.dumps({ + "base_query": "RTX 3080", + "must_include_mode": "groups", + "must_include": "rtx|geforce, 3080", + "must_exclude": "mining,for parts", + "max_price": 300.0, + "min_price": None, + "condition": ["used"], + "category_id": "27386", + "explanation": "Searching for used RTX 3080 GPUs under $300.", + }) + t = _make_translator(db_with_categories, llm_out) + result = t.translate("used RTX 3080 under $300 no mining") + assert result.base_query == "RTX 3080" + assert result.max_price == 300.0 + + +def test_translate_injects_category_hints(db_with_categories): + """The system prompt sent to the LLM must contain category_id hints.""" + llm_out = json.dumps({ + "base_query": "GPU", + "must_include_mode": "all", + "must_include": "", + "must_exclude": "", + "max_price": None, + "min_price": None, + "condition": [], + "category_id": None, + "explanation": "Searching for GPUs.", + }) + t = _make_translator(db_with_categories, llm_out) + t.translate("GPU") + call_args = t._llm_router.complete.call_args + system_prompt = call_args.kwargs.get("system") or call_args.args[1] + # Bootstrap seeds "27386" for Graphics Cards — should appear in the prompt + assert "27386" in system_prompt + + +def test_translate_empty_category_cache_still_works(tmp_path): + """No crash when category cache is empty — prompt uses fallback text.""" + from circuitforge_core.db import get_connection, run_migrations + conn = get_connection(tmp_path / "empty.db") + run_migrations(conn, Path("app/db/migrations")) + # Do NOT seed bootstrap — empty cache + llm_out = json.dumps({ + "base_query": "vinyl", + "must_include_mode": "all", + "must_include": "", + "must_exclude": "", + "max_price": None, + "min_price": None, + "condition": [], + "category_id": None, + "explanation": "Searching for vinyl records.", + }) + t = _make_translator(conn, llm_out) + result = t.translate("vinyl records") + assert result.base_query == "vinyl" + call_args = t._llm_router.complete.call_args + system_prompt = call_args.kwargs.get("system") or call_args.args[1] + assert "If none match" in system_prompt + + +def test_translate_llm_error_raises_query_translator_error(db_with_categories): + from app.platforms.ebay.categories import EbayCategoryCache + cache = EbayCategoryCache(db_with_categories) + mock_router = MagicMock() + mock_router.complete.side_effect = RuntimeError("all backends exhausted") + t = QueryTranslator(category_cache=cache, llm_router=mock_router) + with pytest.raises(QueryTranslatorError, match="LLM backend"): + t.translate("used GPU")