LlamaIndex: el framework de datos para agentes IA con RAG

# Paquete base
pip install llama-index

# Con integracion OpenAI (embeddings + LLM)
pip install llama-index llama-index-llms-openai llama-index-embeddings-openai

# Con Anthropic Claude
pip install llama-index llama-index-llms-anthropic

# Con modelos locales vía Ollama
pip install llama-index llama-index-llms-ollama llama-index-embeddings-ollama

# Con Chroma como base de datos vectorial
pip install llama-index llama-index-vector-stores-chroma

import os
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

# Configuración global (evita pasar el LLM en cada llamada)
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0.1)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512      # Tamaño de cada nodo en tokens
Settings.chunk_overlap = 50    # Solapamiento entre nodos consecutivos

# 1. Cargar documentos desde un directorio (PDF, TXT, MD, etc.)
documents = SimpleDirectoryReader("./data/").load_data()
print(f"Cargados {len(documents)} documentos")

# 2. Indexar los documentos (crea embeddings y los almacena en memoria)
index = VectorStoreIndex.from_documents(documents)

# 3. Crear el query engine
query_engine = index.as_query_engine(
    similarity_top_k=5,   # Recupera los 5 nodos más similares
    streaming=True,       # Respuesta en streaming para mejor UX
)

# 4. Hacer una pregunta
respuesta = query_engine.query(
    "¿Cuál es la política de devolución de productos?"
)
print(respuesta)

import chromadb
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext
from llama_index.vector_stores.chroma import ChromaVectorStore

# Inicializar Chroma con persistencia en disco
chroma_client = chromadb.PersistentClient(path="./chroma_db")
chroma_collection = chroma_client.get_or_create_collection("documentos_empresa")

# Crear el vector store de LlamaIndex apuntando a Chroma
vector_store = ChromaVectorStore(chroma_collection=chroma_collection)
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# Primera ejecucion: indexar documentos y persistir en Chroma
documents = SimpleDirectoryReader("./data/").load_data()
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context,
)

# Ejecuciones posteriores: cargar desde Chroma sin re-indexar
# index = VectorStoreIndex.from_vector_store(vector_store)

# Query engine con metadata filtering
query_engine = index.as_query_engine(
    similarity_top_k=5,
    # Filtrar por metadatos del documento (si se definieron al cargar)
    # filters=MetadataFilters(filters=[ExactMatchFilter(key="tipo", value="contrato")])
)

respuesta = query_engine.query("Resumen de los contratos vigentes en 2026")
print(respuesta.source_nodes[0].score)  # Puntuación de relevancia del primer nodo

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.tools import QueryEngineTool, FunctionTool
from llama_index.core.agent import ReActAgent
from llama_index.llms.anthropic import Anthropic

# Herramienta 1: base de conocimiento propia
documentos = SimpleDirectoryReader("./docs/").load_data()
indice = VectorStoreIndex.from_documents(documentos)
herramienta_docs = QueryEngineTool.from_defaults(
    query_engine=indice.as_query_engine(),
    name="base_conocimiento",
    description=(
        "Busca información en la documentación interna de la empresa. "
        "Usa esta herramienta para preguntas sobre políticas, productos "
        "y procedimientos internos."
    ),
)

# Herramienta 2: función Python personalizada
def obtener_tipo_cambio(divisa: str) -> str:
    """Devuelve el tipo de cambio actual de la divisa indicada frente al euro."""
    # Aquí iría la llamada a una API de tipos de cambio
    tipos_ejemplo = {"USD": 1.08, "GBP": 0.86, "JPY": 163.5}
    tasa = tipos_ejemplo.get(divisa.upper(), "No disponible")
    return f"1 EUR = {tasa} {divisa.upper()}"

herramienta_cambio = FunctionTool.from_defaults(fn=obtener_tipo_cambio)

# Crear el agente con Claude Sonnet
llm = Anthropic(model="claude-sonnet-4-6", temperature=0)
agente = ReActAgent.from_tools(
    tools=[herramienta_docs, herramienta_cambio],
    llm=llm,
    verbose=True,     # Muestra el razonamiento interno (Thought/Action/Observación)
    max_iterations=10,
)

respuesta = agente.chat(
    "¿Cuál es la política de descuento para pedidos internacionales "
    "y cuánto sería en dólares un pedido de 500 euros?"
)
print(respuesta)

from llama_index.core.workflow import (
    Workflow, Event, StartEvent, StopEvent, step
)
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from pydantic import BaseModel

# Definir los eventos del workflow (tipado con Pydantic)
class ResultadoBusqueda(Event):
    nodos: list[str]
    pregunta: str

class RespuestaFinal(BaseModel):
    respuesta: str
    fuentes: list[str]

# Definir el workflow
class WorkflowRAG(Workflow):

    @step
    async def recuperar(self, ev: StartEvent) -> ResultadoBusqueda:
        """Paso 1: recuperar nodos relevantes del índice."""
        indice = VectorStoreIndex.from_documents(
            SimpleDirectoryReader("./docs/").load_data()
        )
        retriever = indice.as_retriever(similarity_top_k=5)
        nodos = await retriever.aretrieve(ev.pregunta)
        textos = [nodo.get_content() for nodo in nodos]
        return ResultadoBusqueda(nodos=textos, pregunta=ev.pregunta)

    @step
    async def sintetizar(self, ev: ResultadoBusqueda) -> StopEvent:
        """Paso 2: generar respuesta a partir de los nodos recuperados."""
        contexto = "\n\n---\n\n".join(ev.nodos)
        # Aqui iria la llamada al LLM para sintetizar la respuesta
        respuesta = f"Respuesta basada en {len(ev.nodos)} fragmentos: [respuesta generada]"  # noqa
        return StopEvent(result=respuesta)


# Ejecutar el workflow
async def main():
    workflow = WorkflowRAG(timeout=60)
    resultado = await workflow.run(pregunta="¿Cómo solicitar vacaciones?")
    print(resultado)

import asyncio
asyncio.run(main())