話題の組み込み型のベクトルデータベース LanceDB を Amazon S3 と Amazon Bedrock で試してみた

[
  {"id": "doc-01", "title": "Amazon S3", "text": "Amazon S3 は、業界をリードするスケーラビリティ、データ可用性、セキュリティ、パフォーマンスを提供するオブジェクトストレージサービスです。データレイク、バックアップ、アーカイブなど幅広い用途で利用されます。"},
  {"id": "doc-02", "title": "AWS Lambda", "text": "AWS Lambda は、サーバーのプロビジョニングや管理なしでコードを実行できるサーバーレスコンピューティングサービスです。イベント駆動型のアプリケーション構築に適しています。"},
  {"id": "doc-03", "title": "Amazon EC2", "text": "Amazon EC2 は、クラウド上で仮想サーバーを提供するサービスです。多様なインスタンスタイプから用途に応じた計算リソースを選択できます。"},
  {"id": "doc-04", "title": "Amazon DynamoDB", "text": "Amazon DynamoDB は、フルマネージドの NoSQL データベースサービスです。1桁ミリ秒のレイテンシで、あらゆる規模のアプリケーションに対応します。"},
  {"id": "doc-05", "title": "Amazon RDS", "text": "Amazon RDS は、リレーショナルデータベースをクラウド上で簡単にセットアップ、運用、スケーリングできるマネージドサービスです。MySQL、PostgreSQL、Oracle などに対応しています。"},
  {"id": "doc-06", "title": "Amazon Redshift", "text": "Amazon Redshift は、クラウドデータウェアハウスサービスです。ペタバイト規模のデータに対して高速な分析クエリを実行できます。Serverless オプションも提供されています。"},
  {"id": "doc-07", "title": "Amazon Athena", "text": "Amazon Athena は、S3 上のデータに対して標準 SQL でクエリを実行できるサーバーレスのインタラクティブ分析サービスです。スキャンしたデータ量に応じた課金体系です。"},
  {"id": "doc-08", "title": "AWS Glue", "text": "AWS Glue は、サーバーレスのデータ統合サービスです。ETL ジョブの実行、データカタログの管理、データ品質の検証などが可能です。"},
  {"id": "doc-09", "title": "Amazon Bedrock", "text": "Amazon Bedrock は、主要な AI 企業が提供する高性能な基盤モデルを単一の API で利用できるフルマネージドサービスです。テキスト生成や埋め込みベクトルの生成が可能です。"},
  {"id": "doc-10", "title": "Amazon SageMaker", "text": "Amazon SageMaker は、機械学習モデルの構築、トレーニング、デプロイを支援するフルマネージドサービスです。データサイエンティストと開発者向けの統合開発環境を提供します。"},
  {"id": "doc-11", "title": "Amazon OpenSearch Service", "text": "Amazon OpenSearch Service は、ログ分析や全文検索、ベクトル検索を実行できるマネージド検索サービスです。k-NN 検索によるセマンティック検索にも対応しています。"},
  {"id": "doc-12", "title": "Amazon Kinesis Data Streams", "text": "Amazon Kinesis Data Streams は、ストリーミングデータをリアルタイムに収集・処理できるサービスです。大規模なデータストリームを低レイテンシで扱えます。"},
  {"id": "doc-13", "title": "Amazon QuickSight", "text": "Amazon QuickSight は、クラウドネイティブな BI サービスです。ダッシュボードの作成や、自然言語による質問応答機能 Amazon Q in QuickSight を利用できます。"},
  {"id": "doc-14", "title": "AWS Step Functions", "text": "AWS Step Functions は、分散アプリケーションのワークフローを視覚的に構築できるサーバーレスオーケストレーションサービスです。Lambda などの AWS サービスを連携させられます。"},
  {"id": "doc-15", "title": "Amazon API Gateway", "text": "Amazon API Gateway は、REST API や WebSocket API を作成、公開、管理できるフルマネージドサービスです。Lambda と組み合わせたサーバーレス API の構築によく使われます。"},
  {"id": "doc-16", "title": "Amazon CloudFront", "text": "Amazon CloudFront は、低レイテンシでコンテンツを配信するグローバルな CDN サービスです。エッジロケーションを利用して静的・動的コンテンツを高速配信します。"},
  {"id": "doc-17", "title": "Amazon VPC", "text": "Amazon VPC は、AWS クラウド内に論理的に分離されたプライベートネットワーク空間を構築するサービスです。サブネットやルートテーブルを自由に設計できます。"},
  {"id": "doc-18", "title": "AWS IAM", "text": "AWS IAM は、AWS リソースへのアクセスを安全に管理するサービスです。ユーザー、グループ、ロール、ポリシーによるきめ細かなアクセス制御が可能です。"},
  {"id": "doc-19", "title": "Amazon CloudWatch", "text": "Amazon CloudWatch は、AWS リソースとアプリケーションのモニタリングサービスです。メトリクスの収集、ログの分析、アラームの設定ができます。"},
  {"id": "doc-20", "title": "AWS CloudFormation", "text": "AWS CloudFormation は、インフラをコードとして管理する IaC サービスです。テンプレートを使って AWS リソースをプロビジョニングできます。"},
  {"id": "doc-21", "title": "Amazon ECS", "text": "Amazon ECS は、コンテナ化されたアプリケーションを実行するフルマネージドなコンテナオーケストレーションサービスです。Fargate と組み合わせるとサーバーレスでコンテナを実行できます。"},
  {"id": "doc-22", "title": "Amazon EKS", "text": "Amazon EKS は、マネージドな Kubernetes サービスです。Kubernetes コントロールプレーンの運用を AWS に任せて、コンテナワークロードを実行できます。"},
  {"id": "doc-23", "title": "AWS Fargate", "text": "AWS Fargate は、サーバーを管理せずにコンテナを実行できるサーバーレスコンピューティングエンジンです。ECS および EKS で利用できます。"},
  {"id": "doc-24", "title": "Amazon SQS", "text": "Amazon SQS は、フルマネージドのメッセージキューイングサービスです。マイクロサービス間の疎結合な非同期通信を実現します。"},
  {"id": "doc-25", "title": "Amazon SNS", "text": "Amazon SNS は、フルマネージドの Pub/Sub メッセージングサービスです。アプリケーション間通知やモバイルプッシュ通知に利用できます。"},
  {"id": "doc-26", "title": "Amazon EventBridge", "text": "Amazon EventBridge は、イベント駆動型アーキテクチャを構築するためのサーバーレスイベントバスサービスです。SaaS アプリケーションとの連携も可能です。"},
  {"id": "doc-27", "title": "AWS KMS", "text": "AWS KMS は、暗号化キーの作成と管理を行うマネージドサービスです。多くの AWS サービスと統合されており、保管時のデータ暗号化に利用されます。"},
  {"id": "doc-28", "title": "Amazon Aurora", "text": "Amazon Aurora は、クラウド向けに構築された MySQL および PostgreSQL 互換のリレーショナルデータベースです。高い性能と可用性を提供し、Serverless v2 にも対応しています。"},
  {"id": "doc-29", "title": "Amazon ElastiCache", "text": "Amazon ElastiCache は、Redis および Memcached 互換のインメモリキャッシュサービスです。データベースの負荷軽減やセッション管理に利用されます。"},
  {"id": "doc-30", "title": "AWS Secrets Manager", "text": "AWS Secrets Manager は、データベース認証情報や API キーなどのシークレットを安全に管理するサービスです。シークレットの自動ローテーションに対応しています。"},
  {"id": "doc-31", "title": "Amazon S3 Tables", "text": "Amazon S3 Tables は、Apache Iceberg テーブルに最適化されたストレージを提供する S3 の機能です。分析ワークロード向けにテーブルデータを効率的に管理できます。"},
  {"id": "doc-32", "title": "AWS Lake Formation", "text": "AWS Lake Formation は、セキュアなデータレイクを構築・管理するサービスです。テーブルや列レベルのきめ細かなアクセス制御を実現します。"},
  {"id": "doc-33", "title": "Amazon EMR", "text": "Amazon EMR は、Apache Spark や Hive などのビッグデータフレームワークを実行するマネージドクラスタープラットフォームです。Serverless オプションも利用できます。"},
  {"id": "doc-34", "title": "AWS Batch", "text": "AWS Batch は、バッチコンピューティングジョブを効率的に実行するフルマネージドサービスです。ジョブのスケジューリングとコンピューティングリソースの管理を自動化します。"},
  {"id": "doc-35", "title": "Amazon Cognito", "text": "Amazon Cognito は、Web およびモバイルアプリケーションの認証、認可、ユーザー管理を提供するサービスです。ソーシャル ID プロバイダーとの連携も可能です。"},
  {"id": "doc-36", "title": "AWS AppSync", "text": "AWS AppSync は、GraphQL API を構築するフルマネージドサービスです。リアルタイムデータ同期やオフライン対応のアプリケーション開発を支援します。"},
  {"id": "doc-37", "title": "Amazon MSK", "text": "Amazon MSK は、Apache Kafka 互換のフルマネージドストリーミングサービスです。Kafka クラスターの運用負荷を軽減し、ストリーミングアプリケーションを構築できます。"},
  {"id": "doc-38", "title": "AWS Amplify", "text": "AWS Amplify は、フルスタックの Web・モバイルアプリケーションを迅速に構築、デプロイできる開発プラットフォームです。ホスティングやバックエンド構築を統合的に支援します。"},
  {"id": "doc-39", "title": "Amazon Neptune", "text": "Amazon Neptune は、フルマネージドのグラフデータベースサービスです。ソーシャルネットワークやレコメンデーションなど、関係性の分析に適しています。Neptune Analytics ではベクトル検索もサポートしています。"},
  {"id": "doc-40", "title": "Amazon S3 Vectors", "text": "Amazon S3 Vectors は、ベクトルデータの保存とクエリをネイティブにサポートする S3 の機能です。専用のベクトルデータベースを運用せずに、低コストでベクトル検索を実現します。"}
]

import json
import os
import time

import boto3
import lancedb
import pyarrow as pa

REGION = "ap-northeast-1"
DB_URI = os.environ["LANCEDB_URI"]  # 例: s3://lancedb-blog-xxxx/db
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)

def embed(text: str) -> list[float]:
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=EMBED_MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

def main():
    with open("sample_docs.json") as f:
        docs = json.load(f)

    print(f"埋め込み生成: {len(docs)} 件")
    start = time.perf_counter()
    data = [
        {
            "id": doc["id"],
            "title": doc["title"],
            "text": doc["text"],
            "vector": embed(doc["text"]),
        }
        for doc in docs
    ]
    print(f"埋め込み生成完了: {time.perf_counter() - start:.1f} 秒")

    schema = pa.schema(
        [
            pa.field("id", pa.string()),
            pa.field("title", pa.string()),
            pa.field("text", pa.string()),
            pa.field("vector", pa.list_(pa.float32(), 1024)),
        ]
    )

    db = lancedb.connect(DB_URI, storage_options={"region": REGION})
    start = time.perf_counter()
    table = db.create_table("aws_services", data=data, schema=schema, mode="overwrite")
    print(f"テーブル作成完了: {time.perf_counter() - start:.1f} 秒")
    print(f"登録件数: {table.count_rows()}")

if __name__ == "__main__":
    main()

"""S3 上の LanceDB テーブルに対してベクトル検索（インデックスなし）を実行する"""
import json
import os
import time

import boto3
import lancedb

REGION = "ap-northeast-1"
DB_URI = os.environ["LANCEDB_URI"]
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)

def embed(text: str) -> list[float]:
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=EMBED_MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

def main():
    db = lancedb.connect(DB_URI, storage_options={"region": REGION})
    table = db.open_table("aws_services")

    queries = [
        "サーバーを管理せずにコンテナを実行したい",
        "ログを集めて分析、監視したい",
        "ベクトル検索ができるサービスは？",
    ]
    for query in queries:
        qvec = embed(query)
        start = time.perf_counter()
        results = table.search(qvec).limit(3).to_pandas()
        elapsed = (time.perf_counter() - start) * 1000

        print(f"\nクエリ: {query}（検索時間: {elapsed:.0f} ms ※参考値）")
        for _, row in results.iterrows():
            print(f"  {row['_distance']:.4f}  {row['title']}: {row['text'][:45]}...")

if __name__ == "__main__":
    main()

import os

import lancedb

REGION = "ap-northeast-1"
DB_URI = os.environ["LANCEDB_URI"]

def main():
    db = lancedb.connect(DB_URI, storage_options={"region": REGION})
    table = db.open_table("aws_services")
    print(f"件数: {table.count_rows()}")

    try:
        table.create_index(metric="l2", num_partitions=4, num_sub_vectors=16)
        print("インデックス作成: 成功")
        print(table.list_indices())
    except Exception as e:
        print(f"インデックス作成: 失敗 ({type(e).__name__})")
        print(e)

if __name__ == "__main__":
    main()

import json
import os

import boto3
import lancedb

REGION = "ap-northeast-1"
DB_URI = os.environ["LANCEDB_URI"]
EMBED_MODEL_ID = "amazon.titan-embed-text-v2:0"
CHAT_MODEL_ID = "jp.anthropic.claude-haiku-4-5-20251001-v1:0"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)

def embed(text: str) -> list[float]:
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=EMBED_MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

def main():
    db = lancedb.connect(DB_URI, storage_options={"region": REGION})
    table = db.open_table("aws_services")

    question = "コストを抑えてベクトル検索を実現できる AWS のサービスや構成を教えてください"

    results = table.search(embed(question)).limit(3).to_pandas()
    context = "\n\n".join(
        f"【{row['title']}】{row['text']}" for _, row in results.iterrows()
    )

    prompt = (
        "以下の参考ドキュメントの内容に基づいて質問に回答してください。\n"
        "参考ドキュメントに記載のない情報は推測せず、その旨を明記してください。\n\n"
        f"# 参考ドキュメント\n{context}\n\n# 質問\n{question}"
    )

    res = bedrock.converse(
        modelId=CHAT_MODEL_ID,
        messages=[{"role": "user", "content": [{"text": prompt}]}],
        inferenceConfig={"maxTokens": 600, "temperature": 0},
    )

    print(f"質問: {question}\n")
    print("検索でヒットしたドキュメント:")
    for _, row in results.iterrows():
        print(f"  - {row['title']} (distance: {row['_distance']:.4f})")
    print("\n回答:")
    print(res["output"]["message"]["content"][0]["text"])
    usage = res["usage"]
    print(f"\ntokens: input={usage['inputTokens']}, output={usage['outputTokens']}")

if __name__ == "__main__":
    main()