结合网络搜索与 RAG
混合检索：本地向量存储加实时网络搜索，获取最新答案。
结合网络搜索与 RAG 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
混合检索的挑战
大多数实际应用中的智能体都需要两类知识：静态领域知识（公司文档、产品手册、政策）和当前信息（新闻、实时价格、近期事件）。
本地向量存储处理前者，网页搜索处理后者。将两者结合可以兼得其优势。
架构：本地 RAG + 网页搜索
混合系统会将每个问题路由到合适的检索来源：
- 向量存储（RAG） — 已建立索引的文档、稳定知识、私有数据
- 网页搜索 — 当前事件、近期发布内容、实时数据
- 两者 — 当问题同时需要文档中的上下文和当前信息时
class HybridRetrievalAgent:
def __init__(self, vector_store, search_client):
self.vector_store = vector_store # e.g., ChromaDB or FAISS
self.search_client = search_client # e.g., TavilyClient
def answer(self, question):
route = self.classify_question(question)
if route == 'static':
context = self.vector_store.query(question, n_results=5)
elif route == 'current':
context = self.web_search(question)
else: # 'both'
local = self.vector_store.query(question, n_results=3)
web = self.web_search(question)
context = local + web
return self.generate_answer(question, context)
if __name__ == '__main__':
class DemoAgent(HybridRetrievalAgent):
def classify_question(self, question):
return 'static' if 'company' in question.lower() else 'current'
def web_search(self, question):
return [('Web result about ' + question, {})]
def generate_answer(self, question, context):
return f'Answer using {len(context)} context item(s).'
class FakeVectorStore:
def query(self, question, n_results=5):
return [('Local doc snippet', {})]
agent = DemoAgent(FakeVectorStore(), search_client=None)
print(agent.answer('What is our company policy on refunds?'))
路由分类器
分类器决定使用哪种检索来源。你可以将其实现为 LLM 调用、关键词规则集或小型训练分类器。基于 LLM 的路由器最为灵活。
ROUTING_PROMPT = '''Classify this question into one of three categories:
- "static": answered from company documents, product docs, or stable technical knowledge
- "current": requires up-to-date information (news, prices, recent events, latest releases)
- "both": needs both company context and current information
Question: {question}
Respond with exactly one word: static, current, or both.'''
def classify_question(question):
response = llm_call(ROUTING_PROMPT.format(question=question))
route = response.strip().lower()
if route not in ('static', 'current', 'both'):
return 'both' # safe default
return route设置本地向量存储
对于静态知识库，请使用 ChromaDB，这是一种在进程内运行的轻量级向量数据库。只需为文档建立一次索引，然后在运行时进行查询。
使用 pip install chromadb openai 进行安装。
import chromadb
from chromadb.utils import embedding_functions
import os
client = chromadb.PersistentClient(path='./vector_db')
ef = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv('OPENAI_API_KEY'),
model_name='text-embedding-3-small'
)
collection = client.get_or_create_collection(
name='company_docs',
embedding_function=ef
)
def index_document(doc_id, text, metadata=None):
collection.add(
ids=[doc_id],
documents=[text],
metadatas=[metadata or {}]
)
def local_retrieve(question, n_results=5):
results = collection.query(
query_texts=[question],
n_results=n_results
)
return list(zip(results['documents'][0], results['metadatas'][0]))网页搜索检索
网页搜索路径使用 Tavily 获取当前信息。请统一格式化结果，以便将其与本地 RAG 结果合并到相同的提示词结构中。
from tavily import TavilyClient
import os
tavily = TavilyClient(api_key=os.getenv('TAVILY_API_KEY'))
def web_retrieve(question, n_results=3):
results = tavily.search(
query=question,
max_results=n_results,
search_depth='basic'
)
# Normalize to same format as local results
return [
(
r['content'][:600], # text
{'source': r['url'], 'title': r['title'], 'type': 'web'} # metadata
)
for r in results.get('results', [])
]合并本地结果与网页结果
同时使用两种来源时，合并结果并为每个结果标记来源。这样 LLM 就能合理地权衡它们：本地文档用于公司特定事实，网页用于当前数据。
def merge_results(local_results, web_results):
merged = []
for text, meta in local_results:
merged.append({
'content': text,
'source': meta.get('source', 'internal document'),
'type': 'local',
'title': meta.get('title', 'Company Document')
})
for text, meta in web_results:
merged.append({
'content': text,
'source': meta.get('source', 'web'),
'type': 'web',
'title': meta.get('title', 'Web Result')
})
return merged
def format_merged_for_prompt(merged_results):
parts = []
for i, r in enumerate(merged_results, 1):
tag = '[INTERNAL]' if r['type'] == 'local' else '[WEB]'
parts.append(f'[{i}] {tag} {r["title"]}\n{r["content"]}')
return '\n\n'.join(parts)
if __name__ == '__main__':
local = [('Refunds are processed within 5 business days.', {'source': 'handbook', 'title': 'Refund Policy'})]
web = [('Company X reported Q2 earnings today.', {'source': 'reuters.com', 'title': 'Q2 Earnings'})]
merged = merge_results(local, web)
print(format_merged_for_prompt(merged))
检测对时效敏感的问题
除了使用 LLM 分类器，还可以使用关键词启发式规则来检测需要当前信息的问题。对于明显的情况，这种方式更快，也能避免额外的 LLM 调用。
CURRENT_EVENTS_SIGNALS = [
'latest', 'current', 'today', 'now', 'recent',
'this week', 'this month', 'this year',
'just released', 'new version', 'updated',
'price', 'stock', 'news', 'announcement',
'2024', '2025'
]
STATIC_SIGNALS = [
'how does', 'what is', 'explain', 'tutorial',
'documentation', 'our product', 'company policy',
'internal', 'handbook'
]
def fast_route(question):
lower = question.lower()
current_score = sum(1 for s in CURRENT_EVENTS_SIGNALS if s in lower)
static_score = sum(1 for s in STATIC_SIGNALS if s in lower)
if current_score > static_score:
return 'current'
elif static_score > current_score:
return 'static'
else:
return 'both'
if __name__ == '__main__':
for q in ['What is our company handbook policy on PTO?', 'What is the latest stock price today?']:
print(f'{fast_route(q)!r} <- "{q}"')
处理来源之间的冲突
当本地文档表达一种说法，而网页结果表达另一种说法时，就会发生冲突。例如：你的内部定价文档显示价格为 $50/月，但网页结果显示价格已变更为 $80/月。
请指示 LLM 标记冲突，并对时间敏感的事实优先采用网页来源。
HYBRID_ANSWER_PROMPT = '''You are answering a question using two types of sources:
- [INTERNAL] sources: company documents (may be outdated)
- [WEB] sources: current web information
For factual claims about current state (prices, versions, availability):
PREFER [WEB] sources over [INTERNAL] ones.
For company-specific processes, policies, and architecture:
PREFER [INTERNAL] sources.
If sources conflict, note the discrepancy in your answer.
Sources:
{sources}
Question: {question}
Answer:'''
def generate_hybrid_answer(question, merged_results):
sources_text = format_merged_for_prompt(merged_results)
return llm_call(HYBRID_ANSWER_PROMPT.format(
sources=sources_text,
question=question
))检测本地文档是否过时
本地文档会随时间推移而过时。请添加过时性检查：如果本地文档早于某个时间阈值，即使路由器将问题分类为“静态”，也应通过网页搜索补充信息。
from datetime import datetime, timedelta
STALENESS_THRESHOLD_DAYS = 90
def check_staleness(metadata):
indexed_at = metadata.get('indexed_at')
if not indexed_at:
return False # unknown age — assume fresh
indexed_date = datetime.fromisoformat(indexed_at)
age = datetime.now() - indexed_date
return age > timedelta(days=STALENESS_THRESHOLD_DAYS)
def smart_retrieve(question, route):
local_results = []
web_results = []
if route in ('static', 'both'):
local_results = local_retrieve(question, n_results=4)
# Check if any local results are stale
stale = any(check_staleness(meta) for _, meta in local_results)
if stale:
print('Stale local docs — adding web search')
web_results = web_retrieve(question, n_results=2)
if route in ('current', 'both'):
web_results = web_retrieve(question, n_results=3)
return merge_results(local_results, web_results)置信度评分
为每条检索到的上下文附加置信度分数。高置信度来源（较新、来自权威域名、嵌入相似度高）应在最终答案中获得更大的权重。
def score_result(result, query_embedding):
score = 0.5 # base score
# Recency bonus for web results
if result.get('type') == 'web':
pub_date = result.get('published_date', '')
if '2024' in pub_date or '2025' in pub_date:
score += 0.2
# Embedding similarity to query
if result.get('content'):
result_emb = embed(result['content'][:500])
sim = cosine_similarity(query_embedding, result_emb)
score += sim * 0.3
# Domain authority
from urllib.parse import urlparse
domain = urlparse(result.get('source', '')).netloc
if any(auth in domain for auth in ['docs.', 'developer.', 'official.']):
score += 0.1
return min(score, 1.0)完整的混合检索流程
将所有步骤整合起来：快速路由 → 从一个或两个来源进行智能检索 → 补充过时信息 → 合并 → 评分 → 格式化 → 生成答案。
def hybrid_answer(question):
# 1. Route (fast heuristic first, LLM fallback for ambiguous)
route = fast_route(question)
if route == 'both':
route = classify_question(question) # LLM for ambiguous cases
print(f'Route: {route}')
# 2. Retrieve
merged = smart_retrieve(question, route)
if not merged:
return 'I could not find relevant information to answer your question.'
# 3. Generate
answer = generate_hybrid_answer(question, merged)
return answer
# Usage
print(hybrid_answer('What is our refund policy?')) # -> static/local
print(hybrid_answer('What is GPT-4 pricing today?')) # -> current/web知识检查
混合检索智能体应在何时优先采用网页搜索结果，而不是本地文档结果？
回顾：将网页搜索与 RAG 结合
混合检索将本地向量存储（用于静态、私有或特定领域的知识）与网页搜索（用于当前的公开信息）结合起来。路由分类器会将每个问题定向到合适的来源；在需要时也可以同时使用两者。
关键技术包括：使用关键词信号进行快速启发式路由、检测本地文档是否过时、在提示词中加入冲突解决指令，以及通过置信度评分为检索到的上下文分配权重。
常见问题解答
「结合网络搜索与 RAG」课时是免费的吗？
是的 — 「结合网络搜索与 RAG」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「结合网络搜索与 RAG」这节课中我会学到什么？
混合检索：本地向量存储加实时网络搜索，获取最新答案。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。
「结合网络搜索与 RAG」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 使用 Tavily 和 SerpAPI 搜索
- 对搜索结果进行排序与筛选
- 深度研究循环模式
- 结合网络搜索与 RAG