Création d’agents IA avec FastAPI, LangGraph et architecture propre
Frontières de couche, état du graphe défini par type, services testables, et structure de déploiement permettant de maintenir les agents en bon état.
Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : Comment structurer des agents d’IA de production avec FastAPI, LangGraph et l’architecture propre. L’accent est mis sur des étapes opérationnelles, des vérifications explicites, ainsi que du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner son intention. Pour une vue d’ensemble, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stockages de secrets et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphe.
@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 est une interface — pas l’application
Puisque FastAPI est une interface — et non l’application elle-même, il convient de définir les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Documentez conjointement le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages échoués font partie intégrante du produit, et non d’améliorations apportées ultérieurement. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité basée sur des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion ou de mise à jour prévisibles, plutôt que de créer des doublons en masse.
@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
L’application communique avec des capacités, pas avec des technologies
Puisque l’application communique avec des capacités et non avec des technologies, il convient de définir les entrées, le responsable de l’étape et les critères de fin avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion ou de mise à jour prévisibles, plutôt que de créer des doublons en masse.
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
Les ports définissent la frontière
Pour Ports Create the Boundary, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute mise à jour partielle silencieuse. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse. Pour Ports Create the Boundary, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’intégralité du code.
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
Où se situe Amazon Bedrock
Pour déterminer où se situe Amazon Bedrock, il faut définir les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez à la fois le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations ultérieures. Vérifiez un échantillon de données après le premier lot. Les sondes Cypher permettent de détecter plus facilement les écarts dans les étiquettes et les propriétés manquantes que les tests de conversation du début à la fin.
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
Où se situe RAG
Pour déterminer où RAG doit être intégré, il convient de définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Vérifiez un échantillon du voisinage après le premier lot. Les sondes Cypher permettent de détecter plus facilement les écarts dans les étiquettes et les propriétés manquantes que les tests de conversation du début à la fin.
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 fournit du contexte. Il ne prend pas de décisions.
Pour RAG Gives Context. It Doesn’t Own the Decision., il faut définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites les vérifications de succès et refusez les terminaisons partielles silencieuses. Inspectez un échantillon du voisinage après le premier lot. Les sondes Cypher permettent de détecter plus facilement les écarts dans les étiquettes et les propriétés manquantes que les tests de conversation du début à la fin. Pour RAG Gives Context. It Doesn’t Own the Decision., il faut définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets et les flags fonctionnels doivent être regroupés en un seul endroit accessible aux opérateurs.
Il est possible d’effectuer un audit sans avoir à lire l’ensemble du graphe.RAG
↓
LLM
↓
"Looks fine"
↓
Create Supplier
RAG
↓
Relevant Context
↓
Agent Reasoning
↓
Application
↓
Domain Rules
↓
Decision
Où se situe LangGraph
Pour déterminer où se situe LangGraph, il faut d’abord définir les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations ultérieures. Modélisez les nœuds et les relations pour les questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux.
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 ne devrait pas devenir le nouveau monolithe
Afin que LangGraph ne devienne pas le nouveau monolithe, il convient de définir les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Modélisez les nœuds et les relations en fonction des questions que vous souhaitez poser, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux.
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
Où s’insère MCP
Pour déterminer où MCP s’insère, définitz les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définites des vérifications de succès et refusez toute exécution partielle silencieuse. Modélisez les nœuds et les relations en fonction des questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux. Pour déterminer où MCP s’insère, définitz les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphique.
LLM
↓
PostgreSQL
LLM
↓
SAP
Agent
↓
MCP Tool
↓
Application API
↓
Use Case
↓
Domain
↓
Infrastructure
create_supplier
execute_sql
PostgreSQL est également un adaptateur
Pour PostgreSQL est également un adaptateur, il faut définir les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’améliorations ultérieures. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations upsert prévisibles, plutôt que de créer des doublons en masse.
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 doit se trouver derrière une autre frontière
Pour SAP, qui doit se trouver derrière une autre frontière, il convient de définir les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Préférez des unités petites et testables aux scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.
CreateSupplierUseCase
↓
Persist State
↓
SupplierCreated
↓
AWS SQS
↓
Integration Worker
↓
SAP
Mettre en place l’architecture
Pour mettre en place l’architecture, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute mise à jour partielle silencieuse. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse. Pour mettre en place l’architecture, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans difficulté.
Traduisez l’ensemble du graphe.
USER
│
▼
FastAPI
│
▼
LangGraph
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
RAG LLM MCP
│ │ │
▼ ▼ │
OpenSearch Bedrock │
▼
Application
│
▼
Domain
│
┌─────────┴─────────┐
│ │
▼ ▼
PostgreSQL AWS SQS
│
▼
Worker
│
▼
SAP
Deux mondes
Pour Deux Mondes, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie du produit, et non d’une mise en forme ultérieure. Inspectez un échantillon de voisinage après le premier lot. Les sondes Cypher permettent de détecter les écarts dans les étiquettes et l’absence de propriétés à moindre coût que les tests de chat bout en bout.
PROBABILISTICLLM
RAG
Natural Language
Intent Understanding
Agent Reasoning
Tool Selection
DETERMINISTIC
Authorization
Business Rules
Transactions
Persistence
Idempotency
Integration
Auditability
Un test pratique pour l’architecture
Pour un test pratique de l’architecture, définissez les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Vérifiez un échantillon de données après le premier lot. Les outils de sondage Cypher permettent de détecter plus facilement les écarts dans les étiquettes et les propriétés manquantes que les tests de conversation du début à la fin.
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
Changer le modèle devrait être ennuyeux
Pour que la modification du modèle soit ennuyeuse, il faut définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les terminations partielles silencieuses. Inspectez un échantillon du voisinage après le premier lot. Les sondes Cypher détectent plus facilement les écarts de labels et les propriétés manquantes que les tests de conversation bout en bout. Pour que la modification du modèle soit ennuyeuse, il faut définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à les lire.
de tout le graphe.
LLMProvider
▲
│
┌─────────┴─────────┐
│ │
BedrockProvider AnotherProvider
FastAPI
Application
Domain
Business Rules
L’architecture propre n’est pas une structure de dossiers
Dans l’approche où l’architecture propre n’est pas une structure de dossiers, il convient de définir les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Modélisez les nœuds et les relations en fonction des questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux.
src/
├── api/
├── application/
├── domain/
├── agents/
├── ports/
└── infrastructure/
L’IA rend ces limites encore plus importantes
Puisque l’IA rend ces limites encore plus importantes, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Préférez des unités petites et testables à des scripts complexes. Lorsqu’une étape échoue, l’échec doit indiquer une seule responsabilité plutôt qu’un processus embrouillé. Modélisez les nœuds et les relations pour les questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux.
L’architecture en une phrase
Pour « L’architecture en une phrase », définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute mise en œuvre partielle silencieuse. Modélisez les nœuds et les relations en fonction des questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux. Pour « L’architecture en une phrase », définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Gardez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire le reste du code.
tout le graphe.Pensée finale
Pour la pensée finale, définissez les entrées, le responsable de l’étape et les critères de sortie avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie du produit, et non d’une mise en forme ultérieure. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.
Prompt
↓
LLM
↓
Tool
↓
Database