LLMZ25-5 Review : Arquitectura RAG del Proyecto Ziweidoushu

En este post, exploramos cómo LangChain se integra en un sistema RAG (Retrieval-Augmented Generation) real para crear un asistente de astrología china (紫微斗数). El proyecto ziweidoushu es un ejemplo de arquitectura de producción que utiliza LangChain para orquestar un flujo complejo de procesamiento de documentos, recuperación híbrida, y generación de respuestas personalizadas.

Repositorio del Proyecto: github.com/clementlwm94/ziweidoushu

Ziweidoushu es un sistema que:

1. Genera cartas natales chinas desde datos de nacimiento
2. Recupera conocimiento relevante de una base de datos vectorial (Qdrant)
3. Combina información específica del usuario con conocimiento del dominio
4. Genera respuestas personalizadas usando LLMs (GPT-4)

---

¿Qué es LangChain y por qué usarlo?

LangChain es un framework de Python diseñado para construir aplicaciones basadas en LLMs (Large Language Models). Proporciona:

• Chains (Cadenas): Flujos de trabajo pre-construidos que conectan múltiples componentes
• Agents (Agentes): Componentes que pueden tomar decisiones y usar herramientas
• Memory (Memoria): Gestión de estado conversacional
• Prompts: Plantillas reutilizables para interacciones con LLMs
• Retrievers: Integraciones con bases de datos vectoriales y sistemas de búsqueda

Ventajas de usar LangChain:

• ✅ Modularidad: Componentes intercambiables
• ✅ Escalabilidad: Fácil agregar nuevas funcionalidades
• ✅ Observabilidad: Integración nativa con Phoenix/OpenTelemetry
• ✅ Estándares: Protocolo LCEL (LangChain Expression Language)

---

Arquitectura del Proyecto Ziweidoushu

El flujo de trabajo de ziweidoushu implementa un pipeline RAG de 9 etapas usando LangChain:

Usuario → Generación de Carta → Generación de Consultas → Recuperación → Resumen → Generación de Respuesta → Respuesta Final

---

Componentes LangChain Utilizados

1. PromptTemplate: Gestión de Prompts Especializados

LangChain PromptTemplate permite crear prompts reutilizables con variables dinámicas. En el proyecto, se definen múltiples prompts especializados:

from langchain.prompts import PromptTemplate

query_prompt_text = """
角色: 你是一个专业的紫微斗数检索查询生成器

目标: 基于用户的命盘关键信息与具体问题，生成若干高质量、可直接用于检索的查询字符串

约束:
- 仅使用用户的命盘中出现的宫位、主星、辅星、煞星；不要编造
- 使用标准中文术语与命名
- 生成 1-3 条查询，按相关性从高到低排序

输入:
- 用户命盘: {user_chart}
- 用户问题: {user_question}

输出: 仅输出 JSON 字符串数组
"""

query_prompt = PromptTemplate(
    input_variables=["user_question", "user_chart"],
    template=query_prompt_text,
)

Ventajas:

• Separación de lógica y prompts
• Reutilización en múltiples contextos
• Fácil mantenimiento y actualización

---

2. RunnableLambda: Orquestación del Pipeline

RunnableLambda permite encadenar funciones Python en un flujo de trabajo secuencial. El proyecto implementa un pipeline de 9 etapas:

from langchain_core.runnables import RunnableLambda

def _with_chart(inputs: Dict[str, Any]) -> Dict[str, Any]:
    """Etapa 1: Genera la carta natal"""
    chart = full_chart_generation(inputs["birth_date"], inputs["birth_hour"], inputs["gender"])
    return {**inputs, "user_chart": chart}

def _with_queries(inputs: Dict[str, Any]) -> Dict[str, Any]:
    """Etapa 4: Genera consultas de búsqueda"""
    queries = generate_queries(
        inputs["translated_question"],
        inputs["user_chart"],
        top_n=inputs.get("top_n_queries", 3),
        llm=inputs.get("llm"),
    )
    return {**inputs, "queries": queries}

# Pipeline completo con RunnableLambda
workflow_chain = (
    RunnableLambda(_with_chart)          # 1. Generar carta
    | RunnableLambda(_with_llm)          # 2. Inicializar LLM
    | RunnableLambda(_with_translated_question)  # 3. Traducir pregunta
    | RunnableLambda(_with_queries)      # 4. Generar consultas
    | RunnableLambda(_with_documents)     # 5. Recuperar documentos
    | RunnableLambda(_with_documents_text) # 6. Formatear documentos
    | RunnableLambda(_with_summary)       # 7. Resumir pasajes
    | RunnableLambda(_with_answer)         # 8. Generar respuesta
    | RunnableLambda(_with_localized_answer) # 9. Localizar respuesta
    | RunnableLambda(_finalize)           # Finalizar
)

¿Por qué RunnableLambda?

• Expresa flujo secuencial claramente
• Permite paso de estado entre etapas
• Integración con herramientas de observabilidad (Phoenix)
• Composición flexible con operador |

---

3. ParentDocumentRetriever: Recuperación Jerárquica

ParentDocumentRetriever implementa un patrón de recuperación jerárquico que combina chunks pequeños (para búsqueda precisa) con documentos padres completos (para contexto):

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import LocalFileStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Almacén de documentos padres
fs = LocalFileStore(STORE_PATH)
doc_store = create_kv_docstore(fs)

# Splitter para chunks hijos
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# Retriever jerárquico
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,           # Busca chunks pequeños
    docstore=doc_store,                 # Trae documentos padres completos
    child_splitter=child_splitter,
    search_kwargs={"k": 10},           # Top-10 chunks
    id_key="source_id"
)

¿Cómo funciona?

1. Busca en vector DB usando chunks pequeños (400 chars)
2. Recupera documentos padres completos desde docstore
3. Proporciona contexto más rico que solo chunks

Ventajas:

• ✅ Precisión de búsqueda (chunks pequeños)
• ✅ Contexto completo (documentos padres)
• ✅ Mejor calidad de RAG

---

4. ContextualCompressionRetriever: Re-ranking con Flashrank

ContextualCompressionRetriever mejora la calidad de documentos recuperados mediante re-ranking contextual:

from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_community.document_compressors import FlashrankRerank

# Compresor de contexto
compressor = FlashrankRerank()

# Retriever con compresión
compression_retriever = ContextualCompressionRetriever(
    base_retriever=retriever,
    base_compressor=compressor,
)

¿Qué hace FlashrankRerank?

• Usa un modelo de re-ranking (MRM3) entrenado para relevancia
• Reordena documentos por relevancia contextual
• Mejora la precisión del Top-K

Ventajas:

• ✅ Reduce ruido de recuperación inicial
• ✅ Mejora precisión en Top-5/Top-10
• ✅ Mejor calidad de contexto para LLM

---

5. ChatOpenAI: Integración con LLMs

ChatOpenAI proporciona una interfaz unificada para interactuar con modelos OpenAI:

from langchain_openai import ChatOpenAI

def _create_llm(api_key: Optional[str], llm_model: Optional[str]) -> ChatOpenAI:
    target_model = llm_model or LLM_MODEL  # "gpt-4o-mini"
    return ChatOpenAI(
        model=target_model,
        temperature=LLM_TEMPERATURE,  # 0.3
        api_key=api_key
    )

Ventajas:

• API unificada (no importa si usas GPT-4, Claude, etc.)
• Integración con LCEL (LangChain Expression Language)
• Manejo automático de rate limiting
• Streaming support

---

6. QdrantVectorStore: Búsqueda Híbrida

QdrantVectorStore integra Qdrant con LangChain para búsqueda híbrida (dense + sparse):

from langchain_qdrant import QdrantVectorStore, FastEmbedSparse, RetrievalMode
from langchain_community.embeddings.fastembed import FastEmbedEmbeddings

# Embeddings densos (semánticos)
embeddings = FastEmbedEmbeddings(model_name="jinaai/jina-embeddings-v2-base-zh")

# Embeddings sparse (keyword-based)
sparse_embeddings = FastEmbedSparse(model_name="Qdrant/BM25")

# Vector store con modo híbrido
vectorstore = QdrantVectorStore(
    embedding=embeddings,                 # Dense embeddings
    client=client,
    collection_name=COLLECTION_NAME,
    sparse_embedding=sparse_embeddings,   # Sparse embeddings
    retrieval_mode=RetrievalMode.HYBRID,  # 🎯 Modo híbrido
)

¿Qué es búsqueda híbrida?

• Dense: Captura similitud semántica
• Sparse (BM25): Captura coincidencias de keywords
• Híbrida: Combina ambos para mejor recall

Ventajas:

• ✅ Mejor recall que solo dense o solo sparse
• ✅ Optimizado para preguntas específicas de dominio
• ✅ Embeddings optimizados para chino (jina-v2-base-zh)

---

Flujo de Trabajo Completo

El pipeline completo de ziweidoushu con LangChain:

graph TD
    A[User Question + Birth Data] --> B[Generate Chart]
    B --> C[Translate Question]
    C --> D[Generate Queries]
    D --> E[Hybrid Retrieval]
    E --> F[Flashrank Rerank]
    F --> G[Summarize Passages]
    G --> H[Generate Answer]
    H --> I[Translate Answer]
    I --> J[Final Response]

Implementación con LangChain:

def run_qdrant_rag_workflow(
    question: str,
    birth_date: str,
    birth_hour: int,
    gender: str,
    top_n_queries: int = 5,
    api_key: Optional[str] = None,
    llm_model: Optional[str] = None,
) -> Dict[str, Any]:
    """Ejecutar el workflow RAG completo de Qdrant."""

    # Crear payload inicial
    payload = {
        "question": question,
        "birth_date": birth_date,
        "birth_hour": birth_hour,
        "gender": gender,
        "top_n_queries": top_n_queries,
        "api_key": api_key,
        "llm_model": llm_model,
    }

    # Ejecutar pipeline completo
    result = workflow_chain.invoke(payload)

    return result

---

Ventajas de Usar LangChain

1. Modularidad Extrema

   • Cada componente es intercambiable
   • Fácil cambiar de Qdrant a Pinecone
   • Fácil cambiar de GPT-4 a Claude

2. Observabilidad Integrada

   • Traces automáticos con Phoenix
   • Mide latencia, tokens, costos
   • Debug más fácil con spans detallados

3. LCEL (LangChain Expression Language)

   • Composición fluida con |
   • Streaming nativo
   • Batching automático

4. Retrievers Reutilizables

   • ParentDocumentRetriever con docstore
   • ContextualCompressionRetriever con re-ranking
   • Fácil cambiar estrategias de recuperación

5. Ecosistema Maduro

   • Integración con 100+ herramientas
   • Documentación excelente
   • Comunidad activa

---

Lecciones Aprendidas

1. Usa RunnableLambda para pipelines secuenciales complejos

   • Más flexibilidad que SequentialChain
   • Permite paso de estado complejo
   • Integración mejor con observabilidad

2. Separa prompts del código

   • PromptTemplate hace código más limpio
   • Fácil iterar en prompts sin tocar lógica
   • Reutilización en múltiples contextos

3. Invierte en retrieval quality

   • ParentDocumentRetriever + ContextualCompressionRetriever = mejor RAG
   • Búsqueda híbrida mejora recall
   • Re-ranking mejora precisión

4. Monitorea todo

   • Integra Phoenix desde el inicio
   • Mide métricas clave (latency, quality)
   • Debug con traces

5. Optimiza para tu dominio

   • Embeddings específicos (jina-v2-base-zh para chino)
   • Prompts especializados por tarea
   • Chunking adaptado a tu contenido

---

Conclusión

El proyecto ziweidoushu demuestra cómo LangChain puede usarse para construir un sistema RAG de producción robusto y escalable. Los componentes clave que hacen esto posible son:

• RunnableLambda: Orquestación flexible del pipeline
• PromptTemplate: Gestión de prompts especializados
• ParentDocumentRetriever: Recuperación jerárquica
• ContextualCompressionRetriever: Re-ranking contextual
• QdrantVectorStore: Búsqueda híbrida
• ChatOpenAI: Integración LLM

La arquitectura resultante es modular, observable y mantenible, lo que permite:

• Iteración rápida en prompts
• Cambio fácil de componentes
• Debug eficiente con Phoenix
• Escalabilidad hacia más funcionalidades

---

Referencias

• LangChain Documentation
• Arize Phoenix