Busque los libros de procedimientos, no los pegue: un pequeño camino RAG citado en LangChain
Cree una ruta de recuperación en dos pasos en LangChain: cargue los runbooks, córtelos en fragmentos, realice búsquedas con consultas específicas, conserve la metadatos, exija IDs de fuentes citadas y pruébelo sin usar embeddings pagados.
Búsquedas en los manuales de procedimientos en lugar de llenar el prompt
A estas alturas, la aplicación del servicio de soporte se divide según el tipo de ticket. Un ticket relacionado con bases de datos incluye de inmediato el catálogo de servicios, los incidentes recientes y las implementaciones recientes, mientras que un ticket de acceso sigue otro camino. Lo que aún falta son los pasos detallados para seguir una recomendación.
Se podría pegar cada manual de procedimientos en el prompt. Funciona con tres archivos cortos, pero deja de funcionar a medida que la colección crece. Los prompts grandes cuestan más, se ejecutan más lentamente y dificultan determinar qué parte del texto sirvió de base para una respuesta. También pueden entregar al modelo documentos a los que el usuario actual nunca tuvo permiso para leer. La solución es buscar.
El proceso aquí descrito consta de dos fases. En el momento del indexado, los manuales de procedimientos en formato Markdown se convierten en objetos Document que contienen metadatos de origen, y estos se dividen en fragmentos más pequeños:
indexing time
Markdown runbooks
-> Document objects with source metadata
-> smaller chunks
-> embeddings
-> vector store
En el momento de la solicitud, la clasificación del ticket genera una consulta de búsqueda específica, que se envía al recuperador:
request time
ticket classification
-> focused search query
-> retriever
-> a few runbook passages with source IDs
-> recommendation chain
-> source-ID check
La gente suele denominar a este proceso generación mejorada mediante recuperación en dos pasos, o RAG de dos pasos. La etiqueta tiene menos importancia que el orden: primero se realiza la búsqueda, y la generación solo recibe los fragmentos obtenidos. El modelo nunca decide si se realizará una búsqueda, ni puede elegir una fuente de datos arbitraria. No se necesita ningún agente.
El ticket utilizado en todo el ejemplo
El ejemplo práctico es un ticket que indica: las solicitudes de pago en la UE fallan después del despliegue a las 14:05, y los registros de checkout-api muestran “database connection refused”. Su clasificación estructurada es la siguiente:
{
"ticket_id": "INC-2048",
"summary": "EU checkout fails after a deployment.",
"category": "database",
"severity": "SEV2",
"service": "checkout-api",
"confidence": 0.91,
"evidence": ["database connection refused"],
"missing_information": ["Affected request percentage"]
}
Una buena búsqueda devuelve el manual de operaciones de la base de datos de pagos, y no una guía sobre latencia en los pagos ni una guía de acceso a identidades. También debe contener suficientes metadatos como para responder a cuatro preguntas: de qué archivo proviene el texto, qué servicio lo posee, dónde comenzaba el fragmento en el archivo y qué IDs de fuente citó la recomendación. Es difícil trabajar con texto que no tenga metadatos de origen.
Las seis partes de la recuperación
Cada parte necesita una definición antes de ser utilizada.
Documento. Considere un Documento como un bloque de texto al que se le han añadido etiquetas. El texto puede provenir de un archivo Markdown, un PDF, una página web, una wiki o incluso de una fila en una tabla, y las etiquetas pueden registrar el ID de fuente, el título, el servicio, la ruta, el propietario, la versión o quiénes pueden verlo.
Chunk. Los documentos se dividen en partes que pueden encontrarse de forma independiente. Un mismo manual puede describir conjuntamente el diagnóstico, la recuperación y el retroceso, pero una consulta sobre las credenciales de la base de datos no tiene por qué devolver siempre el material de retroceso.
Embedding. Un modelo de embedding convierte el texto en una lista de números, y los textos con significado similar deberían quedar cerca uno del otro según el método de comparación elegido. No se trata de una verificación de hechos. Los textos similares pueden seguir estando erróneos, desactualizados o fuera de alcance.
Almacén de vectores. Un almacén de vectores guarda los vectores de los documentos y localiza aquellos cercanos. Algunos existen únicamente en memoria, mientras que otros se encuentran en una base de datos o en un servicio alojado. El lugar donde se almacenan los vectores no sustituye al control de acceso.
Recuperación. Si se le proporciona a un sistema de recuperación un texto de consulta libre, devuelve objetos Document. Se trata de una concepción más amplia que la búsqueda vectorial: detrás de la misma interfaz se puede utilizar la coincidencia por palabras clave, un almacén vectorial, un servicio de búsqueda híbrido o una API de búsqueda interna.
Generación. La etapa final toma los fragmentos seleccionados y genera una respuesta estructurada. Sea lo que sea que escriba, no puede fingir que la búsqueda devolvió algo que en realidad no existía.
La recuperación no es memoria
Un conjunto de manuales de operaciones constituye un cuerpo de conocimiento, y consultarlos se considera recuperación. El historial de chat es una secuencia de mensajes, los datos del usuario guardados son información ordinaria de la aplicación, y los puntos de control del flujo de trabajo registran el estado de ejecución. Sus reglas de retención y permisos difieren, por lo que la siguiente etapa las separa en el código.
Diseño e instalación del proyecto
La captura de estado para esta etapa tiene la siguiente estructura:
langchain-helpdesk/
├── app.py
├── chains.py
├── embeddings.py
├── retrieval.py
├── schemas.py
├── workflow.py
├── runbooks/
│ ├── checkout-database.md
│ ├── identity-access.md
│ └── payment-latency.md
└── tests/
└── test_retrieval.py
Instale los paquetes:
python -m pip install -U \
langchain \
langchain-openai \
langchain-text-splitters \
pydantic \
python-dotenv \
pytest
La aplicación hace llamadas a los embeddings del proveedor. Las pruebas sustituyen estos por un pequeño reemplazo local, de modo que no se producen llamadas al proveedor.
Paso 1: escribir guías de ejecución identificables
La guía de ejecución para el proceso de pago comienza con una línea de título:
# Checkout database connection failures
A continuación, el servicio, el propietario y la carrocería:
Service: checkout-api
Owner: commerce-platformUse this runbook when checkout-api reports database connection refused,
pool exhaustion or authentication failures.## First checks1. Compare the active database host, port and database name with the
previous successful deployment.
2. Confirm that the application secret points to the expected
credential version.
3. Test network reachability from the application environment without
printing the credential.
4. Compare connection-pool limits with the database connection limit.
El título, el servicio y el propietario son texto plano, y el cargador también copia los valores seleccionados en los metadatos. En un sistema empresarial, nunca permita que un encabezado defina el acceso. Las etiquetas de acceso deben provenir de un sistema confiable y ser verificadas cada vez que se realice una recuperación.
Paso 2: cargar cada archivo en un documento
Coloque load_runbooks en retrieval.py. Primero necesita la función auxiliar de rutas:
from pathlib import Path
Luego, el propio cargador:
from langchain_core.documents import Document
def load_runbooks(runbook_dir: Path) -> list[Document]:
documents: list[Document] = [] for path in sorted(runbook_dir.glob("*.md")):
text = path.read_text(encoding="utf-8")
title = text.splitlines()[0].removeprefix("# ").strip()
service_line = next(
line for line in text.splitlines()
if line.startswith("Service:")
)
service = service_line.split(":", 1)[1].strip() documents.append(
Document(
page_content=text,
metadata={
"source_id": path.stem,
"title": title,
"service": service,
"path": str(path),
},
)
) if not documents:
raise ValueError(f"No Markdown runbooks found in {runbook_dir}") return documents
Proceso detallado:
glob("*.md")selecciona los archivos Markdown de un directorio conocido.sorted(...)mantiene el orden de indexación local de manera reproducible. La calidad de la búsqueda no debería depender del orden del sistema de archivos, pero una entrada reproducible facilita la depuración.- El archivo se decodifica como UTF-8.
- La extracción del título y del servicio se realiza de forma sencilla intencionadamente, ya que los archivos de ejemplo siguen un formato fijo. Un cargador real debería validar la metadatos e informar sobre archivos mal formados en lugar de adivinar.
page_contentcontiene el texto que se puede buscar.metadataalmacena los campos que deben acompañar a cada bloque posterior.- La verificación final evita que un índice vacío parezca un éxito.
Paso 3: dividir los documentos
Utilice RecursiveCharacterTextSplitter:
from langchain_text_splitters import RecursiveCharacterTextSplitter
def split_runbooks(documents: list[Document]) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=420,
chunk_overlap=60,
add_start_index=True,
)
return splitter.split_documents(documents)
chunk_size=420 solicita trozos de aproximadamente 420 caracteres, lo cual es adecuado para este pequeño ejemplo. chunk_overlap=60 repite una pequeña parte del texto entre los trozos adyacentes, lo que permite mantener una oración junto a su explicación cuando el límite cae en medio. Una mayor superposición no es necesariamente mejor: implica más texto almacenado, más trabajo de incrustación y más coincidencias casi idénticas.
Con add_start_index=True, cada trozo registra la posición del carácter donde comenzó, de modo que se puede volver al archivo original. split_documents devuelve objetos Document nuevos, y cada trozo hereda los metadatos de su padre.
Elegir el tamaño del trozo
Copiar un número de un tutorial no es una decisión a tomar al azar. Primero, observe cómo están organizados sus documentos y qué es lo que realmente preguntan las personas. Un paso de solución de problemas puede requerir que su encabezado, sus condiciones y sus advertencias estén dentro del mismo bloque de texto, mientras que una larga tabla de políticas podría necesitar un divisor que tenga en cuenta la estructura. Luego, pruebe la recuperación de información basándose en preguntas etiquetadas.
Paso 4: crear el recuperador de almacenamiento vectorial
Para mantener el ejemplo sencillo, la aplicación utiliza InMemoryVectorStore:
from langchain_core.vectorstores import InMemoryVectorStore
def build_vector_retriever(runbook_dir: Path, embeddings, *, k: int = 3):
chunks = split_runbooks(load_runbooks(runbook_dir))
store = InMemoryVectorStore.from_documents(
chunks,
embedding=embeddings,
)
return store.as_retriever(search_kwargs={"k": k})
from_documents incrusta y almacena los fragmentos, mientras que as_retriever expone la interfaz estándar de recuperación. k=3 solicita tres documentos. Eso no significa que los tres sean relevantes, ni garantiza que entre ellos haya el pasaje correcto.
La aplicación crea incrustaciones de los proveedores:
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(
model=os.environ["EMBEDDING_MODEL"]
)
retriever = build_vector_retriever(
Path(__file__).parent / "runbooks",
embeddings,
)
Los modelos disponibles y sus nombres dependen de la cuenta de su proveedor, por lo que debe conservar el nombre del modelo en la configuración. Además, InMemoryVectorStore pierde su índice cuando finaliza el proceso. Esto está bien para ejemplos didácticos y pruebas pequeñas, pero no constituye un índice compartido para producción.
Paso 5: crear una consulta específica a partir de campos confiables
No realice búsquedas con una gran cantidad de datos que contengan todo el estado de la aplicación. Elija los campos de clasificación adecuados:
def build_query(classification: TicketClassification) -> str:
values = [
classification.service or "unknown service",
classification.category,
classification.summary,
*classification.evidence,
]
return " | ".join(values)
Para el boleto continuo, la consulta sale más o menos así:
checkout-api | database | EU checkout fails after a deployment. |
database connection refused
La cadena combina el servicio probable, la categoría del problema, un breve resumen y las pruebas extraídas de la solicitud. Los secretos, las instrucciones ocultas y el historial completo de mensajes no se incluyen en ella.
Paso 6: transferir los metadatos durante la conversión
Lo que devuelve el recuperador son objetos Document, y cada uno de ellos debe convertirse en un pasaje con tipo definido:
class RetrievedPassage(BaseModel):
source_id: str
title: str
service: str
start_index: int
text: str
Luego asocie el Documento con él:
def to_passage(document: Document) -> RetrievedPassage:
return RetrievedPassage(
source_id=str(document.metadata["source_id"]),
title=str(document.metadata["title"]),
service=str(document.metadata["service"]),
start_index=int(document.metadata.get("start_index", 0)),
text=document.page_content.strip(),
)
Los campos obligatorios se indican entre corchetes; por lo tanto, un fragmento que carezca de source_id, title o service genera un error de inmediato. start_index tiene un valor por defecto, ya que los fragmentos almacenados anteriormente pueden no contarlo, y la migración en producción debe decidir si este valor alternativo es aceptable.
Paso 7: crear una entrada de recomendación específica
El escritor que interactúa con el modelo recibe este contrato:
class RecommendationInput(BaseModel):
ticket_id: str
ticket_summary: str
category: Category
severity: Severity
service: str | None
evidence: list[str]
passages: list[RetrievedPassage] = Field(
min_length=1,
max_length=4,
)
Cuando no se devuelven pasajes, no hay recomendación fundamentada:
def prepare_recommendation_input(
classification: TicketClassification,
passages: list[RetrievedPassage],
) -> RecommendationInput:
if not passages:
raise ValueError("At least one runbook passage is required")
return RecommendationInput(
ticket_id=classification.ticket_id,
ticket_summary=classification.summary,
category=classification.category,
severity=classification.severity,
service=classification.service,
evidence=classification.evidence,
passages=passages[:4],
)
Si la búsqueda no encuentra nada, no se debe dejar que el modelo improvise a partir de lo que sabe en general. Dirija el ticket a una persona o utilice una solución alternativa que ya haya sido aprobada por separado.
Paso 8: solicitar IDs de origen en una salida estructurada
El esquema de resultados incluye citaciones:
class GroundedRecommendation(BaseModel):
explanation: str = Field(min_length=20, max_length=500)
next_checks: list[str] = Field(min_length=1, max_length=4)
source_ids: list[str] = Field(min_length=1, max_length=4)
needs_human_review: bool
El texto del prompt dice:
Use only the supplied ticket fields and runbook passages.
Every recommended check must be supported by at least one supplied passage.
Return the source_id values you used.
Do not claim that a check or production change was performed.
Las instrucciones del prompt son útiles, pero las verificaciones a nivel de aplicación siguen siendo importantes.
Paso 9: rechazar las citas que no se obtuvieron al realizar la búsqueda
Una vez que el escritor haya ejecutado el proceso, compare sus IDs de fuente con los fragmentos recuperados:
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError(
"Recommendation cited sources that retrieval did not return: "
+ ", ".join(sorted(unknown))
)
Esto detecta un ID de fuente fabricado y nada más. Un modelo puede citar un documento que realmente fue devuelto y aún así interpretarlo incorrectamente, por lo que siguen siendo necesarias la evaluación y la revisión humana.
Paso 10: conectar el flujo de trabajo corregido
El código en el momento de la solicitud es sencillo:
def run_retrieval_workflow(
classification: TicketClassification,
retriever,
recommendation_chain,
) -> RetrievalResult:
query = build_query(classification)
passages = retrieve_passages(retriever, query)
values = prepare_recommendation_input(classification, passages)
recommendation = recommendation_chain.invoke(values)
available = {passage.source_id for passage in passages}
unknown = set(recommendation.source_ids) - available
if unknown:
raise ValueError("Recommendation cited an unavailable source") return RetrievalResult(
query=query,
recommendation=recommendation,
passages=passages,
)
No hay bucle oculto. La búsqueda se ejecuta una vez, el generador de texto se ejecuta una vez y la verificación de origen también se ejecuta una vez.
Qué se obtiene
Un resultado luce así:
{
"query": "checkout-api | database | EU checkout fails after a deployment. | database connection refused",
"recommendation": {
"explanation": "The error matches the checkout database connection runbook.",
"next_checks": [
"Compare the active database host, port and database name with the previous successful deployment.",
"Confirm that the application secret points to the expected credential version."
],
"source_ids": ["checkout-database"],
"needs_human_review": false
},
"passages": [
{
"source_id": "checkout-database",
"title": "Checkout database connection failures",
"service": "checkout-api",
"start_index": 0
}
]
}
La API puede convertir cada ID de origen en un enlace, pero solo después de verificar que el llamante actual puede abrirlo.
Prueba de recuperación sin pagar por los embeddings
La captura de pantalla complementaria incluye KeywordEmbeddings, un pequeño sustituto de prueba que cuenta con un vocabulario fijo:
VOCABULARY = (
"database",
"connection",
"checkout",
"deployment",
"credential",
"pool",
"payment",
"latency",
"timeout",
"identity",
"access",
"role",
"claim",
)
No lo confunda con un incrustador de producción. Su único propósito es mantener las pruebas rápidas, sin conexión e deterministas. La primera prueba verifica el ticket en ejecución:
passages = retrieve_passages(
retriever,
build_query(classification()),
)
assert passages[0].source_id == "checkout-database"
assert passages[0].service == "checkout-api"
assert passages[0].start_index >= 0
Las segundas mediciones se registraron a las dos en tres consultas etiquetadas:
score = hit_at_k(
retriever,
[
("checkout database connection refused", "checkout-database"),
("payment provider latency timeout", "payment-latency"),
("identity role claim access denied", "identity-access"),
],
)
assert score == 1.0
Esto plantea una pequeña pregunta: ¿se encuentra la fuente esperada entre los dos primeros resultados? Un conjunto de pruebas real necesita muchas preguntas, negativos difíciles, documentos similares y reglas de acceso actuales. Una puntuación perfecta en tres ejemplos solo demuestra que esos tres cumplen los requisitos.
La tercera prueba introduce una recomendación falsa que menciona not-returned, y el flujo de trabajo debe rechazarla. Ejecute todo con:
pytest -q
El resultado esperado:
3 passed
Denominación de las causas de fallo en la recuperación
Cada tipo de fallo merece su propio nombre, para que sepa dónde buscar.
- Fallo de indexación: no se pudo leer un archivo, faltaban metadatos o falló una solicitud de incrustación. Nunca publique un índice incompleto como si estuviera terminado.
- Fallo de actualidad: cambió una política, pero la búsqueda siguió utilizando el manual anterior. Registre las versiones y las marcas de tiempo, y decida cómo se propagan las ediciones y eliminaciones en el índice.
- Fallo de permisos: el índice mostró un texto al que este usuario no tiene derecho a leer. Aplique filtros de acceso antes de que cualquier texto llegue al modelo, ya que ocultar el enlace después es demasiado tarde.
- Fallo de clasificación: el documento correcto está indexado, pero falta entre los resultados principales. Ajuste la consulta, los filtros, el particionamiento, las incrustaciones o la clasificación, y mida nuevamente.
El texto recuperado no es fiable
Los manuales de operaciones contienen errores y, a veces, líneas que parecen órdenes dirigidas al modelo. Trate los pasajes recuperados como material de fuente citado y nunca como reglas de la aplicación con mayor prioridad. Mantenga un comportamiento estable en el prompt del sistema, delimite el texto recuperado y restrinja las herramientas en el código de la aplicación. Ningún manual de operaciones debería poder autorizarse a sí mismo para activar una herramienta que escriba.
Cuando la búsqueda vectorial es la herramienta incorrecta
Algunas búsquedas no requieren vectores. Un ID de ticket exacto se maneja mediante una consulta sencilla a la base de datos, y un nombre de servicio conocido se relaciona directamente con una entrada en el catálogo. Los códigos de error exactos y los identificadores poco comunes se manejan mejor con búsquedas por palabras clave. Si la empresa ya cuenta con un servicio de búsqueda que indexa los documentos adecuados y controla los permisos, úselo. La búsqueda por vectores es solo una de las herramientas disponibles.
Qué falta aún para su uso en producción
Aún faltan varios elementos antes de que una empresa pueda confiar en este enfoque:
- un almacenamiento persistente en lugar de la memoria
- una persona responsable clara de cada documento
- filtros de acceso aplicados en el momento de la consulta
- manejo de versiones y eliminaciones de documentos
- un plan para el día en que cambie el modelo de incrustación
- un límite sobre la cantidad de texto que se devuelve
- pruebas etiquetadas para la recuperación de información
El marco conecta las diferentes partes, y la aplicación toma esas decisiones por sí misma.
Dónde se encuentra la aplicación de servicio al cliente
Ahora puede buscar en un conjunto controlado de manuales de procedimientos y conservar sus fuentes:
classification
-> focused query
-> retriever
-> typed passages with metadata
-> recommendation
-> source-ID check
Lo siguiente es separar el historial de mensajes, los datos del usuario guardados, el estado del flujo de trabajo en tiempo real y los puntos de control en elementos distintos. Trazar esa línea también indica si una cadena sencilla de LangChain sigue siendo suficiente o si LangGraph es más adecuado.
Leer más:
- https://docs.langchain.com/oss/python/langchain/retrieval
- https://docs.langchain.com/oss/python/langchain/knowledge-base
- https://docs.langchain.com/oss/python/integrations/splitters/index
- https://docs.langchain.com/oss/python/integrations/retrievers
Leer relacionado
- Apache Doris 4.1: Búsqueda vectorial, almacenamiento JSON, OLAP y Lakehouse — Una explicación detallada de las novedades de Apache Doris 4.1: índices vectoriales IVF, función de búsqueda BM25(), documentos JSON de 100 MB, Segment V3, cachés de consultas, escritura en Iceberg, comandos MERGE INTO y TIMESTAMPTZ.
- Incorporar un agente de LangChain en FastAPI: Herramientas, búsqueda manual, transmisión en tiempo real — Crea un asistente integrado en la aplicación con FastAPI y LangChain: un manual en PDF almacenado en ChromaDB como herramienta, contexto por usuario, historial con puntos de control y respuestas en streaming.