结构还原
用两次请求从丢失了格式的纯文本中重建 Markdown:一次把硬换行的行重新拼接起来,一次对每个块(标题、列表、代码、提示框)进行分类,并配上只有在相关时才会被读取的伴随问题。
本实践手册拿到的是一段标记已被剥离的纯文本(行在句子中间被硬换行,没有标题标记,没有列表符号),并把它还原成 Markdown 结构:标题、段落、列表、引用、代码、提示框。输入是一份正处于这种状态的团队备忘录。
文本生成模型可以把文本改写为 Markdown,但改写也可能改变用词。在这里,模型从不生成文本:它只回答关于文档的狭窄问题(这一行是否承接了上一句?这个块属于什么内容?),渲染由代码完成,因此输出的每一个字符都来自输入,而每一个判断都带有概率。
整条流水线对每份文档依次执行两次 API 请求:
- 第 1 遍,拼接: 对每一对相邻的行提出一个
Noul问题(一种是/否问题,其答案就是「是」正确的概率),询问这个换行是否把一句话拆到了两行。所有行对都放进一次请求里,承接被拆句子的行会被重新合并成块。 - 第 2 遍,分类: 对每个合并后的块提出一个
Choice问题(从列表中挑选一个选项,每个选项都有对应概率),在标题、段落、列表项、引用、代码或提示框(与正文区分开的 note、tip 或 warning)之间做出选择。这些块只有在第 1 遍给出答案之后才存在,所以这是第二次请求;它还为每个块携带了伴随问题(标题层级、步骤顺序、提示框类型),只有当块的类型使这些问题变得相关时才会读取它们的答案。 - 直接证据留在代码里。 空行和显式标记(
-、1.、#)都由代码读取,从不送给模型去重新考虑;这份备忘录保留了空行,但丢失了所有标记。模型只会拿到代码无法从文本中回答的那些问题。
全部行为都规定在第 2 遍的问题 criteria 中:三个由单行描述构成的 dict,加上 classify_questions 里步骤问题的 true/false criteria。其余代码都只是围绕它们的管道。成本与延迟数字在附录中:这份备忘录需要两次往返、10,211 个 token、0.8 秒、$0.0015。
环境准备
pip install ipython "typesafe-sdk>=0.5.7" cooksafe --extra-index-url https://pypi.typesafe.ai/
然后设置 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 tooling
第 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.
第 2 遍:对块进行分类
每个拼接后的块都会得到一个 Choice 问题:这是什么类型的内容? 下面这三个 dict,加上 classify_questions 中步骤问题的 true/false criteria,就是分类器的全部规范。没有其他逻辑。要把这条流水线适配到你自己的文档,修改这些描述即可。
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",
}
下面的一切都是管道:构造问题、发送一次请求、把答案读回来。如果返回的类型是 heading,渲染器需要一个标题层级;如果是 list_item,需要知道顺序是否重要;如果是 callout,需要知道是哪一种。这些类型此时还未知,而等待它们就意味着第三次往返,因此伴随问题会在同一次请求中提前问出。这些答案大多数永远不会被读取:一个段落的 step 概率毫无意义,直接忽略即可。额外的问题带来的开销很小,因为状态占了大部分 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(它们会渲染成有序列表),三条团队名称所在行的值接近 0.1(渲染成无序列表),而那条关于 doctor 脚本的未标记警告被归类为 kind 为 warning 的 callout。附录会考察模型唯一不确定的那个块。
渲染
代码根据这些判断组装页面。连续的列表项会合并成一个列表,当各项 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.
上面的每一个词都来自输入。流水线只选择了边界、类型和标记。
在演练场中打开
这个分享链接包含拼接后的块以及完整的第 2 遍问题集。打开即可实时重跑分类。
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 演练场中打开拼接后的备忘录 + 问题 →
附录
成本与延迟
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.8 秒、$0.0015。
连接阈值的来源
第 1 遍得到的逐行连接概率:
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 及以上,而作者本意的换行得分接近零。但阈值放在两个区间之间的哪个位置,取决于上一行如何结尾,这是代码可以直接读到的事实:
- 在一个悬空行(没有任何句末标点的行)之后,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」
这条流水线的第一个版本问的是那个显而易见的问题:「这两行属于同一段吗?」它以某种特定的方式失败了。一个标题下面的一串短行(一个打成没有符号的列表)在宽泛意义上确实是一段:这些行挨在一起、共享同一个主题。当问的是段落时,模型对每一对都回答「是」,于是拼接这一遍把整个列表合并成一个很长的块。
同一份文档、同样的请求结构,只是措辞变了:
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
引入团队列表的这个句子确实有歧义——它点出了后文的内容(像标题),本身是一个完整的句子(像段落),又处在通常会放提示框的位置。概率也相应地分散开来(paragraph 0.53、list_item 0.24、callout 0.19),UI 可以把这一点呈现出来——例如,把类型置信度(胜出选项背后的概率)低于 0.55 的块加下划线以供复审。