Módulo 6: RAG Multimodal
6. Implementación con LangChain
Descripción
En las cápsulas anteriores construiste cada pieza del RAG multimodal de forma manual: embeddings, indexación, retrieval, procesamiento de documentos. Funciona, pero escribiste mucho código de infraestructura. LangChain abstrae gran parte de esa infraestructura: document loaders, text splitters, vector stores, retrievers y chains ya están implementados. En esta cápsula vas a re-implementar el RAG multimodal usando LangChain, comparar con la implementación manual, y entender cuándo conviene cada enfoque.
Por qué importa: En producción, no quieres mantener código custom de chunking, embedding y retrieval si existe un framework probado que lo hace. LangChain te da componentes intercambiables: puedes cambiar de ChromaDB a Pinecone, o de OpenAI embeddings a Cohere, cambiando una línea. Pero también necesitas entender qué pasa debajo, y por eso primero lo hicimos manual.
Conexión con el módulo: LangChain es la implementación que usarás en el proyecto de la cápsula 08. Los conceptos de las cápsulas 02-05 (embeddings, indexación, retrieval, documentos) se mapean directamente a componentes de LangChain.
Setup
pip install langchain langchain-openai langchain-community chromadb pymupdf
from dotenv import load_dotenv
import os
load_dotenv()
assert os.getenv("OPENAI_API_KEY"), "Falta OPENAI_API_KEY en .env"
Componentes de LangChain para RAG
Mapa de componentes
Manual (lo que hiciste) → LangChain equivalente
─────────────────────────────────────────────────────────
fitz.open() + get_text() → PyMuPDFLoader
chunk_text() / chunk_by_section → RecursiveCharacterTextSplitter
OpenAI embeddings.create() → OpenAIEmbeddings
chromadb.Client() + add() → Chroma.from_documents()
collection.query() → vectorstore.as_retriever()
client.chat.completions.create() → ChatOpenAI + RetrievalQA chain
Cada componente manual que escribiste tiene un equivalente en LangChain. La ventaja es que son intercambiables: cambiar el vector store, el modelo de embeddings, o el LLM requiere cambiar una sola línea.
Document Loaders
Cargar PDFs
from langchain_community.document_loaders import PyMuPDFLoader
def load_pdf(pdf_path: str) -> list:
loader = PyMuPDFLoader(pdf_path)
documents = loader.load()
return documents
# docs = load_pdf("manual_tecnico.pdf")
# print(f"Páginas cargadas: {len(docs)}")
# print(f"Primera página ({len(docs[0].page_content)} chars):")
# print(docs[0].page_content[:200])
# print(f"Metadata: {docs[0].metadata}")
Cada Document tiene:
page_content: el texto de la páginametadata: información adicional (source, page, etc.)
Cargar múltiples formatos
from langchain_community.document_loaders import (
PyMuPDFLoader,
TextLoader,
UnstructuredMarkdownLoader,
)
from pathlib import Path
def load_document(file_path: str) -> list:
path = Path(file_path)
ext = path.suffix.lower()
loaders = {
".pdf": PyMuPDFLoader,
".txt": TextLoader,
".md": UnstructuredMarkdownLoader,
}
loader_class = loaders.get(ext)
if not loader_class:
raise ValueError(f"Formato no soportado: {ext}")
loader = loader_class(file_path)
return loader.load()
def load_directory(dir_path: str, glob_pattern: str = "**/*.pdf") -> list:
from langchain_community.document_loaders import DirectoryLoader
loader = DirectoryLoader(
dir_path,
glob=glob_pattern,
loader_cls=PyMuPDFLoader,
show_progress=True
)
return loader.load()
Loader custom para documentos con imágenes
LangChain no tiene un loader nativo que extraiga imágenes de PDFs y genere descripciones. Puedes crear uno.
from langchain.schema import Document
from openai import OpenAI
import fitz
import base64
oai_client = OpenAI()
def load_pdf_with_images(
pdf_path: str,
describe_images: bool = True
) -> list[Document]:
doc = fitz.open(pdf_path)
documents = []
for page_num in range(len(doc)):
page = doc[page_num]
text = page.get_text().strip()
if text:
documents.append(Document(
page_content=text,
metadata={
"source": pdf_path,
"page": page_num + 1,
"type": "text",
}
))
if describe_images:
img_list = page.get_images()
for img_idx, img_ref in enumerate(img_list):
xref = img_ref[0]
try:
base_image = doc.extract_image(xref)
if not base_image or not base_image.get("image"):
continue
if len(base_image["image"]) < 5000:
continue
b64 = base64.b64encode(base_image["image"]).decode("utf-8")
ext = base_image.get("ext", "png")
mime = f"image/{ext}" if ext != "jpg" else "image/jpeg"
response = oai_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": "Describe esta imagen en 1-2 oraciones para indexación."},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{b64}"}}
]
}],
max_tokens=100
)
description = response.choices[0].message.content
documents.append(Document(
page_content=description,
metadata={
"source": pdf_path,
"page": page_num + 1,
"type": "image_description",
"image_index": img_idx,
}
))
except Exception as e:
print(f"Error procesando imagen p.{page_num+1}: {e}")
doc.close()
return documents
Text Splitters
RecursiveCharacterTextSplitter
El splitter más usado en LangChain. Divide por párrafos, luego por oraciones, luego por palabras, intentando mantener chunks de tamaño uniforme.
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_documents(
documents: list[Document],
chunk_size: int = 1000,
chunk_overlap: int = 200
) -> list[Document]:
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["\n\n", "\n", ". ", " ", ""],
length_function=len,
)
text_docs = [d for d in documents if d.metadata.get("type") != "image_description"]
image_docs = [d for d in documents if d.metadata.get("type") == "image_description"]
split_text = text_splitter.split_documents(text_docs)
all_docs = split_text + image_docs
return all_docs
MarkdownTextSplitter
Para documentos markdown (o texto con headers), divide por secciones.
from langchain.text_splitter import MarkdownTextSplitter
def split_markdown(
documents: list[Document],
chunk_size: int = 1000,
chunk_overlap: int = 100
) -> list[Document]:
splitter = MarkdownTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
return splitter.split_documents(documents)
Comparar splitters
def compare_splitters(
documents: list[Document],
chunk_size: int = 1000
) -> dict:
recursive = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, chunk_overlap=200
)
markdown = MarkdownTextSplitter(
chunk_size=chunk_size, chunk_overlap=100
)
text_docs = [d for d in documents if d.metadata.get("type") != "image_description"]
recursive_chunks = recursive.split_documents(text_docs)
markdown_chunks = markdown.split_documents(text_docs)
def stats(chunks):
lengths = [len(c.page_content) for c in chunks]
return {
"count": len(chunks),
"avg_length": round(sum(lengths) / len(lengths)) if lengths else 0,
"min_length": min(lengths) if lengths else 0,
"max_length": max(lengths) if lengths else 0,
}
return {
"recursive": stats(recursive_chunks),
"markdown": stats(markdown_chunks),
}
Vector Store con Chroma
Crear vector store desde documentos
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
def create_vectorstore(
documents: list[Document],
persist_directory: str = "./chroma_langchain_db",
collection_name: str = "multimodal_rag"
) -> Chroma:
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
documents=documents,
embedding=embeddings,
persist_directory=persist_directory,
collection_name=collection_name,
)
return vectorstore
Cargar vector store existente
def load_vectorstore(
persist_directory: str = "./chroma_langchain_db",
collection_name: str = "multimodal_rag"
) -> Chroma:
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
persist_directory=persist_directory,
collection_name=collection_name,
embedding_function=embeddings,
)
return vectorstore
Añadir documentos a un vector store existente
def add_documents_to_store(
vectorstore: Chroma,
new_documents: list[Document]
) -> None:
vectorstore.add_documents(new_documents)
Retrievers
Retriever básico
def create_retriever(
vectorstore: Chroma,
k: int = 5,
search_type: str = "similarity"
) -> object:
retriever = vectorstore.as_retriever(
search_type=search_type,
search_kwargs={"k": k}
)
return retriever
Retriever con filtros de metadata
def create_filtered_retriever(
vectorstore: Chroma,
filter_dict: dict,
k: int = 5
) -> object:
retriever = vectorstore.as_retriever(
search_kwargs={
"k": k,
"filter": filter_dict
}
)
return retriever
# text_retriever = create_filtered_retriever(vectorstore, {"type": "text"}, k=5)
# image_retriever = create_filtered_retriever(vectorstore, {"type": "image_description"}, k=3)
Retriever con score threshold
def create_threshold_retriever(
vectorstore: Chroma,
score_threshold: float = 0.7,
k: int = 10
) -> object:
retriever = vectorstore.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"k": k,
"score_threshold": score_threshold
}
)
return retriever
Multi-retriever: combinar resultados de múltiples retrievers
from langchain.retrievers import EnsembleRetriever
def create_ensemble_retriever(
vectorstore: Chroma,
weights: list[float] = None
) -> EnsembleRetriever:
text_retriever = create_filtered_retriever(
vectorstore, {"type": "text"}, k=5
)
image_retriever = create_filtered_retriever(
vectorstore, {"type": "image_description"}, k=3
)
if weights is None:
weights = [0.6, 0.4]
ensemble = EnsembleRetriever(
retrievers=[text_retriever, image_retriever],
weights=weights
)
return ensemble
Chains RAG
RetrievalQA: el chain más simple
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
def create_qa_chain(
vectorstore: Chroma,
model: str = "gpt-4o",
k: int = 5
) -> RetrievalQA:
llm = ChatOpenAI(model=model, temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
qa = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True,
)
return qa
# qa = create_qa_chain(vectorstore)
# result = qa.invoke({"query": "¿Cómo se conectan los microservicios?"})
# print(result["result"])
# for doc in result["source_documents"]:
# print(f" Fuente: {doc.metadata}")
Chain con prompt custom
from langchain.prompts import PromptTemplate
RAG_PROMPT = PromptTemplate(
template=(
"Eres un asistente que responde preguntas basándose SOLO en el contexto proporcionado.\n"
"El contexto puede incluir descripciones de imágenes marcadas como tipo 'image_description'.\n"
"Si encuentras información de imágenes, menciónala en tu respuesta.\n"
"Si no encuentras la respuesta en el contexto, di que no tienes información suficiente.\n"
"Cita las fuentes (página, tipo de contenido) al final de tu respuesta.\n\n"
"Contexto:\n{context}\n\n"
"Pregunta: {question}\n\n"
"Respuesta:"
),
input_variables=["context", "question"]
)
def create_custom_qa_chain(
vectorstore: Chroma,
model: str = "gpt-4o",
k: int = 5
) -> RetrievalQA:
llm = ChatOpenAI(model=model, temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
qa = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True,
chain_type_kwargs={"prompt": RAG_PROMPT},
)
return qa
Chain con LCEL (LangChain Expression Language)
LCEL es la forma moderna de construir chains en LangChain. Más flexible y composable.
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
def create_lcel_rag_chain(
vectorstore: Chroma,
model: str = "gpt-4o",
k: int = 5
):
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
llm = ChatOpenAI(model=model, temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", (
"Eres un asistente que responde preguntas usando el contexto proporcionado. "
"El contexto puede incluir texto y descripciones de imágenes de documentos. "
"Responde de forma precisa y cita las fuentes."
)),
("human", "Contexto:\n{context}\n\nPregunta: {question}"),
])
def format_docs(docs):
formatted = []
for doc in docs:
doc_type = doc.metadata.get("type", "text")
page = doc.metadata.get("page", "?")
prefix = f"[{doc_type.upper()} - p.{page}]"
formatted.append(f"{prefix}\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)
chain = (
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
}
| prompt
| llm
| StrOutputParser()
)
return chain
# chain = create_lcel_rag_chain(vectorstore)
# answer = chain.invoke("¿Qué muestra el diagrama de arquitectura?")
# print(answer)
Chain con fuentes explícitas
from langchain_core.runnables import RunnableParallel
def create_rag_with_sources(
vectorstore: Chroma,
model: str = "gpt-4o",
k: int = 5
):
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
llm = ChatOpenAI(model=model, temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", (
"Responde la pregunta basándote en el contexto. "
"Si incluye descripciones de imágenes, refiérelas. "
"Al final, lista las fuentes usadas."
)),
("human", "Contexto:\n{context}\n\nPregunta: {question}"),
])
def format_docs(docs):
parts = []
for doc in docs:
page = doc.metadata.get("page", "?")
doc_type = doc.metadata.get("type", "text")
parts.append(f"[{doc_type} p.{page}] {doc.page_content}")
return "\n\n".join(parts)
def get_sources(docs):
sources = []
for doc in docs:
sources.append({
"page": doc.metadata.get("page"),
"type": doc.metadata.get("type"),
"source": doc.metadata.get("source"),
"preview": doc.page_content[:100],
})
return sources
chain = RunnableParallel(
answer=(
{
"context": retriever | format_docs,
"question": RunnablePassthrough(),
}
| prompt
| llm
| StrOutputParser()
),
sources=retriever | get_sources,
)
return chain
# result = chain.invoke("¿Cómo funciona la autenticación?")
# print(f"Respuesta: {result['answer']}")
# print(f"Fuentes: {result['sources']}")
Pipeline Completo con LangChain
End-to-end: PDF → indexar → query → respuesta
def build_multimodal_rag(
pdf_paths: list[str],
persist_dir: str = "./chroma_langchain_db",
collection_name: str = "multimodal_rag",
chunk_size: int = 1000,
describe_images: bool = True
):
all_documents = []
for pdf_path in pdf_paths:
if describe_images:
docs = load_pdf_with_images(pdf_path, describe_images=True)
else:
docs = load_pdf(pdf_path)
all_documents.extend(docs)
split_docs = split_documents(all_documents, chunk_size=chunk_size)
print(f"Total documentos: {len(split_docs)}")
text_count = sum(1 for d in split_docs if d.metadata.get("type") != "image_description")
image_count = sum(1 for d in split_docs if d.metadata.get("type") == "image_description")
print(f" Texto: {text_count}, Imágenes: {image_count}")
vectorstore = create_vectorstore(
split_docs,
persist_directory=persist_dir,
collection_name=collection_name
)
chain = create_rag_with_sources(vectorstore)
return chain, vectorstore
# chain, vs = build_multimodal_rag(["doc1.pdf", "doc2.pdf"])
# result = chain.invoke("¿Qué muestra el diagrama principal?")
Comparación: Manual vs LangChain
Código para la misma tarea
TAREA: Indexar un PDF con imágenes y responder preguntas
Manual (cápsulas 02-05):
- extract_text_chunks_from_pdf() ~ 25 líneas
- extract_images_from_pdf() ~ 20 líneas
- describe_image_for_indexing() ~ 15 líneas
- add_text_chunks() ~ 20 líneas
- add_image_descriptions() ~ 25 líneas
- search_by_text() ~ 15 líneas
- hybrid_retrieve() ~ 20 líneas
- generate_answer() ~ 25 líneas
Total: ~165 líneas
LangChain:
- load_pdf_with_images() ~ 40 líneas (custom loader)
- split_documents() ~ 10 líneas
- Chroma.from_documents() ~ 3 líneas
- create_lcel_rag_chain() ~ 20 líneas
Total: ~73 líneas
Cuándo usar cada enfoque
| Criterio | Manual | LangChain |
|---|---|---|
| Control fino | Total — decides cada detalle | Limitado por abstracciones |
| Velocidad de desarrollo | Lento — escribes todo | Rápido — componentes listos |
| Debugging | Directo — ves cada paso | Más opaco — abstracciones ocultan detalles |
| Intercambiabilidad | Reescribes código al cambiar componentes | Cambias una línea |
| Dependencias | Mínimas (openai, chromadb) | LangChain + sus dependencias |
| Producción | Más predecible, menos magia | Más rápido de iterar |
| Aprendizaje | Entiendes los fundamentos | Entiendes el framework |
Recomendación
Prototipo rápido → LangChain
- Quieres probar una idea
- Los defaults son suficientes
- Vas a iterar mucho
Producción con requerimientos específicos → Manual (o LangChain + customización)
- Necesitas control de costos fino
- La lógica de retrieval es custom
- El pipeline tiene pasos no estándar
Aprendizaje → Ambos
- Primero manual para entender
- Luego LangChain para ser productivo
Troubleshooting
Error: "Collection already exists"
Al re-ejecutar el script, intentas crear un vector store que ya existe.
vectorstore = load_vectorstore(persist_dir, collection_name)
# Si necesitas recrear:
# import shutil
# shutil.rmtree(persist_dir)
# vectorstore = create_vectorstore(docs, persist_dir, collection_name)
El retriever no encuentra documentos de imagen
Verifica que los documentos de imagen se indexaron correctamente.
def debug_vectorstore(vectorstore: Chroma) -> dict:
collection = vectorstore._collection
all_docs = collection.get(include=["metadatas"])
types = {}
for meta in all_docs["metadatas"]:
t = meta.get("type", "unknown")
types[t] = types.get(t, 0) + 1
return {"total": len(all_docs["ids"]), "by_type": types}
LangChain deprecation warnings
LangChain evoluciona rápido. Usa los imports actuales.
# VIEJO (deprecated)
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
# NUEVO
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
El chain no retorna fuentes
Asegúrate de usar return_source_documents=True en RetrievalQA o construir el chain con fuentes explícitas usando LCEL.
Memoria excesiva con documentos grandes
def process_in_batches(
documents: list[Document],
vectorstore: Chroma,
batch_size: int = 100
) -> None:
for i in range(0, len(documents), batch_size):
batch = documents[i:i + batch_size]
vectorstore.add_documents(batch)
print(f"Batch {i//batch_size + 1}: {len(batch)} docs indexados")
Ejercicios
Ejercicio 1: Retriever con filtro dinámico
Crea un retriever que filtre por tipo de contenido según el query. Si el query menciona "diagrama", "imagen" o "figura", busca solo en image_description. Si no, busca en todo.
Ver solución
def smart_retriever(
vectorstore: Chroma,
query: str,
k: int = 5
) -> list[Document]:
visual_keywords = {"diagrama", "imagen", "figura", "gráfico", "tabla", "captura", "foto"}
query_words = set(query.lower().split())
if query_words & visual_keywords:
retriever = create_filtered_retriever(
vectorstore, {"type": "image_description"}, k=k
)
else:
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
return retriever.invoke(query)
# docs = smart_retriever(vectorstore, "muestra el diagrama de arquitectura")
# for doc in docs:
# print(f" [{doc.metadata.get('type')}] {doc.page_content[:60]}...")
Ejercicio 2: RAG con historial de conversación
Extiende el chain para que mantenga historial y pueda hacer follow-up questions.
Ver solución
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain
def create_conversational_rag(
vectorstore: Chroma,
model: str = "gpt-4o",
k: int = 5
):
llm = ChatOpenAI(model=model, temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": k})
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
output_key="answer"
)
chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
memory=memory,
return_source_documents=True,
)
return chain
# chain = create_conversational_rag(vectorstore)
# r1 = chain.invoke({"question": "¿Qué es el API Gateway?"})
# print(r1["answer"])
# r2 = chain.invoke({"question": "¿Cómo se conecta con la base de datos?"})
# print(r2["answer"])
Ejercicio 3: Comparar resultados manual vs LangChain
Dada la misma query, ejecuta el retrieval manual (cápsula 04) y el retrieval de LangChain. Compara qué documentos recupera cada uno.
Ver solución
def compare_retrieval(
manual_collection,
langchain_vectorstore: Chroma,
query: str,
k: int = 5
) -> dict:
manual_results = manual_collection.query(
query_texts=[query],
n_results=k
)
manual_ids = manual_results["ids"][0]
lc_retriever = langchain_vectorstore.as_retriever(search_kwargs={"k": k})
lc_docs = lc_retriever.invoke(query)
lc_previews = [d.page_content[:80] for d in lc_docs]
print(f"Query: {query}\n")
print("Manual retrieval:")
for i, (doc_id, doc_text) in enumerate(zip(manual_ids, manual_results["documents"][0])):
print(f" {i+1}. {doc_id}: {doc_text[:60]}...")
print("\nLangChain retrieval:")
for i, (doc, preview) in enumerate(zip(lc_docs, lc_previews)):
print(f" {i+1}. [{doc.metadata.get('type')}] {preview}...")
return {
"manual_count": len(manual_ids),
"langchain_count": len(lc_docs),
}
Resumen
- LangChain abstrae el pipeline RAG en componentes intercambiables: loaders, splitters, vector stores, retrievers, chains.
- Document loaders cargan PDFs, markdown, texto. Para imágenes, necesitas un loader custom.
- Text splitters dividen documentos en chunks.
RecursiveCharacterTextSplitteres el más versátil. - Chroma se integra nativamente con LangChain para crear vector stores persistentes.
- Retrievers buscan documentos relevantes. Puedes filtrar por metadata, usar score thresholds, o combinar múltiples retrievers con
EnsembleRetriever. - LCEL es la forma moderna de construir chains: composable, tipada, y con soporte para streaming.
- Manual vs LangChain: manual da control total, LangChain da velocidad de desarrollo. Ambos son válidos.
- Para producción, LangChain con customización (loaders custom, prompts específicos) suele ser el sweet spot.
Recursos Adicionales
- LangChain RAG Tutorial — Tutorial oficial de RAG
- LangChain Document Loaders — Todos los loaders disponibles
- LangChain Retrievers — Tipos de retrievers
- Chroma + LangChain — Integración ChromaDB
- LCEL (LangChain Expression Language) — Guía de LCEL