import numpy as np
 
# ベクトル DB で使用される主な距離関数
 
def cosine_distance(a, b):
    """コサイン距離: 0 (同一) ～ 2 (正反対)
    テキスト Embedding で最も一般的"""
    return 1 - np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
 
def euclidean_distance(a, b):
    """ユークリッド距離 (L2): 0 ～ ∞
    画像特徴量やクラスタリングに適する"""
    return np.linalg.norm(a - b)
 
def dot_product_distance(a, b):
    """内積: 正規化済みベクトルではコサインと等価
    最も計算が速い"""
    return -np.dot(a, b)  # 負にして距離化
 
# 選択ガイド:
# - テキスト検索 → コサイン距離 or 内積
# - 画像検索 → ユークリッド距離
# - 正規化済み → 内積 (最速)

from pinecone import Pinecone, ServerlessSpec
 
# 初期化
pc = Pinecone(api_key="YOUR_API_KEY")
 
# インデックス作成
pc.create_index(
    name="products",
    dimension=1024,
    metric="cosine",
    spec=ServerlessSpec(cloud="aws", region="us-east-1"),
)
 
index = pc.Index("products")
 
# ベクトルの挿入 (Upsert)
index.upsert(
    vectors=[
        {
            "id": "prod-001",
            "values": [0.1, 0.2, ...],  # 1024次元
            "metadata": {
                "name": "ワイヤレスイヤホン",
                "category": "electronics",
                "price": 15000,
            },
        },
    ],
    namespace="jp-products",
)
 
# クエリ (メタデータフィルタ付き)
results = index.query(
    vector=[0.15, 0.22, ...],
    top_k=10,
    filter={
        "category": {"$eq": "electronics"},
        "price": {"$lte": 20000},
    },
    include_metadata=True,
    namespace="jp-products",
)
 
for match in results.matches:
    print(f"ID: {match.id}, Score: {match.score:.4f}")
    print(f"  {match.metadata}")

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue, Range,
)
 
# 接続
client = QdrantClient(url="http://localhost:6333")
 
# コレクション作成
client.create_collection(
    collection_name="products",
    vectors_config=VectorParams(
        size=1024,
        distance=Distance.COSINE,
    ),
)
 
# ベクトル挿入
client.upsert(
    collection_name="products",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],
            payload={
                "name": "ワイヤレスイヤホン",
                "category": "electronics",
                "price": 15000,
            },
        ),
    ],
)
 
# フィルタ付き検索
results = client.search(
    collection_name="products",
    query_vector=[0.15, 0.22, ...],
    query_filter=Filter(
        must=[
            FieldCondition(key="category", match=MatchValue(value="electronics")),
            FieldCondition(key="price", range=Range(lte=20000)),
        ]
    ),
    limit=10,
)

import weaviate
from weaviate.classes.config import Property, DataType, Configure
 
# 接続
client = weaviate.connect_to_local()
 
# コレクション作成 (スキーマ定義)
collection = client.collections.create(
    name="Product",
    vectorizer_config=Configure.Vectorizer.none(),  # 外部Embedding使用
    properties=[
        Property(name="name", data_type=DataType.TEXT),
        Property(name="category", data_type=DataType.TEXT),
        Property(name="price", data_type=DataType.INT),
    ],
)
 
# データ挿入
collection.data.insert(
    properties={"name": "ワイヤレスイヤホン", "category": "electronics", "price": 15000},
    vector=[0.1, 0.2, ...],
)
 
# ハイブリッド検索 (ベクトル + キーワード)
results = collection.query.hybrid(
    query="高音質イヤホン",
    vector=[0.15, 0.22, ...],
    alpha=0.5,  # 0=キーワードのみ, 1=ベクトルのみ
    limit=10,
)
 
client.close()

import psycopg2
from pgvector.psycopg2 import register_vector
 
# 接続
conn = psycopg2.connect("postgresql://localhost/mydb")
register_vector(conn)
 
cur = conn.cursor()
 
# 拡張とテーブル作成
cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
cur.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id SERIAL PRIMARY KEY,
        name TEXT NOT NULL,
        category TEXT,
        price INTEGER,
        embedding vector(1024)
    )
""")
 
# HNSW インデックス作成
cur.execute("""
    CREATE INDEX ON products
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 200)
""")
 
# データ挿入
cur.execute(
    "INSERT INTO products (name, category, price, embedding) VALUES (%s, %s, %s, %s)",
    ("ワイヤレスイヤホン", "electronics", 15000, [0.1, 0.2, ...]),
)
 
# 検索 (SQL で記述可能)
cur.execute("""
    SELECT name, price, 1 - (embedding <=> %s::vector) AS similarity
    FROM products
    WHERE category = 'electronics' AND price <= 20000
    ORDER BY embedding <=> %s::vector
    LIMIT 10
""", ([0.15, 0.22, ...], [0.15, 0.22, ...]))
 
for row in cur.fetchall():
    print(f"{row[0]}: ¥{row[1]:,} (類似度: {row[2]:.4f})")
 
conn.commit()

import chromadb
from chromadb.utils import embedding_functions
 
# クライアント作成 (インメモリ or 永続化)
client = chromadb.Client()  # インメモリ
# client = chromadb.PersistentClient(path="./chroma_db")  # ディスク永続化
 
# Embedding 関数の設定 (内蔵 or 外部)
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key="YOUR_API_KEY",
    model_name="text-embedding-3-small",
)
 
# コレクション作成
collection = client.create_collection(
    name="documents",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine"},  # 距離関数を指定
)
 
# ドキュメント追加 (自動的に Embedding が生成される)
collection.add(
    documents=[
        "Pythonは汎用プログラミング言語です",
        "機械学習はAIの一分野です",
        "寿司は日本料理です",
    ],
    metadatas=[
        {"category": "programming"},
        {"category": "ai"},
        {"category": "food"},
    ],
    ids=["doc1", "doc2", "doc3"],
)
 
# 検索 (テキストクエリで自動 Embedding)
results = collection.query(
    query_texts=["AIプログラミング"],
    n_results=2,
    where={"category": {"$ne": "food"}},
)
print(results)
 
# Chroma の利点:
# - セットアップが最も簡単 (pip install chromadb)
# - Embedding 関数を内蔵
# - 開発・プロトタイプに最適
# - LangChain/LlamaIndex と統合が容易

# Qdrant でのバッチアップサート
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct
 
client = QdrantClient(url="http://localhost:6333")
 
def batch_upsert(collection: str, data: list[dict], batch_size: int = 100):
    """大量データの効率的な挿入"""
    points = [
        PointStruct(id=d["id"], vector=d["vector"], payload=d["metadata"])
        for d in data
    ]
 
    # バッチに分割して挿入
    for i in range(0, len(points), batch_size):
        batch = points[i:i + batch_size]
        client.upsert(
            collection_name=collection,
            points=batch,
            wait=False,  # 非同期挿入 (高速化)
        )
 
    # 最後に同期を待つ
    client.upsert(
        collection_name=collection,
        points=[],
        wait=True,
    )

-- pgvector のパフォーマンス最適化
 
-- 1. HNSW インデックスの作成 (推奨)
CREATE INDEX idx_documents_embedding ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
 
-- 検索時の ef を設定
SET hnsw.ef_search = 100;  -- top_k の 2-4 倍
 
-- 2. IVFFlat インデックス (構築が速い、大量データ向け)
CREATE INDEX idx_documents_ivf ON documents
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 1000);  -- sqrt(N) が目安
 
-- 検索時のプローブ数
SET ivfflat.probes = 10;  -- lists の 1-5%
 
-- 3. パフォーマンス確認
EXPLAIN ANALYZE
SELECT name, 1 - (embedding <=> $1::vector) AS similarity
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 10;
 
-- 4. 部分インデックス (フィルタが多い場合)
CREATE INDEX idx_electronics_embedding ON documents
USING hnsw (embedding vector_cosine_ops)
WHERE category = 'electronics';
 
-- 5. メモリ設定
SET maintenance_work_mem = '2GB';  -- インデックス構築時のメモリ
SET work_mem = '256MB';             -- 検索時のメモリ

from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, Distance, PointStruct, NamedVector,
)
 
client = QdrantClient(url="http://localhost:6333")
 
# マルチベクトルコレクション
client.create_collection(
    collection_name="products_multi",
    vectors_config={
        "title": VectorParams(size=384, distance=Distance.COSINE),
        "description": VectorParams(size=1024, distance=Distance.COSINE),
        "image": VectorParams(size=512, distance=Distance.COSINE),
    },
)
 
# 複数ベクトルでのデータ挿入
client.upsert(
    collection_name="products_multi",
    points=[
        PointStruct(
            id=1,
            vector={
                "title": [0.1, 0.2, ...],       # タイトル Embedding
                "description": [0.3, 0.4, ...],  # 説明文 Embedding
                "image": [0.5, 0.6, ...],         # 画像 Embedding
            },
            payload={"name": "ワイヤレスイヤホン", "price": 15000},
        ),
    ],
)
 
# 特定のベクトルフィールドで検索
results = client.search(
    collection_name="products_multi",
    query_vector=NamedVector(
        name="description",  # 説明文ベクトルで検索
        vector=[0.35, 0.45, ...],
    ),
    limit=10,
)

from langchain_openai import OpenAIEmbeddings
from langchain_postgres import PGVector
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
 
# Embedding モデル
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
 
# pgvector ストアの作成
vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="rag_documents",
    connection="postgresql://localhost/rag_db",
)
 
# ドキュメントの読み込みと分割
from langchain_community.document_loaders import TextLoader
loader = TextLoader("knowledge_base.txt")
documents = loader.load()
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
)
chunks = splitter.split_documents(documents)
 
# ベクトル DB に挿入
vectorstore.add_documents(chunks)
 
# RAG チェーンの構築
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5},
)
 
qa_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    chain_type="stuff",
    retriever=retriever,
)
 
# 質問応答
answer = qa_chain.invoke("RAGの仕組みを説明してください")
print(answer["result"])

from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings,
)
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
 
# Qdrant クライアント
qdrant_client = QdrantClient(url="http://localhost:6333")
 
# ベクトルストア設定
vector_store = QdrantVectorStore(
    client=qdrant_client,
    collection_name="llama_docs",
    enable_hybrid=True,  # ハイブリッド検索を有効化
)
 
# 設定
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
 
# ドキュメントの読み込みとインデックス作成
documents = SimpleDirectoryReader("./docs/").load_data()
index = VectorStoreIndex.from_documents(
    documents,
    vector_store=vector_store,
)
 
# クエリエンジン
query_engine = index.as_query_engine(
    similarity_top_k=5,
    vector_store_query_mode="hybrid",  # ハイブリッド検索
    alpha=0.5,
)
 
response = query_engine.query("ベクトルDBの選び方は？")
print(response)

from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, Distance, OptimizersConfigDiff,
    CollectionParams,
)
 
client = QdrantClient(url="http://localhost:6333")
 
# シャーディング付きコレクション作成
client.create_collection(
    collection_name="large_scale",
    vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
    shard_number=4,           # 4つのシャードに分割
    replication_factor=2,     # 2重レプリケーション
    write_consistency_factor=1,
    optimizers_config=OptimizersConfigDiff(
        indexing_threshold=20000,  # インデックス構築の閾値
        memmap_threshold=50000,    # メモリマッピングの閾値
    ),
)

import requests
from datetime import datetime
 
def monitor_qdrant(base_url: str = "http://localhost:6333"):
    """Qdrant のヘルスチェックとメトリクス取得"""
 
    # ヘルスチェック
    health = requests.get(f"{base_url}/healthz").json()
    print(f"ステータス: {health}")
 
    # コレクション情報
    collections = requests.get(f"{base_url}/collections").json()
    for col in collections["result"]["collections"]:
        name = col["name"]
        info = requests.get(f"{base_url}/collections/{name}").json()
        result = info["result"]
 
        print(f"\nコレクション: {name}")
        print(f"  ベクトル数: {result['vectors_count']:,}")
        print(f"  ポイント数: {result['points_count']:,}")
        print(f"  セグメント数: {len(result.get('segments', []))}")
        print(f"  ディスク使用: {result.get('disk_data_size', 0) / 1024**2:.1f} MB")
        print(f"  RAM 使用: {result.get('ram_data_size', 0) / 1024**2:.1f} MB")
 
    # テレメトリ (Prometheus 形式)
    metrics = requests.get(f"{base_url}/metrics").text
    return metrics
 
# 定期監視
monitor_qdrant()

-- pgvector のパフォーマンス監視
 
-- 1. インデックスの利用状況
SELECT
    schemaname,
    tablename,
    indexname,
    idx_scan,
    idx_tup_read,
    idx_tup_fetch,
    pg_size_pretty(pg_relation_size(indexrelid)) AS index_size
FROM pg_stat_user_indexes
WHERE indexname LIKE '%embedding%';
 
-- 2. 遅いクエリの検出
SELECT
    query,
    calls,
    mean_exec_time,
    max_exec_time,
    total_exec_time
FROM pg_stat_statements
WHERE query LIKE '%<=>%'  -- ベクトル検索クエリ
ORDER BY mean_exec_time DESC
LIMIT 10;
 
-- 3. テーブルサイズの確認
SELECT
    pg_size_pretty(pg_total_relation_size('products')) AS total_size,
    pg_size_pretty(pg_relation_size('products')) AS table_size,
    pg_size_pretty(pg_indexes_size('products')) AS index_size;

# NG: 1億件を全てメモリに格納
client.create_collection(
    collection_name="huge_data",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
    # → 1536 * 4bytes * 100M = 614GB のメモリが必要!
)
 
# OK: 量子化 + ディスクバックのインデックス
from qdrant_client.models import ScalarQuantization, ScalarQuantizationConfig
 
client.create_collection(
    collection_name="huge_data",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantizationConfig(type="int8", always_ram=False),
    ),
    on_disk_payload=True,  # ペイロードもディスクに
)

-- NG: pgvector でインデックスなし
SELECT * FROM documents
ORDER BY embedding <=> $1
LIMIT 10;
-- → 100万件でフルスキャン: 数秒かかる
 
-- OK: HNSW インデックスを作成
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);
-- → 同じクエリが 1ms 以下に

# NG: 異なる種類のコンテンツを同一コレクションに
collection.upsert([
    # 商品説明と技術文書を混在 → 検索精度が低下
    {"id": 1, "vector": product_embedding, "type": "product"},
    {"id": 2, "vector": doc_embedding, "type": "documentation"},
])
 
# OK: コンテンツタイプ別にコレクションを分割
product_collection = client.create_collection("products", ...)
doc_collection = client.create_collection("documentation", ...)
# または Named Vectors で分離

# NG: ベクトル DB のバックアップを取っていない
# → 障害時にすべてのベクトルを再計算する必要がある
 
# OK: 定期的なスナップショット取得
# Qdrant の場合
snapshot = requests.post(
    f"{base_url}/collections/products/snapshots"
).json()
 
# pgvector の場合
# 通常の pg_dump がそのまま使える
# pg_dump -Fc mydb > mydb_backup.dump
 
# 元データの保持が最重要
# ベクトルは元テキスト + Embedding モデルから再構築可能

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]