三、项目A · 案例一：数据整理

案例目标：把金融、医疗、法规三份非结构化 txt，用 Ollama 的 qwen3:7B 模型批量转成 指令微调数据集（≥200条，三类各≥60条），保存为 dataset.json 并 FTP 上传。

图 3-1　案例一数据生成管道：读取 → 分块 → Prompt生成 → 质检 → 保存上传

3.1 步骤1-1：加载文档 + 构建 LangChain 生成管道

LangChain 是"大模型应用开发框架"，这里只用它最核心的功能：把 prompt 模板和 LLM 连起来。Ollama 是本地的"模型服务"，启动后会在 http://localhost:11434 提供类似 OpenAI 的接口。

from langchain_community.llms import Ollama
from langchain.prompts import PromptTemplate

# 1. 连接 Ollama 服务里的 qwen3:7b（temperature 越高越有创造性，出题适合 0.7 左右）
llm = Ollama(model="qwen3:7b", base_url="http://localhost:11434", temperature=0.7)

# 2. 设计 Prompt 模板 —— 三要素：角色设定 / 任务说明 / 格式要求（三要素缺一不可！）
prompt = PromptTemplate.from_template("""你是一个擅长总结和出题的AI助手。
请阅读下面的文本片段，从中提炼出 1 个有价值的问题，并给出标准答案。

要求：
1. 问题必须具体、有明确答案，答案必须忠实于原文，不得编造；
2. 严格只输出如下 JSON，不要输出任何解释、前缀或多余内容：
{{"instruction": "生成的问题", "input": "", "output": "标准答案"}}

文本片段：
{text}""")

# 3. 先用一小段文本试运行，确认输出是合法 JSON 再批量跑（重要！先验证再量产）
resp = (prompt | llm).invoke({"text": "次级抵押贷款是指向信用记录较差的借款人发放的贷款..."})
print(resp)

3.2 步骤1-2：批量生成 ≥200 条有效数据

把文档切块（chunk_size≈512），逐块调用模型；解析失败就重试一次，仍失败则跳过；最后去重、统计各领域条数。下面是完整可直接运行的脚本：

# gen_dataset.py —— 生成指令微调数据集
import json, re, time
from langchain_community.llms import Ollama
from langchain.prompts import PromptTemplate

llm = Ollama(model="qwen3:7b", base_url="http://localhost:11434", temperature=0.7)
chain = (prompt | llm)   # prompt 模板同上

# ---------- ① 读取 + 分块 ----------
def read_chunks(path, chunk_size=512):
    """把 txt 按 512 字符分块；优先在句号处切，避免句子被拦腰斩断"""
    text = open(path, encoding="utf-8").read()
    text = re.sub(r"\s+", "", text)          # 压掉空白/换行，提高切块密度
    chunks, start = [], 0
    while start < len(text):
        end = min(start + chunk_size, len(text))
        if end < len(text):                    # 未到结尾：向前找最近的句号切
            dot = text.rfind("。", start, end)
            if dot > start: end = dot + 1
        chunks.append(text[start:end])
        start = end
    return chunks

docs = {"金融": "/home/user/workspace/data/美国次贷危机.txt",
        "医疗": "/home/user/workspace/data/尿毒性心包炎.txt",
        "法规": "/home/user/workspace/data/生成式人工智能服务管理暂行办法.txt"}

# ---------- ② 批量生成 + 解析 + 质检 ----------
dataset, seen = [], set()
def parse_json(s):
    """从模型输出中抠出 JSON（模型有时会加 ```json 包裹或前后废话）"""
    m = re.search(r"\{.*\}", s, re.S)
    return json.loads(m.group()) if m else None

for domain, path in docs.items():
    chunks = read_chunks(path)
    cnt = 0
    for i, ch in enumerate(chunks):
        if len(ch) < 50:  continue          # 太短的块出不了好题
        try:
            out = chain.invoke({"text": ch})
            item = parse_json(out)
            if not item or not item.get("instruction") or not item.get("output"):
                continue                     # 字段缺失 → 无效
            key = item["instruction"]
            if key in seen: continue          # 按问题去重，保证多样性
            seen.add(key)
            dataset.append({"instruction": item["instruction"],
                            "input": "",
                            "output": item["output"]})
            cnt += 1
        except Exception as e:
            print(f"[{domain}#{i}] 解析失败，跳过：{e}")
            time.sleep(1)
    print(f"{domain} 完成：{cnt} 条")

# ---------- ③ 保存 ----------
with open("/home/user/workspace/model_b/data/dataset.json", "w", encoding="utf-8") as f:
    json.dump(dataset, f, ensure_ascii=False, indent=2)
print(f"总计 {len(dataset)} 条")

3.3 本任务涉及的库详解

— AI训练师技术交流教程 · 仅供学习交流 —