LLM 结构化输出学习笔记

1. 流式输出：只打印内容，不打印 JSON

问题

原始代码直接 print(chunk.model_dump_json())，输出的是每个 chunk 的完整 JSON，可读性差。

解决

真正要显示的内容在 chunk.choices[0].delta.content，Qwen3 系列还可能有 delta.reasoning_content（思考过程）。最后一个带 usage 的 chunk 的 choices 是空列表，需要跳过。

python

for chunk in completion:
    if not chunk.choices:          # 最后一个 usage chunk，choices 为空
        continue
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)
print()

完整版（含思考过程 + usage 统计）

python

from openai import OpenAI
import os

def get_response():
    client = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
    )
    completion = client.chat.completions.create(
        model="qwen3.8-flash",
        messages=[{'role': 'system', 'content': 'You are a helpful assistant.'},
                  {'role': 'user', 'content': '你是谁？'}],
        stream=True,
        stream_options={"include_usage": True},
    )

    usage = None
    for chunk in completion:
        if not chunk.choices:
            usage = chunk.usage
            continue

        delta = chunk.choices[0].delta

        reasoning = getattr(delta, "reasoning_content", None)
        if reasoning:
            print(reasoning, end="", flush=True)

        if delta.content:
            print(delta.content, end="", flush=True)

    print()
    if usage:
        print(f"[tokens] prompt={usage.prompt_tokens}, "
              f"completion={usage.completion_tokens}, total={usage.total_tokens}")

关键字段对照

小贴士：flush=True 很重要，否则终端缓冲会让输出看起来"卡住"；end="" 让 token 连续拼接。

2. 思考过程与正式内容的输出顺序

结论：正常情况下顺序固定

对于开启思考模式的模型，流式 chunk 严格分为两阶段：

1. 第一阶段：逐个 chunk 返回 delta.reasoning_content，此时 delta.content 为空
2. 第二阶段：思考结束后，开始逐个 chunk 返回 delta.content，此时 delta.reasoning_content 为空

官方文档明确"先输出 reasoning_content，再输出 content"。

⚠️ 三个边界情况

健壮写法：独立处理，不用互斥判断

python

for chunk in completion:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta

    if getattr(delta, "reasoning_content", None):
        print(delta.reasoning_content, end="", flush=True)
    if delta.content:
        print(delta.content, end="", flush=True)

3. 结构化 JSON 输出的四级可靠性阶梯

从"依赖提示词"逐步升级到"由解码器强制约束"，可靠性依次递增：

text

第一级  提示词工程        → 仅"尽力而为"，不保证合法
第二级  JSON Object 模式  → 保证语法合法，不保证结构
第三级  JSON Schema 严格  → 100% 匹配字段名/类型/必填项（推荐）
第四级  Function Calling  → 借用工具参数的 Schema 约束

第一级：纯提示词工程

在 Prompt 里写"请只返回 JSON"。

• 不保证：可能夹带 `json 标记、额外解释文字、字段名漂移
• 适用：仅原型验证

第二级：JSON Object 模式

通过 response_format={"type": "json_object"} 开启。

• ✅ 保证：输出是语法合法的 JSON 字符串，无多余散文
• ❌ 不保证：字段名、类型、必填项完全不受控
• ⚠️ 硬性要求：messages 中必须包含单词 "JSON"，否则 400 报错
• 适用：只需"一块合法 JSON"，对内部结构无严格约定

第三级：JSON Schema 严格模式（推荐生产）

通过 response_format={"type": "json_schema", "json_schema": {..., "strict": true}} 开启。

• ✅ 保证：100% 匹配 Schema——字段名、类型、必填项、枚举值全部受控，不会生成 schema 之外的额外键
• 原理：约束发生在解码阶段，推理引擎在采样时屏蔽任何违反 schema 的 token，而非生成后再检查
• strict 硬性要求：所有字段都必须在 required 中；可选字段用 type: ["string", "null"] 表达；additionalProperties: false 隐式设置
• ⚠️ Qwen 注意：思考模式下部分模型即使设了 response_format，返回内容可能不是严格合法 JSON。结构要求高时考虑关闭思考模式

第四级：Function Calling

把想要的 JSON 结构定义为"工具"的 parameters，通过 tool_choice 强制模型"调用"。

• 与第三级区别：Function Calling 目的是让模型执行动作，输出是 tool_calls；Structured Outputs 目的是约束回复本身结构
• 适用：Agent 架构，需要模型决定"调用哪个工具、传什么参数"
• 同样支持 strict: true

4. 第二级：JSON Object 模式完整代码

python

from openai import OpenAI
import os
import json

def get_json_response():
    client = OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
    )

    completion = client.chat.completions.create(
        model="qwen3.8-flash",
        messages=[
            {
                "role": "system",
                "content": (
                    "你是一个信息提取助手。"
                    "请从用户提供的文本中提取姓名、年龄、邮箱和职业，"
                    "并以 JSON 格式返回，包含字段：name、age、email、occupation。"
                    "只返回 JSON，不要包含任何其他文字或 markdown 代码块标记。"
                ),
            },
            {
                "role": "user",
                "content": "提取以下信息：张三，28岁，邮箱 zhang@example.com，职业是软件工程师。",
            },
        ],
        response_format={"type": "json_object"},   # 关键：开启 JSON Object 模式
        temperature=0.1,                            # 降低随机性
        stream=False,                               # JSON 模式建议先用非流式
    )

    raw = completion.choices[0].message.content
    print("=== 模型原始返回 ===")
    print(raw)

    try:
        data = json.loads(raw)
    except json.JSONDecodeError as e:
        print(f"JSON 解析失败: {e}")
        return None

    print(f"姓名: {data.get('name')}")
    print(f"年龄: {data.get('age')}")
    print(f"邮箱: {data.get('email')}")
    print(f"职业: {data.get('occupation')}")
    return data

补充说明

• 流式场景：JSON 逐 token 到达，不能对每个 chunk 做 json.loads。简单做法是流式接收、拼接完再一次性解析：

  python

  buffer = ""
  for chunk in completion:
      if chunk.choices and chunk.choices[0].delta.content:
          buffer += chunk.choices[0].delta.content
  data = json.loads(buffer)

• 兜底校验：该模式不保证字段名和类型，模型可能返回 {"age": "28"}（字符串）。下游按字段消费时，建议用 Pydantic 做后置校验。

5. 踩坑一：文件名 json.py 导致的循环导入

现象

text

ImportError: cannot import name 'OpenAI' from partially initialized module 'openai'
(most likely due to a circular import)

报错信息误导性地提示"循环导入"，但真正根因是文件名遮蔽标准库。

原因链条

1. 脚本名为 json.py，Python 启动时把脚本所在目录加到 sys.path 最前面
2. 脚本第一行 from openai import OpenAI → openai 内部 import httpx2
3. httpx2/_models.py 执行 import json as jsonlib——本意导入标准库 json
4. 但 sys.path 最前面有同名文件，Python 找到的是你自己的 json.py
5. 你的 json.py 又被执行，第一行又是 from openai import OpenAI，此时 openai 还在初始化中
6. 抛出 partially initialized module 错误

解决

重命名文件即可：

bash

mv /Users/songyang/project/ai/json.py /Users/songyang/project/ai/json_demo.py
rm -rf /Users/songyang/project/ai/__pycache__

危险文件名黑名单

经验法则：脚本名不要和任何可能 import 的模块同名。统一加后缀（_demo.py、_app.py）最省事。

6. 第三级：JSON Schema 严格模式完整代码

保存为 structured_output_demo.py（不要命名 json.py）。

python

from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional
import os
import json

# ---------- 1. 用 Pydantic 定义严格结构 ----------
class UserInfo(BaseModel):
    name: str = Field(description="用户姓名")
    age: int = Field(description="用户年龄")
    email: Optional[str] = Field(default=None, description="邮箱地址，没有则为 null")
    occupation: str = Field(description="职业")

# ---------- 2. 初始化客户端 ----------
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url="https://maas.qianwenaiapi.com/compatible-mode/v1",
)

# ---------- 3. 发起请求，开启 JSON Schema 严格模式 ----------
completion = client.chat.completions.create(
    model="qwen3.8-flash",
    messages=[
        {
            "role": "system",
            "content": "你是一个信息提取助手。请从用户提供的文本中精确提取信息，并严格按照 schema 返回。",
        },
        {
            "role": "user",
            "content": "张三，28岁，邮箱 zhang@example.com，职业是软件工程师。",
        },
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "user_info",
            "schema": UserInfo.model_json_schema(),
            "strict": True,
        },
    },
    temperature=0.1,
    # ⚠️ 重要：不要设置 max_tokens
)

# ---------- 4. 解析并校验结果 ----------
raw_content = completion.choices[0].message.content
print("=== 模型原始返回 ===")
print(raw_content)

try:
    data_dict = json.loads(raw_content)
except json.JSONDecodeError as e:
    print(f"JSON 解析失败（strict 模式下不应发生）: {e}")
    exit(1)

try:
    user = UserInfo.model_validate(data_dict)
except Exception as e:
    print(f"Schema 校验失败（strict 模式下不应发生）: {e}")
    exit(1)

print(f"姓名: {user.name}")
print(f"年龄: {user.age}（类型: {type(user.age).__name__}）")
print(f"邮箱: {user.email}")
print(f"职业: {user.occupation}")
print(user.model_dump())

为什么这份代码能"保证"结构

strict: True 的约束发生在解码阶段，任何会导致违反 Schema 的 token 在采样时就被屏蔽。Pydantic 的 model_json_schema() 自动处理 strict 模式的硬性要求：

• 所有字段（含 Optional）都进入 required 数组
• 可选字段类型自动生成为 ["string", "null"]
• additionalProperties: false 被隐式设置

两个必须注意的坑

坑一：绝对不要设 max_tokens。
官方文档明确警告：结构化输出开启时设置 max_tokens 可能把 JSON 字符串截断在半途，产生非法 JSON 导致解析失败。

坑二：流式模式下 strict 行为可能不可靠。
实测记录显示，开启 stream=True 后返回结构的顶层字段可能异常（例如只剩 content，丢失其他元数据）。需要严格结构保证时，使用非流式调用。 若业务必须流式，退回第二级 JSON Object 模式，或采用"流式接收、拼接完再解析"策略。

7. 选型速查表

生产环境最佳实践

机制强制 + 兜底校验
优先用 JSON Schema 严格模式从解码层杜绝结构错误；同时在业务代码保留一次 json.loads + Pydantic 校验作为最后防线（尤其在使用思考模式或较老模型时）。