Inicio / Artículos / RAG: De los principios fundamentales a tres agentes esenciales

rag embeddings bedrock langgraph retrieval aws

RAG: De los principios fundamentales a tres agentes esenciales

Por qué los pesos fallan con datos privados, cómo funcionan los embeddings y la búsqueda por coseno, y tres agentes progresivos desde una base de conocimientos gestionada hasta la ingestión autogestionada.

4464 palabras

Los modelos base congelados no pueden responder a preguntas empresariales privadas basándose únicamente en sus pesos. El entrenamiento nunca incluyó políticas internas de recursos humanos, contratos ni guías operativas; los puntos de corte congelan el tiempo; y cuando faltan datos, el generador sigue produciendo suposiciones coherentes. Por lo tanto, enseñar a una red lo que sabe una empresa se convierte en un problema de sistemas, no en un problema relacionado con puntos de verificación adicionales.

Por qué los pesos por sí solos fallan

Tres limitaciones estructurales impiden las preguntas y respuestas empresariales:

Los corpus privados nunca se incluyeron en el entrenamiento. Las reglas de retención y las macros de tickets no son texto público en la web.
El tiempo se detiene en el punto de corte. Las modificaciones legales del mes pasado son invisibles para una instantánea congelada.
El silencio no es la opción por defecto. La predicción del siguiente token prefiere una continuación plausible antes que admitir abiertamente lo desconocido.

Tres formas de introducir conocimiento y cuáles son escalables

Opción 1: Ajuste fino en el corpus

El entrenamiento adicional puede inculcar estilo o habilidades, pero difumina los hechos entre los parámetros sin citaciones, y no permite controlar la visibilidad por usuario ya que todos comparten las mismas ponderaciones. Para preguntas y respuestas basadas en políticas que requieren citaciones y control de acceso, el ajuste fino es el mecanismo inadecuado.

Opción 2: Pegar todo en la instrucción del sistema

Entregar texto al hacer la solicitud convierte la recuperación de información en una lectura, lo cual es el mecanismo adecuado. Funciona bien con conjuntos pequeños y estables. Falla cuando el volumen de datos supera los límites, cuando se cobra por cada token en cada solicitud, cuando las respuestas quedan ocultas entre otras instrucciones y degradan su calidad, y cuando un texto estático no permite filtrar según el rol.

Opción 3: Pegar solo lo importante

Mantener el mecanismo de lectura al hacer la solicitud y corregir su escala. Obtener un número limitado de párrafos relevantes y luego generar la respuesta. Eso es, en una frase, generación mejorada mediante recuperación de información.

Qué es RAG, de principio a fin

Offline: se ingieren archivos, se dividen en segmentos, se incrustan los segmentos y se almacenan los vectores (y a menudo índices léxicos). En línea: se incrusta la pregunta, se obtienen los segmentos más relevantes, se agrupan con la pregunta en una instrucción y se genera una respuesta basada en esos segmentos.

La parte difícil no es la última llamada de generación, sino lograr que la obtención de datos sea precisa, económica y respetuosa con los permisos.

Vectores e incrustaciones sin misticismo

Un vector aquí es una lista de números que ubica un fragmento de texto en un espacio donde los elementos con significados similares se encuentran cerca unos de otros. Una red de incrustación mapea el texto a ese espacio. La similitud coseno compara más la dirección que la longitud, por lo que domina en la puntuación de recuperación. Las métricas de distancia que ignoran la normalización rinden peor cuando los segmentos tienen longitudes desiguales.

Elegir la dimensionalidad y el modelo de incrustación es una decisión relacionada con los recursos: más dimensiones pueden separar mejor los significados, pero aumentan el costo de almacenamiento y la latencia; utilizar modelos diferentes en momentos distintos, como durante la indexación o las consultas, afecta negativamente a la búsqueda de vecinos más cercanos.

Aritmética de la similitud

El coseno corresponde al producto escalar de vectores unitarios. Dos fragmentos sobre restablecimiento de contraseñas con redacción diferente siguen siendo similares si el modelo de incrustación capturó la intención subyacente. La búsqueda por palabras clave no detecta esa similitud; en cambio, la recuperación densa sí la identifica — aunque podría pasar por alto tokens exactos de SKU a menos que se combine con una búsqueda léxica híbrida.

Creación de tres agentes completos en entornos de inferencia en la nube gestionados

El resto de esta guía materializa las ideas a través de tres agentes progresivos en modelos gestionados al estilo AWS Bedrock: una base de conocimientos curada, un orquestador LangGraph y un pipeline de ingestión autogestionado. Cada descripción representa un boceto completo del agente; véanlas como arquitecturas ejecutables, no como fragmentos.

Agente 1 — Ruta de la base de conocimientos gestionada

"""
AGENT 1: Managed Knowledge Base. The 30-line version.
Bedrock does the chunking, embedding, indexing and searching. You wire it up.
"""
import boto3
from langchain_aws import AmazonKnowledgeBasesRetriever, ChatBedrockConverse
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

REGION = "eu-west-1"
KB_ID = "ABCD1234EF"

retriever = AmazonKnowledgeBasesRetriever(
    knowledge_base_id=KB_ID,
    region_name=REGION,
    retrieval_config={
        "vectorSearchConfiguration": {
            "numberOfResults": 5,
            "overrideSearchType": "HYBRID",
        }
    },
)

llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
    max_tokens=1024,
)

PROMPT = ChatPromptTemplate.from_messages([
    ("system",
     "Answer only from the numbered context. Cite every claim as [n]. "
     "If the context does not contain the answer, say so plainly."),
    ("human", "Context:\n{context}\n\nQuestion: {question}"),
])


def format_docs(docs) -> str:
    return "\n\n".join(
        f"[{i}] source={d.metadata.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{d.page_content}"
        for i, d in enumerate(docs, start=1)
    )


chain = (
    RunnableParallel(context=retriever | format_docs, question=RunnablePassthrough())
    | PROMPT
    | llm
    | StrOutputParser()
)

if __name__ == "__main__":
    print(chain.invoke("How fast must we submit a shortlist for a niche engineering role?"))

Esta opción se basa en una base de conocimientos alojada: los documentos se almacenan, un indexador gestionado genera embeddings y el agente responde con las citas proporcionadas por dicho servicio. Las ventajas operativas incluyen menos código de indexación personalizado; las desventajas, un control menor sobre los límites de los segmentos y su fusión.

Agente 2 — Orquestación con LangGraph de las herramientas

"""
AGENT 2: LangGraph agent that decides WHETHER and WHICH index to search.
Retrieval is a tool. Identity travels in typed context, never in the prompt.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any

import boto3
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware
from langchain.tools import ToolRuntime
from langchain_aws import ChatBedrockConverse
from langchain_core.tools import tool

REGION = "eu-west-1"
POLICY_KB = "ABCD1234EF"
ENGINEERING_KB = "WXYZ5678GH"

agent_runtime = boto3.client("bedrock-agent-runtime", region_name=REGION)


@dataclass
class RequestContext:
    """Built from validated JWT claims. The model cannot read or forge it."""
    subject: str
    roles: tuple[str, ...]
    region: str


def _retrieve(kb_id: str, query: str, ctx: RequestContext, k: int = 5) -> str:
    """Bedrock Retrieve API with a server-side metadata filter for tenant + residency."""
    resp = agent_runtime.retrieve(
        knowledgeBaseId=kb_id,
        retrievalQuery={"text": query},
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": k,
                "overrideSearchType": "HYBRID",
                "filter": {
                    "andAll": [
                        {"in": {"key": "acl_roles", "value": list(ctx.roles)}},
                        {"in": {"key": "region", "value": [ctx.region, "GLOBAL"]}},
                    ]
                },
            }
        },
    )
    results: list[dict[str, Any]] = resp.get("retrievalResults", [])
    if not results:
        return "NO_AUTHORISED_RESULTS: nothing in this index that you may read."
    return "\n\n".join(
        f"[{i}] score={r.get('score'):.3f} "
        f"source={r.get('location', {}).get('s3Location', {}).get('uri', '?')}\n"
        f"{r['content']['text']}"
        for i, r in enumerate(results, start=1)
    )


@tool
def search_policy_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search HR, compliance and commercial policy.

    Use for shortlist SLAs and service credits, candidate data retention and consent,
    GDPR erasure, statement-of-work approval thresholds, background check rules and
    automated decision governance.
    """
    return _retrieve(POLICY_KB, query, runtime.context)


@tool
def search_engineering_docs(query: str, runtime: ToolRuntime[RequestContext]) -> str:
    """Search platform runbooks and integration documentation.

    Use for the requisition sync service, Workday and SuccessFactors webhooks, Kafka
    topics and event schemas, idempotency and retry behaviour, and vector index refresh.
    """
    return _retrieve(ENGINEERING_KB, query, runtime.context)


agent = create_agent(
    model=ChatBedrockConverse(
        model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
        region_name=REGION,
        temperature=0,
    ),
    tools=[search_policy_docs, search_engineering_docs],
    system_prompt=(
        "You are an enterprise knowledge assistant.\n"
        "- Answer only from tool results and cite the [n] markers they return.\n"
        "- If a tool returns NO_AUTHORISED_RESULTS, say you cannot access a source and "
        "name the document owner. Never fill the gap from memory.\n"
        "- For questions spanning policy and platform, call both tools before answering."
    ),
    context_schema=RequestContext,
    middleware=[
        ModelCallLimitMiddleware(thread_limit=8, run_limit=6, exit_behavior="end"),
        ToolCallLimitMiddleware(thread_limit=6, run_limit=5, exit_behavior="end"),
    ],
)

if __name__ == "__main__":
    claims = {"sub": "asha@example.com", "roles": ["recruiter"], "region": "EU"}
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "How is the vector index refreshed?"}]},
        context=RequestContext(
            subject=claims["sub"],
            roles=tuple(claims["roles"]),
            region=claims["region"],
        ),
    )
    print(result["messages"][-1].content)

Aquí el flujo de control es un grafo explícito: se clasifica la intención, se llaman las herramientas de recuperación, opcionalmente se consulta a un humano y luego se responde. Los puntos de control y las interrupciones convierten los hilos de soporte de larga duración en sesiones duraderas en lugar de diccionarios en memoria.

Agente 3: Ingestión y recuperación autogestionadas

"""
AGENT 3: Self-managed pipeline: your chunking, your index, your fusion, your rerank.
Chosen when Bedrock KB's fixed options are not enough (custom fusion, per-stage timing,
an abstention gate, or a chunking rule the managed service will not express).
"""
from __future__ import annotations
import time
from typing import Annotated, Any, TypedDict

import boto3
from langchain_aws import BedrockEmbeddings, ChatBedrockConverse
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from langgraph.graph import END, START, StateGraph

REGION = "eu-west-1"
CONFIDENCE_FLOOR = 0.34
CANDIDATE_K, CONTEXT_K = 20, 4

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v2:0",
    region_name=REGION,
    model_kwargs={"dimensions": 1024, "normalize": True},
)
llm = ChatBedrockConverse(
    model="eu.anthropic.claude-sonnet-4-5-20250929-v1:0",
    region_name=REGION,
    temperature=0,
)
rerank_client = boto3.client("bedrock-agent-runtime", region_name=REGION)


# ---------------------------------------------------------------- ingest
def chunk_policy_document(markdown: str, source: str, acl: list[str], region: str):
    """Structure-aware first, size-capped second. Headings are the natural unit for
    policy text; the recursive splitter only intervenes on oversized sections."""
    by_heading = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("#", "title"), ("##", "section"), ("###", "clause")]
    ).split_text(markdown)

    capper = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)
    chunks: list[Document] = []
    for part in capper.split_documents(by_heading):
        heading = " > ".join(
            v for k, v in part.metadata.items() if k in ("title", "section", "clause")
        )
        chunks.append(
            Document(
                # contextual retrieval: the chunk carries its own scope
                page_content=f"{heading}\n{part.page_content}",
                metadata={**part.metadata, "source": source, "acl": acl, "region": region},
            )
        )
    return chunks


# ---------------------------------------------------------------- state
class State(TypedDict, total=False):
    question: str
    roles: list[str]
    region: str
    candidates: list[Document]
    context_docs: list[Document]
    confidence: float
    answer: str
    timings: Annotated[dict[str, float], lambda a, b: {**(a or {}), **(b or {})}]


def _ms(name: str, t0: float) -> dict[str, float]:
    return {name: round((time.perf_counter() - t0) * 1000, 1)}


# ---------------------------------------------------------------- nodes
def retrieve(state: State, *, store) -> State:
    """Hybrid search with an ACL pre-filter pushed into the index query."""
    t0 = time.perf_counter()
    docs = store.similarity_search(
        state["question"],
        k=CANDIDATE_K,
        filter={"acl": {"$in": state["roles"]}, "region": {"$in": [state["region"], "GLOBAL"]}},
    )
    return {"candidates": docs, "timings": _ms("retrieve", t0)}


def rerank(state: State) -> State:
    """Bedrock Rerank: a cross-encoder scoring each (query, chunk) pair jointly."""
    t0 = time.perf_counter()
    cands = state["candidates"]
    if not cands:
        return {"context_docs": [], "confidence": 0.0, "timings": _ms("rerank", t0)}

    resp = rerank_client.rerank(
        queries=[{"type": "TEXT", "textQuery": {"text": state["question"]}}],
        sources=[
            {"type": "INLINE", "inlineDocumentSource": {
                "type": "TEXT", "textDocument": {"text": d.page_content}}}
            for d in cands
        ],
        rerankingConfiguration={
            "type": "BEDROCK_RERANKING_MODEL",
            "bedrockRerankingConfiguration": {
                "numberOfResults": CONTEXT_K,
                "modelConfiguration": {
                    "modelArn": f"arn:aws:bedrock:{REGION}::foundation-model/amazon.rerank-v1:0"
                },
            },
        },
    )
    ranked = resp["results"]
    top = [cands[r["index"]] for r in ranked]
    return {
        "context_docs": top,
        "confidence": round(float(ranked[0]["relevanceScore"]), 3),
        "timings": _ms("rerank", t0),
    }


def generate(state: State) -> State:
    """Abstain below the floor rather than answering from weak context."""
    t0 = time.perf_counter()
    if not state["context_docs"] or state["confidence"] < CONFIDENCE_FLOOR:
        return {
            "answer": "I cannot ground an answer for that in a source you are permitted "
                      "to read. Routing to the document owner.",
            "timings": _ms("generate", t0),
        }
    context = "\n\n".join(
        f"[{i}] {d.metadata.get('source')}\n{d.page_content}"
        for i, d in enumerate(state["context_docs"], start=1)
    )
    prompt = ChatPromptTemplate.from_messages([
        ("system", "Answer only from the numbered context and cite each claim as [n]."),
        ("human", "Context:\n{context}\n\nQuestion: {question}"),
    ])
    reply = (prompt | llm).invoke({"context": context, "question": state["question"]})
    return {"answer": reply.content, "timings": _ms("generate", t0)}


def build_graph(store):
    g = StateGraph(State)
    g.add_node("retrieve", lambda s: retrieve(s, store=store))
    g.add_node("rerank", rerank)
    g.add_node("generate", generate)
    g.add_edge(START, "retrieve")
    g.add_edge("retrieve", "rerank")
    g.add_edge("rerank", "generate")
    g.add_edge("generate", END)
    return g.compile()

Esta variante se encarga del particionamiento, de las llamadas a modelos de embedding, del almacenamiento vectorial y de la composición de prompts. Los equipos la eligen cuando las políticas de segmentación, la fusión híbrida léxica y densa o los filtros de tenencia deben ser parte integral del código en lugar de opciones configurables desde la consola.

Elegir entre las tres opciones

Empiece con la opción gestionada cuando sea importante la rapidez para demostraciones y los corpus sean modestos. Pase a LangGraph cuando predominen las ramificaciones, la revisión humana y el mantenimiento de la memoria a lo largo de las interacciones. Tome el control total del proceso cuando la calidad de la recuperación de identificadores, ACLs o segmentaciones de dominio se convierta en el factor clave del producto.

Evaluación que realmente mejora la calidad

Controlar el Recall@k y la Precisión@k en un conjunto de preguntas fijo, además de la fidelidad de las respuestas a los segmentos recuperados. Asociar cada cambio en el tamaño del segmento o en el mecanismo de incrustación con las puntuaciones antes y después. Rechazar los resultados cuando los segmentos principales sean débiles: un silencio calibrado es mejor que una invención fluida.

Cuándo no usar RAG

Omitir la recuperación de información para datos triviales que la red base ya conoce, en tareas puramente creativas, para números que cambian continuamente y que se manejan mejor con APIs, y en casos de latencia ultra baja donde no se puede permitir una solicitud adicional. A veces, unas instrucciones más claras o un ligero ajuste en el formato son suficientes.

Cierre

RAG es como la plomería: una entrada limpia, una recuperación honesta, instrucciones precisas para la generación y controles medibles. Los tres agentes muestran la misma idea en tres niveles de control diferentes. Elija el nivel que se ajuste a sus necesidades de gestión, no el que suene más avanzado en una diapositiva.

Lista de verificación operativa: (1) un incrustador para índice y búsquedas, (2) metadatos para ACL y fechas desde el primer día, (3) búsqueda híbrida cuando aparecen identificadores, (4) reclasificación cuando es importante la precisión, (5) preguntas clave en CI, (6) disparadores de reconstrucción cuando cambian las fuentes. Trate el plano de conocimiento como un producto con responsables, SLOs y mecanismos de reversión, no como un índice del fin de semana que nadie monitorea. A medida que crecen los grafos de agentes, mida el costo en tokens junto con la calidad de las respuestas; una orquestación que consume quince veces más recursos que un chat simple sin mantener la fidelidad es una regresión disfrazada de arquitectura. Mantenga manuales para casos de resultados vacíos, incidentes de índice obsoleto y desajustes en los incrustadores, para que el personal de soporte no tenga que improvisar basándose únicamente en los registros de chat.

Documente cada fuente del corpus con un responsable y una frecuencia de actualización; los contenedores abandonados se convierten en problemas silenciosos y crónicos.

Preferir una superposición temprana de segmentos; solo utilizar separadores semánticos cuando las evaluaciones demuestren que las ventanas fijas no funcionan.

Registrar los IDs de las solicitudes junto con las respuestas para poder diagnosticar si las respuestas defectuosas se deben a fallos en la obtención de datos o a desviaciones en la generación.

Separar a los evaluadores de la red que responde cuando sea posible, a fin de reducir el sesgo por preferencia propia.

Asignar un presupuesto distinto para los gastos de API relacionados con el indexado y otro para el servicio interactivo, de modo que los departamentos financieros puedan ver ambos montos.

Mantener plantillas de prompts de versión junto a las compilaciones del índice; un cambio en los prompts sin reevaluación es la forma en que se introducen regresiones.

En entornos multiinquilinos, probar que los filtros no puedan ser eludidos mediante inyección de prompts que pretendan roles de administrador.

Tener un interruptor de emergencia que desactive la generación y devuelva únicamente citas cuando los monitores de fidelidad detecten problemas.

Volver a revisar los pesos léxicos híbridos cada vez que los códigos de producto dominen la combinación de tráfico.

Capacite al personal de soporte para que lea las paneles de citación antes de elevar los tickets relacionados con “bots incorrectos”.

Documente cada fuente del corpus indicando su responsable y la frecuencia de actualización; los contenedores desatendidos se convierten en un problema silencioso.

Prefiera la superposición de segmentos desde el principio; solo adopte divisores semánticos cuando las evaluaciones demuestren que los ventanales fijos no funcionan.

Registre los IDs de obtención junto con las respuestas para poder diagnosticar si las respuestas defectuosas se deben a fallos en la obtención de datos o a desviaciones en la generación.

Separe a los evaluadores de la red que responde cuando sea posible, a fin de reducir el sesgo por preferencia propia.

Asigne un presupuesto separado para los gastos de API relacionados con el indexado y uno distinto para el servicio interactivo, de modo que el departamento financiero pueda ver ambos conceptos.

Versione los modelos de prompts junto con las construcciones del índice; un cambio en el prompt sin reevaluación es la forma en que se introducen regresiones.

En entornos multiinquilinos, pruebe que los filtros no puedan ser eludidos mediante inyección de prompts que pretendan asumir roles de administrador.

Mantenga un interruptor de apagado que desactive la generación y muestre solo las citas cuando disminuyan los monitores de fidelidad.

Vuelva a revisar las ponderaciones léxicas híbridas cada vez que los códigos de producto dominen la combinación de tráfico.

Capacite al personal de soporte para que lea las pantallas de citas antes de elevar los tickets de “bot incorrecto”.

Documente cada fuente del corpus con su responsable y la frecuencia de actualización; los contenedores abandonados se convierten en un problema silencioso.

Prefiera la superposición de segmentos desde el principio; solo adopte divisores semánticos cuando las evaluaciones demuestren que los períodos fijos fallan.

Registre los IDs de obtención junto con las respuestas para poder diagnosticar si las respuestas defectuosas se deben a fallos en la obtención o a desviaciones en la generación.

Separe a los jueces de evaluación de la red que responde cuando sea posible, para reducir el sesgo por preferencia propia.

Asigne un presupuesto separado para los gastos de API relacionados con el indexado y uno distinto para el servicio interactivo, de modo que las finanzas puedan ver ambos conceptos.

Plantillas de prompts de versión junto a las compilaciones del índice; un aumento en los prompts sin reevaluación es cómo se introducen las regresiones.

En entornos multiinquilinos, pruebe que los filtros no puedan ser eludidos mediante inyección de prompts que pretendan roles de administrador.

Mantenga un interruptor de emergencia que desactive la generación y devuelva solo citas cuando los monitores de fidelidad fallen.

Vuelva a revisar las ponderaciones léxicas híbridas cada vez que los códigos de producto dominen la combinación de tráfico.

Capacite al personal de soporte para que lea las pantallas de citas antes de elevar tickets relacionados con “bot incorrecto”.

Mantenga las evaluaciones emparejadas y las métricas de rechazo junto a la latencia para que el costo y la calidad aparezcan en la misma hoja.

Todos los artículos Enviar el brief