Take To

Appearance

结构化输出：类型校验与结果获取 ​

3、关于类型校验 ​

验证不同schema定义方式在数据格式不匹配时的处理方式。

3.1 fake server ​

我们构造一个虚拟的DeepSeek服务端，客户端接收到的响应是人为构造的，主要目的是为了观察不同模式下LangChain会从响应中的哪些字段中抽取信息。

服务器端如下：

python
import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
python
class FakeDeepSeekHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get("Content-Length", 0))
        raw_body = self.rfile.read(content_length).decode("utf-8")
python
print("\n" + "=" * 100)
python
json_body = None
try:
    json_body = json.loads(raw_body)
    print("[JSON BODY]")
    print(json.dumps(json_body, ensure_ascii=False, indent=2))
except Exception as e:
    print("[JSON PARSE ERROR]")
python
print(repr(e))
python
# 固定返回一个普通 JSON 字符串
response = {
    "id": "chatcmpl-test",
    "object": "chat.completion",
    "created": int(time.time()),
    "model": "any",
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "call_1",
                        "type": "function",
                        "function": {
                            "name": json_body["tools"][0]["function"]["name"],
                            "arguments": json.dumps(
                                {'title1': '盗梦空间', 'year2': 2010,'director': '克里斯托弗·诺兰',
                                 'rating': 9.3},
                                ensure_ascii=False
                            )
                        }
                    }
text
                ]
            },
            "finish_reason": "stop"
        }
    ],
    "usage": {
        "prompt_tokens": 1,
        "completion_tokens": 1,
        "total_tokens": 2
    }
}
python
print("\n" + "=" * 100)
print("[RESPONSE]")
print(json.dumps(response, ensure_ascii=False, indent=2))
python
body = json.dumps(response, ensure_ascii=False).encode("utf-8")
python
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
python
def log_message(self, format, *args):
    pass
python
def main():
    server = HTTPServer(("127.0.0.1", 8889), FakeDeepSeekHandler)
    print("Fake DeepSeek server running at http://127.0.0.1:8889")
    server.serve_forever()
python
if __name__ == "__main__":
    main()

不必关注代码实现，只需要知道我们返回的核心内容是

text
{
 'title1': '盗梦空间',
 'year2': 2010,
 'director': '克里斯托弗·诺兰',
 'rating': 9.3
}

LangChain客户端希望模型返回的是

text
{
    'title': 'xxx',
    'year': xxxx,
    'director': 'xxx',
    'rating': xxx
}

这里的前两个字段不匹配。

启动服务端，然后执行四种定义方式的客户端代码

3.2 四种模式的校验 ​

3.2.1 Pydantic ​

python
from pydantic import BaseModel, Field, SecretStr
python
from langchain_deepseek import ChatDeepSeek
python
model = ChatDeepSeek(
    model="deepseek-v4-flash",
    api_base="http://localhost:8889",
    api_key=SecretStr("<KEY>")
)
python
class MovieModel(BaseModel):
    """
    电影的详细信息
    """
    title: str = Field(description="电影标题")
    year: int = Field(description="电影上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="电影评分，满分十分")
python
model_with_structure = model.with_structured_output(MovieModel)
python
response = model_with_structure.invoke("给出盗梦空间的信息")
print(response)
print(type(response))

输出

python
Traceback...
ValidationError: 2 validation errors for MovieModel
title
Field required [type=missing, input_value={'title1': '盗梦空间'...诺兰','rating': 9.3}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.12/v/missing
year
Field required [type=missing, input_value={'title1': '盗梦空间'...诺兰','rating': 9.3}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.12/v/missing

因为title 和year 字段不匹配抛出异常。

3.2.2 TypedDict ​

python
from typing_extensions import TypedDict, Annotated
python
from langchain_deepseek import ChatDeepSeek
python
model = ChatDeepSeek(
    model="deepseek-v4-flash",
    api_base="http://localhost:8889",
    api_key=SecretStr("<KEY>")
)
python
class MovieDict(TypedDict):
    """
    电影的详细信息
    """
    title: Annotated[str,"电影标题"]
    year: Annotated[int,"电影上映年份"]
    director: Annotated[str,"导演"]
    rating: Annotated[float,"电影评分，满分十分"]
python
structured_model = model.with_structured_output(MovieDict)
response = structured_model.invoke("给出盗梦空间的信息")
print(response)
print(type(response))

输出

text
{'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰','rating': 9.3}
<class 'dict'>

按照字典形式输出，不报错。

3.2.3 JSON schema ​

python
from langchain_deepseek import ChatDeepSeek
python
model = ChatDeepSeek(
    model="deepseek-v4-flash",
    api_base="http://localhost:8889",
    api_key=SecretStr("<KEY>")
)
python
json_schema = {
    "title": "Movie",
    "description": "A movie with details",
    "type": "object",
    "properties": {
        "title": {
            "type": "string",
            "description": "The title of the movie"
        },
        "year": {
            "type": "integer",
            "description": "The year the movie was released"
        },
        "director": {
            "type": "string",
            "description": "The director of the movie"
        },
        "rating": {
            "type": "number",
            "description": "The movie's rating out of 10"
        }
    },
    "required": ["title", "year", "director", "rating"]
}
python
structured_model = model.with_structured_output(
    json_schema,
    method="json_schema"
)
python
response = structured_model.invoke("给出盗梦空间的信息")
print(response)
print(type(response))

输出如下

text
{'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰','rating': 9.3}
<class 'dict'>

按照字典形式输出，不报错。

3.3.4 @dataclass ​

python
from dataclasses import dataclass
python
from langchain_deepseek import ChatDeepSeek
python
model = ChatDeepSeek(
    model="deepseek-v4-flash",
    api_base="http://localhost:8889",
    api_key=SecretStr("<KEY>")
)
python
@dataclass
class Movie():
    """
    电影的详细信息
    """
    title: str = Field(description="电影标题")
    year: int = Field(description="电影上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="电影评分，满分十分")
python
structured_model = model.with_structured_output(Movie)
python
response = structured_model.invoke("给出盗梦空间的信息")
print(response)
print(type(response))

输出如下

text
{'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰','rating': 9.3}
<class 'dict'>

按照字典形式输出，不报错。

3.2.5 小 结 ​

用Pydantic定义schema，在接收到响应后会进行校验，字段不匹配则抛出异常，其余三种方式不校验。

4、获取结构化结果方式 ​

以上定义输出结构的四种模式中，我们都是通过调用“with_structured_output”来获取结构化输出结果，除了这种方式外，还可以通过使用输出解释器来获取结构化输出结果。下面介绍这两种获取结构化结果的方式。

4.1 使用with_structured_output ​

这种方式是最新、最简洁的API，直接让模型“理解”你需要的数据结构，并返回解析好的对象。

此外，我们可以在with_structured_output方法中传入include_raw=True 参数，表示返回解析前的原 始AIMessage ，从而访问令牌用量等元数据。

举例：

python
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
python
# 从.env文件中加载环境变量
load_dotenv(override=True)
python
CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")
python
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=CLOSEAI_API_KEY,
    base_url=CLOSEAI_BASE_URL
)
python
from pydantic import BaseModel, Field
from rich import print as rprint
python
class Movie(BaseModel):
    """电影信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
    director: str = Field(description="导演")
    rating: float = Field(description="评分（10分制）")
python
# 设置模型结构化输出
model_with_structure  = model.with_structured_output(Movie,include_raw=True)
python
# 调用模型并获取结构化输出
resp = model_with_structure.invoke("给我介绍下电影《星际穿越》")
python
print(type(resp))
rprint(resp)

以上代码运行结果如下：

text
<class 'dict'>
{
    'raw': AIMessage(content = '', additional_kwargs = {
        'refusal': None
    }, response_metadata = {
        'token_usage': {
            'completion_tokens': 93,
            'prompt_tokens': 373,
            'total_tokens': 466,
text
            'completion_tokens_details': None,
            'prompt_tokens_details': {
                'audio_tokens': None,
                'cached_tokens': 320
            },
            'prompt_cache_hit_tokens': 320,
            'prompt_cache_miss_tokens': 53
        },
        'model_provider': 'deepseek',
        'model_name': 'deepseek-chat',
        'system_fingerprint': 'fp_eaab8d114b_prod0820_fp8_kvcache',
        'id': '679228cd-956f-49bf-85b5-38ec684baca3',
        'finish_reason': 'tool_calls',
        'logprobs': None
    }, id = 'lc_run--019bbc7f-2c63-7611-a2e6-c413c2d11e52-0',tool_calls = [{
        'name': 'Movie',
        'args': {
            'title': '星际穿越',
            'year': 2014,
            'director': '克里斯托弗·诺兰',
            'rating': 9.3
        },
        'id': 'call_00_WCnHeO1OYKS1gjuJfvmB1QHz',
        'type': 'tool_call'
    }], invalid_tool_calls = [], usage_metadata = {
        'input_tokens': 373,
        'output_tokens': 93,
        'total_tokens': 466,
        'input_token_details': {
            'cache_read': 320
        },
        'output_token_details': {}
    }),
    'parsed': Movie(title = '星际穿越', year = 2014, director = '克里斯托弗·诺兰', rating = 9.3),
    'parsing_error': None
}

输出包含了完整的输出响应，包含三个字段

  • raw：返回的原始AIMessage。

  • parsed：解析后的输出

  • parsing_error：解析错误，当前用的是Pydantic，校验，格式不符合schema会导致报错。其它三种方式不符合schema不会导致报错。

4.2 使用输出解析器(不推荐) ​

这种方法更传统，依赖于在提示词中明确指示模型输出特定格式的文本，然后使用解析器进行转换。

其流程是：提示词指导 (引导生成指定类型）→ 模型生成文本 → 解析器转换。

举例：

python
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
python
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
python
# 1. 创建提示词模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system","回答用户问题,必须始终输出一个包含title(电影标题)和year(上映年份)的JSON 对象"),
    ("human","问题：{question}")
])
python
# 2. 模型初始化
# 从.env文件中加载环境变量
load_dotenv(override=True)
python
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)
python
# 3. 定义结构
class Movie(BaseModel):
    """电影信息"""
    title: str = Field(description="电影标题")
    year: int = Field(description="上映年份")
python
# 4. 创建输出解析器
parser = JsonOutputParser(pydantic_object=Movie)
python
# 5. 创建链
chain = prompt_template | model | parser
python
# 6. 调用（返回字典）
response = chain.invoke({"question": "介绍电影《盗梦空间》"})
#
# response =parser.invoke(model.invoke(prompt_template.invoke({"question":"介绍电影《盗梦空间》"})))
python
print(response)
text
{'title': '盗梦空间', 'year': 2010}