跳到主内容
创建时间：最后更新：作者：白川

处理不完整输出与校验失败

只接收完整且通过校验的结果，处理流式片段、截断和错误，看清 Agent 循环如何回填错误让模型重试，并在明确条件下做一次纠正。

处理不完整输出与校验失败

结构化输出的失败不只有 JSON 少一个括号。请求可能中止，回复可能被截断，也可能字段都正确却引用了不存在的源码。不同问题需要不同的处理，不能统一套一个无限重试循环。

本篇沿用前面建立的 connection.ts、lesson.ts 和 schema.ts。先定义最终接收入口，再把流式显示和有限纠正接到它上面。

失败分三层，处理位置不同

失败层表现谁处理
请求层
网络错误、认证失败、超时
Provider，表现为 stopReason: "error" / "aborted"
生成层
输出被截断、模型要工具
你检查 stopReason
契约层
JSON 语法错、字段不符、业务约束不成立
你的校验器

三层不能互相替代。输出被截断时，即使剩下的片段碰巧是合法 JSON，也不能接收——检查顺序必须是先状态、后解析、再校验。

只在完整结果上解析与校验

创建 src/recover-output.ts：

typescript
import { writeFile } from "node:fs/promises"
import type { AssistantMessage, Context } from "@earendil-works/pi-ai"
import { createConnection } from "./connection.js"
import { extractionPrompt } from "./lesson.js"
import { type FunctionAnalysis, ResultValidationError, validateAnalysis } from "./schema.js"

export class IncompleteOutputError extends Error {}

export function acceptFinalMessage(reply: AssistantMessage): FunctionAnalysis {
  if (reply.stopReason !== "stop") {
    throw new IncompleteOutputError(`本轮不是完整文本结果：${reply.stopReason}`)
  }
  const text = reply.content
    .filter((block) => block.type === "text")
    .map((block) => block.text)
    .join("")
  const value: unknown = JSON.parse(text)
  return validateAnalysis(value)
}

export async function runRecovery() {
  const { models, model } = createConnection()
  const context: Context = {
    messages: [{ role: "user", content: extractionPrompt, timestamp: Date.now() }],
  }
  // 整个任务共用时间上限，不因重试重新延长。
  const records: Array<{ attempt: number; stopReason: string; text: string; error?: string }> = []
  const signal = AbortSignal.timeout(30_000)
  for (let attempt = 0; attempt < 2; attempt += 1) {
    const reply = await models.completeSimple(model, context, { signal, maxTokens: 1024 })
    const record: (typeof records)[number] = {
      attempt: attempt + 1,
      stopReason: reply.stopReason,
      text: reply.content
        .filter((block) => block.type === "text")
        .map((block) => block.text)
        .join(""),
    }
    records.push(record)
    await writeFile("recovery-attempts.json", JSON.stringify(records, null, 2), "utf8")
    try {
      const result = acceptFinalMessage(reply)
      console.log(JSON.stringify(result, null, 2))
      return
    } catch (error) {
      record.error = error instanceof Error ? error.message : String(error)
      await writeFile("recovery-attempts.json", JSON.stringify(records, null, 2), "utf8")
      const correctable = error instanceof SyntaxError || error instanceof ResultValidationError
      // 无法用通用文本识别所有服务的拒绝，只有显式开关才允许一次纠正。
      if (
        !correctable ||
        signal.aborted ||
        attempt === 1 ||
        process.env.ALLOW_FORMAT_RETRY !== "true"
      ) {
        throw error
      }
      console.error("结果未通过检查，尝试一次格式纠正。")
      context.messages.push(reply)
      const instruction =
        error instanceof ResultValidationError && error.stage === "business"
          ? "上一份结果的源码引用不成立，请重新核对函数名和证据，不要编造。"
          : "上一份结果未通过 JSON 或 Schema 校验，请按原字段要求重新提交纯 JSON。"
      context.messages.push({ role: "user", content: instruction, timestamp: Date.now() })
    }
  }
  throw new Error("没有得到可接受结果")
}

入口分成三道检查：结束状态是否适合接收，文本能否解析，数据能否通过 Schema 和源码引用检查。任何一步失败，都不会返回可供业务使用的 FunctionAnalysis。

即使被截断的片段碰巧是一个合法 JSON，也不能绕过结束状态检查。相同逻辑适用于请求错误和主动取消，部分内容只用于诊断或预览。

三道检查各自拦什么

acceptFinalMessage() 里三道检查的职责不能互换：

检查拦下的失败抛出的异常
reply.stopReason !== "stop"
截断、错误、中止、待定稿、延迟
IncompleteOutputError
JSON.parse(text)
语法错误、Markdown 包裹
SyntaxError
validateAnalysis(value)
字段形状不对、业务引用不成立
ResultValidationError

第一道检查拦的是根本没拿到完整结果。它的判断依据是 stopReason，不是文本内容——被截断的文本可能恰好是一个合法的 JSON 片段，但那个片段是残缺的。

第二道检查拦的是拿到了完整文本但格式不对。JSON.parse() 是严格的：多一个逗号、多一段说明文字，都会抛 SyntaxError。这正是本篇不剥围栏、不用正则截取的原因。

第三道检查拦的是格式对但内容不可信。分两个阶段，ResultValidationError.stage 记录是哪一层：

typescript
export class ResultValidationError extends Error {
  constructor(
    readonly stage: "schema" | "business",
    message: string,
  ) {
    super(message)
  }
}

"schema" 阶段是字段形状问题（parameters 不是数组、多了一个 debug 字段）。"business" 阶段是引用关系问题（函数名对不上、证据不在源码里）。

只有 stage 决定了纠正指令的内容——业务错误要提示重新核对源码，格式错误要提示重新输出纯 JSON。用错提示会让第二次尝试重复同一个错误。

失败时模型的四种处境

模型请求失败时，回复消息的几种形态：

stopReasonerrorMessagecontent能作为结果吗
stop
无
文本
可以
length
无
部分文本
不可以
error
有
可能为空
不可以
aborted
有
部分文本
不可以

error 和 aborted 的区别是谁发起的。aborted 通常是 signal 被触发（超时或用户取消），error 是请求本身失败（认证、网络、服务端错误）。

aborted 时已经产生的 token 仍然计入用量。取消不等于免费。

有一个边界情况要注意：error 和 aborted 的回复里 content 可能有内容也可能为空。别假设"出错了就一定有错误文本"——errorMessage 才是原因所在。

流式片段只作预览

创建 src/stream-json.ts：

typescript
import { createConnection } from "./connection.js"
import { extractionPrompt } from "./lesson.js"
import { acceptFinalMessage } from "./recover-output.js"

const { models, model } = createConnection()
const stream = models.streamSimple(
  model,
  {
    messages: [{ role: "user", content: extractionPrompt, timestamp: Date.now() }],
  },
  { signal: AbortSignal.timeout(30_000), maxTokens: 1024 },
)
for await (const event of stream) {
  if (event.type === "text_delta") {
    // 片段只用于预览，不作为业务对象处理。
    process.stderr.write(event.delta)
  }
}
const analysis = acceptFinalMessage(await stream.result())
console.log(JSON.stringify(analysis, null, 2))

运行：

bash
node --env-file=.env --import tsx src/stream-json.ts

预览写到标准错误，最终接受的对象写到标准输出，避免两种内容混在一起。程序不对每个 text_delta 执行 JSON 解析，也不在收到第一个完整字段时触发业务动作。

如果服务支持原生流式结构化对象，可以在明确 SDK 契约后展示局部字段；本例消费的是文本流，不假定已经拥有这种能力。

两种重试回路不要混

Pi 提供了两条不同的重试路径，层次不同：

一、Agent 循环内置的回路——工具校验失败时自动重试。 这条不需要你写代码。模型提交工具参数，校验失败，运行时把错误作为工具结果回填，模型下一轮就能看到并纠正。

二、应用层的回路——文本结果校验失败时重发请求。 这条要你自己写，就是上面 runRecovery() 里的 for 循环。

区别在于：

Agent 循环内置应用层循环
触发条件
工具参数校验失败
文本结果解析或校验失败
反馈形式
结构化的错误工具结果
你自己写的纠正指令
重试次数
由循环决定
由你的预算决定
是否需要开关
不需要
本例用 ALLOW_FORMAT_RETRY 控制

如果结果是通过工具提交的，内置回路已经覆盖了大部分纠正场景，通常不需要再叠加应用层循环。

Agent 循环的自动重试有一个前提：模型必须看到错误。错误工具结果的 content 里带的是校验错误消息，格式是：

text
Validation failed for tool "add":
  - b: expected number, got string

Received arguments:
{
  "a": 1,
  "b": "x"
}

末尾的 Received arguments 让模型看到自己实际发的是什么，比一句"参数错误"有用得多。

只有明确允许时纠正一次

创建 src/recover-main.ts：

typescript
import { runRecovery } from "./recover-output.js"

await runRecovery()

默认运行：

bash
node --env-file=.env --import tsx src/recover-main.ts

默认失败就退出。对于这份确定的普通代码练习，如果你确认只是格式或字段错误，可以在 .env 中设置：

dotenv
ALLOW_FORMAT_RETRY=true

再次运行时最多发出两次应用层请求，两次共用一个 30 秒信号。底层 SDK 自身的重试行为仍需另外核验，不能把这里的循环次数当成全部网络请求次数。

纠正消息只要求修正结果，不要求模型突破拒绝或忽略约束。通用文本入口无法可靠区分所有 Provider 的拒绝回复和格式错误，因此默认不自动纠正；启用开关只适用于已经确认的练习输入。真实业务若有明确拒绝状态，应先单独处理。

失败信息怎样保留

每次回复和错误原因写入 recovery-attempts.json，保留次数、结束原因和文本，便于确认第二次请求是否修正了同一问题。该文件已加入忽略列表，示例不写入 Key 或认证头。

记录可能含任务材料和模型内容，不应直接作为公开日志。单次练习脚本覆盖自己的记录，不提供并发写入、日志轮换或生产存储能力。

源码引用失败时，程序要求重新核对函数名和证据；语法或 Schema 错误时，要求重新提供规定字段。它不会把一份错误对象悄悄补全为成功结果，也不会在失败时返回上一次缓存的有效答案。

诊断信息在消息上

流式失败时，AssistantMessage.diagnostics 数组里可能有结构化诊断。它的类型是脱敏后的诊断记录数组：

typescript
interface AssistantMessage {
	// ...
	diagnostics?: AssistantMessageDiagnostic[]; // Redacted provider/runtime diagnostics for failures and recoveries.
	// ...
}

传输层的失败诊断带 phase 字段，取值是 before_message_stream_start 或 after_message_stream_start——这决定了能不能自动降级重试。详见处理模型回复与流式输出。

排查时可以看三样东西：

位置内容
reply.stopReason
本轮结局
reply.errorMessage
错误原因文本
reply.diagnostics
结构化诊断（含传输阶段）
reply.usage
已消耗的用量（取消也可能已产生）

取消不代表账单归零。已经产生的 Token 仍会计入用量。

检查失败出口

准备六类情况：合法 JSON、字段类型错误、无效源码证据、Markdown 包裹文本、截断和取消。前者可以接受，其余应该在对应阶段拒绝。开启纠正后，第二份仍无效也必须结束。

本轮对最终消息校验、工具提交和错误状态做了模拟验证；真实模型纠正效果、真实网络取消与服务拒绝行为没有联调。结构化输出的价值在于建立可检查的接收边界，不能保证模型的每个业务结论都正确。

参考资料