DeepSeek データエンジニアリングチュートリアル

データ収集から品質評価まで、大規模言語モデルのデータエンジニアリングパイプラインを完全に習得。SFTデータセット構築、RLHF選好データ準備、データクリーニングと重複排除、データ拡張、大規模データ処理などの中核的な要素を網羅し、Python実践コード付き。

学習を始める

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

データはLLMの魂

大規模モデルのトレーニングにおいて、データ品質はモデルの能力の上限を直接決定します。高品質なデータセットは小さなモデルが大きなモデルを凌駕することを可能にし、低品質なデータは数兆パラメータのモデルでも平凡な性能に留めます。データエンジニアリングは、生データの収集から最終的なトレーニングデータの提供までの全チェーンをカバーします。

第1章

データエンジニアリング概要

LLMトレーニングにおけるデータの中心的な役割を理解し、データ品質とデータ量のトレードオフを把握し、データエンジニアリングパイプラインの全体像を構築します。

LLMトレーニングにおけるデータの中心的な役割

LLMトレーニングの3要素（アルゴリズム、計算リソース、データ）の中で、データはしばしば過小評価されますが、その重要性は他の2つをはるかに上回ります。広く信じられている見解は、モデルアーキテクチャの限界収益は減少する一方、データ品質の限界収益は増加するというものです。以下に、相互に関連する3つの核となる事実を示します。

データ品質がモデル能力の上限を決定する：最も先進的なモデルアーキテクチャ（例：MoE）を使用しても、トレーニングデータの品質が低ければ、モデルは高品質な応答を生成できません。DeepSeek-V3とR1の強力な能力は、慎重に構築されたトレーニングデータに大きく起因しています。
データの多様性が汎化能力を決定する：単一ドメインのデータは過学習を引き起こす可能性がありますが、多様なデータにより、モデルは未見のタスクでも良好に機能します。DeepSeekのデータ混合戦略は、数学、コード、推論、対話、クリエイティブライティングなど、複数のドメインをカバーしています。
データ規模が知識の境界を決定する：モデルの知識範囲は、トレーニングデータのカバー範囲を超えることはありません。モデルに医学を理解させるには、高品質な医学データが必要です。プログラミングを可能にするには、十分なコードデータが必要です。

データ品質 vs データ量

限られた計算リソースの下では、データ品質はデータ量よりも重要です。以下に、両者のトレードオフ分析を示します。

次元	量を追求	質を追求
トレーニング効率	トレーニング時間が長く、収束が遅い	トレーニングが速く、収束が安定
モデル性能	ノイズが多く、出力が不安定	出力が正確で、幻覚が少ない
コスト	GPUコストが高く、期間が長い	データクリーニングコストは高いが、総コストは低い
典型的な戦略	ウェブデータをクロールし、粗いフィルタリング	データソースを厳選し、多重フィルタリング、人手によるアノテーション

DeepSeekチームは実践で、高品質データの1/10でトレーニングしたモデルが、指示追従と推論能力において、全量の粗いデータでトレーニングしたモデルをしばしば上回ることを発見しました。これは、データエンジニアリングがLLM開発において最も重要な部分である理由も説明しています。

データエンジニアリングパイプラインの全体像

完全なデータエンジニアリングパイプラインには、以下の段階が含まれます。

データ収集：公開データセット、ウェブクロール、API呼び出し、合成生成などのチャネルから生データを取得
データクリーニング：重複の除去、低品質のフィルタリング、欠損値の処理、機密情報のフィルタリング
データアノテーション：SFT、RLHF、DPOなど、異なるトレーニング段階に合わせて異なる形式のデータを構築
データ拡張：Self-Instruct、Evol-Instruct、逆翻訳などの技術を使用してデータを拡張
品質評価：多様性、難易度、指示の複雑さなどの次元からデータ品質を評価
データ混合：異なるドメインのデータを比率で混合し、最終的なトレーニングデータセットを形成
バージョン管理：DVCなどのツールを使用してデータバージョンを管理し、再現性を確保

第2章

データ収集とソース

LLMトレーニングデータの主要なソース（公開データセット、ウェブクロール、合成データ生成、データコンプライアンスの考慮事項）について学びます。

一般的な公開データセット

データセット名	規模	タイプ	適用シナリオ
Alpaca	52K	SFT指示データ	指示微調整の入門
ShareGPT	90K	マルチターン対話	対話能力のトレーニング
UltraChat	1.5M	マルチターン対話	大規模対話トレーニング
OpenOrca	4M	SFT指示データ	大規模指示微調整
CodeAlpaca	20K	コード生成	コード能力のトレーニング
MathInstruct	260K	数学的推論	数学能力のトレーニング

Hugging Faceから公開データセットを読み込む

from datasets import load_dataset

# Alpacaデータセットを読み込む
alpaca = load_dataset("tatsu-lab/alpaca")
print(f"Alpacaトレーニングセット: {len(alpaca['train'])} サンプル")
print(f"例: {alpaca['train'][0]}")

# ShareGPT 会話データの読み込み
sharegpt = load_dataset("anon8231489123/ShareGPT_Vicuna_unfiltered",
                         data_files="ShareGPT_V3_unfiltered_cleaned_split.json")
print(f"ShareGPT: {len(sharegpt['train'])} 件の会話")

# OpenOrca データの読み込み（必要に応じてサブセット）
orca = load_dataset("Open-Orca/OpenOrca", split="train[:50000]")
print(f"OpenOrca サブセット: {len(orca)} 件")

Webスクレイピングとデータ収集

特定分野のデータには、Webスクレイピングが重要な補完手段です。以下はドキュメントやチュートリアル向けのクローラーの例です：

import requests
from bs4 import BeautifulSoup
import time
from urllib.parse import urljoin, urlparse

def crawl_documentation(base_url, max_pages=100, delay=1.0):
    """ドキュメントサイトをクロールして本文を抽出"""
    visited = set()
    to_visit = [base_url]
    results = []

    while to_visit and len(visited) < max_pages:
        url = to_visit.pop(0)
        if url in visited:
            continue
        visited.add(url)

        try:
            resp = requests.get(url, timeout=10, headers={
                "User-Agent": "Mozilla/5.0 (compatible; DataBot/1.0)"
            })
            resp.raise_for_status()
            soup = BeautifulSoup(resp.text, "html.parser")

            # スクリプトとスタイルタグを削除
            for tag in soup(["script", "style", "nav", "footer"]):
                tag.decompose()

            text = soup.get_text(separator=" ", strip=True)
if len(text) > 200:
                results.append({
                    "url": url,
                    "title": soup.title.string if soup.title else "",
                    "content": text,
                })

            # 新しいリンクを発見
            for link in soup.find_all("a", href=True):
                href = urljoin(url, link["href"])
                if urlparse(href).netloc == urlparse(base_url).netloc:
                    if href not in visited:
                        to_visit.append(href)

            time.sleep(delay)

        except Exception as e:
            print(f"クロール失敗 {url}: {e}")

    return results

# 使用例
docs = crawl_documentation("https://docs.python.org/3/", max_pages=50)
print(f"クロールしたページ数: {len(docs)}")

合成データ生成

公開データが特定のドメインを十分にカバーできない場合、合成データ生成（Synthetic Data Generation）技術を使用できます。DeepSeek などの強力なモデルを使用して高品質なトレーニングデータを生成します：

from openai import OpenAI

client = OpenAI(
    api_key="sk-your-api-key",
    base_url="https://api.deepseek.com/v1",
)

def generate_synthetic_data(topic, num_samples=10):
    """DeepSeek を使用して特定ドメインの合成データを生成"""
    prompt = f"""「{topic}」に関する高品質な指示データを {num_samples} 件生成してください。

各データには以下が含まれます：
- instruction: 明確で具体的な指示または質問
- input: 補足コンテキスト（空文字列でも可）
- output: 専門的で正確かつ詳細な回答

出力形式は JSON 配列です。"""

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
temperature=0.8,
        max_tokens=4096,
    )
    return response.choices[0].message.content

# Python機械学習関連データの生成
synthetic = generate_synthetic_data("Python機械学習", num_samples=10)
print(synthetic)

データコンプライアンス注意事項

公開データセットやWebスクレイピングデータを使用する際は、必ずデータライセンス契約を確認してください。一部のデータセット（ShareGPTなど）には特定の使用制限があります。Webスクレイピングはrobots.txtプロトコルを遵守し、スクレイピング頻度を制御して、対象サイトに負担をかけないようにしてください。個人のプライバシー情報を含むデータは、匿名化処理が必要です。

第3章

データクリーニングと重複排除

データクリーニングは、データエンジニアリングにおいて最も時間がかかるが最も重要なステップです。高品質なデータクリーニングは、モデルトレーニングの結果を大幅に向上させることができ、品質フィルタリング、重複排除、機密情報のフィルタリングなどが含まれます。

品質フィルタリングパイプライン

完全な品質フィルタリングパイプラインには、複数の次元でのフィルタリングが含まれます：

import re
from typing import List, Dict

class DataQualityFilter:
    """多次元データ品質フィルター"""

    def __init__(self, min_length=20, max_length=8000, min_words=5):
        self.min_length = min_length
        self.max_length = max_length
        self.min_words = min_words

    def filter_by_length(self, text: str) -> bool:
        """長さフィルタリング：短すぎるまたは長すぎるテキストはノイズの可能性"""
        return self.min_length <= len(text) <= self.max_length

    def filter_by_word_count(self, text: str) -> bool:
        """単語数フィルタリング：断片化しすぎたテキストをフィルタリング"""
        words = text.split()
        return len(words) >= self.min_words

    def filter_repetitive(self, text: str, threshold=0.3) -> bool:
        """重複コンテンツフィルタリング：重複n-gram比率を検出"""
        if len(text) < 100:
            return True
        words = text.split()
        unigrams = len(set(words))
        if unigrams / len(words) < threshold:
            return False  # 繰り返し率が高すぎる
        return True

    def filter_special_chars(self, text: str, max_ratio=0.3) -> bool:
        """特殊文字フィルタリング：文字化けや異常なテキストをフィルタリング"""

        special = len(re.findall(r'[^\w\s\u4e00-\u9fff.,;:!?()\-\+\=]', text))
        return special / max(len(text), 1) < max_ratio

    def filter_empty_output(self, sample: Dict) -> bool:
        """空出力フィルター：出力が短すぎるか、プレースホルダーのみを含む"""
        output = sample.get("output", "")
        placeholder_patterns = [
            r'^(sorry|unfortunately|i cannot|as an ai)',
            r'^(抱歉|对不起|作为.*AI|我无法)',
        ]
        for pattern in placeholder_patterns:
            if re.match(pattern, output.strip().lower()):
                return False
        return len(output.strip()) > 20

    def apply_all(self, samples: List[Dict]) -> List[Dict]:
        """すべてのフィルターを適用"""
        filtered = []
        for s in samples:
            text = s.get("instruction", "") + " " + s.get("output", "")
            if all([
                self.filter_by_length(text),
                self.filter_by_word_count(text),
                self.filter_repetitive(text),
                self.filter_special_chars(text),
                self.filter_empty_output(s),
            ]):
                filtered.append(s)
        print(f"フィルター前: {len(samples)} 件, フィルター後: {len(filtered)} 件")
        return filtered

MinHash LSH 重複排除

MinHash + LSH（Locality-Sensitive Hashing）は、業界標準の大規模重複排除手法であり、近似重複文書を効率的に検出できます：

from datasketch import MinHash, MinHashLSH
import re

def tokenize_chinese(text):
    """中国語トークン化：文字2-gramで分割"""



    # 簡易版トークン化、本番ではjiebaまたはpkusegを推奨
    text = re.sub(r'[^\u4e00-\u9fff\w]', ' ', text.lower())
    # 2-gramトークン化
    return [text[i:i+2] for i in range(len(text)-1)]

def deduplicate_with_minhash(samples, num_perm=128, threshold=0.8):
    """MinHash LSHを使用した近似重複排除"""
    lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
    unique_samples = []

    for idx, sample in enumerate(samples):
        text = sample.get("instruction", "") + " " + sample.get("output", "")
        tokens = tokenize_chinese(text)

        if len(tokens) < 10:
            continue

        m = MinHash(num_perm=num_perm)
        for token in tokens:
            m.update(token.encode("utf-8"))

        # 近似重複が既に存在するか確認
        if len(lsh.query(m)) == 0:
            lsh.insert(str(idx), m)
            unique_samples.append(sample)

    print(f"重複排除前: {len(samples)} 件, 重複排除後: {len(unique_samples)} 件")
    return unique_samples

意味的重複排除

MinHashは文字レベルの重複排除に適していますが、意味的に類似しているが表現が異なるデータには、埋め込みベクトルを使用した意味的重複排除が必要です：

from sentence_transformers import SentenceTransformer
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

def semantic_deduplication(samples, model_name="BAAI/bge-small-zh-v1.5",
                             threshold=0.95, batch_size=256):
    """意味的埋め込みに基づく重複排除"""
    model = SentenceTransformer(model_name)

    # すべてのテキストを抽出



    texts = [s.get("instruction", "") for s in samples]

    # バッチエンコード
    embeddings = model.encode(texts, batch_size=batch_size, show_progress_bar=True)

    # 類似度行列を計算し、重複をマーク
    unique_indices = []
    seen = set()

    for i in range(len(embeddings)):
        if i in seen:
            continue
        unique_indices.append(i)

        # 現在のサンプルと類似度が高いサンプルを見つける
        sims = cosine_similarity([embeddings[i]], embeddings[i+1:])[0]
        duplicates = np.where(sims > threshold)[0] + i + 1
        for d in duplicates:
            seen.add(int(d))

    print(f"意味的類似重複除去: {len(samples)} -> {len(unique_indices)} 件")
    return [samples[i] for i in unique_indices]

機密情報のフィルタリング

トレーニングデータでは、個人のプライバシー情報や機密コンテンツをフィルタリングする必要があります：

import re

class SensitiveContentFilter:
    """機密情報フィルター"""

    # 一般的な機密情報の正規表現パターン
    PATTERNS = {
        "email": r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
        "phone_cn": r'1[3-9]\d{9}',
        "id_card": r'\d{17}[\dXx]',
        "ip_address": r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
        "url": r'https?://[^\s<>"{}|\\^`\[\]]+',
        "api_key": r'(?:sk|api[_-]?key|token)[=:]\s*[\w-]+',
    }

    def has_sensitive(self, text: str) -> bool:
        """機密情報が含まれているか検出"""
        for name, pattern in self.PATTERNS.items():



            if re.search(pattern, text):
                return True
        return False

    def mask_sensitive(self, text: str) -> str:
        """マスキング：機密情報をプレースホルダーに置換"""
        masked = text
        replacements = {
            "email": "[EMAIL]",
            "phone_cn": "[PHONE]",
            "id_card": "[ID_CARD]",
            "ip_address": "[IP]",
            "api_key": "[API_KEY]",
        }
        for name, replacement in replacements.items():
            if name in self.PATTERNS:
                masked = re.sub(self.PATTERNS[name], replacement, masked)
        return masked

# 使用例
filter = SensitiveContentFilter()
text = "admin@example.com に連絡するか、13800138000 にお電話ください"
print(f"機密情報を含む: {filter.has_sensitive(text)}")  # True
print(f"マスク後: {filter.mask_sensitive(text)}")

第4章

データ形式とアノテーション

トレーニング段階によって必要なデータ形式は異なります。SFTはInstruction-Input-Output形式を使用し、ChatMLは対話シナリオに使用され、RLHF/DPOでは選好比較データが必要です。これらの形式を理解することは、高品質なデータセットを構築するための基礎です。

SFTデータ形式（Instruction-Input-Output）

SFT（Supervised Fine-Tuning）データは最も基本的なトレーニングデータ形式です。各データには命令、オプションの入力、期待される出力が含まれます：

{
  "instruction": "Pythonでクイックソートアルゴリズムを実装",
  "input": "",
  "output": "def quicksort(arr):\n    if len(arr) <= 1:\n        return arr\n    pivot = arr[len(arr) // 2]\n    left = [x for x in arr if x < pivot]\n    middle = [x for x in arr if x == pivot]\n    right = [x for x in arr if x > pivot]\n    return quicksort(left) + middle + quicksort(right)"
}

ChatML対話形式

ChatML（Chat Markup Language）はOpenAIが定義した対話形式で、SFTトレーニングでも広く使用されています。複数ターンの対話を標準形式に構造化します：

def format_chatml(conversations):
    """会話リストをChatML形式の文字列に変換"""
    formatted = ""
    for turn in conversations:
        role = turn["role"]
        content = turn["content"]
        if role == "system":
            formatted += f"<|im_start|>system\n{content}<|im_end|>\n"
        elif role == "user":
            formatted += f"<|im_start|>user\n{content}<|im_end|>\n"
        elif role == "assistant":
            formatted += f"<|im_start|>assistant\n{content}<|im_end|>\n"
    return formatted.strip()

# 会話例
conversation = [
    {"role": "system", "content": "あなたはプロのPythonプログラミングアシスタントです。"},
    {"role": "user", "content": "CSVファイルを読み取るには？"},
    {"role": "assistant", "content": "pandasを使用するとCSVファイルを簡単に読み取れます：\nimport pandas as pd\ndf = pd.read_csv('data.csv')"},

]
]

chatml = format_chatml(conversation)
print(chatml)

RLHF選好データ形式

RLHF（Reinforcement Learning from Human Feedback）とDPO（Direct Preference Optimization）では、同じプロンプトに対してどの応答が良いかをアノテーションする選好比較データが必要です：

# DPOデータ形式
dpo_sample = {
    "prompt": "機械学習とは何か説明してください",
    "chosen": "機械学習は人工知能の一分野であり、明示的なプログラミングなしにコンピュータがデータからパターンを学習できるようにします。一般的な手法には、教師あり学習、教師なし学習、強化学習があります。",
    "rejected": "機械学習は機械に物事を学習させることです。",
}

# RLHF比較データ形式
rlhf_comparison = {
    "prompt": "春についての詩を書いてください",
    "responses": [
        {"text": "春風が頬を撫で、柳は煙のよう...", "score": 4.5},
        {"text": "春が来て、花が咲いた。", "score": 1.0},
    ]
}

データ形式の選択に関する推奨事項

基本的なSFTトレーニングにはInstruction-Input-Output形式で十分です。複数ターンの対話シナリオではChatML形式を推奨します。RLHFまたはDPOトレーニングが必要な場合は、選好比較データを準備する必要があります。DeepSeekシリーズのモデルはAlpaca形式とChatML形式の両方をサポートしています。

第5章

データ拡張技術

既存データが不十分な場合、データ拡張技術を利用することで、少数のシードデータから大量の高品質なトレーニングデータを生成できます。Self-Instruct と Evol-Instruct は最も主流な2つの手法です。

Self-Instruct 自己生成手法

Self-Instruct の核となる考え方は、強力なモデルを使ってシードタスクから新しい指示-出力ペアを自動生成することです：

import json
import random
from openai import OpenAI

client = OpenAI(
    api_key="sk-your-api-key",
    base_url="https://api.deepseek.com/v1",
)

def generate_instructions(seed_tasks, num_to_generate=50):
    """シードタスクに基づいて新しい指示を生成"""
    # シードタスクをランダムにサンプリングしてコンテキストとする
    seed_context = random.sample(seed_tasks, min(8, len(seed_tasks)))
    seed_text = "\n".join([f"- {t['instruction']}" for t in seed_context])

    prompt = f"""あなたはデータアノテーションの専門家です。以下のシードタスクに基づいて、{num_to_generate} 個の新しい多様な指示を生成してください。

シードタスクの例：
{seed_text}

要件：
1. 新しい指示は異なる分野と難易度をカバーすること
2. 指示は明確で具体的、実行可能であること
3. シードタスクとの重複を避けること
4. 出力形式はJSON配列で、各項目にはinstructionフィールドが含まれること

JSON配列のみを出力し、他のテキストを含めないでください。"""

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.9,
        max_tokens=4096,
    )

    try:
        new_instructions = json.loads(response.choices[0].message.content)
        return new_instructions
    except json.JSONDecodeError:
        return []

def generate_output(instruction):
    """指定された指示に対して高品質な出力を生成"""
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": instruction}],
        temperature=0.3,
        max_tokens=2048,
    )
    return response.choices[0].message.content

# 使用例
seed_tasks = [
{"instruction": "Pythonでバブルソートを実装する"},
    {"instruction": "RESTful APIとは何か説明する"},
    {"instruction": "フィボナッチ数列を計算する関数を書く"},
]
new_instructions = generate_instructions(seed_tasks, num_to_generate=10)
print(f"{len(new_instructions)} 個の新しい指示を生成しました")

Evol-Instruct 進化生成

Evol-Instruct は、指示の複雑さを段階的に増やすことで、より挑戦的なデータを生成します。DeepSeek チームはトレーニングで同様の進化戦略を多用しました：

def evolve_instruction(instruction, evolution_type="deepen"):
    """指示を進化させる：深さ、広さ、複雑さを増やす"""
    evolution_prompts = {
        "deepen": """以下の指示をより深く、専門的に書き換えてください。基本原理、実装の詳細、エッジケースについての議論を追加してください。

元の指示：{instruction}

深化後の指示：""",

        "broaden": """以下の指示をより広く書き換えて、複数の関連するサブトピックや異なるシナリオをカバーするようにしてください。

元の指示：{instruction}

拡張後の指示：""",

        "increase_reasoning": """以下の指示を、多段階の推論を必要とする複雑なタスクに書き換えてください。

元の指示：{instruction}

推論要件を増やした指示：""",
    }

    prompt = evolution_prompts.get(evolution_type, evolution_prompts["deepen"])
    prompt = prompt.format(instruction=instruction)

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
        max_tokens=1024,
    )
    return response.choices[0].message.content

# 進化の例
original = "データベースインデックスとは何か説明する"
deepened = evolve_instruction(original, "deepen")
print(f"元の指示: {original}")
print(f"深化後: {deepened}")

バックトランスレーションによるデータ拡張

バックトランスレーション（Back Translation）は、多言語データ拡張の古典的な手法であり、翻訳と逆翻訳を通じて意味的に等価で表現の異なるデータを生成します：

def back_translate(text, source_lang="中国語", pivot_lang="英語"):
    """バックトランスレーション：中国語 -> 英語 -> 中国語、意味的に等価なバリアントを生成"""

    # ステップ1: ターゲット言語に翻訳
    translate_prompt = f"以下の{source_lang}テキストを{pivot_lang}に翻訳し、翻訳結果のみを出力してください：\n{text}"

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": translate_prompt}],
        temperature=0.3,
    )
    translated = response.choices[0].message.content

    # ステップ2: ソース言語に翻訳し直す
    back_prompt = f"以下の{pivot_lang}テキストを{source_lang}に翻訳し、翻訳結果のみを出力してください：\n{translated}"

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": back_prompt}],
        temperature=0.3,
    )
    return response.choices[0].message.content

# 使用例
original = "深層学習は、多層ニューラルネットワークを使用してデータの表現を学習する機械学習のサブフィールドです。"
augmented = back_translate(original)
print(f"原文: {original}")
print(f"拡張: {augmented}")

第6章

DeepSeek 特定データ準備

DeepSeek-R1 の推論データには特別な形式要件があり、thinking タグと CoT（Chain of Thought）推論チェーンが含まれます。この章では、DeepSeek モデル用のデータ準備方法を詳しく説明します。

DeepSeek-R1 推論データ形式

DeepSeek-R1 の重要な革新は、トレーニングデータに明示的な思考プロセス（thinking/reasoning）が含まれていることです。以下は標準的な R1 推論データ形式です：

# DeepSeek-R1 推論データ形式
r1_format_sample = {
    "messages": [
        {
            "role": "user",
            "content": "直方体の縦・横・高さがそれぞれ 3cm、4cm、5cm です。表面積と体積を求めてください。"
        },
        {
            "role": "assistant",
            "content": "<think>これは直方体の表面積と体積の計算問題です。\n表面積 = 2*(縦*横 + 縦*高さ + 横*高さ)\n体積 = 縦*横*高さ\n\n数値を代入：\n縦=3, 横=4, 高さ=5\n表面積 = 2*(3*4 + 3*5 + 4*5) = 2*(12+15+20) = 2*47 = 94\n体積 = 3*4*5 = 60\n\n検証：表面積と体積は単位が異なりますが、数値計算は正しいです。</think>\n\n直方体の表面積は 94 平方センチメートル、体積は 60 立方センチメートルです。"
        }
    ]
}

CoT データ構築

Chain of Thought データを構築する鍵は、モデルに「まず考えてから答える」ことを学習させることです。以下のコードは、数学問題に対して CoT 推論チェーンを自動生成します：

def generate_cot_data(question, answer):
    """質問と回答に対して CoT 推論プロセスを生成する"""
    prompt = f"""以下の数学問題に対して、詳細なステップバイステップの推論プロセスを生成してください。

問題：{question}
正解：{answer}

以下の形式で出力してください：
<think>
（詳細なステップバイステップの推論。使用する公式、中間ステップ、検証プロセスを含む）
</think>

（最終回答、簡潔明瞭に）

推論プロセスが明確で完全で検証可能であることを確認してください。"""

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=4096,
    )
    return response.choices[0].message.content

# 数学問題の CoT を生成
cot_result = generate_cot_data(
    question="x^2 - 5x + 6 = 0 のとき、x の値を求めよ。",
    answer="x = 2 または x = 3"
)
print(cot_result)

コードデータ準備

コードデータは DeepSeek トレーニングデータの重要な構成要素です。高品質なコードデータには、問題の説明、コード実装、コメントが含まれるべきです：

def format_code_data(problem, code, language="python", difficulty="medium"):
    """コードトレーニングデータをフォーマットする"""
    return {
        "messages": [



            {
                "role": "system",
                "content": f"あなたはプロの{language.upper()}プログラミングアシスタントです。明確で効率的なコード実装を提供してください。"
            },
            {
                "role": "user",
                "content": problem
            },
            {
                "role": "assistant",
                "content": code
            }
        ],
        "metadata": {
            "language": language,
            "difficulty": difficulty,
            "type": "code_generation",
        }
    }

# 例
code_sample = format_code_data(
    problem="LRUキャッシュ（最も最近使用されていないもの）を実装し、get と put 操作を O(1) の時間複雑度でサポートします。",
    code="""from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.capacity = capacity
        self.cache = OrderedDict()

    def get(self, key: int) -> int:
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key: int, value: int) -> None:
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)""",
    difficulty="medium"
)

数学データ準備

数学推論データには、数式、導出ステップ、最終回答を含める必要があります。LaTeX形式は数学表現の標準です：

math_sample = {
    "messages": [
        {
            "role": "user",
            "content": "極限 \\lim_{x \\to 0} \\frac{\\sin x}{x} を求める"
        },
        {
            "role": "assistant",
            "content": "<think>これは古典的な極限問題です。\n\n方法1：ロピタルの定理を使用\nx→0 のとき、分子 sin x → 0、分母 x → 0 で、0/0 の不定形を満たします。\n分子と分母をそれぞれ微分します：\nlim(sin x / x) = lim(cos x / 1) = cos(0) = 1\n\n方法2：はさみうちの定理を使用\n0 < |x| < pi/2 のとき、cos x < sin x / x < 1 が成り立ちます。\nx→0 のとき、cos x → 1 なので、極限は 1 です。\n\n検証：この結果は正しいです。</think>\n\n\\lim_{x \\to 0} \\frac{\\sin x}{x} = 1"
        }
    ]
}

第7章

データ品質評価

データ品質評価はデータエンジニアリングの最後の防衛線です。多様性、難易度、指示の複雑さ、自動スコアリングなど複数の次元からデータセットの品質を評価し、トレーニングデータが期待される基準を満たしていることを確認します。

多様性評価

多様性評価により、データセットが十分に広い領域とタスクタイプをカバーしていることを確認します：

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
import numpy as np

def evaluate_diversity(samples, n_clusters=10):
    """データセットの多様性を評価する"""
    # 指示テキストを抽出
    instructions = [s.get("instruction", "") for s in samples]

    # TF-IDFベクトル化
    vectorizer = TfidfVectorizer(max_features=5000, ngram_range=(1, 2))
    vectors = vectorizer.fit_transform(instructions)

    # クラスタ分析
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    labels = kmeans.fit_predict(vectors.toarray())

    # 各クラスタのサンプル数を集計
    cluster_counts = np.bincount(labels)
    cluster_ratio = cluster_counts / len(samples)

    # 多様性スコアを計算（クラスタ分布に基づくエントロピー）
    entropy = -np.sum(cluster_ratio * np.log(cluster_ratio + 1e-10))
    max_entropy = np.log(n_clusters)
    diversity_score = entropy / max_entropy

    print(f"クラスタ数: {n_clusters}")
    print(f"各クラスタのサンプル数: {cluster_counts}")
    print(f"多様性スコア: {diversity_score:.4f} (1.0 = 完全な一様分布)")

    return {
        "diversity_score": diversity_score,
        "cluster_counts": cluster_counts.tolist(),
        "cluster_ratio": cluster_ratio.tolist(),
    }

難易度評価

データセット内の各指示の難易度を評価し、データの難易度分布が合理的であることを確認します：

def evaluate_difficulty(instruction):



    """LLMを使用して指示の難易度を評価する"""
    prompt = f"""以下の指示の難易度を評価し、1〜5のスコアを返してください（1=非常に簡単、5=非常に難しい）。

評価基準：
- 1点：単純な事実に関する質問で、推論は不要
- 2点：基本的な理解または簡単な操作が必要
- 3点：複数ステップの思考または総合的な知識が必要
- 4点：高度な推論または専門知識が必要
- 5点：深い推論、創造性、または専門家レベルの知識が必要

指示：{instruction}

スコアのみを出力してください（1〜5の整数）。"""

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=10,
    )
    try:
        return int(response.choices[0].message.content.strip())
    except ValueError:
        return 3

def analyze_difficulty_distribution(samples):
    """データセットの難易度分布を分析する"""
    from collections import Counter

    difficulties = []
    for s in samples[:100]:  # サンプル評価
        score = evaluate_difficulty(s.get("instruction", ""))
        difficulties.append(score)

    dist = Counter(difficulties)
    for level in range(1, 6):
        print(f"難易度 {level}: {dist.get(level, 0)} 件 ({dist.get(level,0)/len(difficulties)*100:.1f}%)")

    print(f"\n平均難易度: {np.mean(difficulties):.2f}")
    return difficulties

指示の複雑さスコアリング

指示の複雑さスコアリングは、複数の側面から指示の品質を評価します：

評価側面	説明	評価方法
明確性	指示が明確で曖昧さがないか	LLMスコア 1-5
完全性	指示が十分な文脈を含んでいるか	情報密度の計算
実行可能性	指示が完了可能かどうか	LLMによる実行可能性の判断
創造性	指示が創造的思考を促すかどうか	LLMスコア 1-5

自動品質スコアリング

LLMを判定者として使用し、データ品質を自動的にスコアリングします：

def auto_quality_score(sample):
    """LLMを使用して単一データの品質を自動評価する"""
    instruction = sample.get("instruction", "")
    output = sample.get("output", "")

    prompt = f"""以下のQ&Aペアの品質を1〜10点で評価してください。

評価の側面：
1. 指示が明確で曖昧さがないか（1〜3点）
2. 回答が正確で完全か（1〜3点）
3. 回答が専門的で役立つか（1〜2点）
4. 回答の長さが適切か（1〜2点）

指示：{instruction}
回答：{output}

JSON形式で出力してください：
{{"total_score": X, "reasons": ["理由1", "理由2"]}}"""

    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        max_tokens=512,
    )
    try:
        return json.loads(response.choices[0].message.content)
    except json.JSONDecodeError:
        return {"total_score": 5, "reasons": ["スコアを解析できません"]}

# 使用例
sample = {
    "instruction": "Pythonで二分探索アルゴリズムを実装し、境界条件を処理する",
    "output": "def binary_search(arr, target):\n    left, right = 0, len(arr)-1\n    while left <= right:\n        mid = (left+right)//2\n        if arr[mid] == target:\n            return mid\n        elif arr[mid] < target:\n            left = mid+1\n        else:\n            right = mid-1\n    return -1"
}
score = auto_quality_score(sample)
print(f"品質スコア: {score}")

第8章

データ混合戦略

異なる分野のデータを特定の比率で混合することで、バランスの取れた総合的なモデルを訓練できます。データ混合戦略はモデルの汎用能力と専門能力に直接影響します。

典型的なデータ混合比率

以下はDeepSeek系モデルの訓練で一般的に使用されるデータ混合比率の参考です：

データカテゴリ	推奨比率	説明
一般的な対話	30-40%	日常会話、Q&A、雑談。基本的な対話能力を保証
コード生成	15-20%	Python/JS/Java/C++などの主要言語のコード
数学的推論	10-15%	代数、幾何、微積分、確率統計
論理的推論	10-15%	論理パズル、推論チェーン、多段階推論
クリエイティブライティング	5-10%	詩、物語、コピーライティング、脚本
専門分野	5-10%	医療、法律、金融などの垂直分野
安全性アライメント	3-5%	有害なリクエストの拒否、価値観の整合

データ混合のコード実装

import random
from typing import List, Dict

def mix_datasets(datasets: Dict[str, List], ratios: Dict[str, float],
                  total_size: int = 10000, shuffle: bool = True):
    """比率に基づいて複数のデータセットを混合する"""
    # 比率の合計を検証
    total_ratio = sum(ratios.values())
if abs(total_ratio - 1.0) > 0.01:
        print(f"警告: 比率の合計は {total_ratio:.2f} です。1.0に調整することをお勧めします")

    mixed = []
    stats = {}

    for name, ratio in ratios.items():
        if name not in datasets:
            continue

        # 比率に基づいてサンプリング
        sample_size = int(total_size * ratio)
        available = datasets[name]

        # データが不足している場合は全て使用
        if len(available) < sample_size:
            sampled = available.copy()
            print(f"  {name}: データ不足のため、全 {len(available)} 件を使用")
        else:
            sampled = random.sample(available, sample_size)

        # ソースマーカーを追加
        for s in sampled:
            s["source_dataset"] = name

        mixed.extend(sampled)
        stats[name] = len(sampled)

    if shuffle:
        random.shuffle(mixed)

    print(f"\nデータセットの混合が完了しました。合計 {len(mixed)} 件")
    for name, count in stats.items():
        print(f"  {name}: {count} 件 ({count/len(mixed)*100:.1f}%)")

    return mixed

# 使用例
datasets = {
    "general_chat": [...],     # 一般的な対話データ
    "code": [...],              # コードデータ
    "math": [...],              # 数学データ
    "reasoning": [...],         # 推論データ
    "creative": [...],         # クリエイティブライティングデータ
}

ratios = {

    "general_chat": 0.35,
    "code": 0.20,
    "math": 0.15,
    "reasoning": 0.15,
    "creative": 0.10,
    "safety": 0.05,
}

training_data = mix_datasets(datasets, ratios, total_size=10000)

データアニーリング戦略

データアニーリングは、トレーニング中にデータ混合比率を動的に調整するトレーニング戦略です。初期段階では多様なデータで基礎能力を訓練し、後期には高品質・高難度データの割合を徐々に増やします：

ウォームアップ段階（最初の20%のトレーニングステップ）：一般的な会話データが50%を占め、主に簡単な指示で、モデルの基礎的な対話能力を構築します。
メインのトレーニング段階（20%-80%のトレーニングステップ）：コードと推論データの割合を徐々に増やし、中程度の難易度のタスクを導入します。
アニーリング段階（最後の20%のトレーニングステップ）：高品質・高難度データを大幅に増やし、一般的な会話データを減らし、複雑なタスクでのモデルのパフォーマンスを向上させます。

DeepSeekチームはトレーニングで同様のアニーリング戦略を使用しており、これはDeepSeek-R1が推論能力で優れている重要な理由の一つです。

第9章

大規模データ処理

データ量が数百万から数億に達すると、単一マシンでの処理は不可能になります。この章では、SparkやRayなどの分散フレームワークを使用した大規模データ処理と、データバージョン管理について紹介します。

Rayを使用した並列処理

Rayは軽量な分散コンピューティングフレームワークで、特にPythonエコシステムのデータ処理タスクに適しています：

import ray
from typing import List, Dict

# Rayの初期化
ray.init(num_cpus=8)

# 処理関数をRayリモートタスクとして宣言
@ray.remote
def process_batch(batch: List[Dict]) -> List[Dict]:
    """バッチデータの処理：クリーニング、重複排除、品質フィルタリング"""
    filter = DataQualityFilter()
    # 第3章で定義されたすべてのフィルタリングロジックが含まれると仮定
    return filter.apply_all(batch)

def parallel_process(samples: List[Dict], batch_size=1000):
    """Rayを使用して大規模データを並列処理"""
    # バッチに分割
    batches = [samples[i:i+batch_size] for i in range(0, len(samples), batch_size)]

    # タスクを並列送信
    futures = [process_batch.remote(batch) for batch in batches]

    # 結果を収集
    results = []
    for future in futures:
        results.extend(ray.get(future))

    print(f"並列処理完了: {len(samples)} -> {len(results)} 件")
    return results

# 使用例
# results = parallel_process(all_samples, batch_size=2000)
# ray.shutdown()

PySparkを使用した分散処理

PySparkはTB級の大規模データを処理する標準ツールであり、DataFrame APIとSQL操作をサポートしています：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, length, udf
from pyspark.sql.types import BooleanType

# Sparkセッションの作成
spark = SparkSession.builder \



    .appName("DeepSeekDataProcessing") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# 大規模JSONデータの読み込み
df = spark.read.json("hdfs://data/deepseek/*.jsonl")

# 基本フィルタリング
df_clean = df.filter(
    (length(col("instruction")) > 20) &
    (length(col("output")) > 20) &
    (length(col("instruction")) < 8000)
)

# SQLを使用した複雑なフィルタリング
df_clean.createOrReplaceTempView("data")
df_quality = spark.sql("""
    SELECT *, LENGTH(instruction) + LENGTH(output) AS total_length
    FROM data
    WHERE instruction IS NOT NULL
      AND output IS NOT NULL
      AND LENGTH(output) / LENGTH(instruction) > 0.5
""")

# 統計情報
print(f"処理前: {df.count()} 件")
print(f"処理後: {df_quality.count()} 件")

# 処理済みデータの書き込み
df_quality.write.mode("overwrite").json("hdfs://data/deepseek_cleaned/")

大ファイルのストリーム処理

超大なJSONLファイルの場合、メモリオーバーフローを避けるためにストリーム処理を使用します：

import json
import ijson
import gzip

def stream_process_jsonl(filepath, output_path, batch_size=10000):
    """大型JSONLファイルのストリーム処理"""
    opener = gzip.open if filepath.endswith('.gz') else open
    batch = []
    total_processed = 0
    total_written = 0
    filter = DataQualityFilter()

    with opener(filepath, 'r', encoding='utf-8') as infile, \
         open(output_path, 'w', encoding='utf-8') as outfile:

        for line in infile:
            try:
                sample = json.loads(line.strip())
                batch.append(sample)
                total_processed += 1
            except json.JSONDecodeError:
                continue




            if len(batch) >= batch_size:
                # バッチ処理
                cleaned = filter.apply_all(batch)
                for s in cleaned:
                    outfile.write(json.dumps(s, ensure_ascii=False) + '\n')
                total_written += len(cleaned)
                batch = []

                if total_processed % 100000 == 0:
                    print(f"処理済み {total_processed} 件、書き込み {total_written} 件")

        # 残りのバッチを処理
        if batch:
            cleaned = filter.apply_all(batch)
            for s in cleaned:
                outfile.write(json.dumps(s, ensure_ascii=False) + '\n')
            total_written += len(cleaned)

    print(f"処理完了: {total_processed} -> {total_written} 件")

データバージョン管理（DVC）

DVC（Data Version Control）はデータサイエンスのGitであり、データセットのバージョン管理に使用されます：

# DVCの初期化
dvc init

# データセットの追跡
dvc add data/training_data.jsonl

# Gitにコミット
git add data/training_data.jsonl.dvc data/.gitignore
git commit -m "初期トレーニングデータセットv1.0を追加"

# データ更新後、新しいバージョンを作成
dvc add data/training_data.jsonl
git add data/training_data.jsonl.dvc
git commit -m "トレーニングデータを更新: 10Kコードデータを追加"

# 履歴バージョンに切り替え
git checkout "<commit-hash>"
dvc checkout

# データ変更履歴を表示
dvc diff HEAD~1

第10章

本番グレードのデータパイプライン

これまでのすべてのステップを統合し、継続的なデータ収集、自動品質監視、データドリフト検出、完全なパイプラインコードを含む本番グレードのデータパイプラインを構築します。

完全なデータパイプラインアーキテクチャ

データ収集層：公開データセットの更新を定期的に取得、新しいコンテンツのクロール、合成データAPIの呼び出し
データクリーニング層：品質フィルタリング、重複排除、機密情報のフィルタリング、形式の標準化
データ拡張層：Self-Instruct生成、Evol-Instruct進化、逆翻訳
品質評価層：多様性スコアリング、難易度評価、自動品質スコアリング
データ混合層：比率に基づく混合、データアニーリング戦略、バージョン管理
監視・アラート層：データドリフト検出、品質トレンド監視、異常アラート

完全なパイプラインコード

"""DeepSeek データエンジニアリングパイプライン — 完全実装"""
import json
import os
import logging
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)


class DataPipeline:
    """DeepSeek データエンジニアリングパイプライン"""

    def __init__(self, config: Dict):
        self.config = config
        self.data_dir = Path(config.get("data_dir", "./data"))
        self.output_dir = Path(config.get("output_dir", "./output"))
        self.data_dir.mkdir(exist_ok=True)
        self.output_dir.mkdir(exist_ok=True)
        self.metrics = {}

    def step1_collect(self) -> List[Dict]:
        """ステップ1: データ収集"""
        logger.info("Step 1: データ収集")
        all_data = []

        # 公開データセットの読み込み
        from datasets import load_dataset
        try:
            alpaca = load_dataset("tatsu-lab/alpaca", split="train")
            all_data.extend([dict(s) for s in alpaca])
logger.info(f"  Alpacaを読み込み: {len(alpaca)}件")
        except Exception as e:
            logger.warning(f"  Alpacaの読み込みに失敗: {e}")

        # ローカルJSONLファイルの読み込み
        for f in self.data_dir.glob("*.jsonl"):
            with open(f, 'r', encoding='utf-8') as fh:
                for line in fh:
                    try:
                        all_data.append(json.loads(line.strip()))
                    except json.JSONDecodeError:
                        continue
            logger.info(f"  {f.name}を読み込み")

        self.metrics["collected"] = len(all_data)
        return all_data

    def step2_clean(self, data: List[Dict]) -> List[Dict]:
        """ステップ2: データクリーニングと重複排除"""
        logger.info("Step 2: データクリーニングと重複排除")

        # 長さフィルタリング
        def is_valid(s):
            inst = s.get("instruction", "")
            out = s.get("output", "")
            return 20 <= len(inst) <= 8000 and len(out) >= 20

        cleaned = [s for s in data if is_valid(s)]
        logger.info(f"  長さフィルタリング: {len(data)} -> {len(cleaned)}件")

        # 重複排除（instructionに基づく完全一致）
        seen = set()
        deduped = []
        for s in cleaned:
            key = s.get("instruction", "").strip().lower()

            if key not in seen:
                seen.add(key)
                deduped.append(s)
        logger.info(f"  重複排除: {len(cleaned)} -> {len(deduped)} 件")

        self.metrics["cleaned"] = len(deduped)
        return deduped

    def step3_quality_check(self, data: List[Dict]) -> List[Dict]:
        """ステップ3：品質評価とフィルタリング"""
        logger.info("Step 3: 品質評価")

        # 出力が短すぎる、または拒否パターンを含むデータをフィルタリング
        reject_patterns = [
            "申し訳ありません、できません", "AIとして", "ごめんなさい",
            "I cannot", "I'm sorry", "as an AI",
        ]
        def is_quality(s):
            output = s.get("output", "")
            if len(output) < 50:
                return False
            for p in reject_patterns:
                if p in output[:100]:
                    return False
            return True

        quality = [s for s in data if is_quality(s)]
        logger.info(f"  品質フィルタ: {len(data)} -> {len(quality)} 件")

        self.metrics["quality_passed"] = len(quality)
        return quality

    def step4_export(self, data: List[Dict], filename: str = None):
        """ステップ4：処理済みデータのエクスポート"""
        logger.info("Step 4: データエクスポート")

        if filename is None:



            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            filename = f"deepseek_data_{timestamp}.jsonl"

        output_path = self.output_dir / filename
        with open(output_path, 'w', encoding='utf-8') as f:
            for s in data:
                f.write(json.dumps(s, ensure_ascii=False) + '\n')

        logger.info(f"  エクスポート先: {output_path} ({len(data)} 件)")
        return output_path

    def run(self):
        """パイプライン全体を実行"""
        logger.info("=" * 50)
        logger.info("DeepSeek データエンジニアリングパイプライン起動")
        logger.info("=" * 50)

        start = datetime.now()

        data = self.step1_collect()
        data = self.step2_clean(data)
        data = self.step3_quality_check(data)
        output_path = self.step4_export(data)

        elapsed = (datetime.now() - start).total_seconds()

        # 最終レポートを出力
        logger.info("=" * 50)
        logger.info("パイプライン実行完了")
        logger.info(f"  総所要時間: {elapsed:.1f} 秒")
        for k, v in self.metrics.items():
            logger.info(f"  {k}: {v}")
        logger.info(f"  出力ファイル: {output_path}")
        logger.info("=" * 50)

        return data


# ========== メインプログラム ==========
if __name__ == "__main__":
    config = {
        "data_dir": "./data",
        "output_dir": "./output",
    }



    pipeline = DataPipeline(config)
    pipeline.run()

データドリフト検出

データドリフト（Data Drift）とは、データ分布が時間とともに変化することを指します。本番環境では、データ品質が期待から逸脱していないかを継続的に監視する必要があります：

import numpy as np
from scipy.stats import ks_2samp

def detect_data_drift(reference_data: List[Dict], current_data: List[Dict],
                       threshold=0.05):
    """現在のデータが参照データに対してドリフトしているか検出"""

    # 1. 指示文の長さ分布の比較
    ref_lengths = [len(s.get("instruction", "")) for s in reference_data]
    cur_lengths = [len(s.get("instruction", "")) for s in current_data]

    stat, p_value = ks_2samp(ref_lengths, cur_lengths)

    drift_detected = p_value < threshold
    print(f"KS検定: stat={stat:.4f}, p={p_value:.4f}")
    print(f"平均長: 参照={np.mean(ref_lengths):.0f}, 現在={np.mean(cur_lengths):.0f}")

    if drift_detected:
        print("警告: データドリフトを検出しました！")
    else:
        print("データ分布は正常で、有意なドリフトは検出されませんでした。")

    return {
        "drift_detected": drift_detected,
        "p_value": p_value,
        "ref_mean": np.mean(ref_lengths),
        "cur_mean": np.mean(cur_lengths),
    }

自動品質監視

定期的なタスクを使用して、データ品質メトリクスを継続的に監視します：

import schedule
import time

def monitoring_job():
    """定期監視タスク"""
    logger.info("[監視] データ品質チェックを開始...")

    # パイプラインを実行
    pipeline = DataPipeline(config)



    try:
        data = pipeline.run()

        # 品質メトリクスを確認
        quality_ratio = pipeline.metrics.get("quality_passed", 0) / \
                        max(pipeline.metrics.get("collected", 1), 1)

        if quality_ratio < 0.5:
            logger.warning(f"[警告] データ品質合格率が低すぎます: {quality_ratio:.2%}")
        else:
            logger.info(f"[正常] データ品質合格率: {quality_ratio:.2%}")

    except Exception as e:
        logger.error(f"[エラー] 監視タスクが失敗しました: {e}")

# 定期タスクを設定: 毎日午前2時に実行
# schedule.every().day.at("02:00").do(monitoring_job)
#
# while True:
#     schedule.run_pending()
#     time.sleep(60)

パイプラインの実行

# 1. データディレクトリを準備
mkdir data
# 生データファイルを data/ ディレクトリに配置

# 2. 依存関係をインストール
pip install datasets schedule

# 3. パイプラインを実行
python data_pipeline.py

# 4. 出力を確認
# 処理済みデータは output/ ディレクトリにあります
ls output/

本番環境の推奨事項

Apache Airflow または Prefect を使用してパイプラインタスクをオーケストレーション
品質メトリクスを Prometheus + Grafana にプッシュして可視化監視
アラートルールを設定: データ品質合格率がしきい値を下回る、データ量の異常な変動など
DVC または LakeFS を使用してデータバージョン管理を行い、再現性を確保
定期的にデータ品質を手動でサンプリングし、自動スコアリングと相互検証

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

DeepSeek データエンジニアリングよくある質問

データの質と量、どちらが重要ですか？ +

計算リソースが限られている場合、データの質は量よりもはるかに重要です。高品質な10Kデータでトレーニングされたモデルは、低品質な100Kデータでトレーニングされたモデルよりも優れていることがよくあります。DeepSeekチームの経験によると、厳選されアノテーションされたデータセットは、指示追従と推論能力に質的な飛躍をもたらします。まず品質を確保し、その後量を拡大することをお勧めします。

SFTデータは最低何件必要ですか？ +

高品質なSFTデータの場合、10K〜50K件でモデルは十分な指示追従能力を示します。包括的な能力を追求する場合は、100K〜500K件を推奨します。重要なのは、データが多様なタスクタイプと難易度をカバーしていることであり、単に量を追求することではありません。Alpacaの52Kデータは良い出発点です。

データがDeepSeekによってトレーニングされたかどうかを判断するには？ +

モデルのトレーニングデータに含まれる可能性が低い特異な質問を構築してテストできます。例えば、特定の日付以降のイベントや、特定の形式の架空データを使用します。モデルが正確に回答できれば、データ漏洩の可能性があります。ただし、この方法は絶対的に信頼できるわけではありません。最も正確な方法は、公式のテクニカルレポートのトレーニングデータの説明を参照することです。

合成データ生成のリスクは何ですか？ +

合成データの主なリスクは次のとおりです：1) モデルの幻覚が増幅される—生成モデル自体にエラーがある場合、合成データはこれらのエラーを増幅します；2) 多様性の不足—合成データは実際のデータの多様性に欠けることがよくあります；3) モード崩壊—合成を繰り返すとデータ分布がますます狭くなる可能性があります。合成データと実データを混合し、厳格な品質フィルタリングを行うことをお勧めします。

DeepSeek-R1のthinkingデータはどう準備しますか？ +

DeepSeek-R1のthinkingデータには、<think> と </think> タグで囲まれた明示的な推論プロセスを含める必要があります。強力なモデル（DeepSeek-V3など）を使用して数学や推論問題のCoT推論チェーンを生成し、その後人間が品質をレビューできます。重要なのは、推論プロセスが正確で、完全で、検証可能であることを保証することです。R1自身が生成した推論データをR1の再トレーニングに使用することは、モード崩壊を引き起こす可能性があるため推奨されません。

データクリーニングで貴重な情報が失われることはありますか？ +

過度に積極的なデータクリーニングは確かに貴重な情報を失う可能性があります。段階的なクリーニング戦略を採用することをお勧めします：まず緩いルールで明らかに低品質なデータをフィルタリングし、トレーニング中のモデルのパフォーマンスに基づいてフィルタリングルールを動的に調整します。境界のケースでは、直接破棄するのではなく、サンプルを保持し低信頼度としてマークできます。破棄されたデータを定期的に手動でスポットチェックし、クリーニングルールの妥当性を評価します。

さらに学ぶ

DeepSeek関連チュートリアル

DeepSeekモデルの使用方法、デプロイ、エコシステムツールを深く学びます。

入門チュートリアル

DeepSeekモデルの使い方

4つの使用方法、ゼロからの入門チュートリアル。

モデル微調整

DeepSeek微調整チュートリアル

LoRA/QLoRA微調整、SFT、RLHFの全プロセス。

RAGチュートリアル

DeepSeek RAG知識ベース

ベクトルデータベース、検索拡張生成、ドキュメントQ&A。

デプロイチュートリアル

DeepSeekデプロイチュートリアル

Ollama、Docker、vLLM、K8sデプロイソリューション。

モデル詳細

DeepSeekモデルアーキテクチャ

技術アーキテクチャ、ベンチマーク、モデル選定比較。

エコシステムツール

DeepSeekエコシステムツール

WebUI、IDEプラグイン、Agentフレームワーク、RAGプラットフォーム。