数据工程师用合成数据构建企业知识问答系统数据飞轮

import openai
import json
import hashlib
from typing import Optional
from datasketch import MinHash, MinHashLSH

client = openai.OpenAI()

# 数据工程领域的种子指令（手写 20-30 条）
SEED_INSTRUCTIONS = [
    "解释 Apache Spark 中 RDD、DataFrame 和 Dataset 的区别",
    "用 PySpark 写一个读取 Parquet 文件并做聚合统计的示例",
    "设计一个处理每日 10 亿条日志的数据管道架构",
    "解释 Kafka 中 Partition、Offset 和 Consumer Group 的关系",
    "什么是数据湖、数据仓库、数据湖仓一体（Lakehouse）？各自适用场景？",
    "用 Flink 实现一个实时计算用户活跃度的窗口任务",
    "解释 Hive 分区表和分桶表的区别及使用场景",
    "如何优化一个慢查询的 Spark Job？",
]

def generate_instructions(seed_instructions: list[str], n: int = 10) -> list[str]:
    """用 LLM 基于种子指令生成新指令"""
    seed_sample = "\n".join([f"- {inst}" for inst in seed_instructions[:8]])

    prompt = f"""你是一位资深数据工程师。请参考以下示例指令的风格和难度，生成 {n} 条全新的、不重复的数据工程领域指令。

示例指令：
{seed_sample}

要求：
1. 涵盖 Spark、Flink、Kafka、Hive、数据仓库、数据湖等主题
2. 难度适中到高级
3. 每条指令具体、可操作
4. 直接输出指令列表，每行一条，不加编号

生成 {n} 条指令："""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.9,  # 提高温度增加多样性
    )

    instructions = response.choices[0].message.content.strip().split("\n")
    return [inst.strip() for inst in instructions if inst.strip()]

def generate_answer(instruction: str) -> Optional[str]:
    """为指令生成高质量答案"""
    prompt = f"""你是一位有 10 年经验的数据工程师，请回答以下问题。要求：
1. 回答准确、专业
2. 包含具体的代码示例（如果适用）
3. 说明优缺点和适用场景
4. 控制在 300-800 字

问题：{instruction}"""

    try:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,  # 降低温度保证准确性
        )
        return response.choices[0].message.content.strip()
    except Exception as e:
        print(f"生成答案失败: {e}")
        return None

def score_quality(instruction: str, answer: str) -> float:
    """用 LLM 对问答对打质量分（0-10）"""
    prompt = f"""请评估以下问答对的质量，返回 0-10 的分数（只返回数字）。

评估标准：
- 指令清晰度（2分）：指令是否清晰、具体、可操作
- 答案准确性（3分）：技术内容是否正确
- 答案完整性（3分）：是否覆盖了问题的主要方面
- 实用性（2分）：是否包含代码或具体示例

问题：{instruction}
答案：{answer[:500]}...

质量分（0-10）："""

    response = client.chat.completions.create(
        model="gpt-4o-mini",  # 用小模型评分，省成本
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )

    try:
        return float(response.choices[0].message.content.strip())
    except:
        return 5.0

def deduplicate_minhash(instructions: list[str], threshold: float = 0.8) -> list[str]:
    """使用 MinHash LSH 去除相似指令"""
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    unique_instructions = []

    for i, inst in enumerate(instructions):
        minhash = MinHash(num_perm=128)
        for word in inst.split():
            minhash.update(word.encode("utf-8"))

        if not lsh.query(minhash):
            lsh.insert(str(i), minhash)
            unique_instructions.append(inst)

    return unique_instructions

def build_sft_dataset(
    n_instructions: int = 500,
    quality_threshold: float = 7.0,
    output_file: str = "data_engineering_sft.jsonl"
) -> list[dict]:
    """完整的 SFT 数据生成流程"""

    # Step 1: 生成指令
    print("Step 1: 生成指令...")
    all_instructions = list(SEED_INSTRUCTIONS)

    while len(all_instructions) < n_instructions * 2:  # 生成 2 倍数量，过滤后取目标数量
        new_instructions = generate_instructions(all_instructions, n=20)
        all_instructions.extend(new_instructions)
        print(f"  当前指令数: {len(all_instructions)}")

    # Step 2: 去重
    print("Step 2: MinHash 去重...")
    unique_instructions = deduplicate_minhash(all_instructions)
    print(f"  去重后: {len(unique_instructions)} 条")

    # Step 3: 生成答案 + 质量过滤
    print("Step 3: 生成答案并过滤...")
    dataset = []

    for i, instruction in enumerate(unique_instructions[:n_instructions * 2]):
        answer = generate_answer(instruction)
        if answer is None:
            continue

        score = score_quality(instruction, answer)

        if score >= quality_threshold:
            dataset.append({
                "instruction": instruction,
                "output": answer,
                "quality_score": score,
            })
            print(f"  [{i+1}] 得分 {score:.1f} ")
        else:
            print(f"  [{i+1}] 得分 {score:.1f}  过滤")

        if len(dataset) >= n_instructions:
            break

    # Step 4: 保存
    with open(output_file, "w", encoding="utf-8") as f:
        for item in dataset:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")

    print(f"\n最终数据集: {len(dataset)} 条，保存至 {output_file}")
    return dataset

if __name__ == "__main__":
    dataset = build_sft_dataset(n_instructions=500, quality_threshold=7.0)

文档切片 → LLM 生成问题 → 验证答案可从文档中找到

import openai
import json
from typing import Optional
from langchain.text_splitter import RecursiveCharacterTextSplitter

client = openai.OpenAI()

def chunk_document(text: str, chunk_size: int = 512, overlap: int = 50) -> list[str]:
    """文档切片"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", "。", "！", "？", " "],
    )
    return splitter.split_text(text)

def generate_questions_from_chunk(chunk: str, n_questions: int = 3) -> list[str]:
    """从文档片段生成问题"""
    prompt = f"""基于以下文档片段，生成 {n_questions} 个有价值的问题。

要求：
1. 问题必须可以从文档中找到答案
2. 问题要具体，不要泛泛而问
3. 问题难度适中，有实际意义
4. 每行一个问题，不加编号

文档片段：
{chunk}

生成的问题："""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )

    questions = response.choices[0].message.content.strip().split("\n")
    return [q.strip() for q in questions if q.strip()]

def verify_and_generate_answer(question: str, chunk: str) -> Optional[dict]:
    """验证问题可回答，并生成参考答案"""
    prompt = f"""请基于以下文档片段回答问题。
如果文档中没有足够信息回答，请回复"UNANSWERABLE"。

文档：
{chunk}

问题：{question}

答案："""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,
    )

    answer = response.choices[0].message.content.strip()

    if "UNANSWERABLE" in answer.upper():
        return None

    return {
        "question": question,
        "answer": answer,
        "context": chunk,
        # RAGAS 兼容格式
        "ground_truth": answer,
        "contexts": [chunk],
    }

def build_rag_dataset(
    documents: list[str],
    questions_per_chunk: int = 2,
    output_file: str = "rag_eval_dataset.jsonl"
) -> list[dict]:
    """
    从文档列表构建 RAG 评估数据集
    输出格式与 RAGAS 框架兼容，可直接用于评估
    """
    dataset = []

    for doc_idx, doc in enumerate(documents):
        chunks = chunk_document(doc)
        print(f"文档 {doc_idx + 1}: {len(chunks)} 个片段")

        for chunk_idx, chunk in enumerate(chunks):
            if len(chunk) < 100:  # 过滤太短的片段
                continue

            questions = generate_questions_from_chunk(chunk, n_questions=questions_per_chunk)

            for question in questions:
                qa_pair = verify_and_generate_answer(question, chunk)
                if qa_pair:
                    qa_pair["doc_id"] = doc_idx
                    qa_pair["chunk_id"] = chunk_idx
                    dataset.append(qa_pair)
                    print(f"  问答对 {len(dataset)}: {question[:50]}...")

    # 保存为 RAGAS 兼容格式
    with open(output_file, "w", encoding="utf-8") as f:
        for item in dataset:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")

    print(f"\n生成 {len(dataset)} 个问答对，保存至 {output_file}")
    return dataset

# 使用示例
if __name__ == "__main__":
    # 加载企业文档（以 Spark 官方文档为例）
    with open("spark_documentation.txt", "r") as f:
        documents = [f.read()]

    dataset = build_rag_dataset(
        documents=documents,
        questions_per_chunk=2,
        output_file="spark_rag_eval.jsonl"
    )

    # 直接用 RAGAS 评估
    # from ragas import evaluate
    # from datasets import Dataset
    # eval_dataset = Dataset.from_list(dataset)
    # results = evaluate(eval_dataset)

import openai
import random
from sentence_transformers import SentenceTransformer, util
import torch

client = openai.OpenAI()

def generate_positive_pairs(chunk: str, n: int = 2) -> list[tuple[str, str]]:
    """从同一文档片段生成语义相似的文本对"""
    prompt = f"""基于以下文档，生成 {n} 个语义相同但表达不同的句子对。

原文档：
{chunk[:300]}

要求：改写成不同的表达方式，保持语义一致。每行一对，用 ||| 分隔。

示例格式：
Spark 是分布式计算框架 ||| Apache Spark 是用于大规模数据处理的分布式系统

生成的句子对："""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.7,
    )

    pairs = []
    for line in response.choices[0].message.content.strip().split("\n"):
        if "|||" in line:
            parts = line.split("|||")
            if len(parts) == 2:
                pairs.append((parts[0].strip(), parts[1].strip()))

    return pairs

def mine_hard_negatives(
    query: str,
    candidate_texts: list[str],
    model_name: str = "BAAI/bge-m3",
    top_k: int = 10,
    hard_negative_range: tuple = (2, 6),  # 取第2-6名作为难负例
) -> list[str]:
    """
    难负例挖掘：找语义相近但实际不相关的样本

    hard_negative_range: 跳过第1名（真正相关），取2-6名作为难负例
    """
    model = SentenceTransformer(model_name)

    query_emb = model.encode(query, convert_to_tensor=True)
    cand_embs = model.encode(candidate_texts, convert_to_tensor=True)

    scores = util.cos_sim(query_emb, cand_embs)[0]
    top_indices = torch.argsort(scores, descending=True)

    start, end = hard_negative_range
    hard_negatives = [
        candidate_texts[idx]
        for idx in top_indices[start:end]
    ]

    return hard_negatives

def build_embedding_training_data(
    chunks: list[str],
    output_file: str = "embedding_train.jsonl"
) -> list[dict]:
    """
    构建 Embedding 微调数据集
    格式：{"query": ..., "positive": ..., "negative": ...}
    """
    dataset = []

    for i, chunk in enumerate(chunks):
        # 生成正例对
        pos_pairs = generate_positive_pairs(chunk, n=2)

        for query, positive in pos_pairs:
            # 挖掘难负例（从其他 chunk 中找）
            other_chunks = [c for j, c in enumerate(chunks) if j != i]
            if len(other_chunks) > 5:
                sample_chunks = random.sample(other_chunks, min(50, len(other_chunks)))
                hard_negs = mine_hard_negatives(query, sample_chunks)

                for neg in hard_negs[:2]:  # 每个正例配 2 个难负例
                    dataset.append({
                        "query": query,
                        "positive": positive,
                        "negative": neg,
                    })

    with open(output_file, "w", encoding="utf-8") as f:
        for item in dataset:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")

    return dataset

import numpy as np
from openai import OpenAI

client = OpenAI()

def get_token_probs(text: str, model: str = "gpt-4o-mini") -> dict:
    """
    获取模型对某个问题的输出概率分布
    使用 logprobs 参数获取 token 级别的概率
    """
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "请将以下文本分类为：正面/负面/中性。只输出分类标签。"},
            {"role": "user", "content": text},
        ],
        logprobs=True,
        top_logprobs=5,
        max_tokens=5,
    )

    # 提取第一个 token 的概率分布
    top_logprobs = response.choices[0].logprobs.content[0].top_logprobs

    probs = {}
    for lp in top_logprobs:
        probs[lp.token] = np.exp(lp.logprob)  # logprob -> prob

    return probs

def entropy_score(probs: dict) -> float:
    """计算信息熵。熵越高，模型越不确定"""
    values = np.array(list(probs.values()))
    values = values / values.sum()  # 归一化
    return float(-np.sum(values * np.log(values + 1e-10)))

def margin_score(probs: dict) -> float:
    """计算 Margin。最高概率与次高概率之差越小，模型越不确定"""
    sorted_probs = sorted(probs.values(), reverse=True)
    if len(sorted_probs) < 2:
        return 0.0
    return sorted_probs[0] - sorted_probs[1]

def active_learning_selection(
    unlabeled_samples: list[str],
    budget: int,
    strategy: str = "entropy"  # "entropy" 或 "margin"
) -> list[tuple[str, float]]:
    """
    主动学习：从未标注数据中选出最值得标注的 budget 条

    返回：(样本文本, 不确定性分数) 的列表，按分数降序排列
    """
    scored_samples = []

    for sample in unlabeled_samples:
        probs = get_token_probs(sample)

        if strategy == "entropy":
            score = entropy_score(probs)
        else:  # margin（分数越低越不确定）
            score = -margin_score(probs)  # 取负，统一为越高越不确定

        scored_samples.append((sample, score, probs))

    # 按不确定性排序，取前 budget 条
    scored_samples.sort(key=lambda x: x[1], reverse=True)
    selected = [(sample, score) for sample, score, _ in scored_samples[:budget]]

    print(f"从 {len(unlabeled_samples)} 条未标注数据中选出 {budget} 条高价值样本")
    print(f"平均不确定性分数: {np.mean([s for _, s in selected]):.3f}")

    return selected

# 使用示例
if __name__ == "__main__":
    # 假设有 1000 条用户问题待标注
    unlabeled = [
        "这个数据产品用起来还行",
        "Spark 的 shuffle 为什么这么慢",
        "还可以吧",
        # ... 更多样本
    ]

    # 只有 50 条的标注预算，优先选最有价值的
    selected = active_learning_selection(
        unlabeled_samples=unlabeled,
        budget=50,
        strategy="entropy"
    )

    print("\n最应该标注的前5条：")
    for text, score in selected[:5]:
        print(f"  [{score:.3f}] {text}")

# 使用 2.2 节的 build_rag_dataset 函数
documents = load_pdf_documents("technical_manuals/")  # 500 份手册

# 目标：生成 500 个有代表性的问答对
eval_dataset = build_rag_dataset(
    documents=documents,
    questions_per_chunk=1,  # 每个片段1个问题，避免重复
    output_file="equipment_qa_eval.jsonl"
)

from datetime import datetime
import sqlite3

class InteractionLogger:
    """记录用户交互，为数据飞轮提供燃料"""

    def __init__(self, db_path: str = "interactions.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS interactions (
                id INTEGER PRIMARY KEY,
                session_id TEXT,
                question TEXT,
                retrieved_chunks TEXT,  -- JSON 数组
                answer TEXT,
                user_feedback INTEGER,  -- 1=满意, -1=不满意, 0=无反馈
                user_correction TEXT,   -- 用户修改后的答案（如果有）
                timestamp TEXT,
                latency_ms INTEGER
            )
        """)
        self.conn.commit()

    def log(
        self,
        session_id: str,
        question: str,
        retrieved_chunks: list[str],
        answer: str,
        latency_ms: int,
    ) -> int:
        """记录一次问答交互"""
        import json
        cursor = self.conn.execute(
            """INSERT INTO interactions
               (session_id, question, retrieved_chunks, answer,
                user_feedback, timestamp, latency_ms)
               VALUES (?, ?, ?, ?, 0, ?, ?)""",
            (session_id, question, json.dumps(retrieved_chunks, ensure_ascii=False),
             answer, datetime.now().isoformat(), latency_ms)
        )
        self.conn.commit()
        return cursor.lastrowid

    def update_feedback(self, interaction_id: int, feedback: int, correction: str = None):
        """更新用户反馈（异步）"""
        self.conn.execute(
            "UPDATE interactions SET user_feedback=?, user_correction=? WHERE id=?",
            (feedback, correction, interaction_id)
        )
        self.conn.commit()

    def get_low_quality_interactions(self, limit: int = 100) -> list[dict]:
        """获取低质量交互（用于主动学习选择）"""
        cursor = self.conn.execute(
            """SELECT * FROM interactions
               WHERE user_feedback = -1 OR
                     (user_feedback = 0 AND julianday('now') - julianday(timestamp) > 1)
               ORDER BY timestamp DESC LIMIT ?""",
            (limit,)
        )
        columns = [desc[0] for desc in cursor.description]
        return [dict(zip(columns, row)) for row in cursor.fetchall()]

def select_samples_for_annotation(logger: InteractionLogger, budget: int = 20) -> list[dict]:
    """
    从低质量交互中，用主动学习策略选出最值得人工标注的样本
    """
    low_quality = logger.get_low_quality_interactions(limit=200)

    # 用不确定性评分排序
    questions = [item["question"] for item in low_quality]
    selected_with_scores = active_learning_selection(questions, budget=budget)

    # 对应回原始记录
    selected_questions = {q for q, _ in selected_with_scores}
    selected_items = [item for item in low_quality if item["question"] in selected_questions]

    return selected_items[:budget]

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader

def finetune_embedding_with_corrections(
    corrections: list[dict],  # {"question": ..., "correct_answer": ..., "retrieved_chunks": ...}
    base_model: str = "BAAI/bge-m3",
    output_path: str = "finetuned_embedding",
):
    """
    用人工修正数据微调 Embedding 模型

    核心思路：正确答案所在的 chunk 应该与问题更相似（正例）
    被错误检索的 chunk 应该与问题不那么相似（负例）
    """
    model = SentenceTransformer(base_model)

    train_examples = []
    for item in corrections:
        question = item["question"]
        correct_chunk = item.get("correct_chunk", item["correct_answer"])

        # 从错误检索的 chunks 中选难负例
        wrong_chunks = item.get("retrieved_chunks", [])

        if wrong_chunks:
            # MultipleNegativesRankingLoss 格式
            example = InputExample(texts=[question, correct_chunk] + wrong_chunks[:3])
            train_examples.append(example)

    train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
    train_loss = losses.MultipleNegativesRankingLoss(model)

    model.fit(
        train_objectives=[(train_dataloader, train_loss)],
        epochs=3,
        warmup_steps=100,
        output_path=output_path,
    )

    print(f"微调完成，模型保存至 {output_path}")
    return model

def evaluate_rag_improvement(
    eval_dataset_path: str,
    old_embedding_model: str,
    new_embedding_model: str,
) -> dict:
    """
    对比新旧 Embedding 模型在 RAG 评估集上的表现
    结合 33-LLM评估体系 中的评估框架
    """
    from ragas import evaluate
    from ragas.metrics import faithfulness, answer_relevancy, context_recall
    from datasets import Dataset

    with open(eval_dataset_path) as f:
        eval_data = [json.loads(line) for line in f]

    results = {}
    for model_name, is_new in [(old_embedding_model, False), (new_embedding_model, True)]:
        # 用对应 Embedding 模型重新检索
        dataset = Dataset.from_list(eval_data)
        scores = evaluate(
            dataset=dataset,
            metrics=[faithfulness, answer_relevancy, context_recall],
        )
        results["new" if is_new else "old"] = scores

    improvement = {
        metric: results["new"][metric] - results["old"][metric]
        for metric in results["new"]
    }

    print("\n评估结果对比：")
    for metric, diff in improvement.items():
        direction = "+" if diff > 0 else ""
        print(f"  {metric}: {direction}{diff:.3f}")

    return improvement