Las canalizaciones RAG construidas sobre almacenes de vectores estáticos responden preguntas a partir de datos obsoletos. Agregar una base de búsqueda en vivo significa que el LLM siempre tiene acceso a la información actual cuando el almacén de vectores se queda corto. Este tutorial crea un recuperador híbrido que primero verifica el almacén de vectores y luego recurre a la búsqueda en vivo cuando la confianza es baja. La capa de base de búsqueda utiliza Scavio para extraer datos de Google, Reddit y YouTube a $0,005 por consulta.
Requisitos previos
- Python 3.9+ instalado
- langchain, langchain-openai y faiss-cpu instalados
- Una clave API de Scavio de scavio.dev
- Una clave API de OpenAI para el LLM
Guia paso a paso
Paso 1: Construya el recuperador de puesta a tierra de búsqueda
Cree un recuperador que busque en la web contexto en tiempo real. A diferencia de un almacén de vectores, esto siempre devuelve información actual.
Python
import os, requests
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from typing import List
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
class SearchGroundingRetriever(BaseRetriever):
api_key: str = ''
num_results: int = 5
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.api_key = self.api_key or SCAVIO_KEY
def _get_relevant_documents(self, query: str) -> List[Document]:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.api_key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': self.num_results})
resp.raise_for_status()
return [Document(
page_content=f"{r['title']}\n{r.get('snippet', '')}",
metadata={'source': r['link'], 'type': 'search_grounding'}
) for r in resp.json().get('organic_results', [])]
grounding = SearchGroundingRetriever(num_results=5)
docs = grounding.invoke('latest LangChain features 2026')
print(f'Grounding returned {len(docs)} documents')
for d in docs:
print(f' {d.page_content[:60]}')Paso 2: Construya el perro perdiguero híbrido con lógica alternativa
Combine la recuperación de almacenes de vectores con la base de búsqueda. Si la tienda de vectores arroja resultados de baja relevancia (fragmentos cortos, pocas coincidencias), complételos automáticamente con una búsqueda en vivo.
Python
from langchain_core.retrievers import BaseRetriever
class HybridGroundedRetriever(BaseRetriever):
vector_retriever: BaseRetriever = None
search_retriever: BaseRetriever = None
min_vector_results: int = 2
min_content_length: int = 50
def _get_relevant_documents(self, query: str) -> List[Document]:
# Try vector store first
vector_docs = []
if self.vector_retriever:
vector_docs = self.vector_retriever.invoke(query)
# Check if vector results are sufficient
quality_docs = [d for d in vector_docs
if len(d.page_content) >= self.min_content_length]
if len(quality_docs) >= self.min_vector_results:
return quality_docs
# Supplement with live search grounding
search_docs = self.search_retriever.invoke(query)
# Merge: vector docs first, then search docs
seen_content = set(d.page_content[:50] for d in quality_docs)
for sd in search_docs:
if sd.page_content[:50] not in seen_content:
quality_docs.append(sd)
seen_content.add(sd.page_content[:50])
return quality_docs
# Setup
hybrid = HybridGroundedRetriever(
search_retriever=SearchGroundingRetriever(num_results=5),
min_vector_results=2
)
docs = hybrid.invoke('latest Python release date 2026')
print(f'Hybrid returned {len(docs)} docs')
for d in docs:
source_type = d.metadata.get('type', 'vector')
print(f' [{source_type}] {d.page_content[:50]}')Paso 3: Conéctese a una cadena de control de calidad de LangChain
Conecte el perro perdiguero híbrido a una cadena RetrievalQA. La cadena obtiene automáticamente respuestas fundamentadas cuando el almacén de vectores carece de datos actuales.
Python
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model='gpt-4o-mini', temperature=0)
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type='stuff',
retriever=hybrid,
return_source_documents=True,
chain_type_kwargs={
'prompt': None # Uses default prompt
}
)
def ask(question: str) -> dict:
result = qa_chain.invoke({'query': question})
sources = []
for doc in result.get('source_documents', []):
source_type = doc.metadata.get('type', 'vector')
source_url = doc.metadata.get('source', 'local')
sources.append({'type': source_type, 'url': source_url})
grounded = any(s['type'] == 'search_grounding' for s in sources)
return {
'answer': result['result'],
'grounded': grounded,
'sources': sources,
'cost': 0.005 if grounded else 0
}
result = ask('What are the newest LangChain features in 2026?')
print(f'Answer: {result["answer"][:200]}')
print(f'Grounded: {result["grounded"]}')
print(f'Cost: ${result["cost"]}')
for s in result['sources'][:3]:
print(f' [{s["type"]}] {s["url"]}')Paso 4: Agregar decisiones de base y seguimiento de costos
Realice un seguimiento de cuándo se activa la conexión a tierra y cuánto cuesta. Esto ayuda a optimizar el almacén de vectores para reducir las llamadas de búsqueda innecesarias.
Python
class GroundingTracker:
def __init__(self):
self.total_queries = 0
self.grounded_queries = 0
self.total_cost = 0
self.grounding_triggers = []
def record(self, query: str, grounded: bool, cost: float):
self.total_queries += 1
if grounded:
self.grounded_queries += 1
self.total_cost += cost
self.grounding_triggers.append(query)
def report(self) -> str:
pct = (self.grounded_queries / self.total_queries * 100) if self.total_queries else 0
lines = [
f'Grounding Report',
f'Total queries: {self.total_queries}',
f'Grounded: {self.grounded_queries} ({pct:.0f}%)',
f'Vector-only: {self.total_queries - self.grounded_queries}',
f'Search cost: ${self.total_cost:.3f}',
f'',
f'Recent grounding triggers:'
]
for q in self.grounding_triggers[-5:]:
lines.append(f' - {q}')
return '\n'.join(lines)
tracker = GroundingTracker()
test_queries = [
'What is a Python decorator?', # Vector store likely has this
'Latest Python 3.15 release date', # Needs grounding
'LangChain v0.4 breaking changes 2026', # Needs grounding
]
for q in test_queries:
result = ask(q)
tracker.record(q, result['grounded'], result['cost'])
print(tracker.report())Ejemplo en Python
Python
import os, requests
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
from typing import List
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
class SearchGroundingRetriever(BaseRetriever):
api_key: str = ''
num_results: int = 5
def __init__(self, **kw):
super().__init__(**kw)
self.api_key = self.api_key or SCAVIO_KEY
def _get_relevant_documents(self, query: str) -> List[Document]:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.api_key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': self.num_results})
return [Document(page_content=f"{r['title']}\n{r.get('snippet','')}",
metadata={'source': r['link']}) for r in resp.json().get('organic_results', [])]
retriever = SearchGroundingRetriever()
docs = retriever.invoke('LangChain RAG grounding 2026')
for d in docs:
print(f"{d.page_content[:60]}\n {d.metadata['source']}")Ejemplo en JavaScript
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
async function searchGrounding(query, num = 5) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us', num_results: num })
});
return (await resp.json()).organic_results?.map(r => ({
pageContent: `${r.title}\n${r.snippet || ''}`,
metadata: { source: r.link, type: 'search_grounding' }
})) || [];
}
async function hybridRetrieve(query, vectorDocs = []) {
if (vectorDocs.length >= 2) return vectorDocs;
const searchDocs = await searchGrounding(query);
return [...vectorDocs, ...searchDocs];
}
hybridRetrieve('LangChain features 2026').then(docs => {
docs.forEach(d => console.log(`[${d.metadata.type}] ${d.pageContent.slice(0, 50)}`));
});Salida esperada
JSON
Grounding returned 5 documents
Latest LangChain Features and Updates 2026
LangChain v0.4 Release Notes
Hybrid returned 5 docs
[search_grounding] Latest Python 3.15 Released October
Grounding Report
Total queries: 3
Grounded: 2 (67%)
Vector-only: 1
Search cost: $0.010
Recent grounding triggers:
- Latest Python 3.15 release date
- LangChain v0.4 breaking changes 2026