RAG query pipeline: @rag-search retrieval endpoint + answer generation (#79)

IMPLEMENTATION-79.md

@@ -224,7 +224,7 @@ Goal: `@rag-search` returns `{answer, sources}` end to end.
 
 **Retrieval endpoint (1d):**
 
-- [ ] **3.1** New restapi service `@rag-search`
+- [x] **3.1** New restapi service `@rag-search`
       (`backend/src/kitconcept/solr/services/`, staged behind the toggle):
       embed query (`search_query:` prefix), run
       `{!knn f=content_vector topK=K}` with the same `fq` security/path/
@@ -234,21 +234,74 @@ Goal: `@rag-search` returns `{answer, sources}` end to end.
 
 **Answer generation (1.5d):**
 
-- [ ] **3.2 Prompt template**: fixed instruction template (answer only from
+- [x] **3.2 Prompt template**: fixed instruction template (answer only from
       the provided context; answer in the question's language; decline
       explicitly when no answer is found; refer to sources). Code default;
       registry override is post-MVP.
-- [ ] **3.3 Generation call**: send question + matched chunks (+ parent
+- [x] **3.3 Generation call**: send question + matched chunks (+ parent
       title/URL) to `qwen3:14b`; compose the `{answer, sources}` response;
       structured errors for timeout/unavailable/not-configured.
-- [ ] Validate against the Step 2 questions (smoke level: right documents
-      found, declines when it should).
+- [x] Validate against the Step 2 questions (smoke level: right documents
+      found, declines when it should). Note: Step 2 (the proper corpus,
+      internal ticket 459) was deferred by decision - validation ran
+      against a minimal hand-made corpus; re-run once the corpus lands.
 
 Demo at end of step: full RAG loop over REST on the demo corpus. **This is
 the MVP backend.** The user-facing MVP completes when the search UI from the
 kitconcept.intranet modal project integrates this endpoint (outside this
 repository's estimates).
 
+### Implementation notes — Step 3 (design decisions made on the way)
+
+- **Pipeline extracted from the HTTP layer**: the logic lives in
+  `rag/pipeline.py` (`run_rag_search`) with the prompt in
+  `rag/prompt.py`; the restapi service is a thin wrapper. This keeps
+  the pipeline testable with faked Solr/LLM and reusable by a future
+  evaluation harness.
+- **Structured errors with codes**: the response carries `error`
+  (human readable) plus `error_code` (`not_configured`,
+  `embedding_failed`, `generation_failed`, `solr_unavailable`) so the
+  frontend can degrade to the classic search. A generation failure
+  still returns the retrieved sources. Empty retrieval is NOT an
+  error: `answer: null, sources: []`.
+- **Sources are enriched parent documents**: one extra Solr query
+  fetches parent metadata (`Title`, `Description`, `Type`); each
+  source carries the best-ranked chunk's text as `snippet`. `@type`
+  is Solr's friendly type name (e.g. "Page"), consistent with the
+  classic search results.
+- **Optional filters**: `path_prefix` and `lang` request parameters
+  compose as additional pre-filter queries next to the security
+  filter.
+- **Model bake-off** (`qwen3:14b` vs alternatives) stays in the
+  overflow list per the re-scoping; the model remains fixed.
+- Observation from live testing on the minimal corpus: with only a
+  handful of documents, `topK=5` spans the whole corpus, so weakly
+  related sources appear below the top hit. On a realistic corpus a
+  score threshold (`{!vectorSimilarity}` / `minReturn`) or a smaller
+  source cutoff is worth evaluating - noted for the quality pass.
+
+## Known issues (accepted for the MVP)
+
+1. **The Solr tests and a local dev site share the Solr core.** The
+   test compose project uses the fixed host port 8983 — the same port
+   a dev Solr uses — and the test fixtures clear/recreate the index
+   (`down -v` at session start, `maintenance.clear()` in the portal
+   fixtures; the RAG live e2e tests behave exactly like the
+   pre-existing service tests here). Consequences and workarounds:
+   (a) stop the site's Solr before running the tests so the situation
+   does not arise, or (b) if it has happened, no harm is done — simply
+   reindex with `make solr-activate-and-reindex` before using the
+   site again. A real fix (ephemeral host ports for the test project,
+   wired into the test layer's `collective.solr.port`) is a **post-MVP
+   improvement**, tracked in the overflow list.
+2. **Weak sources on small corpora (topK semantics).** `{!knn topK=5}`
+   returns the 5 *nearest* chunks unconditionally, so on a small
+   corpus weakly related documents appear in the source list (answer
+   quality is unaffected — the grounding prompt handles weak context;
+   verified by the decline behavior). Accepted for now; **re-check
+   with the real corpus** (internal ticket 459) and evaluate a
+   similarity cutoff for displayed sources in the quality pass.
+
 ## Post-MVP follow-ups
 
 **Step P1 — Hybrid RRF + tests (2d), first follow-up:**
@@ -274,6 +327,7 @@ remains):
 | Generation model comparison (`qwen3:14b` vs `qwen3.5:9b-q8_0` vs others) on the test questions | 0.5d |
 | Full configuration surface: registry records for model names, topK, chunk size, prompt override | 0.5d |
 | Acceptance test flow + full CI wiring (after the search-UI integration, so acceptance tests target the real UI) | 1d |
+| Separate Solr core/ports for tests vs. local dev site (see known issue 1) | 0.5d |
 
 **Later roadmap** (tracked, not scheduled): full evaluation harness
 (Recall@k/MRR/nDCG as CI regression gate, RAGAS faithfulness/relevancy with

backend/news/79.feature.2

@@ -0,0 +1 @@
+Add the @rag-search endpoint: single-turn RAG search on a query pipeline (embed, permission-trimmed knn chunk retrieval, parent collapse, grounded answer generation) with structured error results. Exclude Image content from chunking. @reebalazs

backend/src/kitconcept/solr/rag/config.py

@@ -49,6 +49,10 @@
 # Number of texts sent to /api/embed in one request.
 EMBED_BATCH_SIZE = 32
 
+# Number of chunks retrieved for a question (and passed to the
+# generation prompt).
+TOP_K = 5
+
 REGISTRY_ENABLED_KEY = "kitconcept.solr.rag_enabled"
 
 # Endpoint paths, relative to the server root URL. The defaults match

backend/src/kitconcept/solr/rag/pipeline.py

@@ -0,0 +1,209 @@
+"""The RAG query pipeline: question -> retrieved chunks -> answer.
+
+Implements the single-turn RAG search (SPECIFICATION-79.md §4):
+
+1. embed the user's question (``search_query:`` prefix),
+2. retrieve the top chunks via a ``{!knn}`` query — the existing
+   security/path/language filter queries compose with the vector query
+   as HNSW pre-filters, so permission trimming works unchanged,
+3. collapse the chunk hits to their parent documents (the sources),
+4. generate the answer with the general-purpose model, prompted with
+   the matched chunk texts (chunk-level context, decision 9) and
+   constrained to the provided context.
+
+The pipeline is independent of the REST service so it can be tested
+with a faked Solr connection and LLM client, and reused (e.g. by a
+future evaluation harness).
+"""
+
+from collective.solr.exceptions import SolrConnectionException
+from collective.solr.interfaces import ISolrConnectionManager
+from collective.solr.parser import SolrResponse
+from dataclasses import dataclass
+from dataclasses import field
+from kitconcept.solr.rag.client import LLMClient
+from kitconcept.solr.rag.client import LLMClientError
+from kitconcept.solr.rag.config import RagConfig
+from kitconcept.solr.rag.config import TOP_K
+from kitconcept.solr.rag.prompt import build_prompt
+from kitconcept.solr.rag.prompt import strip_thinking
+from kitconcept.solr.rag.prompt import SYSTEM_PROMPT
+from plone import api
+from zope.component import queryUtility
+
+import logging
+
+
+logger = logging.getLogger("kitconcept.solr.rag")
+
+# Error codes for structured error reporting (the frontend degrades
+# gracefully based on these; the message is for humans/logs).
+ERROR_NOT_CONFIGURED = "not_configured"
+ERROR_EMBEDDING_FAILED = "embedding_failed"
+ERROR_GENERATION_FAILED = "generation_failed"
+ERROR_SOLR_UNAVAILABLE = "solr_unavailable"
+
+CHUNK_FIELD_LIST = "UID,parent_uid,parent_title,chunk_text,path_string,score"
+SOURCE_FIELD_LIST = "UID,Title,Description,Type,path_string"
+SNIPPET_LENGTH = 300
+
+
+@dataclass
+class RagResult:
+    answer: str | None = None
+    sources: list = field(default_factory=list)
+    error: str | None = None
+    error_code: str | None = None
+
+    @classmethod
+    def failure(cls, code: str, message: str) -> "RagResult":
+        return cls(error=message, error_code=code)
+
+
+def run_rag_search(
+    question: str,
+    config: RagConfig,
+    security_filter: str,
+    path_prefix: str | None = None,
+    lang: str | None = None,
+) -> RagResult:
+    """Run the single-turn RAG search pipeline.
+
+    :param question: The user's natural language question.
+    :param config: Resolved RAG configuration.
+    :param security_filter: The allowedRolesAndUsers filter query for
+        the current user (from ``services.solr.security_filter``).
+    :param path_prefix: Optional path to restrict the search to.
+    :param lang: Optional language to restrict the search to.
+    """
+    client = LLMClient(config)
+    try:
+        vector = client.embed_query(question)
+    except LLMClientError as e:
+        logger.warning("rag-search: embedding failed: %s", e)
+        return RagResult.failure(ERROR_EMBEDDING_FAILED, str(e))
+
+    conn = get_connection()
+    if conn is None:
+        return RagResult.failure(
+            ERROR_SOLR_UNAVAILABLE, "no Solr connection (solr inactive?)"
+        )
+    try:
+        chunks = search_chunks(conn, vector, security_filter, path_prefix, lang)
+        sources = collapse_sources(conn, chunks) if chunks else []
+    except (SolrConnectionException, OSError) as e:
+        # collective.solr raises raw socket errors (e.g.
+        # ConnectionRefusedError) when the Solr server is down
+        logger.warning("rag-search: Solr unavailable: %s", e)
+        return RagResult.failure(ERROR_SOLR_UNAVAILABLE, str(e))
+    if not chunks:
+        # No matching (visible) content: not an error - the answer is
+        # that there is no answer.
+        return RagResult()
+
+    prompt = build_prompt(question, chunks)
+    try:
+        answer = client.chat(prompt, system=SYSTEM_PROMPT)
+    except LLMClientError as e:
+        logger.warning("rag-search: generation failed: %s", e)
+        result = RagResult.failure(ERROR_GENERATION_FAILED, str(e))
+        result.sources = sources  # retrieval worked; expose the sources
+        return result
+    return RagResult(answer=strip_thinking(answer), sources=sources)
+
+
+def get_connection():
+    manager = queryUtility(ISolrConnectionManager)
+    return manager.getConnection() if manager is not None else None
+
+
+def format_vector(vector: list[float]) -> str:
+    return "[" + ",".join(f"{x:.8f}" for x in vector) + "]"
+
+
+def search_chunks(
+    conn,
+    vector: list[float],
+    security_filter: str,
+    path_prefix: str | None = None,
+    lang: str | None = None,
+) -> list[dict]:
+    """Top-K chunk hits for the query vector, permission trimmed."""
+    filter_queries = [security_filter, "is_rag_chunk:true"]
+    if path_prefix:
+        portal_path = "/".join(api.portal.get().getPhysicalPath())
+        prefix = portal_path + path_prefix.rstrip("/")
+        filter_queries.append(f'path_parents:"{prefix}"')
+    if lang:
+        filter_queries.append(f"Language:({lang} OR any)")
+    response = conn.search(
+        q=f"{{!knn f=content_vector topK={TOP_K}}}{format_vector(vector)}",
+        fq=filter_queries,
+        fl=CHUNK_FIELD_LIST,
+        rows=TOP_K,
+    )
+    try:
+        return list(SolrResponse(response).results())
+    finally:
+        response.close()
+
+
+def collapse_sources(conn, chunks: list[dict]) -> list[dict]:
+    """Parent documents of the matched chunks, in rank order.
+
+    Parent-document retrieval: retrieval matches chunks, but the user
+    sees the parent documents as the sources. Parent metadata is
+    fetched from Solr in one query and merged with a snippet from the
+    best-ranked chunk of each parent.
+    """
+    order: list[str] = []
+    best_chunk: dict[str, dict] = {}
+    for chunk in chunks:
+        parent_uid = chunk.get("parent_uid")
+        if not parent_uid:
+            continue
+        if parent_uid not in best_chunk:
+            order.append(parent_uid)
+            best_chunk[parent_uid] = chunk
+    if not order:
+        return []
+    parents = fetch_parents(conn, order)
+
+    portal = api.portal.get()
+    portal_path = "/".join(portal.getPhysicalPath())
+    portal_url = portal.absolute_url()
+
+    sources = []
+    for parent_uid in order:
+        parent = parents.get(parent_uid, {})
+        chunk = best_chunk[parent_uid]
+        path_string = parent.get("path_string") or chunk.get("path_string", "")
+        url = (
+            portal_url + path_string[len(portal_path) :]
+            if path_string.startswith(portal_path)
+            else path_string
+        )
+        sources.append({
+            "@id": url,
+            "UID": parent_uid,
+            "@type": parent.get("Type", ""),
+            "title": parent.get("Title") or chunk.get("parent_title", ""),
+            "description": parent.get("Description", ""),
+            "snippet": chunk.get("chunk_text", "")[:SNIPPET_LENGTH],
+        })
+    return sources
+
+
+def fetch_parents(conn, uids: list[str]) -> dict[str, dict]:
+    """Metadata of the parent documents, keyed by UID."""
+    query = " OR ".join(f'"{uid}"' for uid in uids)
+    response = conn.search(
+        q=f"UID:({query})",
+        fl=SOURCE_FIELD_LIST,
+        rows=len(uids),
+    )
+    try:
+        results = SolrResponse(response).results()
+    finally:
+        response.close()
+    return {flare["UID"]: flare for flare in results}

backend/src/kitconcept/solr/rag/processor.py

@@ -70,6 +70,13 @@
 # this covers roughly 40k tokens of text per document.
 MAX_CHUNKS_PER_DOCUMENT = 100
 
+# Content types excluded from chunking. An Image's only chunkable text
+# is its title/description - metadata about an illustration, not
+# knowledge - and it pollutes answer sources (a photo cited as a
+# source). File stays included: its title/description locates real
+# documents (and the post-MVP Tika body text will build on it).
+EXCLUDED_PORTAL_TYPES = frozenset({"Image"})
+
 
 def chunk_uid(uid: str, index: int) -> str:
     return f"{uid}#rag-{index}"
@@ -134,6 +141,10 @@ def index(self, obj, attributes=None):
         manager, conn = self._connection()
         if conn is None:
             return
+        if getattr(obj, "portal_type", None) in EXCLUDED_PORTAL_TYPES:
+            # also drop chunks indexed before the type was excluded
+            conn.deleteByQuery(chunk_query(uid))
+            return
         if attributes is not None:
             attributes = set(attributes)
             if attributes & TEXT_ATTRIBUTES:

backend/src/kitconcept/solr/rag/prompt.py

@@ -0,0 +1,46 @@
+"""Prompt building for the RAG answer generation.
+
+The prompt constrains the model to the retrieved context
+(SPECIFICATION-79.md §4): answer only from the provided documents,
+answer in the language of the question, and decline explicitly when
+the answer is not found. Code default for the MVP; a registry override
+is a post-MVP configuration item.
+"""
+
+from kitconcept.solr.rag.config import TOP_K
+
+import re
+
+
+SYSTEM_PROMPT = (
+    "You are the search assistant of an intranet site. Answer the"
+    " user's question based only on the provided context documents."
+    " If the answer is not contained in them, say that you could not"
+    " find the answer in the documentation - never invent information."
+    " Answer in the language of the question. Be concise: one or two"
+    " short paragraphs, no headings and no lists unless the question"
+    " asks for an enumeration."
+)
+
+PROMPT_TEMPLATE = (
+    "Context documents:\n\n{context}\n\n"
+    "Question: {question}\n\n"
+    "Answer the question based only on the context documents above."
+)
+
+# Reasoning models may emit a thinking block; never show it to users.
+THINK_RE = re.compile(r"<think>.*?</think>\s*", re.DOTALL)
+
+
+def build_prompt(question: str, chunks: list[dict]) -> str:
+    """One prompt containing the question and the retrieved context."""
+    parts = []
+    for index, chunk in enumerate(chunks[:TOP_K], start=1):
+        title = chunk.get("parent_title", "")
+        text = chunk.get("chunk_text", "")
+        parts.append(f"[{index}] {title}\n{text}")
+    return PROMPT_TEMPLATE.format(context="\n\n".join(parts), question=question)
+
+
+def strip_thinking(answer: str) -> str:
+    return THINK_RE.sub("", answer).strip()

backend/src/kitconcept/solr/services/configure.zcml

@@ -21,8 +21,6 @@
       name="@solr-suggest"
       />
 
-  <!-- RAG: TESTING - rough draft of the RAG query pipeline, to be
-       replaced by the proper implementation (query pipeline ticket) -->
   <plone:service
       method="GET"
       factory=".rag_search.RagSearch"