在本教程中，我們通過結合Tavily Search API，Chroma，Google Gemini LLM和Langchain Framework的優勢來展示如何構建強大而聰明的提問系統。該管道利用Tavily，使用Chroma Vector Store的語義文檔緩存以及通過Gemini模型生成上下文響應的實時Web搜索。這些工具是通過Langchain的模塊化組件集成的，例如RunnableLambda，ChatPromptTemplate，ConsingBufferMemory和GoogleGenerativeAiembeddings。通過引入混合檢索機制，它超越了簡單的問答，該機制在調用新的Web搜索之前檢查了緩存的嵌入。檢索到的文檔是智能格式，匯總並通過結構化的LLM提示，並註意源歸因，用戶歷史記錄和信心評分。諸如高級及時工程，情感和實體分析以及動態矢量商店更新之類的關鍵功能使該管道適用於高級用例，例如研究幫助，特定領域的摘要和智能代理。
!pip install -qU langchain-community tavily-python langchain-google-genai streamlit matplotlib pandas tiktoken chromadb langchain_core pydantic langchain我們安裝併升級了建立高級AI搜索助手所需的一組綜合庫。它包括用於檢索的工具（Tavily-Python，Chromadb），LLM集成（Langchain-Google-Genai，Langchain），數據處理（Pandas，Pydantic），可視化（Matplotlib，Selllit）和Swikenization（Tiktoken）。這些組件構成了構建實時，上下文感知的質量檢查系統的核心基礎。
import os
import getpass
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np
import json
import time
from typing import List, Dict, Any, Optional
from datetime import datetime我們導入整個筆記本中使用的必需Python庫。它包括用於環境變量的標準庫，安全輸入，時間跟踪和數據類型（OS，GEGPASS，時間，打字，DateTime）。此外，它為數據處理，可視化和數值計算以及用於解析結構化數據的JSON提供了核心數據科學工具，例如熊貓，matplotlib和numpy。
if "TAVILY_API_KEY" not in os.environ:
os.environ("TAVILY_API_KEY") = getpass.getpass("Enter Tavily API key: ")
if "GOOGLE_API_KEY" not in os.environ:
os.environ("GOOGLE_API_KEY") = getpass.getpass("Enter Google API key: ")
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)我們只有在用戶尚未在環境中設置的用戶來確保安全可重複地訪問外部服務時，我們可以通過提示用戶來安全地將API鍵和Google Gemini初始化。它還使用Python的記錄模塊配置標準化的記錄設置，該模塊有助於監視執行流並捕獲整個筆記本中的調試或錯誤消息。
from langchain_community.retrievers import TavilySearchAPIRetriever
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_google_genai import ChatGoogleGenerativeAI, GoogleGenerativeAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains.summarize import load_summarize_chain
from langchain.memory import ConversationBufferMemory我們從Langchain生態系統及其集成中導入關鍵組件。它帶來了用於實時Web搜索的TavilySearchApiREtriever，用於矢量存儲的Chroma和GoogleGenerativeAi模塊用於聊天和嵌入模型。核心Langchain模塊，例如ChatPromptTemplate，RunnableLambda，ConvertyBufferMemory和輸出解析器，可啟用靈活的及時構建，內存處理和管道執行。
class SearchQueryError(Exception):
"""Exception raised for errors in the search query."""
pass
def format_docs(docs):
formatted_content = ()
for i, doc in enumerate(docs):
metadata = doc.metadata
source = metadata.get('source', 'Unknown source')
title = metadata.get('title', 'Untitled')
score = metadata.get('score', 0)
formatted_content.append(
f"Document {i+1} (Score: {score:.2f}):n"
f"Title: {title}n"
f"Source: {source}n"
f"Content: {doc.page_content}n"
)
return "nn".join(formatted_content)我們定義了兩個用於搜索和文檔處理的基本組件。 SearchQueryError類創建了自定義異常，以優雅地管理無效或失敗的搜索查詢。 Format_Docs函數通過提取元數據（例如標題，源和相關得分）並將其格式化為清潔，可讀的字符串來處理檢索的文檔列表。
class SearchResultsParser:
def parse(self, text):
try:
if isinstance(text, str):
import re
import json
json_match = re.search(r'{.*}', text, re.DOTALL)
if json_match:
json_str = json_match.group(0)
return json.loads(json_str)
return {"answer": text, "sources": (), "confidence": 0.5}
elif hasattr(text, 'content'):
return {"answer": text.content, "sources": (), "confidence": 0.5}
else:
return {"answer": str(text), "sources": (), "confidence": 0.5}
except Exception as e:
logger.warning(f"Failed to parse JSON: {e}")
return {"answer": str(text), "sources": (), "confidence": 0.5}SearchResultsparser類提供了一種可靠的方法，用於從LLM響應中提取結構化信息。它試圖從模型輸出中解析類似JSON的字符串，如果解析失敗，則返回純文本響應格式。它優雅地處理字符串輸出和消息對象，確保下游處理一致。如果發生錯誤，它會記錄警告並返回包含原始答案，空源和默認置信度得分的後備響應，從而增強了系統的容錯性。
class EnhancedTavilyRetriever:
def __init__(self, api_key=None, max_results=5, search_depth="advanced", include_domains=None, exclude_domains=None):
self.api_key = api_key
self.max_results = max_results
self.search_depth = search_depth
self.include_domains = include_domains or ()
self.exclude_domains = exclude_domains or ()
self.retriever = self._create_retriever()
self.previous_searches = ()
def _create_retriever(self):
try:
return TavilySearchAPIRetriever(
api_key=self.api_key,
k=self.max_results,
search_depth=self.search_depth,
include_domains=self.include_domains,
exclude_domains=self.exclude_domains
)
except Exception as e:
logger.error(f"Failed to create Tavily retriever: {e}")
raise
def invoke(self, query, **kwargs):
if not query or not query.strip():
raise SearchQueryError("Empty search query")
try:
start_time = time.time()
results = self.retriever.invoke(query, **kwargs)
end_time = time.time()
search_record = {
"timestamp": datetime.now().isoformat(),
"query": query,
"num_results": len(results),
"response_time": end_time - start_time
}
self.previous_searches.append(search_record)
return results
except Exception as e:
logger.error(f"Search failed: {e}")
raise SearchQueryError(f"Failed to perform search: {str(e)}")
def get_search_history(self):
return self.previous_searches增強的TavilyReTriever類是圍繞TavilySearchApirEtriever的自定義包裝器，為搜索操作增加了更大的靈活性，控制和可追溯性。它支持高級功能，例如限制搜索深度，域包含/排除過濾器以及可配置的結果計數。 Invoke方法執行Web搜索並跟踪每個查詢的元數據（時間戳，響應時間和結果計數），將其存儲為以後的分析。
class SearchCache:
def __init__(self):
self.embedding_function = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
self.vector_store = None
self.text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
def add_documents(self, documents):
if not documents:
return
try:
if self.vector_store is None:
self.vector_store = Chroma.from_documents(
documents=documents,
embedding=self.embedding_function
)
else:
self.vector_store.add_documents(documents)
except Exception as e:
logger.error(f"Failed to add documents to cache: {e}")
def search(self, query, k=3):
if self.vector_store is None:
return ()
try:
return self.vector_store.similarity_search(query, k=k)
except Exception as e:
logger.error(f"Vector search failed: {e}")
return ()SearchCache類實現了一個語義緩存層，該語義緩存層使用向量嵌入來存儲和檢索文檔以進行有效的相似性搜索。它使用GoogleGenerativeAiembedDings將文檔轉換為密集的向量並將其存儲在色度矢量數據庫中。 add_documents方法初始化或更新矢量存儲，而搜索方法可以基於語義相似性快速檢索最相關的加速文檔。這減少了冗餘API調用並改善了重複或相關查詢的響應時間，並用作AI Assistant Pipeline中的輕量級混合記憶層。
search_cache = SearchCache()
enhanced_retriever = EnhancedTavilyRetriever(max_results=5)
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
system_template = """You are a research assistant that provides accurate answers based on the search results provided.
Follow these guidelines:
1. Only use the context provided to answer the question
2. If the context doesn't contain the answer, say "I don't have sufficient information to answer this question."
3. Cite your sources by referencing the document numbers
4. Don't make up information
5. Keep the answer concise but complete
Context: {context}
Chat History: {chat_history}
"""
system_message = SystemMessagePromptTemplate.from_template(system_template)
human_template = "Question: {question}"
human_message = HumanMessagePromptTemplate.from_template(human_template)
prompt = ChatPromptTemplate.from_messages((system_message, human_message))
我們初始化了AI助手的核心組件：語義搜索曲目，用於基於Web的查詢的EnhancedTavilyReTriever以及一個跨回合保留聊天歷史記錄的對話袋。它還使用ChatPromptTemplate定義了結構化提示，指導LLM充當研究助理。該及時執行嚴格的規則，以確保事實準確性，上下文使用，源引用和簡潔的答复，從而確保可靠和紮根的回應。
def get_llm(model_name="gemini-2.0-flash-lite", temperature=0.2, response_mode="json"):
try:
return ChatGoogleGenerativeAI(
model=model_name,
temperature=temperature,
convert_system_message_to_human=True,
top_p=0.95,
top_k=40,
max_output_tokens=2048
)
except Exception as e:
logger.error(f"Failed to initialize LLM: {e}")
raise
output_parser = SearchResultsParser()
我們定義get_llm函數，該函數初始化了具有可配置參數的Google Gemini語言模型，例如模型名稱，溫度和解碼設置（例如，TOP_P，TOP_K和MAX代幣）。它可確保使用失敗的模型初始化的錯誤處理，可確保魯棒性。還創建了搜索Resultsparser的一個實例，以標準化和構建LLM的原始響應，從而使答案和元數據的下游處理一致。
def plot_search_metrics(search_history):
if not search_history:
print("No search history available")
return
df = pd.DataFrame(search_history)
plt.figure(figsize=(12, 6))
plt.subplot(1, 2, 1)
plt.plot(range(len(df)), df('response_time'), marker="o")
plt.title('Search Response Times')
plt.xlabel('Search Index')
plt.ylabel('Time (seconds)')
plt.grid(True)
plt.subplot(1, 2, 2)
plt.bar(range(len(df)), df('num_results'))
plt.title('Number of Results per Search')
plt.xlabel('Search Index')
plt.ylabel('Number of Results')
plt.grid(True)
plt.tight_layout()
plt.show()
plot_search_metrics函數可視化使用matplotlib從過去查詢中的性能趨勢。它將搜索歷史記錄轉換為數據框，並繪製了兩個子圖：一個顯示每個搜索響應時間，另一個顯示返回的結果數。這有助於分析系統的效率和搜索質量，隨著時間的流逝，幫助開發人員微調了獵犬或在現實世界中識別瓶頸。
def retrieve_with_fallback(query):
cached_results = search_cache.search(query)
if cached_results:
logger.info(f"Retrieved {len(cached_results)} documents from cache")
return cached_results
logger.info("No cache hit, performing web search")
search_results = enhanced_retriever.invoke(query)
search_cache.add_documents(search_results)
return search_results
def summarize_documents(documents, query):
llm = get_llm(temperature=0)
summarize_prompt = ChatPromptTemplate.from_template(
"""Create a concise summary of the following documents related to this query: {query}
{documents}
Provide a comprehensive summary that addresses the key points relevant to the query.
"""
)
chain = (
{"documents": lambda docs: format_docs(docs), "query": lambda _: query}
| summarize_prompt
| llm
| StrOutputParser()
)
return chain.invoke(documents)這兩個功能增強了助手的智力和效率。 retirevie_with_fallback函數實現了混合檢索機制：它首先嘗試從本地色度cache獲取語義相關文檔，如果不成功，則歸還回到實時tavily Web搜索，從而緩存了新的結果，以備將來使用。同時，摘要_documents利用雙子座LLM從檢索到的文檔中生成簡潔的摘要，並在結構化提示的指導下，以確保與查詢相關。它們共同實現了低延遲，信息性和背景感知的反應。
def advanced_chain(query_engine="enhanced", model="gemini-1.5-pro", include_history=True):
llm = get_llm(model_name=model)
if query_engine == "enhanced":
retriever = lambda query: retrieve_with_fallback(query)
else:
retriever = enhanced_retriever.invoke
def chain_with_history(input_dict):
query = input_dict("question")
chat_history = memory.load_memory_variables({})("chat_history") if include_history else ()
docs = retriever(query)
context = format_docs(docs)
result = prompt.invoke({
"context": context,
"question": query,
"chat_history": chat_history
})
memory.save_context({"input": query}, {"output": result.content})
return llm.invoke(result)
return RunnableLambda(chain_with_history) | StrOutputParser()Advanced_chain函數定義了一個模塊化的端到端推理工作流，用於使用緩存或實時搜索來回答用戶查詢。它初始化了指定的雙子座模型，選擇檢索策略（緩存的後備或直接搜索），構造了響應管道，該響應管道包含聊天歷史記錄（如果已啟用），將文檔格式化為上下文，並使用系統引導的模板提示LLM。該鏈還記錄了內存中的相互作用，並返回最終答案，並解析為乾淨的文本。該設計可以通過模型和檢索策略進行靈活的實驗，同時保持對話連貫性。
qa_chain = advanced_chain()
def analyze_query(query):
llm = get_llm(temperature=0)
analysis_prompt = ChatPromptTemplate.from_template(
"""Analyze the following query and provide:
1. Main topic
2. Sentiment (positive, negative, neutral)
3. Key entities mentioned
4. Query type (factual, opinion, how-to, etc.)
Query: {query}
Return the analysis in JSON format with the following structure:
{{
"topic": "main topic",
"sentiment": "sentiment",
"entities": ("entity1", "entity2"),
"type": "query type"
}}
"""
)
chain = analysis_prompt | llm | output_parser
return chain.invoke({"query": query})
print("Advanced Tavily-Gemini Implementation")
print("="*50)
query = "what year was breath of the wild released and what was its reception?"
print(f"Query: {query}")我們初始化了智能助手的最終組成部分。 QA_Chain是組裝的推理管道，準備使用基於檢索，內存和基於GEMINI的響應生成來處理用戶查詢。 Analyze_Query函數使用Gemini模型和結構化的JSON提示進行了對查詢，提取主要主題，情感，實體和查詢類型的輕量級語義分析。示例查詢有關野外釋放和接收的呼吸的示例，展示瞭如何觸發助手並為全棧推理和語義解釋做好準備。打印的標題標誌著交互式執行的開始。
try:
print("nSearching for answer...")
answer = qa_chain.invoke({"question": query})
print("nAnswer:")
print(answer)
print("nAnalyzing query...")
try:
query_analysis = analyze_query(query)
print("nQuery Analysis:")
print(json.dumps(query_analysis, indent=2))
except Exception as e:
print(f"Query analysis error (non-critical): {e}")
except Exception as e:
print(f"Error in search: {e}")
history = enhanced_retriever.get_search_history()
print("nSearch History:")
for i, h in enumerate(history):
print(f"{i+1}. Query: {h('query')} - Results: {h('num_results')} - Time: {h('response_time'):.2f}s")
print("nAdvanced search with domain filtering:")
specialized_retriever = EnhancedTavilyRetriever(
max_results=3,
search_depth="advanced",
include_domains=("nintendo.com", "zelda.com"),
exclude_domains=("reddit.com", "twitter.com")
)
try:
specialized_results = specialized_retriever.invoke("breath of the wild sales")
print(f"Found {len(specialized_results)} specialized results")
summary = summarize_documents(specialized_results, "breath of the wild sales")
print("nSummary of specialized results:")
print(summary)
except Exception as e:
print(f"Error in specialized search: {e}")
print("nSearch Metrics:")
plot_search_metrics(history)
我們演示了完整的行動管道。它使用qa_chain執行搜索，顯示生成的答案，然後分析情感，主題，實體和類型的查詢。它還可以檢索並打印每個查詢的搜索歷史記錄，響應時間和結果計數。此外，它運行了一個針對與任天堂相關的站點的域過濾搜索，總結了結果，並使用Plot_Search_Metrics可視化搜索性能，從而在實時使用中對助手的功能提供了全面的視圖。
總之，遵循本教程為用戶提供了全面的藍圖，用於創建一個高功能，上下文感知和可擴展的抹布系統，該系統將實時的Web智能與對話式AI橋接起來。 Tavely搜索API使用戶可以直接從Web中獲取新的和相關的內容。 Gemini LLM添加了強大的推理和摘要功能，而Langchain的抽象層則可以在內存，嵌入和模型輸出之間進行無縫的編排。該實現包括高級功能，例如特定於域特異性過濾，查詢分析（情感，主題和實體提取）以及使用Chroma和GoogleGenerativeAiembeddings構建的語義矢量緩存的後備策略。此外，結構化的日誌記錄，錯誤處理和分析儀表板為現實部署提供了透明度和診斷。
查看Colab筆記本。 這項研究的所有信用都歸該項目的研究人員。另外，請隨時關注我們 嘰嘰喳喳 而且不要忘記加入我們的 90K+ ml子雷迪特。
Asif Razzaq是Marktechpost Media Inc.的首席執行官。作為一位有遠見的企業家和工程師，ASIF致力於利用人工智能的潛力來實現社會利益。他最近的努力是推出了人工智能媒體平台Marktechpost，該平台的深入覆蓋了機器學習和深度學習新聞，既在技術上都可以聽起來，既可以通過技術上的聲音，又可以被廣泛的受眾理解。該平台每月有超過200萬個觀點，說明了其在受眾中的受歡迎程度。
🚨建立您可以信任的Genai。 ⭐️Parlant是您的開源引擎，用於受控，合規性和有目的的AI對話 – Github上的Star Parlant！ （晉升）
