这套流水线要解决的问题很具体：领域文献每天都在更新，而模型的参数停留在训练截止那一天。手工把新论文读成训练数据，成本高且跟不上节奏。下面这条链路把「抓论文 → 解析 → 造数据 → 持续微调 → 评测回滚」串成一个可以定时跑的闭环，让模型跟着某个领域（比如 NLP、生物信息、推荐系统）的文献持续进化。
读完可以做出这样一套东西：每天早上自动拉取新增论文，生成一批带原文引用的指令数据，跑一轮轻量微调，用固定评测集和回归集打分，分数不降才保留新权重，否则自动回滚。
前置条件清单
- 一台能跑推理与微调的机器。7B 级别模型用 LoRA 做增量训练，消费级显卡即可；更大的模型考虑量化或租用算力。
- Python 环境与基础依赖：
transformers、peft、trl、datasets、accelerate。 - 一个用于生成数据的教师模型接口。可以直接调商用 API，也可以本地部署一个更大的开源模型。接口参数以所用服务的官方文档为准。
- 磁盘空间：每篇论文 PDF 通常几 MB，解析后的文本会膨胀数倍，加上每周的适配器权重，留出几十 GB 比较稳妥。
- 一个评测集。这是整条链路里最容易被忽略、但决定成败的部分。没有它，无法判断模型是在进化还是在退化。
分步骤搭建
第一步：建目录，装依赖
```bash
mkdir -p paper2llm/{raw,parsed,datasets,ckpt,logs,eval}
cd paper2llm
python -m venv .venv && source .venv/bin/activate
pip install arxiv pymupdf datasets peft transformers trl accelerate
```
目录约定：raw 放 PDF 与元数据，parsed 放结构化文本，datasets 放生成的训练样本，ckpt 放适配器权重，eval 放冻结的评测集。
第二步：增量抓取 arXiv 论文
arXiv 提供官方 API，支持按分类、关键词、提交时间检索。关键是「增量」：每次只拉上次运行之后的新论文，避免重复处理。
```python
fetch.py
import json
import pathlib
from datetime import datetime, timedelta, timezone
import arxiv
OUT = pathlib.Path("raw")
STATE = pathlib.Path("raw/last_run.txt")
def fetch(query: str, days: int = 7, limit: int = 300):
since = datetime.now(timezone.utc) - timedelta(days=days)
client = arxiv.Client(page_size=100, delay_seconds=3, num_retries=3)
search = arxiv.Search(
query=query,
max_results=limit,
sort_by=arxiv.SortCriterion.SubmittedDate,
sort_order=arxiv.SortOrder.Descending,
)
seen = set()
for r in client.results(search):
if r.published.astimezone(timezone.utc) < since:
break
pid = r.get_short_id()
if pid in seen:
continue
seen.add(pid)
pdf_path = OUT / f"{pid.replace('/', '_')}.pdf"
if not pdf_path.exists():
r.download_pdf(dirpath=str(OUT), filename=pdf_path.name)
meta = {
"id": pid,
"title": r.title,
"abstract": r.summary,
"authors": [a.name for a in r.authors],
"published": r.published.isoformat(),
"categories": r.categories,
"pdf": str(pdf_path),
}
(OUT / f"{pid.replace('/', '_')}.json").write_text(
json.dumps(meta, ensure_ascii=False), encoding="utf-8"
)
STATE.write_text(datetime.now(timezone.utc).isoformat(), encoding="utf-8")
if __name__ == "__main__":
fetch("cat:cs.CL OR cat:cs.LG", days=7)
```
delay_seconds 与检索语法以 arxiv 包与 arXiv 官方 API 文档的当前说明为准。请求频率别调太激进，夜间批量跑更合适。
第三步：把 PDF 解析成干净文本块
PDF 解析有三个敌人：双栏排版导致句子交错、参考文献混进正文、公式与表格变成乱码。前两个可以靠规则处理，第三个建议走 arXiv 的 LaTeX 源码（e-print）路线，能拿到公式的原始写法。
```python
parse.py
import pathlib
import re
import fitz # PyMuPDF
NOISE = re.compile(r"^(arxiv:|doi:|preprint|under review|figure \d+|table \d+)", re.I)
def parse_pdf(path: pathlib.Path, min_len: int = 40):
doc = fitz.open(path)
blocks = []
for page in doc:
for b in page.get_text("blocks"):
按 (y, x) 排序，缓解双栏错位
bbox, text = b[:4], b[4]
text = " ".join(text.split())
if len(text) < min_len or NOISE.match(text):
continue
blocks.append((page.number, bbox[1], bbox[0], text))
blocks.sort(key=lambda t: (t[0], round(t[1] / 10), t[2]))
lines = [t[3] for t in blocks]
砍掉参考文献之后的内容
for i in range(int(len(lines) * 0.6), len(lines)):
if re.match(r"^(references|bibliography)\b", lines[i], re.I):
lines = lines[:i]
break
return lines
def chunk(lines, size: int = 1200, overlap: int = 150):
text = "\n".join(lines)
chunks, start = [], 0
while start < len(text):
chunks.append(text[start:start + size])
start += size - overlap
return [c for c in chunks if len(c.strip()) > 200]
if __name__ == "__main__":
out = pathlib.Path("parsed")
for pdf in pathlib.Path("raw").glob("*.pdf"):
lines = parse_pdf(pdf)
chunks = chunk(lines)
(out / f"{pdf.stem}.json").write_text(
__import__("json").dumps(chunks, ensure_ascii=False), encoding="utf-8"
)
```
摘要字段别浪费。很多论文的摘要本身就是高质量的「问题—方法—结论」三元组，可以直接作为数据源。
第四步：生成训练数据，但只生成可验证的题
这一步决定数据质量。一个可靠的原则：优先做抽取式与改写式任务，谨慎做自由生成式任务。前者有原文兜底，后者容易把模型幻觉固化成训练信号。
四类稳妥的题型：
1. 术语解释：给定片段，用一句话解释其中的关键概念，要求引用原文句子。
2. 方法问答：从「方法」章节生成问题，答案必须能在片段里找到。
3. 贡献抽取：从摘要中抽取论文解决的核心问题与提出的做法。
4. 审稿式判断：给定实验设置，指出缺失的对照组或未说明的超参。
生成脚本骨架：
```python
generate.py
import json
import pathlib
from concurrent.futures import ThreadPoolExecutor
PROMPT = """你是一名论文数据标注员。下面是一段论文原文。
要求：
1. 生成 {n} 条问答对，答案必须能在原文中找到依据。
2. 每条答案附上原文中的证据句，逐字引用，不要改写证据句。
3. 如果某条内容无法从原文验证，直接跳过，不要编造。
4. 输出严格的 JSON 数组，每项包含 question / answer / evidence / type。
原文：
{chunk}
"""
def build_sample(chunk, meta, n=3):
call_teacher 请替换为你自己的模型调用
raw = call_teacher(PROMPT.format(n=n, chunk=chunk))
items = safe_json_loads(raw)
samples = []
for it in items:
ev = it.get("evidence", "")
证据句必须真实出现在原文里，否则丢弃
if not ev or ev[:40] not in chunk:
continue
samples.append({
"paper_id": meta["id"],
"type": it.get("type", "qa"),
"instruction": it["question"],
"input": ev,
"output": it["answer"],
"source_chunk": chunk[:200],
})
return samples
```
三个必须加的护栏：
- 证据校验：
evidence的片段必须在原文里逐字出现，这一步能筛掉大部分幻觉。 - 每篇限额：单篇论文最多产出固定条数，比如 20 条。否则一篇长综述会淹没整个数据集。
- 按 paper_id 切分：训练集与评测集必须按论文划分，绝不能按样本随机划分。同一篇论文的片段高度相似，按样本切分等于把答案泄给模型。
第五步：质量过滤与去重
```python
filter.py
import json
import pathlib
JUDGE = """给下面这条训练样本打分（1-5），只输出数字。
5 = 问题清晰、答案准确且有原文支撑
3 = 表述含糊但事实无误
1 = 答案与原文不符或属于常识废话
问题：{q}
答案：{a}
证据：{e}
"""
def keep(sample, threshold=4):
score = int(call_teacher(JUDGE.format(
q=sample["instruction"], a=sample["output"], e=sample["input"]
)).strip()[0])
return score >= threshold
```
去重分两层：指令文本用 MinHash 或 simhash 做近似去重；语义层面用句向量算AI 词典：余弦相似度">余弦相似度，把超过阈值的样本合并。教师模型倾向于反复生成「这篇论文的主要贡献是什么」这类模板题，不做去重的话，数据集会被几十条同质样本占据。
第六步：持续微调
用 LoRA 做增量训练，每周产出一个新适配器，而不是重训全量参数。这样成本低、可回滚、便于对比。
```python
train.py
from datasets import load_dataset
from peft import LoraConfig
from transformers import AutoModelForCausalLM, AutoTokenizer
from trl import SFTConfig, SFTTrainer
BASE = "你的基座模型路径或名称" # 以官方文档当前版本为准
tok = AutoTokenizer.from_pretrained(BASE)
model = AutoModelForCausalLM.from_pretrained(BASE, device_map="auto")
lora = LoraConfig(
r=16,
lora_alpha=32,
lora_dropout=0.05,
target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
task_type="CAUSAL_LM",
)
ds = load_dataset("json", data_files="datasets/current.jsonl", split="train")
cfg = SFTConfig(
output_dir="ckpt/run",
num_train_epochs=1,
per_device_train_batch_size=2,
gradient_accumulation_steps=8,
learning_rate=1e-4,
logging_steps=10,
save_strategy="epoch",
max_length=1024,
)
SFTTrainer(model=model, args=cfg, train_dataset=ds, peft_config=lora,
processing_class=tok).train()
```
要点：target_modules 要按基座模型的层命名来写，不同架构不一致；学习率比全量微调大一个量级是常见做法；训练轮数控制在 1 到 2 轮，多了容易过拟合到那批论文上。
混入通用数据是防遗忘的关键。建议领域数据与通用指令数据按大致 7:3 到 8:2 混合，具体比例要在你的评测集上试。
第七步：评测，然后决定留不留
评测集在项目开始时冻结一份，之后只增不改。每次训练完跑三类指标：
- 领域能力：新论文 QA 的准确率，用另一个更强的模型做裁判，或者人工抽检。
- 回归能力：通用问答、指令遵循、格式输出。分数掉超过阈值就判为「灾难性遗忘」。
- 证据一致性：模型回答里引用的句子是否真的出现在给定片段中。
对比方式：同一条 prompt 分别喂给基座、上一版适配器、新版适配器，三方对照。避免只跟基座比——上一版才是真正的基准线。
```bash
保留策略示意
python eval/run_eval.py --model ckpt/run --suite eval/frozen.jsonl --out logs/run.json
python eval/decide.py --new logs/run.json --baseline logs/last_good.json --tol 0.02
```
decide.py 的逻辑很简单：领域分数上涨，且回归分数下降不超过容差，就把新适配器标记为 last_good；否则归档但不上线。
第八步：串成定时任务
把上面几步写成 shell 脚本，用 cron 或 systemd timer 每周跑一次。每一步落盘一份 manifest，记录论文 ID 列表、数据条数、训练超参、评测分数。这样任何一次「模型变差了」都能顺着 manifest 追回去。
```bash
#!/usr/bin/env bash
set -euo pipefail
cd /path/to/paper2llm
source .venv/bin/activate
python fetch.py
python parse.py
python generate.py --out datasets/$(date +%F).jsonl
python filter.py --in datasets/$(date +%F).jsonl --out datasets/current.jsonl
python train.py
python eval/run_eval.py --model ckpt/run --out logs/$(date +%F).json
python eval/decide.py --new logs/$(date +%F).json
```
常见坑与排错
抓取被限流或超时。 把 delay_seconds 调大，失败重试交给客户端处理，并记录已下载的 ID 做幂等。整条链路要能重跑而不重复下载。
双栏 PDF 文本交错。 单纯按抽取顺序拼接会得到「左栏第一句 + 右栏第一句」的怪句子。按坐标排序是最低成本的缓解，更彻底的办法是走 arXiv 的 LaTeX 源码或用版面分析模型。
参考文献污染训练集。 模型会学会复述作者名和期刊名，看起来像懂了，其实只是背了列表。解析阶段务必截断 References 之后的内容。
生成的题目全是模板。 教师模型有强烈的路径依赖。解决办法是给出多种题型模板，并在 prompt 里随机采样，同时在去重阶段严格过滤。
评测分数虚高。 训练集和评测集里出现了同一篇论文的片段，分数自然好看。按 paper_id 划分是硬性要求，写测试断言去卡。
模型学会了新领域，忘了怎么正常说话。 典型的灾难性遗忘。降低学习率、增加通用数据配比、减少训练轮数，三招一起上。
适配器越存越多，不知道哪个能用。 每个适配器旁边存一份 manifest，至少记录基座模型标识、数据文件哈希、评测分数。名字里带上日期和分数，比 final_v2_new 靠谱得多。
版权与许可。 arXiv 上每篇论文的许可条款不同，用于训练前请核对原文的 license 字段，商用场景尤其要谨慎。
下一步建议
跑通主链路之后，可以往三个方向加码。
一是把 RAG 接进来做对照。很多所谓「模型记住了新知识」的效果，检索增强也能达到，且不用训练、可即时更新。两条路线的评测结果放在一起比，才知道微调到底值不值。
二是引入偏好优化。用评测阶段的模型输出做排序，构造 chosen/rejected 对，在 SFT 之后接一轮偏好训练，通常能改善回答的取舍判断。
三是让数据配比自动调。把「领域数据：通用数据」的比例当成超参，用评测分数做搜索，跑几轮之后会得到一个适合当前领域的配方。这条路径的价值在于，它把「人工调参」变成了流水线里的一环，闭环才算真正合上。