大模型自動化標注：從流程到實戰

發布于 2025-11-14 00:11

瀏覽

0收藏

在 AI 模型開發中，數據標注始終是繞不開的 “痛點”—— 人工標注耗時耗力、成本高昂，還容易因主觀差異導致標注不一致。而隨著大模型能力的成熟，自動化標注正在成為破局關鍵：它能將原本一周的標注工作量壓縮到 1 天，還能保持穩定的標注標準，甚至在專業領域省去人工培訓成本。今天，我們就從核心流程、技術選型、實戰代碼到行業工具，全面拆解大模型自動化標注的落地方法，幫你快速上手這一效率利器。

一、先搞懂：大模型自動化標注的核心邏輯

相比傳統人工標注，大模型自動化標注的核心是 “模型預標注 + 多層復核” 的閉環流程。它不是完全替代人工，而是通過大模型的語義理解能力完成初步標注，再通過復核環節控制質量，最終實現 “效率提升 + 成本降低” 的雙重目標。

1.1 為什么選大模型做標注？3 個核心優勢

效率碾壓人工：對長文本（如客服對話）、高混淆度類目（如相似產品分類），大模型標注速度是人工的數倍倍，還能通過并發進一步提速；
標準絕對穩定：人工標注會因疲勞、理解偏差導致標準漂移，而大模型對每條數據的判斷邏輯一致，標注一致性相比人工標注更高；
專業門檻降低：醫療、金融等領域的標注需專業知識，人工需培訓一定的周期，而大模型只需在 Prompt 中嵌入專業規則，即可直接輸出標注結果。

1.2 關鍵前提：先梳理好你的 “標注規則”

在啟動自動化標注前，必須先明確 “標注目標”，否則大模型會因理解偏差輸出無效結果。以文本分類任務為例，你需要準備 3 類信息：

清晰的類目體系：比如將用戶搜索意圖分為 “導航類、信息類、交易類、娛樂類、教育類”，避免模糊類目（如 “其他” 需明確包含場景）；
類目解釋 + 示例：給每個類目配 1-2 個典型案例，比如 “交易類：包含‘購買手機’‘預訂酒店’等有明確消費意圖的 Query”；
標注邊界說明：明確模糊場景的判定規則，比如 “‘查詢手機價格’歸為信息類，‘對比手機價格’歸為交易類前置需求，也歸為信息類”。

這些信息會作為 Prompt 的核心，直接影響大模型標注的準確性。

二、實戰流程：從預標注到模型訓練的 6 步閉環

以文本分類任務為例，完整的大模型自動化標注流程分為 5 個關鍵步驟，每個環節都有需要注意的細節和經驗技巧。

2.1 預標注 —— 用大模型生成初步標簽

預標注是自動化標注的核心環節，本質是 “讓大模型按照你的規則給數據打標簽”。這里的關鍵是模型選型和Prompt 設計。

2.1.1 模型選型：優先選 “大參數 + Instruct 版本”

參數規模：建議選擇 14B 以上模型，效果更穩定。如果預算有限，MoE（混合專家模型）是性價比之選 —— 比如 “Qwen3-235B-A22B-Instruct-2507”，推理速度比同參數非 MoE 模型快 30%，效果差距卻很小；
模型類型：優先用 Instruct 模型（而非 Thinking 模型），原因是 Instruct 模型響應更快，且召回率更高（適合覆蓋更多邊緣案例）；如果追求極致準確率，可搭配 Thinking 模型做二次驗證。

2.1.2 Prompt 設計：結構化指令比模糊描述更有效

壞的 Prompt：“給這個 Query 分類”；

好的 Prompt：“你是專業的搜索意圖分類員，請將以下 Query 分類到【導航類、信息類、交易類、娛樂類、教育類、其他】中的一個。分類規則：1. 導航類：包含明確網址或 APP 名稱，如‘打開百度’；2. 交易類：有購買 / 預訂意圖，如‘買華為 Mate60’... 請只返回類目名稱，不額外解釋。Query：{query}”

2.1.3 代碼實戰：批量預標注腳本

import json
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from openai import OpenAI, APIError, RateLimitError


# 初始化客戶端（支持多API密鑰輪詢，避免限流）
API_KEYS = os.getenv("API_KEYS", "").split(",")  # 多個密鑰用逗號分隔
BASE_URL = os.getenv("BASE_URL", "https://api.openai.com/v1")
clients = [OpenAI(api_key=key.strip(), base_url=BASE_URL) for key in API_KEYS if key.strip()]


# 分類配置（抽離為獨立字典，方便修改）
CLASSIFICATION_CONFIG = {
    "categories": ["導航類", "信息類", "交易類", "娛樂類", "教育類", "其他"],
    "rules": """1. 導航類：包含具體網址、APP名稱或直接訪問需求（如"打開抖音"、"www.baidu.com"）；
2. 信息類：查詢事實、知識、狀態等無明確行動意圖（如"北京今日天氣"、"Python是什么"）；
3. 交易類：含購買、預訂、下單等消費意圖（如"訂上海到北京的機票"、"買華為P60"）；
4. 娛樂類：追求休閑娛樂體驗（如"看搞笑短視頻"、"聽熱門歌曲"）；
5. 教育類：以學習、技能提升為目的（如"Java入門教程"、"考研英語復習方法"）；
6. 其他：不符合以上類別的所有Query"""
}


def classify_single_query(query, client):
    """單條Query分類（增加重試機制和日志記錄）"""
    system_prompt = f"""你是嚴格執行規則的搜索意圖分類員，需按以下規則和類別完成分類：
類別列表：{','.join(CLASSIFICATION_CONFIG['categories'])}
分類規則：{CLASSIFICATION_CONFIG['rules']}
輸出要求：僅返回類目名稱，不添加任何解釋、標點或額外內容，嚴格匹配類別列表中的選項"""


    max_retries = 3
    retry_delay = 2  # 重試間隔（秒）


    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="qwen-max",
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": f"待分類Query：{query}"}
                ],
                temperature=0.05,  # 更低溫度，進一步保證穩定性
                max_tokens=30,
                timeout=10  # 超時控制
            )
            result = response.choices[0].message.content.strip()
            # 嚴格校驗結果是否在類目列表中
            return result if result in CLASSIFICATION_CONFIG["categories"] else "其他"
        except RateLimitError:
            if attempt < max_retries - 1:
                time.sleep(retry_delay * (attempt + 1))  # 指數退避
                continue
            print(f"API限流，Query：{query} 分類失敗")
            return "其他"
        except (APIError, TimeoutError) as e:
            print(f"分類異常（{type(e).__name__}）：{e}，Query：{query}")
            return "其他"


def batch_pre_annotation(input_path, output_path, max_workers=5):
    """批量預標注（支持并發處理，提升效率）"""
    # 讀取輸入數據（支持JSON和TXT格式）
    if not os.path.exists(input_path):
        raise FileNotFoundError(f"輸入文件不存在：{input_path}")


    queries = []
    if input_path.endswith(".txt"):
        with open(input_path, "r", encoding="utf-8") as f:
            queries = [line.strip() for line in f if line.strip()]
    elif input_path.endswith(".json"):
        with open(input_path, "r", encoding="utf-8") as f:
            data = json.load(f)
            queries = [item["query"] for item in data if "query" in item]


    print(f"共讀取 {len(queries)} 條待標注數據")
    results = []


    # 并發處理（按API密鑰數量限制并發數）
    with ThreadPoolExecutor(max_workers=min(max_workers, len(clients))) as executor:
        # 提交任務，綁定不同客戶端避免限流
        future_to_query = {
            executor.submit(classify_single_query, query, clients[i % len(clients)]): query
            for i, query in enumerate(queries)
        }


        for idx, future in enumerate(as_completed(future_to_query), 1):
            query = future_to_query[future]
            try:
                category = future.result()
                results.append({
                    "query": query,
                    "pre_annotated_category": category,
                    "annotation_time": time.strftime("%Y-%m-%d %H:%M:%S"),
                    "review_status": "pending",
                    "model_version": "qwen-max"
                })
                if idx % 50 == 0:
                    print(f"已完成 {idx}/{len(queries)} 條數據標注")
            except Exception as e:
                print(f"Query：{query} 處理失敗：{e}")
                results.append({
                    "query": query,
                    "pre_annotated_category": "其他",
                    "annotation_time": time.strftime("%Y-%m-%d %H:%M:%S"),
                    "review_status": "error",
                    "error_msg": str(e)
                })


    # 保存結果（創建父目錄，避免路徑不存在）
    os.makedirs(os.path.dirname(output_path), exist_ok=True)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)


    # 輸出統計信息
    category_count = {}
    for item in results:
        cat = item["pre_annotated_category"]
        category_count[cat] = category_count.get(cat, 0) + 1
    print(f"\n預標注完成！結果保存至：{output_path}")
    print("類目分布統計：")
    for cat, count in category_count.items():
        print(f"- {cat}：{count} 條（{count/len(results)*100:.1f}%）")


if __name__ == "__main__":
    # 支持命令行參數傳入，更靈活
    import argparse
    parser = argparse.ArgumentParser(description="大模型批量預標注工具")
    parser.add_argument("--input", required=True, help="輸入文件路徑（TXT/JSON）")
    parser.add_argument("--output", required=True, help="輸出文件路徑（JSON）")
    parser.add_argument("--workers", type=int, default=5, help="并發數")
    args = parser.parse_args()


    batch_pre_annotation(args.input, args.output, args.workers)

2.2 雙重復核 —— 先讓模型 “自查”，再人工 “把關”

預標注結果不能直接用！必須經過 “大模型復核 + 人工復核”，否則可能因 Prompt 歧義、模型理解偏差導致成片錯誤。

2.2.1 大模型復核 —— 讓另一個模型做 “審核員”

原理：用不同模型（或同模型不同 Prompt）判斷預標注結果是否正確，相當于 “交叉驗證”。這樣做能快速篩選出明顯錯誤，減少人工工作量。

復核 Prompt 示例：“你是數據標注審核員，請判斷以下 Query 的預標注結果是否正確。Query：{query}，預標注類別：{pre_annotated}。判斷標準與預標注規則一致。如果正確，回復‘正確’；如果錯誤，回復‘錯誤’并給出正確類別。”

2.2.2 人工復核 —— 必須抽樣，不能省略

即使大模型復核通過，也需要人工抽樣檢查。原因是：大模型可能存在 “系統性偏差”（比如把所有 “查詢價格” 都歸為交易類），這種錯誤只有人工能發現。

操作建議：

抽樣數量：至少 100-200 條，覆蓋所有類目；
質量標準：文本分類任務需達到 90% 以上準確率，否則需優化 Prompt 或更換模型；
重點關注：檢查是否有 “成片錯誤”（如某類目全部標錯），這類問題通常是 Prompt 規則不清晰導致的。

2.2.3 代碼實戰：大模型復核腳本

import json
import os
import time
from openai import OpenAI


# 初始化復核專用客戶端（建議使用與預標注不同的模型）
REVIEW_CLIENT = OpenAI(
    api_key=os.getenv("REVIEW_API_KEY", os.getenv("API_KEY")),
    base_url=os.getenv("BASE_URL", "https://api.openai.com/v1")
)


def load_classification_rules():
    """加載分類規則（與預標注保持一致）"""
    return CLASSIFICATION_CONFIG["rules"]


def review_single_item(item):
    """單條數據復核（優化結果解析邏輯，增加置信度判斷）"""
    query = item["query"]
    pre_cat = item["pre_annotated_category"]
    rules = load_classification_rules()


    review_prompt = f"""你是專業的數據標注審核專家，嚴格按照以下規則審核分類結果：
{rules}
待審核內容：
- 原始Query：{query}
- 預標注類別：{pre_cat}
審核要求：
1. 先判斷預標注結果是否完全符合規則（正確/錯誤）；
2. 若錯誤，必須從【{','.join(CLASSIFICATION_CONFIG['categories'])}】中選擇正確類別；
3. 補充說明判斷依據（不超過50字）；
4. 按以下格式輸出，不添加任何額外內容：
判斷結果：{正確/錯誤}
正確類別：{類目名稱}
判斷依據：{簡要說明}"""


    try:
        response = REVIEW_CLIENT.chat.completions.create(
            model="qwen-plus",  # 復核模型與預標注模型區分
            messages=[{"role": "user", "content": review_prompt}],
            temperature=0.05,
            max_tokens=150,
            timeout=15
        )


        review_content = response.choices[0].message.content.strip()
        lines = [line.strip() for line in review_content.split("\n") if line.strip()]


        # 解析結果（魯棒性更強的解析邏輯）
        result = {"判斷結果": "unclear", "正確類別": pre_cat, "判斷依據": "解析失敗"}
        for line in lines:
            if line.startswith("判斷結果："):
                result["判斷結果"] = line.split("：")[-1].strip()
            elif line.startswith("正確類別："):
                result["正確類別"] = line.split("：")[-1].strip()
                # 校驗正確類別是否合法
                if result["正確類別"] not in CLASSIFICATION_CONFIG["categories"]:
                    result["正確類別"] = pre_cat
            elif line.startswith("判斷依據："):
                result["判斷依據"] = line.split("：")[-1].strip()[:50]  # 限制長度


        # 更新item狀態
        item["review_result"] = result
        if result["判斷結果"] == "正確":
            item["review_status"] = "passed"
            item["final_category"] = pre_cat
        elif result["判斷結果"] == "錯誤":
            item["review_status"] = "rejected"
            item["final_category"] = result["正確類別"]
        else:
            item["review_status"] = "need_manual"
            item["final_category"] = pre_cat


        item["review_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
        item["review_model"] = "qwen-plus"
        return item


    except Exception as e:
        print(f"復核失敗：{type(e).__name__} - {e}，Query：{query}")
        item["review_status"] = "review_error"
        item["review_error"] = str(e)
        item["final_category"] = pre_cat
        return item


def batch_review_annotations(input_path, output_path, manual_sample_size=150):
    """批量復核（增加人工抽樣推薦功能）"""
    # 讀取預標注數據
    with open(input_path, "r", encoding="utf-8") as f:
        pre_annotated_data = json.load(f)


    print(f"開始復核 {len(pre_annotated_data)} 條數據...")
    reviewed_data = []
    for item in pre_annotated_data:
        reviewed_item = review_single_item(item)
        reviewed_data.append(reviewed_item)


    # 推薦人工復核樣本（覆蓋不同狀態和類目）
    manual_review_samples = []
    # 1. 先選狀態異常的樣本（錯誤、解析失敗）
    abnormal_samples = [item for item in reviewed_data if item["review_status"] in ["rejected", "need_manual", "review_error"]]
    manual_review_samples.extend(abnormal_samples[:int(manual_sample_size*0.6)])


    # 2. 再從正常樣本中按類目抽樣
    normal_samples = [item for item in reviewed_data if item["review_status"] == "passed"]
    category_samples = {}
    for item in normal_samples:
        cat = item["final_category"]
        if cat not in category_samples:
            category_samples[cat] = []
        category_samples[cat].append(item)


    # 每個類目至少抽2條
    remaining_slots = manual_sample_size - len(manual_review_samples)
    for cat, samples in category_samples.items():
        if remaining_slots <= 0:
            break
        sample_count = min(len(samples), max(2, remaining_slots // len(category_samples)))
        manual_review_samples.extend(samples[:sample_count])
        remaining_slots -= sample_count


    # 補充剩余名額（隨機抽取）
    if remaining_slots > 0:
        extra_samples = [item for item in normal_samples if item not in manual_review_samples]
        manual_review_samples.extend(extra_samples[:remaining_slots])


    # 保存結果
    output_data = {
        "review_summary": {
            "total_count": len(reviewed_data),
            "passed_count": len([x for x in reviewed_data if x["review_status"] == "passed"]),
            "rejected_count": len([x for x in reviewed_data if x["review_status"] == "rejected"]),
            "need_manual_count": len(manual_review_samples),
            "review_date": time.strftime("%Y-%m-%d %H:%M:%S")
        },
        "all_reviewed_data": reviewed_data,
        "manual_review_recommendations": manual_review_samples
    }


    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(output_data, f, ensure_ascii=False, indent=2)


    print(f"復核完成！結果保存至：{output_path}")
    print(f"復核摘要：")
    print(f"- 總數據量：{output_data['review_summary']['total_count']} 條")
    print(f"- 自動通過：{output_data['review_summary']['passed_count']} 條")
    print(f"- 自動駁回：{output_data['review_summary']['rejected_count']} 條")
    print(f"- 推薦人工復核：{output_data['review_summary']['need_manual_count']} 條")




if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="大模型標注復核工具")
    parser.add_argument("--input", required=True, help="預標注結果文件路徑（JSON）")
    parser.add_argument("--output", required=True, help="復核結果輸出路徑（JSON）")
    parser.add_argument("--manual-sample", type=int, default=150, help="推薦人工復核樣本數")
    args = parser.parse_args()


    batch_review_annotations(args.input, args.output, args.manual_sample)

2.3 數據集整理 —— 適配模型訓練格式

復核通過后，需要將標注數據整理成模型訓練所需的格式（以 LLaMA-Factory 為例，這是目前主流的大模型微調框架）。

核心操作：

過濾無效數據：剔除 “需人工復核”“標注錯誤” 的數據；
劃分訓練 / 驗證 / 測試集：通常按 8:1:1 劃分，確保類別分布均勻（用??stratify??參數）；
格式轉換：按框架要求生成 “instruction-input-output” 結構。

2.3.1 代碼實戰：數據集整理腳本

import json
import os
import random
from sklearn.model_selection import train_test_split
from collections import Counter




def filter_valid_data(reviewed_data_path):
    """過濾有效數據（嚴格篩選+數據清洗）"""
    with open(reviewed_data_path, "r", encoding="utf-8") as f:
        data = json.load(f)


    # 從復核結果中提取所有數據
    all_data = data["all_reviewed_data"]


    # 篩選條件：狀態正常 + 類別合法
    valid_data = []
    for item in all_data:
        # 跳過需人工處理或錯誤的數據
        if item["review_status"] in ["need_manual", "review_error", "error"]:
            continue
        # 確保最終類別合法
        if item["final_category"] not in CLASSIFICATION_CONFIG["categories"]:
            continue
        # 去除空值或異常長度的Query
        if not item["query"] or len(item["query"]) > 500:  # 限制最大長度
            continue
        valid_data.append({
            "query": item["query"].strip(),
            "label": item["final_category"],
            "annotation_source": item.get("model_version", "unknown"),
            "review_source": item.get("review_model", "unknown")
        })


    # 檢查類別分布（避免類別失衡）
    label_counts = Counter([x["label"] for x in valid_data])
    print("有效數據類別分布：")
    for label, count in label_counts.most_common():
        print(f"- {label}：{count} 條（{count/len(valid_data)*100:.1f}%）")


    # 過濾樣本數過少的類別（少于10條則剔除）
    min_sample_threshold = 10
    valid_labels = [label for label, count in label_counts.items() if count >= min_sample_threshold]
    valid_data = [x for x in valid_data if x["label"] in valid_labels]


    print(f"過濾后有效數據總量：{len(valid_data)} 條")
    return valid_data


def convert_to_training_format(valid_data, output_dir, train_ratio=0.8, val_ratio=0.1, test_ratio=0.1):
    """轉換為訓練格式（支持自定義劃分比例，增加數據增強）"""
    # 驗證比例合法性
    assert abs(train_ratio + val_ratio + test_ratio - 1.0) < 0.01, "比例之和必須為1"


    # 按類別分層劃分，確保分布均勻
    labels = [x["label"] for x in valid_data]
    train_data, temp_data, train_labels, temp_labels = train_test_split(
        valid_data, labels, test_size=1-train_ratio, random_state=42, stratify=labels
    )
    val_data, test_data = train_test_split(
        temp_data, test_size=test_ratio/(val_ratio+test_ratio), random_state=42, stratify=temp_labels
    )


    # 數據增強（簡單同義詞替換，可選）
    def simple_data_augmentation(text):
        """簡單數據增強：替換常見同義詞"""
        synonyms = {
            "查詢": ["查找", "搜索", "了解"],
            "購買": ["選購", "訂購", "入手"],
            "打開": ["啟動", "訪問", "進入"],
            "學習": ["研習", "掌握", "進修"]
        }
        words = text.split()
        augmented_words = []
        for word in words:
            if word in synonyms and random.random() < 0.3:  # 30%概率替換
                augmented_words.append(random.choice(synonyms[word]))
            else:
                augmented_words.append(word)
        return " ".join(augmented_words)


    # 對訓練集進行數據增強（避免驗證集/測試集污染）
    augmented_train_data = []
    for item in train_data:
        # 保留原始數據
        augmented_train_data.append(item)
        # 生成增強數據（僅對樣本數較少的類別）
        label_count = Counter([x["label"] for x in train_data])[item["label"]]
        if label_count < 50:  # 樣本數少于50的類別，增強1倍
            augmented_item = {
                "query": simple_data_augmentation(item["query"]),
                "label": item["label"],
                "annotation_source": f"{item['annotation_source']}_augmented",
                "review_source": item["review_source"]
            }
            augmented_train_data.append(augmented_item)


    # 轉換為LLaMA-Factory格式
    def format_for_llama_factory(data_list, task_desc="搜索意圖分類"):
        """轉換為LLaMA-Factory所需格式"""
        formatted = []
        for item in data_list:
            formatted.append({
                "instruction": f"請完成{task_desc}任務，將用戶Query分類到以下類別之一：{','.join(CLASSIFICATION_CONFIG['categories'])}",
                "input": item["query"],
                "output": item["label"],
                "system": "你是一個專業的文本分類助手，嚴格按照給定類別進行分類，僅返回類別名稱。"
            })
        return formatted


    # 格式化數據
    train_formatted = format_for_llama_factory(augmented_train_data)
    val_formatted = format_for_llama_factory(val_data)
    test_formatted = format_for_llama_factory(test_data)


    # 保存數據（創建目錄，支持覆蓋提示）
    os.makedirs(output_dir, exist_ok=True)
    for name, data in [("train", train_formatted), ("val", val_formatted), ("test", test_formatted)]:
        path = os.path.join(output_dir, f"{name}.json")
        # 覆蓋提示
        if os.path.exists(path):
            print(f"警告：{path} 已存在，將被覆蓋")
        with open(path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)


    # 輸出統計信息
    print(f"\n數據集生成完成！輸出目錄：{output_dir}")
    print(f"- 訓練集：{len(train_formatted)} 條（含增強數據）")
    print(f"- 驗證集：{len(val_formatted)} 條")
    print(f"- 測試集：{len(test_formatted)} 條")


    # 生成數據說明文檔
    data_info = {
        "生成時間": time.strftime("%Y-%m-%d %H:%M:%S"),
        "數據來源": "大模型自動化標注+人工復核",
        "原始數據量": len(all_data),
        "有效數據量": len(valid_data),
        "增強后訓練集量": len(train_formatted),
        "類別列表": CLASSIFICATION_CONFIG["categories"],
        "劃分比例": f"訓練集{train_ratio*100:.0f}%、驗證集{val_ratio*100:.0f}%、測試集{test_ratio*100:.0f}%",
        "使用說明": "適配LLaMA-Factory框架，支持LoRA微調"
    }
    with open(os.path.join(output_dir, "data_info.json"), "w", encoding="utf-8") as f:
        json.dump(data_info, f, ensure_ascii=False, indent=2)


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="標注數據轉訓練格式工具")
    parser.add_argument("--reviewed-data", required=True, help="復核結果文件路徑（JSON）")
    parser.add_argument("--output-dir", required=True, help="訓練集輸出目錄")
    parser.add_argument("--augment", action="store_true", help="是否啟用數據增強")
    args = parser.parse_args()


    valid_data = filter_valid_data(args.reviewed_data)
    convert_to_training_format(valid_data, args.output_dir, augment=args.augment)

2.4 模型訓練 —— 用標注數據微調專屬模型

整理好數據后，就可以用 LLaMA-Factory 進行微調。這里推薦用 LoRA（低秩適應）方法，優點是訓練成本低、速度快，且不需要全量參數更新。

2.4.1 配置文件（config.yaml）示例

model_name_or_path: Qwen/Qwen3-4B-Instruct-2507  # 基座模型
dataset_dir: ./train_data  # 訓練集目錄
dataset: train.json,val.json  # 訓練/驗證集
template: qwen  # 模型模板，需與基座模型匹配
finetuning_type: lora  # 微調方式：LoRA
lora_target: q_proj,v_proj,o_proj  # 擴展目標層，提升效果
lora_rank: 8  # LoRA秩，平衡效果與成本
lora_alpha: 16  # LoRA alpha值
lora_dropout: 0.05  # Dropout比例，防止過擬合
output_dir: ./finetuned_model  # 模型輸出目錄
per_device_train_batch_size: 8  # 單設備批次大小（根據GPU顯存調整）
per_device_eval_batch_size: 16  # 驗證批次大小
gradient_accumulation_steps: 2  # 梯度累積步數
lr_scheduler_type: cosine_with_restarts  # 學習率調度器，優化收斂
learning_rate: 2.0e-4  # 學習率（根據模型大小調整）
num_train_epochs: 5  # 訓練輪次
max_steps: -1  # 按輪次訓練，不限制步數
warmup_ratio: 0.1  # 熱身比例
fp16: true  # 混合精度訓練，節省顯存
bf16: false  # 根據GPU支持情況選擇
logging_steps: 5  # 日志打印步數
eval_steps: 50  # 驗證步數
save_steps: 100  # 模型保存步數
save_total_limit: 3  # 保留最新3個模型，節省空間
load_best_model_at_end: true  # 訓練結束后加載最優模型
metric_for_best_model: accuracy  # 以準確率為最優指標
greater_is_better: true  # 指標越大越好
seed: 42  # 隨機種子，保證可復現
disable_tqdm: false  # 顯示進度條
report_to: tensorboard  # 支持TensorBoard可視化
resume_from_checkpoint: auto  # 自動恢復訓練

2.4.2 啟動訓練命令

簡單命令

# 激活虛擬環境
conda activate llama_factory
# 執行訓練
python src/train_bash.py --config config.yaml

或者用啟動腳本

#!/bin/bash
# 訓練腳本：包含環境檢查和日志保存
set -e


# 檢查虛擬環境
if [ -z "$CONDA_DEFAULT_ENV" ]; then
    echo "請激活conda虛擬環境！"
    exit 1
fi


# 檢查LLaMA-Factory環境
if ! command -v python src/train_bash.py &> /dev/null; then
    echo "請在LLaMA-Factory項目根目錄執行！"
    exit 1
fi


# 創建日志目錄
LOG_DIR="./train_logs"
mkdir -p $LOG_DIR
LOG_FILE="$LOG_DIR/train_$(date +%Y%m%d_%H%M%S).log"


# 執行訓練
echo "開始訓練，日志保存至：$LOG_FILE"
python src/train_bash.py \
    --config config.yaml \
    2>&1 | tee $LOG_FILE


echo "訓練完成！模型保存至：./finetuned_model"
echo "日志文件：$LOG_FILE"

2.5 LoRA 權重合并 —— 生成可直接部署的完整模型

LoRA 微調后，模型權重分為 “基座模型權重” 和 “LoRA 增量權重”，無法直接用于推理部署。需要將兩者合并，生成完整的模型文件，才能用 vLLM 等框架高效推理。

2.5.1 核心說明

合并需確保基座模型與 LoRA 權重的兼容性（同一模型版本）；
支持 FP16/FP32 精度選擇，FP16 更節省顯存和存儲；
合并后會生成標準的 Hugging Face 模型格式，可直接用于推理和部署。

2.5.2 代碼實戰：LoRA 權重合并腳本（完整可運行）

import os
import argparse
import torch
from peft import PeftModel, PeftConfig
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig




def merge_lora_weights(peft_model_path, output_dir, dtype="float16"):
    """
    合并LoRA權重與基座模型權重
    Args:
        peft_model_path: LoRA權重保存路徑（LLaMA-Factory輸出目錄）
        output_dir: 合并后模型保存目錄
        dtype: 模型精度（float16/float32/bfloat16）
    """
    # 驗證精度參數
    supported_dtypes = ["float16", "float32", "bfloat16"]
    assert dtype in supported_dtypes, f"支持的精度：{supported_dtypes}，輸入：{dtype}"


    # 加載LoRA配置，獲取基座模型信息
    print(f"加載LoRA配置：{peft_model_path}")
    peft_config = PeftConfig.from_pretrained(peft_model_path)
    base_model_name_or_path = peft_config.base_model_name_or_path
    print(f"基座模型：{base_model_name_or_path}")


    # 配置模型加載參數
    torch_dtype = getattr(torch, dtype)
    device_map = "auto"  # 自動分配設備（CPU/GPU）


    # 加載量化配置（如果基座模型是量化的）
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=False,  # 合并時禁用4bit量化，保證精度
        bnb_4bit_compute_dtype=torch_dtype,
        bnb_4bit_use_double_quant=False,
        bnb_4bit_quant_type="nf4"
    )


    try:
        # 加載基座模型（支持量化模型）
        print(f"加載基座模型（精度：{dtype}）...")
        base_model = AutoModelForCausalLM.from_pretrained(
            base_model_name_or_path,
            torch_dtype=torch_dtype,
            device_map=device_map,
            trust_remote_code=True,
            quantization_config=bnb_config if dtype == "float16" else None,
            low_cpu_mem_usage=True  # 低CPU內存占用模式
        )


        # 加載LoRA權重
        print("加載LoRA權重...")
        lora_model = PeftModel.from_pretrained(
            base_model,
            peft_model_path,
            torch_dtype=torch_dtype,
            device_map=device_map,
            trust_remote_code=True
        )


        # 合并權重
        print("開始合并LoRA權重與基座模型...")
        merged_model = lora_model.merge_and_unload()  # 合并并卸載LoRA適配器
        print("權重合并完成！")


        # 加載對應的Tokenizer
        print("加載Tokenizer...")
        tokenizer = AutoTokenizer.from_pretrained(
            base_model_name_or_path,
            trust_remote_code=True,
            padding_side="right"  # 推理時右填充，避免警告
        )
        # 添加終止符（如果不存在）
        if tokenizer.pad_token is None:
            tokenizer.pad_token = tokenizer.eos_token
            print(f"已設置pad_token為：{tokenizer.pad_token}")


        # 保存合并后的模型和Tokenizer
        os.makedirs(output_dir, exist_ok=True)
        print(f"保存合并后的模型至：{output_dir}")


        merged_model.save_pretrained(
            output_dir,
            torch_dtype=torch_dtype,
            safe_serialization=True  # 安全序列化，避免兼容性問題
        )
        tokenizer.save_pretrained(output_dir)


        # 生成模型配置說明
        config_info = {
            "base_model": base_model_name_or_path,
            "lora_model_path": peft_model_path,
            "merged_dtype": dtype,
            "save_time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "tokenizer_pad_token": tokenizer.pad_token,
            "model_size": f"{sum(p.numel() for p in merged_model.parameters()):,} 參數量"
        }


        with open(os.path.join(output_dir, "merged_config.json"), "w", encoding="utf-8") as f:
            json.dump(config_info, f, ensure_ascii=False, indent=2)


        print("="*50)
        print("權重合并總結：")
        print(f"- 基座模型：{base_model_name_or_path}")
        print(f"- LoRA權重路徑：{peft_model_path}")
        print(f"- 合并后精度：{dtype}")
        print(f"- 模型保存路徑：{output_dir}")
        print(f"- 參數量：{config_info['model_size']}")
        print("="*50)


    except Exception as e:
        print(f"權重合并失敗：{type(e).__name__} - {e}")
        raise e


if __name__ == "__main__":
    import json
    import time


    parser = argparse.ArgumentParser(description="LoRA權重與基座模型合并工具")
    parser.add_argument("--lora-path", required=True, help="LLaMA-Factory輸出的LoRA權重路徑")
    parser.add_argument("--output-dir", required=True, help="合并后模型保存目錄")
    parser.add_argument("--dtype", default="float16", choices=["float16", "float32", "bfloat16"], help="模型精度")
    args = parser.parse_args()


    merge_lora_weights(args.lora_path, args.output_dir, args.dtype)

2.5.3 合并腳本使用說明

安裝依賴（如果未安裝）：

pip install peft transformers accelerate bitsandbytes torch

執行合并命令：

python merge_lora_weights.py \
    --lora-path ./finetuned_model \  # LLaMA-Factory訓練輸出的LoRA路徑
    --output-dir ./merged_model \    # 合并后模型保存目錄
    --dtype float16                  # 精度選擇（根據GPU支持情況）

驗證合并結果：合并完成后，merged_model 目錄會包含以下文件，說明合并成功：

??config.json??：模型配置文件
??model-00001-of-xxxx.bin??：模型權重文件（多個分片）
??tokenizer.json???/??tokenizer_config.json??：Tokenizer 文件
??merged_config.json??：合并配置說明

2.6 批跑測試 —— 評估模型效果

訓練完成后，需要在測試集上評估模型準確率，確保標注數據的質量能支撐模型性能。這里推薦用 vLLM 進行推理，速度比原生 PyTorch 快 10 倍以上。

2.6.1 代碼實戰：模型測試腳本

from vllm import LLM, SamplingParams
import json
import os
import time
from sklearn.metrics import accuracy_score, classification_report
from collections import defaultdict


def load_test_data(test_data_path):
    """加載測試集數據"""
    with open(test_data_path, "r", encoding="utf-8") as f:
        test_data = json.load(f)
    # 提取input和output（適配LLaMA-Factory格式）
    return [(item["input"], item["output"]) for item in test_data]


def batch_predict(model_path, test_data, batch_size=32):
    """批量預測（優化批處理，提升推理速度）"""
    # 初始化vLLM引擎
    llm = LLM(
        model=model_path,
        tensor_parallel_size=1,
        gpu_memory_utilization=0.85,  # 合理利用GPU顯存
        max_num_batched_tokens=4096  # 批處理最大tokens數
    )


    # 推理參數
    sampling_params = SamplingParams(
        temperature=0.01,
        top_p=0.95,
        max_tokens=30,
        stop=["\n", "。"]  # 提前終止符，減少無效輸出
    )


    # 構建Prompt模板
    def build_prompt(query):
        return f"""你是專業的搜索意圖分類助手，需將Query分類到以下類別之一：{','.join(CLASSIFICATION_CONFIG['categories'])}
僅返回類別名稱，不添加任何額外內容。
Query：{query}
類別："""


    # 批量處理
    all_prompts = [build_prompt(query) for query, _ in test_data]
    all_preds = []
    total_batches = (len(all_prompts) + batch_size - 1) // batch_size


    print(f"開始推理，共 {len(all_prompts)} 條數據，分為 {total_batches} 批")
    start_time = time.time()


    for i in range(total_batches):
        start_idx = i * batch_size
        end_idx = min((i+1)*batch_size, len(all_prompts))
        batch_prompts = all_prompts[start_idx:end_idx]


        # 推理
        outputs = llm.generate(batch_prompts, sampling_params)
        batch_preds = [output.outputs[0].text.strip() for output in outputs]


        # 結果清洗（確保類別合法）
        cleaned_preds = []
        for pred in batch_preds:
            cleaned_preds.append(pred if pred in CLASSIFICATION_CONFIG["categories"] else "其他")


        all_preds.extend(cleaned_preds)


        # 打印進度
        if (i+1) % 5 == 0 or i == total_batches - 1:
            elapsed_time = time.time() - start_time
            speed = (i+1)*batch_size / elapsed_time
            print(f"完成 {i+1}/{total_batches} 批，速度：{speed:.2f} 條/秒")


    return all_preds


def evaluate_model_performance(model_path, test_data_path, output_report_path):
    """全面評估模型性能（準確率+分類報告+錯誤分析）"""
    # 加載數據
    test_data = load_test_data(test_data_path)
    true_labels = [label for _, label in test_data]
    queries = [query for query, _ in test_data]


    # 批量預測
    pred_labels = batch_predict(model_path, test_data)


    # 計算整體準確率
    accuracy = accuracy_score(true_labels, pred_labels)


    # 詳細分類報告（精確率、召回率、F1值）
    class_report = classification_report(
        true_labels, pred_labels,
        target_names=CLASSIFICATION_CONFIG["categories"],
        output_dict=True,
        zero_division=0
    )


    # 錯誤分析
    error_analysis = defaultdict(list)
    for query, true_label, pred_label in zip(queries, true_labels, pred_labels):
        if true_label != pred_label:
            error_analysis[f"{true_label}→{pred_label}"].append(query)


    # 生成評估報告
    evaluation_report = {
        "evaluation_time": time.strftime("%Y-%m-%d %H:%M:%S"),
        "model_path": model_path,
        "test_data_count": len(test_data),
        "overall_accuracy": round(accuracy, 4),
        "classification_report": class_report,
        "error_analysis": {
            "error_types": list(error_analysis.keys()),
            "total_error_count": sum(len(queries) for queries in error_analysis.values()),
            "error_details": error_analysis
        },
        "top_error_types": sorted(error_analysis.items(), key=lambda x: len(x[1]), reverse=True)[:5]
    }


    # 保存報告
    with open(output_report_path, "w", encoding="utf-8") as f:
        json.dump(evaluation_report, f, ensure_ascii=False, indent=2)


    # 打印關鍵結果
    print("\n" + "="*50)
    print("模型評估報告")
    print("="*50)
    print(f"整體準確率：{accuracy:.4f}")
    print("\n各類別F1值：")
    for cat in CLASSIFICATION_CONFIG["categories"]:
        if cat in class_report:
            f1_score = class_report[cat]["f1-score"]
            print(f"- {cat}：{f1_score:.4f}")
    print(f"\n錯誤分析：共 {evaluation_report['error_analysis']['total_error_count']} 條錯誤")
    print("Top 3錯誤類型：")
    for error_type, examples in evaluation_report["top_error_types"][:3]:
        print(f"- {error_type}：{len(examples)} 條（示例：{examples[0][:30]}...）")
    print(f"\n評估報告已保存至：{output_report_path}")




if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="模型評估工具")
    parser.add_argument("--model-path", required=True, help="微調后模型路徑")
    parser.add_argument("--test-data", required=True, help="測試集文件路徑（JSON）")
    parser.add_argument("--output-report", required=True, help="評估報告輸出路徑（JSON）")
    parser.add_argument("--batch-size", type=int, default=32, help="推理批大小")
    args = parser.parse_args()


    evaluate_model_performance(args.model_path, args.test_data, args.output_report)

三、行業工具推薦：不止于代碼，這些工具更高效

如果你不想從零寫代碼，這些成熟的自動化標注工具能幫你快速落地：

3.1 Label Studio：開源全能型工具

核心優勢：支持文本、圖像、音頻、視頻等多模態數據，可自定義標注界面，還能集成預訓練模型（如 BERT、YOLO）做輔助標注；
適用場景：中小型團隊，需要靈活適配不同標注任務；
亮點功能：內置主動學習模塊，能自動篩選高價值樣本讓人工標注，進一步提升效率。

3.2 整數智能 “啟真”：企業級國產化平臺

核心優勢：基于華為昇騰算力，搭載 DeepSeek 大模型，支持醫療、金融等專業領域的定制化標注，標注效率提升 5-10 倍；
適用場景：對數據安全要求高的企業（如政務、醫療），需要私有化部署；
亮點功能：內置數百個行業專家模型，可直接調用醫療病歷識別、政策文件解析等專業能力。

3.3 ISAT_with_segment_anything：圖像分割專用工具

核心優勢：基于 Meta 的 SAM 模型，支持 “一鍵標注” 圖像分割掩碼，可導出 COCO、YOLO 等主流格式；
適用場景：自動駕駛、遙感影像等需要圖像分割標注的場景；
亮點功能：支持主動學習和數據增強，能自動生成多樣化訓練樣本。

四、避坑指南：這些問題一定要注意

不要過度依賴大模型：即使模型準確率達 90%，也需人工復核，尤其是關鍵業務場景（如醫療診斷數據）；
控制模型成本：大參數模型推理成本高，可先用小模型（如 7B）做初篩，再用大模型處理復雜案例；
關注召回率：Instruct 模型召回率高但準確率可能低，需在 Prompt 中加入 “拒絕模糊分類” 的規則（如 “不確定時歸為其他”）；
記錄標注過程：保存每個樣本的標注模型、Prompt 版本、復核記錄，方便后續追溯問題。

五、總結

大模型自動化標注不是 “炫技”，而是能切實解決標注痛點的實用技術 —— 它將數據標注從 “體力活” 變成 “技術活”，讓算法工程師能將更多精力放在模型優化和業務理解上。從梳理規則到模型訓練，整個流程的核心是 “閉環思維”：通過預標注、復核、測試的不斷迭代，讓標注質量和模型效果相互促進。

如果你還在被人工標注困擾，不妨從文本分類這類簡單任務開始嘗試，用文中的代碼和工具搭建第一個自動化標注流程。隨著大模型能力的持續提升，自動化標注必將成為 AI 開發的標配，提前掌握這一技能，能讓你在 AI 落地中快人一步。

筆者能力有限，歡迎批評指正或者在留言區討論

參考

??https://mp.weixin.qq.com/s/TG5isq_pvZt9RxtSh9FoaA?? 心法利器[142] | 大模型自動化標注：實戰代碼分享
??https://blog.csdn.net/testManger/article/details/149305594?? 大模型數據集自動化標注技術指南
??https://blog.csdn.net/evm_doc/article/details/146000918?? 大模型時代下的數據標注革命：工具、挑戰與未來趨勢
??https://blog.csdn.net/yzx991013/article/details/148675121?? 如何實現自動標注
??https://blog.csdn.net/lov1993/article/details/145587670?? 【LLM實戰-大模型自動打標】一文搞懂基于LLM大模型進行自動化打標
??https://www.kelote.com/news/case/1025362981531291648?? 大模型標注體系詳解：從傳統NLP到對齊人類偏好的深度探索

本文轉載自??鴻煊的學習筆記??，作者：乘風破浪jxj

標簽

大模型

自動化

標注

已于2025-11-14 00:11:21修改

贊

回復

舉報

社區頭條

熱門內容榜 ? 最近上榜

回復

51CTO

51CTO博客

51CTO學堂