前言

很多开发者在接入大模型时,都会经历一个尴尬的阶段:本地 Demo 跑得挺顺,Prompt 里随手写一句"请返回 JSON",模型也乖乖听话,你对着控制台打印的对象心满意足。可一旦真正上了生产环境,问题就开始冒头。

有时它会在 JSON 前面礼貌地加一句"好的,以下是您要的结果";有时会悄悄漏掉一个必填字段,导致后端 DTO 反序列化直接抛异常;还有时,它会一本正经地补出一个你的业务系统根本不认识的枚举值。

你可能会怀疑是模型不够聪明,于是反复打磨提示词,加更多强调:“务必只返回 JSON,不要任何多余文字!”。但你会发现,这种"强调"在统计意义上永远无法做到 100% 可靠。

问题的根源不在于模型蠢,而在于我们犯了一个根本性的认知错误——我们把"自然语言承诺"错当成了"工程契约"

这篇文章就来深度拆解:如何把大模型这种本质上充满随机性的文本输出,收敛成稳定、可校验、可审计的工程数据结构。

一、背景或问题:为什么"请返回 JSON"不可靠?

1.1 自然语言不是契约

在模型的视角里,"请返回 JSON"只是它接收到的众多自然语言指令中的一条建议。模型在生成下一个 token 时,是在整个词汇表上计算概率分布,它"理解"的是语言的统计规律,而不是一份带法律效力的数据合同。

本质上,大模型的输出是概率采样,而不是契约执行

哪怕你在提示词里写得再严肃、再加感叹号,模型仍然存在一个非零概率去"违反"它——因为它从来没有真正"承诺"过什么。

1.2 单纯依赖 Prompt 的五类翻车点

一旦你只靠 Prompt 来约束输出,在生产环境几乎必然会踩到下面这五类坑:

翻车类型现象后果
格式漂移多轮对话或流式输出中,模型在 JSON 外带出解释性废话,如"好的,结果如下:"json.loads() 直接抛 JSONDecodeError,服务崩溃
字段缺失模型对某个信息没把握,干脆把它省略后端 DTO 缺少必填字段,反序列化失败
类型错误本应是布尔值或数字的字段,被返回成字符串 "true""123"静默的逻辑 bug,运行时才暴露
幻觉枚举模型编造了一个逻辑上合理、但业务系统根本不存在的状态值,如订单状态返回 "已发货" 而系统只有 "shipped"枚举校验失败或走入错误的分支
不稳定性用户输入模糊,或遭遇对抗性指令(Prompt 注入),结构化格式彻底崩掉安全风险,甚至被恶意利用

这五类问题的共同点是:它们不是"偶发 bug",而是概率输出的必然伴生现象。你不能用"把提示词写得更好"来彻底消灭它们,只能用工程手段把它们的概率压到足够低,并在它们发生时优雅地兜住。

二、核心思路:从"提示词约束"到"模型能力 + 工程契约"

要实现稳定的结构化输出,必须建立两层认知:

  1. 模型层:不能仅靠提示词,要结合模型供应商提供的 API 能力(JSON Mode / Structured Outputs / Function Calling),让约束下沉到解码阶段。
  2. 工程层:把大模型当作一个"不可信的外部输入源",像对待用户提交的表单数据一样,在服务端建立完整的校验与防御体系。

三大技术支柱:JSON Mode、Structured Outputs 与 Function Calling

目前主流大模型厂商(OpenAI、Anthropic、阿里通义、字节豆包等)都提供了不同层次的结构化输出能力。我们可以把它们归纳为三大支柱:

维度JSON ModeStructured OutputsFunction Calling
本质输出格式开关结构化生成能力调用意图生成机制
核心约束仅保证语法是合法 JSON严格匹配指定的 JSON Schema映射为工具名和参数对象
典型用途简单、非严格的数据抽取工单分类、信息抽取、Agent 状态管理读写业务系统、操作外部 API
约束强度中(保证语法,不保证结构)强(部分模型支持解码层约束)面向动作的强契约
字段可控性无法保证字段存在与类型字段、类型、枚举均可约束参数 schema 可约束
一个关键认知:Function Calling ≠ 模型执行了代码

这是初学者最容易误解的一点。很多开发者听到"Function Calling"(函数调用),会下意识以为"模型调用了我的函数、执行了我的代码"。

事实完全不是这样。 Function Calling 中,模型做的事情只有一件:生成一段"调用意图",也就是告诉你"我想调用哪个工具,参数应该长这样"。

真正的执行权、校验权和审计权,始终牢牢掌握在你的业务服务端手里:

  • 模型说:“我想调用 refund_order(orderId=123)”;
  • 你的服务端收到这个意图后,由你自己决定要不要执行、参数合不合法、这个用户有没有权限。

模型只是"提议",服务端才是"决策者"。这一点是后面三层校验体系的逻辑起点。

在这里插入图片描述

三、实现步骤:用 JSON Schema 建立数据契约

3.1 JSON Schema 是大模型与后端之间的"数据合同"

JSON Schema 是一份机器可读的规范,它精确定义了:字段叫什么名字、是什么类型、哪些是必填的、合法的枚举范围有哪些。它既是给模型看的"答题卡",也是给后端校验器看的"验收标准"。

一份设计良好的 Schema,能让模型输出的随机性被压缩在一个可控的框架内。在设计 Schema 时,建议遵循下面三个原则:

原则一:原子化(Atomic)

字段拆得越细,后端就越容易逐项校验和路由。避免把多个含义塞进一个自由文本字段。

❌ 不推荐:

{
  "action_info": "用户想退款,订单号是 123,退 50 块"
}

✅ 推荐:

{
  "action": "refund",
  "orderId": 123,
  "amount": 50
}
原则二:明确边界(Description)

每个字段的 description 应该清楚地告诉模型:什么时候该填、什么时候不该填、合法的取值是什么。不要假设模型"应该懂",要把业务语义写进 description。

原则三:版本化(Versioning)

在 Schema 里显式加一个版本字段(如 schemaVersion),用于应对 Prompt 或业务规则变更。当线上同时存在新旧两套规则时,版本号能让你的校验逻辑和灰度发布有据可依。

3.2 一个完整的 JSON Schema 契约示例

下面以一个"客服工单分类 + 退款动作"场景为例,定义一份生产可用的 Schema:

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "CustomerServiceDecision",
  "description": "客服场景下,模型对用户诉求的结构化决策结果",
  "type": "object",
  "properties": {
    "schemaVersion": {
      "type": "string",
      "enum": ["v1", "v2"],
      "description": "数据契约版本号,当前生产环境使用 v1"
    },
    "intent": {
      "type": "string",
      "enum": ["consult", "complaint", "refund", "exchange"],
      "description": "用户意图分类。consult=一般咨询;complaint=投诉;refund=退款;exchange=换货。不确定时填 consult"
    },
    "orderId": {
      "type": "integer",
      "description": "用户提到的订单号。如果用户未提供明确订单号,请填 0,不要编造"
    },
    "amount": {
      "type": "number",
      "minimum": 0,
      "description": "涉及金额(退款/换货差价),单位元。咨询类填 0"
    },
    "urgency": {
      "type": "string",
      "enum": ["low", "medium", "high"],
      "description": "紧急程度。high 仅用于退款/投诉且金额>500 的场景"
    },
    "summary": {
      "type": "string",
      "maxLength": 100,
      "description": "用户诉求的一句话摘要,不超过100字"
    }
  },
  "required": ["schemaVersion", "intent", "orderId", "amount", "urgency", "summary"],
  "additionalProperties": false
}

注意几个关键设计:

  • intentenum 限定死取值范围,从根本上杜绝"幻觉枚举";
  • orderId 在 description 里明确"不知道就填 0,不要编造",抑制幻觉;
  • amountminimum: 0 约束,配合业务层校验退款上限;
  • required 列出所有必填字段,防止字段缺失;
  • additionalProperties: false 拒绝多余字段,让契约保持纯净。

四、代码示例:从调用到三层校验的完整闭环

下面用 Python 给出可复现的完整实现。技术栈:openai SDK + pydantic + jsonschema

4.1 环境准备

# Python 3.10+
pip install openai pydantic jsonschema --break-system-packages

4.2 错误示范:只靠"请返回 JSON"

先看一个典型的"教科书式"错误写法,感受它的脆弱:

from openai import OpenAI

client = OpenAI()  # 默认读取环境变量 OPENAI_API_KEY

BAD_PROMPT = """
你是一个客服助手。请分析用户诉求,并返回 JSON,包含 intent、orderId、amount 字段。
请只返回 JSON,不要任何多余文字!
"""

def bad_classify(user_msg: str) -> dict:
    # 危险:完全相信模型输出,没有任何校验
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": BAD_PROMPT},
            {"role": "user", "content": user_msg},
        ],
    )
    # 这一行随时可能抛 JSONDecodeError
    return eval(resp.choices[0].message.content)  # 更危险:eval 执行任意代码!

这段代码至少有三个致命问题:

  1. eval() 解析返回——这是安全漏洞,模型若返回恶意字符串可执行任意代码;
  2. 没有任何结构校验,字段缺失/类型错误全靠运气;
  3. 没有重试和兜底,一次失败整个请求就崩了。

4.3 正确姿势:Structured Outputs + Pydantic 校验

下面是生产级写法。我们用 Pydantic 定义契约,并通过 response_format 让模型严格按 Schema 输出。

from typing import Literal
from pydantic import BaseModel, Field, field_validator

# 1) 用 Pydantic 定义数据契约(它会被自动转换为 JSON Schema)
class ServiceDecision(BaseModel):
    schemaVersion: Literal["v1", "v2"] = Field(
        default="v1",
        description="数据契约版本号,当前生产环境使用 v1"
    )
    intent: Literal["consult", "complaint", "refund", "exchange"] = Field(
        description="用户意图分类。consult=一般咨询;complaint=投诉;refund=退款;exchange=换货"
    )
    orderId: int = Field(
        description="用户提到的订单号。如果用户未提供明确订单号,请填 0,不要编造"
    )
    amount: float = Field(
        default=0.0,
        ge=0,
        description="涉及金额,单位元。咨询类填 0"
    )
    urgency: Literal["low", "medium", "high"] = Field(
        description="紧急程度。high 仅用于退款/投诉且金额>500 的场景"
    )
    summary: str = Field(
        max_length=100,
        description="用户诉求的一句话摘要,不超过100字"
    )

    # 自定义校验器:业务规则约束
    @field_validator("amount")
    @classmethod
    def check_amount(cls, v: float) -> float:
        if v > 100000:
            raise ValueError("退款金额超过单笔上限 10 万元")
        return v

然后调用时启用 Structured Outputs(以 OpenAI 为例,response_format 传入 Pydantic 模型):

import json
from openai import OpenAI

client = OpenAI()

SYSTEM_PROMPT = """你是一个客服助手,负责分析用户诉求并输出结构化决策。
规则:
1. intent 只能从 consult/complaint/refund/exchange 中选择。
2. 如果用户没有明确给出订单号,orderId 必须填 0,绝不猜测或编造。
3. amount 表示涉及金额,咨询类填 0。
4. urgency 为 high 仅当是退款或投诉,且金额大于 500。"""

def call_model(user_msg: str) -> ServiceDecision | None:
    """调用模型,返回严格类型化的决策对象(失败返回 None)"""
    try:
        resp = client.beta.chat.completions.parse(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": user_msg},
            ],
            response_format=ServiceDecision,  # 结构化输出:解码层约束
        )
        # SDK 已自动完成解析与类型校验
        return resp.choices[0].message.parsed
    except Exception as e:
        # 记录日志,交由上层重试/降级
        print(f"[模型调用失败] {e}")
        return None

注意 client.beta.chat.completions.parse 这一行:传入 response_format=ServiceDecision 后,模型会被约束在 JSON Schema 的"笼子"里生成,resp.choices[0].message.parsed 直接就是一个类型安全的 ServiceDecision 对象。这就是"解码层约束"的威力——它比提示词可靠得多。

4.4 核心:服务端三层校验防御体系

大模型可以"建议"操作,但绝不能"代替"决策。即使开了 Structured Outputs,生产级应用仍必须在服务端建立三层防御。下面的代码把这套体系串起来:

from dataclasses import dataclass

# ============ 业务系统模拟 ============
# 订单库(实际从数据库读取)
ORDERS = {
    1001: {"userId": "u_001", "status": "paid",    "total": 200},
    1002: {"userId": "u_002", "status": "shipped", "total": 800},
    1003: {"userId": "u_001", "status": "paid",    "total": 3000},
}

# 越权订单(属于别的用户)——用于演示权限校验
CROSS_USER_ORDER = 9001  # 属于 u_999

CURRENT_USER = "u_001"
MAX_REFUND_AMOUNT = 10000


@dataclass
class ValidationResult:
    ok: bool
    error: str = ""


# ---------- 第一层:结构校验 ----------
def structural_validation(decision: ServiceDecision) -> ValidationResult:
    """检查返回结果是否符合契约(字段/类型/枚举)"""
    # Pydantic 在 parse 阶段已做基础校验,这里补充额外结构约束
    if decision.orderId == 0 and decision.intent in ("refund", "exchange"):
        return ValidationResult(False, "退款/换货必须提供有效的 orderId")
    if decision.intent == "refund" and decision.amount <= 0:
        return ValidationResult(False, "退款意图的 amount 必须大于 0")
    return ValidationResult(True)


# ---------- 第二层:业务校验 ----------
def business_validation(decision: ServiceDecision) -> ValidationResult:
    """检查数据在业务逻辑上是否合理"""
    if decision.intent != "refund":
        return ValidationResult(True)  # 非退款类不深入校验

    order = ORDERS.get(decision.orderId)
    if order is None:
        return ValidationResult(False, f"订单 {decision.orderId} 不存在")

    # 订单状态是否支持退款
    if order["status"] != "paid":
        return ValidationResult(
            False, f"订单状态为 {order['status']},当前不可退款"
        )
    # 退款金额是否在有效范围
    if decision.amount > order["total"]:
        return ValidationResult(
            False, f"退款金额 {decision.amount} 超过订单实付 {order['total']}"
        )
    if decision.amount > MAX_REFUND_AMOUNT:
        return ValidationResult(False, "退款金额超过单笔上限")
    return ValidationResult(True)


# ---------- 第三层:权限校验 ----------
def permission_validation(decision: ServiceDecision) -> ValidationResult:
    """校验当前用户是否有权操作该资源(最危险,绝不能交给模型)"""
    if decision.intent != "refund":
        return ValidationResult(True)

    order = ORDERS.get(decision.orderId)
    if order and order["userId"] != CURRENT_USER:
        return ValidationResult(
            False,
            f"订单 {decision.orderId} 不属于当前用户,拒绝操作(疑似越权)"
        )
    return ValidationResult(True)


# ---------- 三层串联 ----------
def full_validate(decision: ServiceDecision) -> tuple[bool, str]:
    """依次执行三层校验,任一失败即拦截"""
    for validator, name in [
        (structural_validation, "结构校验"),
        (business_validation,  "业务校验"),
        (permission_validation,"权限校验"),
    ]:
        r = validator(decision)
        if not r.ok:
            print(f"[{name}] 失败: {r.error}")
            return False, f"[{name}] {r.error}"
    return True, ""

4.5 高风险操作:Human-in-the-loop

对于退款、删除、支付这类高风险操作,即使三层校验全过,也不应自动执行,而应进入人工确认环节:

def execute_decision(decision: ServiceDecision) -> str:
    """执行决策(高风险动作需人工确认)"""
    ok, err = full_validate(decision)
    if not ok:
        return f"已拦截: {err}"

    # 高风险操作:进入人工确认队列
    if decision.intent in ("refund",) and decision.amount >= 500:
        ticket_id = create_review_ticket(decision)  # 创建人工审核工单
        return (
            f"退款 {decision.amount} 元已进入人工审核队列,"
            f"审核工单号: {ticket_id}(Human-in-the-loop)"
        )

    # 低风险:可直接执行
    do_refund(decision.orderId, decision.amount)
    return "退款已执行"


def create_review_ticket(decision: ServiceDecision) -> str:
    """模拟创建人工审核工单"""
    return f"TK-{decision.orderId:05d}"


def do_refund(order_id: int, amount: float) -> None:
    """模拟执行退款"""
    print(f"[执行] 订单 {order_id} 退款 {amount} 元")

五、运行结果或效果说明

我们用几个典型用户输入来跑通整个链路,观察不同情况下的表现:

def run_case(user_msg: str) -> None:
    print(f"\n===== 用户输入: {user_msg} =====")
    decision = call_model(user_msg)
    if decision is None:
        print("模型调用失败,进入重试/降级流程")
        return
    print(f"模型决策: {decision.model_dump()}")
    result = execute_decision(decision)
    print(f"执行结果: {result}")


if __name__ == "__main__":
    # 场景1:正常退款(低金额,低风险)
    run_case("我昨天买的耳机,订单号 1001,想退 100 块。")

    # 场景2:高额退款(触发人工审核)
    run_case("订单 1003,我要退款 2500 元。")

    # 场景3:订单不存在
    run_case("订单 8888 退款 50 元。")

    # 场景4:未提供订单号(模型应填 0,被结构校验拦截)
    run_case("我想退款。")

预期输出(示意):

===== 用户输入: 我昨天买的耳机,订单号 1001,想退 100 块。 =====
模型决策: {'schemaVersion': 'v1', 'intent': 'refund', 'orderId': 1001, 'amount': 100.0, 'urgency': 'low', 'summary': '用户要求对订单1001退款100元'}
执行结果: 退款已执行

===== 用户输入: 订单 1003,我要退款 2500 元。 =====
模型决策: {'schemaVersion': 'v1', 'intent': 'refund', 'orderId': 1003, 'amount': 2500.0, 'urgency': 'high', 'summary': '用户要求对订单1003退款2500元'}
执行结果: 退款 2500.0 元已进入人工审核队列,审核工单号: TK-01003(Human-in-the-loop)

===== 用户输入: 订单 8888 退款 50 元。 =====
模型决策: {'schemaVersion': 'v1', 'intent': 'refund', 'orderId': 8888, 'amount': 50.0, 'urgency': 'medium', 'summary': '用户要求对订单8888退款50元'}
[业务校验] 失败: 订单 8888 不存在
执行结果: 已拦截: [业务校验] 订单 8888 不存在

===== 用户输入: 我想退款。 =====
模型决策: {'schemaVersion': 'v1', 'intent': 'refund', 'orderId': 0, 'amount': 0.0, 'urgency': 'low', 'summary': '用户想退款但未提供订单信息'}
[结构校验] 失败: 退款/换货必须提供有效的 orderId
执行结果: 已拦截: [结构校验] 退款/换货必须提供有效的 orderId

可以看到:

  • 场景1:低风险退款,三层校验通过,自动执行;
  • 场景2:高额退款,校验通过但因金额触发 Human-in-the-loop;
  • 场景3:模型可能编造了订单号,业务校验精准拦截;
  • 场景4:模型遵守契约把 orderId 填 0,结构校验补位拦截并引导用户补充信息。

在这里插入图片描述

六、失败后的降级与重试策略

即使做了上述所有工作,结构化输出仍有失败概率(模型接口超时、Schema 冲突、极端输入等)。工程上必须建立闭环:

6.1 有限重试(带具体错误反馈)

校验失败时,不要原样重跑,而是把具体的错误信息反馈给模型,让它"定向修复"。这比盲目重试有效得多:

def call_with_retry(user_msg: str, max_retries: int = 2) -> ServiceDecision | None:
    last_error = ""
    for attempt in range(max_retries + 1):
        # 把上一次的错误塞进 system prompt,引导模型修复
        extra = f"\n\n注意:上次输出有误,请修正:{last_error}" if last_error else ""
        try:
            resp = client.beta.chat.completions.parse(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT + extra},
                    {"role": "user", "content": user_msg},
                ],
                response_format=ServiceDecision,
            )
            decision = resp.choices[0].message.parsed
            ok, err = full_validate(decision)
            if ok:
                return decision
            last_error = err  # 携带具体错误进入下一次重试
            print(f"[第{attempt+1}次] 校验未通过: {err}")
        except Exception as e:
            last_error = str(e)
            print(f"[第{attempt+1}次] 调用异常: {e}")
    return None

关键点:重试次数要有限(建议 1-2 次),避免无限循环放大延迟和成本。

6.2 业务降级

若重试仍失败,不应让请求直接 500,而应进入降级路径:

  • 转入人工队列,由客服/运营人工处理;
  • 使用预设规则兜底,如默认分类为 consult、金额置 0,先保证链路不中断;
  • 对用户返回友好的兜底回复,而不是暴露内部错误。

6.3 全链路审计

每一次模型交互都应记录完整链路,确保事故可追溯:

import logging
import time

logger = logging.getLogger("llm_audit")

def audit_log(user_msg: str, decision, validate_result, action: str):
    """全链路审计:原始输入 / 模型建议 / 校验结果 / 最终动作"""
    logger.info({
        "timestamp": time.time(),
        "input": user_msg,
        "model_suggestion": decision.model_dump() if decision else None,
        "validate_ok": validate_result[0],
        "validate_error": validate_result[1],
        "final_action": action,
    })

记录四个关键节点:原始输入 → 模型建议 → 校验结果 → 最终执行动作。这样一旦出现资损或越权事故,可以完整还原。

在这里插入图片描述

七、常见问题与避坑

Q1:开了 Structured Outputs 是不是就不用做服务端校验了?

不是。Structured Outputs 解决的是"格式和结构"问题,但解决不了"业务合理性"和"权限"问题。模型仍可能生成一个语法合法、但订单号属于别人的退款请求。三层校验一个都不能少。

Q2:为什么我的 Function Calling 有时还是返回多余文字?

通常是流式输出(streaming)和多轮对话上下文的问题。建议:开启 Structured Outputs 或 tool_choice 强制选择工具;在流式场景单独处理拼接逻辑;保持 system prompt 简洁,避免与工具定义冲突。

Q3:用 eval() / json.loads() 解析模型返回有什么风险?

eval()高危操作,等于让模型在你的服务器上执行任意 Python 代码,必须杜绝。即使用 json.loads(),也只解决语法问题,不解决结构问题。正确做法是用 Pydantic / JSON Schema Validator 做严格校验后再使用。

Q4:枚举值经常越界怎么办?

三个手段叠加:① Schema 层用 enum 限定;② description 里把每个枚举的含义和适用场景写清楚;③ 服务端用 Literalenum 做二次校验,越界即重试或降级。

Q5:重试会不会导致用户等待太久?

会。因此重试次数要严格控制(1-2 次),并设置单次调用超时。对于延迟敏感场景,可以采用"先返回兜底结果 + 异步重试回填"的策略,把体验和正确性分开处理。

Q6:不同模型厂商的结构化输出能力一样吗?

不一样。OpenAI 的 Structured Outputs 支持解码层约束,严格度最高;部分厂商只支持 JSON Mode(仅保证语法合法)。接入新厂商时,务必确认其结构化能力的边界,并据此调整服务端校验的强度——模型约束越弱,服务端校验越要强

八、总结

回到开头那个问题:为什么"请返回 JSON"不可靠?因为它本质上是一句自然语言建议,而生产环境需要的是一份工程契约

结构化输出的本质,是把大模型从"生成文本的黑盒"收敛为"遵循契约的接口"。这件事不能只靠模型自觉,需要三件事协同:

  1. 模型能力层:用 Structured Outputs / Function Calling 把约束下沉到解码阶段,而不是停留在提示词;
  2. 数据契约层:用 JSON Schema 把字段、类型、枚举、必填项定义成机器可读的合同;
  3. 服务端防御层:用结构校验 → 业务校验 → 权限校验的三层体系 + 人工确认守住底线,再以有限重试、业务降级、全链路审计兜住失败。

自然语言提示只是引导,服务端的三层校验才是线上应用的底线。别再指望模型"变聪明",用精密的工程设计为它打造一套安全、可控的"装具(Harness)"——这,才是大模型工程化的真正门槛。

更多推荐