hkuds--deeptutor
wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
111 行
3.5 KiB
Python
111 行
3.5 KiB
Python
"""
|
|
Optional RAG lookup helper for block generators.
|
|
|
|
If the block context has a primary KB and ``rag_enabled``, run a single
|
|
``rag_search`` call and return both the synthesised text and a list of
|
|
``SourceAnchor`` objects. Failures are swallowed silently – RAG is treated as
|
|
"nice to have" by every generator.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
import logging
|
|
|
|
from ..models import SourceAnchor
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class RagLookup:
|
|
text: str = ""
|
|
anchors: list[SourceAnchor] = field(default_factory=list)
|
|
used: bool = False
|
|
|
|
|
|
def _coerce_anchors(sources: list[dict] | None) -> list[SourceAnchor]:
|
|
if not sources:
|
|
return []
|
|
anchors: list[SourceAnchor] = []
|
|
for src in sources[:6]:
|
|
if not isinstance(src, dict):
|
|
continue
|
|
ref = src.get("id") or src.get("doc_id") or src.get("path") or src.get("source") or ""
|
|
snippet = src.get("text") or src.get("snippet") or src.get("content") or ""
|
|
anchors.append(
|
|
SourceAnchor(
|
|
kind="kb",
|
|
ref=str(ref)[:200],
|
|
snippet=str(snippet)[:300],
|
|
)
|
|
)
|
|
return anchors
|
|
|
|
|
|
async def optional_rag_lookup(*, query: str, ctx) -> RagLookup:
|
|
"""Cheap, best-effort retrieval helper for block generators.
|
|
|
|
Lookup order (BookEngine v2):
|
|
|
|
1. Local exploration chunks attached to ``ctx`` (free, deterministic).
|
|
2. Live ``rag_search`` against ``ctx.primary_kb`` (network round-trip).
|
|
|
|
Returns an empty ``RagLookup`` if neither path produced anything; failures
|
|
are swallowed silently because every generator treats RAG as optional.
|
|
"""
|
|
if not query.strip():
|
|
return RagLookup()
|
|
|
|
# ── Step 1: try the cached exploration sweep first ───────────────
|
|
local_chunks = []
|
|
try:
|
|
local_chunks = ctx.relevant_chunks(query, limit=4)
|
|
except AttributeError:
|
|
local_chunks = []
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.debug(f"relevant_chunks failed: {exc}")
|
|
local_chunks = []
|
|
|
|
if local_chunks:
|
|
text = "\n\n".join(
|
|
f"- {(c.text or '').strip()}" for c in local_chunks if (c.text or "").strip()
|
|
)
|
|
anchors = [
|
|
SourceAnchor(
|
|
kind=c.source or "kb",
|
|
ref=str(c.ref or c.chunk_id or "")[:200],
|
|
snippet=str(c.text or "")[:300],
|
|
)
|
|
for c in local_chunks
|
|
if (c.text or "").strip()
|
|
]
|
|
if text or anchors:
|
|
return RagLookup(text=text, anchors=anchors, used=True)
|
|
|
|
# ── Step 2: fall back to a live RAG call ─────────────────────────
|
|
if not ctx.rag_enabled or not ctx.primary_kb:
|
|
return RagLookup()
|
|
|
|
try:
|
|
from deeptutor.tools.rag_tool import rag_search
|
|
|
|
result = await rag_search(query=query, kb_name=ctx.primary_kb)
|
|
except Exception as exc:
|
|
logger.debug(f"RAG lookup skipped ({ctx.primary_kb}): {exc}")
|
|
return RagLookup()
|
|
|
|
if not isinstance(result, dict):
|
|
return RagLookup()
|
|
|
|
answer = str(result.get("answer") or result.get("content") or "").strip()
|
|
sources = result.get("sources")
|
|
return RagLookup(
|
|
text=answer,
|
|
anchors=_coerce_anchors(sources if isinstance(sources, list) else None),
|
|
used=bool(answer or sources),
|
|
)
|
|
|
|
|
|
__all__ = ["RagLookup", "optional_rag_lookup"]
|