1
from typing import TypedDict, List, Annotated
2

3
# 使用 TypedDict 定义嵌套结构
4
class Actor(TypedDict):
5
    """演员情况"""
6

7
    name: Annotated[str, "演员姓名"]
8
    role: Annotated[str, "饰演的角色"]
9

10
class Movie(TypedDict):
11
    """电影情况"""
12

13
    title: Annotated[str, "电影标题"]
14
    year: Annotated[int, "上映年份"]
15
    director: Annotated[str, "导演"]
16
    cast: Annotated[List[Actor], "演员列表"]      # ← 嵌套列表定义
17
    rating: Annotated[float, "评分"]
18

19
structured_llm = model.with_structured_output(Movie)
20

21
resp = structured_llm.invoke("给我介绍下电影《盗梦空间》")
22

23
# 访问嵌套数据：字典就用 ['键'] 一层层取
24
print(f"电影名: {resp['title']}")
25
print(f"上映年份: {resp['year']}")
26
print(f"导演: {resp['director']}")
27
print(f"演员列表:{resp['cast']}")
28
print(f"评分: {resp['rating']}")

1
json_schema = {
2
    "title": "Movie",
3
    "description": "A movie with details",
4
    "type": "object",
5
    "properties": {
6
        "title": {"type": "string", "description": "The title of the movie"},
7
        "year": {"type": "integer", "description": "The year the movie was released"},
8
        "director": {"type": "string", "description": "The director of the movie"},
9
        "rating": {"type": "number", "description": "The movie's rating out of 10"},
10
    },
11
    "required": ["title", "year", "director", "rating"],
12
}
13

14
structured_model = model.with_structured_output(
15
    json_schema,
16
    method="json_schema",          # ← 显式指定走服务器的 json_schema 模式
17
)
18

19
response = structured_model.invoke("给出盗梦空间的信息")
20
print(response)      # 返回字典：{'title': '盗梦空间', 'year': 2010, 'director': '克里斯托弗·诺兰', 'rating': 8.8}
21
print(type(response))    # <class 'dict'>

1
import json
2
import time
3
from http.server import BaseHTTPRequestHandler, HTTPServer
4

5
# 故意把 title 写成 title1，并且不返回 year
6
FAKE_ARGS = {"title1": "盗梦空间", "director": "克里斯托弗·诺兰", "rating": 9.3}
7

8
class FakeDeepSeekHandler(BaseHTTPRequestHandler):
9
    def do_POST(self):
10
        content_length = int(self.headers.get("Content-Length", 0))
11
        raw_body = self.rfile.read(content_length).decode("utf-8")
12
        json_body = json.loads(raw_body)
13

14
        # 用客户端请求里的工具名回填，模拟"模型按 Schema 返回"
15
        response = {
16
            "id": "chatcmpl-test",
17
            "object": "chat.completion",
18
            "created": int(time.time()),
19
            "model": "any",
20
            "choices": [
21
                {
22
                    "index": 0,
23
                    "message": {
24
                        "role": "assistant",
25
                        "content": "",
26
                        "tool_calls": [
27
                            {
28
                                "id": "call_1",
29
                                "type": "function",
30
                                "function": {
31
                                    "name": json_body["tools"][0]["function"]["name"],
32
                                    "arguments": json.dumps(FAKE_ARGS, ensure_ascii=False),
33
                                },
34
                            }
35
                        ],
36
                    },
37
                    "finish_reason": "tool_calls",
38
                }
39
            ],
40
            "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
41
        }
42
        data = json.dumps(response, ensure_ascii=False).encode()
43
        self.send_response(200)
44
        self.send_header("Content-Type", "application/json")
45
        self.send_header("Content-Length", str(len(data)))
46
        self.end_headers()
47
        self.wfile.write(data)
48

49
    def log_message(self, *args):
50
        pass
51

52
if __name__ == "__main__":
53
    server = HTTPServer(("127.0.0.1", 8889), FakeDeepSeekHandler)
54
    print("fake server 已启动: http://127.0.0.1:8889")
55
    server.serve_forever()