把七部《哈利波特》压成六句话，这个挑战听起来像内容运营的标题党，实际上是一个相当硬核的长文本压缩问题。七本书英文原文合计百万词量级，六句话大概一百多字，压缩比接近一比七千。任何做过长文档摘要的人都知道，压缩比一旦超过某个阈值，模型输出的就不再是摘要，而是“看起来像摘要的东西”：人物还在，因果链断了，主题被抹平成几句正确的废话。

这篇文章不评测某个具体模型跑分，而是拆一套可复用的长文本六句话摘要流水线：怎么切块、怎么做层次归并、怎么约束输出句数、怎么评估压缩后信息有没有塌陷、怎么批量跑完整个书系、接口怎么调用、成本怎么算、哪里最容易翻车。同时拿“魔法故事为什么能风靡全球”当作一个真实的评估靶子——如果六句话之后，读者仍然能感受到成长叙事、规则清晰的魔法系统、以及选择与代价这条暗线，那这次压缩就是成功的；如果只剩“一个男孩发现自己会魔法并打败了坏人”，那就是失败的。

这套流水线适合几类人：做长文档摘要、会议纪要、论文速读的工程同学；给 RAG 做前置压缩、想把百万字知识库变成可控上下文的知识库维护者；以及做内容运营、需要把长内容改写成短卡片的自媒体作者。前面几节先给能力矩阵和边界，后面直接进代码，跑通分块、摘要、六句话归并、质量评估、批量任务和排错。

1. 核心能力速览

先把这个方案的能力边界摆清楚。它不是某个带安装包的开源项目，而是一条可以挂到任意兼容 OpenAI 协议接口上的处理链：文本进来，六句话出去，中间的分块、归并、句数约束、质量打分全部由你自己控制。

这里必须强调一句：显存占用、单本处理耗时、token 成本这三项，本文不给具体数字，因为不同模型、不同量化精度、不同上下文长度差异极大，任何写死的数字都会误导人。正确做法是在你自己的机器上按第 8 节的方法实测一遍，再决定用哪档配置。

2. 适用场景与使用边界

先说什么情况适合用。第一类是长文档速读，比如几十万字的行业报告、几本书的技术史、一部小说的全套剧情梳理，人读不完，模型可以先用层次摘要压到可读长度。第二类是知识库前置压缩，把原始长文本压成结构化摘要再入向量库，检索时命中率往往比直接切碎原文更高，因为摘要层已经做完了一次语义聚合。第三类是内容改写，把长文变成短卡片、把一本书变成一条推送，本质也是定句数压缩。

再说边界。定句数摘要天然会丢细节，它保的是骨架不是血肉。如果你要做的是合同条款核对、病历摘要、法条比对，这套东西只能当辅助，最终必须以原文为准。六句话读完一本书，读的是“这本书在讲什么”，不是“这本书里每一处怎么写的”。把这两件事混为一谈，是这类流水线最常见的误用。

版权和合规这块必须单独拎出来讲。拿《哈利波特》做实验，处理的应该是你自己合法持有的文本，实验产物（摘要、评估结果）用于个人学习和技术研究没问题，但不能把原文批量复制、拼接、二次分发，也不能把整本书的摘要做成公开的替代性阅读产品。做批量任务时，输入目录里的素材来源要可追溯，输出目录要做好访问控制。涉及真人素材、未公开手稿、企业内部文档的，先确认授权范围，再决定要不要送进第三方接口。本地部署在这件事上有天然优势：文本不出机器，风险面小很多。

还有一条容易被忽略的：摘要结果不能当作事实来源。模型在压缩过程中会做合理推断，也会做不合理推断。人物关系、时间线、因果归属这类信息，如果后面要拿去做决策，必须回原文校验。第 6 节会给一个可操作的抽检方法。

3. 环境准备与前置条件

环境这块给一份通用检查清单，不写死版本号，因为模型服务和依赖更新很快，写死了反而容易踩坑。

• 操作系统：Linux 或 Windows 均可，批处理建议 Linux，路径和并发都好管。
• Python：3.10 及以上，虚拟环境隔离，别和系统环境混用。
• 推理后端：本地推理服务或云端兼容接口，二者用同一套调用代码，只改 base_url。
• 磁盘：原始文本、中间摘要、最终结果分三个目录，中间产物体积通常是原文的数倍。
• 网络：调云端接口要稳定，批量任务建议加超时和重试。
• 端口：本地起服务默认端口容易撞车，起之前先看占用。

先把虚拟环境和依赖装上：

# 创建隔离环境，避免污染系统 Python python -m venv venv # Linux / macOS 激活 source venv/bin/activate # Windows PowerShell 激活 # .\venv\Scripts\Activate.ps1 # 安装依赖，版本按实际可用版本调整 pip install --upgrade pip pip install openai tqdm tenacity

openai只用来做协议客户端，指向本地服务同样能用；tqdm打进度条；tenacity做重试。三个库足够撑起整条流水线，不需要引入重型框架。

然后是目录约定，这一步看着无聊，但批量跑书系的时候能救命：

workspace/ ├── config/ │ └── pipeline.json # 模型地址、分块参数、并发数 ├── inputs/ # 原始文本，按书目/章节命名 │ └── book01/ │ ├── ch01.txt │ └── ch02.txt ├── chunks/ # 分块中间产物 ├── summaries/ # 块级与章节级摘要 ├── outputs/ # 最终六句话结果 └── logs/ # 每次运行的请求日志与错误日志

命名规则建议book{序号}_ch{章节号}，全程不换。中途换命名，后面归并脚本就得写一堆兼容逻辑，纯属给自己找麻烦。

4. 部署与启动方式

流水线本身没有“一键启动”这种说法，它是脚本形态加一个可选的服务外壳。先给配置文件，把可变参数全部外置，后面调参只改 JSON，不动代码。

{ "base_url": "http://127.0.0.1:8000/v1", "api_key": "EMPTY", "model": "your-model-name", "chunk_max_chars": 8000, "chunk_overlap": 300, "summary_temperature": 0.2, "final_temperature": 0.3, "max_concurrency": 4, "timeout_seconds": 180, "retry_times": 3 }

几个参数值得单独解释。chunk_max_chars是字符数不是 token 数，中英文差异很大，中文 8000 字大概对应几千 token，英文 8000 字符大概两千 token 上下，实际要按你的模型上下文留足余量。chunk_overlap是块间重叠，主要防止一个关键情节被切在两块边界上，两边的摘要都没提。summary_temperature压到 0.2 左右是为了让块级摘要稳定，final_temperature稍高一点，让六句话读起来不那么僵硬。并发数别一上来拉满，先用 2 到 4 试。

如果要把流水线暴露成服务给别的系统调，用一个轻量 HTTP 外壳就行，下面是最小可用版本，实际路径和端口按需改：

# server.py import json from fastapi import FastAPI from pydantic import BaseModel from pipeline import run_pipeline # 你的流水线入口函数 app = FastAPI(title="LongText Six-Sentence Summarizer") class SummarizeRequest(BaseModel): text: str sentences: int = 6 language: str = "zh" @app.post("/api/summarize") def summarize(req: SummarizeRequest): result = run_pipeline( text=req.text, target_sentences=req.sentences, language=req.language, ) return {"code": 0, "sentences": result["sentences"], "usage": result["usage"]}

启动命令：

# 仅监听本机，避免暴露到公网 uvicorn server:app --host 127.0.0.1 --port 7861 --workers 1

--workers 1是有意的。多 worker 会让并发控制失效，真正的并发应该在流水线内部用信号量管，而不是靠多进程放大请求量。这一步做错，批量任务很容易把后端服务打挂。

5. 六句话摘要流水线实现

这一节是核心。整条链路拆成四步：清洗分块、块级摘要、层次归并、定句数重写。每一步都能单独测，单独调，不要一次写完整条链再调试，出了问题根本定位不到哪一步。

5.1 文本清洗与分块

分块的第一原则是不要按固定字符数硬切。硬切会把一句话切成两半，把一段对话拆成两个块，摘要质量直接掉档。按段落切，超过阈值再退化成按句子切。

# pipeline.py import re from typing import List def clean_text(text: str) -> str: """去掉多余空行、页眉页脚残留、连续空白。""" text = text.replace("\r\n", "\n").replace("\r", "\n") text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\n{3,}", "\n\n", text) return text.strip() def split_sentences(text: str) -> List[str]: """中英文混排的粗粒度断句，够用即可。""" parts = re.split(r"(?<=[。！？!?\.])\s+", text) return [p.strip() for p in parts if p.strip()] def chunk_text(text: str, max_chars: int = 8000, overlap: int = 300) -> List[str]: """按段落优先、句子兜底的策略切块，带尾部重叠。""" paragraphs = [p.strip() for p in text.split("\n") if p.strip()] chunks: List[str] = [] buf = "" for para in paragraphs: if len(para) > max_chars: # 超长段落先按句子拆开，再逐句塞入 for sent in split_sentences(para): if len(buf) + len(sent) + 1 <= max_chars: buf = f"{buf}\n{sent}" if buf else sent else: if buf: chunks.append(buf) buf = sent[-max_chars:] continue if len(buf) + len(para) + 1 <= max_chars: buf = f"{buf}\n{para}" if buf else para else: if buf: chunks.append(buf) buf = para if buf: chunks.append(buf) # 加上下文重叠，降低边界信息丢失 if overlap > 0 and len(chunks) > 1: overlapped = [chunks[0]] for i in range(1, len(chunks)): tail = chunks[i - 1][-overlap:] overlapped.append(tail + "\n" + chunks[i]) chunks = overlapped return chunks

重叠不是越大越好。重叠 300 字，处理一本 30 万字的书会多出百分之几的重复计算量，可以接受；重叠 2000 字，成本直接翻倍，收益却很小。300 到 500 是个合理区间。

5.2 块级摘要

块级摘要的提示词要写得克制。不要让它一次吐太多东西，目标只有一个：把这一段里发生的事和涉及的人抽出来。

# llm_client.py import os import json from openai import OpenAI CFG = json.load(open("config/pipeline.json", encoding="utf-8")) client = OpenAI( base_url=os.getenv("LLM_BASE_URL", CFG["base_url"]), api_key=os.getenv("LLM_API_KEY", CFG["api_key"]), timeout=CFG["timeout_seconds"], ) def chat(prompt: str, system: str = "", temperature: float = 0.2) -> str: messages = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) resp = client.chat.completions.create( model=CFG["model"], messages=messages, temperature=temperature, ) return resp.choices[0].message.content.strip()

块级摘要的提示词模板：

CHUNK_SYSTEM = "你是一名严谨的文本摘要工程师，只提取文本中明确出现的信息，不做推测。" CHUNK_PROMPT = """请对下面这段文本做压缩摘要，要求： 1. 输出 3 到 5 个要点，每个要点一句话； 2. 保留人物、事件、因果、转折，删掉描写性内容和重复信息； 3. 不要评价，不要抒情，不要补充原文没有的内容； 4. 用中文输出。 文本： {chunk} """

注意第三条。不让模型“评价”，是因为评价会污染后续归并——归并阶段需要的是一堆平铺的事实，不是一堆观点。观点层留到最后六句话那一步再处理。

5.3 层次归并与六句话重写

归并是分层的。章节内的块摘要先归并成章节摘要，章节摘要再归并成书摘要，书摘要再归并成书系摘要。三层足够了，再多一层收益递减。

def merge_summaries(summaries: list[str], level: str = "chapter") -> str: joined = "\n\n".join(f"[片段{i+1}]\n{s}" for i, s in enumerate(summaries)) if len(joined) <= CFG["chunk_max_chars"]: prompt = f"""下面是同一{level}的多段要点摘要，请合并成一段连续摘要： 1. 去重，合并重复信息； 2. 保留时间推进和因果关系； 3. 控制在 300 字以内。 {joined} """ return chat(prompt, system=CHUNK_SYSTEM, temperature=0.2) # 超长则先两两归并，递归压缩 mid = len(summaries) // 2 left = merge_summaries(summaries[:mid], level) right = merge_summaries(summaries[mid:], level) return merge_summaries([left, right], level)

最后一步，定句数重写。这一步的提示词是全文最关键的地方，写不好就前功尽弃。

FINAL_PROMPT = """下面是一部作品的完整摘要。请把它压缩成恰好 {n} 句话。 硬性要求： 1. 必须是 {n} 句话，多一句少一句都不行； 2. 每句只承载一个核心信息点，句间要有逻辑推进，不要并列堆砌； 3. 必须覆盖：主角的核心处境、关键转折、代价或牺牲、最终结局； 4. 保留专有名词，不要用"某个人""一个地方"这种模糊指代； 5. 不要出现"本书讲述了""这个故事告诉我们"这类元叙述； 6. 不要评论作品好坏。 摘要正文： {summary} """

第三条里的“代价或牺牲”不是随便加的。一步到位的爽文压缩出来是流水账，而有分量的故事，骨架里一定有“付出了什么”。这条约束能把流水账和真正的叙事梗概区分开。

6. 功能测试与效果验证

测试要分层做，一章能不能压对，决定了一本书能不能压对。不要跳过分层直接测整本书，出了问题你连是哪一层坏的都不知道。

6.1 单章摘要测试

拿一章原文进来，跑块级摘要加章节归并，输出 300 字以内。判断成功的标准有三条：出现的人名在原文里都真实存在；事件顺序和原文一致；没有出现原文里根本没写的情节。第三条是最容易翻车的，模型在压缩时特别爱“顺手补全”，把两个相似情节合并，或者给一个模糊结局凭空加个收束。

失败原因通常是分块太长。块内信息太多，模型会自己挑重点，挑着挑着就变形了。把chunk_max_chars降到 5000 左右再试一次，多数情况能修好。

6.2 单本摘要测试

一本书的章节摘要喂进归并层，输出 800 字左右的整书梗概。这一步要看的不是文字漂不漂亮，是时间线有没有断。具体做法是把输出里的关键事件按顺序列出来，和目录对一遍，看有没有前后颠倒或者整段消失。

常见问题是中段塌陷。前几章和后几章信息密度高，中间部分容易被压成一句“经历了一系列冒险”。这是模型对“过渡段落”的天然歧视。解法是在归并提示词里显式要求“中间章节的关键转折不得省略”，或者在归并时给每章摘要带一个序号标签，让模型看得见进度。

6.3 六句话输出测试

七部书的书级摘要再归并一次，得到书系级摘要，最后走定句数重写。这一步要验三件事：

• 句数是否严格等于 6。大概率第一次不达标，模型会输出 5 句或者 8 句。解法是把句数校验做成程序化检查，不符合就带反馈重跑一次，不要让模型自己数。
• 六句之间有没有逻辑推进。好的六句是递进的：处境、入场、规则、冲突、代价、落点。坏的六句是并列的六个事件，读完不知道重点在哪。
• 主题骨架有没有保住。这就是开头说的评估靶子。如果六句话之后，“成长”“规则清晰的魔法体系”“选择与代价”这三条还立得住，说明压缩是有效的；如果六句话只是六个人名加六个动作，那压缩失败了。

6.4 稳定性与重复性测试

同一份输入，temperature设成 0 到 0.2，连跑三次，看输出差异。差异大说明提示词约束不够硬，或者分块边界刚好切在关键情节上。实用做法是把三次结果做一次并集对比，如果某个关键信息三次里只出现一次，那它在原文里的权重很可能被低估了，需要回头调分块。

6.5 一个可自动化的抽检脚本

人工读太慢，写一个粗糙但有效的覆盖率检查，能筛掉大部分事故：

import re def key_entities(text: str, top_k: int = 30) -> set: """极简实体抽取：按出现频次取高频专有名词，仅作抽检用。""" tokens = re.findall(r"[\u4e00-\u9fa5]{2,4}", text) freq = {} for t in tokens: freq[t] = freq.get(t, 0) + 1 ranked = sorted(freq.items(), key=lambda x: -x[1])[:top_k] return {w for w, _ in ranked} def coverage(source: str, summary: str) -> float: """高频词在摘要中的命中率，衡量信息保留程度。""" ents = key_entities(source) if not ents: return 0.0 hit = sum(1 for e in ents if e in summary) return hit / len(ents)

这个指标很糙，会出现大量误报——高频词很多是虚词和常用词。但它在批量任务里当报警器足够用：覆盖率突然从 0.6 掉到 0.2，说明那一批任务的输入或提示词出问题了。真要精细评估，得换成人名实体识别加人工抽检，成本高很多。

7. 接口 API 与批量任务

接口调用没什么特别的，标准 chat completions 就够。curl 版本先跑通，再接代码：

curl -s http://127.0.0.1:7861/api/summarize \ -H "Content-Type: application/json" \ -d '{ "text": "把清洗后的章节正文放在这里", "sentences": 6, "language": "zh" }'

Python 侧带重试的批量调用：

import json import time from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed from tenacity import retry, stop_after_attempt, wait_exponential from llm_client import chat, CFG from pipeline import clean_text, chunk_text LOG = Path("logs/run.log") @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=20)) def safe_chunk_summary(chunk: str) -> str: from pipeline import CHUNK_PROMPT, CHUNK_SYSTEM return chat(CHUNK_PROMPT.format(chunk=chunk), system=CHUNK_SYSTEM, temperature=0.2) def process_file(path: Path, out_dir: Path) -> dict: raw = clean_text(path.read_text(encoding="utf-8", errors="ignore")) chunks = chunk_text(raw, CFG["chunk_max_chars"], CFG["chunk_overlap"]) results, errors = [], 0 for i, c in enumerate(chunks): try: results.append(safe_chunk_summary(c)) except Exception as e: errors += 1 with LOG.open("a", encoding="utf-8") as f: f.write(f"{path.name}#{i}\t{type(e).__name__}\t{e}\n") out_path = out_dir / f"{path.stem}.chunks.json" out_path.write_text( json.dumps({"source": path.name, "chunks": results, "errors": errors}, ensure_ascii=False, indent=2), encoding="utf-8", ) return {"file": path.name, "chunks": len(chunks), "ok": len(results), "err": errors} def run_batch(in_dir: Path, out_dir: Path, workers: int = 4): out_dir.mkdir(parents=True, exist_ok=True) files = sorted(in_dir.glob("*.txt")) stats = [] with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(process_file, f, out_dir): f for f in files} for fut in as_completed(futures): try: stats.append(fut.result()) except Exception as e: stats.append({"file": futures[fut].name, "fatal": str(e)}) with LOG.open("a", encoding="utf-8") as f: f.write(f"[batch] {time.strftime('%Y-%m-%d %H:%M:%S')} {json.dumps(stats, ensure_ascii=False)}\n") return stats

批量任务的设计要点只有三条。第一，中间产物必须落盘，块级摘要写完就存 JSON，不要等全部跑完再一起写——跑到一半挂了，前面全白干。第二，失败重试要区分错误类型，超时和限流可以重试，参数错误重试一百次也没用。第三，并发数要小于等于后端服务能承受的并行请求数，本地单卡服务开 8 并发经常不是变快而是集体超时。

文件命名和输出结构建议固定成{书名}.chunks.json、{书名}.chapter.json、{书名}.final.json三级，方便断点续跑。断点续跑的判断逻辑很简单：输出文件存在且errors字段为 0 就跳过。

8. 资源占用与性能观察

这一节不编数字，只给观察方法。因为同一段代码在 8G 显存和 24G 显存、在 4K 上下文和 128K 上下文的表现可以差出一个数量级。

观察项一：单块处理耗时。在日志里记录每次请求的起止时间，跑十个块取中位数。如果中位数明显高于预期，先看是不是并发把显存打满了，触发换页之后速度会断崖式下跌。

观察项二：上下文利用率。分块大小是字符数，模型吃的是 token。用分词器算一下你的块平均 token 数，再对比模型的上下文上限，留出百分之三十以上的余量。留太少，长块会被截断，摘要就少一段内容，而你完全看不出来。

观察项三：总调用次数。一本 30 万字的书按 8000 字一块，大概是 40 个块，块级摘要 40 次加归并若干次，单本就接近 50 次调用。七本加起来三百多次。这个数字决定了你的成本量级，跑之前心里要有数，别跑完才发现账单。

观察项四：CPU 推理和 GPU 推理的差异。这不只是速度差，还有稳定性差。CPU 跑长上下文往往慢到没法批量，但胜在显存零占用，适合小规模验证；GPU 跑得快，但并发一高就 OOM，适合串行加小并发。

降低资源占用的几个实用手段：把块调小、把归并层数减少、把中间摘要的字数上限压下来、对已经处理过的章节做结果缓存。第四条最有效——调提示词的时候反复重跑同一批数据，缓存能省掉绝大部分重复开销。

进程和端口这块也要盯。服务跑完没退出，端口一直占着，下次启动就报地址已占用。排查很简单：

# 查看端口占用（Linux / macOS） lsof -i :7861 # Windows 查看端口占用 netstat -ano | findstr :7861

确认是残留进程再杀，别上来就一把梭，误杀别的服务很麻烦。

9. 常见问题与排查方法

排查有一个通用原则：先定位到层。块级摘要出错和归并出错，现象完全不同。块级出错是细节错、人名错；归并出错是顺序错、整段消失。分不清是哪一层，就逐层打印中间结果，别猜。

10. 最佳实践与使用建议

第一条，先用小样本把链路跑通。挑三章、一本书、一轮六句话，三十分钟能验证完的事，不要一上来就扔七本书进去跑一小时然后发现提示词写错了。

第二条，把提示词当代码管。版本号、改动原因、对应的评估结果都记下来。摘要这类任务，提示词改一个词，输出风格能差出很远，不记录就没法回溯。

第三条，中间产物一定要落盘且带时间戳。调参阶段你会反复重跑，没有缓存就是纯烧时间。

第四条，批量任务必须有日志和失败清单。跑完不检查errors字段，等于没跑。哪怕只有一条失败，也要单独重试，不要整批重来。

第五条，接口服务默认只监听本机。真要对外提供，加鉴权、加限流、加请求体大小限制，别把一个裸接口暴露出去。

第六条，涉及有版权的文本、真实人物素材、企业内部文档时，先确认授权范围再处理。摘要产物也不等于可以自由分发，尤其是整本书级别的梗概，公开发布前想清楚用途。

第七条，发布或商用前必须人工复核。定句数摘要这种强压缩，模型很容易在最后一句硬凑一个“升华”，读起来像模像样，实际是编的。最后一句尤其要盯。

回到开头那个问题：六句话能不能读完《哈利波特》。技术上能，但这六句话的价值不在“省了多少阅读时间”，而在它逼着你去想——一部作品里到底哪些东西是不能丢的。是主角的处境，是关键转折，是选择的代价，是规则的清晰度。这套流水线真正的用处，是把“什么算重要”这件事从直觉变成可配置的参数：你想保留情节，就调事件覆盖率；你想保留主题，就调骨架约束；你想保留风格，就单独跑一遍文风迁移。同一套分块和归并逻辑，换个提示词就能复用。

如果你现在手上正好有一堆读不完的长文档，建议先从一个章节开始，把chunk_max_chars调到 5000，跑出第一版六句话，看看丢的是什么。丢的东西，往往比留下的更能说明你的提示词哪里没写好。