En este tutorial, demostramos cómo construir un sistema de preguntas poderoso poderoso e inteligente combinando las fortalezas de API de búsqueda tavilia, CromaGoogle Gemini LLMS y el marco Langchain. La tubería aprovecha la búsqueda web en tiempo real utilizando el almacenamiento en caché de documentos semánticos de Tavily con el almacén de vectores de Chroma y la generación de respuesta contextual a través del modelo Gemini. Estas herramientas se integran a través de los componentes modulares de Langchain, como Runnablelambda, ChatPromptTemplate, Conversación BufferMemory y GoogleGenerativeaiembeddings. Va más allá de las preguntas y respuestas simples mediante la introducción de un mecanismo de recuperación híbrido que verifica los incrustaciones en caché antes de invocar búsquedas web frescas. Los documentos recuperados están formateados de manera inteligente, resumidos y pasados a través de un indicador de LLM estructurado, con atención a la atribución de la fuente, el historial del usuario y la puntuación de la confianza. Las funciones clave, como la ingeniería rápida avanzada, el análisis de sentimientos y la entidad, y las actualizaciones de la tienda vectorial dinámica hacen que esta tubería sea adecuada para casos de uso avanzado como asistencia de investigación, resumen específico del dominio y agentes inteligentes.
!pip install -qU langchain-community tavily-python langchain-google-genai streamlit matplotlib pandas tiktoken chromadb langchain_core pydantic langchainInstalamos y actualizamos un conjunto integral de bibliotecas necesarias para construir un asistente de búsqueda de IA avanzado. Incluye herramientas para la recuperación (Tavily-Python, ChromAdB), integración LLM (Langchain-Google-Genai, Langchain), Manejo de datos (Pandas, Pydantic), Visualización (Matplotlib, Streamlit) y Tokenization (Tiktoken). Estos componentes forman la base central para construir un sistema de control de calidad en tiempo real consciente de contexto.
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 datetimeImportamos bibliotecas esenciales de Python utilizadas en todo el cuaderno. Incluye bibliotecas estándar para variables de entorno, entrada segura, seguimiento de tiempo y tipos de datos (OS, GetPass, Time, Typing, DateTime). Además, trae herramientas de ciencia de datos básicas como pandas, matplotlib y numpy para el manejo de datos, la visualización y los cálculos numéricos, así como JSON para analizar datos estructurados.
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__)Inicializamos de forma segura las claves API para Tavily y Google Gemini solicitando a los usuarios solo si aún no están configurados en el entorno, asegurando un acceso seguro y repetible a servicios externos. También configura una configuración de registro estandarizada utilizando el módulo de registro de Python, que ayuda a monitorear el flujo de ejecución y capturar mensajes de depuración o error en todo el cuaderno.
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 ConversationBufferMemoryImportamos componentes clave del ecosistema Langchain y sus integraciones. Aporta a TavilySearchApiretriever para la búsqueda web en tiempo real, el Chroma para el almacenamiento vectorial y los módulos de GoogleGenerativeai para los modelos de chat e incrustación. Los módulos Core Langchain como ChatPromptTemplate, RunNablelambda, Conversación BufferMemory y los analizadores de salida permiten la construcción de aviso flexible, el manejo de la memoria y la ejecución de la tubería.
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)Definimos dos componentes esenciales para la búsqueda y el manejo de documentos. La clase SearchQueryError crea una excepción personalizada para administrar consultas de búsqueda no válidas o fallidas con gracia. La función Format_Docs procesa una lista de documentos recuperados extrayendo metadatos como el título, la fuente y el puntaje de relevancia y el formateo de ellos en una cadena limpia y legible.
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}La clase SearchResultSParser proporciona un método robusto para extraer información estructurada de las respuestas LLM. Intenta analizar una cadena tipo JSON desde la salida del modelo, volviendo a un formato de respuesta de texto plano si falla el análisis. Maneja con gracia salidas de cadena y objetos de mensaje, asegurando un procesamiento constante aguas abajo. En el caso de los errores, registra una advertencia y devuelve una respuesta alternativa que contiene la respuesta sin procesar, fuentes vacías y una puntuación de confianza predeterminada, mejorando la tolerancia a las fallas del sistema.
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_searchesLa clase mejorada de TavilyRecriever es un envoltorio personalizado alrededor del TavilySearchapiretriever, agregando mayor flexibilidad, control y trazabilidad a las operaciones de búsqueda. Admite características avanzadas como la profundidad de búsqueda limitante, los filtros de inclusión/exclusión de dominio y los recuentos de resultados configurables. El método de Invoke realiza búsquedas web y rastrea los metadatos de cada consulta (marca de tiempo, tiempo de respuesta y recuento de resultados), almacenándolo para un análisis posterior.
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 []La clase SearchCache implementa una capa de almacenamiento en caché semántico que almacena y recupera documentos utilizando incrustaciones de vectores para una búsqueda de similitud eficiente. Utiliza GoogleGenerativeaiembeddings para convertir documentos en vectores densos y los almacena en una base de datos de vectores de croma. El método add_documents inicializa o actualiza el almacén Vector, mientras que el método de búsqueda permite una recuperación rápida de los documentos en caché más relevantes basados en la similitud semántica. Esto reduce las llamadas de API redundantes y mejora los tiempos de respuesta para consultas repetidas o relacionadas, que sirve como una capa de memoria híbrida liviana en la tubería de asistente de IA.
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])
Inicializamos los componentes centrales del Asistente de IA: un SearchCache semántico, el mejor retriever para consultar la consulta basada en la web y una empresa de conversación para retener el historial de chat en los turnos. También define un aviso estructurado utilizando ChatPromptTemplate, guiando a la LLM a actuar como asistente de investigación. El rápido impone reglas estrictas para la precisión objetiva, el uso de contexto, la cita de origen y la respuesta concisa, asegurando respuestas confiables y fundamentadas.
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()
Definimos la función get_llm, que inicializa un modelo de lenguaje Google Gemini con parámetros configurables, como el nombre del modelo, la temperatura y la configuración de decodificación (por ejemplo, TOP_P, TOP_K y MAX Tokens). Asegura la robustez con el manejo de errores para la inicialización del modelo fallido. También se crea una instancia de SearchResultsParser para estandarizar y estructurar las respuestas sin procesar de la LLM, lo que permite un procesamiento constante de respuestas y metadatos.
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()
La función traza_search_metrics visualiza las tendencias de rendimiento de consultas anteriores usando matplotlib. Convierte el historial de búsqueda en un marco de datos y traza dos subgrafías: uno que muestra el tiempo de respuesta por búsqueda y el otro que muestra el número de resultados devueltos. Esto ayuda a analizar la eficiencia y la calidad de la búsqueda del sistema con el tiempo, ayudando a los desarrolladores a ajustar al recipiente o identificar cuellos de botella en el uso del mundo real.
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)Estas dos funciones mejoran la inteligencia y la eficiencia del asistente. La función Remieve_with_Fallback implementa un mecanismo de recuperación híbrido: primero intenta obtener documentos semánticamente relevantes del caché local de Chroma y, si no tiene éxito, recae en una búsqueda web en tiempo real, almacenando en caché los nuevos resultados para uso futuro. Mientras tanto, Summarize_Documents aprovecha un Gemini LLM para generar resúmenes concisos a partir de documentos recuperados, guiados por un indicador estructurado que garantiza la relevancia para la consulta. Juntos, permiten respuestas de baja latencia, informativa y con el contexto.
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()La función Advanced_Chain define un flujo de trabajo de razonamiento modular de extremo a extremo para responder consultas de los usuarios utilizando la búsqueda en caché o en tiempo real. Inicializa el modelo Gemini especificado, selecciona la estrategia de recuperación (búsqueda en caché o búsqueda directa), construye una tubería de respuesta que incorpora el historial de chat (si está habilitado), formatea documentos en contexto y solicita la LLM utilizando una plantilla guiada por el sistema. La cadena también registra la interacción en la memoria y devuelve la respuesta final, analizada en texto limpio. Este diseño permite una experimentación flexible con modelos y estrategias de recuperación mientras se mantiene la coherencia de la conversación.
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}")Inicializamos los componentes finales del asistente inteligente. QA_Chain es la tubería de razonamiento ensamblada lista para procesar consultas de usuarios utilizando la generación de respuesta de recuperación, memoria y basada en Gemini. La función Analyze_Query realiza un análisis semántico ligero en una consulta, extrayendo el tema principal, el sentimiento, las entidades y el tipo de consulta utilizando el modelo Gemini y un aviso JSON estructurado. La consulta de ejemplo, sobre la liberación y recepción de Breath of the Wild, muestra cómo el asistente se desencadena y se prepara para la inferencia de pila completa e interpretación semántica. El encabezado impreso marca el inicio de la ejecución interactiva.
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)
Demostramos la tubería completa en acción. Realiza una búsqueda utilizando el QA_Chain, muestra la respuesta generada y luego analiza la consulta de sentimiento, tema, entidades y tipo. También recupera e imprime el historial de búsqueda de cada consulta, el tiempo de respuesta y el recuento de resultados. Además, ejecuta una búsqueda filtrada por dominio centrada en sitios relacionados con Nintendo, resume los resultados y visualiza el rendimiento de la búsqueda utilizando Tot_Search_Metrics, ofreciendo una vista integral de las capacidades del asistente en el uso en tiempo real.
En conclusión, seguir este tutorial ofrece a los usuarios un plan integral para crear un contexto altamente capaz, contextable y escalable TRAPO Sistema que une inteligencia web en tiempo real con IA conversacional. La API de búsqueda de Tavily permite a los usuarios extraer directamente contenido fresco y relevante de la web. El Gemini LLM agrega capacidades robustas de razonamiento y resumen, mientras que la capa de abstracción de Langchain permite una orquestación perfecta entre la memoria, las incrustaciones y las salidas del modelo. La implementación incluye características avanzadas como filtrado específico de dominio, análisis de consultas (sentimiento, tema y extracción de entidad) y estrategias de alojamiento utilizando un caché vectorial semántico construido con Chroma y GoogleGenerativeiembeddings. Además, los paneles de registro estructurado, manejo de errores y análisis proporcionan transparencia y diagnósticos para la implementación del mundo real.
Mira el Cuaderno de colab. Todo el crédito por esta investigación va a los investigadores de este proyecto. Además, siéntete libre de seguirnos Gorjeo Y no olvides unirte a nuestro 90k+ ml de subreddit.
Asif Razzaq es el CEO de MarktechPost Media Inc .. Como empresario e ingeniero visionario, ASIF se compromete a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, MarktechPost, que se destaca por su cobertura profunda de noticias de aprendizaje automático y de aprendizaje profundo que es técnicamente sólido y fácilmente comprensible por una audiencia amplia. La plataforma cuenta con más de 2 millones de vistas mensuales, ilustrando su popularidad entre el público.