Bedrock、Zilliz Cloud、LangChainを使用してAWS上にエンタープライズ対応のRAGパイプラインを構築する方法

Retrieval-Augmented Generation（RAG）は、エンタープライズグレードのLLMソリューションの基盤として急速に普及しており、企業の約90%が、信頼できるドメイン固有の知識でLLMをグラウンディングするために利用しています。しかし現実はより複雑です。RAGエコシステムは選択肢が爆発的に増えています。LLM、埋め込みモデル、オーケストレーションフレームワーク、ベクトルデータベースはそれぞれ独自の統合パターンを持っており、チームはエンタープライズの制約内で機能するソリューションを組み合わせるのに苦労しています。

AWSに深く投資している組織にとって、この課題はいっそう切実です。既存のインフラを単に取り替えたり、確立されたセキュリティおよびコンプライアンスポリシーを迂回したりすることはできません。必要なのは、AWS環境とシームレスに統合され、エンタープライズ対応サービスを活用し、将来にも対応できるRAGアーキテクチャです。

このチュートリアルでは、まさにそれを構築していきます。AWS Bedrock（Nova + Titanモデル）、ベクトルデータベースとしてのZilliz Cloud、そしてオーケストレーションのためのLangChainを使用した、エンタープライズ対応のRAGパイプラインです。最後まで進めると、AWSスタックに直接デプロイできる、実用的で安全かつ本番環境対応の基盤が手に入ります。妥協も回避策も不要です。

スケールに対応するRAGをどう設計するか

コードに入る前に、何を構築するのか、そしてそれが実際のエンタープライズ課題をなぜ解決するのかを理解しましょう。

従来のLLMは、エンタープライズ環境では2つの大きな壁にぶつかります。知識がトレーニング時点で止まっているため、最新のレポート、顧客データ、業界動向にはアクセスできません。さらに、推論過程を追跡する手段がないまま頻繁にハルシネーションを起こします。顧客向けアプリケーションを支えるものとしては、決して望ましいものではありません。

RAGはこの状況を根本から変えます。巨大なモデルを再トレーニングするのではなく、まず関連情報を検索し、そのコンテキストに基づいて回答を生成します。メリットはすぐに現れます。精度が25〜40%向上し、ハルシネーションが60%以上減少し、回答の完全なトレーサビリティが得られます。AIは突然、前四半期の業績を把握し、情報源を引用できるようになります。

私たちのエンタープライズRAGシステムは、実績あるMVC（Model-View-Controller）アーキテクチャパターンに従います。

• Modelレイヤー: ドキュメント処理、埋め込み、ベクトルストレージ、LLM推論といった重い処理を担当します

• Viewレイヤー: ユーザーインターフェースとAPIレスポンスを管理します

• Controllerレイヤー: Lambda関数とイベント処理を通じてワークフローをオーケストレーションします

5つの中核エンジンが実装を支えます。

• Query Processing Engine: ユーザーの質問を最適化された検索クエリに変換します

• Vector Retrieval Engine: Zilliz Cloudのセマンティック検索を使用して関連コンテンツを見つけます

• Reranking Module: 関連性とビジネスルールを使用して結果に優先順位を付けます

• Generation Engine: AWS Bedrockを通じてコンテキストを正確な回答に統合します

• Event-Driven Backbone: Amazon EventBridgeを通じて、すべてを疎結合に保ちます

使用するテクノロジースタック

エンタープライズRAGの構築は、単に個々の「最良の」ツールを選ぶことではありません。AWSエコシステムにシームレスに統合されるテクノロジーを組み合わせることです。このチュートリアルでは、コンピューティングにAWS Lambda、埋め込みと生成にAWS Bedrock（Nova + Titanモデル）、ベクトル検索にZilliz Cloud、オーケストレーションにLangChainを組み合わせます。

AWS Lambda: Elastic Compute Foundation

Lambdaはサーバーレスの基盤を提供します。管理すべきサーバーはなく、ゼロから数千リクエストまで即座にスケールし、実行ごとの課金体系です。各RAGステージ（ドキュメント処理、ベクトル化、検索、生成）は、それぞれ独立したLambda関数として実行されます。この設計により、システムはモジュール化され、耐障害性が高く、コスト効率に優れたものになります。

AWS Bedrock: Flexible Model Hub

Bedrockは、50以上のサーバーレス基盤モデル（さらにAmazon、Anthropic、Metaなどによる120以上のマーケットプレイスオプション）へのアクセスを提供します。その最大の特長は？アプリケーションコードを変更せずにモデルを切り替えられることです。つまり、A/Bテスト、レイテンシとコストの最適化、新しいモデルの採用を、再アーキテクチャなしで行えます。

Zilliz Cloud: 最も高性能なエンタープライズ向けベクトルデータベース

オープンソースのMilvusを基盤とするZilliz Cloudは、データに動的に適応するAutoIndexにより、インデックス調整の推測作業を不要にします。そのCardinal検索エンジンは、従来のベクトルデータベースより最大10倍高速なパフォーマンスを実現しながら、数十億ベクトルまでシームレスにスケールします。これはエンタープライズ規模のデプロイメントに不可欠です。

LangChain: オーケストレーションレイヤー

LangChainは、埋め込み、検索、生成の間の流れを管理し、すべてをつなぎ合わせます。AWSとの統合と柔軟な抽象化により、アーキテクチャをクリーンでモジュール化され、本番環境に対応した状態に保ちます。

スタックが整ったので、実際に手を動かして構築を始めましょう。

AWSでエンタープライズ対応RAGの構築を始める

AWS CDKインフラストラクチャのセットアップ

AWS CDK（Cloud Development Kit）を使用して、Lambda関数、API Gateway、S3バケット、CloudFrontといったスタック全体を定義し、環境間で一貫してデプロイします。CDKを使えば、アプリケーションコードと同じようにインフラストラクチャをバージョン管理し、レビューできます。

# 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"
    }
)

CDK Bootstrapステップでは、デプロイメントアーティファクト用のS3バケット、権限用のIAMロール、設定用のSSMパラメータといった基盤となるAWSリソースを作成します。その後、開発、ステージング、本番環境向けに個別のスタックをデプロイできます。

Zilliz Cloudへの接続

Zilliz Cloudのセットアップには、コレクションの作成、インデックスの最適化、接続の確立という3つのステップがあります。検索精度と速度の最適なバランスを得るために、HNSWインデックスを備えた1024次元ベクトルを使用しています。

# 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}
}

大規模なドキュメントコレクションを扱う際の検索パフォーマンスを向上させるために、パーティションを使用してドキュメントタイプやビジネス領域ごとにデータを整理します。

開発ワークフローの効率化

Makefileは、依存関係のインストール、テストの実行、異なる環境へのデプロイ、リソースのクリーンアップのための統一されたコマンドを提供します。

# Standardized development process
install:  # Install dependencies
test:     # Run tests
lint:     # Code checking
deploy:   # Deploy application
clean:    # Clean environment

CI/CDパイプラインは、コード品質チェック、型検証、自動テストを処理します。

コア機能の構築

ドキュメント処理パイプライン

このパイプラインは、解析、コンテンツクリーニング、インテリジェントなチャンキング、メタデータ抽出という4つの段階でドキュメントを処理します。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

当社のチャンキング戦略では、段落の境界を尊重し、関連情報をまとめて保持する、セマンティクスを考慮したセグメンテーションを使用しています。チャンクサイズはドキュメントの種類に基づいて自動的に調整されます。

ベクトル化とストレージ

AWS Bedrock の Titan Embeddings は、効率化のためにドキュメントをバッチで処理します。ベクトルキャッシュにより、以前に処理されたコンテンツの埋め込みを再計算することを防ぎます。

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

階層型ストレージのアプローチでは、頻繁にアクセスされるベクトルを高速ストレージに保持し、古いコンテンツをコスト最適化された階層にアーカイブします。

ハイブリッド検索戦略

ベクトル類似度とキーワードマッチングを、多段階プロセスを通じて組み合わせています。まず広範な初期検索を行い、その後、最良の結果を得るために精密なランキングを行います。

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]

Cross-Encoder モデルがリランキングを処理し、最も関連性の高い結果を特定します。コンテキストウィンドウの最適化により、各クエリに対して適切な量の情報を取得できます。

LangChain 統合

LangChain は、LangChain Hub の実証済み RAG テンプレートを使用して、検索と生成のプロセスをオーケストレーションします。

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})

メモリ管理により、複数ターンのディスカッションにおける会話コンテキストを維持します。ストリーミング応答により、完全な回答を待つのではなく、結果をリアルタイムで表示します。エラーハンドリングにより、問題が発生した際にも円滑に復旧できます。

サーバーレスアーキテクチャのデプロイ

Lambda 関数設計の卓越性

各 Lambda 関数は単一責任の原則に従い、特定のビジネスロジックに集中します。当社のコア機能であるドキュメント処理、ベクトル化、検索、生成は、最大限の柔軟性を実現するために独立してデプロイおよびスケールします。

# 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)

メモリ割り当ては関数の責任に応じて異なります。クエリ関数には 1GB、ドキュメント処理関数には最適なコストパフォーマンスバランスのために 2GB を割り当てます。タイムアウト設定は運用上のニーズを反映しており、クエリには 30 秒、ドキュメント処理には 300 秒を設定します。

クエリ関数に 100 インスタンスの予約済み同時実行を設定することで、重要なユーザー向け操作におけるコールドスタートの影響を排除します。

API Gateway の設定

API Gateway は、包括的なリクエストルーティングを備えた RESTful インターフェースを提供する、当社の統一システムエントリーポイントとして機能します。セキュリティと安定性は、慎重に設定されたレート制限、認証、CORS ポリシーによって実現されます。

# 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

API Gateway レイヤーでのインテリジェントキャッシュにより、戦略的な結果キャッシュを通じてバックエンドの負荷を軽減します。リクエスト検証は入力パラメータの整合性を確保し、無効なリクエストからバックエンドシステムを保護します。

CloudFront CDN 最適化

CloudFront は、高度な静的リソースキャッシュを備えたグローバルコンテンツ配信を提供し、世界中のユーザーアクセス速度を大幅に向上させます。当社の戦略には、動的・静的の分離、インテリジェントルーティング、エッジキャッシュ最適化が組み込まれています。

# 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
    }
]

エッジロケーションの最適化により、戦略的な地理的分散を通じて、グローバルユーザーに対して 100 ミリ秒未満のレスポンスタイムを実現します。

パフォーマンス最適化

コールドスタート緩和戦略

コールドスタートはサーバーレスアーキテクチャの主要な課題であり、当社の包括的な多層最適化アプローチによって対処されています。CloudWatch Events ベースのウォームアップメカニズムにより、重要な関数の実行環境の準備状態を維持します。

# 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)

依存関係の最適化により、パッケージサイズの最小化と軽量ライブラリの選定を通じて、コールドスタート時間を短縮します。グローバルスコープでのコネクションプール初期化により、繰り返し発生する接続オーバーヘッドを排除します。

重要な関数に対する Provisioned Concurrency は、コールドスタートレイテンシを完全に排除します。ビジネスパターンに基づく動的なインスタンス調整により、パフォーマンスとコストのバランスを最適化します。

インテリジェントな並行処理

当社の階層型同時実行制御は、リソース要件に基づいて異なる制限を適用します。Lambda の同時実行設定は関数の特性を反映しています。軽量なクエリ関数は高い同時実行性をサポートする一方、リソース集約型のドキュメント処理関数は、競合を防ぐために制御された制限を使用します。

# Concurrency configuration example
functions_config:
  query_function:
    reserved_concurrency: 100
    memory: 1024
  document_processing:
    reserved_concurrency: 10
    memory: 2048

SQS と SNS を通じた非同期処理によりタスクの疎結合を実現し、同期呼び出しの連鎖的な失敗を防ぎます。バッチ処理の最適化では、類似タスクを集約してリソース利用率を高めます。

マルチティアキャッシュアーキテクチャ

当社の 3 層キャッシュシステムは、さまざまなアクセスパターンに対して最適化されています。

L1 Cache は Lambda 関数メモリ（5 分 TTL、100MB 容量）を使用し、L2 Cache は Redis クラスター（1 時間 TTL、1GB 容量）を採用し、L3 Cache は S3 ストレージ（1 日 TTL、無制限容量）を活用します。

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)

予測キャッシュウォーミングは、履歴クエリパターンを活用してプロアクティブなデータ読み込みを行います。インテリジェントなキャッシュ無効化は、アクティブな更新とパッシブな有効期限切れ戦略の両方を通じてデータ整合性を維持します。

戦略的コスト最適化

きめ細かなリソース設定により、オンデマンド課金の最適化を推進します。動的なリソース調整は、リアルタイムの負荷パターンに基づいて Lambda のメモリとタイムアウト設定を自動的にチューニングします。

安定したワークロード向けの Reserved Instances と Savings Plans は、最大 72% のコンピューティングコスト削減を実現します。Spot Instances は、重要度の低いバッチ処理を担い、さらなるコスト削減を実現します。

# Cost optimization configuration
cost_optimization = {
    'lambda_memory_optimization': True,
    'auto_scaling': True,
    'reserved_capacity': {
        'query_functions': 50,
        'processing_functions': 5
    }
}

包括的な監視と運用

パフォーマンスメトリクスの監視

CloudWatch 連携により、これらの重要なメトリクス全体にわたる完全なパフォーマンス可視性を提供します。

API レスポンスタイムは P50 < 1s、P95 < 3s、P99 < 5s を維持し、Success Rate は > 99.9% を維持し、Concurrent Users はリアルタイム監視を受け、Vector Retrieval Performance は < 200ms を維持し、LLM Generation Time は < 2s のままです。

# 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()
        }]
    )

構造化ログ分析

JSON 形式の構造化ログにより、強力なクエリと分析機能が可能になります。包括的なログは、リクエスト ID、タイムスタンプ、ユーザーコンテキスト、パフォーマンスメトリクス、詳細なエラー情報を取得します。

プロアクティブな障害管理

AWS X-Ray は、パフォーマンスのボトルネックを迅速に特定するためのエンドツーエンドの分散トレーシングを提供します。自動アラートシステムは、設定可能なしきい値とマルチチャネル通知により主要メトリクスを監視します。

# 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'
    }
]

セルフヒーリングメカニズムは、Lambda の自動リトライ機能と Dead Letter Queue (DLQ) 機能を活用して、自律的な障害復旧を行います。キャパシティプランニングは、履歴データと成長予測を使用して、プロアクティブなリソーススケーリングを行います。

チュートリアルから本番環境へ: あなたの RAG システムは準備完了

あなたは今、ほとんどの RAG プロジェクトを停止させる統合の頭痛の種を解決する、完全なエンタープライズ RAG システムを構築しました。これは単なる localhost デモではありません。AWS 上で動作する本番対応インフラストラクチャがあり、ドキュメント処理、Zilliz Cloud を通じたベクトルストレージ、ハイブリッド検索、Bedrock Nova モデルによる LLM 生成を備えています。

モジュラー設計により、ニーズの進化に合わせてコンポーネントを反復改善しながら、AI 機能をすぐにデプロイできます。負荷で壊れる実験的なフレームワークではなく、ビジネスとともにスケールする実証済みのエンタープライズパターンの上に構築しています。

さらに進める準備はできていますか？ 実際のデータでシステムをデプロイし、その可能性を確認してください。あなたの体験や変更点についてぜひお聞かせください。

完全なコードリポジトリ: https://github.com/yincma/AWS-zilliz-RAG/tree/main