用 Ollama 构建长文档本地摘要器：分段总结与数字核对

Ollama 把开源模型跑在你自己的电脑上，并提供本地 API。 这篇做 ollama_summarize.py：把长文档切段，逐段让本地模型写要点，再合并成总摘要；同时检查摘要有没有把原文里的数字弄丢，丢了就标出来让人核对。

先看清成品

• 在 box 上用 qwen2.5-local:1.5b 真实跑通：733 字的示例报告切成 2 段，约 11 秒出摘要。
• 发现并处理两个真实问题：小模型摘要丢数字；“必须保留数字”的重试让模型几乎照抄原文（361/370 字）。
• 最终方案：重试结果太长就不用，改为在 summary.md 末尾列出“需要人工核对的数字”。pytest 7 项通过。

合规提醒：模型在本机运行，数据不出本机；但每个模型有自己的许可证（在 ollama.com/library 的模型页面查看），商用前先确认。本文只用 Ollama 官方安装方式和它自带的本地 API（默认 http://localhost:11434），不使用任何非官方镜像或破解版本。

运行说明：本文所有输出都是在 box 上真实跑出来的（Ollama 0.40.7，8 核 Intel Xeon CPU、15 GB 内存、无 GPU，2026-10-21 20:38–20:56，UTC+8）。box 的网络不允许从官方模型库下载：ollama pull qwen2.5:1.5b 报错 redirect target not allowed: …r2.cloudflarestorage.com resolves to non-public 198.18.0.1。所以我用的是 box 上之前从 GGUF 文件导入的模型，名字带 -local（qwen2.5-local:1.5b、qwen3-local:0.6b、deepseek-r1-distill:1.5b）。你在自己电脑上直接 ollama pull qwen2.5:1.5b，把命令里的模型名换掉即可。速度数据只代表这台 CPU 机器。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

80-ollama-A-src/
├── ollama_summarize.py
├── pilot_report.txt
└── test_summarize.py

第 1 步：主脚本

"""用本地 Ollama 对长文档做“分段摘要 → 合并摘要”。只访问本机 http://localhost:11434。"""
import argparse, json, re, sys, time
from pathlib import Path
import requests

HOST = "http://localhost:11434"

def chunk_text(text, max_chars=1500):
    """按段落切块；单段超长时按句号硬切。"""
    paras = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks, cur = [], ""
    for p in paras:
        while len(p) > max_chars:
            cut = p.rfind("。", 0, max_chars)
            cut = cut + 1 if cut > 0 else max_chars
            if cur: chunks.append(cur); cur = ""
            chunks.append(p[:cut]); p = p[cut:].strip()
        if len(cur) + len(p) + 2 > max_chars and cur:
            chunks.append(cur); cur = p
        else:
            cur = (cur + "\n\n" + p).strip()
    if cur: chunks.append(cur)
    return chunks

def chat(model, messages, *, host=HOST, num_ctx=4096, stream_to=None, session=requests):
    """调用 /api/chat（流式）。返回 (全文, 最后一条统计)。"""
    body = {"model": model, "messages": messages, "stream": True,
            "options": {"temperature": 0, "num_ctx": num_ctx}, "keep_alive": "5m"}
    r = session.post(f"{host}/api/chat", json=body, stream=True, timeout=600)
    if r.status_code == 404:
        raise SystemExit(f"模型 {model} 不存在：先运行 ollama pull {model}（或 ollama list 查看本机模型名）")
    r.raise_for_status()
    out, last = [], {}
    for line in r.iter_lines():
        if not line: continue
        d = json.loads(line)
        if "error" in d: raise RuntimeError(d["error"])
        piece = d.get("message", {}).get("content", "")
        out.append(piece)
        if stream_to and piece: stream_to.write(piece); stream_to.flush()
        if d.get("done"): last = d
    return "".join(out), last

NUM = re.compile(r"\d+(?:\.\d+)?%?")

def missing_numbers(src, summary, top=6):
    """源文本中出现的数字（取前 top 个，按出现顺序去重）有哪些没进摘要。"""
    seen = []
    for n in NUM.findall(src):
        if n not in seen: seen.append(n)
    return [n for n in seen[:top] if n not in summary]

SYS = "你是严谨的中文摘要助手。只依据给出的文本，不添加文本中没有的信息。"

def summarize(text, model, *, max_chars=1500, session=requests, log=sys.stderr):
    chunks = chunk_text(text, max_chars)
    partial, stats, notes = [], [], []
    for i, c in enumerate(chunks, 1):
        t0 = time.time()
        s, st = chat(model, [{"role": "system", "content": SYS},
                             {"role": "user", "content": f"用 3 条以内的要点总结下面这段：\n\n{c}"}], session=session)
        miss = missing_numbers(c, s)
        if len(miss) >= 2:  # 小模型常把数字丢掉：带着清单重试一次
            print(f"  漏掉数字 {miss}，重试", file=log)
            s2, st2 = chat(model, [{"role": "system", "content": SYS},
                                   {"role": "user", "content": f"用 3 条以内的要点总结下面这段，必须保留这些数字：{'、'.join(miss)}\n\n{c}"}], session=session)
            if len(s2.strip()) > 0.6 * len(c):  # 重试结果几乎是照抄原文：不用
                print(f"  重试结果 {len(s2.strip())} 字，接近原文 {len(c)} 字，保留第一次结果", file=log)
            else:
                s, st = s2, st2
            still = missing_numbers(c, s)
            if still: notes.append((i, still))
        partial.append(s.strip()); stats.append(st)
        print(f"[{i}/{len(chunks)}] {len(c)} 字 → {len(s.strip())} 字，{time.time()-t0:.1f}s", file=log)
    if len(partial) == 1:
        return partial[0], partial, stats, notes
    merged = "\n\n".join(f"第{i}段要点：\n{p}" for i, p in enumerate(partial, 1))
    final, st = chat(model, [{"role": "system", "content": SYS},
                             {"role": "user", "content": f"把下面各段要点合并成一份不超过 5 条的总摘要，去掉重复：\n\n{merged}"}], session=session)
    stats.append(st)
    return final.strip(), partial, stats, notes

def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("file"); ap.add_argument("--model", default="qwen2.5:1.5b")
    ap.add_argument("--max-chars", type=int, default=1500); ap.add_argument("-o", default="summary.md")
    a = ap.parse_args(argv)
    text = Path(a.file).read_text(encoding="utf-8")
    final, partial, stats, notes = summarize(text, a.model, max_chars=a.max_chars)
    toks = sum(s.get("eval_count", 0) for s in stats)
    md = f"# 摘要：{Path(a.file).name}\n\n{final}\n\n## 分段要点\n\n" + "\n\n".join(
        f"### 第 {i} 段\n\n{p}" for i, p in enumerate(partial, 1))
    if notes:
        md += "\n\n## 需要人工核对：摘要没有覆盖的数字\n\n" + "\n".join(f"- 第 {i} 段：{'、'.join(n)}" for i, n in notes)
    md += f"\n\n---\n模型 {a.model}，共生成 {toks} tokens，本地运行。\n"
    Path(a.o).write_text(md, encoding="utf-8"); print(f"已写入 {a.o}")

if __name__ == "__main__":
    main()

第 2 步：测试

import json, io
import ollama_summarize as m

def test_chunk_respects_limit():
    text = "\n\n".join(["甲" * 400] * 5)
    cs = m.chunk_text(text, 1000)
    assert all(len(c) <= 1000 for c in cs) and len(cs) == 3

def test_chunk_hard_split_on_sentence():
    text = ("这是一句话。" * 300)
    cs = m.chunk_text(text, 500)
    assert all(len(c) <= 500 for c in cs) and all(c.endswith("。") for c in cs)

class FakeResp:
    def __init__(self, lines, code=200): self.lines, self.status_code = lines, code
    def raise_for_status(self): pass
    def iter_lines(self): return [json.dumps(l).encode() for l in self.lines]

class FakeSession:
    def __init__(self): self.bodies = []
    def post(self, url, json=None, **kw):
        self.bodies.append(json)
        return FakeResp([{"message": {"content": "要点"}, "done": False},
                         {"message": {"content": "A"}, "done": True, "eval_count": 2}])

def test_stream_joined_and_temperature_zero():
    s = FakeSession()
    txt, st = m.chat("x", [{"role": "user", "content": "hi"}], session=s)
    assert txt == "要点A" and st["eval_count"] == 2
    assert s.bodies[0]["options"]["temperature"] == 0 and s.bodies[0]["stream"] is True

def test_map_reduce_calls():
    s = FakeSession()
    final, parts, stats, notes = m.summarize("\n\n".join(["乙" * 900] * 3), "x", max_chars=1000, session=s, log=io.StringIO())
    assert len(parts) == 3 and len(s.bodies) == 4  # 3 段 + 1 次合并

def test_missing_model_message():
    class S:
        def post(self, *a, **k): return FakeResp([], 404)
    try:
        m.chat("nope", [], session=S()); assert False
    except SystemExit as e:
        assert "ollama pull nope" in str(e)

def test_missing_numbers():
    assert m.missing_numbers("成本 2.6 万元，占 58%，共 12 周", "成本约 2.6 万") == ["58%", "12"]

def test_copying_retry_rejected():
    src = "成本 2.6 万元，占 58%，共 12 周。" * 10
    replies = iter(["很贵", src])  # 第一次漏数字；重试照抄原文
    class S:
        def post(self, url, json=None, **kw):
            return FakeResp([{"message": {"content": next(replies)}, "done": True}])
    final, parts, stats, notes = m.summarize(src, "x", max_chars=2000, session=S(), log=io.StringIO())
    assert final == "很贵" and notes and notes[0][0] == 1

第 3 步：运行

$ ollama list
NAME                          ID              SIZE      MODIFIED
deepseek-r1-distill:1.5b      4eab8234d6a6    1.1 GB    4 hours ago
qwen2.5-local:1.5b            7ae7e314f0ab    1.1 GB    4 hours ago
qwen3-local:0.6b              70ddd2951d4b    639 MB    4 hours ago
…
$ python ollama_summarize.py pilot_report.txt --model qwen2.5-local:1.5b --max-chars 400
  漏掉数字 ['2026', '6', '840', '41%', '70%', '18']，重试
[1/2] 371 字 → 49 字，1.6s
  漏掉数字 ['1', '512', '61%', '76%', '14', '3']，重试
  重试结果 361 字，接近原文 370 字，保留第一次结果
[2/2] 370 字 → 80 字，7.3s
已写入 summary.md
$ cat summary.md
# 摘要：pilot_report.txt

试点目标是提高厨余垃圾的正确投放率，从41%提高到70%以上。试点分为宣传和定时定点投放两个阶段。试点引入积分激励，居民投放垃圾获得积分，积分兑换日用品。积分激励推广后厨余垃圾投放率提高，但人力成本高。居民代表提出三点建议，社区计划扩大试点并逐步替换督导员。

## 分段要点

### 第 1 段

试点目标是提高厨余垃圾的正确投放率，从41%提高到70%以上。试点分为宣传和定时定点投放两个阶段。

### 第 2 段

试点引入积分激励，居民投放垃圾获得积分，积分兑换日用品。积分激励推广后厨余垃圾投放率提高，但人力成本高。居民代表提出三点建议，社区计划扩大试点并逐步替换督导员。

## 需要人工核对：摘要没有覆盖的数字

- 第 1 段：2026、6、840、18
- 第 2 段：1、512、61%、76%、14、3

---
模型 qwen2.5-local:1.5b，共生成 165 tokens，本地运行。
$ python -m pytest -q
7 passed

怎么检查结果？

• 对照原文读一遍摘要：小模型可能概括得不准（本例把“三条建议”只写成“提出三点建议”）。
• “需要人工核对的数字”列出的是原文前几个数字中没出现在摘要里的；不是所有数字都必须保留，由你判断哪些重要。
• 同一输入、temperature=0 时结果基本稳定，但不同 Ollama 版本或量化版本可能不同。

调试示例

1. 小模型摘要把关键数字丢了。 第一版只做“分段 → 合并”，第 2 段（积分、结果、成本、建议）的摘要是：

试点引入积分激励，居民投放垃圾正确率达到 76%。

人力成本 2.6 万元、占预算 58%、居民的三条建议全没了。于是我加了 missing_numbers()：漏掉 2 个以上数字就带着清单重试一次。

2. 重试“矫枉过正”。 加了重试后再跑：

  漏掉数字 ['1', '512', '61%', '76%', '14', '3']，重试
[2/2] 370 字 → 361 字，7.5s

模型为了“保留这些数字”，几乎把原文抄了一遍（361/370 字），这已经不是摘要了。修复：重试结果超过原文 60% 长度就丢弃，保留第一次的摘要，并把仍未覆盖的数字写进 summary.md 的“需要人工核对”一节。1.5B 的小模型做不到“又短又全”，与其硬逼它，不如把风险交给人看。

常见问题

模板规则

• 分段摘要后一定要有合并步骤。
• 检查数字等硬信息，丢了就标出来。
• 重试结果要再检查，不能默认更好。
• 小模型的结论都要人工核对。

项目资源

本地模型便宜又私密，但要给它配一个会检查的人。