Saltar al contenido
Lección 12 de 12

Proyecto: Construye una Base de Conocimiento

10 min read

Visión General del Proyecto Final

En esta lección final, construirás un sistema de base de conocimiento completo desde cero. Este proyecto reúne todo lo que has aprendido: procesamiento de documentos, chunking, embeddings, almacenamiento vectorial, recuperación, ingeniería de prompts y medición de calidad. Al final, tendrás un sistema funcional que ingesta documentos y responde preguntas sobre ellos con citas de fuentes.

Lo que Construirás

  • Un pipeline de ingestión de documentos que maneja PDFs y páginas web
  • Un vector store respaldado por ChromaDB con metadatos
  • Un pipeline de recuperación con búsqueda híbrida y re-ranking
  • Una interfaz de chat que genera respuestas fundamentadas
  • Una suite de verificación de calidad para comprobar que tu sistema funciona correctamente

Paso 1: Configuración del Proyecto

Crea la estructura del proyecto e instala dependencias:

mkdir knowledge-base && cd knowledge-base
python -m venv venv
source venv/bin/activate  # En Windows: venv\Scripts\activate

pip install langchain langchain-openai langchain-community chromadb
pip install pypdf beautifulsoup4 requests ragas

Crea la estructura del proyecto:

mkdir -p src data/pdfs data/web
touch src/__init__.py src/ingest.py src/retriever.py src/chain.py src/app.py

Configura tus variables de entorno:

export OPENAI_API_KEY="tu-api-key-aqui"

Paso 2: Pipeline de Ingestión de Documentos

Crea src/ingest.py -- el pipeline que carga, fragmenta y almacena documentos:

"""Pipeline de ingestion de documentos para la base de conocimiento."""
from pathlib import Path
from langchain_community.document_loaders import (
    PyPDFLoader,
    WebBaseLoader,
    DirectoryLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from datetime import datetime

# Configuracion
CHROMA_DIR = "./chroma_db"
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
EMBEDDING_MODEL = "text-embedding-3-small"


def load_pdfs(pdf_dir: str) -> list:
    """Cargar todos los PDFs de un directorio."""
    loader = DirectoryLoader(pdf_dir, glob="**/*.pdf", loader_cls=PyPDFLoader)
    docs = loader.load()

    for doc in docs:
        doc.metadata["source_type"] = "pdf"
        doc.metadata["indexed_at"] = datetime.now().isoformat()

    print(f"Cargadas {len(docs)} paginas de PDFs")
    return docs


def load_web_pages(urls: list[str]) -> list:
    """Cargar contenido de paginas web."""
    loader = WebBaseLoader(web_paths=urls)
    docs = loader.load()

    for doc in docs:
        doc.metadata["source_type"] = "web"
        doc.metadata["indexed_at"] = datetime.now().isoformat()
        # Limpiar espacios en blanco
        doc.page_content = " ".join(doc.page_content.split())

    print(f"Cargadas {len(docs)} paginas web")
    return docs


def chunk_documents(docs: list) -> list:
    """Dividir documentos en fragmentos para embedding."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = splitter.split_documents(docs)
    print(f"Divididos en {len(chunks)} fragmentos")
    return chunks


def create_vectorstore(chunks: list) -> Chroma:
    """Embeber fragmentos y almacenar en ChromaDB."""
    embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=CHROMA_DIR,
        collection_metadata={"hnsw:space": "cosine"},
    )

    print(f"Almacenados {len(chunks)} fragmentos en ChromaDB")
    return vectorstore


def ingest(pdf_dir: str = "./data/pdfs", urls: list[str] = None) -> Chroma:
    """Ejecutar el pipeline completo de ingestion."""
    all_docs = []

    # Cargar PDFs si el directorio existe y tiene archivos
    pdf_path = Path(pdf_dir)
    if pdf_path.exists() and list(pdf_path.glob("**/*.pdf")):
        all_docs.extend(load_pdfs(pdf_dir))

    # Cargar paginas web si se proporcionan URLs
    if urls:
        all_docs.extend(load_web_pages(urls))

    if not all_docs:
        print("No se encontraron documentos. Agrega PDFs a ./data/pdfs o proporciona URLs.")
        return None

    # Fragmentar y almacenar
    chunks = chunk_documents(all_docs)
    vectorstore = create_vectorstore(chunks)

    return vectorstore


if __name__ == "__main__":
    # Ejemplo: ingestar PDFs y algunas paginas web
    urls = [
        "https://docs.python.org/3/tutorial/index.html",
        "https://docs.python.org/3/tutorial/introduction.html",
    ]
    ingest(urls=urls)

Paso 3: Pipeline de Recuperación

Crea src/retriever.py -- la capa de recuperación con búsqueda híbrida:

"""Pipeline de recuperacion con busqueda hibrida y re-ranking."""
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

CHROMA_DIR = "./chroma_db"
EMBEDDING_MODEL = "text-embedding-3-small"


def get_vectorstore() -> Chroma:
    """Cargar vectorstore existente."""
    embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)
    return Chroma(
        persist_directory=CHROMA_DIR,
        embedding_function=embeddings,
    )


def create_retriever(vectorstore: Chroma, k: int = 5):
    """Crear retriever con MMR para diversidad."""
    return vectorstore.as_retriever(
        search_type="mmr",
        search_kwargs={
            "k": k,
            "fetch_k": 20,
            "lambda_mult": 0.7,
        },
    )


def retrieve_with_scores(vectorstore: Chroma, query: str, k: int = 5):
    """Recuperar documentos con puntuaciones de similitud para depuracion."""
    results = vectorstore.similarity_search_with_score(query, k=k)
    return results


def filtered_retrieve(
    vectorstore: Chroma, query: str, source_type: str = None, k: int = 5
):
    """Recuperar con filtrado opcional de metadatos."""
    search_kwargs = {"k": k}

    if source_type:
        search_kwargs["filter"] = {"source_type": source_type}

    retriever = vectorstore.as_retriever(
        search_type="mmr", search_kwargs=search_kwargs
    )
    return retriever.invoke(query)

Paso 4: Cadena RAG con Citas de Fuentes

Crea src/chain.py -- la capa de generación:

"""Cadena RAG con citas de fuentes."""
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

LLM_MODEL = "gpt-4o"


def format_docs_with_sources(docs) -> str:
    """Formatear documentos recuperados con metadatos de fuente."""
    formatted = []
    for i, doc in enumerate(docs):
        source = doc.metadata.get("source", "Desconocido")
        source_type = doc.metadata.get("source_type", "desconocido")
        page = doc.metadata.get("page", "")
        page_info = f", pagina {page}" if page else ""

        formatted.append(
            f"[Fuente {i+1}: {source}{page_info} ({source_type})]\n"
            f"{doc.page_content}"
        )
    return "\n\n---\n\n".join(formatted)


def create_rag_chain(retriever):
    """Crear la cadena RAG con rastreo de fuentes."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Eres un asistente experto que responde preguntas
usando el contexto proporcionado. Sigue estas reglas:

1. Responde SOLO basandote en el contexto proporcionado abajo.
2. Si el contexto no contiene la respuesta, di:
   "No tengo suficiente informacion para responder esta pregunta."
3. Cita tus fuentes usando referencias [Fuente N].
4. Se conciso pero completo.
5. Si las fuentes proporcionan informacion diferente, menciona ambas perspectivas.

Contexto:
{context}"""),
        ("human", "{question}"),
    ])

    llm = ChatOpenAI(model=LLM_MODEL, temperature=0)

    # Cadena que retorna tanto la respuesta como las fuentes
    chain = RunnableParallel(
        answer=(
            {
                "context": retriever | format_docs_with_sources,
                "question": RunnablePassthrough(),
            }
            | prompt
            | llm
            | StrOutputParser()
        ),
        sources=retriever,
    )

    return chain


def create_simple_chain(retriever):
    """Crear una cadena simple que retorna solo la respuesta."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Responde basandote en el contexto abajo. Cita fuentes como
[Fuente: archivo]. Si no estas seguro, dilo.

Contexto:
{context}"""),
        ("human", "{question}"),
    ])

    llm = ChatOpenAI(model=LLM_MODEL, temperature=0)

    chain = (
        {
            "context": retriever | format_docs_with_sources,
            "question": RunnablePassthrough(),
        }
        | prompt
        | llm
        | StrOutputParser()
    )

    return chain

Paso 5: Interfaz de Chat

Crea src/app.py -- una interfaz interactiva simple:

"""Interfaz de chat interactiva para la base de conocimiento."""
from src.retriever import get_vectorstore, create_retriever, retrieve_with_scores
from src.chain import create_rag_chain


def run_chat():
    """Ejecutar una sesion de chat interactiva."""
    print("Chat de Base de Conocimiento")
    print("=" * 50)
    print("Escribe 'salir' para terminar, 'debug' para alternar modo debug")
    print()

    vectorstore = get_vectorstore()
    retriever = create_retriever(vectorstore, k=5)
    chain = create_rag_chain(retriever)

    debug_mode = False

    while True:
        question = input("Tu: ").strip()

        if not question:
            continue
        if question.lower() == "salir":
            print("Hasta luego!")
            break
        if question.lower() == "debug":
            debug_mode = not debug_mode
            print(f"Modo debug: {'ACTIVADO' if debug_mode else 'DESACTIVADO'}")
            continue

        # Obtener respuesta con fuentes
        result = chain.invoke(question)

        print(f"\nAsistente: {result['answer']}")

        if debug_mode:
            print("\n--- Info Debug ---")
            print(f"Recuperados {len(result['sources'])} fragmentos:")
            for i, doc in enumerate(result["sources"]):
                source = doc.metadata.get("source", "Desconocido")
                print(f"  [{i+1}] {source}")
                print(f"      {doc.page_content[:100]}...")

            # Mostrar puntuaciones de similitud
            scored = retrieve_with_scores(vectorstore, question, k=3)
            print("\nPuntuaciones de similitud:")
            for doc, score in scored:
                print(f"  Puntuacion: {score:.4f} | {doc.metadata.get('source', '?')}")
            print("--- Fin Debug ---")

        print()


if __name__ == "__main__":
    run_chat()

Paso 6: Verificaciones de Calidad

Crea src/check_quality.py -- verificaciones automatizadas para comprobar tu sistema:

"""Verificaciones de calidad para la base de conocimiento."""
import json
from datetime import datetime
from src.retriever import get_vectorstore, create_retriever
from src.chain import create_simple_chain


def run_quality_checks(test_questions: list[dict]) -> dict:
    """Ejecutar verificaciones de calidad contra un conjunto de preguntas."""
    vectorstore = get_vectorstore()
    retriever = create_retriever(vectorstore, k=5)
    chain = create_simple_chain(retriever)

    results = []

    for item in test_questions:
        question = item["question"]
        expected_keywords = item.get("expected_keywords", [])

        # Obtener respuesta
        answer = chain.invoke(question)

        # Verificar si las palabras clave esperadas aparecen en la respuesta
        found_keywords = [
            kw for kw in expected_keywords if kw.lower() in answer.lower()
        ]
        keyword_coverage = (
            len(found_keywords) / len(expected_keywords)
            if expected_keywords
            else 1.0
        )

        results.append({
            "question": question,
            "answer": answer,
            "keyword_coverage": keyword_coverage,
            "found_keywords": found_keywords,
            "missing_keywords": [
                kw for kw in expected_keywords if kw not in found_keywords
            ],
        })

        status = "PASA" if keyword_coverage >= 0.5 else "FALLA"
        print(f"[{status}] {question}")
        print(f"       Cobertura: {keyword_coverage:.0%}")
        if keyword_coverage < 1.0:
            missing = [kw for kw in expected_keywords if kw not in found_keywords]
            print(f"       Faltantes: {missing}")
        print()

    # Resumen
    pass_count = sum(1 for r in results if r["keyword_coverage"] >= 0.5)
    total = len(results)
    avg_coverage = sum(r["keyword_coverage"] for r in results) / total

    summary = {
        "timestamp": datetime.now().isoformat(),
        "total_questions": total,
        "passed": pass_count,
        "failed": total - pass_count,
        "average_keyword_coverage": round(avg_coverage, 2),
        "results": results,
    }

    print("=" * 50)
    print(f"Resultados: {pass_count}/{total} pasaron ({avg_coverage:.0%} cobertura promedio)")

    # Guardar resultados
    with open("quality_results.json", "w") as f:
        json.dump(summary, f, indent=2)

    return summary


# Preguntas de prueba - personaliza para tu base de conocimiento
TEST_QUESTIONS = [
    {
        "question": "Que temas cubre la base de conocimiento?",
        "expected_keywords": ["python", "tutorial"],
    },
    {
        "question": "Como se definen variables en Python?",
        "expected_keywords": ["variable", "asignacion", "="],
    },
    {
        "question": "Que es un tema completamente no relacionado como la fisica cuantica?",
        "expected_keywords": [],  # Deberia decir "No tengo informacion"
    },
]

if __name__ == "__main__":
    run_quality_checks(TEST_QUESTIONS)

Paso 7: Uniendo Todo

Crea un punto de entrada main.py:

"""Punto de entrada principal para el sistema de base de conocimiento."""
import sys
from src.ingest import ingest
from src.app import run_chat
from src.check_quality import run_quality_checks, TEST_QUESTIONS


def main():
    if len(sys.argv) < 2:
        print("Uso:")
        print("  python main.py ingest              - Ingestar documentos")
        print("  python main.py chat                - Iniciar interfaz de chat")
        print("  python main.py check               - Ejecutar verificaciones de calidad")
        print("  python main.py ingest --urls URL1 URL2")
        return

    command = sys.argv[1]

    if command == "ingest":
        urls = []
        if "--urls" in sys.argv:
            url_idx = sys.argv.index("--urls") + 1
            urls = sys.argv[url_idx:]

        ingest(urls=urls if urls else None)

    elif command == "chat":
        run_chat()

    elif command == "check":
        run_quality_checks(TEST_QUESTIONS)

    else:
        print(f"Comando desconocido: {command}")


if __name__ == "__main__":
    main()

Ejecutando el Sistema

# Paso 1: Ingestar algunos documentos
python main.py ingest --urls https://docs.python.org/3/tutorial/introduction.html

# Paso 2: Comenzar a chatear
python main.py chat

# Paso 3: Ejecutar verificaciones de calidad
python main.py check

Extendiendo el Proyecto

Ahora que tienes una base funcional, aquí hay formas de extenderlo:

  • Agrega más tipos de documentos. Integra cargadores de Markdown, archivos de código o parsers de CSV.
  • Implementa una interfaz web. Usa Streamlit o Gradio para construir una interfaz basada en navegador con respuestas en streaming.
  • Agrega re-ranking. Integra Cohere Rerank o un cross-encoder para mejorar la precisión de recuperación.
  • Implementa control de acceso. Agrega roles de usuario y filtra documentos basándose en permisos.
  • Configura monitoreo. Registra cada consulta, su latencia, los documentos recuperados y el feedback del usuario.
  • Construye una API. Envuelve la cadena en un endpoint FastAPI para integración con otros sistemas.

Resumen del Curso

A lo largo de doce lecciones, has aprendido:

  1. Por qué RAG importa y qué problemas resuelve
  2. Cómo los embeddings codifican significado en vectores buscables
  3. Cómo las bases de datos vectoriales almacenan y recuperan esos vectores a escala
  4. Cómo procesar documentos de PDFs, HTML, código y más
  5. Cómo las estrategias de chunking afectan la calidad de recuperación
  6. Cómo técnicas avanzadas de recuperación como búsqueda híbrida y re-ranking mejoran resultados
  7. Cómo construir un pipeline RAG completo con LangChain
  8. Cómo patrones avanzados como HyDE, self-RAG y RAG agéntico van más allá de la recuperación básica
  9. Cómo construir sistemas RAG específicos para código
  10. Cómo medir tu sistema con métricas reales
  11. Cómo endurecer todo para producción
  12. Cómo construir una base de conocimiento completa desde cero

RAG es una de las habilidades más prácticas e impactantes en ingeniería de IA hoy. Los sistemas que construyas con estas técnicas conectarán modelos de lenguaje poderosos con datos del mundo real, produciendo respuestas que son precisas, actuales y verificables. Ve y construye algo útil.