ai-data-engineering

AI データエンジニアリング

目的

RAG パイプライン、特徴量ストア、埋め込み生成など、AI/ML システム向けのデータインフラストラクチャを構築します。本番環境の AI アプリケーション向けに、アーキテクチャパターン、オーケストレーションのワークフロー、評価指標を提供します。

使用するタイミング

次の場合にこのスキルを使用します：

• RAG（検索拡張生成）パイプラインを構築する
• セマンティック検索またはベクトルデータベースを実装する
• リアルタイムサービング向けに ML 特徴量ストアを設定する
• 埋め込み生成パイプラインを作成する
• RAGAS 指標を使って RAG の品質を評価する
• AI システム向けのデータワークフローをオーケストレーションする
• フロントエンドスキル（ai-chat、search-filter）と統合する

次の場合はこのスキルを使用しないでください：

• 従来型の CRUD アプリケーションを構築する（databases-relational を使用してください）
• シンプルなキー・バリューストレージを使用する（databases-nosql を使用してください）
• アプリケーションに AI/ML コンポーネントがない

RAG パイプラインのアーキテクチャ

RAG パイプラインには、明確に分かれた5つの段階があります。本番環境で実装するには、このアーキテクチャを理解することが重要です。

┌─────────────────────────────────────────────────────────────┐
│                    RAG Pipeline (5 Stages)                   │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. INGESTION → Load documents (PDF, DOCX, Markdown)        │
│  2. INDEXING → Chunk (512 tokens) + Embed + Store           │
│  3. RETRIEVAL → Query embedding + Vector search + Filters   │
│  4. GENERATION → Context injection + LLM streaming          │
│  5. EVALUATION → RAGAS metrics (faithfulness, relevancy)    │
│                                                              │
└─────────────────────────────────────────────────────────────┘

RAG アーキテクチャと実装パターンの詳細については、以下を参照してください：

• references/rag-architecture.md - 5段階の詳細な内訳
• examples/langchain-rag/basic_rag.py - 動作する実装例

チャンク化の戦略

チャンク化は、RAG の品質を左右する最も重要な決定事項です。不適切なチャンク化は検索の妨げになります。

推奨されるデフォルト設定：

• サイズ： 512トークン
• オーバーラップ： 50～100トークン
• 方式： 固定トークンベース

これらの値を推奨する理由：

• 小さすぎる場合（<256トークン）：コンテキストが失われ、多くの検索が必要になります
• 大きすぎる場合（>1024トークン）：無関係な内容が含まれ、トークン制限に達します
• オーバーラップにより、チャンク境界での情報損失を防ぎます

特殊なケース向けの代替戦略：

# Code-aware chunking (preserves functions/classes)
from langchain.text_splitter import RecursiveCharacterTextSplitter

code_splitter = RecursiveCharacterTextSplitter.from_language(
    language="python",
    chunk_size=512,
    chunk_overlap=50
)

# Semantic chunking (splits on meaning, not tokens)
from langchain.text_splitter import SemanticChunker

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile"  # Split at semantic boundaries
)

参照先： 完全な判断フレームワークについては、references/chunking-strategies.md を参照してください

埋め込みの生成

埋め込みの品質は検索精度に直接影響します。現在、Voyage AI がクラス最高の性能を誇ります。

第一推奨：Voyage AI voyage-3

• 次元数：1024
• MTEB スコア：69.0（2025年12月時点で最高）
• コスト：$$$ ですが、OpenAI より9.74%高性能です
• 用途：最高の検索品質が求められる本番システム

コスト効率の高い代替案：OpenAI text-embedding-3-small

• 次元数：1536
• MTEB スコア：62.3
• コスト：$（voyage-3 より5倍低価格）
• 用途：開発、プロトタイピング、コスト重視のアプリケーション

実装：

from langchain_voyageai import VoyageAIEmbeddings
from langchain_openai import OpenAIEmbeddings

# Production (best quality)
embeddings = VoyageAIEmbeddings(
    model="voyage-3",
    voyage_api_key="your-api-key"
)

# Development (cost-effective)
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key="your-api-key"
)

参照先： プロバイダーの完全な比較については、references/embedding-strategies.md を参照してください

RAGAS 評価指標

従来の指標（BLEU、ROUGE）では RAG の品質を測定できません。RAGAS は LLM による評価を提供します。

4つの主要指標：

指標	測定内容	良好なスコア
忠実性	検索されたコンテキストとの事実の一貫性	> 0.8
回答の関連性	回答がユーザーの質問に答えているか	> 0.7
コンテキスト適合率	検索されたチャンクが実際に関連しているか	> 0.6
コンテキスト再現率	必要なチャンクがすべて検索されたか	> 0.7

簡易評価スクリプト：

# Run RAGAS evaluation (TOKEN-FREE script execution)
python scripts/evaluate_rag.py --dataset eval_data.json --output results.json

手動実装：

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

dataset = {
    "question": ["What is the capital of France?"],
    "answer": ["Paris is the capital of France."],
    "contexts": [["France's capital is Paris."]],
    "ground_truth": ["Paris"]
}

result = evaluate(dataset, metrics=[faithfulness, answer_relevancy])
print(f"Faithfulness: {result['faithfulness']}")
print(f"Answer Relevancy: {result['answer_relevancy']}")

参照先： RAGAS の完全な実装ガイドについては、references/evaluation-metrics.md を参照してください

特徴量ストア

特徴量ストアは、一貫した特徴量計算を提供することで「学習時とサービング時の乖離」の問題を解決します。

第一推奨：Feast - オープンソースで、あらゆるバックエンド（PostgreSQL、Redis、DynamoDB、S3、BigQuery、Snowflake）で動作します

基本的な使い方：

from feast import FeatureStore
store = FeatureStore(repo_path="feature_repo/")

# Online serving (low-latency)
features = store.get_online_features(
    features=["user_features:total_orders"],
    entity_rows=[{"user_id": 1001}]
).to_dict()

参照先： Feast の完全な設定方法と代替案（Tecton、Hopsworks）については、references/feature-stores.md を参照してください

LangChain によるオーケストレーション

LangChain は、最大規模のエコシステム（API リファレンスのスニペットが24,215件以上）を持つ、LLM オーケストレーションの主要なフレームワークです。

Context7 ライブラリ ID： /websites/langchain_oss_python_langchain（信頼度：高、スニペット数：435）

基本的な RAG チェーン：

from langchain_core.prompts import ChatPromptTemplate
from langchain_qdrant import QdrantVectorStore
from langchain_voyageai import VoyageAIEmbeddings

# Setup retriever
vectorstore = QdrantVectorStore(
    client=qdrant_client,
    embedding=VoyageAIEmbeddings(model="voyage-3")
)
retriever = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 5})

# Build chain
prompt = ChatPromptTemplate.from_template(
    "Answer based on context:\n{context}\n\nQuestion: {question}"
)
chain = {"context": retriever, "question": lambda x: x} | prompt | ChatOpenAI() | StrOutputParser()

# Stream response
for chunk in chain.stream("What is the capital of France?"):
    print(chunk, end="", flush=True)

参照先： ストリーミングとハイブリッド検索を含む LangChain 0.3 以降の完全なパターンについては、references/langchain-patterns.md を参照してください

オーケストレーションツール

最新の AI パイプラインには、cron ジョブを超えるワークフローオーケストレーションが必要です。

第一の推奨：Dagster（ML／AI パイプライン向け） - アセット中心の設計、最も優れたリネージ追跡、RAG に最適です

例：埋め込みパイプライン

from dagster import asset
from langchain_voyageai import VoyageAIEmbeddings

@asset
def raw_documents():
    """Load documents from S3."""
    return documents

@asset
def chunked_documents(raw_documents):
    """Split into 512-token chunks with 50-token overlap."""
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=50)
    return splitter.split_documents(raw_documents)

@asset
def embedded_documents(chunked_documents):
    """Generate embeddings with Voyage AI."""
    embeddings = VoyageAIEmbeddings(model="voyage-3")
    return embeddings.embed_documents([doc.page_content for doc in chunked_documents])

参照： 完全な Dagster のパターンと代替ツール（Prefect、Airflow 3.0、dbt）については、references/orchestration-tools.md を参照してください。

フロントエンドスキルとの統合

ai-chat スキル → RAG バックエンド

ai-chat スキルは、ストリーミング応答のために RAG パイプラインの出力を利用します。

バックエンド API（FastAPI）：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

@app.post("/api/rag/stream")
async def stream_rag(query: str):
    async def generate():
        chain = RetrievalQA.from_chain_type(llm=OpenAI(streaming=True), retriever=vectorstore.as_retriever())
        async for chunk in chain.astream(query):
            yield chunk
    return StreamingResponse(generate(), media_type="text/plain")

参照： フロントエンド統合パターンの全容については、references/rag-architecture.md を参照してください。

search-filter スキル → セマンティック検索

search-filter スキルは、ベクトル類似度検索のためにセマンティック検索バックエンドを使用します。

バックエンド（Qdrant + Voyage AI）：

from qdrant_client import QdrantClient
from langchain_voyageai import VoyageAIEmbeddings

@app.post("/api/search/semantic")
async def semantic_search(query: str, filters: dict):
    query_vector = VoyageAIEmbeddings(model="voyage-3").embed_query(query)
    results = QdrantClient().search(
        collection_name="documents",
        query_vector=query_vector,
        query_filter=filters,
        limit=10
    )
    return {"results": results}

データのバージョン管理

第一の推奨：LakeFS（2025 年 11 月に DVC チームを買収）

データレイク上で Git のような操作を行えます。ブランチ、コミット、マージ、タイムトラベルに対応しています。S3／Azure／GCS で動作します。

import lakefs

branch = lakefs.Branch("main").create("experiment-voyage-3")
branch.commit("Updated embeddings to voyage-3")
branch.merge_into("main")

参照： LakeFS のセットアップの全容については、references/data-versioning.md を参照してください。

クイックスタートのワークフロー

1. ベクトルデータベースをセットアップする：

# Run Qdrant setup script (TOKEN-FREE execution)
python scripts/setup_qdrant.py --collection docs --dimension 1024

2. ドキュメントをチャンク化して埋め込む：

# Chunk documents (TOKEN-FREE execution)
python scripts/chunk_documents.py \
  --input data/documents/ \
  --chunk-size 512 \
  --overlap 50 \
  --output data/chunks/

3. RAG パイプラインを実装する：

完全に動作する例については、examples/langchain-rag/basic_rag.py を参照してください。

4. RAGAS で評価する：

# Run evaluation (TOKEN-FREE execution)
python scripts/evaluate_rag.py \
  --dataset data/eval_qa.json \
  --output results/ragas_metrics.json

5. オーケストレーションを使ってデプロイする：

本番環境へのデプロイについては、examples/dagster-pipelines/embedding_pipeline.py を参照してください。

依存関係

必須の Python パッケージ：

# Core RAG
pip install langchain langchain-core langchain-openai langchain-voyageai langchain-qdrant

# Vector database
pip install qdrant-client

# Evaluation
pip install ragas datasets

# Feature stores
pip install feast

# Orchestration
pip install dagster dagster-webserver

# Data versioning
pip install lakefs-client

代替ツールで必要に応じて使用するもの：

# LlamaIndex (alternative to LangChain)
pip install llama-index

# dbt (SQL transformations)
pip install dbt-core dbt-postgres

# Prefect (alternative orchestration)
pip install prefect

トラブルシューティング

よくある問題：

1. 検索品質が低い - チャンクサイズを確認する（512 トークンを試す）、オーバーラップを増やす（50～100）、ハイブリッド検索を試す、Cohere で再ランキングする

2. 埋め込み生成が遅い - ドキュメントをバッチ処理する（100～1000 件）、非同期 API を使用する、Redis でキャッシュする、開発環境では小さいモデルを使用する

3. LLM のコストが高い - 検索するチャンク数を減らす（k=3）、より安価な再ランキングモデルを使用する、頻出するクエリをキャッシュする

参照： トラブルシューティングガイドの全容については、references/rag-architecture.md を参照してください。

ベストプラクティス

チャンク化： デフォルトでは、50 トークンのオーバーラップを設けて 512 トークンにします。複雑なドキュメントにはセマンティックチャンク化を使用します。ソースコードではコード構造を維持します。

埋め込み： 本番環境では Voyage AI voyage-3 を、開発環境では OpenAI text-embedding-3-small を使用します。埋め込みモデルを混在させないでください（変更する場合はすべて再埋め込みします）。

評価： パイプラインを変更するたびに RAGAS のメトリクスを実行します。50 組以上の質問と回答を含むテストデータセットを維持します。メトリクスを継続的に追跡します。

オーケストレーション： ML／AI パイプラインには Dagster を、SQL の変換のみに dbt を使用します。すべてのパイプラインコードをバージョン管理します。

フロントエンドとの統合： LLM の応答は必ずストリーミングします。再試行ロジックを実装します。引用やソースをユーザーに表示します。結果が空の場合も適切に処理します。

追加リソース

リファレンスドキュメント：

• references/rag-architecture.md - RAG パイプラインの完全ガイド
• references/chunking-strategies.md - チャンク化の判断フレームワーク
• references/embedding-strategies.md - 埋め込みモデルの比較
• references/langchain-patterns.md - LangChain 0.3 以降のパターン
• references/feature-stores.md - Feast のセットアップと代替ツール
• references/evaluation-metrics.md - RAGAS の実装ガイド

動作するサンプル：

• examples/langchain-rag/basic_rag.py - シンプルな RAG チェーン
• examples/langchain-rag/streaming_rag.py - ストリーミング応答
• examples/langchain-rag/hybrid_search.py - ベクトル検索 + BM25
• examples/llamaindex-agents/query_engine.py - LlamaIndex の代替実装
• examples/feast-features/ - 完全なフィーチャーストアのセットアップ
• examples/dagster-pipelines/embedding_pipeline.py - 本番環境向けパイプライン

実行可能なスクリプト（トークン不要）：

• scripts/evaluate_rag.py - RAGAS 評価ランナー
• scripts/chunk_documents.py - ドキュメントのチャンク化ユーティリティ
• scripts/benchmark_retrieval.py - 検索品質のベンチマーク
• scripts/setup_qdrant.py - Qdrant コレクションのセットアップ