Architecture de référence pour des systèmes d’IA agentielle d’entreprise de niveau professionnel
Apprenez les couches fondamentales, les stratégies de mémorisation, les mécanismes d’extraction des informations et les contraintes nécessaires pour faire évoluer l’IA agente d’un prototype vers des systèmes de production d’entreprise fiables.
Résumé exécutif
Le passage des applications de grands modèles de langage sans état vers une intelligence artificielle agente de niveau production représente un changement majeur dans la manière dont les entreprises développent des logiciels. Les premières tentatives d’intelligence artificielle au sein des entreprises se sont fortement appuyées sur la génération augmentée par récupération d’informations (RAG) de base — en intégrant des documents sous forme d’embeddings dans la fenêtre de contexte d’une requête. Cette approche fonctionne bien pour les réponses simples, mais elle est insuffisante en ce qui concerne la prise de décisions autonomes, la planification à plusieurs étapes avec gestion de l’état, l’appel fiable d’outils, ainsi que la capacité à s’autocorriger au cours d’un flux de travail.
L’IA agente d’entreprise comble cette lacune en repositionnant les modèles de base : au lieu d’agir comme interface directe avec l’utilisateur, le modèle devient un composant de raisonnement intégré à une plateforme logicielle déterministe. De tels systèmes peuvent percevoir leur environnement, diviser de grands objectifs en étapes plus petites, appeler des services internes de l’entreprise et conserver l’état d’exécution au cours de transactions multi-étapes. Cependant, passer d’un prototype fonctionnel à quelque chose qui peut être mis en production nécessite de résoudre les problèmes liés aux boucles de contrôle non déterministes instables, à la dégradation du contexte au fil des sessions longues, aux risques d’escalade de privilèges et à une consommation incontrôlée de tokens.
Cet article présente une architecture de référence du début à la fin destinée aux architectes et aux responsables du développement d’IA chargés des systèmes d’agents d’entreprise. Il couvre les couches fondamentales nécessaires à tout agent, compare les topologies multi-agents, explore en détail les modèles d’intégration pour des systèmes de récupération de niveau entreprise — en accordant une attention particulière à Amazon Kendra — et inclut des exemples fonctionnels en Python adaptés à un usage en production.
Section 1 : L’architecture canonique de l’agent d’entreprise
Couches principales : Perception, Raisonnement, Planification, Mémoire et Exécution d’outils
Un système d’agents prêt pour la production sépare le modèle fondamental probabiliste de l’environnement d’exécution déterministe qui l’entoure. Cette architecture se compose généralement de cinq couches principales, toutes fonctionnant à l’intérieur d’un conteneur d’environnement d’exécution géré :
- Couche de perception : Elle convertit les signaux bruts provenant des entreprises — télémétrie, messages d’utilisateurs, charges utiles de webhook, réponses API — en entrées propres et typées que le reste du système peut utiliser. Cela inclut la purification des entrées, la limitation des fréquences d’appel et des vérifications de schéma précoce, toutes effectuées avant même que l’LLM ne soit invoqué.
- Moteur de raisonnement : C’est le centre cognitif, alimenté par un LLM sous-jacent (par exemple, les modèles Anthropic Claude 3.5 Sonnet ou AWS Bedrock). Plutôt que d’agir directement, ce moteur interprète le contexte actuel, évalue les différentes options possibles et produit des déclarations structurées indiquant l’intention.
Modèles d’orchestration : boucles à un agent vs. topologies multi-agents
Le choix du bon modèle d’orchestration dépend fortement du degré de complexité du domaine cible :
- Boucles ReAct à un agent : Une boucle de raisonnement qui alterne entre les phases de Réflexion, d’Action et d’Observation. Cela convient aux workflows linéaires et simples qui utilisent moins de 5 à 8 outils distincts. Au-delà, un seul agent risque de rencontrer des fenêtres de contexte surchargées, des instructions floues et de la confusion quant au choix de l’outil à utiliser.
Pour une utilisation en production d’entreprise, la Topologie Multi-Agent Superviseur est la valeur par défaut recommandée, grâce à ses frontières d’état claires, à une auditabilité facilitée et à des coûts contextuels plus prévisibles.
Section 2 : Gestion de la mémoire d’entreprise et hygiène du contexte
Hierarchie de la mémoire : Mémoire de travail, trajectoire à court terme et mémoire à long terme
Considérez la gestion de la mémoire des agents comme un problème de budgétisation des ressources. Laisser la mémoire croître sans contrôle augmente les coûts d’inférence, ajoute de la latence et fait perdre au modèle le fil du contexte à mesure que celui-ci se dégrade. Un agent d’entreprise bien conçu nécessite une structure de mémoire en couches :
- Mémoire de travail (Écran tactile) : La fenêtre contextuelle en temps réel — les instructions système actuellement en vigueur, l’état de la tâche active et les résultats des outils les plus récents.
Stratégies de compression du contexte, d’élagage et d’isolation de l’état
Pour éviter que le contexte ne se détériore, le système en temps de exécution doit appliquer activement des règles d’hygiène :
- Troncature des observations : Les réponses brutes des outils — comme une charge utile JSON contenant 500 enregistrements — ne doivent jamais être directement transférées dans la mémoire de travail. Le système en temps de exécution doit nettoyer, tronquer ou synthétiser ce que l’outil renvoie avant qu’il n’atteigne le moteur de raisonnement.
- Compaction par fenêtre déroulante :Lorsque le espace temporaire dépasse un seuil budgétaire prédéfini (par exemple, 20 % de la limite totale de contexte du modèle), une étape de compression transforme les échanges précédents de la conversation en résumés sémantiques condensés, tout en supprimant les échanges bruts originaux.
- Isolement de l’état des sous-tâches :Lorsque le superviseur confie une tâche à un sous-agent, il crée pour ce dernier un contexte vierge ne contenant que l’objectif spécifique et les paramètres nécessaires — aucune trace de l’historique de raisonnement du superviseur n’est transmise.
Section 3 : Analyse approfondie : Ancrage de la récupération d’informations dans les entreprises via Amazon Kendra
Amazon Kendra en tant que couche de récupération en production
Construire un pipeline de récupération sur une base de données vectorielle générique signifie généralement de développer depuis zéro sa propre logique d’ingestion des documents, de segmentation en blocs, de génération d’encodages vectoriels, ainsi qu’une couche de fusion des recherches hybrides. Amazon Kendra, quant à lui, propose un moteur de recherche d’entreprise entièrement géré qui gère tout cela directement. Il intègre une compréhension du langage naturel en plusieurs étapes, un analyseur structuré qui prend en compte les tableaux, en-têtes et pieds de page, ainsi qu’un modèle de classement hybride combinant des signaux lexicaux et sémantiques.
Dans une pile d’agents d’entreprise, Kendra agit généralement comme le Sous-système de fondation des connaissances central — le composant chargé de permettre aux agents d’obtenir un contexte factuel vérifié à partir de sources de données internes diverses, sans le risque de fragmentation lié à une simple division des données par stockage vectoriel.
Ajustement avancé de la pertinence et suppression dynamique des métadonnées
Afin de permettre une prise de décision autonome en toute sécurité, les systèmes de recherche de niveau entreprise nécessitent une gestion précise des permissions ainsi que des contrôles de pertinence ajustables :
- Listes de contrôle d’accès natives (ACL) : Kendra récupère automatiquement et met en correspondance les ACL au niveau du document définies dans les systèmes d’origine, qu’il s’agisse de SharePoint, Confluence ou S3. Lorsqu’un agent envoie une requête, il transmet l’identité de l’utilisateur authentifié sous forme de jeton
UserContext, et Kendra applique des restrictions de sécurité au niveau de l’index afin que aucun extrait que l’utilisateur demandeur n’a pas le droit de voir ne parvienne à l’agent.
_last_updated_at, selon la catégorie métier, ou en fonction de correspondances exactes dans des champs personnalisés tels que Department ou ProjectCode.Retrieve par rapport à l’API polyvalente Query. L’API Retrieve ignore complètement les métadonnées de recherche orientées vers l’interface utilisateur et renvoie plutôt des extraits détaillés au niveau des passages, conçus spécifiquement pour être directement intégrés dans la fenêtre de contexte d’un LLM.Section 4 : Garde-fous déterministes, validation des outils et disjoncteurs
Contrôles pré-exécution (feedforward) et post-exécution (feedback)
Les agents autonomes ont besoin de limites claires au niveau logiciel pour empêcher l’escalade des privilèges, les appels à des outils mal formatés ou les boucles d’exécution infinies :
- Validation feedforward (avant exécution) : Avant même qu’un appel à un outil n’atteigne un système d’entreprise interne, l’environnement de exécution vérifie ses arguments selon des schémas Pydantic stricts — en s’assurant que les champs requis existent, que les valeurs numériques ou d’énumération se situent dans des plages autorisées, et que le token d’autorisation de l’appelant permet réellement cette action.
Erreur : La table de base de données ‘users_v2’ n’a pas été trouvée. Tables disponibles : ['users', 'orders'] — fournissant à l’agent des informations exploitables pour ajuster sa prochaine action.Sandboxing, budgets d’étapes et disjoncteurs automatiques
- Sandboxing isolé : Tout code généré dynamiquement — par exemple, du Python produit par un agent d’analyse de données — doit s’exécuter à l’intérieur de sandbox isolés et jetables tels que des conteneurs Docker ou des micro-VMs gVisor, avec un accès réseau sortant strictement contrôlé.
- Budgets de pas et de coûts : Chaque tâche d’agent doit être limitée par un plafond strict en termes d’itérations de appel d’outils (par exemple, pas plus de 10) ainsi que par un montant maximal de tokens consommés. Dès que l’un de ces limites est dépassé, le système doit interrompre son exécution et transférer la tâche à un opérateur humain.
- Disjoncteurs d’outils : Si une API backend ou une base de données continue de rencontrer des problèmes lors de tentatives successives de récupération, le système doit déclencher un disjoncteur, marquant cet outil comme
UNAVAILABLEdans le catalogue d’outils de l’agent, afin que la couche de planification opte pour des solutions alternatives plutôt que d’essayer sans cesse de contourner une dépendance défectueuse.
Section 5 : Schéma de mise en œuvre en production
L’exemple ci-dessous est une application Python complète et exécutable qui illustre une architecture multi-agents Supervisor de niveau production. Elle combine la validation des outils basée sur Pydantic, la planification budgétaire par étapes, une gestion des erreurs déterministe, ainsi qu’un outil d’encapsulation pour les recherches via Amazon Kendra.
import os
import json
import time
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError
# =====================================================================
# 1. TOOL SCHEMAS & ENTERPRISE INTEGRATION CONTRACTS
# =====================================================================
class KendraRetrieveInput(BaseModel):
"""Input contract for the Amazon Kendra retrieval tool."""
query_text: str = Field(..., description="The natural language query string to search across corporate documentation.")
department_filter: Optional[str] = Field(None, description="Optional department metadata filter (e.g., 'Engineering', 'HR').")
user_id: str = Field(..., description="Authenticated user ID used for native Kendra ACL security trimming.")
class DatabaseQueryInput(BaseModel):
"""Input contract for enterprise relational database lookups."""
query_type: str = Field(..., description="Must be 'SELECT'. Data mutation operations are strictly prohibited.")
table_name: str = Field(..., description="Target database table name.")
limit: int = Field(default=5, ge=1, le=20, description="Number of records to return.")
# =====================================================================
# 2. MOCK ENTERPRISE SERVICES & KENDRA TOOL ENGINE
# =====================================================================
class MockAmazonKendraClient:
"""Simulates Amazon Kendra's high-density Retrieve API with ACL security trimming."""
def __init__(self):
self._mock_index = [
{
"doc_id": "KENDRA-DOC-001",
"content": "Production deployment requires dual sign-off from Engineering and Security leads.",
"department": "Engineering",
"acl_users": ["user_eng_101", "admin_007"]
},
{
"doc_id": "KENDRA-DOC-002",
"content": "Standard employee travel stipend is capped at $150/day for domestic lodging.",
"department": "HR",
"acl_users": ["user_hr_201", "user_eng_101", "admin_007"]
}
]
def retrieve(self, query_text: str, user_id: str, department_filter: Optional[str] = None) -> List[Dict[str, Any]]:
results = []
for doc in self._mock_index:
# Enforce Document-Level ACL Security Trimming
if user_id not in doc["acl_users"]:
continue
# Apply Optional Metadata Department Filtering
if department_filter and doc["department"].lower() != department_filter.lower():
continue
results.append({
"DocumentId": doc["doc_id"],
"ContentSnippet": doc["content"],
"Department": doc["department"]
})
return results
class MockEnterpriseDatabase:
"""Simulates a secure internal enterprise database."""
def __init__(self):
self._tables = {
"deployments": [
{"id": 1, "service": "auth-service", "status": "COMPLETED", "env": "prod"},
{"id": 2, "service": "payment-api", "status": "PENDING_APPROVAL", "env": "prod"}
]
}
def execute_select(self, query_type: str, table_name: str, limit: int) -> List[Dict[str, Any]]:
if query_type.upper() != "SELECT":
raise ValueError(f"Security Alert: Unauthorized operation '{query_type}'. Only 'SELECT' is permitted.")
if table_name not in self._tables:
raise KeyError(f"Database Error: Table '{table_name}' does not exist. Available tables: {list(self._tables.keys())}")
return self._tables[table_name][:limit]
# =====================================================================
# 3. PRODUCTION AGENT HARNESS & RUNTIME ENGINE
# =====================================================================
class ProductionAgentRuntime:
"""
Deterministic software harness surrounding probabilistic reasoning models.
Enforces budgets, schema validation, sandboxing, and error recovery loops.
"""
def __init__(self, user_id: str, max_step_budget: int = 4):
self.user_id = user_id
self.max_step_budget = max_step_budget
self.kendra_service = MockAmazonKendraClient()
self.db_service = MockEnterpriseDatabase()
def execute_task(self, task_goal: str) -> Dict[str, Any]:
print(f"=== [Harness Started] Initiating Task: '{task_goal}' for User: '{self.user_id}' ===")
step_count = 0
scratchpad_history: List[str] = []
# Simulated dynamic model trajectory outputs (demonstrating multi-step execution & self-correction)
simulated_llm_turns = [
# Turn 1: Attempt invalid database deletion (Caught by Feedforward Schema/Guardrail)
{
"thought": "I will clean up old deployment logs before checking security compliance.",
"action": "execute_db_query",
"args": {"query_type": "DELETE", "table_name": "deployments", "limit": 5}
},
# Turn 2: Corrected database lookup
{
"thought": "I will check active deployment statuses in the enterprise database.",
"action": "execute_db_query",
"args": {"query_type": "SELECT", "table_name": "deployments", "limit": 2}
},
# Turn 3: Ground task using Amazon Kendra Retrieve API
{
"thought": "Now I need to query enterprise policies regarding production deployment sign-off.",
"action": "kendra_retrieve",
"args": {"query_text": "production deployment sign-off rules", "department_filter": "Engineering"}
}
]
while step_count < self.max_step_budget:
step_count += 1
print(f"\n--- [Step {step_count}/{self.max_step_budget}] ---")
# Fetch current simulated model decision turn
turn_data = simulated_llm_turns[min(step_count - 1, len(simulated_llm_turns) - 1)]
print(f"Agent Thought: {turn_data['thought']}")
action = turn_data.get("action")
args = turn_data.get("args", {})
# --- TOOL EXECUTION BRANCH: DATABASE ---
if action == "execute_db_query":
try:
# 1. Pre-execution Feedforward Schema Validation
validated_args = DatabaseQueryInput(**args)
# 2. Tool Execution
db_results = self.db_service.execute_select(
query_type=validated_args.query_type,
table_name=validated_args.table_name,
limit=validated_args.limit
)
observation = f"Database Query Success: {json.dumps(db_results)}"
print(f"[Observation]: {observation}")
scratchpad_history.append(observation)
except ValidationError as ve:
error_msg = f"Schema Validation Blocked Action: {ve.errors()[0]['msg']}"
print(f"[Harness Feedforward Intercept]: {error_msg}")
scratchpad_history.append(error_msg)
except (ValueError, KeyError) as exec_err:
error_msg = f"Tool Execution Failure: {str(exec_err)}"
print(f"[Harness Feedback Sensor Catch]: {error_msg}")
scratchpad_history.append(error_msg)
# --- TOOL EXECUTION BRANCH: AMAZON KENDRA RETRIEVE ---
elif action == "kendra_retrieve":
try:
# Inject authenticated context
args["user_id"] = self.user_id
# 1. Pre-execution Schema Validation
validated_kendra_args = KendraRetrieveInput(**args)
# 2. Execute Kendra Retrieve Call
kendra_passages = self.kendra_service.retrieve(
query_text=validated_kendra_args.query_text,
user_id=validated_kendra_args.user_id,
department_filter=validated_kendra_args.department_filter
)
observation = f"Amazon Kendra Retrieved {len(kendra_passages)} Grounding Snippets: {json.dumps(kendra_passages)}"
print(f"[Observation]: {observation}")
scratchpad_history.append(observation)
# Successful multi-step completion condition reached
return {
"status": "SUCCESS",
"completed_in_steps": step_count,
"trajectory": scratchpad_history
}
except ValidationError as ve:
error_msg = f"Kendra Schema Error: {ve.errors()[0]['msg']}"
print(f"[Harness Intercept]: {error_msg}")
scratchpad_history.append(error_msg)
return {"status": "FAILED", "reason": "Step budget exhausted without completing task goals."}
# =====================================================================
# 4. EXECUTION DRIVER
# =====================================================================
if __name__ == "__main__":
# Instantiate runtime for an authorized engineering user
agent_runtime = ProductionAgentRuntime(user_id="user_eng_101", max_step_budget=4)
# Run Agentic Workflow
final_execution_summary = agent_runtime.execute_task(
task_goal="Verify pending production deployments and confirm authorization policies."
)
print("\n================ FINAL SYSTEM SUMMARY ================")
print(json.dumps(final_execution_summary, indent=2))
Section 6 : Opérations en production, télémétrie et gouvernance
Observabilité : Suivi des trajectoires avec OpenTelemetry
L’exécution de systèmes agents en production exige un niveau de suivi que les tableaux de bord APM ordinaires ne peuvent pas fournir. Étant donné que les agents suivent des chemins d’exécution variables et non déterministes, votre stack de télémétrie doit reconstituer l’ensemble de l’arbre des trajectoires empruntées par un agent, et non seulement une paire demande-réponse unique :
- Granularité des spans : Chaque tour de l’agent doit générer un ensemble de spans OpenTelemetry imbriqués, dont chaque span est dédié à la construction du prompt système, à la mesure du temps nécessaire pour que le modèle réponde, à la vérification que les arguments de l’outil correspondent au schéma attendu, à la chronométrage de l’appel effectif à l’outil, ainsi qu’à l’exécution de toute opération de compactage de la mémoire qui a lieu au cours de ce tour.
- Enregistrement de l’état de la trajectoire : Les spans doivent enregistrer le nombre de tokens dans le contexte d’entrée, le nombre de tokens dans la sortie, les schémas utilisés pour les paramètres de l’outil, ainsi que la longueur brute des observations retournées. C’est ce niveau de détail qui permet d’attribuer avec précision les coûts et de déterminer exactement quelle étape d’une trajectoire a constitué un goulot d’étranglement.
Règles opérationnelles et évaluation (la triade agente)
Pour maintenir un agent en bon état de fonctionnement, il est nécessaire d’effectuer des évaluations automatisées en continu selon trois dimensions clés :
- Taux d’achèvement des objectifs : proportion d’exécutions de l’agent qui atteignent un état final réussi sans épuiser leur budget d’étapes ni générer une exception non gérée.
- Ancrage dans le contexte, ou indice d’hallucination : cette mesure vérifie si la réponse résumée finale de l’agent est effectivement fondée sur des faits provenant de ses outils de recherche (tels que les extraits fournis par Amazon Kendra), plutôt que d’être générée à partir de la mémoire paramétrique du modèle lui-même.
Conclusion avec des recommandations pratiques
Construire des systèmes d’IA agente de niveau entreprise signifie traiter le modèle de base sous-jacent comme un composant de raisonnement probabiliste intégré à un cadre logiciel déterministe et strictement contrôlé. Ce sont les organisations qui parviennent à passer des prototypes aux versions en production que celles qui établissent des frontières architecturales claires entre la perception, la planification, la mémoire et l’exécution des outils, plutôt que de permettre au modèle de contrôler librement tous ces éléments en même temps.
Plan d’action concret pour les équipes d’architecture
- Séparer la réflexion de l’exécution : chaque appel à une outil initié par un LLM doit passer par un mécanisme déterministe qui valide les arguments selon un schéma Pydantic avant exécution et capture proprement les erreurs par la suite.
- Utiliser Amazon Kendra pour l’ancrage contextuel : appelez l’API
Retrievede Kendra afin de fournir à l’agent un contexte dense au niveau des passages, tout en comptant sur son système de contrôle d’accès aux documents au niveau de l’index pour assurer automatiquement le respect des règles de sécurité. - Limiter explicitement l’utilisation de la mémoire : appliquez une compression progressive du contexte et isolez le contexte des sous-tâches afin d’éviter que la dégradation du contexte, les écarts dans les instructions ou une consommation excessive de tokens ne surviennent au cours de sessions prolongées.
Lectures complémentaires
- Pourquoi les coûts de l’IA agente explosent : un modèle de coût axé sur l’architecture — Explique pourquoi les coûts des agents basés sur des LLM doivent être mesurés par tâche accomplie et non par appel, et présente des leviers architecturaux tels que le routage des modèles, les budgets de contexte et le cache pour contrôler les dépenses.
- Neuf piliers architecturaux pour des systèmes AI-agents de niveau production — Découvrez un plan architectural basé sur neuf piliers couvrant le réseau de confiance zéro, les niveaux de données et la liaison aux preuves, afin de créer des systèmes AI-agents auditables de niveau entreprise.
- Concevoir des espaces de travail AI-agents adaptés au fonctionnement réel des équipes — Explique dix principes fondamentaux pour concevoir des espaces de travail AI-agents qui intègrent le contexte, les connaissances, l’autonomie, l’accès aux outils et l’automatisation des workflows en un système unifié.
- Choisir et ajuster des modèles d’embedding pour des systèmes RAG en production — Découvrez comment les modèles d’embedding transforment le texte en vecteurs recherchables, pourquoi le vocabulaire sectoriel entrave la recherche sémantique, et comment sélectionner, compresser et affiner des modèles pour des systèmes RAG en production.