基于静态向量存储构建的 RAG 管道可以回答陈旧数据的问题。添加实时搜索基础意味着当向量存储不足时，法学硕士始终可以访问当前信息。本教程构建了一个混合检索器，它首先检查向量存储，然后在置信度较低时回退到实时搜索。搜索基础层使用 Scavio 从 Google、Reddit 和 YouTube 获取每个查询 0.005 美元的费用。
前置条件
- 已安装 Python 3.9+
- 安装 langchain、langchain-openai 和 faiss-cpu
- 来自 scavio.dev 的 Scavio API 密钥
- LLM 的 OpenAI API 密钥
操作指南
步骤 1: 构建搜索接地检索器
创建一个在网络上搜索实时上下文的检索器。与向量存储不同，它始终返回当前信息。
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]}')步骤 2: 构建具有后备逻辑的混合检索器
将矢量存储检索与搜索基础相结合。如果矢量存储返回低相关性结果（短片段、很少匹配），则会自动补充实时搜索。
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]}')步骤 3: 接入 LangChain QA 链
将混合检索器连接到 RetrievalQA 链。当向量存储缺乏当前数据时，链会自动获得可靠的答案。
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"]}')步骤 4: 添加接地决策和成本跟踪
跟踪何时触发接地以及需要花费多少费用。这有助于优化向量存储以减少不必要的搜索调用。
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())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']}")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)}`));
});预期输出
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