File size: 4,931 Bytes
187966e e73905d 187966e 9c48747 187966e e73905d 187966e 9c48747 187966e e73905d 069d2cd e73905d acad0ab 187966e 9c48747 e73905d 187966e 9c48747 20f9db9 9c48747 e74dda2 9c48747 e74dda2 9c48747 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """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
|