import numpy as np
 
def cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
    """コサイン類似度: -1 ～ 1 (1に近いほど類似)"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
 
def euclidean_distance(a: np.ndarray, b: np.ndarray) -> float:
    """ユークリッド距離: 0 ～ ∞ (0に近いほど類似)"""
    return np.linalg.norm(a - b)
 
def dot_product(a: np.ndarray, b: np.ndarray) -> float:
    """内積: 正規化済みベクトルではコサイン類似度と等価"""
    return np.dot(a, b)
 
# 使用例
vec_cat  = np.array([0.82, 0.15, -0.33])
vec_dog  = np.array([0.79, 0.18, -0.31])
vec_econ = np.array([-0.21, 0.67, 0.44])
 
print(f"猫-犬: {cosine_similarity(vec_cat, vec_dog):.4f}")   # → 0.9987 (高い)
print(f"猫-経済: {cosine_similarity(vec_cat, vec_econ):.4f}") # → -0.2341 (低い)

from openai import OpenAI
 
client = OpenAI()
 
# 単一テキストの埋め込み
response = client.embeddings.create(
    model="text-embedding-3-large",
    input="機械学習とは何ですか？",
    dimensions=1024,  # 次元削減 (3072 → 1024): コスト・速度改善
)
embedding = response.data[0].embedding
print(f"次元数: {len(embedding)}")  # 1024
 
# バッチ処理 (最大2048テキスト)
texts = [
    "Pythonは汎用プログラミング言語です",
    "機械学習は人工知能の一分野です",
    "寿司は日本の伝統的な料理です",
]
response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts,
)
embeddings = [d.embedding for d in response.data]

from sentence_transformers import SentenceTransformer
 
# BGE-M3: 多言語対応の高性能OSSモデル
model = SentenceTransformer("BAAI/bge-m3")
 
texts = [
    "機械学習の基礎を学ぶ",
    "ディープラーニング入門",
    "今日のランチは何にする？",
]
 
embeddings = model.encode(texts, normalize_embeddings=True)
print(f"形状: {embeddings.shape}")  # (3, 1024)
 
# 類似度行列
from sentence_transformers.util import cos_sim
similarity_matrix = cos_sim(embeddings, embeddings)
print(similarity_matrix)

import cohere
 
co = cohere.Client("YOUR_API_KEY")
 
# input_type が重要: クエリとドキュメントで異なる埋め込みを生成
query_embed = co.embed(
    texts=["日本の人口は？"],
    model="embed-multilingual-v3.0",
    input_type="search_query",       # 検索クエリ用
).embeddings[0]
 
doc_embeds = co.embed(
    texts=[
        "日本の人口は約1億2500万人です。",
        "東京は日本の首都です。",
    ],
    model="embed-multilingual-v3.0",
    input_type="search_document",    # 文書用
).embeddings

from google.cloud import aiplatform
from vertexai.language_models import TextEmbeddingModel
 
# Google の多言語 Embedding モデル
model = TextEmbeddingModel.from_pretrained("text-multilingual-embedding-002")
 
embeddings = model.get_embeddings(
    texts=["機械学習の基礎", "ディープラーニング入門"],
    auto_truncate=True,
)
 
for emb in embeddings:
    print(f"次元数: {len(emb.values)}")  # 768
    print(f"統計: {emb.statistics}")

# ONNX ランタイムで高速推論
from optimum.onnxruntime import ORTModelForFeatureExtraction
from transformers import AutoTokenizer
import numpy as np
 
# ONNX 最適化済みモデルをロード
model = ORTModelForFeatureExtraction.from_pretrained(
    "BAAI/bge-m3",
    export=True,  # 初回は PyTorch → ONNX 変換
)
tokenizer = AutoTokenizer.from_pretrained("BAAI/bge-m3")
 
# 推論
inputs = tokenizer(
    ["機械学習の基礎を学ぶ"],
    padding=True,
    truncation=True,
    max_length=512,
    return_tensors="np",
)
 
outputs = model(**inputs)
# [CLS] トークンの出力を Embedding として使用
embedding = outputs.last_hidden_state[:, 0, :]
embedding = embedding / np.linalg.norm(embedding, axis=1, keepdims=True)
print(f"形状: {embedding.shape}")  # (1, 1024)

import numpy as np
from openai import OpenAI
 
client = OpenAI()
 
def semantic_search(query: str, documents: list[str], top_k: int = 5) -> list:
    """セマンティック検索の基本実装"""
 
    # 全テキストを一括で埋め込み
    all_texts = [query] + documents
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=all_texts,
    )
 
    query_vec = np.array(response.data[0].embedding)
    doc_vecs = np.array([d.embedding for d in response.data[1:]])
 
    # コサイン類似度を計算
    similarities = np.dot(doc_vecs, query_vec) / (
        np.linalg.norm(doc_vecs, axis=1) * np.linalg.norm(query_vec)
    )
 
    # 上位 k 件を返す
    top_indices = np.argsort(similarities)[::-1][:top_k]
    return [
        {"text": documents[i], "score": float(similarities[i])}
        for i in top_indices
    ]
 
# 使用例
docs = [
    "Pythonは汎用的なプログラミング言語で、機械学習で広く使われています",
    "JavaScriptはWebブラウザで動作するスクリプト言語です",
    "深層学習はニューラルネットワークを多層にした機械学習手法です",
    "寿司は酢飯と魚介類を組み合わせた日本料理です",
]
results = semantic_search("AIに使われる言語は？", docs, top_k=2)
for r in results:
    print(f"[{r['score']:.4f}] {r['text']}")

import numpy as np
from rank_bm25 import BM25Okapi
import MeCab
 
def hybrid_search(
    query: str,
    documents: list[str],
    alpha: float = 0.5,  # 0=キーワードのみ, 1=ベクトルのみ
    top_k: int = 5,
) -> list[dict]:
    """ハイブリッド検索: BM25 + Embedding"""
 
    # 1. BM25 (キーワード検索)
    mecab = MeCab.Tagger("-Owakati")
    tokenized_docs = [mecab.parse(doc).strip().split() for doc in documents]
    tokenized_query = mecab.parse(query).strip().split()
 
    bm25 = BM25Okapi(tokenized_docs)
    bm25_scores = bm25.get_scores(tokenized_query)
    # 正規化 (0-1)
    if bm25_scores.max() > 0:
        bm25_scores = bm25_scores / bm25_scores.max()
 
    # 2. Embedding (セマンティック検索)
    all_texts = [query] + documents
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=all_texts,
    )
    query_vec = np.array(response.data[0].embedding)
    doc_vecs = np.array([d.embedding for d in response.data[1:]])
 
    cosine_scores = np.dot(doc_vecs, query_vec) / (
        np.linalg.norm(doc_vecs, axis=1) * np.linalg.norm(query_vec)
    )
    # 正規化 (0-1)
    cosine_scores = (cosine_scores + 1) / 2
 
    # 3. スコア統合
    hybrid_scores = alpha * cosine_scores + (1 - alpha) * bm25_scores
 
    # 上位 k 件を返す
    top_indices = np.argsort(hybrid_scores)[::-1][:top_k]
    return [
        {
            "text": documents[i],
            "hybrid_score": float(hybrid_scores[i]),
            "vector_score": float(cosine_scores[i]),
            "bm25_score": float(bm25_scores[i]),
        }
        for i in top_indices
    ]

from sklearn.cluster import KMeans
from sklearn.decomposition import PCA
import matplotlib.pyplot as plt
import numpy as np
 
def cluster_texts(texts: list[str], n_clusters: int = 3):
    """テキストをクラスタリング"""
    # Embedding 取得
    embeddings = get_embeddings(texts)  # [N, dim]
 
    # K-means クラスタリング
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    labels = kmeans.fit_predict(embeddings)
 
    # PCA で2次元に次元削減して可視化
    pca = PCA(n_components=2)
    reduced = pca.fit_transform(embeddings)
 
    # クラスタごとにグループ化
    clusters = {}
    for text, label in zip(texts, labels):
        clusters.setdefault(int(label), []).append(text)
 
    return clusters
 
# 使用例
texts = [
    "Pythonで機械学習", "TensorFlowの使い方", "PyTorchチュートリアル",
    "東京の観光地", "京都の寺院", "大阪の食べ歩き",
    "確定申告の方法", "住民税の計算", "年末調整の手順",
]
clusters = cluster_texts(texts, n_clusters=3)
for label, items in clusters.items():
    print(f"\nクラスタ {label}:")
    for item in items:
        print(f"  - {item}")

import numpy as np
 
def detect_anomalies(
    reference_texts: list[str],
    test_texts: list[str],
    threshold: float = 0.5,
) -> list[dict]:
    """Embedding ベースの異常検知"""
 
    ref_embeddings = np.array(get_embeddings(reference_texts))
    test_embeddings = np.array(get_embeddings(test_texts))
 
    # 参照テキストの重心 (セントロイド) を計算
    centroid = ref_embeddings.mean(axis=0)
    centroid /= np.linalg.norm(centroid)  # 正規化
 
    results = []
    for text, emb in zip(test_texts, test_embeddings):
        emb_norm = emb / np.linalg.norm(emb)
        similarity = np.dot(centroid, emb_norm)
        is_anomaly = similarity < threshold
        results.append({
            "text": text,
            "similarity": float(similarity),
            "is_anomaly": is_anomaly,
        })
 
    return results

def zero_shot_classify(text: str, categories: list[str]) -> dict:
    """Embedding ベースのゼロショット分類"""
 
    # テキストとカテゴリ全てを埋め込み
    all_inputs = [text] + categories
    embeddings = get_embeddings(all_inputs)
 
    text_emb = np.array(embeddings[0])
    cat_embs = np.array(embeddings[1:])
 
    # 各カテゴリとの類似度を計算
    similarities = np.dot(cat_embs, text_emb) / (
        np.linalg.norm(cat_embs, axis=1) * np.linalg.norm(text_emb)
    )
 
    # ソフトマックスで確率に変換
    exp_sim = np.exp(similarities * 10)  # temperature=0.1
    probs = exp_sim / exp_sim.sum()
 
    return {cat: float(prob) for cat, prob in zip(categories, probs)}
 
# 使用例
result = zero_shot_classify(
    "新しいGPUが発表され、AI処理が3倍高速化",
    ["テクノロジー", "スポーツ", "政治", "エンタメ"]
)
# → {'テクノロジー': 0.89, 'スポーツ': 0.03, '政治': 0.04, 'エンタメ': 0.04}

import numpy as np
from itertools import combinations
 
def find_near_duplicates(
    texts: list[str],
    threshold: float = 0.95,
) -> list[tuple[int, int, float]]:
    """Embedding ベースの重複検出"""
 
    embeddings = np.array(get_embeddings(texts))
    # 正規化
    norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
    normalized = embeddings / norms
 
    # 類似度行列を計算
    similarity_matrix = np.dot(normalized, normalized.T)
 
    # 閾値以上のペアを抽出
    duplicates = []
    for i, j in combinations(range(len(texts)), 2):
        sim = similarity_matrix[i, j]
        if sim >= threshold:
            duplicates.append((i, j, float(sim)))
 
    return sorted(duplicates, key=lambda x: -x[2])
 
# 使用例
texts = [
    "Pythonは人気のプログラミング言語です",
    "Pythonは広く使われるプログラミング言語です",  # ほぼ重複
    "JavaScriptはWebで使われる言語です",
    "今日は天気が良いです",
]
 
duplicates = find_near_duplicates(texts, threshold=0.9)
for i, j, sim in duplicates:
    print(f"[{sim:.4f}] '{texts[i]}' ≈ '{texts[j]}'")

import numpy as np
 
class EmbeddingRecommender:
    """Embedding ベースのレコメンデーションエンジン"""
 
    def __init__(self):
        self.items: list[dict] = []
        self.embeddings: np.ndarray | None = None
 
    def add_items(self, items: list[dict]):
        """アイテムを追加 (title, description, metadata)"""
        self.items = items
        texts = [f"{item['title']}: {item['description']}" for item in items]
        self.embeddings = np.array(get_embeddings(texts))
        # 正規化
        norms = np.linalg.norm(self.embeddings, axis=1, keepdims=True)
        self.embeddings = self.embeddings / norms
 
    def recommend_by_text(self, query: str, top_k: int = 5) -> list[dict]:
        """テキストクエリに基づくレコメンド"""
        query_emb = np.array(get_embeddings([query])[0])
        query_emb = query_emb / np.linalg.norm(query_emb)
 
        scores = np.dot(self.embeddings, query_emb)
        top_indices = np.argsort(scores)[::-1][:top_k]
 
        return [
            {**self.items[i], "score": float(scores[i])}
            for i in top_indices
        ]
 
    def recommend_similar(self, item_index: int, top_k: int = 5) -> list[dict]:
        """類似アイテムのレコメンド"""
        scores = np.dot(self.embeddings, self.embeddings[item_index])
        top_indices = np.argsort(scores)[::-1][1:top_k+1]  # 自分自身を除外
 
        return [
            {**self.items[i], "score": float(scores[i])}
            for i in top_indices
        ]
 
    def recommend_by_history(
        self, viewed_indices: list[int], top_k: int = 5
    ) -> list[dict]:
        """閲覧履歴に基づくレコメンド"""
        # 閲覧アイテムの平均ベクトル
        viewed_embs = self.embeddings[viewed_indices]
        profile = viewed_embs.mean(axis=0)
        profile = profile / np.linalg.norm(profile)
 
        scores = np.dot(self.embeddings, profile)
 
        # 既に閲覧したアイテムを除外
        for idx in viewed_indices:
            scores[idx] = -1
 
        top_indices = np.argsort(scores)[::-1][:top_k]
        return [
            {**self.items[i], "score": float(scores[i])}
            for i in top_indices
        ]

from openai import OpenAI
 
client = OpenAI()
 
# 同じテキストを異なる次元数で埋め込み
text = "機械学習の基礎を学ぶ"
 
for dim in [256, 512, 1024, 3072]:
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text,
        dimensions=dim,
    )
    emb = response.data[0].embedding
    print(f"次元数: {dim:4d}, メモリ: {dim * 4:>5d} bytes/vector")
 
# 出力:
# 次元数:  256, メモリ:  1024 bytes/vector
# 次元数:  512, メモリ:  2048 bytes/vector
# 次元数: 1024, メモリ:  4096 bytes/vector
# 次元数: 3072, メモリ: 12288 bytes/vector
 
# MRL の利点:
# - 同一モデルで精度 vs コストを柔軟に調整
# - 先頭の次元ほど重要な情報を保持
# - 段階的な検索 (粗い→細かい) が可能

import asyncio
from openai import AsyncOpenAI
 
async def batch_embed(texts: list[str], batch_size: int = 100) -> list:
    """大量テキストの効率的な埋め込み"""
    client = AsyncOpenAI()
    all_embeddings = []
 
    # バッチに分割
    batches = [texts[i:i+batch_size] for i in range(0, len(texts), batch_size)]
 
    # 並列実行 (レート制限に注意)
    semaphore = asyncio.Semaphore(5)  # 同時5リクエストまで
 
    async def embed_batch(batch):
        async with semaphore:
            response = await client.embeddings.create(
                model="text-embedding-3-small",
                input=batch,
            )
            return [d.embedding for d in response.data]
 
    results = await asyncio.gather(*[embed_batch(b) for b in batches])
 
    for batch_result in results:
        all_embeddings.extend(batch_result)
 
    return all_embeddings

import hashlib
import json
import sqlite3
import numpy as np
from typing import Optional
 
class EmbeddingCache:
    """SQLite ベースの Embedding キャッシュ"""
 
    def __init__(self, db_path: str = "embedding_cache.db", model: str = "text-embedding-3-small"):
        self.model = model
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS embeddings (
                text_hash TEXT PRIMARY KEY,
                model TEXT,
                embedding BLOB,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
 
    def _hash(self, text: str) -> str:
        return hashlib.sha256(f"{self.model}:{text}".encode()).hexdigest()
 
    def get(self, text: str) -> Optional[list[float]]:
        """キャッシュから Embedding を取得"""
        row = self.conn.execute(
            "SELECT embedding FROM embeddings WHERE text_hash = ?",
            (self._hash(text),),
        ).fetchone()
        if row:
            return json.loads(row[0])
        return None
 
    def put(self, text: str, embedding: list[float]):
        """Embedding をキャッシュに保存"""
        self.conn.execute(
            "INSERT OR REPLACE INTO embeddings (text_hash, model, embedding) VALUES (?, ?, ?)",
            (self._hash(text), self.model, json.dumps(embedding)),
        )
        self.conn.commit()
 
    def get_or_create(self, texts: list[str]) -> list[list[float]]:
        """キャッシュミスのテキストのみ API を呼び出し"""
        results = [None] * len(texts)
        uncached_indices = []
 
        for i, text in enumerate(texts):
            cached = self.get(text)
            if cached:
                results[i] = cached
            else:
                uncached_indices.append(i)
 
        if uncached_indices:
            uncached_texts = [texts[i] for i in uncached_indices]
            response = client.embeddings.create(
                model=self.model, input=uncached_texts
            )
            for idx, emb_data in zip(uncached_indices, response.data):
                results[idx] = emb_data.embedding
                self.put(texts[idx], emb_data.embedding)
 
        return results
 
# 使用例
cache = EmbeddingCache()
embeddings = cache.get_or_create(["Hello", "World", "Hello"])  # "Hello" は1回だけAPI呼び出し

from sentence_transformers import (
    SentenceTransformer,
    InputExample,
    losses,
)
from torch.utils.data import DataLoader
 
# ベースモデルをロード
model = SentenceTransformer("BAAI/bge-m3")
 
# 訓練データの準備 (正例ペア)
train_examples = [
    InputExample(texts=["Pythonの使い方", "Pythonプログラミング入門"], label=0.9),
    InputExample(texts=["Pythonの使い方", "Java言語の基礎"], label=0.3),
    InputExample(texts=["機械学習とは", "ディープラーニングの基礎"], label=0.8),
    InputExample(texts=["機械学習とは", "今日の天気"], label=0.05),
    # 1000+ ペアを推奨
]
 
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
 
# CosineSimilarity Loss (回帰型)
train_loss = losses.CosineSimilarityLoss(model)
 
# 訓練
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=3,
    warmup_steps=100,
    output_path="./finetuned-embedding",
)
 
# ファインチューニング済みモデルの利用
finetuned = SentenceTransformer("./finetuned-embedding")
embeddings = finetuned.encode(["テスト文書"])

from sentence_transformers import InputExample, losses
from sentence_transformers.evaluation import InformationRetrievalEvaluator
 
# Triplet Loss (アンカー, 正例, 負例)
triplet_examples = [
    InputExample(texts=[
        "Pythonでのデータ分析",     # アンカー
        "pandas を使ったデータ処理",  # 正例 (類似)
        "JavaScriptフレームワーク",   # 負例 (非類似)
    ]),
    # ...
]
 
triplet_loader = DataLoader(triplet_examples, shuffle=True, batch_size=16)
triplet_loss = losses.TripletLoss(model, distance_metric=losses.TripletDistanceMetric.COSINE)
 
# Multiple Negatives Ranking Loss (大規模データに最適)
mnrl_examples = [
    InputExample(texts=["クエリ1", "関連文書1"]),
    InputExample(texts=["クエリ2", "関連文書2"]),
    # バッチ内の他ペアが自動的に負例になる
]
 
mnrl_loader = DataLoader(mnrl_examples, shuffle=True, batch_size=32)
mnrl_loss = losses.MultipleNegativesRankingLoss(model)
 
# 評価器の設定
evaluator = InformationRetrievalEvaluator(
    queries={"q1": "Python データ分析", "q2": "機械学習 入門"},
    corpus={"d1": "pandasの使い方...", "d2": "scikit-learnチュートリアル..."},
    relevant_docs={"q1": {"d1"}, "q2": {"d2"}},
)
 
model.fit(
    train_objectives=[(mnrl_loader, mnrl_loss)],
    evaluator=evaluator,
    epochs=5,
    evaluation_steps=500,
    output_path="./finetuned-retrieval",
)

from langchain.text_splitter import RecursiveCharacterTextSplitter
 
# 再帰的分割 (最も一般的)
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # トークン数ではなく文字数
    chunk_overlap=50,      # オーバーラップ文字数
    separators=["\n\n", "\n", "。", "、", " ", ""],  # 分割優先順
    length_function=len,
)
 
chunks = splitter.split_text(long_document)
 
# セマンティック分割 (Embedding 類似度ベース)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
 
semantic_splitter = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=90,  # 類似度の低い箇所で分割
)
 
semantic_chunks = semantic_splitter.split_text(long_document)

import numpy as np
from itertools import combinations
 
def validate_embeddings(
    test_pairs: list[tuple[str, str, float]],
    model_name: str = "text-embedding-3-small",
) -> dict:
    """Embedding モデルの品質検証
 
    test_pairs: [(text_a, text_b, expected_similarity), ...]
    expected_similarity: 0.0 (無関係) ～ 1.0 (同義)
    """
    texts_a = [p[0] for p in test_pairs]
    texts_b = [p[1] for p in test_pairs]
    expected = [p[2] for p in test_pairs]
 
    embs_a = np.array(get_embeddings(texts_a, model=model_name))
    embs_b = np.array(get_embeddings(texts_b, model=model_name))
 
    # コサイン類似度を計算
    actual = []
    for a, b in zip(embs_a, embs_b):
        sim = np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
        actual.append(float(sim))
 
    # 相関係数 (期待値との相関)
    from scipy.stats import spearmanr
    correlation, p_value = spearmanr(expected, actual)
 
    # 分類精度 (閾値0.5で正/負を判別)
    correct = sum(
        1 for e, a in zip(expected, actual)
        if (e >= 0.5 and a >= 0.5) or (e < 0.5 and a < 0.5)
    )
    accuracy = correct / len(test_pairs)
 
    return {
        "model": model_name,
        "spearman_correlation": correlation,
        "p_value": p_value,
        "classification_accuracy": accuracy,
        "mean_actual_similarity": np.mean(actual),
    }

# NG: インデックス時と検索時で異なるモデルを使用
index_embeddings = openai_embed(documents)    # text-embedding-3-large
query_embedding = cohere_embed(query)          # embed-v3
# → ベクトル空間が異なるため、類似度計算が無意味
 
# OK: 同一モデルで統一
index_embeddings = openai_embed(documents, model="text-embedding-3-small")
query_embedding = openai_embed(query, model="text-embedding-3-small")

# NG: 10万文字のドキュメントをそのまま埋め込み
embedding = embed(huge_document)  # 情報が圧縮されすぎて精度低下
 
# OK: 適切にチャンク分割してから埋め込み
chunks = split_text(huge_document, chunk_size=512)
chunk_embeddings = [embed(chunk) for chunk in chunks]
# → 各チャンクの意味が保持される

# NG: 同じテキストを何度も API に送信
for query in user_queries:  # 多くは過去のクエリと同一
    embedding = embed(query)  # 毎回 API 呼び出し → コスト膨大
 
# OK: キャッシュを使用
cache = EmbeddingCache()
for query in user_queries:
    embedding = cache.get_or_create([query])[0]

# NG: Embedding の分布を確認せずにデプロイ
# → 想定外のクラスタリング結果やバイアスに気づかない
 
# OK: 定期的に可視化して品質を確認
from sklearn.manifold import TSNE
import matplotlib.pyplot as plt
 
def visualize_embeddings(embeddings, labels, title="Embedding Space"):
    tsne = TSNE(n_components=2, random_state=42, perplexity=30)
    reduced = tsne.fit_transform(np.array(embeddings))
 
    plt.figure(figsize=(10, 8))
    for label in set(labels):
        mask = [l == label for l in labels]
        points = reduced[mask]
        plt.scatter(points[:, 0], points[:, 1], label=label, alpha=0.6)
    plt.legend()
    plt.title(title)
    plt.savefig("embedding_visualization.png")

# 演習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