别再相信“合法 JSON”：用 Schema、自动修复与安全写表搭建可靠的数据提取流水线

批量处理销售沟通记录时，最让人崩溃的往往不是模型直接报错，而是它看起来成功了。

前几十条数据正常写入，第 37 条突然少了 phone，后面的列开始错位；有些手机号被当成数字，导出 Excel 后变成科学计数法；还有一条把意向等级写成“比较感兴趣”，程序没报错，统计报表却再也匹配不到 high。

问题在于：

大模型返回了合法 JSON，不代表它返回了可写入业务系统的数据。

很多结构化提取项目，只做了 json.loads()。它能证明括号和引号基本正确，却无法证明字段完整、类型准确，更无法判断数据是否符合业务要求。

真正可靠的方案，不是再加一句“请严格返回 JSON”，而是建立一条完整流水线：

可写入数据 = JSON 语法正确 + 字段结构正确 + 业务规则通过

本文用“从销售沟通记录中提取客户线索并写入表格”作为贯穿案例，搭建一套包含 Schema 校验、本地修复、模型纠错、有限重试、失败隔离和安全写表 的工程闭环。

---

一、JSON 看起来正确，为什么还是不能用？

先看一段经过脱敏和改写的测试文本：

3/7 沟通记录：

王经理说他们公司内部一般简称“星海科技”，正式名称暂时没确认。
他对智能客服方案比较感兴趣，希望下周再演示一次，但没有留下手机号。

聊天中还提到采购负责人李女士，号码可能在同事通讯录里，目前无法确认。
上次联系时间有人记成 2026/03/05，也有人写 3月5日。

这段文本故意包含了结构化提取中最常见的麻烦：

没有明确手机号；
公司只有简称；
意向等级使用自然语言；
同时出现王经理和李女士；
日期格式不统一；
部分信息无法确认。

即使提示词写得很严，模型仍可能返回以下结果。

失败一：缺少业务必填字段

{
"company": "星海科技",
"phone": null,
"intent_level": "high",
"confidence": 0.82
}

这是合法 JSON，但缺少 name。如果程序直接使用 data["name"]，任务会中断；如果使用 data.get("name")，表格里则会悄悄出现空值。

失败二：字段类型错误

{
"name": "王经理",
"company": "星海科技",
"phone": 13800138000,
"intent_level": "high",
"confidence": "0.82",
"missing_fields": "phone"
}

手机号被写成数字，confidence 变成字符串，missing_fields 也没有返回数组。

其中最危险的是手机号。手机号、身份证号、订单号看起来像数字，本质上却是标识符，不应该参与数学运算，更不应该被 Excel 转成科学计数法。

失败三：枚举值不合法

{
"name": "王经理",
"company": "星海科技",
"phone": null,
"intent_level": "比较感兴趣",
"confidence": 0.82,
"missing_fields": ["phone"]
}

这同样能被解析，但业务系统只接受：

high / medium / low / unknown

如果不做枚举校验，“比较感兴趣”“高意向”“重点跟进”可能同时出现在一列里，后续筛选、统计和自动化规则都会失效。

失败四：JSON 外混入解释文字

根据沟通记录，客户具有较高意向，提取结果如下：

json
{
"name": "王经理",
"company": "星海科技",
"phone": null,
"intent_level": "high",
"confidence": 0.82,
"missing_fields": ["phone"]
}

人看起来毫无问题，json.loads() 却会直接失败。

因此，结构化数据至少要经过三道门：

1. 语法层：是不是合法 JSON；

2. 结构层：字段、类型、枚举是否符合 Schema；

3. 业务层：手机号、日期、必填信息和多联系人冲突是否合理。

---

二、先定义 Schema：把需求变成机器可执行的合同

Schema 不是一段写给模型看的“格式建议”，而是一份可以被程序强制执行的合同。

设计 Schema 时，首先要区分两件事：

业务必须有的字段：例如线索必须明确对应一个联系人；
原文可能没有的字段：例如客户没有留下手机号。

不能因为业务希望收集手机号，就让模型在原文没有手机号时硬填一个值。正确做法是允许 phone 为 null，同时在 missing_fields 中记录缺失。

JSON Schema 示例

{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"additionalProperties": false,
"required": [
"name",
"intent_level",
"confidence",
"missing_fields"
],
"properties": {
"name": {
"type": "string",
"minLength": 1
},
"company": {
"type": ["string", "null"]
},
"phone": {
"type": ["string", "null"],
"pattern": "^\\+?[0-9\\- ]{6,20}$"
},
"intent_level": {
"type": "string",
"enum": ["high", "medium", "low", "unknown"]
},
"contact_date": {
"type": ["string", "null"],
"format": "date"
},
"source_text": {
"type": ["string", "null"],
"default": null
},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 1
},
"missing_fields": {
"type": "array",
"items": {
"type": "string"
},
"default": []
}
}
}

这里有几个值得注意的细节：

additionalProperties: false：禁止模型擅自增加字段；
phone 使用字符串：避免前导零丢失和科学计数法；
intent_level 使用枚举：防止自然语言污染分类；
confidence 限制在 0 到 1；
company、phone 和 contact_date 可以为 null；
name 是业务必填字段，无法确认时不能进入正式表。

使用 Pydantic 做 Python 校验

如果项目使用 Python，Pydantic 更适合直接接入业务代码。

import re
from datetime import date
from typing import Literal

from pydantic import (
BaseModel,
ConfigDict,
Field,
field_validator,
)

class Lead(BaseModel):
model_config = ConfigDict(extra="forbid")

name: str = Field(min_length=1)
company: str | None = None
phone: str | None = None
intent_level: Literal["high", "medium", "low", "unknown"]
contact_date: date | None = None
source_text: str | None = None
confidence: float = Field(ge=0, le=1)
missing_fields: list[str] = Field(default_factory=list)

@field_validator("phone")
@classmethod
def validate_phone(cls, value):
if value is None:
return value

value = value.strip()

if not re.fullmatch(r"\+?[0-9\- ]{6,20}", value):
raise ValueError("手机号格式异常")

return value

不要把列表默认值直接写成：

missing_fields: list[str] = []

部分框架和旧式写法可能带来可变默认值共享问题。生产代码建议统一使用：

Field(default_factory=list)

另外，Schema 只能说明“一个字段是否像日期”，却不一定知道日期是否符合业务逻辑。例如联系日期不能晚于当前日期、合同结束日期不能早于开始日期，这些规则仍应放在业务校验层。

---

三、失败后不要立刻重跑：先做分层自动修复

Schema 能发现问题，但不会自动解决问题。

如果每次验证失败都重新执行完整提取，不仅浪费调用成本，还可能得到另一个完全不同的错误。更合理的策略是把修复分成三层。

第一层：本地确定性修复

不需要模型参与的问题，尽量在本地处理：

去掉 Markdown 代码围栏；
删除 JSON 前后的解释文字；
清理尾随逗号；
将 mobile、telephone 映射为 phone；
将 "0.82" 安全转换为 0.82；
将“高意向”映射为 high；
将可确定的日期格式统一为 YYYY-MM-DD。

本地修复必须遵守两个原则：

1. 只做确定性转换；

2. 不补充原文不存在的信息。

例如，把 "高意向" 映射为 "high" 是标准化；生成一个原文没有的手机号，则是编造。

第二层：让模型根据错误定向修复

如果字段缺失、类型冲突或结构复杂，可以把以下内容一起交给模型：

原始文本；
上一次输出；
Schema；
明确的校验错误。

可直接使用下面这份修复 Prompt：

你是一个结构化数据修复器。

请根据原始文本和校验错误，修复上一次模型输出。

必须遵守以下规则：

1. 只能使用原始文本中明确存在的信息。
2. 不得猜测、推断或编造缺失字段。
3. 无法确定的信息必须使用 null；意向等级无法确定时使用 unknown。
4. 只能返回一个合法 JSON，不要返回 Markdown 代码围栏或解释文字。
5. 返回结果必须严格遵守给定 Schema。
6. 只根据校验错误进行定向修正。
7. 不得修改已经正确且有原文依据的字段。
8. 如果原文出现多个联系人且无法确认目标联系人，不得擅自合并。

原始文本：
{source_text}

上一次模型输出：
{raw_output}

Schema：
{schema}

校验错误：
{validation_errors}

第三层：重新执行原始提取

只有在输出严重截断、任务理解错误，或者修复结果仍不可用时，才重新执行完整提取。

而且每一次修复之后，都必须重新经过：

JSON 解析 → Schema 校验 → 业务规则校验

不能因为输出来自“修复模型”，就默认它一定正确。

---

四、有限重试与失败隔离：不要让任务陷入死循环

重试并不等于把同一个请求原样再发一次。

下一次请求应该携带明确错误、降低随机性，并根据错误类型选择不同策略。

| 错误类型 | 是否重试 | 处理方式 | |---|---:|---| | JSON 混入代码围栏 | 否 | 本地清洗 | | 字段别名不一致 | 视情况 | 仅做白名单映射 | | 必填字段缺失但原文存在 | 是 | 携带错误定向修复 | | 原文没有业务必填信息 | 否 | 标记失败或转人工 | | 429 限流 | 是 | 指数退避并加入随机抖动 | | 401 或鉴权失败 | 否 | 检查密钥和接口配置 | | 上下文被截断 | 是 | 缩短输入或拆分任务 | | 连续多次 Schema 失败 | 否 | 进入失败队列 |

比较稳妥的上限通常是 2—3 次尝试。这不是成功率承诺，而是防止单条异常数据无限消耗资源的工程边界。

网络错误可以使用指数退避：

delay = min(base_delay  (2 * attempt), max_delay) \
+ random.uniform(0, 1)

随机抖动很重要。如果一批任务同时遭遇限流，又同时在固定的 2 秒后重试，就像一群人同时冲向刚打开的闸门，很可能再次拥堵。

完整流程

原始文本
↓
模型提取
↓
JSON 解析 ──失败──→ 本地清洗 / 模型修复
↓
Schema 校验 ─失败→ 携带错误信息定向修复
↓
业务规则校验 ─失败→ 重试或转人工
↓
生成任务 ID / 幂等键
↓
写入暂存表
↓
写入正式表格
↘ 失败队列

生产环境还应持续记录以下指标：

首次通过率；
自动修复成功率；
平均重试次数；
最终失败率；
单条数据处理成本；
各字段缺失率；
重复写入率。

如果没有真实测试结果，就不要写“通过率达到 99.99%”。更可靠的做法，是准备 50—100 条真实脱敏文本，用同一批数据比较不同模型、Prompt 和修复策略。

---

五、完整实战：从模型调用到稳定写入 Excel

下面给出一个可扩展的 Python 核心示例。安装依赖：

pip install pydantic requests openpyxl

模型调用通过环境变量配置：

export MODEL_URL="完整的模型接口地址"
export MODEL_API_KEY="你的密钥"
export MODEL_NAME="模型名称"

不同服务的响应格式可能略有差异，必要时只需修改 call_model()。

import json
import os
import random
import re
import time
import uuid

import requests
from openpyxl import Workbook, load_workbook
from pydantic import ValidationError

SCHEMA = Lead.model_json_schema()

FIELD_ALIASES = {
"mobile": "phone",
"telephone": "phone",
"customer_name": "name",
"intent": "intent_level",
}

INTENT_ALIASES = {
"高意向": "high",
"比较感兴趣": "high",
"一般": "medium",
"低意向": "low",
"无法判断": "unknown",
}

def call_model(prompt: str) -> str:
response = requests.post(
os.environ["MODEL_URL"],
headers={
"Authorization": f"Bearer {os.environ['MODEL_API_KEY']}",
"Content-Type": "application/json",
},
json={
"model": os.environ["MODEL_NAME"],
"messages": [{"role": "user", "content": prompt}],
"temperature": 0,
},
timeout=60,
)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]

def extract_with_model(source_text: str) -> str:
prompt = f"""
从以下销售沟通记录中提取一个主要联系人。

只能依据原文，不得猜测缺失信息。
无法确定的字段使用 null，意向无法判断时使用 unknown。
如果存在多个联系人且无法确认主要联系人，name 返回空字符串，
交由业务校验转人工处理。
只能返回 JSON。

Schema：
{json.dumps(SCHEMA, ensure_ascii=False)}

原始文本：
{source_text}
"""
return call_model(prompt)

def repair_locally(raw_output: str) -> str:
text = raw_output.strip()

fenced = re.search(
r"

(?:json)?\s(.?)\s*``

",
text,
flags=re.DOTALL | re.IGNORECASE,
)
if fenced:
text = fenced.group(1)

start = text.find("{")
end = text.rfind("}")
if start >= 0 and end > start:
text = text[start:end + 1]

text = re.sub(r",\s*([}\]])", r"\1", text)
return text.strip()

def parse_json_safely(raw_output: str) -> dict:
cleaned = repair_locally(raw_output)
data = json.loads(cleaned)

if not isinstance(data, dict):
raise ValueError("模型结果必须是 JSON 对象")

normalized = {}

for key, value in data.items():
normalized[FIELD_ALIASES.get(key, key)] = value

if isinstance(normalized.get("confidence"), str):
normalized["confidence"] = float(normalized["confidence"])

intent = normalized.get("intent_level")
if intent in INTENT_ALIASES:
normalized["intent_level"] = INTENT_ALIASES[intent]

if isinstance(normalized.get("missing_fields"), str):
normalized["missing_fields"] = [
normalized["missing_fields"]
]

if normalized.get("phone") is not None:
normalized["phone"] = str(normalized["phone"]).strip()

return normalized

def validate_business_rules(lead: Lead, source_text: str) -> None:
if not lead.name.strip():
raise ValueError("无法确认主要联系人")

if lead.phone is None and "phone" not in lead.missing_fields:
raise ValueError("phone 为 null 时，missing_fields 应包含 phone")

if lead.phone and lead.phone not in source_text:
raise ValueError("手机号缺少可追溯的原文依据")

def validate_with_schema(data: dict, source_text: str) -> Lead:
lead = Lead.model_validate(data)
validate_business_rules(lead, source_text)
return lead

def repair_with_model(
source_text: str,
raw_output: str,
validation_errors: str,
) -> str:
prompt = f"""
你是一个结构化数据修复器。

只能根据原始文本修复，不得猜测或编造。
无法确定的信息使用 null，意向无法确定时使用 unknown。
只能返回 JSON，严格遵守 Schema。
根据校验错误定向修正。
不得修改已经正确且有原文依据的字段。

原始文本：
{source_text}

上一次模型输出：
{raw_output}

Schema：
{json.dumps(SCHEMA, ensure_ascii=False)}

校验错误：
{validation_errors}
"""
return call_model(prompt)

def should_retry(error: Exception, attempt: int, max_attempts: int) -> bool:
if attempt + 1 >= max_attempts:
return False

if isinstance(error, requests.HTTPError):
status = error.response.status_code
return status == 429 or status >= 500

if isinstance(error, requests.RequestException):
return True

return isinstance(
error,
(json.JSONDecodeError, ValidationError, ValueError),
)

def write_to_sheet(
lead: Lead | None,
task_id: str,
source_text: str,
error: str | None = None,
filename: str = "leads.xlsx",
) -> None:
if os.path.exists(filename):
workbook = load_workbook(filename)
else:
workbook = Workbook()
workbook.active.title = "正式数据"
workbook.create_sheet("失败记录")

workbook["正式数据"].append([
"task_id", "name", "company", "phone",
"intent_level", "contact_date", "confidence"
])
workbook["失败记录"].append([
"task_id", "source_text", "error"
])

if lead:
sheet = workbook["正式数据"]

existing_ids = {
row[0].value for row in sheet.iter_rows(min_row=2)
}
if task_id not in existing_ids:
sheet.append([
task_id,
lead.name,
lead.company,
lead.phone,
lead.intent_level,
lead.contact_date.isoformat()
if lead.contact_date else None,
lead.confidence,
])

# 将手机号列设为文本格式
sheet.cell(sheet.max_row, 4).number_format = "@"
else:
workbook["失败记录"].append([
task_id,
source_text,
error,
])

workbook.save(filename)

def process_one(source_text: str, max_attempts: int = 3):
task_id = str(uuid.uuid4())
raw_output = ""
last_error = None

for attempt in range(max_attempts):
try:
if attempt == 0:
raw_output = extract_with_model(source_text)
else:
raw_output = repair_with_model(
source_text,
raw_output,
str(last_error),
)

data = parse_json_safely(raw_output)
lead = validate_with_schema(data, source_text)

write_to_sheet(
lead=lead,
task_id=task_id,
source_text=source_text,
)

print(
f"[成功] task_id={task_id}, "
f"attempt={attempt + 1}"
)
return lead

except Exception as error:
last_error = error

print(
f"[失败] task_id={task_id}, "
f"attempt={attempt + 1}, error={error}"
)

if not should_retry(error, attempt, max_attempts):
break

if isinstance(error, requests.RequestException):
base_delay = 1
max_delay = 20
delay = min(
base_delay  (2 * attempt),
max_delay,
) + random.uniform(0, 1)
time.sleep(delay)

write_to_sheet(
lead=None,
task_id=task_id,
source_text=source_text,
error=str(last_error),
)
return None

示例日志可能呈现为：

text
[失败] task_id=7f... attempt=1, error=phone 为 null 时，missing_fields 应包含 phone
[成功] task_id=7f... attempt=2



这只是运行格式示例，不代表任何生产通过率。真正评估系统时，要保留每一次原始响应、修复响应和校验错误。

这套流程也不绑定某一家模型服务。你可以把调用封装成统一的 call_model()，再比较不同模型的首次通过率、修复成功率和成本。



如果希望通过统一接口调用和切换模型，可以前往 api.884819.xyz 查看可用接口，并把地址与密钥放进环境变量，避免直接写入代码。

---

六、写入表格之前，还有四道保险

1. 先写暂存表，再进入正式表

校验通过不等于立刻覆盖业务数据。批量任务可以先写入暂存表，经过查重或抽样检查后，再同步到正式表。

2. 失败记录单独保存

失败数据不能直接丢弃，至少应保留：

task_id；


原始文本；
模型原始输出；
校验错误；
尝试次数；
最后一次处理时间。


数据量较小时，可以放进“失败记录”工作表；规模较大时，应进入数据库或死信队列。

3. 给每条任务分配唯一 ID

接口超时只代表客户端没有及时收到响应，不代表服务端一定没有处理成功。如果程序直接重试并再次写表，就可能产生重复数据。

示例代码使用 task_id` 做了最基础的重复检查。生产环境还可以根据来源系统 ID、文本摘要和业务主键生成幂等键。

4. 敏感信息必须脱敏

销售记录可能包含手机号、姓名、公司内部信息。日志、测试数据和模型请求都应遵守所在组织的数据安全要求，避免把完整敏感信息写进公开日志。

---

七、今天就能落地的最低版本

不必第一天就搭建复杂的消息队列和监控系统。一个最低可用版本只需要完成五件事：

1. 定义明确的 Schema；

2. 校验所有模型返回结果；

3. 失败时携带错误信息定向修复一次；

4. 修复仍失败就进入单独记录表；

5. 只有通过 Schema 和业务校验的数据才能写入正式表。

稳定不是永远不失败，而是每次失败都能被发现、分类、恢复或隔离。

想复现本文流程，可以先到 api.884819.xyz 配置模型接口，再用自己的 50—100 条真实脱敏数据测试。不要只看某一次输出是否漂亮，真正值得比较的是首次通过率、最终通过率、平均重试次数和单条成本。

8848AI平台使用用户名和密码即可注册，不需要邮箱验证；没有月租和订阅，按量付费，国产模型如 Deepseek、千问等可免费使用。平台内置 AI 对话功能，注册后可以直接体验。

新用户注册即送体验token。

下一篇，我们会处理这条流水线中更隐蔽的风险：接口超时并不代表任务失败。如果模型已经处理成功，而程序再次重试，就可能写入两条一模一样的数据。届时将继续拆解如何用任务 ID、幂等键、断点续跑和失败队列，避免批量 AI 任务重复写入与中途丢数据。

本文由8848AI原创，转载请注明出处。关注8848AI，带你从零开始学AI。

#AI教程 #结构化数据 #JSONSchema #Pydantic #Python #大模型应用 #数据提取 #8848AI