AI合成数据生成完全教程
从理论到实战，全面掌握AI合成数据生成技术
目录
- 什么是合成数据
- LLM文本数据生成
- 表格数据合成：SDV与CTGAN
- 图像合成：GAN与Diffusion
- 代码合成数据
- 多模态合成数据
- 数据增强策略
- 合成数据质量评估
- 隐私保护与去标识化
- 合成数据在微调与RAG中的应用
- 实战案例：数据稀缺场景完整解决方案
- 总结与最佳实践
1. 什么是合成数据
合成数据（Synthetic Data）是指通过算法、模型或规则生成的人工数据，而非从真实世界直接采集。在AI时代，合成数据已成为解决数据稀缺、隐私保护和模型训练的重要手段。
为什么需要合成数据？
| 痛点 | 合成数据解决方案 |
|---|---|
| 数据稀缺 | 无限生成训练数据 |
| 隐私合规 | 不含真实个人信息 |
| 标注成本高 | 自动生成带标签数据 |
| 数据不均衡 | 按需生成少数类样本 |
| 数据获取困难 | 模拟罕见/危险场景 |
| 版权风险 | 原创生成，无版权问题 |
合成数据的主要类型
- 文本数据：对话、文章、摘要、翻译对
- 表格数据：用户记录、交易数据、传感器数据
- 图像数据：训练图片、增强样本、风格迁移
- 代码数据：编程题目、代码补全、测试用例
- 多模态数据：图文对、视频描述、语音文本
合成数据的典型应用场景
- 模型微调：生成高质量指令-回答对，用于SFT训练
- RAG知识库：生成问答对验证检索增强效果
- 数据增强：扩充训练集，提升模型泛化能力
- 隐私保护：替代真实敏感数据进行开发测试
- A/B测试：生成模拟用户行为数据
- 边缘场景：模拟罕见事件用于模型训练
2. LLM文本数据生成
利用大语言模型生成文本数据是最直接、最灵活的合成数据方法。
2.1 基础文本生成
"""
使用LLM生成合成文本数据的基础框架
支持多种任务类型的数据生成
"""
import json
import requests
from typing import List, Dict, Optional
from tqdm import tqdm
class LLMDataGenerator:
"""
基于LLM的合成数据生成器
支持：
- 问答对生成
- 文本分类数据
- 摘要数据
- 翻译数据
- 对话数据
"""
def __init__(
self,
model="qwen3:1.7b",
api_url="http://localhost:11434/api/chat",
temperature=0.8,
):
self.model = model
self.api_url = api_url
self.temperature = temperature
def _call_llm(self, system_prompt: str, user_prompt: str) -> str:
"""调用LLM"""
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
"stream": False,
"options": {"temperature": self.temperature},
}
response = requests.post(self.url, json=payload, timeout=120)
return response.json()["message"]["content"]
def generate_qa_pairs(
self,
topic: str,
num_pairs: int = 10,
difficulty: str = "中等"
) -> List[Dict]:
"""
生成问答对
Args:
topic: 主题
num_pairs: 生成数量
difficulty: 难度 (简单/中等/困难)
Returns:
问答对列表
"""
system_prompt = """你是一个专业的数据生成专家。
请根据给定的主题生成高质量的问答对。
要求：
1. 问题要具体、清晰
2. 回答要准确、详细（100-300字）
3. 覆盖该主题的不同方面
4. 难度适中
请以JSON数组格式输出，每个元素包含question和answer字段。
只输出JSON数组，不要其他内容。"""
user_prompt = f"""请生成 {num_pairs} 个关于"{topic}"的问答对。
难度级别：{difficulty}
示例格式：
[
{{"question": "...", "answer": "..."}},
{{"question": "...", "answer": "..."}}
]"""
result = self._call_llm(system_prompt, user_prompt)
# 解析JSON
try:
# 提取JSON部分
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
qa_pairs = json.loads(result.strip())
return qa_pairs
except json.JSONDecodeError:
print(f"JSON解析失败，原始输出: {result[:200]}")
return []
def generate_classification_data(
self,
categories: List[str],
num_per_category: int = 20,
style: str = "多样"
) -> List[Dict]:
"""
生成文本分类数据
Args:
categories: 分类类别列表
num_per_category: 每个类别生成数量
style: 文本风格
Returns:
分类数据列表
"""
system_prompt = """你是一个数据生成专家。
请为文本分类任务生成训练数据。
要求：
1. 每条数据属于指定的一个类别
2. 文本内容要真实自然，像真实用户写的
3. 每个类别内的样本要多样化
4. 文本长度在50-200字之间
请以JSON数组格式输出，每个元素包含text和label字段。
只输出JSON数组。"""
categories_str = "、".join(categories)
user_prompt = f"""请为以下类别生成分类数据：{categories_str}
每个类别生成 {num_per_category} 条数据。
文本风格：{style}
输出格式：
[
{{"text": "...", "label": "类别名"}},
...
]"""
result = self._call_llm(system_prompt, user_prompt)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except json.JSONDecodeError:
return []
def generate_summarization_data(
self,
domain: str,
num_samples: int = 10,
summary_ratio: float = 0.3,
) -> List[Dict]:
"""
生成摘要训练数据
Args:
domain: 领域（如科技、医疗、法律）
num_samples: 生成数量
summary_ratio: 摘要占原文比例
Returns:
包含原文和摘要的数据列表
"""
system_prompt = f"""你是一个{domain}领域的数据生成专家。
请生成用于训练文本摘要模型的配对数据。
要求：
1. 先写一篇{domain}领域的短文（300-500字）
2. 然后写该短文的摘要（约{int(summary_ratio*100)}%长度）
3. 摘要要准确概括原文核心信息
4. 内容要专业、真实
请以JSON数组格式输出，每个元素包含article和summary字段。"""
user_prompt = f"请生成 {num_samples} 组{domain}领域的摘要训练数据。"
result = self._call_llm(system_prompt, user_prompt)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except json.JSONDecodeError:
return []
def generate_dialogue_data(
self,
scenario: str,
num_dialogues: int = 10,
turns: int = 5,
) -> List[List[Dict]]:
"""
生成多轮对话数据
Args:
scenario: 对话场景
num_dialogues: 生成数量
turns: 对话轮数
Returns:
对话列表
"""
system_prompt = """你是一个对话数据生成专家。
请根据给定场景生成自然的多轮对话。
要求：
1. 对话要自然流畅，像真实用户对话
2. 包含用户和助手两个角色
3. 对话要有逻辑连贯性
4. 每轮回复50-200字
请以JSON数组格式输出，每个元素是一个对话（消息列表）。
每条消息包含role和content字段。"""
user_prompt = f"""场景：{scenario}
对话轮数：{turns}轮
生成数量：{num_dialogues}个对话
输出格式：
[
[
{{"role": "user", "content": "..."}},
{{"role": "assistant", "content": "..."}},
...
],
...
]"""
result = self._call_llm(system_prompt, user_prompt)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except json.JSONDecodeError:
return []
# 使用示例
if __name__ == "__main__":
generator = LLMDataGenerator(model="qwen3:1.7b")
# 生成问答对
qa_data = generator.generate_qa_pairs(
topic="Python编程基础",
num_pairs=5,
difficulty="中等"
)
print("=== 问答对 ===")
for qa in qa_data:
print(f"Q: {qa['question']}")
print(f"A: {qa['answer'][:100]}...\n")
# 生成分类数据
cls_data = generator.generate_classification_data(
categories=["正面评价", "负面评价", "中性评价"],
num_per_category=5,
)
print("=== 分类数据 ===")
for item in cls_data:
print(f"[{item['label']}] {item['text'][:80]}...")
2.2 Evol-Instruct：指令进化生成
"""
Evol-Instruct 技术
通过逐步增加复杂度来进化指令数据
这是WizardLM等模型的核心数据生成方法
"""
class EvolInstructionGenerator:
"""
指令进化生成器
将简单指令逐步进化为复杂指令，生成高质量训练数据
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def evolve_instruction(self, simple_instruction: str, evolution_type: str) -> str:
"""
进化单条指令
Args:
simple_instruction: 原始简单指令
evolution_type: 进化类型
Returns:
进化后的复杂指令
"""
evolution_prompts = {
"add_constraints": """请在以下指令基础上增加约束条件，使其更具体、更有挑战性。
约束可以包括：格式要求、字数限制、特定视角、技术栈限制等。
原始指令：{instruction}
只输出进化后的指令，不要其他内容。""",
"deepen": """请将以下指令深化，增加其深度和复杂度。
可以要求更深入的分析、更多的步骤、更详细的解释。
原始指令：{instruction}
只输出进化后的指令。""",
"broaden": """请将以下指令扩展，使其涵盖更广泛的范围。
可以要求比较多个方案、考虑更多因素、覆盖更多场景。
原始指令：{instruction}
只输出进化后的指令。""",
"concretize": """请将以下指令具体化，添加具体的场景、数据或示例。
使指令更贴近实际应用场景。
原始指令：{instruction}
只输出进化后的指令。""",
"reasoning": """请将以下指令改写，要求回答者展示推理过程。
要求逐步分析、解释原因、给出论证。
原始指令：{instruction}
只输出进化后的指令。""",
}
prompt = evolution_prompts[evolution_type].format(instruction=simple_instruction)
return self._call_llm("", prompt)
def generate_response(self, instruction: str) -> str:
"""为指令生成高质量回答"""
system = "你是一个知识渊博、表达清晰的AI助手。请详细、准确地回答以下问题。"
return self._call_llm(system, instruction)
def evolve_and_generate(
self,
seed_instructions: List[str],
num_evolutions: int = 3,
evolution_types: List[str] = None,
) -> List[Dict]:
"""
批量进化指令并生成回答
Args:
seed_instructions: 种子指令列表
num_evolutions: 每条指令的进化次数
evolution_types: 使用的进化类型
Returns:
进化后的指令-回答对列表
"""
if evolution_types is None:
evolution_types = ["add_constraints", "deepen", "broaden", "concretize", "reasoning"]
results = []
for seed in tqdm(seed_instructions, desc="进化指令"):
current_instruction = seed
for i in range(num_evolutions):
# 随机选择进化类型
evo_type = evolution_types[i % len(evolution_types)]
# 进化指令
evolved = self.evolve_instruction(current_instruction, evo_type)
# 生成回答
response = self.generate_response(evolved)
results.append({
"original": seed,
"evolution_type": evo_type,
"evolution_level": i + 1,
"instruction": evolved,
"response": response,
})
current_instruction = evolved
return results
# 使用示例
evol_gen = EvolInstructionGenerator(model="qwen3:1.7b")
seeds = [
"解释什么是机器学习",
"写一个排序算法",
"如何学习Python？",
]
results = evol_gen.evolve_and_generate(seeds, num_evolutions=2)
for r in results[:3]:
print(f"原始: {r['original']}")
print(f"进化类型: {r['evolution_type']}")
print(f"进化后: {r['instruction']}")
print(f"回答: {r['response'][:100]}...\n")
2.3 Self-Instruct：自指令生成
"""
Self-Instruct 方法
让模型自己生成指令、输入和输出
这是Alpaca等模型的核心数据生成方法
"""
class SelfInstructGenerator:
"""
Self-Instruct数据生成器
流程：
1. 从种子任务出发
2. 让模型生成新的任务指令
3. 为新指令生成输入和输出
4. 过滤低质量数据
5. 迭代扩充
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def generate_task_instructions(
self,
seed_tasks: List[Dict],
num_new: int = 10,
) -> List[str]:
"""
基于种子任务生成新的任务指令
Args:
seed_tasks: 种子任务列表，每个包含instruction
num_new: 生成的新指令数量
Returns:
新指令列表
"""
seed_examples = "\n".join([
f"{i+1}. {t['instruction']}"
for i, t in enumerate(seed_tasks[:8])
])
system = """你是一个任务生成专家。请根据示例任务，生成新的、不同的任务指令。
要求：
1. 任务类型要多样化（问答、写作、分析、编程、翻译等）
2. 任务要具体、可执行
3. 不要重复示例中的任务
4. 每条指令独立，不依赖其他指令"""
user = f"""以下是一些任务示例：
{seed_examples}
请生成 {num_new} 个新的任务指令。
每行一个指令，以数字编号开头。"""
result = self._call_llm(system, user)
# 解析指令
instructions = []
for line in result.strip().split("\n"):
line = line.strip()
if line and line[0].isdigit():
# 去掉编号
parts = line.split(".", 1)
if len(parts) > 1:
instructions.append(parts[1].strip())
return instructions[:num_new]
def generate_input_output(self, instruction: str) -> Dict:
"""
为指令生成输入和输出
Returns:
{"instruction": ..., "input": ..., "output": ...}
"""
system = """你是一个数据生成专家。
请为给定的任务指令生成一个具体的输入示例和对应的输出。
有些任务不需要额外输入（如写作任务），此时input为空字符串。
请以JSON格式输出：{"instruction": "...", "input": "...", "output": "..."}"""
user = f"任务指令：{instruction}"
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except:
return {"instruction": instruction, "input": "", "output": result}
def generate_dataset(
self,
seed_tasks: List[Dict],
num_iterations: int = 3,
num_per_iteration: int = 10,
) -> List[Dict]:
"""
迭代生成完整数据集
Args:
seed_tasks: 种子任务
num_iterations: 迭代次数
num_per_iteration: 每次迭代生成数量
Returns:
完整数据集
"""
all_data = list(seed_tasks)
for iteration in range(num_iterations):
print(f"\n=== 迭代 {iteration + 1}/{num_iterations} ===")
# 生成新指令
new_instructions = self.generate_task_instructions(
all_data, num_per_iteration
)
print(f"生成了 {len(new_instructions)} 条新指令")
# 为每条指令生成输入输出
for inst in tqdm(new_instructions, desc="生成数据"):
data = self.generate_input_output(inst)
all_data.append(data)
print(f"总数据量: {len(all_data)}")
return all_data
# 使用示例
seed_tasks = [
{"instruction": "请解释什么是深度学习", "input": "", "output": "深度学习是..."},
{"instruction": "将以下英文翻译成中文", "input": "Hello, how are you?", "output": "你好，你怎么样？"},
{"instruction": "写一首关于春天的诗", "input": "", "output": "春风拂面暖..."},
]
self_gen = SelfInstructGenerator(model="qwen3:1.7b")
dataset = self_gen.generate_dataset(seed_tasks, num_iterations=2, num_per_iteration=5)
# 保存数据集
with open("synthetic_dataset.json", "w", encoding="utf-8") as f:
json.dump(dataset, f, ensure_ascii=False, indent=2)
print(f"数据集已保存，共 {len(dataset)} 条")
3. 表格数据合成：SDV与CTGAN
3.1 SDV（Synthetic Data Vault）
SDV是最流行的表格数据合成库，支持多种合成方法。
"""
使用SDV生成合成表格数据
"""
# pip install sdv
import pandas as pd
from sdv.single_table import CTGANSynthesizer, GaussianCopulaSynthesizer
from sdv.metadata import SingleTableMetadata
# 1. 准备真实数据
real_data = pd.DataFrame({
"age": [25, 30, 35, 28, 42, 55, 23, 38, 45, 31],
"income": [5000, 8000, 12000, 6000, 15000, 20000, 4500, 11000, 18000, 7500],
"education": ["本科", "硕士", "博士", "本科", "硕士", "博士", "本科", "硕士", "博士", "本科"],
"city": ["北京", "上海", "深圳", "广州", "北京", "上海", "深圳", "广州", "北京", "上海"],
"purchased": [0, 1, 1, 0, 1, 1, 0, 1, 1, 0],
})
print("真实数据:")
print(real_data.head())
# 2. 定义元数据
metadata = SingleTableMetadata()
metadata.detect_from_dataframe(real_data)
# 查看检测到的元数据
print("\n元数据:")
print(metadata)
# 3. 使用GaussianCopula（快速，适合数值型数据）
gc_synthesizer = GaussianCopulaSynthesizer(metadata)
gc_synthesizer.fit(real_data)
# 生成合成数据
gc_synthetic = gc_synthesizer.sample(num_rows=100)
print("\nGaussianCopula合成数据:")
print(gc_synthetic.head())
print(f"形状: {gc_synthetic.shape}")
# 4. 使用CTGAN（更强大，适合复杂分布）
ctgan_synthesizer = CTGANSynthesizer(
metadata,
epochs=300, # 训练轮数
batch_size=500, # 批大小
generator_dim=(256, 256), # 生成器维度
discriminator_dim=(256, 256), # 判别器维度
verbose=True,
)
ctgan_synthesizer.fit(real_data)
ctgan_synthetic = ctgan_synthesizer.sample(num_rows=100)
print("\nCTGAN合成数据:")
print(ctgan_synthetic.head())
# 5. 评估合成数据质量
from sdv.evaluation.single_table import evaluate_quality
quality_report = evaluate_quality(
real_data=real_data,
synthetic_data=ctgan_synthetic,
metadata=metadata,
)
print(f"\n合成数据质量分数: {quality_report.get_score():.2f}")
print("各列质量:")
for column_name, score in quality_report.get_details("Column Shapes").iterrows():
print(f" {column_name}: {score['Score']:.2f}")
# 6. 可视化对比
from sdv.evaluation.single_table import get_column_plot
# 对比年龄分布
fig = get_column_plot(
real_data=real_data,
synthetic_data=ctgan_synthetic,
column_name="age",
metadata=metadata,
)
fig.savefig("age_comparison.png", dpi=150, bbox_inches="tight")
print("年龄分布对比图已保存")
# 7. 保存和加载模型
ctgan_synthesizer.save("ctgan_model.pkl")
loaded_synthesizer = CTGANSynthesizer.load("ctgan_model.pkl")
new_synthetic = loaded_synthesizer.sample(num_rows=50)
print(f"\n加载模型后生成: {new_synthetic.shape}")
3.2 CTGAN详解与调优
"""
CTGAN模型详解与调优指南
"""
from sdv.single_table import CTGANSynthesizer
from sdv.metadata import SingleTableMetadata
import pandas as pd
import numpy as np
class CTGANTuner:
"""
CTGAN调优工具
CTGAN的核心组件：
- 生成器（Generator）：生成假数据
- 判别器（Discriminator）：区分真假数据
- 条件生成器（Conditional Generator）：处理离散列
调优关键参数：
- epochs: 训练轮数（越大越精确，但越慢）
- batch_size: 批大小（影响训练稳定性）
- generator_dim / discriminator_dim: 网络维度
- generator_lr / discriminator_lr: 学习率
"""
def __init__(self, data: pd.DataFrame):
self.data = data
self.metadata = SingleTableMetadata()
self.metadata.detect_from_dataframe(data)
def quick_generate(self, num_rows: int = 100) -> pd.DataFrame:
"""快速生成（适合数据探索）"""
model = CTGANSynthesizer(
self.metadata,
epochs=100,
batch_size=500,
verbose=False,
)
model.fit(self.data)
return model.sample(num_rows)
def quality_generate(self, num_rows: int = 100) -> pd.DataFrame:
"""高质量生成（适合正式使用）"""
model = CTGANSynthesizer(
self.metadata,
epochs=500,
batch_size=500,
generator_dim=(256, 256),
discriminator_dim=(256, 256),
generator_lr=2e-4,
discriminator_lr=2e-4,
verbose=True,
)
model.fit(self.data)
return model.sample(num_rows)
def compare_distributions(self, synthetic: pd.DataFrame) -> dict:
"""比较真实数据和合成数据的分布"""
results = {}
for col in self.data.columns:
if self.data[col].dtype in ['int64', 'float64']:
# 数值列：比较统计量
real_stats = self.data[col].describe()
syn_stats = synthetic[col].describe()
results[col] = {
"type": "numeric",
"real_mean": real_stats["mean"],
"syn_mean": syn_stats["mean"],
"real_std": real_stats["std"],
"syn_std": syn_stats["std"],
"mean_diff_pct": abs(real_stats["mean"] - syn_stats["mean"]) / real_stats["mean"] * 100,
}
else:
# 类别列：比较分布
real_dist = self.data[col].value_counts(normalize=True)
syn_dist = synthetic[col].value_counts(normalize=True)
# 计算分布差异
all_cats = set(real_dist.index) | set(syn_dist.index)
max_diff = max(
abs(real_dist.get(c, 0) - syn_dist.get(c, 0))
for c in all_cats
)
results[col] = {
"type": "categorical",
"real_dist": real_dist.to_dict(),
"syn_dist": syn_dist.to_dict(),
"max_category_diff": max_diff,
}
return results
def tune_and_report(self, num_rows: int = 200):
"""完整的调优流程"""
print("1. 快速生成测试...")
quick_syn = self.quick_generate(num_rows)
quick_stats = self.compare_distributions(quick_syn)
print("2. 高质量生成...")
quality_syn = self.quality_generate(num_rows)
quality_stats = self.compare_distributions(quality_syn)
print("\n=== 对比报告 ===")
for col in self.data.columns:
q = quality_stats[col]
if q["type"] == "numeric":
print(f"\n{col} (数值):")
print(f" 真实均值: {q['real_mean']:.2f}, 合成均值: {q['syn_mean']:.2f}")
print(f" 均值差异: {q['mean_diff_pct']:.1f}%")
else:
print(f"\n{col} (类别):")
print(f" 最大类别分布差异: {q['max_category_diff']:.2%}")
return quality_syn
# 使用示例
data = pd.DataFrame({
"年龄": np.random.normal(35, 10, 1000).astype(int).clip(18, 70),
"收入": np.random.lognormal(9, 0.5, 1000).astype(int),
"学历": np.random.choice(["高中", "本科", "硕士", "博士"], 1000, p=[0.2, 0.4, 0.3, 0.1]),
"是否购买": np.random.binomial(1, 0.3, 1000),
})
tuner = CTGANTuner(data)
synthetic_data = tuner.tune_and_report(num_rows=1000)
3.3 时间序列数据合成
"""
使用PAR合成时间序列数据
适用于IoT传感器数据、用户行为序列等
"""
# pip install sdv[deep]
import pandas as pd
import numpy as np
from sdv.sequential import PARSynthesizer
from sdv.metadata import MultiTableMetadata
# 1. 创建时间序列数据
def generate_sensor_data(num_sensors=5, days=30):
"""模拟传感器时间序列数据"""
records = []
for sensor_id in range(num_sensors):
base_temp = 20 + np.random.randn() * 5 # 每个传感器基准温度不同
for day in range(days):
for hour in range(24):
# 模拟温度变化（有日周期）
temp = base_temp + 10 * np.sin(hour * np.pi / 12) + np.random.randn() * 2
humidity = 60 + 20 * np.sin(hour * np.pi / 12 + np.pi) + np.random.randn() * 5
records.append({
"sensor_id": f"sensor_{sensor_id}",
"timestamp": pd.Timestamp("2024-01-01") + pd.Timedelta(days=day, hours=hour),
"temperature": round(temp, 1),
"humidity": round(max(0, min(100, humidity)), 1),
"status": np.random.choice(["normal", "warning"], p=[0.95, 0.05]),
})
return pd.DataFrame(records)
real_ts_data = generate_sensor_data(num_sensors=3, days=7)
print(f"真实时间序列数据: {real_ts_data.shape}")
print(real_ts_data.head())
# 2. 使用PAR生成合成时间序列
# 注意：PAR需要较大的数据量才能有效训练
# 对于小数据集，可以使用更简单的方法
# 3. 简单方法：基于统计特征生成
class SimpleTimeSeriesGenerator:
"""基于统计特征的简单时间序列生成器"""
def __init__(self, real_data: pd.DataFrame, sequence_col: str, value_cols: list):
self.real_data = real_data
self.sequence_col = sequence_col
self.value_cols = value_cols
# 计算统计特征
self.stats = {}
for col in value_cols:
self.stats[col] = {
"mean": real_data[col].mean(),
"std": real_data[col].std(),
"min": real_data[col].min(),
"max": real_data[col].max(),
"autocorr": real_data[col].autocorr(lag=1), # 自相关系数
}
def generate(self, length: int) -> pd.DataFrame:
"""生成指定长度的时间序列"""
data = {}
for col in self.value_cols:
s = self.stats[col]
# 使用AR(1)过程生成有自相关的时间序列
values = [s["mean"]]
for _ in range(length - 1):
# AR(1): x_t = autocorr * x_{t-1} + noise
noise = np.random.randn() * s["std"] * np.sqrt(1 - s["autocorr"]**2)
new_val = s["autocorr"] * values[-1] + (1 - s["autocorr"]) * s["mean"] + noise
new_val = max(s["min"], min(s["max"], new_val))
values.append(new_val)
data[col] = values
return pd.DataFrame(data)
# 使用示例
ts_gen = SimpleTimeSeriesGenerator(
real_ts_data,
sequence_col="timestamp",
value_cols=["temperature", "humidity"]
)
synthetic_ts = ts_gen.generate(length=168) # 一周的小时数据
print(f"\n合成时间序列: {synthetic_ts.shape}")
print(synthetic_ts.describe())
4. 图像合成：GAN与Diffusion
4.1 使用Stable Diffusion生成训练图像
"""
使用Stable Diffusion生成合成图像数据
适用于数据增强、数据稀缺场景
"""
# pip install diffusers transformers torch
from diffusers import StableDiffusionPipeline, StableDiffusionXLPipeline
import torch
import os
from typing import List
class ImageDataGenerator:
"""
基于Stable Diffusion的图像数据生成器
"""
def __init__(
self,
model_id="stabilityai/stable-diffusion-xl-base-1.0",
device="cuda" if torch.cuda.is_available() else "cpu",
):
self.device = device
print(f"加载模型: {model_id}")
if "xl" in model_id.lower():
self.pipe = StableDiffusionXLPipeline.from_pretrained(
model_id,
torch_dtype=torch.float16 if device == "cuda" else torch.float32,
)
else:
self.pipe = StableDiffusionPipeline.from_pretrained(
model_id,
torch_dtype=torch.float16 if device == "cuda" else torch.float32,
)
self.pipe = self.pipe.to(device)
# 内存优化
if device == "cuda":
self.pipe.enable_model_cpu_offload()
def generate_images(
self,
prompts: List[str],
output_dir: str = "./generated_images",
num_images_per_prompt: int = 1,
negative_prompt: str = "blurry, low quality, distorted",
guidance_scale: float = 7.5,
num_inference_steps: int = 30,
) -> List[str]:
"""
批量生成图像
Args:
prompts: 提示词列表
output_dir: 输出目录
num_images_per_prompt: 每个提示词生成几张图
negative_prompt: 负面提示词
guidance_scale: 引导强度
num_inference_steps: 推理步数
Returns:
生成的图像路径列表
"""
os.makedirs(output_dir, exist_ok=True)
generated_paths = []
for i, prompt in enumerate(prompts):
print(f"生成 {i+1}/{len(prompts)}: {prompt[:50]}...")
images = self.pipe(
prompt=prompt,
negative_prompt=negative_prompt,
num_images_per_prompt=num_images_per_prompt,
guidance_scale=guidance_scale,
num_inference_steps=num_inference_steps,
).images
for j, image in enumerate(images):
path = os.path.join(output_dir, f"image_{i:04d}_{j:02d}.png")
image.save(path)
generated_paths.append(path)
print(f"共生成 {len(generated_paths)} 张图像")
return generated_paths
def generate_with_style(
self,
base_prompts: List[str],
styles: List[str],
output_dir: str = "./styled_images",
) -> List[str]:
"""
使用不同风格生成图像
Args:
base_prompts: 基础提示词
styles: 风格描述列表
output_dir: 输出目录
"""
styled_prompts = []
for prompt in base_prompts:
for style in styles:
styled_prompts.append(f"{prompt}, {style}")
return self.generate_images(styled_prompts, output_dir)
# 使用示例
generator = ImageDataGenerator(
model_id="stabilityai/stable-diffusion-xl-base-1.0"
)
# 生成产品图片
product_prompts = [
"a modern wireless headphone on a white background, product photography, studio lighting",
"a sleek smartphone on a marble table, minimalist style, product shot",
"a pair of running shoes, dynamic angle, professional product photography",
]
paths = generator.generate_images(
product_prompts,
output_dir="./product_images",
num_images_per_prompt=4,
)
4.2 图像数据增强
"""
传统图像数据增强方法
不需要生成模型，直接对现有图像进行变换
"""
# pip install albumentations opencv-python
import albumentations as A
import cv2
import numpy as np
from typing import List
import os
class ImageAugmentor:
"""
图像数据增强器
支持的增强方法：
- 几何变换：旋转、翻转、缩放、裁剪
- 颜色变换：亮度、对比度、色调
- 噪声与模糊：高斯噪声、运动模糊
- 遮挡：随机擦除、遮挡
"""
def __init__(self):
# 定义增强管道
self.light_aug = A.Compose([
A.HorizontalFlip(p=0.5),
A.RandomRotate90(p=0.5),
A.RandomBrightnessContrast(
brightness_limit=0.2,
contrast_limit=0.2,
p=0.5,
),
])
self.medium_aug = A.Compose([
A.HorizontalFlip(p=0.5),
A.VerticalFlip(p=0.2),
A.RandomRotate90(p=0.5),
A.ShiftScaleRotate(
shift_limit=0.1,
scale_limit=0.1,
rotate_limit=15,
p=0.5,
),
A.RandomBrightnessContrast(
brightness_limit=0.3,
contrast_limit=0.3,
p=0.5,
),
A.HueSaturationValue(
hue_shift_limit=20,
sat_shift_limit=30,
val_shift_limit=20,
p=0.3,
),
A.GaussNoise(var_limit=(10, 50), p=0.3),
])
self.heavy_aug = A.Compose([
A.HorizontalFlip(p=0.5),
A.VerticalFlip(p=0.3),
A.RandomRotate90(p=0.5),
A.ShiftScaleRotate(
shift_limit=0.2,
scale_limit=0.2,
rotate_limit=30,
p=0.5,
),
A.RandomBrightnessContrast(
brightness_limit=0.4,
contrast_limit=0.4,
p=0.5,
),
A.HueSaturationValue(
hue_shift_limit=30,
sat_shift_limit=40,
val_shift_limit=30,
p=0.5,
),
A.GaussNoise(var_limit=(10, 80), p=0.4),
A.MotionBlur(blur_limit=7, p=0.3),
A.CoarseDropout(
max_holes=8,
max_height=32,
max_width=32,
p=0.3,
),
])
def augment_image(
self,
image_path: str,
num_augmented: int = 5,
level: str = "medium",
) -> List[np.ndarray]:
"""
对单张图像进行增强
Args:
image_path: 图像路径
num_augmented: 生成数量
level: 增强级别 (light/medium/heavy)
Returns:
增强后的图像列表
"""
image = cv2.imread(image_path)
image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
aug_map = {
"light": self.light_aug,
"medium": self.medium_aug,
"heavy": self.heavy_aug,
}
transform = aug_map[level]
augmented = []
for _ in range(num_augmented):
result = transform(image=image)
augmented.append(result["image"])
return augmented
def augment_directory(
self,
input_dir: str,
output_dir: str,
num_per_image: int = 5,
level: str = "medium",
):
"""
批量增强目录中的所有图像
"""
os.makedirs(output_dir, exist_ok=True)
extensions = [".jpg", ".jpeg", ".png", ".bmp"]
image_files = [
f for f in os.listdir(input_dir)
if os.path.splitext(f)[1].lower() in extensions
]
total = 0
for filename in image_files:
filepath = os.path.join(input_dir, filename)
name, ext = os.path.splitext(filename)
augmented = self.augment_image(filepath, num_per_image, level)
for i, aug_img in enumerate(augmented):
aug_img_bgr = cv2.cvtColor(aug_img, cv2.COLOR_RGB2BGR)
output_path = os.path.join(output_dir, f"{name}_aug_{i:03d}{ext}")
cv2.imwrite(output_path, aug_img_bgr)
total += 1
print(f"增强完成: {len(image_files)} 张原图 → {total} 张增强图")
# 使用示例
augmentor = ImageAugmentor()
# 增强单张图片
augmented_images = augmentor.augment_image(
"example.jpg",
num_augmented=10,
level="medium"
)
# 批量增强
augmentor.augment_directory(
input_dir="./original_images",
output_dir="./augmented_images",
num_per_image=10,
level="medium"
)
5. 代码合成数据
5.1 代码训练数据生成
"""
生成代码相关的合成训练数据
用于训练代码助手、代码补全等模型
"""
class CodeDataGenerator:
"""
代码数据生成器
支持生成：
- 代码题目与解答
- 代码补全数据
- 代码解释数据
- 代码重构数据
- Bug修复数据
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def generate_coding_problems(
self,
language: str = "Python",
difficulty: str = "中等",
topics: List[str] = None,
num_problems: int = 10,
) -> List[Dict]:
"""
生成编程题目及解答
Args:
language: 编程语言
difficulty: 难度
topics: 知识点列表
num_problems: 生成数量
"""
if topics is None:
topics = ["数组", "字符串", "链表", "树", "动态规划", "排序", "搜索"]
topics_str = "、".join(topics)
system = f"""你是一个编程教育专家。请生成{language}编程题目。
要求：
1. 题目描述清晰完整
2. 包含输入输出示例
3. 提供完整的解答代码
4. 代码要有详细注释
5. 难度：{difficulty}
请以JSON数组格式输出，每个元素包含：
- title: 题目标题
- description: 题目描述
- examples: 输入输出示例列表
- solution: 解答代码
- explanation: 解题思路
- topics: 涉及的知识点"""
user = f"""请生成 {num_problems} 道{language}编程题目。
涉及知识点：{topics_str}
难度说明：
- 简单：基础语法、简单逻辑
- 中等：数据结构、基本算法
- 困难：复杂算法、优化问题"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except:
return [{"raw": result}]
def generate_code_explanation_data(
self,
code_snippets: List[str],
) -> List[Dict]:
"""
为代码片段生成解释数据
Args:
code_snippets: 代码片段列表
Returns:
代码-解释对列表
"""
system = """你是一个编程教师。请为给定的代码生成详细的解释。
解释要求：
1. 先概述代码的功能
2. 逐行或逐块解释关键逻辑
3. 说明使用的技术点
4. 指出可能的优化空间
请以JSON格式输出：{"code": "原始代码", "explanation": "详细解释"}"""
results = []
for snippet in code_snippets:
result = self._call_llm(system, f"请解释以下代码：\n\n```\n{snippet}\n```")
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
parsed = json.loads(result.strip())
results.append(parsed)
except:
results.append({"code": snippet, "explanation": result})
return results
def generate_bug_fix_data(
self,
language: str = "Python",
num_samples: int = 10,
) -> List[Dict]:
"""
生成Bug修复训练数据
Returns:
包含有bug代码、修复代码、bug描述的数据列表
"""
system = f"""你是一个{language}编程专家。请生成Bug修复训练数据。
每条数据包含：
1. buggy_code: 有bug的代码
2. fixed_code: 修复后的代码
3. bug_description: bug描述
4. fix_explanation: 修复说明
要求：
- bug要真实、常见（如索引错误、类型错误、逻辑错误等）
- 修复代码要正确、简洁
- 解释要清晰
请以JSON数组格式输出。"""
user = f"请生成 {num_samples} 组{language}的Bug修复训练数据。"
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except:
return [{"raw": result}]
def generate_test_cases(
self,
function_code: str,
language: str = "Python",
num_cases: int = 10,
) -> List[Dict]:
"""
为函数生成测试用例
Args:
function_code: 函数代码
language: 编程语言
num_cases: 测试用例数量
"""
system = f"""你是一个测试专家。请为给定的{language}函数生成全面的测试用例。
测试用例应覆盖：
1. 正常输入
2. 边界条件
3. 异常输入
4. 特殊情况
请以JSON数组格式输出，每个元素包含：
- input: 测试输入
- expected_output: 期望输出
- description: 测试描述
- category: 测试类别（正常/边界/异常）"""
user = f"""请为以下函数生成 {num_cases} 个测试用例：
```{language}
{function_code}
```"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except:
return [{"raw": result}]
# 使用示例
code_gen = CodeDataGenerator(model="qwen3:1.7b")
# 生成编程题目
problems = code_gen.generate_coding_problems(
language="Python",
difficulty="中等",
topics=["数组", "排序"],
num_problems=3,
)
for p in problems:
print(f"题目: {p.get('title', 'N/A')}")
print(f"描述: {p.get('description', 'N/A')[:100]}...")
print(f"解答: {p.get('solution', 'N/A')[:100]}...\n")
6. 多模态合成数据
6.1 图文对数据生成
"""
生成图文对训练数据
用于训练多模态模型（如CLIP、LLaVA等）
"""
class MultimodalDataGenerator:
"""
多模态数据生成器
生成图像-文本配对数据，用于：
- 图像描述训练
- 视觉问答(VQA)
- 图文检索
"""
def __init__(self, llm_model="qwen3:1.7b"):
self.llm_model = llm_model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.llm_model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def generate_image_captions(
self,
image_descriptions: List[str],
num_captions_per_image: int = 3,
caption_style: str = "多样",
) -> List[Dict]:
"""
为图像描述生成多种文字描述
Args:
image_descriptions: 图像内容描述列表
num_captions_per_image: 每张图生成的描述数
caption_style: 描述风格
Returns:
图文对列表
"""
system = """你是一个图像描述专家。请为给定的图像内容生成多种文字描述。
要求：
1. 每种描述要从不同角度
2. 描述要自然、准确
3. 包含不同详细程度（简短/中等/详细）
请以JSON数组格式输出，每个元素包含：
- image_desc: 原始图像描述
- caption: 生成的文字描述
- detail_level: 详细程度（brief/medium/detailed）
- angle: 描述角度（整体/细节/情感/场景）"""
results = []
for desc in image_descriptions:
user = f"""图像内容：{desc}
请生成 {num_captions_per_image} 种不同的文字描述。"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
parsed = json.loads(result.strip())
results.extend(parsed if isinstance(parsed, list) else [parsed])
except:
results.append({"image_desc": desc, "caption": result})
return results
def generate_vqa_data(
self,
image_descriptions: List[str],
num_questions_per_image: int = 5,
) -> List[Dict]:
"""
生成视觉问答(VQA)数据
Args:
image_descriptions: 图像内容描述
num_questions_per_image: 每张图的问题数
Returns:
VQA数据列表
"""
system = """你是一个视觉问答数据生成专家。请根据图像描述生成问答对。
问题类型应多样化：
1. 识别类：图中有什么？
2. 计数类：图中有几个？
3. 属性类：某个物体的颜色/大小？
4. 关系类：物体之间的位置关系？
5. 推理类：正在发生什么？为什么？
请以JSON数组格式输出，每个元素包含：
- image_desc: 图像描述
- question: 问题
- answer: 答案
- question_type: 问题类型"""
results = []
for desc in image_descriptions:
user = f"图像描述：{desc}\n请生成 {num_questions_per_image} 个问答对。"
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
parsed = json.loads(result.strip())
results.extend(parsed if isinstance(parsed, list) else [parsed])
except:
results.append({"image_desc": desc, "raw": result})
return results
# 使用示例
multimodal_gen = MultimodalDataGenerator(llm_model="qwen3:1.7b")
image_descs = [
"一只橘猫躺在阳光下的窗台上睡觉",
"城市夜景，高楼大厦灯火辉煌",
"雨天的街道，行人撑着伞匆匆走过",
]
# 生成图像描述
captions = multimodal_gen.generate_image_captions(image_descs, num_captions_per_image=3)
print("=== 图像描述 ===")
for c in captions[:3]:
print(f"图像: {c.get('image_desc', 'N/A')}")
print(f"描述: {c.get('caption', 'N/A')}\n")
# 生成VQA数据
vqa_data = multimodal_gen.generate_vqa_data(image_descs, num_questions_per_image=3)
print("=== VQA数据 ===")
for v in vqa_data[:3]:
print(f"问题: {v.get('question', 'N/A')}")
print(f"答案: {v.get('answer', 'N/A')}\n")
7. 数据增强策略
7.1 文本数据增强
"""
NLP文本数据增强技术
不依赖LLM，纯规则和算法实现
"""
import random
import re
from typing import List
class TextAugmentor:
"""
文本数据增强器
支持的增强方法：
- 同义词替换
- 随机插入
- 随机删除
- 随机交换
- 回译（Back Translation）
- 文本生成模板
"""
def __init__(self):
# 中文同义词词典（示例，实际应使用更大的词典）
self.synonyms = {
"好": ["优秀", "不错", "出色", "棒"],
"坏": ["差", "糟糕", "不好", "劣质"],
"大": ["巨大", "庞大", "很大", "大型"],
"小": ["微小", "很小", "小型", "迷你"],
"快": ["迅速", "快速", "高速", "敏捷"],
"慢": ["缓慢", "迟缓", "低速"],
"美丽": ["漂亮", "好看", "美观", "迷人"],
"重要": ["关键", "核心", "主要", "至关重要"],
}
def synonym_replacement(self, text: str, n: int = 2) -> str:
"""
同义词替换：随机替换n个词为其同义词
"""
words = list(text)
new_words = words.copy()
# 找到可替换的词
replaceable = [
(i, word) for i, word in enumerate(words)
if word in self.synonyms
]
if not replaceable:
return text
# 随机选择要替换的词
random.shuffle(replaceable)
for i, word in replaceable[:n]:
synonym = random.choice(self.synonyms[word])
new_words[i] = synonym
return "".join(new_words)
def random_insertion(self, text: str, n: int = 1) -> str:
"""
随机插入：在随机位置插入同义词
"""
words = list(text)
for _ in range(n):
# 随机选择一个有同义词的词
candidates = [w for w in words if w in self.synonyms]
if not candidates:
break
word = random.choice(candidates)
synonym = random.choice(self.synonyms[word])
# 随机位置插入
pos = random.randint(0, len(words))
words.insert(pos, synonym)
return "".join(words)
def random_deletion(self, text: str, p: float = 0.1) -> str:
"""
随机删除：以概率p删除字符
"""
words = list(text)
if len(words) <= 3:
return text
new_words = [w for w in words if random.random() > p]
# 保证至少有一个字
if not new_words:
return text
return "".join(new_words)
def random_swap(self, text: str, n: int = 1) -> str:
"""
随机交换：随机交换两个字符的位置
"""
words = list(text)
for _ in range(n):
if len(words) < 2:
break
i, j = random.sample(range(len(words)), 2)
words[i], words[j] = words[j], words[i]
return "".join(words)
def augment(
self,
text: str,
num_augmented: int = 5,
methods: List[str] = None,
) -> List[str]:
"""
综合增强：使用多种方法生成增强样本
Args:
text: 原始文本
num_augmented: 生成数量
methods: 使用的方法列表
Returns:
增强后的文本列表
"""
if methods is None:
methods = ["synonym", "insertion", "deletion", "swap"]
augmented = []
method_funcs = {
"synonym": lambda t: self.synonym_replacement(t),
"insertion": lambda t: self.random_insertion(t),
"deletion": lambda t: self.random_deletion(t),
"swap": lambda t: self.random_swap(t),
}
for _ in range(num_augmented):
method = random.choice(methods)
augmented_text = method_funcs[method](text)
augmented.append(augmented_text)
return augmented
def augment_dataset(
self,
texts: List[str],
labels: List[str],
num_per_sample: int = 3,
) -> tuple:
"""
增强整个数据集
Returns:
(augmented_texts, augmented_labels)
"""
all_texts = list(texts)
all_labels = list(labels)
for text, label in zip(texts, labels):
augmented = self.augment(text, num_per_sample)
all_texts.extend(augmented)
all_labels.extend([label] * len(augmented))
return all_texts, all_labels
# 使用示例
augmentor = TextAugmentor()
text = "这家餐厅的菜非常好吃，服务态度也很好"
print(f"原文: {text}")
print(f"同义词替换: {augmentor.synonym_replacement(text)}")
print(f"随机插入: {augmentor.random_insertion(text)}")
print(f"随机删除: {augmentor.random_deletion(text)}")
print(f"随机交换: {augmentor.random_swap(text)}")
# 批量增强
augmented = augmentor.augment(text, num_augmented=5)
print("\n增强结果:")
for aug in augmented:
print(f" {aug}")
7.2 使用LLM进行高级文本增强
"""
使用LLM进行高级文本增强
可以实现更自然、更多样的增强效果
"""
class LLMTextAugmentor:
"""
基于LLM的高级文本增强
方法：
- 改写：保持语义，改变表达方式
- 扩展：增加细节和描述
- 缩写：精简表达
- 风格转换：改变语言风格
- 视角转换：改变叙述角度
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def paraphrase(self, text: str, num_variants: int = 3) -> List[str]:
"""改写：保持语义，改变表达"""
system = """请将以下文本改写为不同的表达方式，保持相同的含义。
每行一个改写版本，不要编号。"""
user = f"请改写 {num_variants} 次：\n{text}"
result = self._call_llm(system, user)
return [line.strip() for line in result.strip().split("\n") if line.strip()]
def expand(self, text: str) -> str:
"""扩展：增加细节"""
system = "请将以下简短文本扩展为更详细、更丰富的描述。保持核心含义不变。"
return self._call_llm(system, text)
def condense(self, text: str) -> str:
"""缩写：精简表达"""
system = "请将以下文本精简为更简洁的表达，保留核心信息。"
return self._call_llm(system, text)
def style_transfer(self, text: str, target_style: str) -> str:
"""风格转换"""
system = f"请将以下文本转换为{target_style}风格，保持核心含义。"
return self._call_llm(system, text)
def augment_for_classification(
self,
text: str,
label: str,
num_augmented: int = 5,
) -> List[Dict]:
"""
为分类任务增强数据
确保增强后的文本保持相同的标签
"""
system = f"""请为以下文本生成 {num_augmented} 个相似的文本变体。
这些文本应该保持相同的类别标签：{label}
要求：
1. 语义相似但表达不同
2. 长度相近
3. 保持相同的情感/类别
4. 每行一个变体"""
user = f"原始文本：{text}"
result = self._call_llm(system, user)
variants = [line.strip() for line in result.strip().split("\n") if line.strip()]
return [{"text": v, "label": label} for v in variants]
# 使用示例
llm_aug = LLMTextAugmentor(model="qwen3:1.7b")
text = "这款手机的拍照效果非常出色"
print(f"原文: {text}")
# 改写
paraphrases = llm_aug.paraphrase(text, 3)
print("\n改写:")
for p in paraphrases:
print(f" {p}")
# 扩展
expanded = llm_aug.expand(text)
print(f"\n扩展: {expanded}")
# 风格转换
formal = llm_aug.style_transfer(text, "正式商务")
print(f"\n正式风格: {formal}")
8. 合成数据质量评估
8.1 评估框架
"""
合成数据质量评估框架
从多个维度评估合成数据的质量
"""
import numpy as np
from typing import List, Dict, Any
from collections import Counter
class SyntheticDataEvaluator:
"""
合成数据质量评估器
评估维度：
1. 多样性（Diversity）：数据是否足够多样
2. 真实性（Fidelity）：数据是否像真实数据
3. 一致性（Consistency）：数据内部是否一致
4. 覆盖度（Coverage）：是否覆盖目标分布
5. 实用性（Utility）：对下游任务是否有用
"""
def evaluate_text_diversity(self, texts: List[str]) -> Dict[str, float]:
"""
评估文本数据的多样性
指标：
- 词汇多样性：不同词数/总词数
- 句子长度方差
- 重复率
"""
# 词汇多样性
all_chars = "".join(texts)
unique_chars = set(all_chars)
char_diversity = len(unique_chars) / len(all_chars) if all_chars else 0
# 句子长度统计
lengths = [len(t) for t in texts]
length_mean = np.mean(lengths)
length_std = np.std(lengths)
# 重复率
unique_texts = set(texts)
duplication_rate = 1 - len(unique_texts) / len(texts)
# n-gram多样性
def get_ngrams(text, n=3):
return [text[i:i+n] for i in range(len(text)-n+1)]
all_trigrams = []
for text in texts:
all_trigrams.extend(get_ngrams(text, 3))
unique_trigrams = set(all_trigrams)
trigram_diversity = len(unique_trigrams) / len(all_trigrams) if all_trigrams else 0
return {
"char_diversity": round(char_diversity, 4),
"length_mean": round(length_mean, 1),
"length_std": round(length_std, 1),
"duplication_rate": round(duplication_rate, 4),
"trigram_diversity": round(trigram_diversity, 4),
"total_samples": len(texts),
"unique_samples": len(unique_texts),
}
def evaluate_label_distribution(
self,
synthetic_labels: List[str],
real_labels: List[str] = None,
) -> Dict[str, Any]:
"""
评估标签分布
Args:
synthetic_labels: 合成数据标签
real_labels: 真实数据标签（可选，用于对比）
"""
syn_dist = Counter(synthetic_labels)
syn_total = len(synthetic_labels)
syn_normalized = {k: v/syn_total for k, v in syn_dist.items()}
result = {
"synthetic_distribution": syn_normalized,
"synthetic_counts": dict(syn_dist),
"num_classes": len(syn_dist),
}
if real_labels:
real_dist = Counter(real_labels)
real_total = len(real_labels)
real_normalized = {k: v/real_total for k, v in real_dist.items()}
# 计算分布差异（JS散度近似）
all_classes = set(syn_normalized.keys()) | set(real_normalized.keys())
max_diff = max(
abs(syn_normalized.get(c, 0) - real_normalized.get(c, 0))
for c in all_classes
)
result["real_distribution"] = real_normalized
result["max_distribution_diff"] = round(max_diff, 4)
return result
def evaluate_text_quality(self, texts: List[str]) -> Dict[str, Any]:
"""
评估文本质量
指标：
- 平均长度
- 空文本率
- 过短文本率
- 过长文本率
- 包含乱码率
"""
total = len(texts)
empty_count = sum(1 for t in texts if not t.strip())
too_short = sum(1 for t in texts if 0 < len(t.strip()) < 10)
too_long = sum(1 for t in texts if len(t) > 5000)
# 简单的乱码检测（连续特殊字符）
garbled = sum(
1 for t in texts
if len(re.findall(r'[^\w\s\u4e00-\u9fff，。！？、；：""''（）]', t)) > len(t) * 0.3
)
lengths = [len(t) for t in texts if t.strip()]
return {
"total": total,
"empty_rate": round(empty_count / total, 4),
"too_short_rate": round(too_short / total, 4),
"too_long_rate": round(too_long / total, 4),
"garbled_rate": round(garbled / total, 4),
"avg_length": round(np.mean(lengths), 1) if lengths else 0,
"median_length": round(np.median(lengths), 1) if lengths else 0,
}
def full_evaluation(
self,
texts: List[str],
labels: List[str] = None,
real_labels: List[str] = None,
) -> Dict[str, Any]:
"""
完整评估
"""
report = {
"text_quality": self.evaluate_text_quality(texts),
"diversity": self.evaluate_text_diversity(texts),
}
if labels:
report["label_distribution"] = self.evaluate_label_distribution(
labels, real_labels
)
# 综合评分
quality = report["text_quality"]
diversity = report["diversity"]
score = 100
score -= quality["empty_rate"] * 50
score -= quality["too_short_rate"] * 30
score -= quality["garbled_rate"] * 40
score -= diversity["duplication_rate"] * 30
score = max(0, min(100, score))
report["overall_score"] = round(score, 1)
report["grade"] = (
"A" if score >= 90 else
"B" if score >= 80 else
"C" if score >= 70 else
"D" if score >= 60 else
"F"
)
return report
# 使用示例
evaluator = SyntheticDataEvaluator()
# 评估合成文本
synthetic_texts = [
"这是一条正面评价，产品质量很好",
"服务态度差，不推荐这家店",
"性价比很高，值得购买",
"物流速度很快，包装完好",
"商品与描述不符，很失望",
]
synthetic_labels = ["正面", "负面", "正面", "正面", "负面"]
report = evaluator.full_evaluation(
texts=synthetic_texts,
labels=synthetic_labels,
)
import json
print(json.dumps(report, ensure_ascii=False, indent=2))
9. 隐私保护与去标识化
9.1 数据去标识化
"""
数据去标识化工具
确保合成数据不包含真实个人信息
"""
import re
from typing import List, Dict
class DataAnonymizer:
"""
数据匿名化/去标识化工具
处理的敏感信息：
- 手机号码
- 身份证号
- 邮箱地址
- 姓名
- 地址
- 银行卡号
- IP地址
"""
def __init__(self):
# 定义敏感信息的正则模式
self.patterns = {
"phone": r'1[3-9]\d{9}',
"id_card": r'\d{17}[\dXx]',
"email": r'[\w.+-]+@[\w-]+\.[\w.-]+',
"bank_card": r'\d{16,19}',
"ip_address": r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',
}
# 替换模板
self.replacements = {
"phone": "[手机号]",
"id_card": "[身份证号]",
"email": "[邮箱]",
"bank_card": "[银行卡号]",
"ip_address": "[IP地址]",
}
def detect_sensitive_info(self, text: str) -> List[Dict]:
"""
检测文本中的敏感信息
Returns:
检测到的敏感信息列表
"""
findings = []
for info_type, pattern in self.patterns.items():
matches = re.finditer(pattern, text)
for match in matches:
findings.append({
"type": info_type,
"value": match.group(),
"start": match.start(),
"end": match.end(),
})
return findings
def anonymize(self, text: str, replacement_type: str = "placeholder") -> str:
"""
匿名化文本中的敏感信息
Args:
text: 原始文本
replacement_type: 替换方式 (placeholder/random/mask)
Returns:
匿名化后的文本
"""
result = text
for info_type, pattern in self.patterns.items():
if replacement_type == "placeholder":
replacement = self.replacements[info_type]
elif replacement_type == "mask":
replacement = "***"
else:
replacement = f"[{info_type}_{hash(text) % 10000:04d}]"
result = re.sub(pattern, replacement, result)
return result
def anonymize_dataset(self, texts: List[str]) -> List[str]:
"""批量匿名化"""
return [self.anonymize(text) for text in texts]
def verify_anonymization(self, original: str, anonymized: str) -> Dict:
"""
验证匿名化是否彻底
Returns:
验证结果
"""
remaining_sensitive = self.detect_sensitive_info(anonymized)
return {
"is_clean": len(remaining_sensitive) == 0,
"remaining_issues": remaining_sensitive,
"original_length": len(original),
"anonymized_length": len(anonymized),
}
# 使用示例
anonymizer = DataAnonymizer()
text = """
用户张三的手机号是13812345678，邮箱是zhangsan@example.com。
身份证号：110101199001011234
银行卡号：6222021234567890123
IP地址：192.168.1.100
"""
print("=== 原始文本 ===")
print(text)
# 检测敏感信息
findings = anonymizer.detect_sensitive_info(text)
print("=== 检测到的敏感信息 ===")
for f in findings:
print(f" [{f['type']}] {f['value']}")
# 匿名化
anonymized = anonymizer.anonymize(text)
print("\n=== 匿名化后 ===")
print(anonymized)
# 验证
result = anonymizer.verify_anonymization(text, anonymized)
print(f"\n=== 验证结果 ===")
print(f"是否干净: {result['is_clean']}")
10. 合成数据在微调与RAG中的应用
10.1 生成SFT微调数据
"""
生成高质量的SFT（监督微调）训练数据
这是合成数据最核心的应用场景之一
"""
class SFTDataGenerator:
"""
SFT训练数据生成器
生成格式：
- 指令-回答对
- 多轮对话
- 系统提示+对话
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def generate_domain_sft_data(
self,
domain: str,
num_samples: int = 50,
difficulty_levels: List[str] = None,
) -> List[Dict]:
"""
生成特定领域的SFT数据
Args:
domain: 领域名称
num_samples: 生成数量
difficulty_levels: 难度级别列表
Returns:
SFT训练数据列表
"""
if difficulty_levels is None:
difficulty_levels = ["简单", "中等", "困难"]
system = f"""你是一个{domain}领域的专家和数据生成专家。
请生成高质量的训练数据，用于微调一个{domain}领域的AI助手。
要求：
1. 问题要真实、具体，像真实用户会问的问题
2. 回答要专业、准确、有深度
3. 回答格式要规范（适当使用标题、列表等）
4. 覆盖该领域的不同子话题
5. 包含不同难度级别
请以JSON数组格式输出，每个元素包含：
- instruction: 用户指令/问题
- input: 额外输入（可为空字符串）
- output: 期望的回答
- difficulty: 难度级别
- topic: 子话题"""
user = f"""请生成 {num_samples} 条{domain}领域的SFT训练数据。
难度级别：{', '.join(difficulty_levels)}
注意：
- 问题要多样化，不要重复
- 回答长度在200-500字之间
- 确保内容准确、专业"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
data = json.loads(result.strip())
return data if isinstance(data, list) else [data]
except:
return [{"raw": result}]
def format_for_training(
self,
data: List[Dict],
format_type: str = "alpaca",
) -> List[Dict]:
"""
格式化为训练框架所需的格式
Args:
data: 原始数据
format_type: 格式类型 (alpaca/sharegpt/vicuna)
"""
if format_type == "alpaca":
# Alpaca格式
return [
{
"instruction": d.get("instruction", ""),
"input": d.get("input", ""),
"output": d.get("output", ""),
}
for d in data
]
elif format_type == "sharegpt":
# ShareGPT格式（多轮对话）
formatted = []
for d in data:
conversations = [
{"from": "human", "value": d.get("instruction", "")},
{"from": "gpt", "value": d.get("output", "")},
]
formatted.append({"conversations": conversations})
return formatted
elif format_type == "chatml":
# ChatML格式
return [
{
"messages": [
{"role": "user", "content": d.get("instruction", "")},
{"role": "assistant", "content": d.get("output", "")},
]
}
for d in data
]
return data
# 使用示例
sft_gen = SFTDataGenerator(model="qwen3:1.7b")
# 生成法律领域数据
legal_data = sft_gen.generate_domain_sft_data(
domain="法律咨询",
num_samples=10,
difficulty_levels=["简单", "中等"],
)
print(f"生成了 {len(legal_data)} 条数据")
# 格式化
alpaca_format = sft_gen.format_for_training(legal_data, "alpaca")
chatml_format = sft_gen.format_for_training(legal_data, "chatml")
# 保存
import json
with open("legal_sft_alpaca.json", "w", encoding="utf-8") as f:
json.dump(alpaca_format, f, ensure_ascii=False, indent=2)
print("Alpaca格式数据已保存")
10.2 生成RAG评估数据
"""
生成RAG系统的评估数据
用于测试RAG系统的检索质量和回答质量
"""
class RAGEvalDataGenerator:
"""
RAG评估数据生成器
生成：
- 问题-文档对（用于评估检索）
- 问题-答案对（用于评估回答质量）
- 反事实问题（用于评估幻觉检测）
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
def _call_llm(self, system: str, user: str) -> str:
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def generate_eval_questions(
self,
documents: List[str],
num_questions_per_doc: int = 3,
) -> List[Dict]:
"""
为文档生成评估问题
Args:
documents: 文档列表
num_questions_per_doc: 每个文档生成的问题数
Returns:
评估数据列表
"""
system = """你是一个RAG系统评估专家。请根据给定的文档生成评估问题。
要求：
1. 问题必须基于文档内容，能够从文档中找到答案
2. 问题类型多样化：
- 事实类：直接从文档提取信息
- 推理类：需要综合文档信息
- 比较类：需要对比文档中的不同部分
3. 每个问题标注需要检索到的文档
请以JSON数组格式输出，每个元素包含：
- question: 问题
- answer: 基于文档的标准答案
- question_type: 问题类型
- source_doc_index: 来源文档索引"""
all_questions = []
for i, doc in enumerate(documents):
user = f"""文档内容：
{doc}
请基于此文档生成 {num_questions_per_doc} 个评估问题。
来源文档索引：{i}"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
questions = json.loads(result.strip())
if isinstance(questions, list):
all_questions.extend(questions)
else:
all_questions.append(questions)
except:
all_questions.append({"raw": result, "source_doc_index": i})
return all_questions
def generate_unanswerable_questions(
self,
domain: str,
num_questions: int = 5,
) -> List[Dict]:
"""
生成无法从知识库回答的问题（用于检测幻觉）
"""
system = f"""请生成一些{domain}领域的问题，这些问题在给定的知识库中找不到答案。
这些问题是用来测试AI系统是否会"编造"不存在的信息。
要求：
1. 问题要合理、真实（用户可能会问）
2. 但确保答案不在常见知识库中
3. 正确的回答应该是"无法确定"或"信息不足"
请以JSON数组格式输出。"""
user = f"请生成 {num_questions} 个{domain}领域无法从通用知识库回答的问题。"
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
return json.loads(result.strip())
except:
return [{"raw": result}]
# 使用示例
rag_eval_gen = RAGEvalDataGenerator(model="qwen3:1.7b")
documents = [
"Python是一种高级编程语言，由Guido van Rossum于1991年创建。Python的设计哲学强调代码的可读性和简洁性。",
"机器学习是人工智能的一个分支。主要分为监督学习、无监督学习和强化学习三种类型。",
]
questions = rag_eval_gen.generate_eval_questions(documents, num_questions_per_doc=2)
print("=== 评估问题 ===")
for q in questions:
print(f"问题: {q.get('question', 'N/A')}")
print(f"答案: {q.get('answer', 'N/A')[:100]}")
print(f"类型: {q.get('question_type', 'N/A')}\n")
11. 实战案例：数据稀缺场景完整解决方案
11.1 完整案例：构建客服AI训练数据
"""
实战案例：从零开始构建客服AI的训练数据
覆盖数据生成、质量评估、隐私保护的完整流程
"""
import json
from typing import List, Dict
class CustomerServiceDataPipeline:
"""
客服AI数据生成流水线
步骤：
1. 定义数据需求
2. 生成种子数据
3. 扩充数据集
4. 质量评估
5. 隐私检查
6. 格式化输出
"""
def __init__(self, model="qwen3:1.7b"):
self.model = model
self.url = "http://localhost:11434/api/chat"
self.all_data = []
def _call_llm(self, system: str, user: str) -> str:
import requests
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"stream": False,
}
resp = requests.post(self.url, json=payload, timeout=120)
return resp.json()["message"]["content"]
def step1_define_scenarios(self) -> List[Dict]:
"""步骤1：定义客服场景"""
scenarios = [
{"category": "订单查询", "intent": "查询订单状态"},
{"category": "订单查询", "intent": "查询物流信息"},
{"category": "退换货", "intent": "申请退货"},
{"category": "退换货", "intent": "申请换货"},
{"category": "退换货", "intent": "查询退款进度"},
{"category": "商品咨询", "intent": "商品功能咨询"},
{"category": "商品咨询", "intent": "商品价格咨询"},
{"category": "商品咨询", "intent": "库存查询"},
{"category": "投诉建议", "intent": "服务投诉"},
{"category": "投诉建议", "intent": "商品质量投诉"},
{"category": "账户问题", "intent": "密码重置"},
{"category": "账户问题", "intent": "账户信息修改"},
]
print(f"定义了 {len(scenarios)} 个客服场景")
return scenarios
def step2_generate_seed_data(self, scenarios: List[Dict], num_per_scenario: int = 5) -> List[Dict]:
"""步骤2：为每个场景生成种子数据"""
system = """你是一个客服对话数据生成专家。
请为给定的客服场景生成训练数据。
每条数据是一个完整的客服对话，包含：
- 用户的初始问题
- 客服的回复
- 用户可能的追问
- 客服的后续回复
请以JSON数组格式输出，每个元素包含：
- scenario: 场景类别
- intent: 用户意图
- conversation: 对话列表，每条包含role和content
- resolution: 问题是否解决（resolved/unresolved/escalated）"""
all_data = []
for scenario in scenarios:
user = f"""场景：{scenario['category']} - {scenario['intent']}
请生成 {num_per_scenario} 条客服对话数据。
对话要真实自然，像真实的在线客服对话。"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
data = json.loads(result.strip())
if isinstance(data, list):
all_data.extend(data)
else:
all_data.append(data)
except:
all_data.append({
"scenario": scenario["category"],
"intent": scenario["intent"],
"raw": result,
})
print(f"生成了 {len(all_data)} 条种子数据")
return all_data
def step3_expand_dataset(self, seed_data: List[Dict], target_size: int = 200) -> List[Dict]:
"""步骤3：扩充数据集"""
# 使用改写和变体生成来扩充
expanded = list(seed_data)
system = """请为以下客服对话生成变体。
保持相同的问题类型和解决方式，但改变具体的表述方式。
以JSON数组格式输出变体对话。"""
remaining = target_size - len(expanded)
iterations = remaining // len(seed_data) + 1
for i in range(min(iterations, 3)):
for item in seed_data[:10]: # 取前10条生成变体
if len(expanded) >= target_size:
break
conversation = item.get("conversation", [])
if not conversation:
continue
user = f"""原始对话：
{json.dumps(conversation, ensure_ascii=False, indent=2)}
请生成2个变体对话。"""
result = self._call_llm(system, user)
try:
if "```" in result:
result = result.split("```")[1]
if result.startswith("json"):
result = result[4:]
variants = json.loads(result.strip())
if isinstance(variants, list):
expanded.extend(variants)
except:
pass
print(f"扩充后数据量: {len(expanded)}")
return expanded[:target_size]
def step4_quality_check(self, data: List[Dict]) -> Dict:
"""步骤4：质量检查"""
evaluator = SyntheticDataEvaluator()
# 提取对话文本
texts = []
for item in data:
conv = item.get("conversation", [])
for msg in conv:
texts.append(msg.get("content", ""))
report = evaluator.evaluate_text_quality(texts)
diversity = evaluator.evaluate_text_diversity(texts)
return {"quality": report, "diversity": diversity}
def step5_privacy_check(self, data: List[Dict]) -> List[Dict]:
"""步骤5：隐私检查"""
anonymizer = DataAnonymizer()
cleaned = []
issues_found = 0
for item in data:
conv = item.get("conversation", [])
cleaned_conv = []
for msg in conv:
content = msg.get("content", "")
findings = anonymizer.detect_sensitive_info(content)
if findings:
issues_found += len(findings)
content = anonymizer.anonymize(content)
cleaned_conv.append({**msg, "content": content})
cleaned.append({**item, "conversation": cleaned_conv})
print(f"隐私检查完成: 发现并处理了 {issues_found} 处敏感信息")
return cleaned
def step6_format_output(self, data: List[Dict], output_path: str = "cs_training_data.json"):
"""步骤6：格式化输出"""
# 转换为ChatML格式
formatted = []
for item in data:
conv = item.get("conversation", [])
if not conv:
continue
messages = []
for msg in conv:
role = "user" if msg.get("role") == "user" else "assistant"
messages.append({"role": role, "content": msg.get("content", "")})
formatted.append({
"messages": [
{"role": "system", "content": "你是一个专业的客服助手。请友好、专业地回答用户的问题。"},
*messages,
],
"metadata": {
"scenario": item.get("scenario", ""),
"intent": item.get("intent", ""),
"resolution": item.get("resolution", ""),
}
})
with open(output_path, "w", encoding="utf-8") as f:
json.dump(formatted, f, ensure_ascii=False, indent=2)
print(f"已保存 {len(formatted)} 条训练数据到 {output_path}")
return formatted
def run_pipeline(self, target_size: int = 50):
"""运行完整的数据生成流水线"""
print("=" * 50)
print("客服AI训练数据生成流水线")
print("=" * 50)
# 步骤1
print("\n[1/6] 定义客服场景...")
scenarios = self.step1_define_scenarios()
# 步骤2
print("\n[2/6] 生成种子数据...")
seed_data = self.step2_generate_seed_data(scenarios, num_per_scenario=3)
# 步骤3
print("\n[3/6] 扩充数据集...")
expanded_data = self.step3_expand_dataset(seed_data, target_size)
# 步骤4
print("\n[4/6] 质量检查...")
quality_report = self.step4_quality_check(expanded_data)
print(f"质量报告: {json.dumps(quality_report, ensure_ascii=False, indent=2)}")
# 步骤5
print("\n[5/6] 隐私检查...")
clean_data = self.step5_privacy_check(expanded_data)
# 步骤6
print("\n[6/6] 格式化输出...")
final_data = self.step6_format_output(clean_data)
print("\n" + "=" * 50)
print(f"流水线完成！共生成 {len(final_data)} 条训练数据")
print("=" * 50)
return final_data
# 运行完整流水线
if __name__ == "__main__":
pipeline = CustomerServiceDataPipeline(model="qwen3:1.7b")
training_data = pipeline.run_pipeline(target_size=30)
12. 总结与最佳实践
12.1 合成数据生成检查清单
✅ 数据规划
□ 明确数据用途（训练/测试/验证）
□ 确定数据规模需求
□ 定义数据质量标准
□ 规划数据分布
✅ 数据生成
□ 选择合适的生成方法
□ 设置合理的生成参数
□ 多样化生成策略
□ 迭代优化生成质量
✅ 质量控制
□ 自动化质量评估
□ 人工抽检验证
□ 分布一致性检查
□ 去重和去噪
✅ 隐私保护
□ 敏感信息检测
□ 数据匿名化处理
□ 合规性验证
□ 访问权限控制
✅ 持续优化
□ 基于模型效果反馈优化
□ 定期更新生成策略
□ 积累领域知识库
□ 建立数据版本管理
12.2 常见问题与解决方案
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 数据多样性不足 | 生成提示词单一 | 多样化提示词模板 |
| 数据质量参差不齐 | 模型输出不稳定 | 增加质量过滤和人工审核 |
| 合成数据有偏见 | 基础模型偏见 | 使用多个模型交叉生成 |
| 生成速度慢 | LLM推理瓶颈 | 批量处理、并发调用 |
| 隐私泄露风险 | 生成过程中记忆真实数据 | 加强匿名化和后处理 |
| 标签不准确 | 模型理解偏差 | 多模型验证和人工校验 |
12.3 工具与资源推荐
- LLM数据生成：Ollama + 本地SLM，隐私安全
- 表格数据合成：SDV（Synthetic Data Vault）
- 图像生成：Stable Diffusion、DALL-E
- 数据增强：nlpaug（文本）、albumentations（图像）
- 质量评估：SDV评估工具、自定义评估框架
- 隐私保护：Presidio、自定义匿名化工具
本文档持续更新。 合成数据技术发展迅速，建议定期关注最新研究和工具更新。
