Saltar al contenido
Lección 12 de 12

Proyecto: Construye una Base de Conocimiento

10 min read

Vision General del Proyecto Final

En esta leccion final, construiras un sistema de base de conocimiento completo desde cero. Este proyecto reune todo lo que has aprendido: procesamiento de documentos, chunking, embeddings, almacenamiento vectorial, recuperacion, ingenieria de prompts y medicion de calidad. Al final, tendras un sistema funcional que ingesta documentos y responde preguntas sobre ellos con citas de fuentes.

Lo que Construiras

  • Un pipeline de ingestion de documentos que maneja PDFs y paginas web
  • Un vector store respaldado por ChromaDB con metadatos
  • Un pipeline de recuperacion con busqueda hibrida y re-ranking
  • Una interfaz de chat que genera respuestas fundamentadas
  • Una suite de verificacion de calidad para comprobar que tu sistema funciona correctamente

Paso 1: Configuracion 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 Ingestion 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 Recuperacion

Crea src/retriever.py -- la capa de recuperacion con busqueda hibrida:

"""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 generacion:

"""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, aqui hay formas de extenderlo:

  • Agrega mas tipos de documentos. Integra cargadores de Markdown, archivos de codigo 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 precision de recuperacion.
  • Implementa control de acceso. Agrega roles de usuario y filtra documentos basandose 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 integracion con otros sistemas.

Resumen del Curso

A lo largo de doce lecciones, has aprendido:

  1. Por que RAG importa y que problemas resuelve
  2. Como los embeddings codifican significado en vectores buscables
  3. Como las bases de datos vectoriales almacenan y recuperan esos vectores a escala
  4. Como procesar documentos de PDFs, HTML, codigo y mas
  5. Como las estrategias de chunking afectan la calidad de recuperacion
  6. Como tecnicas avanzadas de recuperacion como busqueda hibrida y re-ranking mejoran resultados
  7. Como construir un pipeline RAG completo con LangChain
  8. Como patrones avanzados como HyDE, self-RAG y RAG agentico van mas alla de la recuperacion basica
  9. Como construir sistemas RAG especificos para codigo
  10. Como medir tu sistema con metricas reales
  11. Como endurecer todo para produccion
  12. Como construir una base de conocimiento completa desde cero

RAG es una de las habilidades mas practicas e impactantes en ingenieria de IA hoy. Los sistemas que construyas con estas tecnicas conectaran modelos de lenguaje poderosos con datos del mundo real, produciendo respuestas que son precisas, actuales y verificables. Ve y construye algo util.