1
import os
2
from dataclasses import dataclass
3
from typing import List
4

5
from dotenv import load_dotenv
6
from langchain.chat_models import init_chat_model
7
from pydantic import BaseModel, Field
8
from typing_extensions import Annotated, TypedDict
9

10
load_dotenv(override=True)
11

12
model = init_chat_model(
13
    model="deepseek-v4-flash",
14
    model_provider="openai",
15
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
16
    api_key=os.getenv("DEEPSEEK_API_KEY"),
17
)
18

19
# ---------- 2-1 TypedDict 版 Schema ----------
20
class Actor(TypedDict):
21
    """演员情况"""
22

23
    name: Annotated[str, "演员姓名"]
24
    role: Annotated[str, "饰演的角色"]
25

26
class MovieDict(TypedDict):
27
    """电影情况"""
28

29
    title: Annotated[str, ..., "电影标题"]
30
    year: Annotated[int, ..., "上映年份"]
31
    director: Annotated[str, ..., "导演"]
32
    cast: Annotated[List[Actor], "演员列表"]
33
    rating: Annotated[float, "电影评分，满分十分"]
34

35
structured_llm = model.with_structured_output(MovieDict)
36
resp = structured_llm.invoke("给我介绍下电影《盗梦空间》")
37

38
print(type(resp))                 # <class 'dict'>：走的是字典路线
39
print(resp["title"])              # 取值一律用方括号，不是点号
40
print(resp["cast"][0]["name"])    # 嵌套的列表里也是普通字典
41
for actor in resp["cast"]:
42
    print(actor["name"], "饰演", actor["role"])
43

44
# ---------- 2-2 dataclass 版 Schema ----------
45
@dataclass
46
class MovieDc:
47
    """电影的详细信息"""
48

49
    title: str = Field(description="电影标题")
50
    year: int = Field(description="电影上映年份")
51
    director: str = Field(description="导演")
52
    rating: float = Field(description="电影评分，满分十分")
53

54
resp2 = model.with_structured_output(MovieDc).invoke("给出盗梦空间的信息")
55
print(type(resp2))                # <class 'dict'>：和 TypedDict 一样，都不返回对象
56
print(resp2["title"])
57

58
# 挑战：手写 __init__ 的普通类不能当 Schema
59
from langchain_core.utils.function_calling import convert_to_openai_tool
60

61
class PlainMovie:
62
    """手写 __init__ 的普通类"""
63

64
    def __init__(self, title: str, year: int):
65
        self.title = title
66
        self.year = year
67

68
print(convert_to_openai_tool(MovieDc)["function"]["parameters"]["properties"])
69
# {'title': {'description': '电影标题', 'type': 'string'}, ...}  ← 有类型、有描述
70
try:
71
    print(convert_to_openai_tool(PlainMovie)["function"]["parameters"]["properties"])
72
except Exception as e:
73
    # 本机（langchain-core 1.2.18）直接报错；有些版本是"不报错，但每个字段都变成空对象 {}"
74
    # —— 两种结果都说明：普通类当 Schema 等于盲抽，要用 @dataclass
75
    print("普通类转工具定义失败：", type(e).__name__)
76

77
# ---------- 2-3 include_raw 看 token 用量 ----------
78
class Movie(BaseModel):
79
    """电影信息"""
80

81
    title: str = Field(description="电影标题")
82
    year: int = Field(description="上映年份")
83
    director: str = Field(description="导演")
84
    rating: float = Field(description="评分（10分制）")
85

86
model_with_structure = model.with_structured_output(Movie, include_raw=True)
87
resp3 = model_with_structure.invoke("给我介绍下电影《星际穿越》")
88

89
print(type(resp3))                    # <class 'dict'>
90
print(list(resp3.keys()))             # ['raw', 'parsing_error', 'parsed']
91
print(resp3["parsed"])                # 解析后的 Movie 对象
92
print(resp3["raw"].usage_metadata)    # {'input_tokens': ..., 'output_tokens': ..., 'total_tokens': ...}
93
print(resp3["parsing_error"])         # 正常时是 None
94

95
# ---------- 2-4 JSON Schema 手写 + 显式强约束模式 ----------
96
json_schema = {
97
    "title": "Movie",                     # 注意：title 会被当成工具名，不能是中文
98
    "description": "A movie with details",
99
    "type": "object",
100
    "properties": {
101
        "title": {"type": "string", "description": "The title of the movie"},
102
        "year": {"type": "integer", "description": "The year the movie was released"},
103
        "director": {"type": "string", "description": "The director of the movie"},
104
        "rating": {"type": "number", "description": "The movie's rating out of 10"},
105
    },
106
    "required": ["title", "year", "director", "rating"],
107
}
108

109
structured_model = model.with_structured_output(
110
    json_schema,
111
    method="json_schema",      # ← 显式指定走服务器自己的 json_schema 强约束模式
112
)
113

114
response = structured_model.invoke("给出盗梦空间的信息")
115
print(response)
116
print(type(response))          # <class 'dict'>
117
# 提醒：DeepSeek 的模型服务不支持 json_schema 模式，这行要挑支持它的平台跑

1
# ---------- fake_server.py ----------
2
import json
3
import time
4
from http.server import BaseHTTPRequestHandler, HTTPServer
5

6
# 故意把 title 写成 title1、year 写成 year2（两个字段都错）
7
FAKE_ARGS = {"title1": "盗梦空间", "year2": 2010, "director": "克里斯托弗·诺兰", "rating": 9.3}
8

9
class FakeDeepSeekHandler(BaseHTTPRequestHandler):
10
    def do_POST(self):
11
        content_length = int(self.headers.get("Content-Length", 0))
12
        raw_body = self.rfile.read(content_length).decode("utf-8")
13
        json_body = json.loads(raw_body)
14
        print("[收到请求] tools[0].name =", json_body["tools"][0]["function"]["name"])
15
        print("[收到请求] tool_choice =", json_body.get("tool_choice"))
16

17
        response = {
18
            "id": "chatcmpl-test",
19
            "object": "chat.completion",
20
            "created": int(time.time()),
21
            "model": "any",
22
            "choices": [
23
                {
24
                    "index": 0,
25
                    "message": {
26
                        "role": "assistant",
27
                        "content": "",
28
                        "tool_calls": [
29
                            {
30
                                "id": "call_1",
31
                                "type": "function",
32
                                "function": {
33
                                    "name": json_body["tools"][0]["function"]["name"],
34
                                    "arguments": json.dumps(FAKE_ARGS, ensure_ascii=False),
35
                                },
36
                            }
37
                        ],
38
                    },
39
                    "finish_reason": "stop",      # ← 故意写成 stop，看解析会不会受影响
40
                }
41
            ],
42
            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
43
        }
44
        data = json.dumps(response, ensure_ascii=False).encode()
45
        self.send_response(200)
46
        self.send_header("Content-Type", "application/json")
47
        self.send_header("Content-Length", str(len(data)))
48
        self.end_headers()
49
        self.wfile.write(data)
50

51
    def log_message(self, *args):
52
        pass
53

54
if __name__ == "__main__":
55
    HTTPServer(("127.0.0.1", 8889), FakeDeepSeekHandler).serve_forever()
56

57
# ---------- client.py（另开一个终端，先启动 fake_server 再运行它）----------
58
from langchain_deepseek import ChatDeepSeek
59
from pydantic import BaseModel, Field, SecretStr
60
from typing_extensions import Annotated, TypedDict
61

62
model = ChatDeepSeek(
63
    model="deepseek-v4-flash",
64
    api_base="http://localhost:8889",
65
    api_key=SecretStr("<KEY>"),
66
)
67

68
# 第 2 步：Pydantic 模式 → 抛 ValidationError（2 个 Field required）
69
class MovieModel(BaseModel):
70
    """电影的详细信息"""
71

72
    title: str = Field(description="电影标题")
73
    year: int = Field(description="电影上映年份")
74
    director: str = Field(description="导演")
75
    rating: float = Field(description="电影评分，满分十分")
76

77
try:
78
    resp = model.with_structured_output(MovieModel).invoke("给出盗梦空间的信息")
79
    print(resp, type(resp))
80
except Exception as e:
81
    print("Pydantic 抛异常:", type(e).__name__)
82
    print(str(e)[:200])
83
    # 2 validation errors for MovieModel
84
    # title → Field required；year → Field required
85

86
# 第 3 步：TypedDict 模式 → 原样返回错误键名
87
class MovieDict(TypedDict):
88
    """电影的详细信息"""
89

90
    title: Annotated[str, ..., "电影标题"]
91
    year: Annotated[int, ..., "电影上映年份"]
92
    director: Annotated[str, ..., "导演"]
93
    rating: Annotated[float, ..., "电影评分，满分十分"]
94

95
resp = model.with_structured_output(MovieDict).invoke("给出盗梦空间的信息")
96
print("TypedDict 返回:", resp, type(resp))
97
# {'title1': '盗梦空间', 'year2': 2010, 'director': '克里斯托弗·诺兰', 'rating': 9.3} <class 'dict'>
98
# 结论：finish_reason 写成 "stop" 完全不影响解析——只要 tool_calls 里有内容就照单全收