Capítulo 3: Self-Query y filtrado por metadatos

¿Qué es Self-Query?

El LLM extrae tanto la consulta semántica como los filtros estructurados de la entrada del usuario:

Usuario: "Buscar políticas de RRHH de 2023"

Extraído:
- Consulta semántica: "políticas de RRHH"
- Filtros: {department: "HR", year: 2023}

Implementación con LangChain

from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo

# Define metadata attributes
metadata_field_info = [
    AttributeInfo(
        name="department",
        description="El departamento (HR, Engineering, Sales)",
        type="string"
    ),
    AttributeInfo(
        name="year",
        description="Año de publicación",
        type="integer"
    ),
    AttributeInfo(
        name="document_type",
        description="Tipo de documento (policy, guide, memo)",
        type="string"
    )
]

# Create self-query retriever
retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents="Company internal documents",
    metadata_field_info=metadata_field_info
)

# Query with natural language filters
results = retriever.get_relevant_documents(
    "Muéstreme las guías de ingeniería de 2024"
)
# Automatically filters: department=Engineering, year=2024

Comparaciones soportadas

Operador	Ejemplo
$eq	year == 2023
$ne	status != "draft"
$gt	year > 2020
$gte	year >= 2020
$lt	year < 2024
$in	department in ["HR", "Legal"]

Mejores prácticas

1. Esquema de metadatos claro - Atributos bien definidos
2. Buenas descripciones - Ayudan al LLM a entender los campos
3. Lógica de respaldo - Manejar fallos en la extracción
4. Pruebas exhaustivas - Diversos patrones de consulta

Combinando enfoques

# Multi-query + Self-query
def advanced_retrieve(query, retriever):
    # Generate variations
    queries = generate_variations(query)
    
    # Self-query each variation
    all_results = []
    for q in queries:
        results = self_query_retriever.get_relevant_documents(q)
        all_results.extend(results)
    
    # Deduplicate and rank
    return rank_and_dedupe(all_results)