企業向けナレッジベースや内部Q&Aシステムの実装において、ドキュメントの埋め込み（Embedding）処理はRAG（Retrieval-Augmented Generation）アーキテクチャの基盤となる工程です。単にAPIエンドポイントを呼び出すだけでなく、モデル選定、バッチリクエストの制御、次元整合性の管理が検索精度と運用コストに直結します。2026年の開発現場では、主に以下の3つの実装パスが標準的に採用されています。

主要な実装アプローチ比較

開発フェーズでは精度検証用にローカル推論を採用し、本番リリースではAPIゲートウェイ経由のクラウド型を選択するのが一般的な構成です。以下に各方式の実装コードと統合パipelineを示します。

開発環境の設定

# メイン依存パッケージ
pip install openai numpy tiktoken

# ローカル推論用パッケージ
pip install sentence-transformers torch accelerate

パターン1：APIゲートウェイ経由のクラウド実装

統一されたインターフェースで複数のLLMベンダーの埋め込みモデルを利用できる構成です。ネットワーク遅延の抑制とキー管理の一元化が可能です。

import numpy as np
from openai import OpenAI

class CloudEmbeddingFacade:
    def __init__(self, token: str, gateway_endpoint: str):
        self.api_handler = OpenAI(api_key=token, base_url=gateway_endpoint)

    def extract_vector_representations(self, target_documents: list[str], model_id: str = "text-embedding-3-large") -> list[list[float]]:
        payload = self.api_handler.embeddings.create(input=target_documents, model=model_id)
        return [doc_item.embedding for doc_item in payload.data]

    @staticmethod
    def calculate_cosine_alignment(vec_x: list[float], vec_y: list[float]) -> float:
        matrix_a, matrix_b = np.array(vec_x), np.array(vec_y)
        dot_val = np.dot(matrix_a, matrix_b)
        magnitude_prod = np.linalg.norm(matrix_a) * np.linalg.norm(matrix_b)
        return float(dot_val / magnitude_prod) if magnitude_prod != 0 else 0.0

# 検証用テストケース
query_samples = [
    "Kubernetesクラスターの構築手順",
    "K8sセットアップガイド",
    "現在の天候状況に関する報告"
]

client = CloudEmbeddingFacade(token="YOUR_API_KEY", gateway_endpoint="https://api.gateway-provider.com/v1")
vectors = client.extract_vector_representations(query_samples)

print(f"関連性が高いペア: {client.calculate_cosine_alignment(vectors[0], vectors[1]):.4f}")
print(f"関連性が低いペア: {client.calculate_cosine_alignment(vectors[0], vectors[2]):.4f}")

パターン2：ローカル環境でのオープンソースモデル推論

データプライバシー要件が厳格な場合や、大量のリクエストに対するコスト抑制を求める場合に有効です。BAAIが開発するbgeシリーズは中文および日本語の構文理解において高い実績を持っています。

import numpy as np
from sentence_transformers import SentenceTransformer

LOCAL_ENCODER_ID = "BAAI/bge-large-zh-v1.5"
RETRIEVAL_PROMPT_TEMPLATE = "Retrieve semantically matching articles for this statement: {}"

class OnPremiseEmbedder:
    def __init__(self, model_ref: str):
        self.encoder_pipeline = SentenceTransformer(model_ref)

    def produce_local_vectors(self, raw_texts: list[str]) -> np.ndarray:
        processed_inputs = [RETRIEVAL_PROMPT_TEMPLATE.format(txt) for txt in raw_texts]
        encoded_output = self.encoder_pipeline.encode(processed_inputs, normalize_embeddings=True)
        return encoded_output

    @staticmethod
    def evaluate_vector_proximity(mat_left: np.ndarray, mat_right: np.ndarray) -> float:
        numerator = np.dot(mat_left, mat_right.T)
        denominator = np.linalg.norm(mat_left) * np.linalg.norm(mat_right)
        return float(numerator / denominator)

# ローカル推論テスト
documents = [
    "Kubernetesクラスターの構築手順",
    "K8sセットアップガイド",
    "現在の天候状況に関する報告"
]

local_processor = OnPremiseEmbedder(LOCAL_ENCODER_ID)
local_vecs = local_processor.produce_local_vectors(documents)

sim_related = local_processor.evaluate_vector_proximity(local_vecs[0], local_vecs[1])
sim_unrelated = local_processor.evaluate_vector_proximity(local_vecs[0], local_vecs[2])

print(f"関連性が高いペア: {sim_related:.4f}")
print(f"関連性が低いペア: {sim_unrelated:.4f}")

パターン3：RAGシステム全体の統合パイプライン

実際のプロダクションでは、テキスト分割、バッチ埋め込み生成、インデックス登録、コンテキスト検索、LLM回答生成までの一連の流れを統一的に扱う必要があります。以下は最小限の構成をモジュール化した実装例です。

import numpy as np
from openai import OpenAI
import json

CLIENT_CONFIG = {
    "token": "YOUR_GATEWAY_KEY",
    "endpoint": "https://api.gateway-provider.com/v1"
}

def segment_document(raw_content: str, max_segment_len: int = 500, overlap_size: int = 50) -> list[str]:
    segments = []
    current_idx = 0
    total_len = len(raw_content)
    
    while current_idx < total_len:
        boundary = min(current_idx + max_segment_len, total_len)
        segments.append(raw_content[current_idx:boundary])
        current_idx = boundary - overlap_size
    return segments

def ingest_and_index(segments: list[str], client_wrapper: OpenAI) -> list[list[float]]:
    all_processed_vectors = []
    batch_cap = 100
    
    for idx in range(0, len(segments), batch_cap):
        current_batch = segments[idx:idx + batch_cap]
        api_response = client_wrapper.embeddings.create(input=current_batch, model="text-embedding-3-large")
        all_processed_vectors.extend([emb.embedding for emb in api_response.data])
        print(f"Indexed: {min(idx + batch_cap, len(segments))}/{len(segments)}")
        
    return all_processed_vectors

class InMemoryVectorRepository:
    def __init__(self):
        self.text_catalog = []
        self.vector_matrix = []

    def register_entries(self, contents: list[str], embeddings: list[list[float]]):
        self.text_catalog.extend(contents)
        self.vector_matrix.extend(embeddings)

    def retrieve_top_k(self, search_vector: list[float], hit_count: int = 3) -> list[str]:
        search_arr = np.array(search_vector)
        match_scores = []
        
        for i, stored_vec in enumerate(self.vector_matrix):
            vec_arr = np.array(stored_vec)
            sim = float(np.dot(search_arr, vec_arr) / (np.linalg.norm(search_arr) * np.linalg.norm(vec_arr)))
            match_scores.append((sim, i))
            
        match_scores.sort(key=lambda x: x[0], reverse=True)
        return [self.text_catalog[idx] for _, idx in match_scores[:hit_count]]

def generate_factual_response(user_query: str, repository: InMemoryVectorRepository, llm_client: OpenAI) -> str:
    query_vec_resp = llm_client.embeddings.create(input=[user_query], model="text-embedding-3-large")
    query_embedding = query_vec_resp.data[0].embedding
    
    matched_chunks = repository.retrieve_top_k(query_embedding, hit_count=3)
    context_block = "\n---\n".join(matched_chunks)
    
    completion_obj = llm_client.chat.completions.create(
        model="gpt-5",
        messages=[
            {"role": "system", "content": "User queries must be answered solely based on the provided reference materials. If information is absent, state that clearly."},
            {"role": "user", "content": f"Reference Data:\n{context_block}\n\nQuestion: {user_query}"}
        ]
    )
    return completion_obj.choices[0].message.content

if __name__ == "__main__":
    llm_engine = OpenAI(api_key=CLIENT_CONFIG["token"], base_url=CLIENT_CONFIG["endpoint"])
    
    sample_manual = """Kubernetes(K8s)はコンテナオーケストレーションを自動化するオープンソース基盤です。
主なコンポーネントにはAPI Server、etcd、Scheduler、Controller Managerが含まれます。
Podは最も小さなデプロイ単位であり、1つ以上のコンテナをホストします。
ServiceはPodのネットワーク到達性を提供し、ClusterIP、NodePort、LoadBalancerをサポートします。
DeploymentはReplicaSetを管理し、ローリングアップデートとロールバック機能を備えています。"""
    
    text_splits = segment_document(sample_manual, max_segment_len=200, overlap_size=30)
    vec_store = InMemoryVectorRepository()
    vec_store.register_entries(text_splits, ingest_and_index(text_splits, llm_engine))
    
    final_answer = generate_factual_response("K8sにおけるServiceの公開モードにはどのような種類がありますか？", vec_store, llm_engine)
    print(final_answer)

実装上の技術的注意点

• 埋め込み次元の不一致による検索失敗：開発時はローカルモデル(1024次元)、本番はクラウドAPI(3072次元)のように変更すると、既存のベクターDBスキーマと整合性が取れなくなります。モデルの変更時には全データの再インデックス化が必要となります。
• APIレートリミットとトークン超過：主要なEmbedding APIは1リクエストあたりの最大件数制限(通常2048件)と総トークン上限を設定しています。ingest_and_index関数のように100件ずつ区切って非同期またはバッチ実行することで、429 Too Many Requestsエラーを回避できます。
• 文脈を考慮したテキスト分割戦略：単純な文字数固定切断（例：chunk_size=500）を行うと、「容器编排」のような複合語が中途半端に切り離され、埋め込みの语义的意味が崩壊します。句読点や改行をブレイクポイントとして検知し、意味のかたまりを保ったままサイズ制限を適用するスライディングウィンドウアルゴリズムの導入が推奨されます。
• 類似度計算時のベクトル正規化処理：一部のエンコーダーは出力時にL2ノルムを適用しない仕様があります。正規化されていない場合、ユークリッド距離や単純な内積計算が異常値を示す可能性があります。ローカル推論時はnormalize_embeddings=Trueを明示し、クラウドAPI側はレスポンスヘッダーで既に単位ベクトル化されていることを確認してください。