第 3 章 · API 实战与结构化输出

本章目标：把模型调用的"工程细节"吃透——返回结构、流式输出、异常处理、结构化输出，然后做出第一个真正的工具。

3.1 环境准备

只装一个包。DeepSeek 完全兼容 OpenAI 协议，直接用 OpenAI 的 SDK 就行：

pip install openai

准备你的 API key：

打开 https://platform.deepseek.com，注册并实名认证。
左侧找到「API Keys」，创建一个 key，形如 sk-xxxxxxxx。
把 key 存成环境变量，千万别硬编码进代码或提交到 git：

import os
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "sk-你的key")

生产环境用环境变量或密钥管理服务；学习阶段直接写个变量即可，但记得别把 key 传进公开仓库。

3.2 第一次调用，看懂返回结构

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-你的key"),
    base_url="https://api.deepseek.com",
)

response = client.chat.completions.create(
    model="deepseek-v4-flash",              # v4-flash 对话模型；更强推理可换 deepseek-v4-pro 或开启思考模式
    messages=[
        {"role": "system", "content": "你是一个乐于助人的助手。"},
        {"role": "user", "content": "用一句话介绍你自己。"},
    ],
    stream=False,
)

# 模型说的话在这里
print(response.choices[0].message.content)

返回的 response 对象里，几个你以后反复用到的字段：

字段	含义
`response.choices[0].message.content`	模型输出的正文（最常用）
`response.choices[0].message.role`	输出者角色，一般是 `assistant`
`response.choices[0].finish_reason`	结束原因：`stop`（正常）/`length`（被 max_tokens 截断）/其他
`response.usage.prompt_tokens`	本次输入的 token 数
`response.usage.completion_tokens`	本次输出的 token 数
`response.usage.total_tokens`	两者之和（计费依据）

finish_reason 特别重要：如果它是 length，说明答案被 max_tokens 砍断了，你需要调大上限或精简输入。

一个实用的调试习惯：每次调用后打印 usage，你就能直观感受到"这条 prompt 花了多少 token、多少钱"。这是后面做成本控制（第 6 章）的基础。

3.3 流式输出（SSE）：让文字"打字机"式出现

上面 stream=False 是"等模型全部生成完，一次性返回"。这在真实产品里体验很差——用户盯着白屏等好几秒。

流式输出让模型生成一个 token 就吐一个，前端逐字显示，体验极佳。几乎所有 AI 应用都用它。

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-你的key"), base_url="https://api.deepseek.com")

stream = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "写一首关于秋天的五言绝句。"}],
    stream=True,                        # 关键：开启流式
)

for chunk in stream:
    # 每个 chunk 携带一小段增量文本
    delta = chunk.choices[0].delta
    content = getattr(delta, "content", None)
    if content:
        print(content, end="", flush=True)   # 不换行，边生成边打印

理解它：流式模式下，返回的是一个"迭代器"，每次迭代拿到一小块新生成的文字（delta.content）。你把它拼起来就是完整答案。

工程上，流式输出通常配合 SSE（Server-Sent Events）把内容从后端推到前端。你在浏览器里看到 ChatGPT 逐字蹦出来，就是这个原理。

3.4 异常处理：模型调用一定会出问题

调用外部 API，出错是常态，不是意外。至少要处理这几种：

网络超时：请求卡住。
限流（Rate Limit）：请求太频繁被拒。
鉴权失败：key 错了。
模型输出不符合预期：返回了非 JSON、格式错乱。

一个稳健的调用封装：

import time
import os
from openai import OpenAI, APIError, RateLimitError, APITimeoutError

client = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-你的key"), base_url="https://api.deepseek.com")

def chat_with_retry(messages, max_retries=3):
    """带重试的对话调用，网络/限流问题自动重试。"""
    for attempt in range(max_retries):
        try:
            resp = client.chat.completions.create(
                model="deepseek-v4-flash",
                messages=messages,
                timeout=60,             # 60 秒超时
            )
            return resp.choices[0].message.content
        except (RateLimitError, APITimeoutError) as e:
            wait = 2 ** attempt        # 指数退避：1s、2s、4s
            print(f"请求失败({type(e).__name__})，{wait} 秒后重试...")
            time.sleep(wait)
        except APIError as e:
            print(f"API 错误：{e}")
            break                      # 鉴权失败等，重试无意义
    return None

要点：限流和超时用"指数退避"重试；鉴权、参数错误直接失败，别傻重试。

3.5 结构化输出：让模型吐"程序能用的数据"

前面所有例子里，模型输出的是"给人看的自然语言"。但要让 AI 成为你系统里的一个组件，它得输出"程序能解析的数据"——通常是 JSON。

这是"AI 应用开发"和"玩 ChatGPT"的分水岭。两种做法：

做法 A：JSON 模式（response_format）

resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {
            "role": "system",
            "content": "你只输出 JSON，不要任何其他文字。"
        },
        {
            "role": "user",
            "content": (
                "把下面这句话里的关键信息提取成 JSON：\n"
                "'2026年8月20日，禾丰科技智粮系统入库玉米 5000 公斤。'\n"
                "字段：date(日期)、company(公司)、weight_kg(重量)、grain(品种)。"
            ),
        },
    ],
    response_format={"type": "json_object"},   # 强制 JSON 输出
)

import json
data = json.loads(resp.choices[0].message.content)
print(data)          # {'date': '2026-08-20', 'company': '禾丰科技', ...}
print(data["weight_kg"])

注意两个坑：

用 response_format={"type":"json_object"} 时，prompt 里必须出现 "json" 这个词（上面 system 里写了），否则可能报错。
模型偶尔还是会输出坏 JSON。所以 json.loads 一定要包在 try/except 里，失败了重试或降级。

做法 B：用 Pydantic 校验（更稳，推荐生产用）

先定义你要的数据结构，再用模型输出去"套"，套不上就重试：

from pydantic import BaseModel, Field
import json

class GrainRecord(BaseModel):
    date: str = Field(description="入库日期 YYYY-MM-DD")
    company: str = Field(description="公司名")
    weight_kg: float = Field(description="入库重量，单位公斤")
    grain: str = Field(description="粮食品种")

# 让模型按这个 schema 输出（把字段和含义喂给模型）
schema_hint = '{"date":"2026-08-20","company":"禾丰科技","weight_kg":5000,"grain":"玉米"}'

resp = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {"role": "system", "content": "你只输出 JSON，不要其他文字。"},
        {"role": "user", "content": (
            f"提取关键信息，输出 JSON，字段格式参考：{schema_hint}\n"
            "文本：'2026年8月20日，禾丰科技智粮系统入库玉米 5000 公斤。'"
        )},
    ],
    response_format={"type": "json_object"},
)

try:
    record = GrainRecord.model_validate_json(resp.choices[0].message.content)
    print("校验通过：", record.model_dump())
except Exception as e:
    print("输出不符合预期，需要重试或兜底：", e)

为什么 Pydantic 这套更好：它把"模型输出"变成了"经过类型校验的对象"。weight_kg 一定是 float，date 一定是 str——下游代码可以放心用。这本质上就是给不可靠的 AI 输出套上了一层可靠的类型约束，是你作为工程师最大的价值所在。

3.6 实战项目：自然语言查数据

现在做一个真正有用的东西：用户用大白话问数据，系统转成 SQL 去查库，返回结果。

这是你转型路上的第一个"里程碑项目"，也是最能让你体会到"AI 组件"价值的例子。

完整代码

import os
from openai import OpenAI
import sqlite3

client = OpenAI(api_key=os.environ.get("DEEPSEEK_API_KEY", "sk-你的key"), base_url="https://api.deepseek.com")

# 1. 准备一张示例表（真实项目里换成你的 MySQL/PG，用对应驱动即可）
conn = sqlite3.connect(":memory:")
conn.execute("""
CREATE TABLE grain_inbound (
    id INTEGER PRIMARY KEY,
    batch_no TEXT,
    grain_type TEXT,
    weight_kg REAL,
    inbound_time TEXT
)""")
conn.executemany(
    "INSERT INTO grain_inbound (batch_no, grain_type, weight_kg, inbound_time) VALUES (?,?,?,?)",
    [
        ("B20260801", "玉米", 5000, "2026-08-01 09:00"),
        ("B20260802", "小麦", 3000, "2026-08-05 10:00"),
        ("B20260803", "玉米", 7000, "2026-08-10 11:00"),
        ("B20260804", "水稻", 2000, "2026-08-15 14:00"),
    ],
)
conn.commit()

# 2. 表结构描述（喂给模型，让它知道有哪些字段）
TABLE_SCHEMA = """
表名：grain_inbound
字段：
- id (整数, 主键)
- batch_no (文本, 批次号)
- grain_type (文本, 粮食品种)
- weight_kg (实数, 入库重量, 公斤)
- inbound_time (文本, 入库时间, 格式 YYYY-MM-DD HH:MM)
"""

def text_to_sql(question: str) -> str:
    """把自然语言问题翻译成 SQLite SQL。"""
    resp = client.chat.completions.create(
        model="deepseek-v4-flash",
        temperature=0,                  # SQL 要确定，温度设 0
        messages=[
            {
                "role": "system",
                "content": (
                    "你是数据库专家。根据给定表结构，把用户问题翻译成一条 SQLite 查询语句。\n"
                    f"表结构如下：\n{TABLE_SCHEMA}\n"
                    "要求：只输出 SQL 本身，不要解释，不要代码块，不要分号。"
                ),
            },
            {"role": "user", "content": question},
        ],
    )
    return resp.choices[0].message.content.strip()

def ask_data(question: str):
    sql = text_to_sql(question)
    print(f"[生成SQL] {sql}")
    try:
        rows = conn.execute(sql).fetchall()
        print(f"[查询结果] {rows}")
        return rows
    except Exception as e:
        print(f"[执行失败] {e}")
        return None

# 3. 试试看
ask_data("玉米一共入库了多少公斤？")
ask_data("8月10号之后入库了几批？")
ask_data("每种粮食各入库了多少？")

这段代码值得你停下来想的三件事

temperature=0：SQL 是确定性任务，随机性越接近 0 越好。
表结构喂进 system：这是"给模型上下文"的典型姿势——它不知道你的库长啥样，你必须告诉它。
SQL 要程序执行，不是给人看：模型只负责"翻译"，真正查库的是你的 conn.execute。模型生成 SQL 永远不可直接信任，生产上必须加白名单校验（只允许 SELECT、只允许查询指定表、过滤危险关键字），防止注入和越权。

3.7 本章小结与练习

你该记住的：

choices[0].message.content 是正文，usage 是计费，finish_reason 是结束原因。
流式输出用 stream=True 逐块读 delta.content，是所有 AI 应用体验的关键。
外部调用必加超时 + 限流重试（指数退避）。
结构化输出（JSON 模式 + Pydantic 校验）是把 AI 变成"可用组件"的核心手段。

练习：

把 3.6 的项目跑通，然后改表结构为你的真实业务表（比如订单表、用户表），加 2~3 个查询。
给 text_to_sql 加一个"SQL 安全校验"函数：只放行 SELECT，且表名必须在白名单里。
把 3.6 改成流式输出版本，观察 SQL 是怎么一点点"长出来"的。

下一章：解决模型"不知道你的私有数据、会乱编"的问题——RAG 检索增强生成，做出企业知识库问答。