LangChain 跨 Provider 结构化输出与降级 -Yòuyou

LLM 最自然的输出是一段文本：

可以继续追问：
1. 如何创建应用？
2. 怎样接入模型？
3. 工作流有什么作用？

页面只是把这段内容显示出来时，文本已经够用。程序如果要把三个问题分别渲染成按钮，就需要另一种结果：

{
  "questions": [
    "如何创建应用？",
    "怎样接入模型？",
    "工作流有什么作用？"
  ]
}

两者的差别不只是有没有花括号。下游代码需要确认 questions 存在、它是 list、每一项都是 string，而且数量不超过三条。结构化输出（Structured Output）就是让模型结果满足这类可由程序检查的契约。

单个 Provider 中，LangChain 的 with_structured_output() 可以把 Schema 绑定到 ChatModel。问题出现在模型可以切换时：相同的方法名背后可能使用原生 JSON Schema、JSON Mode 或 Function Calling。一个模型能正常聊天，不代表它支持当前选择的结构化机制。

从文本到 Pydantic 对象发生了什么

Pydantic Schema 可以描述程序期望的数据：

from pydantic import BaseModel, Field
 
 
class SuggestedQuestions(BaseModel):
    questions: list[str] = Field(
        min_length=1,
        max_length=3,
        description="根据对话生成的后续问题",
    )

如果得到普通 Python 数据，Pydantic 可以完成最终校验：

valid = SuggestedQuestions.model_validate({
    "questions": [
        "如何创建应用？",
        "怎样接入模型？",
    ]
})
 
print(valid.questions)

下面几种结果都不符合契约：

# 缺少 questions
{}
 
# questions 不是 list
{"questions": "如何创建应用？"}
 
# 元素类型不对
{"questions": [1, 2, 3]}
 
# 超过 Schema 声明的数量
{"questions": ["A", "B", "C", "D"]}

LangChain 的模型级结构化输出写法是：

structured_model = model.with_structured_output(SuggestedQuestions)
result = structured_model.invoke("根据当前对话生成后续问题")

从调用到结果，中间至少存在五步：

Pydantic Schema
      │
      ▼
LangChain 把 Schema 转成 Provider 能理解的请求
      │
      ▼
Provider / Model 生成 JSON 或 Tool Call
      │
      ▼
LangChain 解析返回值
      │
      ▼
Pydantic 校验并产生 SuggestedQuestions

with_structured_output() 统一的是最上层调用形状，并没有让所有 Provider 使用同一种底层协议。

三种常见结构化策略

LangChain 当前模型文档把常见 method 分成 json_schema、function_calling 和 json_mode。具体类与模型支持哪些方式，需要查看相应 Provider 集成文档：Models - Structured output。

JSON Schema

Provider 原生接受一份 Schema，并在生成阶段约束输出：

Schema
  │
  ▼
response_format / native structured-output field
  │
  ▼
Provider 按 Schema 生成

strict=True 是否可用由 Provider 和 LangChain 集成决定。支持时，字段名称、类型和必填项会由服务端约束。

Function Calling

LangChain 把 Schema 转成一个工具的参数定义，让模型产生 Tool Call：

{
  "name": "SuggestedQuestions",
  "arguments": {
    "questions": [
      "如何创建应用？"
    ]
  }
}

这里不一定真的执行某个业务工具。Tool Call 的参数只是被用作结构化数据载体。模型或兼容端点如果不支持 Tool Calling、具体 tool_choice 或相应 Schema，这条路径会在请求阶段失败。

JSON Mode

JSON Mode 只要求输出是合法 JSON：

请返回 JSON
      │
      ▼
{"questions": [...]}

它通常不接收完整 Schema 约束。字段名、数量和嵌套结构仍要写进 Prompt，并在返回后用 Pydantic 校验。合法 JSON 不等于合法业务对象：

{"answer": "这是合法 JSON，但没有 questions 字段"}

Prompt + Parser

某些模型没有可用的原生结构化能力，只能在 Prompt 中加入格式说明，再用 PydanticOutputParser 解析文本：

Prompt 中说明格式
      │
      ▼
模型生成普通文本
      │
      ▼
Parser 提取 JSON
      │
      ▼
Pydantic 校验

约束发生得越晚，越容易得到 Markdown 代码块、额外解释、无效 JSON 或缺失字段。

策略	Schema 在哪里生效	常见失败
JSON Schema	Provider 生成阶段	Provider 或模型不支持 Schema
Function Calling	Tool Call 参数	不支持工具、`tool_choice` 或参数 Schema
JSON Mode	只保证 JSON 语法	字段缺失、类型和数量错误
Prompt + Parser	返回后的文本解析	额外文本、无效 JSON、提取失败

为什么普通聊天成功，结构化输出仍会失败

普通聊天只要求：

messages -> model -> text

结构化输出增加了能力要求：

messages
  + Schema
  + method
  + 可能存在的 strict / tool_choice
        │
        ▼
Provider 与具体模型必须同时支持

因此，下面四句话不能互相推出：

Provider 提供 OpenAI-compatible Chat Completions
这个模型支持普通聊天
这个模型支持 Tool Calling
这个模型支持某一种结构化输出策略

例如 ChatDeepSeek 集成文档分开标记模型能力：对话模型可以支持 Tool Calling 和结构化输出，推理模型可能不支持。这说明能力属于“Provider + 具体模型 + 当前集成版本”，不能只写成“DeepSeek 支持”或“DeepSeek 不支持”。

实际的跨 Provider 故障常有两个阶段：

阶段	现象	说明
请求阶段	Provider 拒绝 `tool_choice`、Schema 或参数	当前策略不被接受
返回阶段	得到文本、`dict`、`None` 或字段不完整	请求完成，但结果没有满足契约

继续修改 Prompt 只能影响第二类问题的一部分。第一类问题发生在 Provider 解析请求时，Prompt 还没有机会发挥作用。

策略为什么属于模型适配层

如果每个业务 Service 自己调用：

model.with_structured_output(
    SuggestedQuestions,
    method="function_calling",
)

Service 就必须知道当前模型支持哪种方法。增加 Provider 后，同一份判断会散落在标题、分类、候选问题和工作流节点中。

结构化策略可以作为 Provider 默认值，由具体模型覆盖：

- name: openai
  structured_output_strategy: json_schema
  structured_output_strict: true
 
- name: deepseek
  structured_output_strategy: json_mode
 
- name: zhipu
  structured_output_strategy: function_calling

模型级覆盖处理同一 Provider 内的能力差异：

model: provider-model-name
structured_output_strategy: prompt
structured_output_strict: false

这张表描述的是某个项目在当前依赖版本和已验证模型下的选择，不是 Provider 的永久能力清单。Provider 升级 API、LangChain 改变默认行为或模型换代后，需要重新验证。

业务只表达两件事：

使用哪个模型配置
结果要符合哪个 Schema

方法选择交给统一模型工厂。

一个完整的结构化模型工厂

下面的实现包含策略选择、Pydantic 校验和有限重试：

from enum import Enum
from typing import Any
 
from langchain_core.exceptions import OutputParserException
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.runnables import Runnable, RunnableLambda
from pydantic import BaseModel, ValidationError
 
 
class StructuredMethod(str, Enum):
    JSON_SCHEMA = "json_schema"
    FUNCTION_CALLING = "function_calling"
    JSON_MODE = "json_mode"
    PROMPT = "prompt"
 
 
def create_structured_chat_model(
    model,
    schema: type[BaseModel],
    method: StructuredMethod,
    strict: bool = False,
    max_attempts: int = 2,
) -> Runnable[Any, BaseModel]:
    if max_attempts < 1:
        raise ValueError("max_attempts must be at least 1")
 
    if method is StructuredMethod.PROMPT:
        structured_model = model | PydanticOutputParser(
            pydantic_object=schema
        )
    else:
        options: dict[str, Any] = {"method": method.value}
 
        if method is StructuredMethod.JSON_SCHEMA:
            options["strict"] = strict
 
        structured_model = model.with_structured_output(
            schema,
            **options,
        )
 
    def validate(output: Any) -> BaseModel:
        if isinstance(output, schema):
            return output
        return schema.model_validate(output)
 
    validated_model = structured_model | RunnableLambda(validate)
 
    return validated_model.with_retry(
        retry_if_exception_type=(
            ValidationError,
            OutputParserException,
        ),
        wait_exponential_jitter=False,
        stop_after_attempt=max_attempts,
    )

假设模型目录选择：

method = function_calling
strict = false
schema = SuggestedQuestions
max_attempts = 2

运行时会经历：

1. with_structured_output() 把 SuggestedQuestions 转成工具参数
2. 模型返回 Tool Call
3. LangChain 提取 arguments
4. validate() 检查结果
5. 第一次若缺字段或类型错误，重新调用一次
6. 第二次仍失败，把异常交给业务层

strict 只在选中的方法确实支持时传入。把 strict=True 无条件发送给所有方法，可能让原本支持 JSON Mode 的 Provider 因未知参数失败。

Schema、解析和校验是三层

结构化输出常被一句“加个 Pydantic Schema”带过，实际包含三个边界：

Schema 描述

字段名称、类型、说明和数量限制会被转换成 Prompt、JSON Schema 或 Tool Schema。它影响模型怎样生成。

LangChain 解析

LangChain 从 Provider 返回值中取得 JSON、Tool Call arguments 或普通文本，并尝试转换成 Python 数据。

Pydantic 校验

最终对象仍要满足业务约束：

def validate_result(
    result: object,
    schema: type[BaseModel],
) -> BaseModel:
    if isinstance(result, schema):
        return result
 
    return schema.model_validate(result)

这一步不能只检查 isinstance(result, dict)。下面的对象是 dict，但仍然无效：

{
    "questions": [
        "A",
        "B",
        "C",
        "D",
    ]
}

错误消息应尽量保留期望 Schema、实际类型和验证详情。否则真正的“缺少 questions”会在更远处变成 NoneType has no attribute questions。

重试只处理可能漂移的输出

模型同一个请求的输出可能变化。第一次返回无效 JSON，第二次可能满足 Schema，所以解析和校验失败可以做有限重试。

适合重试的错误：

无效 JSON
缺少字段
字段类型错误
返回 None 或其他无法校验的值
文本解析失败

不适合原样重试的错误：

API Key 缺失或无效
Provider、模型不存在
当前模型不支持选中的结构化方法
请求参数被 Provider 明确拒绝
Schema 本身无法被当前方法表达

后一组错误在配置不变时会稳定复现。把 Exception 全部纳入重试，只会重复付费请求，并把配置错误延迟到最后一次。

重试次数也需要有上限：

第 1 次：调用 + 校验失败
第 2 次：再次调用 + 校验失败
停止重试，把失败交给业务层

重试解决的是“同一契约下偶发没有满足”，不是“契约与 Provider 根本不兼容”。

先判断任务是否真的需要结构化输出

不是所有 LLM 调用都需要 Schema。判断标准是下游代码如何使用结果。

会话标题

标题只需要一段短文本：

from langchain_core.output_parsers import StrOutputParser
 
chain = prompt | model | StrOutputParser()
title = chain.invoke({"query": query})

后续只做：

normalized = " ".join(title.split()).strip("`'\"")
normalized = normalized[:50]

给标题增加 {"subject": "..."} 不会增加业务信息，却会增加一次结构化能力依赖。

候选问题

候选问题要分别渲染和限制数量，Schema 有明确价值：

class SuggestedQuestions(BaseModel):
    questions: list[str] = Field(
        min_length=1,
        max_length=3,
    )

任务	下游怎样使用	输出契约
标题	显示一段短文本	`str`
摘要	保存一段文本	`str`
候选问题	渲染多个独立按钮	Pydantic list
分类	按类别进入不同分支	Enum 或 Pydantic
工具参数	传给确定性函数	Tool Schema

结构化输出不是“比文本更正式”的包装，而是下游确实需要字段、类型或枚举时才增加的契约。

业务降级不能放进模型工厂

结构化工厂知道：

这个结果有没有满足 SuggestedQuestions

它不知道：

候选问题失败后应该返回空列表、旧结果，还是让整个请求失败

后一个问题属于业务语义，所以降级留在 Service：

import logging
 
 
def generate_suggested_questions(
    structured_model,
    histories: str,
) -> list[str]:
    try:
        result = structured_model.invoke({
            "histories": histories,
        })
        return result.questions[:3]
    except Exception:
        logging.exception(
            "生成候选问题失败，回退为空列表"
        )
        return []

标题则有不同的降级值：

def generate_conversation_name(
    title_chain,
    query: str,
) -> str:
    fallback = normalize_title(query)
 
    try:
        generated = title_chain.invoke({"query": query})
        return normalize_title(generated) or fallback
    except Exception:
        return fallback

辅助任务	失败后的业务值	为什么可降级
会话标题	用户问题的短文本	标题质量不影响回答
候选问题	空列表	页面可以不显示按钮
历史摘要	保留旧摘要	当前消息仍可保存
非核心标签	跳过标签	主记录仍然有效

主聊天生成失败通常不能用空字符串假装成功；辅助任务失败则不应反过来破坏已经生成的回答。降级位置由任务重要性决定，不由 Provider 决定。

完整边界最终是：

模型目录
  声明 Provider / Model 当前使用的结构化策略
        │
        ▼
模型工厂
  创建对应 Runnable
        │
        ▼
解析与 Pydantic 校验
  把输出漂移变成明确异常
        │
        ▼
有限重试
  只重试可能变化的解析和校验错误
        │
        ▼
业务降级
  决定空列表、旧摘要或原始文本

这条链可以分别离线测试：

每个 Provider 是否选择预期 method
strict 是否只传给支持的方法
第一次返回 None、第二次返回合法 dict 时是否只重试一次
两次都失败时是否把异常交给 Service
标题失败是否回退到原问题
候选问题失败是否返回空列表
摘要失败是否仍然保存主消息

这些测试不调用真实模型，只验证策略选择、数据校验和业务失败边界。真实 Provider 是否接受当前 Schema，仍需要单独的 opt-in smoke test。

Yòuyou

探索

LangChain 跨 Provider 结构化输出与降级