DeepSeek Data Engineering Tutorial

From data collection to quality evaluation, fully master the LLM data engineering pipeline. Covers core aspects such as SFT dataset construction, RLHF preference data preparation, data cleaning and deduplication, data augmentation, and large-scale data processing, with Python practical code.

Start Learning

DeepSeek Data Engineering

Data is the Soul of LLM

In large model training, data quality directly determines the upper limit of model capability. High-quality datasets can enable small models to surpass large models, while low-quality data can make trillion-parameter models perform mediocrely. Data engineering covers the entire chain from raw data collection to final training data delivery.

Chapter 1

Data Engineering Overview

Understand the central role of data in LLM training, grasp the trade-off between data quality and quantity, and build a comprehensive understanding of the data engineering pipeline.

The Central Role of Data in LLM Training

Among the three elements of LLM training (algorithm, compute, data), data is often underestimated, but its importance far exceeds the other two. A widely held view is that the marginal returns of model architecture diminish, while the marginal returns of data quality increase. Here are three interrelated core facts:

Data quality determines the upper bound of model capability: Even with the most advanced model architecture (e.g., MoE), if the training data quality is poor, the model cannot produce high-quality responses. The strong capabilities of DeepSeek-V3 and R1 are largely attributed to carefully constructed training data.
Data diversity determines generalization ability: Data from a single domain can cause the model to overfit, while diverse data enables the model to perform well on unseen tasks. DeepSeek's data mixing strategy covers multiple domains such as mathematics, code, reasoning, dialogue, and creative writing.
Data scale determines the knowledge boundary: The model's knowledge scope does not exceed the coverage of the training data. To make the model understand medicine, high-quality medical data is necessary; to make the model capable of programming, sufficient code data is required.

Data Quality vs. Data Quantity

Under limited computational resources, data quality is more important than data quantity. Here is a trade-off analysis of the two:

Dimension	Pursuing Quantity	Pursuing Quality
Training Efficiency	Longer training time, slower convergence	Faster training, more stable convergence
Model Performance	More noise, unstable outputs	Accurate outputs, fewer hallucinations
Cost	High GPU cost, long cycle	Higher data cleaning cost, but lower total cost
Typical Strategy	Crawl web data, coarse filtering	Curated data sources, multiple filtering, human annotation

The DeepSeek team found in practice that models trained on 1/10 of high-quality data often outperform models trained on the full set of coarse data in instruction following and reasoning capabilities. This also explains why data engineering is the most critical part of LLM development.

Data Engineering Pipeline Overview

A complete data engineering pipeline includes the following stages:

Data Collection: Obtain raw data from public datasets, web crawling, API calls, synthetic generation, and other channels
Data Cleaning: Remove duplicates, filter low quality, handle missing values, filter sensitive information
Data Annotation: Construct data in different formats for different training stages such as SFT, RLHF, DPO
Data Augmentation: Expand data using techniques like Self-Instruct, Evol-Instruct, back-translation
Quality Evaluation: Evaluate data quality from dimensions such as diversity, difficulty, instruction complexity
Data Mixing: Mix data from different domains in proportions to form the final training dataset
Version Management: Use tools like DVC to manage data versions, ensuring reproducibility

Chapter 2

Data Collection and Sources

Learn about the main sources of LLM training data, including public datasets, web crawling, synthetic data generation, and data compliance considerations.

Common Public Datasets

Dataset Name	Size	Type	Use Case
Alpaca	52K	SFT instruction data	Instruction fine-tuning basics
ShareGPT	90K	Multi-turn dialogue	Dialogue capability training
UltraChat	1.5M	Multi-turn dialogue	Large-scale dialogue training
OpenOrca	4M	SFT instruction data	Large-scale instruction fine-tuning
CodeAlpaca	20K	Code generation	Code capability training
MathInstruct	260K	Mathematical reasoning	Mathematical capability training

Loading Public Datasets from Hugging Face

from datasets import load_dataset

# Load the Alpaca dataset
alpaca = load_dataset("tatsu-lab/alpaca")
print(f"Alpaca training set: {len(alpaca['train'])} samples")
print(f"Example: {alpaca['train'][0]}")

# Load ShareGPT conversation data
sharegpt = load_dataset("anon8231489123/ShareGPT_Vicuna_unfiltered",
                         data_files="ShareGPT_V3_unfiltered_cleaned_split.json")
print(f"ShareGPT: {len(sharegpt['train'])} conversations")

# Load OpenOrca data (load subset as needed)
orca = load_dataset("Open-Orca/OpenOrca", split="train[:50000]")
print(f"OpenOrca subset: {len(orca)} items")

Web Scraping and Data Collection

For domain-specific data, web scraping is an important supplementary method. Below is an example crawler for documentation and tutorials:

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):
    """Crawl documentation websites and extract main content"""
    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")

            # Remove script and style tags
            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,
                })

            # Discover new links
            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"Crawl failed {url}: {e}")

    return results

# Usage example
docs = crawl_documentation("https://docs.python.org/3/", max_pages=50)
print(f"Crawled {len(docs)} pages")

Synthetic Data Generation

When public data is insufficient to cover a specific domain, synthetic data generation techniques can be used. High-quality training data can be generated using strong models like 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):
    """Generate synthetic data for a specific domain using DeepSeek"""
    prompt = f"""Please generate {num_samples} high-quality instruction data about 「{topic}」.

Each data item contains:
- instruction: clear, specific instruction or question
- input: supplementary context (can be empty string)
- output: professional, accurate, detailed answer

Output format is a JSON array."""

    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

# Generate Python machine learning related data
synthetic = generate_synthetic_data("Python machine learning", num_samples=10)
print(synthetic)

Data Compliance Notes

When using public datasets and web-scraped data, be sure to check the data license agreement. Some datasets (such as ShareGPT) have specific usage restrictions. Web scraping should comply with the robots.txt protocol, control the scraping frequency, and avoid burdening the target website. Data involving personal privacy information needs to be desensitized.

Chapter 3

Data Cleaning and Deduplication

Data cleaning is the most time-consuming but crucial step in data engineering. High-quality data cleaning can significantly improve model training results, including quality filtering, deduplication, and sensitive information filtering.

Quality Filtering Pipeline

A complete quality filtering pipeline includes filtering from multiple dimensions:

import re
from typing import List, Dict

class DataQualityFilter:
    """Multi-dimensional data quality filter"""

    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:
        """Length filtering: too short or too long text may be noise"""
        return self.min_length <= len(text) <= self.max_length

    def filter_by_word_count(self, text: str) -> bool:
        """Word count filtering: filter overly fragmented text"""
        words = text.split()
        return len(words) >= self.min_words

    def filter_repetitive(self, text: str, threshold=0.3) -> bool:
        """Repetitive content filtering: detect duplicate n-gram ratio"""
        if len(text) < 100:
            return True
        words = text.split()
        unigrams = len(set(words))
        if unigrams / len(words) < threshold:
            return False  # repetition rate too high
        return True

    def filter_special_chars(self, text: str, max_ratio=0.3) -> bool:
        """Special character filtering: filter garbled or abnormal text"""

        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:
        """Empty output filter: output too short or only contains placeholders"""
        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]:
        """Apply all filters"""
        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"Before filtering: {len(samples)} items, after filtering: {len(filtered)} items")
        return filtered

MinHash LSH Deduplication

MinHash + LSH (Locality-Sensitive Hashing) is an industry-standard large-scale deduplication solution that efficiently detects near-duplicate documents:

from datasketch import MinHash, MinHashLSH
import re

def tokenize_chinese(text):
    """Chinese tokenization: split by character 2-gram"""



    # Simplified tokenization, for production use jieba or pkuseg
    text = re.sub(r'[^\u4e00-\u9fff\w]', ' ', text.lower())
    # 2-gram tokenization
    return [text[i:i+2] for i in range(len(text)-1)]

def deduplicate_with_minhash(samples, num_perm=128, threshold=0.8):
    """Approximate deduplication using 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"))

        # Check if approximate duplicate already exists
        if len(lsh.query(m)) == 0:
            lsh.insert(str(idx), m)
            unique_samples.append(sample)

    print(f"Before deduplication: {len(samples)} items, after deduplication: {len(unique_samples)} items")
    return unique_samples

Semantic Deduplication

MinHash is suitable for literal-level deduplication, but for data that is semantically similar but expressed differently, embedding vectors are needed for semantic deduplication:

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):
    """Deduplication based on semantic embeddings"""
    model = SentenceTransformer(model_name)

    # Extract all texts



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

    # Batch encode
    embeddings = model.encode(texts, batch_size=batch_size, show_progress_bar=True)

    # Compute similarity matrix, mark duplicates
    unique_indices = []
    seen = set()

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

        # Find samples highly similar to current one
        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"Semantic deduplication: {len(samples)} -> {len(unique_indices)} items")
    return [samples[i] for i in unique_indices]

Sensitive Information Filtering

In training data, personal privacy information and sensitive content must be filtered out:

import re

class SensitiveContentFilter:
    """Sensitive information filter"""

    # Common sensitive information regex patterns
    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:
        """Detect if sensitive information is present"""
        for name, pattern in self.PATTERNS.items():



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

    def mask_sensitive(self, text: str) -> str:
        """Masking: replace sensitive information with placeholders"""
        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

# Usage example
filter = SensitiveContentFilter()
text = "Please contact admin@example.com or call 13800138000"
print(f"Contains sensitive information: {filter.has_sensitive(text)}")  # True
print(f"Masked: {filter.mask_sensitive(text)}")

Chapter 4

Data Formats and Annotation

Different training stages require different data formats. SFT uses the Instruction-Input-Output format, ChatML is used for dialogue scenarios, and RLHF/DPO requires preference comparison data. Understanding these formats is the foundation for building high-quality datasets.

SFT Data Format (Instruction-Input-Output)

SFT (Supervised Fine-Tuning) data is the most basic training data format. Each data point contains an instruction, optional input, and expected output:

{
  "instruction": "Implement quicksort algorithm in 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 Conversation Format

ChatML (Chat Markup Language) is a conversation format defined by OpenAI, also widely used in SFT training. It structures multi-turn conversations into a standard format:

def format_chatml(conversations):
    """Convert a list of conversations to ChatML format string"""
    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()

# Example conversation
conversation = [
    {"role": "system", "content": "You are a professional Python programming assistant."},
    {"role": "user", "content": "How to read a CSV file?"},
    {"role": "assistant", "content": "You can easily read a CSV file using pandas:\nimport pandas as pd\ndf = pd.read_csv('data.csv')"},

]
]

chatml = format_chatml(conversation)
print(chatml)

RLHF Preference Data Format

RLHF (Reinforcement Learning from Human Feedback) and DPO (Direct Preference Optimization) require preference comparison data, i.e., for the same prompt, annotate which response is better:

# DPO data format
dpo_sample = {
    "prompt": "Explain what machine learning is",
    "chosen": "Machine learning is a branch of artificial intelligence that enables computers to learn patterns from data without explicit programming. Common methods include supervised learning, unsupervised learning, and reinforcement learning.",
    "rejected": "Machine learning is just making machines learn things.",
}

# RLHF comparison data format
rlhf_comparison = {
    "prompt": "Write a poem about spring",
    "responses": [
        {"text": "Spring breeze brushes the face, willows like smoke...", "score": 4.5},
        {"text": "Spring has come, flowers bloom.", "score": 1.0},
    ]
}

Data Format Selection Recommendations

For basic SFT training, the Instruction-Input-Output format is sufficient; for multi-turn dialogue scenarios, the ChatML format is recommended; if RLHF or DPO training is needed, preference comparison data must be prepared. DeepSeek series models support both Alpaca format and ChatML format.

Chapter 5

Data Augmentation Techniques

When existing data is insufficient, data augmentation techniques can help you generate large amounts of high-quality training data from a small set of seed data. Self-Instruct and Evol-Instruct are the two most popular methods.

Self-Instruct Self-Generation Method

The core idea of Self-Instruct is to use a strong model to automatically generate new instruction-output pairs from seed tasks:

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):
    """Generate new instructions based on seed tasks"""
    # Randomly sample seed tasks as context
    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"""You are a data annotation expert. Based on the following seed tasks, generate {num_to_generate} new, diverse instructions.

Seed task examples:
{seed_text}

Requirements:
1. New instructions should cover different domains and difficulty levels
2. Instructions should be clear, specific, and executable
3. Avoid duplication with seed tasks
4. Output format is a JSON array, each item contains an instruction field

Please output only the JSON array, do not include any other text."""

    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):
    """Generate high-quality output for a given 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

# Usage example
seed_tasks = [
{"instruction": "Implement bubble sort in Python"},
    {"instruction": "Explain what a RESTful API is"},
    {"instruction": "Write a function to compute the Fibonacci sequence"},
]
new_instructions = generate_instructions(seed_tasks, num_to_generate=10)
print(f"Generated {len(new_instructions)} new instructions")

Evol-Instruct Evolutionary Generation

Evol-Instruct generates more challenging data by gradually increasing the complexity of instructions. The DeepSeek team used similar evolutionary strategies extensively in training:

def evolve_instruction(instruction, evolution_type="deepen"):
    """Evolve instruction: increase depth, breadth, or complexity"""
    evolution_prompts = {
        "deepen": """Rewrite the following instruction to be more in-depth and professional. Add discussion of underlying principles, implementation details, or edge cases.

Original instruction: {instruction}

Deepened instruction: """,

        "broaden": """Rewrite the following instruction to be broader, requiring coverage of multiple related subtopics or different scenarios.

Original instruction: {instruction}

Expanded instruction: """,

        "increase_reasoning": """Rewrite the following instruction into a complex task that requires multi-step reasoning to complete.

Original instruction: {instruction}

Instruction with increased reasoning requirements: """,
    }

    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

# Evolution example
original = "Explain what a database index is"
deepened = evolve_instruction(original, "deepen")
print(f"Original instruction: {original}")
print(f"Deepened: {deepened}")

Back Translation Data Augmentation

Back Translation is a classic method for multilingual data augmentation, generating semantically equivalent but differently expressed data through translation and back-translation:

def back_translate(text, source_lang="Chinese", pivot_lang="English"):
    """Back translation: Chinese -> English -> Chinese, generate semantically equivalent variants"""

    # Step 1: Translate to target language
    translate_prompt = f"Please translate the following{source_lang}text into{pivot_lang}, output only the translation result:\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

    # Step 2: Translate back to source language
    back_prompt = f"Please translate the following{pivot_lang}text into{source_lang}, output only the translation result:\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

# Usage example
original = "Deep learning is a subfield of machine learning that uses multi-layer neural networks to learn representations of data."
augmented = back_translate(original)
print(f"Original: {original}")
print(f"Augmented: {augmented}")

Chapter 6

DeepSeek Specific Data Preparation

DeepSeek-R1's reasoning data has special format requirements, including thinking tags and CoT (Chain of Thought) reasoning chains. This chapter details how to prepare data for DeepSeek models.

DeepSeek-R1 Reasoning Data Format

The key innovation of DeepSeek-R1 is that the training data includes explicit thinking/reasoning processes. The following is the standard R1 reasoning data format:

# DeepSeek-R1 Reasoning Data Format
r1_format_sample = {
    "messages": [
        {
            "role": "user",
            "content": "A rectangular prism has length, width, and height of 3cm, 4cm, and 5cm respectively. Find its surface area and volume."
        },
        {
            "role": "assistant",
            "content": "<think>This is a problem of calculating the surface area and volume of a rectangular prism.\nSurface area = 2*(length*width + length*height + width*height)\nVolume = length*width*height\n\nSubstitute the values:\nlength=3, width=4, height=5\nSurface area = 2*(3*4 + 3*5 + 4*5) = 2*(12+15+20) = 2*47 = 94\nVolume = 3*4*5 = 60\n\nVerification: Surface area and volume have different units, but the numerical calculations are correct.</think>\n\nThe surface area of the rectangular prism is 94 square centimeters, and the volume is 60 cubic centimeters."
        }
    ]
}

CoT Data Construction

The key to constructing Chain of Thought data is to let the model learn to "think first, then answer." The following code automatically generates CoT reasoning chains for math problems:

def generate_cot_data(question, answer):
    """Generate CoT reasoning process for a question and answer"""
    prompt = f"""Please generate a detailed step-by-step reasoning process for the following math problem.

Problem: {question}
Correct answer: {answer}

Please output in the following format:
<think>
(detailed step-by-step reasoning, including formulas used, intermediate steps, verification process)
</think>

(final answer, concise and clear)

Please ensure the reasoning process is clear, complete, and verifiable."""

    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

# Generate CoT for a math problem
cot_result = generate_cot_data(
    question="If x^2 - 5x + 6 = 0, find the value of x.",
    answer="x = 2 or x = 3"
)
print(cot_result)

Code Data Preparation

Code data is an important part of DeepSeek training data. High-quality code data should include problem descriptions, code implementations, and comments:

def format_code_data(problem, code, language="python", difficulty="medium"):
    """Format code training data"""
    return {
        "messages": [



            {
                "role": "system",
                "content": f"You are a professional{language.upper()} programming assistant. Please provide clear and efficient code implementations."
            },
            {
                "role": "user",
                "content": problem
            },
            {
                "role": "assistant",
                "content": code
            }
        ],
        "metadata": {
            "language": language,
            "difficulty": difficulty,
            "type": "code_generation",
        }
    }

# Example
code_sample = format_code_data(
    problem="Implement an LRU cache (Least Recently Used) that supports get and put operations with O(1) time complexity.",
    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"
)

Math Data Preparation

Math reasoning data needs to include formulas, derivation steps, and final answers. LaTeX format is the standard for mathematical expressions:

math_sample = {
    "messages": [
        {
            "role": "user",
            "content": "Find the limit \\lim_{x \\to 0} \\frac{\\sin x}{x}"
        },
        {
            "role": "assistant",
            "content": "<think>This is a classic limit problem.\n\nMethod 1: Use L'Hôpital's rule\nWhen x→0, the numerator sin x → 0, the denominator x → 0, satisfying the 0/0 indeterminate form.\nDifferentiate the numerator and denominator separately:\nlim(sin x / x) = lim(cos x / 1) = cos(0) = 1\n\nMethod 2: Use the Squeeze theorem\nFor 0 < |x| < pi/2, we have cos x < sin x / x < 1\nWhen x→0, cos x → 1, so the limit is 1.\n\nVerification: This result is correct.</think>\n\n\\lim_{x \\to 0} \\frac{\\sin x}{x} = 1"
        }
    ]
}

Chapter 7

Data Quality Evaluation

Data quality evaluation is the last line of defense in data engineering. Evaluate dataset quality from multiple dimensions such as diversity, difficulty, instruction complexity, and automatic scoring to ensure training data meets expected standards.

Diversity Evaluation

Diversity evaluation ensures the dataset covers a sufficiently wide range of domains and task types:

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

def evaluate_diversity(samples, n_clusters=10):
    """Evaluate the diversity of the dataset"""
    # Extract instruction texts
    instructions = [s.get("instruction", "") for s in samples]

    # TF-IDF vectorization
    vectorizer = TfidfVectorizer(max_features=5000, ngram_range=(1, 2))
    vectors = vectorizer.fit_transform(instructions)

    # Cluster analysis
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    labels = kmeans.fit_predict(vectors.toarray())

    # Count samples per cluster
    cluster_counts = np.bincount(labels)
    cluster_ratio = cluster_counts / len(samples)

    # Calculate diversity score (entropy based on cluster distribution)
    entropy = -np.sum(cluster_ratio * np.log(cluster_ratio + 1e-10))
    max_entropy = np.log(n_clusters)
    diversity_score = entropy / max_entropy

    print(f"Number of clusters: {n_clusters}")
    print(f"Samples per cluster: {cluster_counts}")
    print(f"Diversity score: {diversity_score:.4f} (1.0 = perfect uniform distribution)")

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

Difficulty Evaluation

Evaluate the difficulty of each instruction in the dataset to ensure a reasonable difficulty distribution:

def evaluate_difficulty(instruction):



    """Use LLM to evaluate the difficulty of instructions"""
    prompt = f"""Please evaluate the difficulty of the following instruction, return a score from 1-5 (1=very easy, 5=very difficult).

Evaluation criteria:
- 1 point: Simple factual question, no reasoning required
- 2 points: Requires basic understanding or simple operation
- 3 points: Requires multi-step thinking or comprehensive knowledge
- 4 points: Requires advanced reasoning or professional knowledge
- 5 points: Requires deep reasoning, creativity, or expert-level knowledge

Instruction: {instruction}

Please output only the score (an integer from 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):
    """Analyze the difficulty distribution of the dataset"""
    from collections import Counter

    difficulties = []
    for s in samples[:100]:  # sample evaluation
        score = evaluate_difficulty(s.get("instruction", ""))
        difficulties.append(score)

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

    print(f"\nAverage difficulty: {np.mean(difficulties):.2f}")
    return difficulties

Instruction Complexity Scoring

Instruction complexity scoring evaluates the quality of instructions from multiple dimensions:

Scoring Dimension	Description	Scoring Method
Clarity	Whether the instruction is clear and unambiguous	LLM score 1-5
Completeness	Whether the instruction contains sufficient context	Information density calculation
Executability	Whether the instruction can be completed	LLM executability judgment
Creativity	Whether the instruction encourages creative thinking	LLM score 1-5

Automatic Quality Scoring

Use LLM as a judge to automatically score data quality:

def auto_quality_score(sample):
    """Use LLM to automatically evaluate the quality of a single data item"""
    instruction = sample.get("instruction", "")
    output = sample.get("output", "")

    prompt = f"""Please evaluate the quality of the following Q&A pair, scoring from 1-10.

Evaluation dimensions:
1. Is the instruction clear and unambiguous (1-3 points)
2. Is the answer accurate and complete (1-3 points)
3. Is the answer professional and helpful (1-2 points)
4. Is the answer length appropriate (1-2 points)

Instruction: {instruction}
Answer: {output}

Please output in JSON format:
{{"total_score": X, "reasons": ["reason1", "reason2"]}}"""

    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": ["Unable to parse score"]}

# Usage example
sample = {
    "instruction": "Implement binary search algorithm in Python, handling boundary conditions",
    "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"Quality score: {score}")

Chapter 8

Data Mixing Strategy

Data from different domains needs to be mixed in specific proportions to train a comprehensive and balanced model. The data mixing strategy directly affects the model's general capabilities and specialized abilities.

Typical Data Mixing Ratios

The following are reference data mixing ratios commonly used in DeepSeek-like model training:

Data Category	Suggested Ratio	Description
General Conversation	30-40%	Daily conversation, Q&A, chit-chat, ensuring basic conversational ability
Code Generation	15-20%	Code in mainstream languages like Python/JS/Java/C++
Mathematical Reasoning	10-15%	Algebra, geometry, calculus, probability and statistics
Logical Reasoning	10-15%	Logic puzzles, reasoning chains, multi-step reasoning
Creative Writing	5-10%	Poetry, stories, copywriting, scripts
Professional Domains	5-10%	Vertical domains such as medicine, law, finance
Safety Alignment	3-5%	Rejecting harmful requests, value alignment

Code Implementation for Data Mixing

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):
    """Mix multiple datasets according to ratios"""
    # Validate sum of ratios
    total_ratio = sum(ratios.values())
if abs(total_ratio - 1.0) > 0.01:
        print(f"Warning: The sum of ratios is {total_ratio:.2f}, it is recommended to adjust it to 1.0")

    mixed = []
    stats = {}

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

        # Sample according to ratio
        sample_size = int(total_size * ratio)
        available = datasets[name]

        # If data is insufficient, use all
        if len(available) < sample_size:
            sampled = available.copy()
            print(f"  {name}: Insufficient data, using all {len(available)} items")
        else:
            sampled = random.sample(available, sample_size)

        # Add source marker
        for s in sampled:
            s["source_dataset"] = name

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

    if shuffle:
        random.shuffle(mixed)

    print(f"\nDataset mixing completed, total {len(mixed)} items")
    for name, count in stats.items():
        print(f"  {name}: {count} items ({count/len(mixed)*100:.1f}%)")

    return mixed

# Usage example
datasets = {
    "general_chat": [...],     # General conversation data
    "code": [...],              # Code data
    "math": [...],              # Math data
    "reasoning": [...],         # Reasoning data
    "creative": [...],         # Creative writing data
}

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)

Data Annealing Strategy

Data annealing is a training strategy that dynamically adjusts data mixing ratios during training. Early on, diverse data is used to train basic capabilities, and later, the proportion of high-quality, high-difficulty data is gradually increased:

Warm-up phase (first 20% of training steps): General chat data accounts for 50%, mainly simple instructions, helping the model build basic conversational abilities.
Main training phase (20%-80% of training steps): Gradually increase the proportion of code and reasoning data, introducing medium-difficulty tasks.
Annealing phase (last 20% of training steps): Substantially increase high-quality, high-difficulty data, reduce general chat data, and improve the model's performance on complex tasks.

The DeepSeek team used a similar annealing strategy in their training, which is one of the important reasons why DeepSeek-R1 excels in reasoning capabilities.

Chapter 9

Large-scale Data Processing

When data volume reaches millions or even hundreds of millions, single-machine processing is no longer feasible. This chapter introduces distributed frameworks like Spark and Ray for large-scale data processing, as well as data version management.

Parallel Processing with Ray

Ray is a lightweight distributed computing framework, particularly suitable for data processing tasks in the Python ecosystem:

import ray
from typing import List, Dict

# Initialize Ray
ray.init(num_cpus=8)

# Declare the processing function as a Ray remote task
@ray.remote
def process_batch(batch: List[Dict]) -> List[Dict]:
    """Process a batch of data: cleaning, deduplication, quality filtering"""
    filter = DataQualityFilter()
    # Assume this includes all filtering logic defined in Chapter 3
    return filter.apply_all(batch)

def parallel_process(samples: List[Dict], batch_size=1000):
    """Process large-scale data in parallel using Ray"""
    # Split into batches
    batches = [samples[i:i+batch_size] for i in range(0, len(samples), batch_size)]

    # Submit tasks in parallel
    futures = [process_batch.remote(batch) for batch in batches]

    # Collect results
    results = []
    for future in futures:
        results.extend(ray.get(future))

    print(f"Parallel processing completed: {len(samples)} -> {len(results)} records")
    return results

# Usage example
# results = parallel_process(all_samples, batch_size=2000)
# ray.shutdown()

Distributed Processing with PySpark

PySpark is the standard tool for processing TB-level large-scale data, supporting DataFrame API and SQL operations:

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

# Create Spark session
spark = SparkSession.builder \



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

# Read large-scale JSON data
df = spark.read.json("hdfs://data/deepseek/*.jsonl")

# Basic filtering
df_clean = df.filter(
    (length(col("instruction")) > 20) &
    (length(col("output")) > 20) &
    (length(col("instruction")) < 8000)
)

# Use SQL for complex filtering
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
""")

# Statistics
print(f"Before processing: {df.count()} records")
print(f"After processing: {df_quality.count()} records")

# Write processed data
df_quality.write.mode("overwrite").json("hdfs://data/deepseek_cleaned/")

Stream Processing Large Files

For extremely large JSONL files, use stream processing to avoid memory overflow:

import json
import ijson
import gzip

def stream_process_jsonl(filepath, output_path, batch_size=10000):
    """Stream process large JSONL files"""
    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:
                # Batch processing
                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"Processed {total_processed} records, wrote {total_written} records")

        # Process remaining batch
        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"Processing complete: {total_processed} -> {total_written} records")

Data Version Control (DVC)

DVC (Data Version Control) is the Git for data science, used to manage dataset versions:

# Initialize DVC
dvc init

# Track dataset
dvc add data/training_data.jsonl

# Commit to Git
git add data/training_data.jsonl.dvc data/.gitignore
git commit -m "Add initial training dataset v1.0"

# After data update, create new version
dvc add data/training_data.jsonl
git add data/training_data.jsonl.dvc
git commit -m "Update training data: added 10K code data"

# Switch to historical version
git checkout "<commit-hash>"
dvc checkout

# View data change history
dvc diff HEAD~1

Chapter 10

Production-Grade Data Pipeline

Integrate all previous steps into a complete production-grade data pipeline, including continuous data collection, automated quality monitoring, data drift detection, and complete pipeline code.

Complete Data Pipeline Architecture

Data Collection Layer: Scheduled tasks to fetch public dataset updates, crawl new content, call synthetic data APIs
Data Cleaning Layer: Quality filtering, deduplication, sensitive information filtering, format standardization
Data Augmentation Layer: Self-Instruct generation, Evol-Instruct evolution, back-translation
Quality Evaluation Layer: Diversity scoring, difficulty assessment, automatic quality scoring
Data Mixing Layer: Proportional mixing, data annealing strategy, version management
Monitoring and Alerting Layer: Data drift detection, quality trend monitoring, anomaly alerting

Complete Pipeline Code

"""DeepSeek Data Engineering Pipeline — Complete Implementation"""
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 Data Engineering Pipeline"""

    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]:
        """Step 1: Data Collection"""
        logger.info("Step 1: Data Collection")
        all_data = []

        # Load public dataset
        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"  Loaded Alpaca: {len(alpaca)} entries")
        except Exception as e:
            logger.warning(f"  Failed to load Alpaca: {e}")

        # Load local JSONL files
        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"  Loaded {f.name}")

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

    def step2_clean(self, data: List[Dict]) -> List[Dict]:
        """Step 2: Data Cleaning and Deduplication"""
        logger.info("Step 2: Data Cleaning and Deduplication")

        # Length filtering
        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"  Length filtering: {len(data)} -> {len(cleaned)} entries")

        # Deduplication (exact dedup based on 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"  Deduplicated: {len(cleaned)} -> {len(deduped)} items")

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

    def step3_quality_check(self, data: List[Dict]) -> List[Dict]:
        """Step 3: Quality assessment and filtering"""
        logger.info("Step 3: Quality assessment")

        # Filter out data with too short output or containing rejection patterns
        reject_patterns = [
            "Sorry, I cannot", "As an AI", "I'm sorry",
            "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"  Quality filter: {len(data)} -> {len(quality)} items")

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

    def step4_export(self, data: List[Dict], filename: str = None):
        """Step 4: Export processed data"""
        logger.info("Step 4: Export data")

        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"  Exported to: {output_path} ({len(data)} items)")
        return output_path

    def run(self):
        """Run the complete pipeline"""
        logger.info("=" * 50)
        logger.info("DeepSeek Data Engineering Pipeline Started")
        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()

        # Output final report
        logger.info("=" * 50)
        logger.info("Pipeline run completed")
        logger.info(f"  Total time: {elapsed:.1f} seconds")
        for k, v in self.metrics.items():
            logger.info(f"  {k}: {v}")
        logger.info(f"  Output file: {output_path}")
        logger.info("=" * 50)

        return data


# ========== Main program ==========
if __name__ == "__main__":
    config = {
        "data_dir": "./data",
        "output_dir": "./output",
    }



    pipeline = DataPipeline(config)
    pipeline.run()

Data Drift Detection

Data drift refers to changes in the data distribution over time. In production environments, it is necessary to continuously monitor whether data quality deviates from expectations:

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):
    """Detect whether current data has drifted relative to reference data"""

    # 1. Compare instruction length distributions
    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 test: stat={stat:.4f}, p={p_value:.4f}")
    print(f"Average length: reference={np.mean(ref_lengths):.0f}, current={np.mean(cur_lengths):.0f}")

    if drift_detected:
        print("Warning: Data drift detected!")
    else:
        print("Data distribution is normal, no significant drift detected.")

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

Automated Quality Monitoring

Use scheduled tasks to continuously monitor data quality metrics:

import schedule
import time

def monitoring_job():
    """Scheduled monitoring task"""
    logger.info("[Monitor] Starting data quality check...")

    # Run pipeline
    pipeline = DataPipeline(config)



    try:
        data = pipeline.run()

        # Check quality metrics
        quality_ratio = pipeline.metrics.get("quality_passed", 0) / \
                        max(pipeline.metrics.get("collected", 1), 1)

        if quality_ratio < 0.5:
            logger.warning(f"[Alert] Data quality pass rate is too low: {quality_ratio:.2%}")
        else:
            logger.info(f"[Normal] Data quality pass rate: {quality_ratio:.2%}")

    except Exception as e:
        logger.error(f"[Error] Monitoring task failed: {e}")

# Configure scheduled task: run at 2 AM daily
# schedule.every().day.at("02:00").do(monitoring_job)
#
# while True:
#     schedule.run_pending()
#     time.sleep(60)

Running the Pipeline

# 1. Prepare data directory
mkdir data
# Put your raw data files into the data/ directory

# 2. Install dependencies
pip install datasets schedule

# 3. Run the pipeline
python data_pipeline.py

# 4. View output
# Processed data is in the output/ directory
ls output/

Production Environment Recommendations

Use Apache Airflow or Prefect to orchestrate pipeline tasks
Push quality metrics to Prometheus + Grafana for visual monitoring
Configure alert rules: data quality pass rate below threshold, abnormal data volume fluctuations, etc.
Use DVC or LakeFS for data version management to ensure reproducibility
Regularly manually sample data quality and cross-validate with automatic scoring

Data Engineering FAQ

DeepSeek Data Engineering FAQ

Is data quality or data quantity more important? +

When computational resources are limited, data quality is far more important than data quantity. A model trained on high-quality 10K data often outperforms one trained on low-quality 100K data. The DeepSeek team's experience shows that carefully curated and annotated datasets can bring a qualitative leap in instruction following and reasoning capabilities. It is recommended to ensure quality first, then consider expanding quantity.

What is the minimum number of SFT data samples? +

For high-quality SFT data, 10K-50K samples are sufficient for the model to show decent instruction following capabilities. If pursuing comprehensive abilities, 100K-500K samples are recommended. The key is that the data should cover diverse task types and difficulty levels, rather than simply pursuing quantity. Alpaca's 52K data is a good starting point.

How to determine if data has been trained on by DeepSeek? +

You can test by constructing specific questions that are unlikely to appear in the model's training data. For example, use events after a specific date, fictional data in a specific format, etc. If the model can answer accurately, it may indicate data leakage. However, this method is not absolutely reliable; the most accurate way is to refer to the training data description in the official technical report.

What are the risks of synthetic data generation? +

The main risks of synthetic data include: 1) Model hallucination is amplified—if the generating model itself has errors, synthetic data will amplify these errors; 2) Insufficient diversity—synthetic data often lacks the diversity of real data; 3) Mode collapse—multiple iterations of synthesis may cause the data distribution to become increasingly narrow. It is recommended to mix synthetic data with real data and perform strict quality filtering.

How to prepare thinking data for DeepSeek-R1? +

DeepSeek-R1's thinking data needs to include explicit reasoning processes, wrapped with <think> and </think> tags. You can use a strong model (such as DeepSeek-V3) to generate CoT reasoning chains for math and reasoning problems, then manually review the quality. The key is to ensure the reasoning process is correct, complete, and verifiable. It is not recommended to use R1's own generated reasoning data to retrain R1, as it may lead to mode collapse.

Can data cleaning lose valuable information? +

Overly aggressive data cleaning can indeed lose valuable information. It is recommended to adopt a progressive cleaning strategy: first use lenient rules to filter obviously low-quality data, then dynamically adjust filtering rules based on model performance during training. For borderline cases, you can keep samples and mark them as low confidence rather than discarding them directly. Regularly conduct manual spot checks on discarded data to evaluate the reasonableness of cleaning rules.

Learn More

DeepSeek Related Tutorials

Dive deeper into DeepSeek model usage, deployment, and ecosystem tools.

Getting Started

How to Use DeepSeek Models

Four usage methods, zero-basics tutorial.

Fine-tuning

DeepSeek Fine-tuning Tutorial

LoRA/QLoRA fine-tuning, SFT, RLHF full pipeline.

RAG Tutorial

DeepSeek RAG Knowledge Base

Vector databases, retrieval-augmented generation, document Q&A.

Deployment

DeepSeek Deployment Tutorial

Ollama, Docker, vLLM, K8s deployment solutions.

Model Details

DeepSeek Model Architecture

Technical architecture, benchmarks, model selection comparison.

Ecosystem Tools

DeepSeek Ecosystem Tools

WebUI, IDE plugins, Agent frameworks, RAG platforms.