app/tools/knowledge_tool.py

"""Tool: search the knowledge base (RAG)."""

from __future__ import annotations

import os
from typing import TYPE_CHECKING, Any

from sqlalchemy import select

from app.models import KnowledgeSource
from app.rag import index

# Cosine-similarity floor: over-fetch a wide candidate pool, then drop chunks
# below this so an irrelevant question (e.g. shipping when the KB has no shipping
# page) returns the HONEST "no info" path instead of 8 junk chunks that push weak
# models to invent or redirect. Tunable via RAG_MIN_SCORE; conservative default.
_RAG_MIN_SCORE = float(os.getenv("RAG_MIN_SCORE", "0.25"))
_RAG_KEEP = 8           # max chunks fed to the model after the floor
_RAG_OVERFETCH = 30     # candidate pool before the floor

if TYPE_CHECKING:
    from app.tools.registry import ToolContext


def _primary_url(source: KnowledgeSource) -> str | None:
    """The best real URL to cite for a source (its location if public, else the
    first discovered page). Never invented — these come from indexing."""
    if (source.location or "").startswith(("http://", "https://")):
        return source.location
    for u in (source.urls or []):
        if u.startswith(("http://", "https://")):
            return u
    return None


async def run(args: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
    query = (args.get("query") or "").strip()
    if not query:
        return {"context": "", "sources": [], "note": "empty query"}
    results = await index.search(ctx.db, query, k=_RAG_OVERFETCH, tenant_id=ctx.tenant_id)
    # Drop everything below the relevance floor; if nothing clears it, take the
    # honest "no info" path (better than feeding junk the model invents around).
    kept = [(c, s) for c, s in results if s >= _RAG_MIN_SCORE][:_RAG_KEEP]
    # #20: a CORRECT cross-language semantic match (e.g. "warranty" -> "garantía"
    # content) scores ~0.19 cosine with the multilingual MiniLM — below the 0.25
    # keyword-tuned floor — and was being dropped, so the bot wrongly said "no
    # info". If nothing clears the floor, still surface the single best chunk when
    # it clears a lower semantic floor; let the model judge relevance.
    if not kept and results:
        best_c, best_s = max(results, key=lambda cs: cs[1])
        if best_s >= _RAG_MIN_SCORE * 0.6:
            kept = [(best_c, best_s)]
    if not kept:
        # Log the gap so the merchant sees what the bot couldn't answer and can
        # add it to their knowledge. Tells the model to NOT invent an answer.
        from app import analytics

        analytics.record(
            ctx.db, ctx.tenant_id, "unresolved",
            session_id=ctx.session.id, meta={"q": query[:300]},
        )
        return {
            "context": "",
            "sources": [],
            "note": "no relevant information found",
            "hint": "No tienes información sobre esto. NO la inventes: dile al "
            "cliente que no estás seguro y ofrece pasarlo con el equipo "
            "(escalate_to_human) para confirmárselo.",
        }
    blocks = []
    sources: list[str] = []
    source_ids = []
    for chunk, _score in kept:
        name = (chunk.meta or {}).get("source_name", "fuente")
        blocks.append(f"[{name}] {chunk.text}")
        if name not in sources:
            sources.append(name)
        if chunk.source_id not in source_ids:
            source_ids.append(chunk.source_id)

    # Surface the REAL link(s) for the matched sources so the model can cite the
    # correct URL instead of guessing one (which would 404). Only verified URLs.
    links: list[dict[str, str]] = []
    if source_ids:
        rows = (
            await ctx.db.execute(
                select(KnowledgeSource).where(
                    KnowledgeSource.id.in_(source_ids),
                    KnowledgeSource.tenant_id == ctx.tenant_id,  # defense in depth
                )
            )
        ).scalars().all()
        for s in rows:
            url = _primary_url(s)
            if url:
                links.append({"name": s.name, "url": url})

    out: dict[str, Any] = {"context": "\n\n".join(blocks), "sources": sources}
    if links:
        out["links"] = links
        # Proactive citation builds real trust (the customer can verify a policy
        # claim themselves) instead of just asking the shopper to take the bot's
        # word for it -- especially for concrete/checkable facts (returns,
        # shipping windows, warranty, prices). Not every trivial reply needs one.
        out["hint"] = (
            "Cuando cites una política o un dato concreto y verificable "
            "(devoluciones, envíos, garantía, precios, plazos), añade el enlace "
            "real al final en formato [nombre](url) para que el cliente pueda "
            "comprobarlo — genera confianza. Usa EXACTAMENTE una de estas URLs, "
            "nunca inventes ni modifiques rutas."
        )
    return out