Agentes de IA para producción con FastAPI, LangGraph y arquitectura limpia
Límites de capas, estado del grafo tipado, servicios verificables y una estructura de despliegue que permite mantener los agentes en buen estado.
Esta guía reconstruye el proceso desde las materias primas hasta un sistema funcional para: Cómo estructuro agentes de IA de producción con FastAPI, LangGraph y arquitectura limpia. El enfoque está en pasos operativos, verificaciones explícitas y código que se puede incorporar directamente a un repositorio sin tener que adivinar su propósito. Para obtener una visión general, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el grafo.
@router.post("/suppliers")
async def create_supplier(request: SupplierRequest):
policies = opensearch.search(
index="supplier-policies",
query=request.description,
)
response = bedrock.converse(
modelId=MODEL_ID,
messages=build_messages(request, policies),
)
supplier = Supplier(
name=request.name,
tax_id=request.tax_id,
)
db.add(supplier)
db.commit()
sqs.send_message(
QueueUrl=SUPPLIER_QUEUE,
MessageBody=serialize(supplier),
)
return {"status": "created"}
FastAPI es una interfaz, no la aplicación
Ya que FastAPI es una interfaz y no la aplicación en sí, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras aplicadas posteriormente. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves de negocio convierte las fusiones posteriores en actualizaciones predecibles, en lugar de situaciones de duplicados.
@router.post("/suppliers")
async def create_supplier(
request: CreateSupplierRequest,
use_case: CreateSupplierUseCase = Depends(
get_create_supplier_use_case
),
):
command = CreateSupplierCommand(
name=request.name,
tax_id=request.tax_id,
country=request.country,
)
result = await use_case.execute(command)
return CreateSupplierResponse.from_result(result)
HTTP Request
↓
FastAPI
↓
Application
↓
Domain
↓
Ports
↓
Adapters
La aplicación se comunica con capacidades, no con tecnologías
Para que la aplicación hable de capacidades y no de tecnologías, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves empresariales convierte las fusiones posteriores en operaciones de inserción o actualización predecibles, en lugar de situaciones de duplicados excesivos.
class CreateSupplierUseCase:
def __init__(
self,
repository: SupplierRepository,
policy_service: SupplierPolicyService,
event_publisher: EventPublisher,
):
self.repository = repository
self.policy_service = policy_service
self.event_publisher = event_publisher
async def execute(
self,
command: CreateSupplierCommand,
) -> Supplier:
existing = await self.repository.find_by_tax_id(
command.tax_id
)
if existing:
raise SupplierAlreadyExists(command.tax_id)
policy = await self.policy_service.evaluate(
country=command.country
)
supplier = Supplier.create(
name=command.name,
tax_id=command.tax_id,
country=command.country,
requires_approval=policy.requires_approval,
)
await self.repository.add(supplier)
await self.event_publisher.publish(
SupplierCreated(supplier.id)
)
return supplier
Los puertos crean el límite
Para Ports, cree los límites, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Trate esta fase como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves empresariales convierte las fusiones posteriores en actualizaciones predecibles en lugar de situaciones de duplicados. Para Ports, cree los límites, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar que los operadores puedan auditar sin tener que leer todo el código.
ph.
from typing import Protocol
class SupplierRepository(Protocol):
async def find_by_tax_id(
self,
tax_id: str,
) -> Supplier | None:
...
async def add(
self,
supplier: Supplier,
) -> None:
...
class PolicyRetriever(Protocol):
async def retrieve(
self,
query: str,
) -> list[PolicyDocument]:
...
class LLMProvider(Protocol):
async def reason(
self,
context: AgentContext,
) -> AgentDecision:
...
class EventPublisher(Protocol):
async def publish(
self,
event: DomainEvent,
) -> None:
...
SupplierRepository
PolicyRetriever
LLMProvider
EventPublisher
PostgreSQL
Amazon OpenSearch
Amazon Bedrock
AWS SQS
Dónde encaja Amazon Bedrock
Para determinar dónde encaja Amazon Bedrock, se deben definir las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Se debe documentar tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Se debe inspeccionar un conjunto de ejemplos después del primer lote. Las pruebas con Cypher detectan con mayor eficiencia los cambios en las etiquetas y la falta de propiedades que las pruebas de chat de extremo a extremo.
class BedrockLLMProvider(LLMProvider):
def __init__(self, client, model_id: str):
self.client = client
self.model_id = model_id
async def reason(
self,
context: AgentContext,
) -> AgentDecision:
response = self.client.converse(
modelId=self.model_id,
messages=build_messages(context),
)
return map_bedrock_response(response)
Application
↓
LLMProvider
↑
BedrockLLMProvider
Dónde encaja RAG
Para determinar dónde encaja RAG, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando una etapa falla, el error debe indicar una única responsabilidad y no un proceso complicado. Inspeccione un grupo de ejemplos después del primer lote. Las pruebas Cypher detectan con mayor facilidad cambios en las etiquetas y propiedades faltantes que las pruebas de chat de extremo a extremo.
FastAPI
↓
OpenSearch
↓
LLM
class OpenSearchPolicyRetriever(PolicyRetriever):
def __init__(
self,
opensearch_client,
embedding_provider,
):
self.client = opensearch_client
self.embedding_provider = embedding_provider
async def retrieve(
self,
query: str,
) -> list[PolicyDocument]:
vector = await self.embedding_provider.embed(query)
results = self.client.search(
index="supplier-policies",
body=build_vector_query(vector),
)
return map_documents(results)
RAG proporciona contexto, pero no toma las decisiones.
Para RAG proporciona contexto; no toma las decisiones. Defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina pruebas de éxito y rechace las completaciones parciales silenciosas. Inspeccione un grupo de ejemplos después del primer lote; las pruebas Cypher detectan con mayor eficiencia los desvíos en las etiquetas y las propiedades faltantes que las pruebas de chat de extremo a extremo. Para RAG proporciona contexto; no toma las decisiones. Defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación; los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un único lugar accesible para los operadores.
Se puede auditar sin necesidad de leer todo el grafo.RAG
↓
LLM
↓
"Looks fine"
↓
Create Supplier
RAG
↓
Relevant Context
↓
Agent Reasoning
↓
Application
↓
Domain Rules
↓
Decision
Dónde encaja LangGraph
Para determinar dónde encaja LangGraph, se deben definir las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Se deben documentar tanto la ruta óptima como la ruta de recuperación. Las intentonas repetidas, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Se deben modelar nodos y relaciones para las preguntas que se vayan a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos y misteriosos.
Understand Request
↓
Retrieve Policies
↓
Evaluate Information
↓
Need More Data?
↙ ↘
Yes No
↓ ↓
Ask User Continue
↓
Approval Required?
↙ ↘
Yes No
↓ ↓
Human Approval Continue
↘ ↙
Request Action
graph = StateGraph(AgentState)
graph.add_node(
"understand_intent",
understand_intent,
)
graph.add_node(
"retrieve_policies",
retrieve_policies,
)
graph.add_node(
"evaluate",
evaluate_request,
)
graph.add_node(
"human_approval",
request_human_approval,
)
graph.add_node(
"request_action",
request_application_action,
)
graph.add_edge(
"understand_intent",
"retrieve_policies",
)
graph.add_edge(
"retrieve_policies",
"evaluate",
)
graph.add_conditional_edges(
"evaluate",
determine_next_step,
{
"approval": "human_approval",
"execute": "request_action",
},
)
LangGraph no debería convertirse en el nuevo monolito
Para que LangGraph no se convierta en el nuevo monolito, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables sobre scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Modele nodos y relaciones para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos y misteriosos.
def evaluate_node(state):
if state.amount > 100_000:
state.requires_approval = True
if state.country == "BR":
...
if state.supplier_type == "CRITICAL":
...
decision = supplier_policy.evaluate(
supplier=supplier,
context=context,
)
LangGraph → orchestrates
Domain → decides
Dónde encaja MCP
Para determinar dónde encaja MCP, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Modele los nodos y las relaciones para las preguntas que planteará, no para cada sustantivo del documento. Los enlaces escasos y tipados son mejores que grafos densos y misteriosos. Para determinar dónde encaja MCP, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos confidenciales y las banderas de funcionalidad deben estar en un lugar que los operadores puedan auditar sin tener que leer todo el grafo.
LLM
↓
PostgreSQL
LLM
↓
SAP
Agent
↓
MCP Tool
↓
Application API
↓
Use Case
↓
Domain
↓
Infrastructure
create_supplier
execute_sql
PostgreSQL también es un adaptador
En el caso de PostgreSQL también como adaptador, se deben definir las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Se debe documentar tanto la ruta normal como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Se deben utilizar restricciones e índices antes de la inserción masiva. La unicidad en las claves de negocio convierte las fusiones posteriores en operaciones upsert predecibles, en lugar de situaciones de duplicados.
class SupplierRepository(Protocol):
async def add(
self,
supplier: Supplier,
) -> None:
...
class PostgresSupplierRepository(
SupplierRepository
):
def __init__(self, session):
self.session = session
async def add(
self,
supplier: Supplier,
) -> None:
entity = SupplierModel.from_domain(
supplier
)
self.session.add(entity)
FastAPI → doesn't know PostgreSQL exists
Application → doesn't know PostgreSQL existsDomain → doesn't know PostgreSQL existsInfrastructure → does
SAP debe encontrarse detrás de otro límite
Para SAP, que debería encontrarse detrás de otro límite, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves de negocio convierte las fusiones posteriores en operaciones de inserción o actualización predecibles, en lugar de situaciones de duplicados.
CreateSupplierUseCase
↓
Persist State
↓
SupplierCreated
↓
AWS SQS
↓
Integration Worker
↓
SAP
Construyendo la arquitectura
Para armar la arquitectura, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves de negocio convierte las fusiones posteriores en operaciones predictibles de inserción o actualización, en lugar de situaciones de duplicados. Para armar la arquitectura, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar que los operadores puedan auditar sin dificultad.
En todo el grafo.
USER
│
▼
FastAPI
│
▼
LangGraph
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
RAG LLM MCP
│ │ │
▼ ▼ │
OpenSearch Bedrock │
▼
Application
│
▼
Domain
│
┌─────────┴─────────┐
│ │
▼ ▼
PostgreSQL AWS SQS
│
▼
Worker
│
▼
SAP
Dos mundos
Para “Dos mundos”, defina las entradas, el propietario de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Inspeccione un vecindario de muestra después del primer lote. Las pruebas con Cypher detectan el desvío en las etiquetas y la falta de propiedades a un costo menor que las pruebas de chat de extremo a extremo.
PROBABILISTICLLM
RAG
Natural Language
Intent Understanding
Agent Reasoning
Tool Selection
DETERMINISTIC
Authorization
Business Rules
Transactions
Persistence
Idempotency
Integration
Auditability
Una prueba práctica para la arquitectura
Para una prueba práctica de la arquitectura, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y probables sobre scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Inspeccione un grupo de ejemplos después del primer lote. Las pruebas Cypher detectan cambios en las etiquetas y propiedades faltantes de manera más económica que las pruebas de chat de extremo a extremo.
async def test_supplier_requires_approval():
repository = FakeSupplierRepository()
policies = FakePolicyService(
requires_approval=True
)
events = FakeEventPublisher()
use_case = CreateSupplierUseCase(
repository=repository,
policy_service=policies,
event_publisher=events,
)
supplier = await use_case.execute(
CreateSupplierCommand(
name="ACME",
tax_id="123",
country="BR",
)
)
assert supplier.requires_approval is True
Cambiar el modelo debería ser aburrido
Para que cambiar el modelo sea algo aburrido, se deben definir las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea desde un punto de control conocido sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Inspeccione un grupo de ejemplos después del primer lote. Las pruebas Cypher detectan con mayor eficiencia los desvíos en las etiquetas y las propiedades faltantes que las pruebas de chat de extremo a extremo. Para que cambiar el modelo sea algo aburrido, se deben definir las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea desde un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben estar en un lugar que los operadores puedan auditar sin necesidad de leerlo todo.
Todo el grafo.
LLMProvider
▲
│
┌─────────┴─────────┐
│ │
BedrockProvider AnotherProvider
FastAPI
Application
Domain
Business Rules
La arquitectura limpia no es una estructura de carpetas
En “La arquitectura limpia no es una estructura de carpetas”, se deben definir las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Modele nodos y relaciones para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos y misteriosos.
src/
├── api/
├── application/
├── domain/
├── agents/
├── ports/
└── infrastructure/
La IA hace que estos límites sean aún más importantes
Dado que la IA hace que estos límites sean aún más importantes, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Prefiera unidades pequeñas y verificables en lugar de scripts extensos. Cuando un paso falla, el error debe indicar una única responsabilidad y no un proceso complicado. Diseñe nodos y relaciones de modelo para las preguntas que vaya a hacer, no para cada sustantivo del documento; los enlaces escasos y tipados son mejores que grafos densos e inexplicables.
La arquitectura en una frase
Para “La arquitectura en una frase”, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea desde un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas. Modele nodos y relaciones para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos y misteriosos. Para “La arquitectura en una frase”, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea desde un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar que los operadores puedan auditar sin tener que leerlo.
todo el grafo.Pensamiento final
Para el pensamiento final, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Utilice restricciones e índices antes de la ingesta masiva. La unicidad en las claves de negocio convierte las fusiones posteriores en inserciones predecibles en lugar de situaciones de duplicados excesivos.
Prompt
↓
LLM
↓
Tool
↓
Database
Lista de verificación operativa
Lecturas relacionadas
- Anatomía de un equipo comercial de IA: Orquestando agentes con LangGraph y FastAPI — Una explicación detallada de una plataforma multiagente de código abierto: cómo se orquestan los agentes cofundador, gerente y especialista, comparten información, se detienen para obtener aprobación e informan.
- Orquestar agentes de investigadores y programadores con LangGraph — Crear una aplicación multiagente con LangGraph que utilice prompts especializados, modelos Ollama, búsquedas en Tavily, herramientas de transferencia de tareas y puntos de control en Postgres para flujos de trabajo dirigidos por un orquestador.