Skip to main content

结构恢复

机器翻译译自英文(docs.typesafe.ai)本页 39 段里 6 段还没译

文档索引

在以下地址获取完整的文档索引:https://docs.typesafe.ai/llms.txt 使用此文件可以在进一步探索之前发现所有可用页面。

结构恢复

通过两次请求从丢失格式的纯文本中重建 Markdown:一次将硬换行行拼接回去,另一次为每个块(标题、列表、代码、提示框)分类,并附带仅在相关时才读取的伴随问题。

本 cookbook 接收格式标记已被剥离的纯文本(行在句子中间硬换行,没有标题标记,没有列表项目符号)并将其结构重建为 Markdown:标题、段落、列表、引用、代码、提示框。输入正是处于此种状态的一份团队备忘录。

文本生成模型可以将文本重写为 Markdown,但重写也会改变文字。这里模型从不生成文本:它回答有关文档的狭窄问题(这行是否从句子中间接续?这个块是什么类型的内容?),而代码负责渲染,因此输出的每个字符都来自输入,且每个判断都带有概率。

整个流水线针对每个文档包含两个 API 请求,按顺序执行:

  • 第一遍,拼接: 针对相邻行对,每个对应一个 Noul 问题(一个是非问题,其答案为“是”正确的概率),询问换行是否将句子拆分到两行。所有对在同一请求中发送,继续被拆分句子的行将合并回块。
  • 第二遍,分类: 针对每个合并后的块,对应一个 Choice 问题(从列表中选择一个选项,每个选项附带概率),在标题、段落、列表项、引用、代码或提示框(与正文区分开的笔记、提示或警告)中选择。块仅在第一遍回答后存在,因此这是第二个请求;它还携带每个块的伴随问题(标题级别、步骤顺序、提示框类型),仅在块的类型使其相关时才读取这些答案。
  • 直接证据保留在代码中。 空行和显式标记(- , 1. , #)在代码中读取,绝不发送给模型重新考虑;此备忘录保留了空行但丢失了所有标记。模型仅接收代码无法从文本中回答的问题。

所有行为均由第 2 次传递的问题标准指定:三个包含单行描述字典,加上 classify_questions 中步骤问题的真/假标准。其余代码只是围绕它们的水管工作。成本和延迟数字在附录中:两次往返,10,211 个 token,0.8 秒,此备忘录成本 $0.0015。

Setup

pip install ipython "typesafe-sdk>=0.5.7" cooksafe --extra-index-url https://pypi.typesafe.ai/

然后设置 TYPESAFE_API_KEY。所有 API 调用都缓存在 json_cache.json 中,该文件随 cookbook 一起提供,因此重新渲染时会重播已发布数字,而无需调用 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 是模型作为状态一部分读取的普通文本,问题和答案通过 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.

第二遍:分类块

每个拼接块都获得一个 Choice 问题:*这是哪种类型的内容?*下面 classify_questions 中的这三个字典以及步骤问题的真/假标准,构成了分类器的完整规范。没有其他逻辑。要将管道适应为你自己的文档,请编辑这些描述。

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 且无论何种情况仅发送一次,而额外的往返则会增加完整的请求延迟。

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.

表格中记录了每个块的判断结果,旁边的伴生列展示了被投入使用的初步答案:三行“周一之前要做的事”的步骤概率接近 0.9(将渲染为有序列表),三行团队成员的概率接近 0.1(无序列表),而未标记的关于 doctor 脚本的警告被分类为 warning 类型的提示框。附录探讨了模型不确定的那个块。

渲染

Code assembles the page from the judgments. Consecutive list items become one list, numbered when the mean of the items' step probabilities is at least 0.5. That threshold is a group-level decision no single question asked directly.

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.

上述每个单词均来自输入内容。流水线仅选择了边界、类型和标记。

Open it in the playground

此分享链接包含拼接后的块和完整的第二遍问题集。打开它以实时重新运行分类。

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})"))

Open the stitched memo + questions in the TypeSafe playground →


附录

成本与延迟

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。

合并阈值的来源

第一遍的每行拼接概率:

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 的切割点并将列表合并到引入它的句子中。没有单一的阈值适用于两种情况;一旦代码首先检查标点符号,两个区间便分离开来。

Why the question is "mid-sentence" and not "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,两个列表都会塌陷。备忘录合并为几个冗长的块。“同一段落”要求模型判断主题是否延续,而在列表项之间,主题确实延续。“从句子中间接续”询问的是文本本身。当判断馈入阈值时,问题应指出决定该判断的最狭窄事实。这里的措辞导致了 17 个块与 12 个块之间的差异。

The lowest-confidence block

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

引入团队列表的句子确实存在歧义——它命名了以下内容(像标题),是一个完整的句子(像段落),并且位于提示框应该出现的位置。概率相应分布(段落 0.53,列表项 0.24,提示框 0.19),用户界面可以展示这一点——例如,对类型置信度(获胜选择背后的概率)低于 0.55 的任何块进行下划线以进行审阅。