結構還原
用兩次請求從丟失了格式的純文本重建 Markdown：一次把硬換行的行重新拼接，一次對每個塊分類（標題、列表、程式碼、提示框）。
這個 cookbook 拿到一段被剝掉標記的純文本（行在句子中間被硬換行，沒有標題標記，也沒有列表專案符號），把它重建為 Markdown 結構：標題、段落、列表、引用、程式碼、提示框。輸入正是一份處於這種狀態的團隊備忘。
文本生成模型可以把文本改寫成 Markdown，但改寫也可能改動文字。這裡模型從不生成文本：它只回答關於文件的狹窄問題（這一行是否在句子中間接續？這個塊是什麼內容？），渲染由程式碼完成，因此輸出的每個字元都來自輸入，每個判斷都帶有機率。
整條流水線對每份文件是兩次 API 請求，按順序執行：
- 第 1 遍，拼接： 對每一對相鄰的行提一個
Noul問題（yes/no 問題，答案是「yes 正確」的機率），問這次換行是否把一個句子拆到了兩行上。所有行對放在一個請求裡；繼續某個被拆開句子的行會被合併回塊。 - 第 2 遍，分類： 對每個合併後的塊提一個
Choice問題（從列表裡選一個選項，每個選項都有機率），在標題、段落、列表項、引用、程式碼或提示框（與正文區分開的 note、tip 或 warning）之間選擇。塊只有等第 1 遍回答後才會存在，所以這是第二次請求；它還攜帶每個塊的伴隨問題（標題層級、步驟順序、提示框種類），只有當塊的型別讓這些問題變得相關時才會讀取它們的答案。 - 直接證據留在程式碼裡。 空行和明確的標記（
-、1.、#）在程式碼裡讀取，絕不交給模型重新判斷；這份備忘保留了空行，但丟掉了所有標記。模型只會拿到程式碼無法從文本中回答的那些問題。
全部行為都寫在第 2 遍的問題 criteria 裡：三個「一行描述」的字典，外加 classify_questions 裡那一步問題的 true/false criteria。其餘程式碼都是圍繞它們的管道。成本與延遲資料見附錄：這份備忘用兩次往返、10,211 個 token、0.8 秒、$0.0015。
環境準備
pip install ipython 'cooksafe>=0.2.0,<0.3.0'然後設定 TYPESAFE_API_KEY。每次 API 呼叫都快取在隨 cookbook 一起提供的 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 拉取，這樣 cookbook 的數字保持可復現。
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 是普通文本，模型把它當作 state 的一部分讀取，問題和答案用這些 id 來指代行（與 語義搜尋 cookbook 同一套方案）。
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.Pass 2: classifying blocks
每個拼接後的塊拿到一個 Choice 問題：這是什麼內容？ 下面這三個字典，加上 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，需要知道是哪一種。型別此時還不知道，等型別出來意味著第三次往返，所以伴隨問題在同一請求裡就提前問了。這些答案大多數根本不會被讀取：段落的步驟機率毫無意義，直接忽略。多問一個問題增加的成本很小，因為 token 大部分來自 state、而且無論如何只發一次，而多一次往返則會多出整整一個請求的延遲。
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」的步驟機率接近 0.9（會渲染成有序列表），三行團隊行接近 0.1（渲染成專案符號），而那條沒有標記的、關於 doctor 指令碼的警告被歸類為種類為 warning 的提示框。附錄看了模型唯一拿不準的那個塊。
渲染
程式碼根據這些判斷拼出頁面。連續的列表項合併成一個列表，當各項步驟機率的均值至少為 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 裡開啟
這個分享連結包含拼接後的塊和完整的第 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})"))附錄
成本與延遲
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 的閾值，把列表並進引出它的那個句子裡。沒有哪個單一閾值對兩種情況都成立；只要程式碼先檢查標點，兩個帶就分開了。
為什麼問的是「句子中間」而不是「同一段」
這條流水線的第一版問了那個顯而易見的問題：「這兩行屬於同一段嗎？」它以某種具體的方式失敗了。標題下一串短行（一個沒打專案符號的列表）在寬泛意義確實是一段：這些行挨在一起，共享一個主題。一問到段落，模型對每一對都說「是」，拼接這一步就把整個列表併成一個長塊。
同一份文件，同樣的請求形狀，只改措辭：
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 以上，兩個列表都塌掉了。備忘併成幾個拖沓的長塊。「同一段」讓模型判斷主題是否延續，而在列表項之間主題確實延續。「句子中間接續」問的則是文本本身。當一個判斷會餵給閾值時，問題就該點明決定它的最狹窄的事實。這裡，措辭的差別就是 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 的塊加下劃線以便複核。