STORM：按章节检索，再写长文

用户想要一份完整的杭州茶文化旅游攻略。直接把所有资料塞进一个 prompt 让模型写出来，结果是：结构松散、证据混在不同段落、引用难以审计。

STORM 的做法：先列大纲，每个章节单独检索、单独写，最后编辑合成。

一句话

STORM 把"一次检索一次写"变成"大纲 → 按章节检索 → 按章节写 → 编辑合成"，让长文的结构和证据可控。

它修什么失败

问题	表面看起来	实际风险
一次性生成长文	快	结构松散
所有资料塞一个 context	信息很多	证据跨章节混用
没有编辑环节	内容存在	重复、语气不一致

它引入什么复杂度

谁	负责什么
大纲步骤	定义章节
章节检索器	只检索该章节需要的资料
章节写作者	基于该章节证据写内容
编辑/组装器	合并、去重、统一语气

和其他模式的关系

flowchart LR
  ARAG["Agentic RAG"] -.->|"长文"| STM["STORM"]
  RL["Retrieval Loop"] -.->|"长文"| STM
  STM -.->|"事实验证"| COVE["CoVe"]
  STM -.->|"质量审查"| MC["Maker-Checker"]

Agentic RAG / Retrieval Loop：STORM 的每个章节内部就是一次 Retrieval Loop。
CoVe：写完后可以对每个章节的断言做 CoVe 验证。
Maker-Checker：每个章节或最终文章可以过一轮 Maker-Checker。

完整实现：杭州茶文化旅游攻略

"""storm_travel.py — 按章节检索和写作的长文生成"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any


# ── 极简运行时 ──────────────────────────────────────────

@dataclass(frozen=True)
class Message:
    role: str
    content: str

class MockLLM:
    def __init__(self, responses: list[str]):
        self._responses = list(responses)
        self._idx = 0
    def complete(self, messages: list[Message]) -> str:
        if self._idx >= len(self._responses):
            raise RuntimeError("MockLLM 回复用完了")
        out = self._responses[self._idx]
        self._idx += 1
        return out

@dataclass(frozen=True)
class Document:
    doc_id: str
    text: str

class SimpleSearchIndex:
    def __init__(self, docs: list[Document]):
        self._docs = docs
    def search(self, query: str, k: int = 3) -> list[Document]:
        terms = self._tokenize(query)
        if not terms:
            return []
        scored = []
        for doc in self._docs:
            doc_tokens = self._tokenize(doc.text)
            score = sum(1 for t in terms for d in doc_tokens if t == d)
            if score > 0:
                scored.append((doc, score))
        scored.sort(key=lambda x: x[1], reverse=True)
        return [doc for doc, _ in scored[:k]]
    @staticmethod
    def _tokenize(text: str) -> list[str]:
        tokens, buf = [], ""
        for ch in text.lower():
            if ch.isalnum() or '\u4e00' <= ch <= '\u9fff':
                buf += ch
            else:
                if buf:
                    tokens.append(buf)
                    buf = ""
        if buf:
            tokens.append(buf)
        return tokens


# ── STORM 数据结构 ──────────────────────────────────────

@dataclass
class Section:
    title: str
    content: str
    evidence_ids: list[str] = field(default_factory=list)

@dataclass
class Article:
    topic: str
    sections: list[Section]
    final_text: str


# ── STORM 核心逻辑 ──────────────────────────────────────

def storm_write(
    model: MockLLM,
    *,
    topic: str,
    index: SimpleSearchIndex,
    top_k: int = 3,
) -> Article:
    """
    1. 列大纲
    2. 每个章节：生成 query → 检索 → 写
    3. 组装最终文章
    """
    # 步骤 1：列大纲
    raw_outline = model.complete([
        Message(role="system", content="为主题创建简短大纲。只返回 JSON。"),
        Message(role="user", content=f"主题：{topic}"),
    ])
    outline = json.loads(raw_outline)["sections"]
    print(f"[大纲] {len(outline)} 个章节：{outline}")

    # 步骤 2：逐章节检索和写作
    sections: list[Section] = []
    for title in outline:
        # 2a：生成检索 query
        raw_query = model.complete([
            Message(role="system", content="为这个章节生成检索 query。只返回 JSON。"),
            Message(role="user", content=f"主题：{topic}\n章节：{title}"),
        ])
        query = json.loads(raw_query)["query"]

        # 2b：检索
        docs = index.search(query, k=top_k)
        evidence_ids = [d.doc_id for d in docs]
        notes = "\n".join(f"- [{d.doc_id}] {d.text[:100]}" for d in docs)
        print(f"[章节 '{title}'] query='{query}', 命中 {len(docs)} 条")

        # 2c：写章节
        content = model.complete([
            Message(
                role="system",
                content="用证据写这个章节。引用来源 [doc_id]。",
            ),
            Message(
                role="user",
                content=f"主题：{topic}\n章节：{title}\n\n证据：\n{notes}",
            ),
        ])
        sections.append(Section(title=title, content=content, evidence_ids=evidence_ids))

    # 步骤 3：组装
    sections_text = "\n\n".join(f"## {s.title}\n{s.content}" for s in sections)
    final = model.complete([
        Message(role="system", content="把各章节组装成完整文章。统一语气，去重。"),
        Message(role="user", content=f"主题：{topic}\n\n{sections_text}"),
    ])
    print(f"[组装] 完成")

    return Article(topic=topic, sections=sections, final_text=final)


# ── 运行 ────────────────────────────────────────────────

def main() -> None:
    docs = [
        Document("tea_history", "龙井茶始于宋代，盛于明清。西湖龙井是中国十大名茶之一。"),
        Document("tea_museum", "中国茶叶博物馆，龙井路 88 号，免费，9:00-17:00，展示中国茶文化历史。"),
        Document("tea_village", "龙井村可体验采茶，春季最佳，步行友好，约 2 小时。"),
        Document("tea_ceremony", "杭州多家茶馆提供茶艺表演，推荐湖畔居、青藤茶馆。"),
        Document("tea_food", "龙井虾仁是杭州名菜，用龙井茶嫩芽炒制。龙井茶酥也是特色茶点。"),
        Document("weather", "杭州春季多雨，建议备伞。4-5 月气温 15-25 度。"),
    ]
    index = SimpleSearchIndex(docs)

    model = MockLLM([
        # 大纲
        json.dumps({
            "sections": ["龙井茶的历史", "茶文化体验", "茶主题美食"]
        }, ensure_ascii=False),

        # 章节 1：query
        '{"query":"龙井茶历史起源"}',
        # 章节 1：写
        "龙井茶始于宋代，盛于明清，是中国十大名茶之一 [tea_history]。中国茶叶博物馆详细展示了这段历史 [tea_museum]。",

        # 章节 2：query
        '{"query":"杭州茶文化体验采茶"}',
        # 章节 2：写
        "到龙井村可以亲手采茶，春季最佳，步行约 2 小时 [tea_village]。市区内湖畔居、青藤茶馆有茶艺表演 [tea_ceremony]。",

        # 章节 3：query
        '{"query":"杭州茶美食龙井虾仁"}',
        # 章节 3：写
        "龙井虾仁用龙井茶嫩芽炒制，是杭州名菜 [tea_food]。龙井茶酥也值得尝试 [tea_food]。",

        # 组装
        (
            "# 杭州茶文化旅游攻略\n\n"
            "## 龙井茶的历史\n"
            "龙井茶始于宋代，盛于明清，是中国十大名茶之一 [tea_history]。"
            "中国茶叶博物馆（龙井路 88 号，免费）详细展示了这段历史 [tea_museum]。\n\n"
            "## 茶文化体验\n"
            "到龙井村可以亲手采茶，春季最佳，步行约 2 小时 [tea_village]。"
            "市区内湖畔居、青藤茶馆有专业茶艺表演 [tea_ceremony]。\n\n"
            "## 茶主题美食\n"
            "龙井虾仁用龙井茶嫩芽炒制，是杭州必尝名菜。"
            "龙井茶酥也值得尝试 [tea_food]。"
        ),
    ])

    article = storm_write(model, topic="杭州茶文化旅游攻略", index=index)
    print(f"\n{'='*50}")
    print(article.final_text)
    print(f"{'='*50}")
    print(f"\n章节证据：")
    for s in article.sections:
        print(f"  {s.title}: {s.evidence_ids}")


if __name__ == "__main__":
    main()

运行日志

[大纲] 3 个章节：['龙井茶的历史', '茶文化体验', '茶主题美食']
[章节 '龙井茶的历史'] query='龙井茶历史起源', 命中 2 条
[章节 '茶文化体验'] query='杭州茶文化体验采茶', 命中 3 条
[章节 '茶主题美食'] query='杭州茶美食龙井虾仁', 命中 1 条
[组装] 完成

==================================================
# 杭州茶文化旅游攻略

## 龙井茶的历史
龙井茶始于宋代，盛于明清，是中国十大名茶之一 [tea_history]。中国茶叶博物馆（龙井路 88 号，免费）详细展示了这段历史 [tea_museum]。

## 茶文化体验
到龙井村可以亲手采茶，春季最佳，步行约 2 小时 [tea_village]。市区内湖畔居、青藤茶馆有专业茶艺表演 [tea_ceremony]。

## 茶主题美食
龙井虾仁用龙井茶嫩芽炒制，是杭州必尝名菜。龙井茶酥也值得尝试 [tea_food]。
==================================================

章节证据：
  龙井茶的历史: ['tea_history', 'tea_museum']
  茶文化体验: ['tea_village', 'tea_ceremony', 'tea_museum']
  茶主题美食: ['tea_food']

调试：trace 里看什么

要看的	怎么看	说明什么
大纲质量	章节标题是否有意义	太泛的标题 → 检索也会泛
每章节 query	trace 里的 query	query 和章节不匹配 → 证据跑偏
证据隔离	每个章节的 evidence_ids	A 章节引用了 B 章节的来源 = 串味
最终文章 vs 章节	对比 final_text 和各 section	组装时丢了引用或加了幻觉
章节间重复	读最终文章	编辑步骤没去重

工程笔记

成本公式

大纲 = 1 次调用
每章节 = 1（query）+ 1 次检索 + 1（写） = 2 次模型调用
组装 = 1 次调用
总调用 = 1 + n_sections × 2 + 1 = 2 + 2n

3 个章节 = 8 次模型调用。10 个章节 = 22 次。

常见坑

坑	现象	修法
大纲太浅	每个章节都很泛	写完大纲先审查
证据串味	A 章节引用 B 章节的来源	按章节隔离证据账本
context 爆炸	所有文档全进最终 prompt	按章节摘要
假引用	引用不存在	强制要求检索器返回的 doc id
组装太弱	拼接但没编辑	明确要求组装步骤统一语气、去重

什么时候用

输出是文章、报告、攻略。
章节结构很重要。
证据要按章节隔离。
最终编辑环节有用。

什么时候别用

用户要一句话答案。
预算不支持多次检索和写作。
不需要引用。
大纲不确定，应该先让人列好。

读完以后

STORM 是长文写作的检索工作流。

如果是动态问答而不是长文，读 Agentic RAG。如果写完后要逐条验证事实，读 CoVe。如果只需要简单检索，退回 RAG。