{
  "Character": "cardiologist",
  "Question_Type": "diagnosis",
  "Difficulty": "university"
}

# 輸入配置
config = {
  "Character": "farmer",
  "Question_Type": "keywords",
  "Difficulty": "high_school",
  "Language": "Chinese",
  "Length": 12
}


# 生成查詢
query = "春季養(yǎng)鴨飼料調(diào)整方法"

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer




class Qwen3DataGenerator:
    def __init__(self, model_path="Qwen/Qwen3-32B"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path, 
            torch_dtype=torch.bfloat16,
            device_map="auto"
        )
        self.persona_hub = ["farmer", "researcher", "engineer", "doctor", "teacher"]


    def generate_config(self, document, language="English"):
        """生成文檔的三維配置"""
        prompt = f"""
Given a document, generate a JSON configuration with Character, Question_Type, Difficulty.
Document: {document}
Language: {language}
Character candidates: {self.persona_hub}
Question_Type options: keywords, acquire_knowledge, summary, yes_or_no, background
Difficulty options: high_school, university, phd
Output JSON:
"""
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
        outputs = self.model.generate(
            **inputs,
            max_new_tokens=100,
            temperature=0.7,
            top_p=0.9
        )
        config_json = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        return eval(config_json)  # 解析生成的JSON

from sentence_transformers import SentenceTransformer, util
from datasets import load_dataset
import numpy as np




class QueryGenerator:
    def __init__(self, embedding_model="Qwen/Qwen3-Embedding-0.6B"):
        self.embedding_model = SentenceTransformer(embedding_model)


    def generate_query(self, config, document):
        """根據(jù)配置生成查詢"""
        role = config["Character"]
        q_type = config["Question_Type"]
        difficulty = config["Difficulty"]
        lang = config.get("Language", "Chinese")


        # 構(gòu)建生成提示詞
        prompt = f"""
Generate a {lang} query based on the document and config.
Document: {document}
Character: {role}
Question_Type: {q_type}
Difficulty: {difficulty}
Query:
"""
        # 這里使用Qwen3-32B生成查詢，實(shí)際應(yīng)用中需替換為真實(shí)生成邏輯
        return f"示例查詢: 根據(jù){role}需求生成的{q_type}查詢"


    def filter_high_quality_pairs(self, pairs, threshold=0.7):
        """篩選高相似度數(shù)據(jù)對(duì)"""
        queries = [p["query"] for p in pairs]
        docs = [p["document"] for p in pairs]


        # 生成嵌入向量
        query_emb = self.embedding_model.encode(queries, batch_size=32)
        doc_emb = self.embedding_model.encode(docs, batch_size=32)


        # 計(jì)算余弦相似度
        similarities = util.cos_sim(query_emb, doc_emb).diag().cpu().numpy()


        # 篩選相似度>閾值的樣本
        high_quality_indices = np.where(similarities > threshold)[0]
        return [pairs[i] for i in high_quality_indices]

from datasets import Dataset
import pandas as pd




def mine_hard_negatives(dataset_path, output_path):
    """挖掘難負(fù)樣本"""
    # 加載數(shù)據(jù)集
    data = pd.read_json(dataset_path)
    dataset = Dataset.from_pandas(data)


    # 加載Qwen3嵌入模型
    model = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")


    # 挖掘難負(fù)樣本
    hard_negatives = util.mine_hard_negatives(
        dataset,
        model,
        anchor_column_name="query",
        positive_column_name="document",
        num_negatives=5,
        range_min=20,
        range_max=50,
        max_score=0.8,
        absolute_margin=0.1,
        sampling_strategy="top",
        batch_size=64
    )


    # 保存結(jié)果
    hard_negatives_dataset = Dataset.from_dict(hard_negatives)
    hard_negatives_dataset.to_json(output_path, orient="records", lines=True)


# 執(zhí)行難負(fù)樣本挖掘
mine_hard_negatives(
    dataset_path="agriculture_data.json",
    output_path="agriculture_hard_negatives.json"
)

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from sentence_transformers import SentenceTransformer, util
from datasets import Dataset, load_dataset
import numpy as np
import pandas as pd
import os




class Qwen3DataGenerator:
    def __init__(self, model_path="Qwen/Qwen3-32B"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_path, 
            torch_dtype=torch.bfloat16,
            device_map="auto"
        )
        self.persona_hub = ["farmer", "researcher", "engineer", "doctor", "teacher"]


    def generate_config(self, document, language="English"):
        prompt = f"""
Given a document, generate a JSON configuration with Character, Question_Type, Difficulty.
Document: {document}
Language: {language}
Character candidates: {self.persona_hub}
Question_Type options: keywords, acquire_knowledge, summary, yes_or_no, background
Difficulty options: high_school, university, phd
Output JSON:
"""
        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device)
        outputs = self.model.generate(
            **inputs,
            max_new_tokens=100,
            temperature=0.7,
            top_p=0.9
        )
        config_json = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        return eval(config_json)




class QueryGenerator:
    def __init__(self, embedding_model="Qwen/Qwen3-Embedding-0.6B"):
        self.embedding_model = SentenceTransformer(embedding_model)


    def generate_query(self, config, document):
        role = config["Character"]
        q_type = config["Question_Type"]
        difficulty = config["Difficulty"]
        lang = config.get("Language", "Chinese")
        return f"示例查詢: 作為{role}，{q_type}類型的{difficulty}難度問題：{document[:20]}..."


    def filter_high_quality_pairs(self, pairs, threshold=0.7):
        queries = [p["query"] for p in pairs]
        docs = [p["document"] for p in pairs]
        query_emb = self.embedding_model.encode(queries, batch_size=32)
        doc_emb = self.embedding_model.encode(docs, batch_size=32)
        similarities = util.cos_sim(query_emb, doc_emb).diag().cpu().numpy()
        high_quality_indices = np.where(similarities > threshold)[0]
        return [pairs[i] for i in high_quality_indices]




def mine_hard_negatives(dataset_path, output_path):
    data = pd.read_json(dataset_path)
    dataset = Dataset.from_pandas(data)
    model = SentenceTransformer("Qwen/Qwen3-Embedding-0.6B")
    hard_negatives = util.mine_hard_negatives(
        dataset,
        model,
        anchor_column_name="query",
        positive_column_name="document",
        num_negatives=5,
        range_min=20,
        range_max=50,
        max_score=0.8,
        absolute_margin=0.1,
        sampling_strategy="top",
        batch_size=64
    )
    hard_negatives_dataset = Dataset.from_dict(hard_negatives)
    hard_negatives_dataset.to_json(output_path, orient="records", lines=True)




def generate_embedding_data(documents, output_path):
    generator = Qwen3DataGenerator()
    query_gen = QueryGenerator()
    data_pairs = []


    for doc in documents:
        config = generator.generate_config(doc)
        query = query_gen.generate_query(config, doc)
        data_pairs.append({
            "query": query,
            "document": doc,
            "config": config
        })


    high_quality_pairs = query_gen.filter_high_quality_pairs(data_pairs)
    Dataset.from_list(high_quality_pairs).to_json(output_path, orient="records", lines=True)
    return high_quality_pairs




def run_swift_sft():
    os.system("""
swift sft \
--model /path/to/Qwen3-Embedding-0.6B \
--task_type embedding \
--model_type qwen3_emb \
--train_type lora \
--dataset /path/to/agriculture_data.json \
--split_dataset_ratio 0.05 \
--eval_strategy steps \
--output_dir output/qwen3-agriculture \
--eval_steps 100 \
--num_train_epochs 1 \
--save_steps 100 \
--per_device_train_batch_size 4 \
--per_device_eval_batch_size 4 \
--gradient_accumulation_steps 4 \
--learning_rate 6e-6 \
--loss_type infonce \
--label_names labels \
--dataloader_drop_last true
""")




def evaluate_model(model_path, test_data_path):
    model = SentenceTransformer(model_path)
    test_data = load_dataset("json", data_files=test_data_path)["train"]
    queries = test_data["query"]
    docs = test_data["document"]


    query_emb = model.encode(queries, batch_size=32)
    doc_emb = model.encode(docs, batch_size=32)


    similarities = util.cos_sim(query_emb, doc_emb)
    # 計(jì)算NDCG等評(píng)估指標(biāo)
    return similarities




if __name__ == "__main__":
    # 示例文檔數(shù)據(jù)
    sample_documents = [
        "春季養(yǎng)鴨時(shí)，應(yīng)適當(dāng)增加能量飼料的比例以應(yīng)對(duì)低溫天氣",
        "發(fā)財(cái)樹春季養(yǎng)護(hù)需要保證充足光照",
        "食品包裝中常用的阻隔性材料主要用于保質(zhì)保鮮"
    ]


    # 1. 生成數(shù)據(jù)配置與查詢
    generated_data = generate_embedding_data(sample_documents, "generated_data.json")


    # 2. 挖掘難負(fù)樣本
    mine_hard_negatives("generated_data.json", "hard_negatives.json")


    # 3. 執(zhí)行LoRA微調(diào)
    run_swift_sft()


    # 4. 模型評(píng)估
    evaluation_results = evaluate_model("output/qwen3-agriculture", "hard_negatives.json")
    print("模型評(píng)估相似度矩陣:", evaluation_results.shape)

筆者能力有限，歡迎批評(píng)指正或者在留言區(qū)討論