文档目录
结构恢复
用两次请求从丢失格式的纯文本中重建 Markdown:一次把硬换行的行重新拼回,一次给每个块分类(标题、列表、代码块、提示框)。
非官方中文翻译 · 内容以 英文原文 ↗ 为准 · 同步于 2026-10-15
本实战示例接收的输入是一份标记被剥除的纯文本(行在句子中间硬换行切断,没有标题标记、没有列表项目符号),并把它的结构重建为 Markdown:标题、段落、列表、引用、代码块、提示框。输入是一份恰好处于这种状态的团队备忘录。
文本生成模型可以把文本改写成 Markdown,但改写也会改动措辞。这里的模型从不生成文本:它只回答关于文档的窄义问题(这一行是不是接着上一句?这个块是什么类型的内容?),渲染由代码完成,因此输出中的每一个字符都来自输入,每一次判断都带有一个概率。
整个流水线对每个文档执行两轮 API 请求,按顺序进行:
- Pass 1,拼接: 对每一对相邻行提一个
Noul(是非题)问题(其答案就是"是"为正确的概率),问这两行之间的换行是否把一个句子撕成了两段。所有这些对都放进同一次请求中,那些承接断句的行会被合并回同一个块。 - Pass 2,分类: 对每个合并后的块提一个
Choice(选择题)问题(从一个列表中选一个,并对每个选项给出一个概率),从标题、段落、列表项、引用、代码块或提示框(脱离正文的一条注释、提示或警告)中选一。这些块只有在 Pass 1 完成之后才存在,所以这是第二次请求;它同时为每个块附带一些陪问题(标题档位、步骤顺序、提示框种类),答案只有在块的类型让它们有意义时才读取。 - 直接证据由代码读取。 空行和显式标记(
-、1.、#)在代码里读取,绝不发回给模型让它重新判断;这份备忘录保留了空行,但丢失了所有标记。模型只拿到代码无法从文本中回答的那些问题。
所有行为都规定在 Pass 2 的问题判定标准里:三组一行一句的描述字典,外加放在 classify_questions 里步骤题的对/错判定标准。其余代码都是围绕这些的管道。开销和延迟数据见附录:本备忘录两轮往返,10,211 tokens,0.8 秒,$0.0015。
安装与配置
pip install ipython 'cooksafe>=0.2.0,<0.3.0'然后设置 TYPESAFE_API_KEY。每次 API 调用都会缓存在 json_cache.json 中,该文件随本实战示例一起发布,因此重新渲染会直接复用已发布的数据,无需再次调用 API。删除该文件即可重新实时执行全部流程。
import os
import re
import urllib.request
from pathlib import Path
from time import perf_counter
from cooksafe import JsonCache, make_playground_link
from IPython.display import Markdown, display
from typesafe_sdk import Choice, Noul, NoulCriteria, TypeSafeClient
TYPESAFE_MODEL = "jev-1.12"
PRICE = (0.042, 0.00) # $ per 1M tokens (input, output); TypeSafe jev-1.12 as of 2026-09
client = TypeSafeClient(api_key=os.environ["TYPESAFE_API_KEY"], timeout=120.0)
json_cache = JsonCache(Path("json_cache.json"))测试文档:一份丢失格式的团队备忘录
测试文档是一份关于构建系统迁移的备忘录,正好处于到达纯文本收件箱时的状态:段落被硬换行打断,一句 shell 命令单独成行,两段列表既没有项目符号也没有编号,一段警告没有任何标记表明它是警告。文本从一个固定版本的 gist 拉取,以保证本实战示例的数据可复现。
GIST = (
"https://gist.githubusercontent.com/eugene-shvarts/6df7daf97233bf92bcdd6b386a0fa561"
"/raw/5da03690611fb6ddcbaabdb91fb9f91d9751b113/build-memo.txt"
)
@json_cache
def fetch_document(url: str) -> str:
request = urllib.request.Request(url, headers={"User-Agent": "typesafe-cookbook/1.0"})
with urllib.request.urlopen(request) as response:
return response.read().decode()
RAW = fetch_document(GIST)
print(RAW[:560])Migration to the new build system
Hi everyone, quick heads up about the build system migration that is
happening next week. We have been running the new pipeline in shadow
mode for three weeks and the results look solid, so it is time to
make the switch for real.
What changes for you
The old make targets keep working until the end of the month. The new
entrypoint is a single command that wraps everything, including the
docs build that used to be separate.
bun run build
Generated artifacts no longer need to be committed. The new pipeline
uploads them切行、追踪空行、为每行打 id 都在代码里完成,不涉及模型。每行得到一个短 id(L014| );这些 id 是普通文本,模型作为上下文的一部分读取它们,问题与答案都通过这些 id 引用各行(与语义搜索实战示例使用同一套方案)。
def to_lines(text: str) -> list[dict]:
lines, gap = [], False
for raw in text.split("\n"):
stripped = re.sub(r"[\t ]+", " ", raw).strip()
if not stripped:
gap = bool(lines) # a leading blank is not a break
continue
lines.append({"text": stripped, "gap": gap})
gap = False
return lines
def tag(items: list[dict], prefix: str) -> str:
return "\n".join(
f"{chr(10) if item['gap'] else ''}{prefix}{i:03d}| {item['text']}"
for i, item in enumerate(items)
)
def line_id(i: int) -> str:
return f"L{i:03d}"
def block_id(i: int) -> str:
return f"B{i:03d}"
LINES = to_lines(RAW)
print(f"{len(LINES)} non-blank lines. The model sees, e.g.:")
print("\n".join(tag(LINES, "L").splitlines()[19:24]))28 non-blank lines. The model sees, e.g.:
L013| The cutover touches three teams, so check whether you are on this
L014| list before you plan anything for Monday:
L015| The platform team
L016| The web client team
L017| Whoever still owns the release toolingPass 1:拼接断句
每一对相邻行问一个 Noul(是非题)问题,全部放进同一次请求;中间隔有空行的相邻行对会被跳过。问题被刻意设计得很窄("这一行是不是接着上一句?"),它接近文本本身的一个客观事实。措辞的选择以及合并阈值的推导过程都放在附录中。
def join_question(i: int) -> Noul:
return Noul(
instructions=f"Does line {line_id(i)} pick up mid-sentence, continuing a sentence left unfinished at the end of line {line_id(i - 1)}?",
criteria=NoulCriteria(
true="The line starts in the middle of a sentence that began on the previous line - the line break tore the sentence apart",
false="The line begins a new sentence, item, heading, or thought of its own",
),
)
@json_cache
def stitch(wording: str = "mid-sentence") -> dict:
make = join_question if wording == "mid-sentence" else naive_join_question
questions = {line_id(i): make(i) for i in range(1, len(LINES)) if not LINES[i]["gap"]}
started = perf_counter()
response = client.system_one(
state=tag(LINES, "L"), questions=questions, model=TYPESAFE_MODEL
)
return {
"joins": [
response.answers[line_id(i)].noul if line_id(i) in response.answers else 0.0
for i in range(len(LINES))
],
"seconds": round(perf_counter() - started, 2),
"usage": [response.usage.input_tokens, response.usage.output_tokens],
}
result = stitch()
print(f"{sum(1 for l in LINES if not l['gap']) - 1} pair questions, one request, "
f"{result['seconds']}s")16 pair questions, one request, 0.32s合并的阈值取决于上一行是如何结尾的。在未结束的行(行末没有句末标点)之后,合并概率 ≥ 0.2 就把这两行合并;在出现终止标点(. ! ? : ;)之后,阈值上调到 0.5。附录会逐步展示这两个数字背后的概率分布。
JOIN_AFTER_DANGLING, JOIN_AFTER_TERMINAL = 0.2, 0.5
def ends_terminal(text: str) -> bool:
return re.search(r'[.!?:;…]["\')\]]*$', text) is not None
def merge(joins: list[float]) -> list[dict]:
blocks = []
for i, line in enumerate(LINES):
bar = (
JOIN_AFTER_TERMINAL
if i and ends_terminal(LINES[i - 1]["text"])
else JOIN_AFTER_DANGLING
)
if blocks and not line["gap"] and joins[i] >= bar:
blocks[-1]["text"] += " " + line["text"]
blocks[-1]["lines"].append(i)
else:
blocks.append({"text": line["text"], "lines": [i], "gap": line["gap"]})
return blocks
blocks = merge(result["joins"])
healed = len(LINES) - len(blocks)
print(f"{len(LINES)} lines -> {len(blocks)} blocks ({healed} line breaks healed)")
for i, block in enumerate(blocks):
n = len(block["lines"])
print(f"{block_id(i)} {n} line{'s' if n > 1 else ' '} {block['text'][:62]}")28 lines -> 17 blocks (11 line breaks healed)
B000 1 line Migration to the new build system
B001 4 lines Hi everyone, quick heads up about the build system migration t
B002 1 line What changes for you
B003 3 lines The old make targets keep working until the end of the month.
B004 1 line bun run build
B005 3 lines Generated artifacts no longer need to be committed. The new pi
B006 2 lines The cutover touches three teams, so check whether you are on t
B007 1 line The platform team
B008 1 line The web client team
B009 1 line Whoever still owns the release tooling
B010 1 line Things to do before Monday
B011 1 line Update your local toolchain to version 2.4 or later
B012 1 line Delete the old build cache directory
B013 1 line Run the doctor script and fix anything it flags
B014 3 lines If the doctor script reports a red result on the toolchain che
B015 2 lines As Dana put it in the kickoff, "a migration nobody notices is
B016 1 line Thanks, and shout if anything looks off.Pass 2:块的分类
每个拼接块都会收到一个 Choice 问题:这段内容属于什么类型?。这三份字典,外加下面 classify_questions 中 step 问题的真假判定标准,就是分类器的全部规格说明。没有别的逻辑。想把这条流水线适配到自己的文档上,就去改这几段描述。
TYPE_CRITERIA = {
"heading": "A short label or title that names the document or the section that follows it - not a full sentence of content",
"paragraph": "Running prose: one or more complete sentences of explanatory or narrative text",
"list_item": "One entry in a list of parallel items - an ingredient, a feature, a task, an attendee; reads as one of several sibling entries",
"quote": "Words attributed to a person or source - quoted speech, a citation, an excerpt someone else wrote",
"code": "Computer code, a shell command, terminal output, or a config snippet meant to be read verbatim",
"callout": "A warning, tip, or important note that interrupts the flow to flag something the reader must not miss",
}
HLEVEL_CRITERIA = {
"title": "The title of the whole document",
"section": "A major section heading within the document",
"subsection": "A minor heading nested under a section",
}
CALLOUT_CRITERIA = {
"note": "Neutral extra information the reader should be aware of",
"tip": "A helpful suggestion or shortcut that makes things easier",
"warning": "A caution about something that can go wrong or cause harm",
}下面全是胶水代码:构造问题,发一次请求,把答案读回来。如果 type 字段返回 heading,渲染器就需要一个标题层级;如果返回 list_item,就要知道顺序是否重要;如果返回 callout,就要知道是哪一种。类型在请求时尚不可知,等类型结果回来再问就意味着第三轮往返,所以配套问题在同一请求里一并提前问出。不过大部分答案最终用不上:段落(paragraph)这种类型的 step 概率毫无意义,被直接忽略。多加一个问题几乎没什么代价,因为 state 占了请求里绝大部分 token,反正无论如何都要发一次;而多走一次往返则要额外增加整整一次请求的延迟。
HEADING_MAX_CHARS = 90 # longer blocks can't render as headings, so don't ask
def classify_questions(texts: list[str]) -> dict:
questions = {}
for i, text in enumerate(texts):
bid = block_id(i)
questions[f"type_{bid}"] = Choice(
instructions=f"What kind of content is block {bid}?", criteria=TYPE_CRITERIA
)
if len(text) <= HEADING_MAX_CHARS:
questions[f"hlevel_{bid}"] = Choice(
instructions=f"As a heading, what level would block {bid} occupy in this document's structure?",
criteria=HLEVEL_CRITERIA,
)
questions[f"step_{bid}"] = Noul(
instructions=f"Is block {bid} an instruction in a sequence where the order of the items matters?",
criteria=NoulCriteria(
true="It is one step of a procedure - the items around it must happen in order",
false="Order is irrelevant - it is a loose collection, or not a list item at all",
),
)
questions[f"callout_{bid}"] = Choice(
instructions=f"What kind of aside is block {bid}?", criteria=CALLOUT_CRITERIA
)
return questions
@json_cache
def classify(texts: list[str], gaps: list[bool]) -> dict:
tagged = tag([{"text": t, "gap": g} for t, g in zip(texts, gaps)], "B")
questions = classify_questions(texts)
started = perf_counter()
response = client.system_one(state=tagged, questions=questions, model=TYPESAFE_MODEL)
judgments = []
for i in range(len(texts)):
bid = block_id(i)
type_answer = response.answers[f"type_{bid}"]
hlevel = response.answers.get(f"hlevel_{bid}")
judgments.append(
{
"type": type_answer.choice,
"confidence": type_answer.confidence,
"probabilities": type_answer.probabilities,
"hlevel": hlevel.choice if hlevel else "section",
"step": response.answers[f"step_{bid}"].noul,
"callout": response.answers[f"callout_{bid}"].choice,
}
)
return {
"judgments": judgments,
"n_questions": len(questions),
"seconds": round(perf_counter() - started, 2),
"usage": [response.usage.input_tokens, response.usage.output_tokens],
}
classified = classify([b["text"] for b in blocks], [b["gap"] for b in blocks])
for block, judgment in zip(blocks, classified["judgments"]):
block.update(judgment)
print(f"{classified['n_questions']} questions about {len(blocks)} blocks, one request, "
f"{classified['seconds']}s\n")
print(f"{'block':<6}{'type':<11}{'conf':<6}{'companion used':<18}text")
for i, b in enumerate(blocks):
companion = {
"heading": f"level={b['hlevel']}",
"list_item": f"step={b['step']:.2f}",
"callout": f"kind={b['callout']}",
}.get(b["type"], "-")
print(f"{block_id(i):<6}{b['type']:<11}{b['confidence']:.2f} {companion:<18}"
f"{b['text'][:46]}")62 questions about 17 blocks, one request, 0.51s
block type conf companion used text
B000 heading 0.99 level=title Migration to the new build system
B001 paragraph 0.98 - Hi everyone, quick heads up about the build sy
B002 heading 0.75 level=section What changes for you
B003 paragraph 0.89 - The old make targets keep working until the en
B004 code 1.00 - bun run build
B005 paragraph 0.90 - Generated artifacts no longer need to be commi
B006 paragraph 0.43 - The cutover touches three teams, so check whet
B007 list_item 0.99 step=0.15 The platform team
B008 list_item 1.00 step=0.16 The web client team
B009 list_item 0.99 step=0.12 Whoever still owns the release tooling
B010 heading 0.96 level=section Things to do before Monday
B011 list_item 0.98 step=0.86 Update your local toolchain to version 2.4 or
B012 list_item 0.99 step=0.87 Delete the old build cache directory
B013 list_item 0.92 step=0.90 Run the doctor script and fix anything it flag
B014 callout 0.65 kind=warning If the doctor script reports a red result on t
B015 quote 0.99 - As Dana put it in the kickoff, "a migration no
B016 paragraph 0.92 - Thanks, and shout if anything looks off.每个块的判定都列在那张表里,companion 列展示了提前问出的答案各自派上了什么用场:「Things to do before Monday」下面三行的 step 概率都在 0.9 附近(最终渲染成有序列表),那三行介绍 team 的项 step 落在 0.1 附近(用项目符号),而那段没有额外标记、只提到 doctor script 的警告被归为 kind 为 warning 的 callout。附录会去分析模型信心不足的那个块。
Rendering
代码根据这些判定把页面拼出来。相邻的列表项会合并成一个列表,当这些项的 step 概率均值不低于 0.5 时,就用有序编号。这个阈值是组一级的决策,没有任何单一问题会直接问到它。
STEP_THRESHOLD = 0.5
HEADING_MARK = {"title": "#", "section": "##", "subsection": "###"}
CALLOUT_MARK = {"note": "NOTE", "tip": "TIP", "warning": "WARNING"}
def to_markdown(blocks: list[dict]) -> str:
groups = []
for b in blocks:
if b["type"] in ("list_item", "code") and groups and groups[-1][0] == b["type"]:
groups[-1][1].append(b)
else:
groups.append((b["type"], [b]))
parts = []
for kind, items in groups:
if kind == "list_item":
ordered = sum(b["step"] for b in items) / len(items) >= STEP_THRESHOLD
parts.append("\n".join(
f"{n + 1}. {b['text']}" if ordered else f"- {b['text']}"
for n, b in enumerate(items)
))
elif kind == "code":
parts.append("```\n" + "\n".join(b["text"] for b in items) + "\n```")
elif kind == "heading":
parts.append(f"{HEADING_MARK[items[0]['hlevel']]} {items[0]['text']}")
elif kind == "quote":
parts.append(f"> {items[0]['text']}")
elif kind == "callout":
parts.append(f"> [!{CALLOUT_MARK[items[0]['callout']]}]\n> {items[0]['text']}")
else:
parts.append(items[0]["text"])
return "\n\n".join(parts) + "\n"
markdown = to_markdown(blocks)
print(markdown)# Migration to the new build system
Hi everyone, quick heads up about the build system migration that is happening next week. We have been running the new pipeline in shadow mode for three weeks and the results look solid, so it is time to make the switch for real.
## What changes for you
The old make targets keep working until the end of the month. The new entrypoint is a single command that wraps everything, including the docs build that used to be separate.
```
bun run build
```
Generated artifacts no longer need to be committed. The new pipeline uploads them to the registry automatically, and checking them in just creates merge conflicts.
The cutover touches three teams, so check whether you are on this list before you plan anything for Monday:
- The platform team
- The web client team
- Whoever still owns the release tooling
## Things to do before Monday
1. Update your local toolchain to version 2.4 or later
2. Delete the old build cache directory
3. Run the doctor script and fix anything it flags
> [!WARNING]
> If the doctor script reports a red result on the toolchain check, do not proceed with the migration. Ping the infra channel first and we will sort it out together.
> As Dana put it in the kickoff, "a migration nobody notices is the only kind worth shipping."
Thanks, and shout if anything looks off.上面每个字都来自输入。流水线所决定的,仅是边界、类型和标记。
在 playground 中打开
此分享链接包含拼接后的 blocks 以及完整的 pass-2 问题 (question) 集。打开它即可实时重新运行分类。
playground_link = make_playground_link(
tag(blocks, "B"),
classify_questions([b["text"] for b in blocks]),
models=[TYPESAFE_MODEL],
)
display(Markdown(f"🔗 [Open the stitched memo + questions in the TypeSafe playground]({playground_link})"))在 TypeSafe playground 中打开拼接后的 memo + 问题 →
附录
成本与延迟
tokens = [result["usage"], classified["usage"]]
total_in, total_out = sum(t[0] for t in tokens), sum(t[1] for t in tokens)
cost = total_in / 1e6 * PRICE[0] + total_out / 1e6 * PRICE[1]
n_joins = sum(1 for l in LINES if not l["gap"]) - 1
print(f"pass 1 {n_joins} questions {result['seconds']}s")
print(f"pass 2 {classified['n_questions']} questions {classified['seconds']}s")
print(f"total {total_in + total_out:,} tokens "
f"{result['seconds'] + classified['seconds']:.1f}s ${cost:.4f}")pass 1 16 questions 0.32s
pass 2 62 questions 0.51s
total 10,211 tokens 0.8s $0.0003两轮往返,10,211 token,0.8s,$0.0015。
join 阈值的来源
pass 1 的逐行 join 概率:
print("join line")
for i, line in enumerate(LINES[:18]):
join = " " if i == 0 or line["gap"] else f"{result['joins'][i]:.2f}"
print(f"{join} {line_id(i)}| {line['text'][:66]}")join line
L000| Migration to the new build system
L001| Hi everyone, quick heads up about the build system migration that
0.77 L002| happening next week. We have been running the new pipeline in shad
0.62 L003| mode for three weeks and the results look solid, so it is time to
0.39 L004| make the switch for real.
L005| What changes for you
L006| The old make targets keep working until the end of the month. The
0.42 L007| entrypoint is a single command that wraps everything, including th
0.59 L008| docs build that used to be separate.
L009| bun run build
L010| Generated artifacts no longer need to be committed. The new pipeli
0.48 L011| uploads them to the registry automatically, and checking them in
0.40 L012| just creates merge conflicts.
L013| The cutover touches three teams, so check whether you are on this
0.50 L014| list before you plan anything for Monday:
0.22 L015| The platform team
0.11 L016| The web client team
0.12 L017| Whoever still owns the release tooling这些概率落在两个彼此分开的区间:会切断句子的换行得分在 0.39 及以上,而作者本意保留的换行得分接近零。但区间之间的分界点该划在哪里,取决于上一行是如何结尾的,这一信息代码可以直接读出来:
在一个 dangling 行(没有句末标点的行)之后,得分在 0.2 及以上就算作延续。这里真正的延续得分低到 0.39(
L004| make the switch for real.),所以一个统一的 0.5 阈值会把本该连成一段的段落拆散。在终止性标点(结束句子或分句的字符:
.!?:;)之后,阈值上升到 0.5。备忘录中的团队列表说明了原因:L015| The platform team紧跟在冒号后面,得分 0.22。这是一个较低但非零的"延续此句"信号,会通过 0.2 阈值并把列表合并到引入它的句子中。没有哪个单一阈值能同时兼顾这两种情形;只要代码先检查标点,两个区间就能分开。
为什么问的是"mid-sentence"而不是"same paragraph"
这个流水线的第一个版本问了一个最直观的问题:"这两行属于同一段落吗?"它以特定的方式失败了。标题下的一连串短行(一个不带项目符号的列表)在宽松意义上确实算一个段落:这些行排列在一起,共享一个主题。一旦用段落来问,模型会对每一对都回答 yes,stitch 阶段于是把整个列表合并成一个长块。
同一份文档、同一种请求结构,只是措辞发生了变化:
def naive_join_question(i: int) -> Noul:
return Noul(
instructions=f"Are lines {line_id(i - 1)} and {line_id(i)} part of the same paragraph?",
criteria=NoulCriteria(
true="The two lines belong to the same paragraph of running text",
false="The two lines belong to different paragraphs or different pieces of content",
),
)
naive = stitch("same-paragraph")
print(f"{'':14}{'mid-sentence':>13}{'same paragraph':>16}")
for i in (15, 16, 17, 20, 21):
print(f"{line_id(i)}{'':2}{LINES[i]['text'][:36]:<38}"
f"{result['joins'][i]:>7.2f}{naive['joins'][i]:>13.2f}")
print(f"\nblocks after merge: {len(blocks)} (mid-sentence) vs "
f"{len(merge(naive['joins']))} (same paragraph)") mid-sentence same paragraph
L015 The platform team 0.22 0.77
L016 The web client team 0.11 0.81
L017 Whoever still owns the release tooli 0.12 0.78
L020 Delete the old build cache directory 0.08 0.88
L021 Run the doctor script and fix anythi 0.05 0.91
blocks after merge: 17 (mid-sentence) vs 12 (same paragraph)换成段落的措辞后,每个无标记的列表项得分都高于 0.75,两个列表都被压扁了。备忘录合并成了几块连绵的长段。"Same paragraph" 让模型判断主题是否延续,而列表项之间主题确实是延续的。"Picks up mid-sentence" 问的是文本本身。当一个判断结果要喂给阈值时,问题应该点到那个起决定作用的最小事实。这里的措辞,就是 17 个块和 12 个块之间的差距。
置信度最低的块
uncertain = min(blocks, key=lambda b: b["confidence"])
print(f'"{uncertain["text"]}"')
print(f"confidence {uncertain['confidence']:.2f}: ", end="")
print(", ".join(f"{k} {v:.2f}" for k, v in
sorted(uncertain["probabilities"].items(), key=lambda kv: -kv[1])[:3]))"The cutover touches three teams, so check whether you are on this list before you plan anything for Monday:"
confidence 0.43: paragraph 0.53, list_item 0.24, callout 0.19引入团队列表的那句话本身确实是模棱两可的——它点出接下来会出现的内容(像标题),自身又是一个完整句子(像段落),所处的位置也正是 callout 通常出现的地方。概率因此分散开来(paragraph 0.53,list_item 0.24,callout 0.19),UI 可以把这种情形呈现出来——例如,把类型置信度(即获胜选项背后的概率)低于 0.55 的块加上下划线,提示人工复核。
本文档由 Mintlify 构建和托管。