Incrustar un agente de LangChain en FastAPI: Herramientas, búsqueda en el manual y transmisión en tiempo real.
Crear un asistente dentro de la aplicación con FastAPI y LangChain: un manual en PDF almacenado en ChromaDB que se expone como herramienta, contexto por usuario, historial con puntos de control y respuestas en tiempo real.
Cuando una aplicación supera unas pocas pantallas, su documentación comienza a crecer descontroladamente y los usuarios dejan de leerla. Un manual de 20 páginas que explica las funciones, la configuración y las reglas del dominio es valioso, pero solo si las personas pueden encontrar respuestas sin tener que buscar mucho. Un asistente integrado en el producto puede cerrar esa brecha: responde a preguntas como “¿cómo funciona esto?” basadas en el manual y “¿qué hay en mi proyecto?” a partir de los datos propios de la aplicación.
Esta guía explica paso a paso una versión compacta y funcional de dicho asistente. Conectarás un agente LangChain a un servicio FastAPI, le proporcionarás una herramienta para leer datos de la aplicación y otra para buscar un manual en PDF almacenado en ChromaDB, pasarás al agente al usuario autenticado cuando sea necesario, guardarás el historial de conversaciones con un checkpointer de LangGraph y transmitirás la respuesta al cliente. A lo largo del proceso señalaremos las deficiencias en el código mínimo que debes corregir para que funcione sin problemas, así como los cambios necesarios antes de ponerlo en producción.
El escenario y los componentes involucrados
Imagina un equipo que desarrolla una herramienta para diseñar instalaciones fotovoltaicas. El producto comenzó siendo sencillo, pero luego incorporó paneles, inversores, estimaciones de producción, planos de techos y una larga lista de reglas de diseño. Su manual ahora cuenta con más de 20 páginas. Constantemente surgen dos tipos de preguntas:
- Preguntas sobre el producto en sí: qué hace una función, dónde se encuentra una configuración, qué regla se aplica. Las respuestas están en la documentación.
- Preguntas sobre el propio trabajo del usuario: qué inversor eligió, cuánta energía produce su sistema, qué techos ha conectado. Las respuestas se encuentran en la base de datos de la aplicación, y por defecto ningún modelo las conoce.
La Generación Aumentada por Recuperación de Información (RAG) se ocupa del primer tipo: indexa el manual, busca los pasajes relevantes cuando llega una pregunta y se los entrega al modelo como contexto. Las herramientas se encargan del segundo tipo: pequeñas funciones que el modelo puede llamar para obtener datos de la aplicación. Al combinar ambos se obtiene un asistente que puede tanto explicar el producto como razonar sobre un proyecto específico.
El sistema real detrás de este escenario cuenta con muchas más herramientas y una lógica de dominio mucho más extensa. Lo que se presenta a continuación está deliberadamente reducido a lo esencial para que la arquitectura siga siendo visible:
- FastAPI expone la API HTTP y determina quién es el solicitante.
- El agente de LangChain (que funciona en LangGraph) se encarga del bucle de razonamiento y del estado de la conversación.
- Un LLM interpreta cada solicitud y decide si necesita información externa.
- Herramientas brindan al agente acceso controlado a las funcionalidades de la aplicación.
- RAG permite al agente buscar en la documentación.
- ChromaDB almacena los fragmentos del manual y realiza búsquedas vectoriales.
- Transmisión en tiempo real envía tokens al cliente mientras aún se está generando la respuesta.
La ventaja de esta estructura es que el asistente funciona dentro de una aplicación full-stack existente con usuarios reales y datos reales, en lugar de operar como un chatbot genérico independiente.
Estructura del proyecto
Cada funcionalidad tiene su propio paquete: enrutamiento HTTP, autenticación, lógica del agente, herramientas y el pipeline RAG. Esto mantiene cada archivo de tamaño reducido y hace evidente dónde debe colocarse una nueva capacidad.
project/
│
├── main.py
├── .env
├── .gitignore
│
├── auth/
│ ├── __init__.py
│ └── dependencies.py
│
├── routers/
│ ├── __init__.py
│ └── chat.py
│
├── llm/
│ ├── __init__.py
│ ├── agent.py
│ ├── context.py
│ ├── orchestrator.py
│ ├── prompts.py
│ ├── provider.py
│ │
│ ├── tools/
│ │ ├── __init__.py
│ │ ├── demo_tool.py
│ │ └── manual_tool.py
│ │
│ └── rag/
│ ├── __init__.py
│ ├── config.py
│ ├── context.py
│ │
│ ├── ingestion/
│ │ ├── __init__.py
│ │ ├── loader.py
│ │ ├── chunker.py
│ │ ├── chroma.py
│ │ └── indexer.py
│ │
│ └── retrieval/
│ ├── __init__.py
│ └── retriever.py
│
├── scripts/
│ ├── __init__.py
│ └── index_manual.py
│
├── docs/
│ └── manual.pdf
│
└── chroma_data/ (*generated locally, not commited or deployed)
Función de cada parte:
main.pycrea la aplicación FastAPI.auth/alberga la dependencia de autenticación simulada.routers/contiene los puntos de enlace HTTP.llm/incluye todo lo relacionado con el agente.llm/tools/guarda las funciones que el agente puede llamar.
llm/rag/ alberga el pipeline de recuperación, dividido en ingestion/ (cargar, fragmentar e indexar el PDF) y retrieval/ (realizar consultas a ChromaDB).scripts/ contiene los comandos que se ejecutan manualmente, como el indexado del manual.docs/ guarda el PDF original.chroma_data/ se genera localmente y nunca debe ser comprometido ni desplegado.No creará todo esto de una sola vez. El orden de construcción es: la capa API, luego el modelo, después las herramientas, seguido del pipeline RAG, y finalmente el contexto, la transmisión en tiempo real y el historial.
Paso 1: Un esqueleto de FastAPI con un usuario simulado
Comience instalando todo lo que necesitará el proyecto. La lista incluye el servidor web, LangChain y LangGraph, el cliente de chat compatible con OpenAI, ChromaDB, los cargadores de PDF y divisores de texto, además de python-dotenv para la configuración.
pip install fastapi uvicorn langchain langgraph langchain-openai chromadb langchain-community langchain-text-splitters pypdf python-dotenv
Se accederá al modelo a través de OpenRouter, por lo que cree un archivo .env en la raíz del proyecto que contenga la clave de API.
OPENROUTER_API_KEY=your_api_key_here
Agregue .env a .gitignore de inmediato. Una clave que llegue al control de versiones una vez debe considerarse filtrada.
Punto de entrada de la aplicación
main.py permanece muy simple. Crea la aplicación y registra el enrutador de chat; nada relacionado con el modelo o el agente debe estar aquí.
from fastapi import FastAPI
from routers.chat import chat_router
app = FastAPI(
title="AI Agent Demo",
)
app.include_router(chat_router)
Un primer endpoint de chat
En routers/chat.py, defina un enrutador bajo el prefijo /chat con una única ruta POST. Por ahora, solo reproduce el mensaje recibido, lo cual es suficiente para confirmar que todo funciona antes de involucrar a la IA.
from fastapi import APIRouter
chat_router = APIRouter(
prefix="/chat",
tags=["Chat"],
)
@chat_router.post("")
def ask_ai(
message: str,
):
return {
"message": message,
}
Tenga en cuenta que message: str en una ruta POST sin un modelo de cuerpo hace que FastAPI lo lea desde la cadena de consulta. Esto es conveniente para probar cosas en Swagger UI, pero para un cliente real normalmente se aceptaría un cuerpo JSON definido con un modelo Pydantic, ya que las cadenas de consulta terminan en los registros de acceso y tienen límites prácticos de longitud.
Una dependencia de autenticación simulada
Una aplicación en producción verificaría una cookie de sesión o un JWT y cargaría al usuario desde la base de datos. Aquí, un encabezado personalizado sustituye todo eso. Cree auth/dependencies.py con una pequeña dataclass MockUser y una función get_current_user que lee el encabezado X-Demo-User y rechaza la solicitud con un 401 cuando está ausente.
from dataclasses import dataclass
from fastapi import Header, HTTPException
@dataclass
class MockUser:
id: str
name: str
def get_current_user(
x_demo_user: str | None = Header(default=None),
) -> MockUser:
if x_demo_user is None:
raise HTTPException(
status_code=401,
detail="Missing X-Demo-User header",
)
return MockUser(
id=x_demo_user,
name=x_demo_user,
)
La inyección de dependencias de FastAPI ahora entrega al usuario al endpoint. Basta con declarar un parámetro con Depends(get_current_user).
from fastapi import APIRouter, Depends
from auth.dependencies import (
MockUser,
get_current_user,
)
chat_router = APIRouter(
prefix="/chat",
tags=["Chat"],
)
@chat_router.post("")
def ask_ai(
message: str,
current_user: MockUser = Depends(
get_current_user,
),
):
return {
"user": current_user.name,
"message": message,
}
Un cliente se identifica enviando una cabecera como esta:
X-Demo-User: user-123
En cada solicitud, FastAPI ejecuta primero get_current_user() y pasa el objeto MockUser resultante a ask_ai. El punto clave del diseño es la división de tareas: FastAPI se encarga de la autenticación, mientras que la capa de IA simplemente recibe un objeto de usuario en el que puede confiar. Más tarde, ese objeto de usuario permite a las herramientas devolver datos pertenecientes a la persona correcta. Cambiar el modelo simulado por una autenticación real solo modifica esta única dependencia.
Paso 2: Conectar un modelo a través de OpenRouter
Con un endpoint funcional y un llamante conocido, el servicio necesita un modelo. OpenRouter expone una API compatible con OpenAI, por lo que la clase ChatOpenAI de LangChain funciona con ella una vez que se indica base_url como OpenRouter y se pasa la clave de OpenRouter.
Ponga esto en llm/provider.py. Este archivo carga .env, falla rápidamente con un error claro si la clave está ausente, y envuelve la clave en SecretStr de Pydantic para que no se imprima accidentalmente en los registros o representaciones.
import os
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_openai import ChatOpenAI
load_dotenv()
api_key = os.getenv(
"OPENROUTER_API_KEY",
)
if not api_key:
raise RuntimeError(
"OPENROUTER_API_KEY environment variable is not set."
)
model = ChatOpenAI(
model="YOUR_MODEL",
api_key=SecretStr(api_key),
base_url="https://openrouter.ai/api/v1",
)
Dado que la clave proviene del entorno, nunca aparece en el código fuente. En esta etapa ya se podrían enviar solicitudes al modelo y obtener respuestas, pero eso sería una simple llamada a un LLM. El objetivo es tener un agente que pueda decidir por sí mismo cuándo necesita una herramienta.
Elegir un modelo
El argumento model es simplemente un identificador de modelo de OpenRouter, por lo que puedes cambiar de modelos sin tocar el resto del código. Al comparar las opciones, verifica:
- la compatibilidad con llamadas a herramientas, de la que depende el agente;
- la compatibilidad con transmisión en tiempo real;
- el tamaño de la ventana de contexto;
- los límites de velocidad;
- si existe una versión gratuita.
OpenRouter ofrece algunos modelos de forma gratuita, lo cual es útil mientras haces pruebas. El catálogo cambia con regularidad, así que consulta la lista actual y filtra los modelos gratuitos en lugar de confiar en una recomendación fija. Lo que elijas se pasa directamente al constructor:
model = ChatOpenAI(
model="YOUR_MODEL_ID",
api_key=SecretStr(api_key),
base_url="https://openrouter.ai/api/v1",
)
Por ejemplo, si el catálogo muestra un identificador como el siguiente (un ejemplo tomado en el momento de redactar este texto; es posible que ya no esté disponible), deberías pasar esa cadena exacta como model:
google/gemma-4-26b-a4b-it:free
Tenga en cuenta que los modelos gratuitos comparten capacidad. Suelen verse limitados en su tasa de uso o dejar de estar disponibles, a menudo en el peor momento durante una demostración. Cambiar a otro modelo o adjuntar su propia clave de proveedor a través de OpenRouter suele solucionar el problema. Para entornos de producción, elija en función de la fiabilidad, las capacidades, la latencia y el costo, no solo del precio.
Paso 3: De un modelo a un agente
Una llamada directa al modelo implica un solo paso: el texto del usuario entra y sale una respuesta completada. Un agente introduce un bucle de toma de decisiones. El modelo examina la solicitud, decide si puede responder de inmediato o necesita algo primero, llama a una herramienta si es necesario, lee el resultado y solo entonces genera la respuesta final. En resumen:
- Llamada simple: usuario, luego LLM, luego respuesta;
El módulo del agente
En llm/agent.py, create_agent de LangChain construye el agente a partir de un modelo y una prompt del sistema. Bajo el capó, genera un grafo LangGraph que ejecuta el bucle de modelo y herramientas por usted.
from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
agent = create_agent(
model=model,
system_prompt=SYSTEM_PROMPT,
)
Este agente aún no tiene herramientas, por lo que se comporta muy similar al modelo puro. Primero, dele instrucciones.
La prompt del sistema
llm/prompts.py contiene una prompt breve que indica al modelo para qué sirve, prohíbe inventar datos y especifica qué tipo de pregunta se corresponde con qué tipo de búsqueda.
SYSTEM_PROMPT = """
You are an AI assistant for our demo application.
You help users understand the application and navigate the system.
Never invent data.
When information about the demo system
is required, use the available application tools.
When answering questions about the application,
use the documentation search tool.
Always answer in clear, conversational language.
""".strip()
La prompt establece dos fuentes de verdad:
- datos de la aplicación (lo que hay en la cuenta del usuario) proviene de las herramientas de la aplicación;
- Conocimiento de la aplicación (cómo funciona el producto) proviene de la búsqueda en la documentación.
Una aclaración antes de continuar. Las herramientas y RAG se presentan por separado más abajo porque así resulta más fácil entender cada una, pero en el agente final la búsqueda en la documentación es en sí misma una herramienta. No existe un segundo mecanismo: el agente ve una lista de funciones llamables, y buscar en el manual es una de ellas.
Paso 4: La primera herramienta
Una herramienta es una función que el agente tiene permiso para llamar. Esto es lo que permite que la arquitectura escale: en lugar de incluir toda la información de la aplicación en el prompt, se exponen operaciones específicas y el modelo puede solicitarlas solo cuando sea necesario para responder a una pregunta.
Para la demostración, llm/tools/demo_tool.py define una herramienta que devuelve un bloque fijo de información del proyecto.
from langchain.tools import tool
@tool
def get_my_demo_data() -> str:
"""
Return information about the demonstration data.
This is just for demo data. But in production, make a more detailed instruction.
"""
return """
Project: Aperture Analytics Dashboard
Owner: Jordan Lee
Status: In Progress
Team size: 6
Budget: $84,000
Deadline: 2026-11-15
Description: An internal dashboard for visualizing customer usage
metrics, built with FastAPI and React, integrating with the
company's data warehouse.
""".strip()
Hay dos detalles importantes aquí. El decorador @tool convierte una función ordinaria de Python en una herramienta de LangChain, tomando su nombre y el esquema de argumentos de la firma de la función. Además, la documentación se convierte en la descripción de la herramienta, que es lo que el modelo lee al decidir si debe llamarla. En un sistema real, esa descripción merece una atención especial: se debe indicar con precisión qué devuelve la herramienta, cuándo es apropiado hacerlo y cuándo no. Una descripción vaga es una de las causas más comunes por las que un agente llama a la herramienta incorrecta o no llama a ninguna en absoluto.
Registre la herramienta pasándola a create_agent:
from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import get_my_demo_data
agent = create_agent(
model=model,
tools=[
get_my_demo_data,
],
system_prompt=SYSTEM_PROMPT,
)
Tu código nunca decide cuándo se ejecuta la función. Si un usuario pregunta “¿Qué datos de demostración tengo?”, el modelo reconoce que necesita información específica de la cuenta y llama a get_my_demo_data(). Si el usuario pregunta “¿Qué es una demostración?”, no se necesita ninguna búsqueda y el modelo responde directamente. Esa decisión se toma en cada turno, dentro del bucle del agente.
Paso 5: Creación de la pipeline RAG para el manual
El agente ahora puede obtener datos de la aplicación, pero aún no sabe nada sobre cómo funciona el producto. Pegar un manual de 20 páginas en la instrucción del sistema desperdiciaría tokens en cada solicitud y resultaría difícil de mantener actualizado. RAG evita ambos problemas.
Vale la pena ser preciso sobre qué es y qué no es RAG. Nada se entrena ni se ajusta a partir de la documentación. En el momento de la consulta, el sistema busca en el manual los pasajes más relevantes para la pregunta y les proporciona esos pasajes al modelo como contexto, y el modelo responde a partir de ellos.
El proceso consta de dos fases:
- Ingestión, que se ejecuta por separado de la aplicación web: carga el PDF, lo divide en fragmentos y almacena esos fragmentos junto con sus embeddings en ChromaDB.
- Recuperación, que se lleva a cabo dentro de una solicitud: toma la pregunta, busca en ChromaDB, recopila los mejores fragmentos y los pasa al modelo.
Si desea una visión más amplia de los conceptos, la reseña del blog sobre la recuperación de conocimiento actualizado bajo demanda los aborda con mayor profundidad; aquí nos centramos en la implementación.
Cargando el PDF
llm/rag/ingestion/loader.py utiliza PyPDFLoader de LangChain, el cual convierte cada página del PDF en un Document.
from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader
PDF_PATH = Path("docs/manual.pdf")
def load_manual():
loader = PyPDFLoader(
str(PDF_PATH),
)
documents = loader.load()
return documents
Document contiene dos elementos: el texto extraído en page_content, y un diccionario metadata que describe de dónde proviene ese texto. Es el metadata lo que posteriormente permite que una respuesta cite una página. Conceptualmente, cada página cargada se ve así:
Document
├── page_content
│ └── "To create a new demo data..."
│
└── metadata
├── source: docs/manual.pdf
└── page: 12
PyPDFLoader suele registrar tanto un índice de page basado en cero como una page_label legible para el usuario. El generador de contexto a continuación utiliza page_label, que coincide con los números de página que ven los lectores en el PDF.
Dividir páginas en bloques
Buscar páginas completas, y mucho menos todo el documento como un único bloque, produce resultados poco precisos. llm/rag/ingestion/chunker.py divide los documentos con RecursiveCharacterTextSplitter.
from langchain_text_splitters import (
RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document
def chunk_documents(
documents: list[Document],
) -> list[Document]:
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=150,
separators=[
"\n\n",
"\n",
". ",
" ",
"",
],
)
return splitter.split_documents(
documents,
)
El divisor tiene como objetivo crear fragmentos de aproximadamente 1,000 caracteres con una superposición de 150 caracteres. Se prueban las listas de separadores en orden: prefiere dividir en los límites de párrafos, luego en saltos de línea, después al final de las oraciones, seguido de los espacios, y solo como último recurso en medio de una palabra. La superposición existe porque un mismo hecho puede abarcar varios puntos de división; repetir un poco de texto a cada lado reduce la probabilidad de que la oración relevante quede dividida por la mitad.
Estos números son puntos de partida, no reglas. El tamaño adecuado depende de cómo estén redactados sus documentos y del grado de precisión necesario en la recuperación, por lo que trátelos como valores a ajustar según las necesidades reales. El artículo del blog sobre el fragmentado que preserva las pruebas profundiza más en este equilibrio.
Una colección Chroma persistente
llm/rag/ingestion/chroma.py abre un PersistentClient que almacena los datos en el disco y devuelve la colección del manual, creándola la primera vez que se utiliza.
import chromadb
from llm.rag.config import (
CHROMA_PATH,
MANUAL_COLLECTION_NAME,
)
def get_chroma_client():
return chromadb.PersistentClient(
path=CHROMA_PATH,
)
def get_manual_collection():
client = get_chroma_client()
return client.get_or_create_collection(
name=MANUAL_COLLECTION_NAME,
)
Las rutas y nombres provienen de llm/rag/config.py, el cual lee las variables de entorno y recurre a valores predeterminados razonables si es necesario:
import os
CHROMA_PATH = os.getenv(
"CHROMA_PATH",
"./chroma_data",
)
MANUAL_PATH = os.getenv(
"MANUAL_PATH",
"docs/manual.pdf",
)
MANUAL_COLLECTION_NAME = os.getenv(
"MANUAL_COLLECTION_NAME",
"manual",
)
Agregue las entradas correspondientes a .env:
CHROMA_PATH=./chroma_data
MANUAL_PATH=docs/manual.pdf
MANUAL_COLLECTION_NAME=manual
No se ha configurado ningún modelo de embedding en ninguna parte, y eso es intencional para la demostración. Cuando se crea una colección sin una función de embedding explícita, Chroma utiliza su valor predeterminado integrado: cada vez que se añaden documentos, Chroma calcula sus embeddings por sí misma y los almacena junto con el texto y los metadatos. El modelo predeterminado se ejecuta localmente y se carga la primera vez que se utiliza, por lo que la primera operación de indexación podría detenerse mientras se descarga.
El resultado es un almacén de vectores local y persistente en el directorio chroma_data/. Dado que se deriva íntegramente del PDF, agréguelo a .gitignore junto con .env.
La tarea de indexación
llm/rag/ingestion/indexer.py une los pasos de ingestión.
from pathlib import Path
from llm.rag.config import MANUAL_PATH
from llm.rag.ingestion.loader import load_manual
from llm.rag.ingestion.chunker import chunk_documents
from llm.rag.ingestion.chroma import get_manual_collection
def index_manual():
collection = get_manual_collection()
if collection.count() > 0:
print(
f"Manual already indexed "
f"({collection.count()} chunks)."
)
return
manual_path = Path(
MANUAL_PATH,
)
if not manual_path.exists():
raise FileNotFoundError(
f"Manual not found: {manual_path}"
)
documents = load_manual()
print(
f"Loaded {len(documents)} pages."
)
chunks = chunk_documents(
documents,
)
print(
f"Created {len(chunks)} chunks."
)
collection.add(
ids=[
f"manual-chunk-{i}"
for i in range(len(chunks))
],
documents=[
chunk.page_content
for chunk in chunks
],
metadatas=[
chunk.metadata
for chunk in chunks
],
)
print(
f"Stored {len(chunks)} chunks."
)
Analicemos qué hace. Abre la colección y devuelve un resultado temprano si ya contiene fragmentos, lo que hace que las ejecuciones repetidas no causen problemas. Verifica si el PDF existe y, de lo contrario, genera un error claro. Luego carga las páginas, las divide en fragmentos y agrega todo a Chroma con una sola llamada, utilizando identificadores estables (manual-chunk-0, manual-chunk-1, etc.), los textos de los fragmentos y su metadatos. Los mensajes de progreso indican cuántas páginas y fragmentos se han procesado.
Un problema surge de ese retorno prematuro: si editas el manual y ejecutas la script nuevamente, no ocurre nada, porque la colección no está vacía. Para aplicar los cambios, debes eliminar la colección (o el directorio chroma_data/) antes de volver a indexarla, o reemplazar la lógica de protección por una que realice inserciones o reconstrucciones de forma intencionada.
La lista no muestra scripts/index_manual.py; solo necesita importar index_manual y llamarlo. Ejútalo una vez como módulo desde la raíz del proyecto:
python -m scripts.index_manual
Esa única ejecución lee el PDF, lo divide en fragmentos, incrusta esos fragmentos y los almacena. La salida de la terminal indica los conteos. En la demostración simplificada, el manual es un PDF de una página que contiene una sola regla: “Los datos de la demostración solo pueden entregarse a usuarios administradores”, lo cual es suficiente para mostrar si la recuperación funciona. A partir de ese momento, iniciar FastAPI no afecta en absoluto al PDF, ya que los vectores ya se han persistido.
Consultar la colección
El indexado por sí solo no le proporciona nada al agente; necesita una forma de buscar. llm/rag/retrieval/retriever.py encapsula la API de consultas de Chroma.
from dataclasses import dataclass
from typing import Any
from llm.rag.ingestion.chroma import (
get_manual_collection,
)
@dataclass
class RetrievedChunk:
content: str
metadata: dict[str, Any]
distance: float
def retrieve_manual(
query: str,
n_results: int = 5,
) -> list[RetrievedChunk]:
collection = get_manual_collection()
results = collection.query(
query_texts=[query],
n_results=n_results,
include=[
"documents",
"metadatas",
"distances",
],
)
documents = results["documents"] or []
metadatas = results["metadatas"] or []
distances = results["distances"] or []
retrieved_chunks = []
for document, metadata, distance in zip(
documents[0],
metadatas[0],
distances[0],
):
retrieved_chunks.append(
RetrievedChunk(
content=document,
metadata=dict(metadata)
if metadata else {},
distance=distance,
)
)
return retrieved_chunks
La función envía la pregunta como query_texts, solicita hasta cinco resultados y pide los documentos, sus metadatos y sus distancias. Chroma devuelve una lista por consulta, por lo que el código lee el índice [0] de cada campo; las soluciones alternativas or [] protegen contra la falta de campos. Cada resultado se empaqueta como una clase RetrievedChunk para que el resto del código no dependa de la estructura de respuesta de Chroma.
Dado que la consulta se integra con el mismo modelo que los fragmentos almacenados, la correspondencia es semántica. Una pregunta como “¿Cómo agrego nuevos datos de demostración?” encuentra pasajes sobre la creación o concesión de datos de demostración, incluso si nunca se utilizan las palabras “agregar nuevos”. La distancia indica cuán cercano es cada resultado; cuanto más baja, mayor es la similitud. Ese valor resulta útil posteriormente si se desea descartar las coincidencias débiles en lugar de pasar siempre cinco fragmentos al modelo.
Con esto en marcha, funciona la parte de recuperación de RAG. Lo que queda es pasar los resultados al modelo.
Convierte fragmentos en contexto
llm/rag/context.py formatea los resultados en una sola cadena que el modelo puede leer.
from llm.rag.retrieval.retriever import (
RetrievedChunk,
)
def build_context(
chunks: list[RetrievedChunk],
) -> str:
context_parts = []
for chunk in chunks:
page = chunk.metadata.get(
"page_label",
)
context_parts.append(
f"Source: User Guide, page {page}\n"
f"{chunk.content}"
)
return "\n\n---\n\n".join(
context_parts,
)
Cada fragmento está precedido por una línea de código que indica el nombre de la guía del usuario y su etiqueta de página, y los fragmentos están separados por un divisor. La línea de código es lo que permite al modelo indicar de dónde proviene una respuesta, y también brinda a los usuarios una forma de verificarlo.
Paso 6: Exponer el manual como una herramienta
El proceso está completo: las páginas se cargan y se dividen en fragmentos, estos se almacenan en ChromaDB, y los relevantes pueden ser encontrados y formateados. Sin embargo, el agente no tiene idea de que todo esto existe. Aquí es donde la nota arquitectónica anterior resulta útil: la búsqueda de documentación se convierte en simplemente otra herramienta.
llm/tools/manual_tool.py define search_user_manual, que recibe una consulta, obtiene cinco fragmentos de texto y los devuelve como contexto formateado. Si no se encuentra nada, devuelve un mensaje claro indicando que el manual no aborda esa pregunta, lo cual permite al modelo transmitir información honesta en lugar de una cadena vacía.
from langchain.tools import tool
from llm.rag.context import build_context
from llm.rag.retrieval.retriever import retrieve_manual
@tool
def search_user_manual(
query: str,
) -> str:
"""
Search the application user manual.
Use this tool when the user asks about application
behavior, instructions, rules, limitations, or
how something works.
"""
chunks = retrieve_manual(
query=query,
n_results=5,
)
if not chunks:
return (
"The manual does not contain enough "
"information to answer this question."
)
return build_context(
chunks,
)
Al igual que antes, la documentación es la forma en que la herramienta se presenta al modelo. Indica que debe utilizarse esta herramienta para preguntas sobre comportamiento, instrucciones, reglas, limitaciones y funcionamiento de las cosas, lo cual coincide con el prompt del sistema.
Ahora registre ambas herramientas en el agente:
from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import (
get_my_solar_system,
)
from llm.tools.manual_tool import (
search_user_manual,
)
agent = create_agent(
model=model,
tools=[
get_my_solar_system,
search_user_manual,
],
system_prompt=SYSTEM_PROMPT,
)
Fíjese en la importación que aparece allí: hace referencia a get_my_solar_system, un nombre de la aplicación completa, mientras que el módulo de la herramienta de demostración define get_my_demo_data. Utilice get_my_demo_data tanto en la importación como en la lista de tools, de lo contrario el módulo no podrá importarse.
Por qué un diseño basado en herramientas sigue siendo viable a medida que la aplicación crece
El agente nunca necesita una instrucción enorme que describa todo lo que sabe la aplicación. En su lugar, cuenta con capacidades específicas y controladas. Agregar una función al asistente implica escribir una nueva herramienta y registrarla; la capa HTTP no cambia. La demostración mantiene exactamente una herramienta de datos y una herramienta de documentación para que el patrón sea fácil de seguir, pero la misma estructura permite contar con un conjunto mucho más amplio de herramientas en la aplicación completa.
Paso 7: Pasar al agente al usuario autenticado
La herramienta de demostración sigue devolviendo datos codificados de forma fija. Una herramienta real debe saber quién está haciendo la solicitud, y la aplicación ya lo sabe: FastAPI identificó al usuario en la dependencia de autenticación. Lo que falta es pasar ese usuario al proceso del agente. LangChain denomina a esto contexto de ejecución.
Defina la estructura del contexto en llm/context.py:
from dataclasses import dataclass
from auth.dependencies import MockUser
@dataclass
class AgentContext:
user: MockUser
Este objeto se proporciona cuando se invoca al agente, y esta distinción es importante. El usuario representa información relacionada con una solicitud específica; pertenece a la petición HTTP actual, no a la conversación en general, y nunca debe almacenarse como un mensaje que el modelo pueda leer o reescribir. Al mantenerlo fuera del historial de mensajes, también se evita que un prompt haga que el agente actúe como otro usuario. En la aplicación completa, el mismo objeto de contexto también contiene elementos como la sesión de base de datos y el ID del proyecto que se está editando.
El código de demostración se detiene en la definición de la clase, por lo que quedan dos conexiones por establecer; vale la pena consultar la documentación actual de LangChain para conocer la API exacta. Primero, declare el esquema al crear el agente, generalmente con un argumento context_schema=AgentContext en la función create_agent. Segundo, haga que las herramientas lo lean: en LangChain 1.x una herramienta puede aceptar un parámetro de ejecución (por ejemplo, anotado como ToolRuntime[AgentContext]) y leer al usuario desde su atributo context, el cual está oculto de la vista del modelo sobre los argumentos de la herramienta. Allí es donde una verdadera función get_my_demo_data buscaría registros para user.id.
Paso 8: Una capa de orquestación que transmite datos en tiempo real
En lugar de llamar al agente desde el interior del router, coloque la interacción en llm/orchestrator.py. De esta manera, el router se mantendrá enfocado en HTTP, y el orquestador se encargará de que un mensaje se convierta en la ejecución de un agente.
from collections.abc import Iterator
from langchain_core.messages import (
AIMessage,
AIMessageChunk,
BaseMessage,
ToolMessage,
)
from langchain_core.runnables import RunnableConfig
from llm.agent import agent
from llm.context import AgentContext
def chat_stream(
user_message: str,
user,
) -> Iterator[str]:
config: RunnableConfig = {
"configurable": {
"thread_id": f"user:{user.id}",
}
}
context = AgentContext(
user=user,
)
for chunk, metadata in agent.stream(
{
"messages": [
{
"role": "user",
"content": user_message,
}
]
},
config=config,
context=context,
stream_mode="messages",
):
if not isinstance(
chunk,
BaseMessage,
):
continue
if isinstance(
chunk,
ToolMessage,
):
continue
if not isinstance(
chunk,
(
AIMessage,
AIMessageChunk,
),
):
continue
if isinstance(
chunk.content,
str,
):
yield chunk.content
Hay mucho contenido en esta función, así que trátela por partes.
El ID del hilo selecciona la conversación
El primer bloque crea la configuración de ejecución:
config = {
"configurable": {
"thread_id": f"user:{user.id}",
}
}
Un checkpointer de LangGraph almacena el estado de la conversación mediante una clave thread_id. Cada ejecución con el mismo ID de hilo continúa la misma conversación, y sus mensajes pueden ser leídos posteriormente. Aquí el ID de hilo se deriva del ID del usuario, lo que significa que cada usuario tiene exactamente una conversación. Eso está bien para una demostración; una aplicación real crearía IDs de conversación adecuados, permitiría varios por usuario y verificaría en cada solicitud que quien llama es el propietario del hilo al que se accede.
La descripción asume que existe un checkpointer, pero ninguno de los fragmentos del agente lo cumple. Sin él, thread_id no tiene efecto y nada se recuerda entre las solicitudes. Cree un único InMemorySaver en llm/agent.py y páselo a create_agent a través de su argumento checkpointer, para que tanto el agente como las funciones de historial puedan importar la misma instancia.
El contexto en tiempo de ejecución viaja con la ejecución
A continuación, el orquestador envuelve al usuario en el objeto de contexto:
context = AgentContext(
user=user,
)
Ese objeto se pasa como context= a agent.stream(). Este es el camino por el cual la identidad establecida en FastAPI llega al agente y, a través de él, a las herramientas.
Filtrado del flujo
agent.stream() se llama con stream_mode="messages", lo que genera pares de un fragmento de mensaje y metadatos a medida que el modelo produce tokens. No todo lo que hay en ese flujo debe llegar al usuario. El bucle omite cualquier elemento que no sea un mensaje de LangChain, ignora los objetos ToolMessage (salida bruta de herramientas como texto manual recuperado), mantiene únicamente los mensajes y fragmentos de mensaje generados por la IA, y muestra su contenido cuando se trata de una cadena de texto simple. El contenido que algunos proveedores envían como lista de partes es descartado silenciosamente por esa última verificación; por lo tanto, si cambia de modelo y ve respuestas vacías, allí debe buscar la causa.
Paso 9: Devolver una respuesta en streaming
Hacer de chat_stream() un generador fue una decisión intencionada. Esperar a obtener la respuesta completa antes de enviar un byte hace que el usuario permanezca mirando un indicador de carga, y las respuestas de los LLM pueden tardar varios segundos. Al transmitir en streaming, las primeras palabras se muestran casi de inmediato, lo que hace que el asistente parezca mucho más reactivo.
StreamingResponse de FastAPI acepta directamente un generador. Actualice routers/chat.py:
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from auth.dependencies import (
MockUser,
get_current_user,
)
from llm.orchestrator import chat_stream
chat_router = APIRouter(
prefix="/chat",
tags=["Chat"],
)
@chat_router.post("")
def ask_ai(
message: str,
current_user: MockUser = Depends(
get_current_user,
),
):
return StreamingResponse(
chat_stream(
user_message=message,
user=current_user,
),
media_type="text/plain",
)
La respuesta se envía en formato text/plain, y cada fragmento generado se escribe en la conexión tan pronto como se produce. Dado que chat_stream es un generador regular (sincrónico), Starlette lo itera en un hilo de trabajo, por lo que no bloquea el bucle de eventos. Si más adelante necesita eventos estructurados en el cliente (por ejemplo, para mostrar “buscando en el manual...” mientras se ejecuta una herramienta), los Server-Sent Events constituyen el siguiente paso lógico.
La ruta completa de la solicitud ahora es la siguiente: el cliente envía datos a /chat, FastAPI autentica al solicitante, chat_stream() inicia la ejecución del agente, el modelo decide si debe llamar a una herramienta, cualquier herramienta se ejecuta y devuelve su resultado, el modelo escribe la respuesta, y los tokens vuelven al cliente.
La característica clave es que FastAPI nunca ejecuta el modelo por sí mismo. El enrutador maneja las comunicaciones HTTP, el orquestador controla al agente, el agente decide qué información necesita y las herramientas se encargan de obtenerla. Cada capa puede modificarse sin afectar a las demás.
Paso 10: Leer el historial de conversaciones
Dado que el agente tiene puntos de verificación, su estado se almacena después de cada turno. Esto permite mostrar a los usuarios que regresan sus conversaciones anteriores. Dos funciones auxiliares se encargan de esta tarea.
Leer el punto de verificación en bruto
La primera función carga el último punto de control para un hilo y devuelve el canal messages, o una lista vacía si el hilo nunca se ha utilizado:
def get_conversation_messages(
thread_id: str,
) -> list[BaseMessage]:
config: RunnableConfig = {
"configurable": {
"thread_id": thread_id,
}
}
checkpoint = checkpointer.get(
config,
)
if checkpoint is None:
return []
return checkpoint[
"channel_values"
].get(
"messages",
[],
)
Si copias este código, corrige el sangrado de la asignación config: debe estar sangrada dentro del cuerpo de la función, de lo contrario Python genera un error. La función también necesita que se importen BaseMessage, RunnableConfig y la instancia compartida checkpointer.
Lo que se devuelve es el estado bruto del agente, y eso incluye más información de la que recuerda el usuario en el chat. Cuando el agente llama a una herramienta, LangGraph registra un mensaje de IA que contiene la llamada a la herramienta y un mensaje separado con el resultado. Estos son detalles de implementación que el frontend no debería tener que interpretar.
Diseño de los mensajes para su visualización
La segunda función crea la vista para el usuario:
def get_conversation_messages_for_display(
thread_id: str,
) -> list[dict[str, str]]:
display = []
for message in get_conversation_messages(
thread_id,
):
if isinstance(
message,
HumanMessage,
):
content = _extract_text_content(
message.content,
)
if content.strip():
display.append(
{
"type": "human",
"content": content,
}
)
continue
if isinstance(
message,
AIMessage,
):
content = _extract_text_content(
message.content,
)
if content.strip():
display.append(
{
"type": "ai",
"content": content,
}
)
return display
Solo mantiene objetos HumanMessage y AIMessage, extrae su texto, descarta aquellos que están vacíos y devuelve diccionarios simples con un type y un content. Los mensajes de herramienta nunca aparecen porque no pertenecen a ninguno de los dos tipos aceptados. También se descartan los mensajes de IA vacíos, lo cual es importante ya que un mensaje de IA que solo solicita una llamada a herramienta generalmente no contiene texto.
Esta función depende de una ayuda, _extract_text_content, que no se muestra aquí. Su función es devolver el contenido sin cambios cuando se trata de una cadena y, cuando se trata de una lista de partes de contenido, unir esas partes en un texto único. También es necesario importar HumanMessage y AIMessage.
El endpoint de historial
Exponga la vista de visualización a través de una ruta GET en routers/chat.py. Obtiene el mismo ID de hilo que utiliza el endpoint de chat y lo devuelve junto con los mensajes.
@chat_router.get("/current")
def get_current_conversation(
current_user: MockUser = Depends(
get_current_user,
),
):
thread_id = (
f"user:{current_user.id}"
)
messages = (
get_conversation_messages_for_display(
thread_id,
)
)
return {
"thread_id": thread_id,
"messages": messages,
}
Una interfaz de chat puede llamar a esta función al abrirse la página y mostrar la conversación existente antes de que el usuario escriba algo:
GET /chat/current
La respuesta se ve así:
{
"thread_id": "user:user-123",
"messages": [
{
"type": "human",
"content": "How much demo data do I have?"
},
{
"type": "ai",
"content": "You currently have 18 demo data."
}
]
}
¿Por qué no devolver el estado en bruto?
El estado interno del agente y la conversación que ve el usuario son cosas diferentes. A medida que se añaden funcionalidades, el estado acumula llamadas a herramientas, resultados de esas herramientas, pasos intermedios, metadatos del modelo y otros registros. Devolver todo ello vincularía la interfaz frontal con los componentes internos del agente y podría hacer que se filtre información de las herramientas que no se pretendía mostrar. El backend debe definir cuál es el historial de conversación público y devolver únicamente ese contenido.
Cómo fluye una única solicitud de principio a fin
Con todas las piezas en su lugar, vale la pena aclarar un aspecto importante: el modelo nunca accede a su base de datos ni a sus archivos PDF. Solo puede solicitar que se ejecute una herramienta. Dicha herramienta, que funciona como código Python ordinario con controles de acceso normales, realiza la operación y devuelve texto, el cual el modelo utiliza para formular su respuesta. Es este límite lo que garantiza que el asistente sea seguro para integrarse en una aplicación con datos reales.
Probándolo en Swagger UI
FastAPI genera documentación interactiva de forma automática, por lo que no se necesita un cliente separado durante las pruebas. Inicie el servidor:
python -m uvicorn main:app --reload
Luego abra la documentación API interactiva disponible en /docs, establezca el encabezado X-Demo-User e intente tres interacciones:
- Una pregunta sobre datos, como preguntar qué datos de demostración se tienen. El agente debe decidir que necesita la herramienta de demostración, llamarla y responder con los detalles del proyecto devueltos. En la aplicación completa, el mismo tipo de herramienta consulta los registros reales del usuario.
- Una pregunta sobre documentación, como preguntar quién puede recibir datos de demostración. El agente debe realizar una búsqueda manual, recuperar la única regla indexada y responder que solo los usuarios administradores pueden hacerlo.
- El endpoint de historial,
GET /chat/current, que debe devolver los turnos del humano y de la IA correspondientes a las dos preguntas anteriores, sin ningún mensaje de herramienta.
Si las dos primeras generan respuestas basadas en la salida de la herramienta y la tercera muestra una transcripción limpia, todas las capas están funcionando correctamente.
Antes de llevarlo a producción
La demostración simplifica intencionadamente varios componentes. Estos son los que hay que revisar antes de que los usuarios reales dependan del servicio.
Estatuto duradero de las conversaciones
InMemorySaver es perfecto para el desarrollo, pero todo lo que almacena desaparece cuando se reinicia el proceso, y no puede compartirse entre varias instancias de API detrás de un balanceador de carga. Utilice un punto de control respaldado por una base de datos u otro almacenamiento duradero para que las conversaciones sobrevivan a los despliegues y todas las instancias vean el mismo estado. Para saber qué almacena realmente InMemorySaver y cómo lo hace, consulte la guía del blog sobre cómo InMemorySaver organiza los puntos de control, las escrituras y los blobs.
Despliegue real de un almacén de vectores
Un directorio local de Chroma es adecuado para demostrar el flujo de trabajo, pero no constituye una infraestructura de producción. Ejecute Chroma como un servicio persistente o pásese a una base de datos vectorial gestionada que se adapte a su stack. Lo que sea que elija debe ser persistente, tener copias de seguridad y ser accesible desde cada instancia de la aplicación.
El indexado queda fuera de la API
La demostración ya toma una decisión importante correctamente: el indexado es un script separado que se ejecuta por sí solo, mientras que la API solo realiza búsquedas.
python -m scripts.index_manual
El servidor nunca carga el PDF, ni lo divide en partes ni calcula embeddings al iniciar. El indexado es un proceso de ingestión fuera de línea; la recuperación forma parte del procesamiento de una solicitud. Al mantenerlos separados, la API no necesita detectar cambios en la documentación ni reconstruir nada.
En producción, se da el siguiente paso y se ejecuta el mismo código de indexación como un trabajo dedicado de ingestión, activado desde un pipeline de despliegue, según un horario establecido, o por un proceso siempre que se suba nueva documentación. La arquitectura permanece igual; el trabajo simplemente se vuelve automático, repetible e independiente para su despliegue. Las responsabilidades se dividen entonces de manera clara:
- Trabajo de ingestión: cargar documentos, dividirlos en fragmentos, incrustarlos y actualizar el almacén vectorial.
- Servicio FastAPI: aceptar consultas, recuperar los fragmentos relevantes y generar respuestas.
- Almacén vectorial: conservar las representaciones indexadas que se utilizan al realizar consultas.
Recuerde la protección de retorno anticipado en el indexador cuando lo automatice; un trabajo que omite silenciosamente la reindexación es peor que no tener ningún trabajo.
Un modelo de incrustación explícito
Depender de la función de incrustación predeterminada de Chroma permite que la demostración no requiera configuraciones adicionales, pero un sistema en producción debe elegir y configurar explícitamente su modelo de incrustación. Esto hace que los resultados sean reproducibles y le brinda control sobre la calidad, el costo, la latencia y el lugar donde se calculan las incrustaciones. Hay una regla innegociable: se debe utilizar el mismo modelo de incrustación tanto para el indexado como para las consultas. Cambiarlo implica volver a indexar todo.
Observabilidad y manejo de errores
Una vez que un agente está en funcionamiento, saber qué hizo es tan importante como hacer que funcione. Una sola solicitud puede involucrar varias llamadas al modelo, una o más llamadas a herramientas y un paso de recuperación antes de obtener la respuesta final. Registrar solo esa respuesta final casi no le dice nada cuando algo sale mal. Instrumente todo el flujo:
- Llamadas a herramientas: qué herramientas se ejecutaron, con qué argumentos y cuánto tiempo tardó cada una.
El objetivo es que el agente nunca sea una caja negra. En cualquier ejecución, se debe poder saber qué hizo, qué herramientas utilizó, cuánto tiempo duró cada paso y dónde falló. Las herramientas específicas dependen de tu stack tecnológico; el principio, no.
Puntos clave
- No se necesita una plataforma de IA grande para añadir un asistente útil a una aplicación con muchos dominios. Comienza con el conjunto más pequeño de componentes que resuelva un problema real.
- Considera la búsqueda de documentación como una herramienta más entre otras. De esta manera, el agente cuenta con un método único y uniforme para acceder tanto al conocimiento del producto como a los datos de los usuarios.
- Mantén la identidad en el contexto de ejecución, no en los mensajes. El usuario pertenece a la solicitud, y las herramientas deben leerla desde allí.
- Separa HTTP, orquestación, agente y herramientas. Cada capa debe mantenerse pequeña, y añadir una funcionalidad implica incorporar una herramienta adicional.
Lecturas relacionadas
- Memoria híbrida de agente: fusión de BM25 y búsqueda vectorial con RRF en Python — Aprenda por qué la búsqueda vectorial pura falla como memoria de agente, cómo la fusión de rango recíproco combina BM25 y resultados densos en Python, y cuándo los resúmenes de GraphRAG son útiles.
- Diseñando una memoria de agente de cuatro capas con LangGraph y Amazon Bedrock — Aprenda a dotar a los agentes de LLM de memoria episódica, semántica y procedural funcional en Bedrock y LangGraph, así como a protegerla contra envenenamiento, fugas de PII y filtraciones de datos entre usuarios.
- Dejando que Gemini elija la fuente: FAISS, Tavily y respuestas directas en LangGraph — Cree un pequeño flujo de trabajo con LangGraph en el que Gemini dirige cada pregunta a una base de conocimientos FAISS, a búsquedas en la web con Tavily o a una respuesta directa, mediante un sistema de enrutamiento de salida estructurada.
- Enrutamiento de Preguntas a Través de Herramientas SQL y Búsqueda Web con un Agente Gemini — Cómo un agente de llamadas a herramientas de LangChain en Vertex AI elige entre tres herramientas de texto a SQL de SQLite y la búsqueda web en tiempo real, además de los errores relacionados con datos, dependencias y autenticación que se pueden esperar.
- Transmisión de Agentes LangGraph a React sin Filtrar la Información Interna de las Herramientas — Cómo serializar la salida astream_events de LangGraph para Server-Sent Events, redactar cargas útiles sensibles de llamadas a herramientas en FastAPI y mostrar insignias amigables de herramientas en React.
- Evaluación del comportamiento de los agentes: resultados, trayectorias y retroalimentación en la producción — Cómo juzgar a los sistemas agentes más allá de la respuesta final: asociar LangChain, LangGraph y LangSmith a roles, calificar trayectorias y contexto, e incorporar los fallos en la producción en las evaluaciones.