Comment créer un pipeline RAG prêt pour l’entreprise sur AWS avec Bedrock, Zilliz Cloud et LangChain
La génération augmentée par récupération (RAG) est rapidement devenue l’épine dorsale des solutions LLM de niveau entreprise : près de 90 % des entreprises s’appuient dessus pour ancrer les LLM dans des connaissances fiables et spécifiques à leur domaine. Mais la réalité est plus compliquée : l’écosystème RAG a explosé avec une multitude d’options. Les LLM, les modèles d’embeddings, les frameworks d’orchestration et les bases de données vectorielles ont chacun leurs propres schémas d’intégration, laissant les équipes peiner à assembler des solutions qui fonctionnent dans les contraintes de l’entreprise.
Pour les organisations fortement investies dans AWS, ce défi est encore plus aigu. Vous ne pouvez pas simplement arracher l’infrastructure existante ni contourner les politiques de sécurité et de conformité établies. Ce dont vous avez besoin, c’est d’une architecture RAG qui s’intègre parfaitement à votre environnement AWS, exploite des services prêts pour l’entreprise et reste pérenne.
Dans ce tutoriel, nous allons construire exactement cela : un pipeline RAG prêt pour l’entreprise utilisant AWS Bedrock (modèles Nova + Titan), Zilliz Cloud comme base de données vectorielle, et LangChain pour l’orchestration. À la fin, vous disposerez d’une base pratique, sécurisée et prête pour la production que vous pourrez déployer directement dans votre stack AWS — sans compromis, sans solutions de contournement.
Comment nous concevons RAG pour le passage à l’échelle
Avant de nous plonger dans le code, comprenons ce que nous construisons et pourquoi cela résout de véritables problèmes d’entreprise.
Les LLM traditionnels se heurtent à deux murs majeurs dans les contextes d’entreprise. Leurs connaissances s’arrêtent au moment de l’entraînement — aucun accès à vos derniers rapports, à vos données clients ou aux évolutions du secteur. De plus, ils hallucinent fréquemment sans moyen de retracer leur raisonnement. Ce n’est pas exactement ce que vous souhaitez pour alimenter des applications destinées aux clients.
RAG change complètement la donne. Au lieu de réentraîner des modèles massifs, vous récupérez d’abord les informations pertinentes, puis générez des réponses à partir de ce contexte. Les avantages sont immédiats : 25 à 40 % de précision en plus, plus de 60 % d’hallucinations en moins et une traçabilité complète des réponses. Votre IA connaît soudainement les résultats du dernier trimestre et peut citer ses sources.
Notre système RAG d’entreprise suit le modèle éprouvé de l’architecture MVC (Model-View-Controller) :
Couche Model : Gère le gros du travail — traitement des documents, embeddings, stockage vectoriel et inférence LLM
Couche View : Gère les interfaces utilisateur et les réponses API
Couche Controller : Orchestre le workflow via des fonctions Lambda et la gestion des événements
Cinq moteurs principaux alimentent notre implémentation :
Moteur de traitement des requêtes : Transforme les questions des utilisateurs en requêtes de recherche optimisées
Moteur de récupération vectorielle : Trouve du contenu pertinent grâce à la recherche sémantique de Zilliz Cloud
Module de reranking : Priorise les résultats à l’aide de règles de pertinence et métier
Moteur de génération : Synthétise le contexte en réponses précises via AWS Bedrock
Épine dorsale événementielle : Maintient l’ensemble faiblement couplé via Amazon EventBridge
La stack technologique que nous utiliserons
Construire un RAG d’entreprise ne consiste pas seulement à choisir les “meilleurs” outils individuels : il s’agit d’assembler des technologies qui s’intègrent parfaitement à votre écosystème AWS. Dans ce tutoriel, nous combinerons AWS Lambda pour le calcul, AWS Bedrock (modèles Nova + Titan) pour les embeddings et la génération, Zilliz Cloud pour la recherche vectorielle, et LangChain pour l’orchestration.
AWS Lambda : fondation de calcul élastique
Lambda nous offre une épine dorsale serverless : aucun serveur à gérer, une mise à l’échelle instantanée de zéro à des milliers de requêtes, et une tarification à l’exécution. Chaque étape RAG — traitement des documents, vectorisation, récupération et génération — s’exécute comme sa propre fonction Lambda indépendante. Cette conception rend le système modulaire, tolérant aux pannes et rentable.
AWS Bedrock : hub de modèles flexible
Bedrock donne accès à plus de 50 modèles de fondation serverless (plus plus de 120 options de marketplace d’Amazon, Anthropic, Meta, et d’autres). Sa fonctionnalité phare ? Le remplacement de modèles sans modifier le code de votre application. Cela signifie que vous pouvez effectuer des tests A/B, optimiser la latence par rapport au coût, ou adopter de nouveaux modèles sans réarchitecturer.
Zilliz Cloud : Base de données vectorielle d’entreprise la plus performante
Basé sur l’open-source Milvus, Zilliz Cloud élimine les approximations liées au réglage des index avec AutoIndex, qui s’adapte dynamiquement à vos données. Son moteur de recherche Cardinal offre des performances jusqu’à 10 fois plus rapides que les bases de données vectorielles traditionnelles, tout en évoluant de manière fluide jusqu’à des milliards de vecteurs — un élément essentiel pour les déploiements à l’échelle de l’entreprise.
LangChain : Couche d’orchestration
LangChain relie le tout — en gérant le flux entre l’embedding, la récupération et la génération. Avec ses intégrations AWS et ses abstractions flexibles, il garde notre architecture propre, modulaire et prête pour la production.
Une fois notre stack en place, il est temps de passer à la pratique et de commencer à construire.
Premiers pas pour construire un RAG prêt pour l’entreprise sur AWS
Configuration de l’infrastructure AWS CDK
Nous utiliserons AWS CDK (Cloud Development Kit) pour définir l’ensemble de votre stack — fonctions Lambda, API Gateway, buckets S3, CloudFront — et le déployer de manière cohérente dans différents environnements. CDK vous permet de versionner et de réviser votre infrastructure exactement comme du code applicatif.
# Core Lambda function configuration
lambda_function = lambda_.Function(
self, "RAGQueryFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
memory_size=3008,
timeout=Duration.seconds(30),
reserved_concurrency=100,
environment={
"ZILLIZ_ENDPOINT": self.zilliz_endpoint,
"BEDROCK_MODEL_ID": "amazon.nova-pro-v1:0"
}
)
L’étape CDK Bootstrap crée les ressources AWS fondamentales : buckets S3 pour les artefacts de déploiement, rôles IAM pour les autorisations, et paramètres SSM pour la configuration. Vous pouvez ensuite déployer des stacks séparées pour les environnements de développement, de préproduction et de production.
Connexion à Zilliz Cloud
La configuration de Zilliz Cloud implique trois étapes : créer une collection, optimiser l’indexation et établir les connexions. Nous utilisons des vecteurs à 1024 dimensions avec une indexation HNSW pour un équilibre optimal entre précision de recherche et vitesse.
# Zilliz connection configuration
connections.connect(
alias="default",
uri=ZILLIZ_ENDPOINT,
token=ZILLIZ_TOKEN,
timeout=30
)
# Create optimized collection
collection = Collection("rag_collection")
index_params = {
"metric_type": "IP",
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 128}
}
Utilisez des partitions pour organiser les données par type de document ou domaine métier afin d’améliorer les performances de recherche lors du traitement de grandes collections de documents.
Rationalisation de votre workflow de développement
Le Makefile fournit des commandes unifiées pour installer les dépendances, exécuter les tests, déployer dans différents environnements et nettoyer les ressources.
# Standardized development process
install: # Install dependencies
test: # Run tests
lint: # Code checking
deploy: # Deploy application
clean: # Clean environment
Les pipelines CI/CD gèrent les contrôles de qualité du code, la validation des types et les tests automatisés.
Construction des fonctionnalités principales
Pipeline de traitement des documents
Le pipeline traite les documents en quatre étapes : analyse, nettoyage du contenu, découpage intelligent et extraction des métadonnées.class DocumentProcessor:
class DocumentProcessor:
def process(self, document):
# Document Parsing
parsed_content = self.parse_document(document)
# Content cleaning and preprocessing
cleaned_text = self.clean_content(parsed_content)
# Intelligent chunking
chunks = self.chunk_text(cleaned_text,
chunk_size=1000,
overlap=100)
# Metadata extraction
metadata = self.extract_metadata(document)
return processed_chunks
Notre stratégie de découpage utilise une segmentation sémantique qui respecte les limites des paragraphes et conserve ensemble les informations liées. Les tailles des segments s’ajustent automatiquement en fonction du type de document.
Vectorisation et stockage
Titan Embeddings d’AWS Bedrock traite les documents par lots pour plus d’efficacité. La mise en cache des vecteurs évite de recalculer les embeddings pour le contenu déjà traité.
class VectorProcessor:
def __init__(self):
self.embedding_model = TitanEmbeddings()
self.batch_size = 32
def vectorize_batch(self, texts):
# Batch vectorization
embeddings = self.embedding_model.embed_documents(texts)
# Vector normalization
normalized_embeddings = self.normalize_vectors(embeddings)
return normalized_embeddings
L’approche de stockage par niveaux conserve les vecteurs fréquemment consultés dans un stockage haute vitesse, tout en archivant les contenus plus anciens dans des niveaux optimisés pour les coûts.
Stratégie de récupération hybride
Nous combinons la similarité vectorielle avec la correspondance par mots-clés au moyen d’un processus en plusieurs étapes : une récupération initiale large, puis un classement de précision pour obtenir les meilleurs résultats.
class HybridRetriever:
def retrieve(self, query, top_k=10):
# Vector retrieval
vector_results = self.vector_search(query, top_k*2)
# Keyword retrieval
keyword_results = self.keyword_search(query, top_k*2)
# Result fusion
merged_results = self.merge_results(
vector_results, keyword_results
)
# Reranking
reranked_results = self.rerank(query, merged_results)
return reranked_results[:top_k]
Les modèles Cross-Encoder gèrent le reclassement afin d’identifier les résultats les plus pertinents. L’optimisation de la fenêtre de contexte garantit que vous obtenez la bonne quantité d’informations pour chaque requête.
Intégration LangChain
LangChain orchestre le processus de récupération et de génération à l’aide de modèles RAG éprouvés provenant de LangChain Hub.
from langchain.chains import RetrievalQA
from langchain.retrievers import VectorStoreRetriever
# Build RAG chain
qa_chain = RetrievalQA.from_chain_type(
llm=BedrockLLM(model_id="amazon.nova-pro-v1:0"),
chain_type="stuff",
retriever=ZillizRetriever(
collection=collection,
search_params={"top_k": 5}
),
return_source_documents=True
)
# Execute query
result = qa_chain.invoke({"query": user_question})
La gestion de la mémoire maintient le contexte de la conversation pour les échanges à plusieurs tours. Les réponses en streaming affichent les résultats en temps réel au lieu d’attendre des réponses complètes. La gestion des erreurs garantit une récupération en douceur lorsque des problèmes surviennent.
Déploiement d’une architecture sans serveur
Excellence de la conception des fonctions Lambda
Chaque fonction Lambda suit les principes de responsabilité unique, en se concentrant sur une logique métier spécifique. Nos fonctions principales — traitement de documents, vectorisation, récupération et génération — se déploient et évoluent indépendamment pour une flexibilité maximale.
# Query processing Lambda function
def lambda_handler(event, context):
try:
# Initialize connections (outside handler)
query = event['query']
# Vector retrieval
retriever = ZillizRetriever()
relevant_docs = retriever.search(query, top_k=5)
# LLM generation
llm = BedrockLLM()
response = llm.generate(query, relevant_docs)
return {
'statusCode': 200,
'body': json.dumps(response)
}
except Exception as e:
logger.error(f"Error: {str(e)}")
return error_response(e)
L’allocation de mémoire varie selon la responsabilité de la fonction : les fonctions de requête obtiennent 1 Go, tandis que les fonctions de traitement de documents reçoivent 2 Go pour un équilibre optimal entre coût et performance. Les configurations de délai d’expiration reflètent les besoins opérationnels : 30 secondes pour les requêtes, 300 secondes pour le traitement des documents.
La concurrence réservée avec 100 instances pour les fonctions de requête élimine l’impact du démarrage à froid sur les opérations critiques destinées aux utilisateurs.
Configuration d’API Gateway
API Gateway sert de point d’entrée unifié de notre système, fournissant des interfaces RESTful avec un routage complet des requêtes. La sécurité et la stabilité proviennent de politiques de limitation du débit, d’authentification et CORS soigneusement configurées.
# API Gateway configuration
endpoints:
- path: /query
method: POST
integration: lambda
rate_limit: 1000/min
auth: IAM
- path: /documents
method: POST
integration: lambda
rate_limit: 100/min
auth: IAM
La mise en cache intelligente au niveau d’API Gateway réduit la charge du backend grâce à une mise en cache stratégique des résultats. La validation des requêtes garantit l’intégrité des paramètres d’entrée, protégeant les systèmes backend contre les requêtes invalides.
Optimisation du CDN CloudFront
CloudFront assure une distribution mondiale du contenu avec une mise en cache sophistiquée des ressources statiques, améliorant considérablement les vitesses d’accès des utilisateurs dans le monde entier. Notre stratégie intègre la séparation dynamique-statique, le routage intelligent et l’optimisation de la mise en cache en périphérie.
# CloudFront cache configuration
cache_behaviors = [
{
'path_pattern': '/api/*',
'ttl': 300, # API response short-term cache
'headers': ['Authorization']
},
{
'path_pattern': '/static/*',
'ttl': 86400, # Static resources long-term cache
'compress': True
}
]
L’optimisation des emplacements périphériques garantit des temps de réponse inférieurs à 100 millisecondes pour les utilisateurs du monde entier grâce à une distribution géographique stratégique.
Optimisation des performances
Stratégies d’atténuation des démarrages à froid
Les démarrages à froid représentent le principal défi de l’architecture serverless, traité grâce à notre approche complète d’optimisation multicouche. Les mécanismes de préchauffage basés sur CloudWatch Events maintiennent la disponibilité de l’environnement d’exécution pour les fonctions critiques.
# Warm-up Lambda configuration
def warm_up_handler(event, context):
if event.get('source') == 'aws.events':
return {'statusCode': 200, 'body': 'warmed up'}
# Normal business logic
return business_logic(event, context)
L’optimisation des dépendances réduit la durée des démarrages à froid grâce à des tailles de packages minimisées et à la sélection de bibliothèques légères. L’initialisation du pool de connexions dans la portée globale élimine la surcharge liée aux connexions répétées.
Provisioned Concurrency pour les fonctions critiques élimine complètement la latence des démarrages à froid. L’ajustement dynamique des instances en fonction des modèles métier optimise l’équilibre entre performances et coûts.
Traitement concurrent intelligent
Notre contrôle de concurrence en couches applique différentes limites selon les besoins en ressources. Les paramètres de concurrence Lambda reflètent les caractéristiques des fonctions : les fonctions de requête légères prennent en charge une forte concurrence tandis que les fonctions de traitement de documents gourmandes en ressources utilisent des limites contrôlées afin d’éviter la contention.
# Concurrency configuration example
functions_config:
query_function:
reserved_concurrency: 100
memory: 1024
document_processing:
reserved_concurrency: 10
memory: 2048
Le traitement asynchrone via SQS et SNS permet le découplage des tâches, évitant les défaillances en cascade des appels synchrones. L’optimisation du traitement par lots agrège des tâches similaires pour une meilleure utilisation des ressources.
Architecture de mise en cache multiniveau
Notre système de mise en cache à trois niveaux optimise différents modèles d’accès :
Le cache L1 utilise la mémoire des fonctions Lambda (TTL de 5 minutes, capacité de 100 Mo), le cache L2 utilise un cluster Redis (TTL d’1 heure, capacité de 1 Go), et le cache L3 s’appuie sur le stockage S3 (TTL d’1 jour, capacité illimitée).
class CacheManager:
def get(self, key):
# L1 cache query
if key in self.memory_cache:
return self.memory_cache[key]
# L2 cache query
value = self.redis_client.get(key)
if value:
self.memory_cache[key] = value
return value
# L3 cache query
return self.s3_cache.get(key)
Le préchauffage prédictif du cache exploite les schémas de requêtes historiques pour un chargement proactif des données. L’invalidation intelligente du cache maintient la cohérence des données grâce à des stratégies à la fois de mises à jour actives et d’expiration passive.
Optimisation stratégique des coûts
La configuration fine des ressources favorise l’optimisation de la facturation à la demande. L’ajustement dynamique des ressources règle automatiquement la mémoire Lambda et les paramètres de délai d’expiration en fonction des schémas de charge en temps réel.
Les Reserved Instances et les Savings Plans pour les charges de travail stables offrent jusqu’à 72 % d’économies sur les coûts de calcul. Les Spot Instances gèrent le traitement par lots non critique pour des réductions de coûts supplémentaires.
# Cost optimization configuration
cost_optimization = {
'lambda_memory_optimization': True,
'auto_scaling': True,
'reserved_capacity': {
'query_functions': 50,
'processing_functions': 5
}
}
Surveillance et opérations complètes
Surveillance des métriques de performance
L’intégration CloudWatch offre une visibilité complète sur les performances à travers ces métriques critiques :
Le temps de réponse de l’API maintient P50 < 1 s, P95 < 3 s, P99 < 5 s, le taux de réussite reste > 99,9 %, les utilisateurs simultanés bénéficient d’une surveillance en temps réel, les performances de récupération vectorielle restent < 200 ms, et le temps de génération LLM demeure < 2 s.
# Custom metrics sending
def send_metrics(metric_name, value, unit='Count'):
cloudwatch = boto3.client('cloudwatch')
cloudwatch.put_metric_data(
Namespace='RAG/System',
MetricData=[{
'MetricName': metric_name,
'Value': value,
'Unit': unit,
'Timestamp': datetime.utcnow()
}]
)
Analyse structurée des journaux
La journalisation structurée au format JSON permet de puissantes capacités d’interrogation et d’analyse. Des journaux complets capturent les identifiants de requête, les horodatages, le contexte utilisateur, les métriques de performance et les informations d’erreur détaillées.
Gestion proactive des défaillances
AWS X-Ray fournit un traçage distribué de bout en bout pour l’identification rapide des goulots d’étranglement de performance. Les systèmes d’alerte automatisés surveillent les métriques clés avec des seuils configurables et des notifications multicanaux.
# Alert rule configuration
alerts = [
{
'metric': 'ResponseTime',
'threshold': 3000, # 3 seconds
'comparison': 'GreaterThanThreshold',
'action': 'sns_notification'
},
{
'metric': 'ErrorRate',
'threshold': 1, # 1%
'comparison': 'GreaterThanThreshold',
'action': 'auto_scaling'
}
]
Les mécanismes d’auto-réparation exploitent les capacités de nouvelle tentative automatique de Lambda et la fonctionnalité Dead Letter Queue (DLQ) pour une récupération autonome après défaillance. La planification de la capacité utilise les données historiques et les projections de croissance pour une mise à l’échelle proactive des ressources.
Du tutoriel à la production : votre système RAG est prêt
Vous venez de créer un système RAG d’entreprise complet qui résout le casse-tête d’intégration bloquant la plupart des projets RAG. Ce n’est pas une énième démonstration localhost : vous disposez d’une infrastructure prête pour la production exécutée sur AWS avec traitement de documents, stockage vectoriel via Zilliz Cloud, recherche hybride et génération LLM via les modèles Bedrock Nova.
La conception modulaire vous permet de déployer immédiatement des fonctionnalités d’IA tout en itérant sur les composants à mesure que les besoins évoluent. Vous vous appuyez sur des modèles d’entreprise éprouvés qui évoluent avec votre activité, plutôt que sur des frameworks expérimentaux qui cèdent sous la charge.
Prêt à aller plus loin ? Déployez le système avec vos données réelles et voyez ce qu’il peut faire. Nous serions ravis de connaître votre expérience et vos modifications.
Dépôt de code complet : https://github.com/yincma/AWS-zilliz-RAG/tree/main
Continuer à lire
Announcing VDBBench 1.0: Open-Source VectorDB Benchmarking with Your Real-World Production Workloads
Discover VDBBench 1.0, an open-source tool for benchmarking vector databases with real-world production data, streaming ingestion, and concurrent workloads.
Vector Databases vs. Object-Relational Databases
Use a vector database for AI-powered similarity search; use an object-relational database for complex data modeling with both relational integrity and object-oriented features.
Bringing AI to Legal Tech: The Role of Vector Databases in Enhancing LLM Guardrails
Discover how vector databases enhance AI reliability in legal tech, ensuring accurate, compliant, and trustworthy AI-powered legal solutions.