书架 · Python 学习系列 · 04 · LLM 应用开发（API / SDK 调用）下一章：05 Agent 开发 →

04 · LLM 应用开发（API / SDK 调用）

本章目标：不用任何 Agent 框架，直接用 openai SDK 完成：普通对话、多轮对话、流式输出、结构化输出、工具调用、多模态、embedding + RAG 最小实现。这是所有上层框架的地基（第 05 章会证明）。

前置：01、02；03 的异步部分建议先看。

📌 本章所有接入参数（base_url / 模型名）均核对自各家官方文档，访问日期 2026-09-05，出处见文末。模型迭代快，用前请点出处链接复核。

目录

1. 核心概念：messages / token / 采样参数
2. 一套 SDK 通吃：OpenAI 兼容接口
3. 第一次调用与多轮对话
4. 流式输出
5. 结构化输出
6. 工具调用（function calling）初体验
7. 多模态：图片输入
8. Embedding 与 RAG 最小实现
9. 工程问题：超时、重试、错误、成本
10. 自测清单
11. 小练习
12. 参考与出处

1. 核心概念：messages / token / 采样参数

调 LLM 的 HTTP 接口本质就一件事：POST 一个 messages 数组到 /chat/completions，拿回一个回复。

// messages：对话历史，按 role 分角色
[
  { "role": "system",    "content": "你是一个严谨的助理" },   // 全局人设，优先级最高
  { "role": "user",      "content": "什么是 GIL？" },          // 用户说的话
  { "role": "assistant", "content": "GIL 是全局解释器锁……" },  // 模型历史回复
  { "role": "tool",      "content": "42" }                     // 工具执行结果（第 6 节）
]

• token：模型处理文本的计费与长度单位，中文 1 字 ≈ 1~2 token，英文 1 词 ≈ 1.3 token。请求返回的 usage 字段带精确计数
• temperature / top_p：采样随机度。0 ≈ 每次都选最高概率词（稳定，适合抽取/分类/结构化输出）；0.7~1.0 更发散（适合创作）。一般调一个即可
• max_tokens（部分新模型叫 max_completion_tokens）：限制回复长度上限

☕ 给 Java 开发者的类比：把模型想成一个无状态的 HTTP 接口——它没有"会话"概念，每次请求都要把完整历史 messages 重新发过去（所以才有第 05 章的"上下文管理"问题）。所谓"多轮对话"，就是你自己在内存/Redis 里维护这个数组。

2. 一套 SDK 通吃：OpenAI 兼容接口

openai 官方 Python SDK 是事实上的通用客户端：任何兼容 OpenAI Chat Completions 协议的服务，改 base_url + api_key + model 三个参数即可接入。以下为官方文档核实的三家（访问日期 2026-09-05）：

uv add openai

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],   # 永远从环境变量读，严禁写进代码
    base_url="https://api.deepseek.com",      # ← 换服务商只改这三处
)
# model 参数填谁家的模型名就是谁

⚠️ 两个工程细节（来自官方文档的坑）： 1. 百炼的 API Key 按地域绑定：用北京的 key 调弗吉尼亚 endpoint 返回 401 invalid_api_key——看起来像 key 失效，实际是地域不匹配（官方页明确说明，见上表出处） 2. 智谱官方要求 OpenAI SDK ≥ 1.0.0（旧版有兼容问题） 3. OpenAI 官方另有新一代 Responses API（2025-03 随 Agents SDK 一起发布）；但 Chat Completions 仍是国产模型兼容的事实标准，本套资料以它为主。依据：openai-python 官方仓库同时维护两套 API，见文末。

3. 第一次调用与多轮对话

3.1 单轮

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com",
)

resp = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[
        {"role": "system", "content": "你是一个 Python 助教"},
        {"role": "user", "content": "用一句话解释生成器"},
    ],
    # temperature=0.7,          # 可选
)
print(resp.choices[0].message.content)     # 模型回复文本
print(resp.usage.total_tokens)             # 本次消耗 token 数

返回对象是 Pydantic 模型，resp.model_dump_json() 可看全量 JSON（排障神器）。

3.2 多轮：自己维护 messages

import os
from openai import OpenAI

# 接 3.1 的 client（换服务商只改这两行）
client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com")

def chat(messages: list[dict]) -> str:
    resp = client.chat.completions.create(
        model="deepseek-v4-pro",
        messages=messages,
    )
    return resp.choices[0].message.content

messages = [{"role": "system", "content": "你是简洁的助手"}]

while True:
    user_input = input("你: ")
    if user_input in {"exit", "quit"}:
        break
    messages.append({"role": "user", "content": user_input})      # ① 记录用户输入
    reply = chat(messages)                                        # ② 连同历史一起发送
    messages.append({"role": "assistant", "content": reply})      # ③ 记录模型回复
    print(f"助手: {reply}")
# 就这么简单：多轮对话 = 追加 messages 数组，没有任何魔法

4. 流式输出

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com")

stream = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": "写一首关于程序员的诗"}],
    stream=True,                                     # ← 打开流式
    stream_options={"include_usage": True},          # 最后一个 chunk 附带 token 用量
)

full = []
for chunk in stream:
    delta = chunk.choices[0].delta                   # 增量内容在 delta 里
    if delta.content:                                # 结束 chunk 的 content 可能为 None
        full.append(delta.content)
        print(delta.content, end="", flush=True)     # 逐字打印的打字机效果
print("".join(full))

异步版（FastAPI/批量并发场景，见第 03/07 章）：

import os
from openai import AsyncOpenAI

aclient = AsyncOpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                      base_url="https://api.deepseek.com")

async def chat_stream(prompt: str):
    stream = await aclient.chat.completions.create(
        model="deepseek-v4-pro",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    async for chunk in stream:                       # async for 消费异步流
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content     # 做成异步生成器 → 第 07 章 SSE 直接用

☕ 流式就是服务端把一个 JSON 拆成 N 个 SSE chunk 依次推给你（data: {...}\n\n 格式）。SDK 帮你做了拼装，你拿到的是生成器——本质是"服务器推、客户端生成器收"，和第 02 章的生成器知识完全对上。

5. 结构化输出

让模型"必须返回能被程序解析的 JSON"，是 AI 应用落地的关键一步（抽取、分类、填表）。

方式一：JSON mode + Pydantic 手工解析（兼容面最广）

from pydantic import BaseModel, ValidationError

class BookInfo(BaseModel):          # Pydantic：运行时数据校验库（第 07 章细讲）
    title: str
    authors: list[str]
    year: int | None = None

resp = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[
        {"role": "system", "content": "你是图书信息抽取器，只输出 JSON。"},
        {"role": "user", "content": "抽取《三体》的信息"},
    ],
    response_format={"type": "json_object"},   # JSON 输出模式（各家支持度见其文档）
    temperature=0,                            # 结构化输出建议 0
)
try:
    book = BookInfo.model_validate_json(resp.choices[0].message.content)
    print(book.title, book.year)
except ValidationError as e:
    print("模型输出不合法:", e)                 # 校验失败要兜底（重试/降级）

方式二：openai SDK 的 .parse()（官方 SDK 内置 helper，一步到位）

import os
from openai import OpenAI
from pydantic import BaseModel

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com")

class BookInfo(BaseModel):        # 沿用方式一的数据类
    title: str
    authors: list[str]
    year: int | None = None

completion = client.chat.completions.parse(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": "抽取《三体》的信息"}],
    response_format=BookInfo,      # 直接传 Pydantic 类：自动生成 JSON Schema 并解析
)
if completion.choices[0].message.parsed:
    print(completion.choices[0].message.parsed.title)

.parse() 的原理：把 Pydantic 模型转成 JSON Schema 塞给模型 + 拿回文本自动 model_validate——依然是第 3、4 节那套协议的封装。出处：openai-python 官方仓库 helpers 文档（见文末）。

6. 工具调用（function calling）初体验

让模型决定调用哪个函数、生成什么参数（函数永远由你本地执行）：

import json
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com")

# ① 用 JSON Schema 向模型声明工具
tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "查询指定城市当前天气",       # description 决定模型什么时候想起它
        "parameters": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名，如：成都"},
            },
            "required": ["city"],
        },
    },
}]

# ② 本地实现（真实项目里换成调天气 API）
def get_weather(city: str) -> str:
    return json.dumps({"city": city, "temp": "26℃", "cond": "多云"}, ensure_ascii=False)

messages = [{"role": "user", "content": "成都今天热不热？"}]

# ③ 第一次调用：模型不直接回答，而是返回"我要调工具"
resp = client.chat.completions.create(
    model="deepseek-v4-pro", messages=messages, tools=tools,
)
msg = resp.choices[0].message
print(msg.tool_calls)     # [ChatCompletionMessageToolCall(...)] ← 模型的"工具调用请求"

if msg.tool_calls:
    messages.append(msg)                        # ④ 先把带 tool_calls 的 assistant 消息入历史
    for tc in msg.tool_calls:
        args = json.loads(tc.function.arguments)     # 模型生成的参数（JSON 字符串）
        result = get_weather(**args)                 # ⑤ 本地执行
        messages.append({                           # ⑥ 结果以 role="tool" 回填
            "role": "tool",
            "tool_call_id": tc.id,                  # 必须回带 id，模型才知道对应哪个调用
            "content": result,
        })
    # ⑦ 带着工具结果再调一次 → 模型生成最终自然语言回答
    final = client.chat.completions.create(
        model="deepseek-v4-pro", messages=messages, tools=tools,
    )
    print(final.choices[0].message.content)      # "成都现在 26℃，多云，不算热……"

这 7 步就是"工具调用"的全部：tools 声明 → 模型返回 tool_calls → 本地执行 → role="tool" 回填 → 再调一次。把它放进 while 循环，就是第 05 章/ReAct HTML 里的完整 Agent。

📌 历史注脚：OpenAI 2023-11 DevDay 起用 tools/tool_choice 取代老的 functions/function_call 参数，老教程里的写法已过时。国产兼容端点普遍支持 tools 形态，个别差异以各家文档为准。

7. 多模态：图片输入

OpenAI 兼容协议的多模态格式：content 从字符串变成分段数组：

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com")

resp = client.chat.completions.create(
    model="deepseek-v4-flash-vision-exp",   # DeepSeek 实验视觉模型（官方文档示例，见文末）
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "这张图里有什么？用一句话描述"},
            {"type": "image_url",
             "image_url": {"url": "https://example.com/cat.jpg"}},  # 也可传 base64 data URI
        ],
    }],
)
print(resp.choices[0].message.content)

📌 视觉模型各家命名不同（DeepSeek 是 deepseek-v4-flash-vision-exp，百炼是 Qwen-VL 系列），且部分视觉端点对 tools/stream 支持有限制，用前查各家文档——本章表格出处可直达。

8. Embedding 与 RAG 最小实现

Embedding = 把文本压成向量（如 1024 维浮点数组），语义相近 → 向量夹角小。据此可以做检索（RAG 的核心）、聚类、去重。

8.1 拿 embedding

import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"],
                base_url="https://api.deepseek.com")

resp = client.embeddings.create(
    model="deepseek-v4-flash",            # 各家的 embedding 模型名不同，以文档为准
    input=["什么是GIL", "全局解释器锁是什么", "今天天气不错"],
)
vectors = [d.embedding for d in resp.data]   # 每个 list[float]

8.2 RAG = 检索增强生成，五步流水线（无框架实现）

"""最小 RAG：文档 → 切块 → 向量化 → 相似度检索 → 拼 prompt。
只依赖 openai + numpy，向量存内存（生产换向量库：Qdrant/Milvus/pgvector 等）。"""
import json
import numpy as np
from openai import OpenAI

client = OpenAI(api_key=..., base_url=...)

docs = [
    "我们公司的年假制度：入职满1年5天，满3年10天，满5年15天。",
    "报销流程：先在OA提交发票，主管审批后财务7个工作日打款。",
    "服务器故障应急预案：先看监控大盘，然后通知值班SRE，最后写复盘报告。",
]

# ①② 切块（这里每条就是一个块）→ ③ 向量化
emb = client.embeddings.create(model="deepseek-v4-flash", input=docs)
matrix = np.array([d.embedding for d in emb.data])          # (3, dim) 矩阵

def search(query: str, top_k: int = 1) -> list[str]:
    q = np.array(client.embeddings.create(
        model="deepseek-v4-flash", input=[query]).data[0].embedding)
    # 余弦相似度：点积 / 模长乘积（矩阵化一次算完）
    sims = matrix @ q / (np.linalg.norm(matrix, axis=1) * np.linalg.norm(q))
    return [docs[i] for i in np.argsort(sims)[::-1][:top_k]]

def ask(question: str) -> str:
    hits = search(question, top_k=2)                        # ④ 检索最相关的块
    context = "\n".join(f"[资料{i+1}] {h}" for i, h in enumerate(hits))
    resp = client.chat.completions.create(                  # ⑤ 拼进 prompt 生成
        model="deepseek-v4-pro",
        messages=[
            {"role": "system", "content": "仅依据以下资料回答，资料没有就说不知道。\n" + context},
            {"role": "user", "content": question},
        ],
    )
    return resp.choices[0].message.content

print(ask("年假有几天？"))   # 会命中资料1

真实项目还要处理：切块策略（按段落/固定 token 滑窗）、持久化向量、混合检索（关键词+向量）、重排（rerank）。但骨架就是这 30 行，框架（LangChain 的 Retriever、LlamaIndex）只是把这些步骤组件化。

9. 工程问题：超时、重试、错误、成本

from openai import OpenAI, APIConnectionError, RateLimitError, APIStatusError

client = OpenAI(
    api_key=..., base_url=...,
    timeout=30.0,        # SDK 内置 HTTP 超时（底层 httpx）
    max_retries=3,       # SDK 内置自动重试（连接错误/429/5xx，指数退避）
)

def safe_chat(messages):
    try:
        resp = client.chat.completions.create(model="deepseek-v4-pro", messages=messages)
        u = resp.usage
        print(f"[token] prompt={u.prompt_tokens} completion={u.completion_tokens} total={u.total_tokens}")
        return resp.choices[0].message.content
    except RateLimitError:
        raise RuntimeError("限流：降低并发或加 Semaphore")     # fail fast，严禁吞异常
    except APIStatusError as e:
        raise RuntimeError(f"服务端错误 {e.status_code}: {e.message}")
    except APIConnectionError as e:
        raise RuntimeError(f"网络错误: {e}")

工程清单：

• 成本：按 usage 字段记账；批量任务先小样本估单价；缓存重复问题的答案
• 限流：并发用 asyncio.Semaphore（第 03 章）；429 走 SDK 内置退避重试
• 密钥：环境变量 / .env（uv add python-dotenv），绝不进代码和 git
• 可观测：把每次请求的 model / usage / 耗时打进日志（第 03 章 logging）

10. 自测清单

• [ ] messages 有哪几种 role？"多轮对话"是谁在维护？
• [ ] 换一家模型服务商，SDK 侧要改哪三个参数？
• [ ] 流式输出里增量文本在哪个字段？stream_options 的 include_usage 干嘛的？
• [ ] JSON mode 和 .parse() 各自怎么做结构化输出？校验失败为什么要兜底？
• [ ] 工具调用的 7 步流程背下来（tools 声明→…→再调一次）
• [ ] tool_call_id 不回带会发生什么？
• [ ] RAG 五步是什么？余弦相似度怎么算？
• [ ] SDK 哪两个构造参数解决超时和重试？

11. 小练习

练习 1：把 3.2 的多轮对话加上 deque(maxlen=20) 截断历史（第 02 章知识复用），并打印每轮 token 用量。

练习 2：写 extract(text) -> BookInfo 函数：JSON mode + Pydantic 校验 + 失败自动重试 1 次（重试时在 user 消息里附上校验错误信息让模型修正）。

练习 3：给第 6 节加第二个工具 calculate(a, b)（用 eval 前先自己实现安全四则运算），让模型自己决定查天气还是算数。

练习 4：把 8.2 的 RAG 示例改成从本地 markdown 文件加载，按 \n\n 切块，检索 top_k=3。

12. 参考与出处

以下全部为官方一手来源（访问日期：2026-09-05）：

⬅️ 返回目录 ｜ ➡️ 下一章：05-Agent开发 ｜ 🔥 配套交互页：ReAct 深剖（HTML）