Frameworks de Orquestación

¿Por Qué Orquestación?

Las aplicaciones LLM reales involucran múltiples pasos: recuperar datos, formatear prompts, llamar al modelo, parsear respuestas, manejar errores, hacer retry. Los frameworks de orquestación abstraen esta complejidad.

LangChain

El framework más popular para construir aplicaciones con LLMs.

Conceptos Core

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

# 1. Modelo
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 2. Prompt Template
prompt = ChatPromptTemplate.from_messages([
    ("system", "Eres un experto en {topic}. Responde en {language}."),
    ("human", "{question}"),
])

# 3. Chain (LCEL - LangChain Expression Language)
chain = prompt | llm | StrOutputParser()

# 4. Ejecutar
result = chain.invoke({
    "topic": "Python",
    "language": "español",
    "question": "¿Qué son los decoradores?"
})

RAG Chain con LangChain

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import RunnablePassthrough

# Setup
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Prompt RAG
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """Responde basándote en el contexto. Si no sabes, di "No lo sé".
    
Contexto: {context}"""),
    ("human", "{question}"),
])

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# RAG Chain
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("¿Cómo configuro Docker?")

LlamaIndex

Especializado en conectar LLMs con datos. Excelente para RAG.

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.llms.openai import OpenAI

# Cargar documentos
documents = SimpleDirectoryReader("./data").load_data()

# Crear índice (chunking + embeddings automáticos)
index = VectorStoreIndex.from_documents(documents)

# Query engine
query_engine = index.as_query_engine(
    llm=OpenAI(model="gpt-4o"),
    similarity_top_k=5,
)

response = query_engine.query("¿Cuál es la política de vacaciones?")
print(response.response)
print(f"Fuentes: {[n.metadata for n in response.source_nodes]}")

Chat Engine

chat_engine = index.as_chat_engine(
    chat_mode="condense_plus_context",
    llm=OpenAI(model="gpt-4o"),
)

response = chat_engine.chat("¿Cuántos días de vacaciones tengo?")
response = chat_engine.chat("¿Y si soy senior?")  # Mantiene contexto

Comparación

Construir Sin Framework

A veces, un framework es overkill. Para casos simples:

class SimpleLLMPipeline:
    """Pipeline ligero sin dependencias externas."""
    
    def __init__(self, client, model="gpt-4o"):
        self.client = client
        self.model = model
        self.steps = []
    
    def add_step(self, name: str, fn):
        self.steps.append({"name": name, "fn": fn})
        return self
    
    def run(self, input_data: dict) -> dict:
        data = input_data.copy()
        
        for step in self.steps:
            try:
                data = step["fn"](data, self.client, self.model)
                data["_last_step"] = step["name"]
            except Exception as e:
                data["_error"] = f"{step['name']}: {str(e)}"
                break
        
        return data

# Uso
def extract_entities(data, client, model):
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": f"Extrae entidades de: {data['text']}"}],
        response_format={"type": "json_object"},
    )
    data["entities"] = json.loads(resp.choices[0].message.content)
    return data

def classify_intent(data, client, model):
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": f"Clasifica: {data['text']}"}],
    )
    data["intent"] = resp.choices[0].message.content.strip()
    return data

pipeline = SimpleLLMPipeline(client)
pipeline.add_step("extract", extract_entities)
pipeline.add_step("classify", classify_intent)

result = pipeline.run({"text": "Quiero cancelar mi pedido #12345"})

¿Cuándo Usar Qué?

Resumen

LangChain y LlamaIndex son los frameworks dominantes. LangChain para orquestación compleja y agentes, LlamaIndex para RAG rápido. Para producción, evalúa si realmente necesitas un framework o si un pipeline propio es más mantenible y con menos dependencias.