[ユーザーの質問]
   ↓
[Bedrock Titan Embeddings V2] → クエリを埋め込みベクトル化
   ↓
[DynamoDB SearchVectors API] → 類似ドキュメントを取得
   ↓
[Claude（Bedrock）] → 取得結果を文脈として回答生成
   ↓
[ユーザーへの回答]

① ユーザーの質問 → 埋め込みモデルで数値ベクトル化
② ベクトル DB に類似検索 → 関連ドキュメントを取得
③ 取得したドキュメントを LLM のプロンプトに含める
④ LLM が文脈を踏まえた回答を生成

import os
import boto3
from moto import mock_aws

# ダミー認証情報を設定（moto モック使用時に必要）
os.environ.setdefault("AWS_ACCESS_KEY_ID", "testing")
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "testing")
os.environ.setdefault("AWS_SECURITY_TOKEN", "testing")
os.environ.setdefault("AWS_SESSION_TOKEN", "testing")
os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")

@mock_aws
def main():
    dynamodb = boto3.client("dynamodb", region_name="us-east-1")

    response = dynamodb.create_table(
        TableName="RagDocuments",
        KeySchema=[
            {"AttributeName": "doc_id", "KeyType": "HASH"},
            {"AttributeName": "chunk_id", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "doc_id", "AttributeType": "S"},
            {"AttributeName": "chunk_id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",  # サーバーレス課金（オンデマンド）
    )

    print("テーブル作成ステータス:", response["TableDescription"]["TableStatus"])

main()

import math
import random

def generate_embedding(text: str, dimensions: int = 1536) -> list[float]:
    """モック版: Bedrock 不要。テキストから決定論的なL2正規化ベクトルを生成する"""
    # hash を seed にすることで同じテキストは常に同じベクトルを返す
    rng = random.Random(hash(text) % (2 ** 32))
    raw = [rng.gauss(0, 1) for _ in range(dimensions)]
    # normalize=True 相当: L2 正規化
    norm = math.sqrt(sum(v ** 2 for v in raw))
    return [v / norm for v in raw]

# 動作確認
test_embedding = generate_embedding("DynamoDBのベクトル検索とは")
print(f"次元数: {len(test_embedding)}")  # → 1536
print(f"先頭5次元: {test_embedding[:5]}")

import boto3
import json
from decimal import Decimal

dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
table = dynamodb.Table("RagDocuments")

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

def generate_embedding(text: str, dimensions: int = 1536) -> list[float]:
    """Bedrock Titan Embeddings V2 で埋め込みを生成する"""
    body = json.dumps({
        "inputText": text,
        "dimensions": dimensions,  # 256 / 512 / 1536 から選択
        "normalize": True,          # Cosine 類似度の精度向上に有効
    })
    
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=body,
        contentType="application/json",
        accept="application/json",
    )
    
    result = json.loads(response["body"].read())
    return result["embedding"]

def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """テキストをオーバーラップ付きでチャンク分割する"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start += chunk_size - overlap  # オーバーラップ分だけ戻る
    return chunks

def put_document(doc_id: str, title: str, category: str, full_text: str):
    """ドキュメントをチャンク分割してDynamoDBに保存する"""
    chunks = chunk_text(full_text)
    
    with table.batch_writer() as batch:
        for i, chunk in enumerate(chunks):
            embedding = generate_embedding(chunk)
            
            batch.put_item(Item={
                "doc_id": doc_id,
                "chunk_id": f"{i:04d}",  # "0000", "0001", ...
                "content": chunk,
                "title": title,
                "category": category,
                "created_at": "2026-08-05T00:00:00Z",
                # DynamoDB の数値型は Decimal で保存する
                "embedding": [Decimal(str(v)) for v in embedding],
            })
    
    print(f"✅ {doc_id}: {len(chunks)} チャンクを保存しました")

# 実行例
put_document(
    doc_id="doc-001",
    title="DynamoDB ネイティブベクトル検索ガイド",
    category="aws",
    full_text="Amazon DynamoDB は 2026年8月にネイティブベクトル検索をサポートしました..."
)

{
  "TableName": "RagDocuments",
  "VectorSecondaryIndexUpdates": [
    {
      "Create": {
        "IndexName": "EmbeddingIdx",
        "VectorAttribute": {
          "AttributeName": "embedding",
          "Dimensions": 1536
        },
        "IndexOptions": {
          "SimilarityFunction": "COSINE"
        },
        "FilterableAttributes": [
          { "AttributeName": "category" },
          { "AttributeName": "created_at" }
        ]
      }
    }
  ]
}

import boto3

dynamodb = boto3.client("dynamodb", region_name="us-east-1")

response = dynamodb.update_table(
    TableName="RagDocuments",
    VectorSecondaryIndexUpdates=[
        {
            "Create": {
                "IndexName": "EmbeddingIdx",
                "VectorAttribute": {
                    "AttributeName": "embedding",
                    "Dimensions": 1536,
                },
                "IndexOptions": {
                    "SimilarityFunction": "COSINE",
                },
                "FilterableAttributes": [
                    {"AttributeName": "category"},
                    {"AttributeName": "created_at"},
                ],
            }
        }
    ],
)

print("インデックス作成リクエスト送信完了")

import time

def wait_for_index_active(table_name: str, index_name: str, timeout_sec: int = 600):
    """ベクトルインデックスが ACTIVE になるまで待機する"""
    dynamodb = boto3.client("dynamodb", region_name="us-east-1")
    elapsed = 0
    
    while elapsed < timeout_sec:
        response = dynamodb.describe_table(TableName=table_name)
        indexes = response["Table"].get("VectorSecondaryIndexes", [])
        
        for idx in indexes:
            if idx["IndexName"] == index_name:
                status = idx["IndexStatus"]
                print(f"  インデックスステータス: {status} ({elapsed}秒経過)")
                if status == "ACTIVE":
                    print("✅ インデックスが ACTIVE になりました")
                    return True
        
        time.sleep(15)
        elapsed += 15
    
    print("⚠️ タイムアウト。コンソールで手動確認してください")
    return False

wait_for_index_active("RagDocuments", "EmbeddingIdx")

import boto3
import json
from decimal import Decimal

dynamodb = boto3.client("dynamodb", region_name="us-east-1")
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")

def search_similar_documents(
    query_text: str,
    top_k: int = 5,
    category_filter: str | None = None,
) -> list[dict]:
    """
    クエリテキストに意味的に近いドキュメントチャンクを返す
    """
    # Step 1: クエリを埋め込みベクトルに変換
    body = json.dumps({
        "inputText": query_text,
        "dimensions": 1536,
        "normalize": True,
    })
    embed_response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=body,
        contentType="application/json",
        accept="application/json",
    )
    query_vector = json.loads(embed_response["body"].read())["embedding"]
    
    # Step 2: SearchVectors API でベクトル検索
    search_params = {
        "TableName": "RagDocuments",
        "IndexName": "EmbeddingIdx",
        "QueryVector": [Decimal(str(v)) for v in query_vector],
        "TopK": top_k,
    }
    
    # カテゴリフィルタを追加（省略可）
    if category_filter:
        search_params["FilterExpression"] = "category = :cat"
        search_params["ExpressionAttributeValues"] = {
            ":cat": {"S": category_filter}
        }
    
    response = dynamodb.search_vectors(**search_params)
    
    # Step 3: 結果を整形して返す
    results = []
    for item in response.get("Items", []):
        results.append({
            "doc_id": item["doc_id"]["S"],
            "chunk_id": item["chunk_id"]["S"],
            "title": item["title"]["S"],
            "content": item["content"]["S"],
            "score": float(item["_similarity_score"]["N"]),  # 類似度スコア
        })
    
    return results

# 実行例
results = search_similar_documents("DynamoDBのベクトル検索の料金体系は？", top_k=3)
for r in results:
    print(f"スコア: {r['score']:.4f} | タイトル: {r['title']}")
    print(f"  内容: {r['content'][:100]}...")
    print()

スコア: 0.9123 | タイトル: DynamoDB ネイティブベクトル検索ガイド
  内容: Amazon DynamoDB のベクトル検索は PAY_PER_REQUEST モードで...

スコア: 0.8741 | タイトル: AWS サービス料金まとめ
  内容: DynamoDB の料金は読み書きリクエスト数に基づいており...

スコア: 0.7654 | タイトル: サーバーレスアーキテクチャ設計
  内容: コスト最適化の観点から、オンデマンド課金モードは...

# カテゴリと日付で絞り込む例
search_params["FilterExpression"] = "category = :cat AND created_at >= :date"
search_params["ExpressionAttributeValues"] = {
    ":cat": {"S": "aws"},
    ":date": {"S": "2026-01-01T00:00:00Z"},
}

┌─────────────────────────────────────────────────────┐
│                     クライアント                      │
└─────────────────────┬───────────────────────────────┘
                      │ HTTP リクエスト
                      ▼
┌─────────────────────────────────────────────────────┐
│            AWS Lambda + API Gateway                  │
│  ┌────────────────────────────────────────────────┐  │
│  │  1. クエリ受信                                  │  │
│  │  2. Bedrock Titan Embeddings → 埋め込み生成     │  │
│  │  3. DynamoDB SearchVectors → 類似ドキュメント取得│  │
│  │  4. プロンプト構築（文脈注入）                   │  │
│  │  5. Bedrock Claude → 回答生成                   │  │
│  └────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────┘
                      │
          ┌───────────┴───────────┐
          ▼                       ▼
┌─────────────────┐    ┌──────────────────────┐
│  Amazon Bedrock  │    │  Amazon DynamoDB      │
│  (Embeddings +   │    │  (データ + ベクトル   │
│   Claude LLM)   │    │   インデックス)       │
└─────────────────┘    └──────────────────────┘

def build_rag_prompt(query: str, search_results: list[dict]) -> str:
    """検索結果をコンテキストとして含むプロンプトを構築する"""
    
    context_parts = []
    for i, result in enumerate(search_results, 1):
        context_parts.append(
            f"[ドキュメント {i}]\n"
            f"タイトル: {result['title']}\n"
            f"内容: {result['content']}\n"
        )
    
    context = "\n---\n".join(context_parts)
    
    return f"""以下の参考ドキュメントを基に、質問に答えてください。
参考ドキュメントに記載のない内容については、「提供された情報には記載がありません」と答えてください。

# 参考ドキュメント
{context}

# 質問
{query}

# 回答（回答の最後に「参照：[タイトル名]」の形式で出典を記載してください）"""

#!/usr/bin/env python3
"""
RAGドキュメント投入スクリプト
使い方: python ingest.py
"""
import boto3
import json
from decimal import Decimal

REGION = "us-east-1"
TABLE_NAME = "RagDocuments"

dynamodb = boto3.resource("dynamodb", region_name=REGION)
bedrock = boto3.client("bedrock-runtime", region_name=REGION)
table = dynamodb.Table(TABLE_NAME)

# サンプルドキュメント
SAMPLE_DOCS = [
    {
        "doc_id": "doc-001",
        "title": "DynamoDB ベクトル検索 入門",
        "category": "aws",
        "content": """Amazon DynamoDB は 2026年8月、ネイティブベクトル検索を正式サポートしました。
これにより、既存の DynamoDB テーブルに埋め込みベクトルを保存し、
SearchVectors API を使って意味的類似検索ができるようになりました。
料金体系は通常の DynamoDB と同じく PAY_PER_REQUEST が選択でき、
アイドル時のコストは発生しません。""",
    },
    {
        "doc_id": "doc-002",
        "title": "Bedrock Titan Embeddings の使い方",
        "category": "aws",
        "content": """Amazon Bedrock の Titan Embeddings V2 は、テキストを高精度な
埋め込みベクトルに変換するモデルです。256・512・1536 次元を選択できます。
InvokeModel API を通じて呼び出し、normalize=True を指定することで
コサイン類似度での検索精度が向上します。""",
    },
]

def generate_embedding(text: str) -> list[float]:
    body = json.dumps({"inputText": text, "dimensions": 1536, "normalize": True})
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=body,
        contentType="application/json",
        accept="application/json",
    )
    return json.loads(response["body"].read())["embedding"]

def ingest_document(doc: dict):
    embedding = generate_embedding(doc["content"])
    table.put_item(Item={
        "doc_id": doc["doc_id"],
        "chunk_id": "0000",
        "title": doc["title"],
        "category": doc["category"],
        "content": doc["content"],
        "embedding": [Decimal(str(v)) for v in embedding],
    })
    print(f"✅ 保存完了: {doc['title']}")

if __name__ == "__main__":
    for doc in SAMPLE_DOCS:
        ingest_document(doc)
    print("\n全ドキュメントの投入が完了しました！")

#!/usr/bin/env python3
"""
RAG質問応答スクリプト
使い方: python rag_query.py "DynamoDBのベクトル検索の料金は？"
"""
import sys
import boto3
import json
from decimal import Decimal

REGION = "us-east-1"
TABLE_NAME = "RagDocuments"
INDEX_NAME = "EmbeddingIdx"

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

def generate_embedding(text: str) -> list[float]:
    body = json.dumps({"inputText": text, "dimensions": 1536, "normalize": True})
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=body, contentType="application/json", accept="application/json",
    )
    return json.loads(response["body"].read())["embedding"]

def search_documents(query_text: str, top_k: int = 5) -> list[dict]:
    query_vector = generate_embedding(query_text)
    response = dynamodb.search_vectors(
        TableName=TABLE_NAME,
        IndexName=INDEX_NAME,
        QueryVector=[Decimal(str(v)) for v in query_vector],
        TopK=top_k,
    )
    results = []
    for item in response.get("Items", []):
        results.append({
            "title": item["title"]["S"],
            "content": item["content"]["S"],
            "score": float(item["_similarity_score"]["N"]),
        })
    return results

def generate_answer(query: str, context_docs: list[dict]) -> str:
    context = "\n---\n".join(
        f"[{i+1}] {d['title']}\n{d['content']}" for i, d in enumerate(context_docs)
    )
    prompt = f"""参考ドキュメント:
{context}

質問: {query}

上記の参考ドキュメントに基づいて回答してください。
回答の最後に「参照: [タイトル]」の形式で出典を記載してください。"""
    
    body = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": prompt}],
    })
    response = bedrock.invoke_model(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        body=body, contentType="application/json", accept="application/json",
    )
    result = json.loads(response["body"].read())
    return result["content"][0]["text"]

if __name__ == "__main__":
    query = sys.argv[1] if len(sys.argv) > 1 else "DynamoDBのベクトル検索とは？"
    print(f"🔍 質問: {query}\n")
    
    docs = search_documents(query)
    print(f"📄 {len(docs)} 件のドキュメントを取得しました\n")
    
    answer = generate_answer(query, docs)
    print(f"💬 回答:\n{answer}")

from langchain_core.retrievers import BaseRetriever
from langchain_core.documents import Document
from langchain_core.callbacks import CallbackManagerForRetrieverRun

class DynamoDBVectorRetriever(BaseRetriever):
    """DynamoDB SearchVectors を使った LangChain カスタム Retriever"""
    
    table_name: str
    index_name: str
    top_k: int = 5
    
    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> list[Document]:
        # 埋め込み生成
        embedding = generate_embedding(query)  # 前掲の関数を使用
        
        # DynamoDB で検索
        results = search_documents(query, self.top_k)  # 前掲の関数を使用
        
        # LangChain Document 形式に変換
        return [
            Document(
                page_content=r["content"],
                metadata={"title": r["title"], "score": r["score"]},
            )
            for r in results
        ]

# 使用例
retriever = DynamoDBVectorRetriever(
    table_name="RagDocuments",
    index_name="EmbeddingIdx",
    top_k=5,
)
docs = retriever.get_relevant_documents("DynamoDBの料金は？")

def save_conversation_memory(
    session_id: str,
    turn_id: str,
    user_message: str,
    assistant_message: str,
):
    """会話ログをベクトル化してエージェントメモリとして保存する"""
    memory_text = f"ユーザー: {user_message}\nアシスタント: {assistant_message}"
    embedding = generate_embedding(memory_text)
    
    table.put_item(Item={
        "doc_id": f"memory-{session_id}",
        "chunk_id": turn_id,
        "content": memory_text,
        "category": "memory",
        "session_id": session_id,
        "embedding": [Decimal(str(v)) for v in embedding],
    })

def recall_relevant_memories(current_query: str, top_k: int = 3) -> list[dict]:
    """現在のクエリに関連する過去の会話を想起する"""
    return search_documents(current_query, top_k=top_k)

既に DynamoDB を使っている？
     ├─ YES → DynamoDB ネイティブベクトル検索を強く推奨
     └─ NO
          │
          ▼
     AWS 以外のクラウドを使っている？
          ├─ YES → Pinecone（マルチクラウド対応）を検討
          └─ NO（AWS）
               │
               ▼
          フルテキスト検索も同時に必要？
               ├─ YES → OpenSearch Serverless を検討
               └─ NO → DynamoDB ネイティブベクトル検索を推奨

# テナント分離の検索例
response = dynamodb.search_vectors(
    TableName="RagDocuments",
    IndexName="EmbeddingIdx",
    QueryVector=query_vector,
    TopK=10,
    FilterExpression="tenant_id = :tid",
    ExpressionAttributeValues={":tid": {"S": "tenant-abc"}},
)