长文档问答

把长文档塞进 Claude 的 200K 上下文,标记缓存,流式输出答案。第一轮之后的每一个追问都按 1/10 输入价命中缓存。

请求结构

三层,顺序固定:

1. System block — 怎么回答的指令(可缓存)。
2. System block — 文档本身,带清晰分隔符(必缓存)。
3. User message — 实际问题(不缓存)。

{
  "model": "claude-sonnet-4-6",
  "max_tokens": 1024,
  "stream": true,
  "system": [
    {
      "type": "text",
      "text": "你只能基于下面文档回答。文档里没有的内容直接说没有。涉及来源时原文引用。"
    },
    {
      "type": "text",
      "text": "<DOCUMENT name=\"contract.pdf\">\n... 25,000 tokens 的文档文本 ...\n</DOCUMENT>",
      "cache_control": {"type": "ephemeral"}
    }
  ],
  "messages": [
    {"role": "user", "content": "终止条款是什么,需要提前多少天通知?"}
  ]
}

选模型

完整流式样例

"""
长文档问答,带流式。文档第一次写入缓存,后续每个问题命中缓存。
依赖:pip install anthropic
"""
import os, pathlib
from anthropic import Anthropic

client = Anthropic(
    base_url="https://buzzai.cc",
    api_key=os.environ["BUZZ_API_KEY"],
)

DOC_PATH = pathlib.Path("contract.txt")
DOC_TEXT = DOC_PATH.read_text()

INSTRUCTIONS = (
    "你只能基于下面文档回答。"
    "文档里没有的内容直接说没有。"
    "涉及来源时原文引用。"
)

def system_blocks():
    return [
        {"type": "text", "text": INSTRUCTIONS},
        {
            "type": "text",
            "text": f'<DOCUMENT name="{DOC_PATH.name}">\n{DOC_TEXT}\n</DOCUMENT>',
            "cache_control": {"type": "ephemeral"},
        },
    ]


def ask(question: str):
    print(f"\nQ: {question}\nA: ", end="", flush=True)
    usage = None
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system_blocks(),
        messages=[{"role": "user", "content": question}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
        final = stream.get_final_message()
        usage = final.usage
    print()
    print(
        f"  [usage] input={usage.input_tokens} "
        f"cache_create={usage.cache_creation_input_tokens} "
        f"cache_read={usage.cache_read_input_tokens} "
        f"output={usage.output_tokens}"
    )


if __name__ == "__main__":
    ask("终止条款是什么,需要提前多少天通知?")
    ask("有没有竞业限制?期限多长?")
    ask("用三条要点总结赔偿条款。")

// 长文档问答 + 流式。文档第一次写入缓存。
// 依赖:npm i @anthropic-ai/sdk
import Anthropic from "@anthropic-ai/sdk";
import { readFileSync } from "node:fs";

const client = new Anthropic({
  baseURL: "https://buzzai.cc",
  apiKey: process.env.BUZZ_API_KEY,
});

const DOC_PATH = "contract.txt";
const DOC_TEXT = readFileSync(DOC_PATH, "utf8");

const INSTRUCTIONS =
  "你只能基于下面文档回答。" +
  "文档里没有的内容直接说没有。" +
  "涉及来源时原文引用。";

function systemBlocks() {
  return [
    { type: "text", text: INSTRUCTIONS },
    {
      type: "text",
      text: `<DOCUMENT name="${DOC_PATH}">\n${DOC_TEXT}\n</DOCUMENT>`,
      cache_control: { type: "ephemeral" },
    },
  ];
}

async function ask(question) {
  process.stdout.write(`\nQ: ${question}\nA: `);
  const stream = client.messages.stream({
    model: "claude-sonnet-4-6",
    max_tokens: 1024,
    system: systemBlocks(),
    messages: [{ role: "user", content: question }],
  });

  for await (const event of stream) {
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      process.stdout.write(event.delta.text);
    }
  }
  const final = await stream.finalMessage();
  console.log(
    `\n  [usage] input=${final.usage.input_tokens} ` +
      `cache_create=${final.usage.cache_creation_input_tokens} ` +
      `cache_read=${final.usage.cache_read_input_tokens} ` +
      `output=${final.usage.output_tokens}`
  );
}

await ask("终止条款是什么,需要提前多少天通知?");
await ask("有没有竞业限制?期限多长?");
await ask("用三条要点总结赔偿条款。");

usage 长这样

Q: 终止条款是什么...
A: 任一方提前 60 天书面通知即可终止...
  [usage] input=22 cache_create=24187 cache_read=0 output=98

Q: 有没有竞业限制...
A: 第 8.2 条规定 12 个月竞业限制...
  [usage] input=22 cache_create=0 cache_read=24187 output=84

Q: 用三条要点总结赔偿条款...
A: - 双方在知识产权侵权方面互相赔偿...
  [usage] input=22 cache_create=0 cache_read=24187 output=120

第一次调用把 24K token 写入缓存。后续每次都按输入价的 10% 读 24K。三次调用总共只花~1.2 倍的单次输入成本,而不是 3 倍。

文档预处理

用分隔符包裹

Claude 对显式边界很敏感。用清晰的开闭标签 + name 属性,模型能反向引用给你:

<DOCUMENT name="employment-agreement-2026.pdf">
... 文本 ...
</DOCUMENT>

多文档

2-3 个文档时,直接用各自的标签拼接,都放在同一个缓存块里:

<DOCUMENT name="contract-a.pdf">...</DOCUMENT>

<DOCUMENT name="contract-b.pdf">...</DOCUMENT>

几十个文档时,改用「先检索后缓存」:每次问题取 top 3-5 块,塞进 system,接受检索集变化时缓存会部分失效。

PDF 和 Office 文件

先转文本。常见管线:pdftotext、pandoc,或者 pypdf / pdfjs-dist 之类的库。把页眉页脚和页码去掉,这些东西污染上下文还会影响引用。

流式输出

设 stream: true。Claude 输出 SSE 事件;SDK 提供 text_stream 迭代器,只吐可见文本。最终 usage 在结尾通过 get_final_message() / finalMessage() 拿到。

不用 SDK 的客户端看到的原始 SSE:

event: message_start
data: {"type":"message_start","message":{"id":"msg_...","usage":{"input_tokens":22,"cache_creation_input_tokens":24187,"cache_read_input_tokens":0,...}}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"任一方"}}
...
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":98,...}}

event: message_stop
data: {"type":"message_stop"}

Cache TTL

Anthropic 默认 ephemeral 缓存自最后一次命中后存活 5 分钟。如果用户连续问问题,基本够用,每次新问题都会刷新计时器。会话有较长停顿时,选 1 小时 TTL:

"cache_control": {"type": "ephemeral", "ttl": "1h"}

1 小时 TTL 写入价高一些(约 5 分钟版的 2 倍),但读价相同。会话间隔超过 5 分钟时值得开。

让回答有据可查

两个能稳定降幻觉的 prompt 模式:

模式 1:不在文档里就拒答

如果文档里没有答案,只回复:
"NOT_IN_DOCUMENT: <简短原因>"

不要用文档之外的知识。

模式 2:引用原文

回答里每一个论断后面必须跟一段文档原文,
用 <quote>...</quote> 标签包裹。
没有原文支持的论断,不要写。

高风险场景(法务、合规)两个一起用。面向消费者的摘要,模式 1 单独够用。

什么场景不适合

相关链接

• Prompt Caching 概念
• 流式处理指南
• POST /v1/messages 参考
• Prompt Cache 生产实战