文本切分 token

如何估算 Token 数量？

• 快速估算公式：
  • 中文文本：字数 × 0.5 ~ 1
  • 英文文本：字数 × 0.25
  • 代码：字符数 / 4

API Key

在 Python 项目中使用

对于 Python 项目，也可以使用 python-dotenv 库动态加载 .env 文件：

from dotenv import load_dotenv
import os

# 加载 .env 文件
load_dotenv(override=True)

# 读取环境变量
api_key = os.getenv("OPENROUTER_API_KEY")

采用这种方式后，所有的 API 密钥都集中管理在 .env 文件中，代码仓库中不会出现任何敏感信息，既安全又便于维护。

OpenAI SDK 兼容格式

SDK 是对 API 的封装，让开发者不需要手写复杂的 HTTP 请求，而是用简洁的代码就能调用 API。

不用 SDK（手写 HTTP 请求）

import requests
import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv(override=True)
# 获取 API Key
api_key = os.getenv("OPENROUTER_API_KEY")

# 使用 OpenRouter 的 base_url
response = requests.post(
    "https://openrouter.ai/api/v1/chat/completions",  # OpenRouter 地址
    headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
    json={"model": "openai/gpt-5", "messages": [{"role": "user", "content": "你好"}]}
)

result = response.json()
print("不用 SDK 的结果：", result['choices'][0]['message']['content'])

使用 SDK（简洁明了）

from openai import OpenAI

client = OpenAI(
    api_key=api_key,
    base_url=" https://openrouter.ai/api/v1"  # 添加 base_url
)

response = client.chat.completions.create(
    model="openai/gpt-5",
    messages=[{"role": "user", "content": "你好"}]
)
print("使用 SDK 的结果：", response.choices[0].message.content)

可以看到，SDK 帮我们处理了 HTTP 头部、JSON 序列化、错误处理等细节，让代码更简洁易读。openai 就是 OpenAI 官方提供的 Python SDK。

在大模型 API 领域，一个有趣的现象是：几乎所有的平台都声称"兼容 OpenAI API 格式"。这意味着什么？
OpenAI 在 2020 年发布 GPT-3 API 时，设计了一套简洁的调用接口。其核心是 chat.completions.create() 方法，接受 model 和 messages 两个必填参数。这个设计因其简洁性和扩展性，逐渐成为了行业事实标准。

目前，DeepSeek、阿里百炼、智谱清言、OpenRouter 等平台都支持这套格式。这带来了巨大的便利性——我们只需要修改两个配置项,就能在不同平台之间无缝切换。

Chat Completions API 接入大模型

中转站

OpenRouter

阿里百炼

Chat Completions API 接入大模型

安装核心依赖库

conda activate ai-learn
pip install openai transformers tiktoken python-dotenv requests httpx

包说明：

• openai：OpenAI Python 官方 SDK，用于调用大模型 API
• transformers：Hugging Face Transformers 库，用于加载 tokenizer
• tiktoken：OpenAI 官方 tokenizer 工具
• python-dotenv：环境变量加载工具，加载 .env 文件中的环境变量
• requests：HTTP 请求库
• httpx：HTTP 客户端库（支持异步）

1. 查看当前版本

# 查看已安装的依赖包版本
import importlib.metadata

# 定义需要检查的包列表
packages = ['openai', 'transformers', 'tiktoken', 'python-dotenv', 'requests', 'httpx']

# 循环检查每个包的版本
for package in packages:
    try:
        version = importlib.metadata.version(package)
        print(f'{package:<20} v{version}')
    except importlib.metadata.PackageNotFoundError:
        print(f'{package:<20} 未安装')

openai               v 2.16.0
transformers         v 5.0.0
tiktoken             v 0.12.0
python-dotenv        v 1.2.1
requests             v 2.32.5
httpx                v 0.28.1

如果导入成功并显示版本号，说明依赖库已经正确安装。openai 库的版本应该在 1.0 以上，这是支持最新 API 格式的版本。

2. 配置 .env 文件

# 示例：创建 .env 文件（实际使用时请替换为真实的 API Key）
env_content = """
# OpenRouter API Key
OPENROUTER_API_KEY=sk-or-v1-xxxxxxxxxxxxxxxx

# DeepSeek API Key
DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxx

# 阿里云百炼 API Key
DASHSCOPE_API_KEY=sk-xxxxxxxxxxxxxxxx

# 智谱 AI API Key
ZHIPUAI_API_KEY=xxxxxxxxxxxxxxxx
"""

# 写入 .env 文件
with open('.env', 'w', encoding='utf-8') as f:
    f.write(env_content.strip())

print("✅ .env 文件已创建，请替换为你的真实 API Key")

加载环境变量：

from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv(override=True)

# 验证环境变量是否加载成功
keys_to_check = ["OPENROUTER_API_KEY", "DEEPSEEK_API_KEY", "DASHSCOPE_API_KEY", "ZHIPUAI_API_KEY"]

for key_name in keys_to_check:
    key_value = os.getenv(key_name)
    if key_value and not key_value.startswith("xxx"):
        print(f"✅ {key_name}: {key_value[:10]}... (已加载)")
    else:
        print(f"⚠️ {key_name}: 未配置或使用占位符")

✅ OPENROUTER_API_KEY: sk-proj-jv... (已加载)
✅ DEEPSEEK_API_KEY: sk-ddea 2 fd... (已加载)
✅ DASHSCOPE_API_KEY: sk-2904274... (已加载)
✅ ZHIPUAI_API_KEY: 4 db 0 cf 2 aa 2... (已加载)

如果看到 ✅ 标记，说明环境变量已成功加载。如果显示 ⚠️ 警告，请检查 .env 文件中对应的 Key 是否正确填写。至此，环境准备工作全部完成，我们可以开始第一个 API 调用了。

第一个 API 调用：Hello World

1. 调用示例

from openai import OpenAI

# 创建客户端，指向 DeepSeek 平台
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

# 调用 API
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "user", "content": "你是谁？"}
    ]
)

# 提取回复内容
answer = response.choices[0].message.content
print("模型回复：")
print(answer)

这段代码完成了以下几个关键步骤：

1. 创建客户端：使用 OpenAI() 创建一个客户端对象，通过 api_key 指定身份凭证，通过 base_url 指定 DeepSeek 的 API 地址。
2. 构造请求：调用 client.chat.completions.create()，指定模型名称和消息列表。
3. 解析响应：从 response.choices[0].message.content 中提取模型生成的文本。

2. 计算 Token 调用量

我们可以通过不同的方式来对调用 api 后大模型的输入以及输出来计算 Token，能够通过 Token 的控制，来管理上下文的长度，从而控制大模型的输出和成本的控制。那么有很多框架内部集成了 Token 计算的功能，能直接通过 UI 看板的形式来观察 Token 的使用情况。

方法一：通过 API 直接获取（推荐）

所有国内平台的 API 响应都会返回实际消耗的 Token 数量，这是最准确的方式：

from openai import OpenAI

# 以 DeepSeek 为例（Qwen、GLM 用法相同，只需替换 base_url）
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"  # 或其他平台地址
)

text = "你好,世界!Hello World!"
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": text}],
    max_tokens=1  # 只生成1个token以节省费用
)

# 提取回复内容
answer = response.choices[0].message.content
print("模型回复：")
print(answer)

# 直接从响应中获取 Token 消耗
print(f"输入 Token: {response.usage.prompt_tokens}")
print(f"输出 Token: {response.usage.completion_tokens}")
print(f"总计 Token: {response.usage.total_tokens}")
# 输出示例: 输入 Token: 7, 输出 Token: 1, 总计 Token: 8

模型回复：
你好
输入 Token: 11
输出 Token: 1
总计 Token: 12

方法二：使用各平台官方 Tokenizer（本地计算）

# DeepSeek: 使用 Hugging Face transformers
from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("deepseek-ai/DeepSeek-V3.2")
text = "你好,世界!Hello World!"
tokens = tokenizer.encode(text)
print(f"Token 数量: {len(tokens)}")
# 中文约 1 字 ≈ 0.6 token

# Qwen (通义千问): 使用 Qwen tokenizer
tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-7B")
tokens = tokenizer.encode(text)
print(f"Token 数量: {len(tokens)}")
# 中文约 1.5-1.8 字 ≈ 1 token

Token 数量: 7
Token 数量: 7

方法三：OpenAI 模型使用 tiktoken

# 仅适用于 OpenAI / GPT 系列模型
import tiktoken

encoding = tiktoken.encoding_for_model("gpt-5")
text = "你好,世界!Hello World!"

tokens = encoding.encode(text)
print(f"Token 数量: {len(tokens)}")

Token 数量: 7

3. 理解消息结构：三角色对话模型

在上面的代码中，messages 参数是一个列表，包含了对话中的所有消息。每条消息都是一个字典，必须包含 role 和 content 两个字段。

OpenAI API 定义了三种角色：

• system：系统角色，用于设定 AI 的行为规范、角色定位、回复风格等。这是"幕后导演"，用户看不到，但会影响整个对话的基调。
• user：用户角色，代表人类的提问或输入。
• assistant：助手角色，代表 AI 的回复。在构造多轮对话时，需要手动添加历史回复。

让我们通过一个更完整的例子来理解这三种角色的作用：

# 使用三角色构造一个完整的对话
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一位专业的 Python 编程导师，擅长用简洁明了的语言解释复杂概念。"},
        {"role": "user", "content": "什么是列表推导式？"},
    ]
)

print("模型回复：")
print(response.choices[0].message.content)

system 消息的作用非常强大，可以用来：

• 设定角色（如"你是一位律师"、"你是一位翻译专家"）
• 规定输出格式（如"请用 JSON 格式回复"、"只回答是或否"）
• 限制回复范围（如"只回答 Python 相关问题"）
• 设定语言和风格（如"用口语化的方式回复"、"用英文回复"）

在实际应用中，合理使用 system 消息可以显著提升 AI 的回复质量和可控性。

4. 解析响应结果：理解 response 对象

API 返回的 response 对象包含了丰富的信息。让我们完整地查看一下它的结构：

# 完整查看 response 对象
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "用一句话介绍 Python"}]
)

print("=" * 60)
print("Response 对象结构：")
print("=" * 60)
print(f"模型名称: {response.model}")
print(f"响应 ID: {response.id}")
print(f"创建时间: {response.created}")
print(f"对象类型: {response.object}")
print()
print("消息内容:")
print(f"  角色: {response.choices[0].message.role}")
print(f"  内容: {response.choices[0].message.content}")
print()
print("Token 使用情况:")
print(f"  输入 Token: {response.usage.prompt_tokens}")
print(f"  输出 Token: {response.usage.completion_tokens}")
print(f"  总计 Token: {response.usage.total_tokens}")

============================================================
Response 对象结构：
============================================================
模型名称: deepseek-chat
响应 ID: 56505e88-6414-4a61-8359-8b0fa6387852
创建时间: 1769137073
对象类型: chat.completion

消息内容:
  角色: assistant
  内容: Python 是一门简洁易读、功能强大的高级编程语言。

Token 使用情况:
  输入 Token: 8
  输出 Token: 13
  总计 Token: 21

核心参数调优：temperature 与 max_tokens

如何通过参数来控制模型的行为。最重要的两个参数是 temperature 和 max_tokens ——前者控制输出的随机性和创造性，后者限制输出的最大长度。

理解并合理使用这两个参数，可以让你精确控制模型的输出风格和成本。不同的应用场景需要不同的参数配置：严肃的文档生成需要低 temperature，创意写作需要高 temperature；简短回复需要小 max_tokens，长文本生成需要大 max_tokens。

1. temperature：控制输出的随机性

temperature 参数控制模型输出的随机性，取值范围通常是 0 到 2（有些平台支持更高值）：

• temperature = 0：输出最确定，每次运行结果几乎相同，适合需要稳定输出的场景（如数据提取、代码生成）
• temperature = 0.7（默认值）：平衡了创造性和稳定性，适合大多数场景
• temperature = 1.5 或更高：输出高度随机和创造性，适合创意写作、头脑风暴

# 对比不同 temperature 的输出
prompt = "用一句话描述春天"

# 定义待测试的不同温度值，用于对比输出的随机性
temperatures = [0, 0.7, 1.5]

for temp in temperatures:
    # 调用 API，传入不同的 temperature 参数
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        temperature=temp
    )
    
    # 格式化打印输出结果，便于观察对比
    print(f"\n{'='*60}")
    print(f"Temperature = {temp}")
    print(f"{'='*60}")
    print(response.choices[0].message.content)

============================================================
Temperature = 0
============================================================
春天是万物从冻土中醒来，用新绿与暖风重写世界的季节。

============================================================
Temperature = 0.7
============================================================
春天是万物抖开冬被、争先恐后向光生长的季节。

============================================================
Temperature = 1.5
============================================================
春天是万物醒来，风与阳光都变得温柔，让所有生长都理直气壮的季节。

在实际应用中，推荐的 temperature 配置：

不同场景的 Temperature 推荐值

2. max_tokens：控制输出长度

max_tokens 参数限制模型生成的最大 Token 数量。这是控制成本的关键参数，因为输出 Token 的价格通常比输入高 3-5 倍。

需要注意的是：

• max_tokens 只是上限，模型可能生成更短的内容
• 如果回复在达到 max_tokens 时被截断，response.choices[0].finish_reason 会是 "length"
• 不同语言的 Token 消耗不同（回顾第 1.1 节：中文约 1.5-2 Token/字，英文约 0.75 Token/词）

# 测试不同 max_tokens 的效果
prompt = "详细介绍 Python 的历史发展"

# 定义不同的最大 token 限制进行测试
token_limits = [50, 200, 500]

for max_tok in token_limits:
    # 调用 API，通过 max_tokens 参数限制生成内容的长度上限
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tok      # 注意：max_tokens 仅为上限，模型可能提前结束生成
    )
    
    # 解析响应中的内容、结束原因及实际消耗的 token 数
    content = response.choices[0].message.content
    finish_reason = response.choices[0].finish_reason
    actual_tokens = response.usage.completion_tokens
    
    # 打印格式化的输出结果及元数据
    print(f"\n{'='*60}")
    print(f"max_tokens = {max_tok}")
    print(f"实际输出 Token: {actual_tokens}")
    print(f"结束原因: {finish_reason}")
    print(f"{'='*60}")
    print(content)
    
    # 判断是否因达到 token 上限而导致内容未生成完毕
    if finish_reason == "length":
        print("\n⚠️ 输出被截断！考虑增加 max_tokens")

============================================================
max_tokens = 50
实际输出 Token: 50
结束原因: length
============================================================
## Python 的历史发展

### 一、诞生与早期阶段（1980年代末-1990年代）

**1. 起源（1989年）**
- **创始人**：吉多·范罗苏姆（Guido van

⚠️ 输出被截断！考虑增加 max_tokens

============================================================
max_tokens = 200
实际输出 Token: 200
结束原因: length
============================================================
## Python 的历史发展

### 一、起源与诞生（1980年代末-1991年）

**创始人**：吉多·范罗苏姆（Guido van Rossum）
- 荷兰程序员，当时在荷兰国家数学与计算机科学研究所（CWI）工作
- 曾参与ABC语言开发，从中汲取经验

**诞生背景**：
- 1989年圣诞节期间，吉多为打发时间开始编写新语言
- 设计目标：创建一种**易于阅读、易于学习、易于维护**的语言
- 名称来源：取自英国喜剧团体Monty Python的《飞翔的马戏团》

**首次发布**：1991年2月（版本0.9.0）
- 已包含类、继承、异常处理、函数等核心特性
- 采用模块系统，支持函数式编程和面向对象编程

### 二、早期发展（1990年代）

**重要里程碑**：
- **199

⚠️ 输出被截断！考虑增加 max_tokens

============================================================
max_tokens = 500
实际输出 Token: 500
结束原因: length
============================================================
## Python 的历史发展

### **诞生背景（1980年代末）**
Python 由荷兰程序员 **吉多·范罗苏姆（Guido van Rossum）** 于 **1989年圣诞节期间** 在荷兰数学和计算机科学研究所（CWI）开始开发。其设计初衷是：
- 替代 **ABC 语言**（一种教学语言），解决其扩展性不足的问题
- 提供一种**易于阅读、简洁明了**的脚本语言
- 吸收 Unix shell 和 C 语言的优点，同时避免它们的复杂性

### **关键版本演进**

#### **1. Python 0.x 时代（1991-1994）**
- **1991年2月**：发布第一个公开版本 Python 0.9.0
- 已包含**类、继承、异常处理、函数、核心数据类型**（list, dict, str）
- 采用 **Modula-3** 的模块系统
- 使用**缩进作为语法结构**（这一特色延续至今）

#### **2. Python 1.x（1994-2000）**
- **1994年1月**：Python 1.0 发布
- 新增 **函数式编程工具**（`lambda`, `map`, `filter`, `reduce`）
- 引入**垃圾回收机制**（引用计数为主）

#### **3. Python 2.x（2000-2020）**
- **2000年10月**：Python 2.0 发布
- 加入**列表推导式**、完整的垃圾回收系统
- **2003年**：Python 2.3 引入 `set` 类型
- **2006年**：Python 2.5 加入 `with` 语句
- **2008年12月**：Python 2.6 成为最后一个主要 2.x 版本
- **2010年**：Python 2.7 发布（最终维护版）
- **2020年1月1日**：Python 2 正式停止支持

#### **4. Python 3.x（2008-至今）—— 不兼容的革命**
- **2008年12月**：Python 3.0（代号 "Python 3000"）发布
- **破坏性改变**：解决 2.x 的设计缺陷，不向后兼容
-

⚠️ 输出被截断！考虑增加 max_tokens

从输出可以看到：

• 当 max_tokens=50 时，回复明显被截断，finish_reason 为 "length"
• 当 max_tokens=200 时，可能刚好够用，或仍然被截断
• 当 max_tokens=500 时，回复通常能够完整，finish_reason 为 "stop"（模型主动结束）

在实际应用中，推荐的 max_tokens 配置策略：

• 简短回复（摘要、标题）：50-100 tokens
• 中等回复（问答、对话）：200-500 tokens
• 长文本生成（文章、报告）：1000-4000 tokens
• 如果不确定长度：可以设置较大值（如 2000），让模型自行决定何时结束

成本优化技巧：如果只需要简短回复，务必设置合理的 max_tokens，避免模型生成不必要的长文本浪费费用。

多平台无缝切换：一套代码调用所有模型

通过修改 api_key、base_url 和 model 三个参数，我们可以用同一套代码调用不同平台的模型。

1.平台配置字典

# 多平台配置字典
PLATFORM_CONFIGS = {
    "deepseek": {
        "api_key": os.getenv("DEEPSEEK_API_KEY"),
        "base_url": "https://api.deepseek.com",
        "model": "deepseek-chat"
    },
    "openrouter": {
        "api_key": os.getenv("OPENROUTER_API_KEY"),
        "base_url": "https://openrouter.ai/api/v1",
        "model": "openai/gpt-5-mini"  # 使用免费或低价模型
    },
    "dashscope": {
        "api_key": os.getenv("DASHSCOPE_API_KEY"),
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        "model": "qwen-turbo"
    },
    "zhipu": {
        "api_key": os.getenv("ZHIPUAI_API_KEY"),
        "base_url": "https://open.bigmodel.cn/api/paas/v4",
        "model": "glm-4"
    }
}

print("✅ 平台配置字典已创建")

有了这个配置字典，我们可以编写一个通用的调用函数：

def call_llm(platform_name, prompt, temperature=0.7, max_tokens=200):
    """
    通用的大模型调用函数
    
    Args:
        platform_name: 平台名称（deepseek/openrouter/dashscope/zhipu）
        prompt: 用户输入
        temperature: 温度参数
        max_tokens: 最大输出 Token 数
    
    Returns:
        模型回复内容
    """
    # 获取对应平台的配置信息
    config = PLATFORM_CONFIGS[platform_name]
    
    # 初始化 OpenAI 客户端（大多数国产大模型 API 均兼容 OpenAI 格式）
    client = OpenAI(
        api_key=config["api_key"],
        base_url=config["base_url"]
    )
    
    # 调用大模型聊天接口
    response = client.chat.completions.create(
        model=config["model"],
        messages=[{"role": "user", "content": prompt}],
        temperature=temperature,
        max_tokens=max_tokens
    )
    
    # 返回模型生成的文本内容
    return response.choices[0].message.content

# 测试：使用相同 prompt 调用不同平台
test_prompt = "用一句话解释什么是 AI"

print("使用 dashscope:")
print(call_llm("dashscope", test_prompt))
print("\n" + "="*60 + "\n")

使用 dashscope:
AI是模拟人类智能的计算机系统，能够执行需要人类智慧的任务，如学习、推理、感知和决策。

============================================================

现在切换平台变得非常简单，只需修改第一个参数即可。这个函数封装了所有平台差异，让你可以专注于业务逻辑。

2. 批量对比测试

利用配置字典和通用函数，我们可以轻松实现批量测试，对比不同平台的输出质量：

# 批量测试多个平台
test_prompt = "写一首关于程序员的打油诗"

platforms_to_test = ["deepseek", "openrouter", "dashscope"]  # 可以继续添加其他平台

for platform in platforms_to_test:
    try:
        print(f"\n{'='*60}")
        print(f"平台: {platform.upper()}")
        print(f"模型: {PLATFORM_CONFIGS[platform]['model']}")
        print(f"{'='*60}")
        
        result = call_llm(platform, test_prompt, temperature=1.0, max_tokens=150)
        print(result)
        
    except Exception as e:
        print(f"❌ 调用失败: {e}")

流式输出：打字机效果

OpenAI API 提供了 流式输出（Streaming） 功能，通过设置 stream=True，可以让模型边生成边返回内容。这不仅提升了用户体验，还能让用户在生成过程中提前终止，节省成本。

1. 基础流式输出

启用流式输出非常简单，只需在调用时加上 stream=True 参数：

import time

# 创建客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

# 流式调用
print("模型正在生成回复（流式输出）：\n")

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "用三句话介绍人工智能的发展历程"}],
    stream=True  # 启用流式输出
)

# 逐块接收并打印
for chunk in stream:
    # 提取增量内容
    delta_content = chunk.choices[0].delta.content
    
    if delta_content:
        print(delta_content, end="", flush=True)  # 实时打印，不换行
        time.sleep(0.1)  # 模拟打字机效果（可选）

print("\n\n✅ 流式输出完成")

这段代码的关键点：

1. stream=True：告诉 API 使用流式模式返回结果
2. 迭代 stream 对象：返回值是一个迭代器，每次返回一小块内容
3. chunk.choices[0].delta.content ：提取增量内容（注意是 delta 而不是 message）
4. print(..., end="", flush=True) ：实时打印不换行，flush=True 确保立即显示

2. 流式输出的完整处理

这个封装后的函数同时实现了：

• 实时显示流式输出（用户体验）
• 保存完整内容（便于后续处理）
• 返回生成结果（可用于日志、数据库存储等）

流式输出特别适合以下场景：

• 聊天机器人：用户看到逐字生成，体验更自然
• 长文本生成：用户可以边看边等，不会觉得卡顿
• 交互式应用：用户可以在生成过程中判断是否继续等待

注意：流式模式下无法直接获取 usage 信息（Token 统计），如果需要统计成本，建议在非流式模式下测试，或使用第 1.1 节介绍的[[#方法三：OpenAI 模型使用 tiktoken]]。

错误处理：优雅应对异常

在实际应用中，API 调用可能遇到各种异常：API Key 错误、余额不足、网络超时、请求频率超限等。如果不做错误处理，程序会直接崩溃，用户体验极差。

1. 常见错误类型与分类捕获

OpenAI SDK 定义了多种异常类型，我们可以分类捕获并给出不同的处理方式：

from openai import (
    OpenAI,
    AuthenticationError,  # 认证错误（API Key 无效）
    RateLimitError,       # 速率限制错误（请求过快）
    APIConnectionError,   # 网络连接错误
    APIError              # 通用 API 错误
)

def safe_call_llm(prompt, max_retries=3):
    """
    带错误处理的 API 调用
    
    Args:
        prompt: 用户输入
        max_retries: 最大重试次数
    
    Returns:
        模型回复或错误信息
    """
    # 初始化 OpenAI 客户端，配置 DeepSeek 的 API Key 和 Base URL
    client = OpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com"
    )
    
    # 循环尝试 API 调用，最多重试 max_retries 次
    for attempt in range(max_retries):
        try:
            # 发起 API 调用
            response = client.chat.completions.create(
                model="deepseek-chat",
                messages=[{"role": "user", "content": prompt}],
                timeout=30.0  # 设置超时时间（秒）
            )
            return response.choices[0].message.content
        
        except AuthenticationError as e:
            # 认证错误，无需重试
            return f"❌ API Key 无效或已过期，请检查环境变量配置"
        
        except RateLimitError as e:
            # 速率限制，等待后重试
            wait_time = 2 ** attempt  # 指数退避：1秒、2秒、4秒...
            print(f"⚠️ 请求过快，等待 {wait_time} 秒后重试...")
            time.sleep(wait_time)
            continue
        
        except APIConnectionError as e:
            # 网络错误，重试
            print(f"⚠️ 网络连接失败（第 {attempt+1}/{max_retries} 次），重试中...")
            time.sleep(1)
            continue
        
        except APIError as e:
            # 通用 API 错误
            return f"❌ API 调用失败: {str(e)}"
        
        except Exception as e:
            # 其他未知错误
            return f"❌ 未知错误: {str(e)}"
    
    return f"❌ 重试 {max_retries} 次后仍然失败，请检查网络或稍后再试"

# 测试错误处理
test_prompt = "用一句话介绍一下 Python"
result = safe_call_llm(test_prompt)
print(result)

• AuthenticationError：API Key 无效，直接返回错误提示，不重试（因为重试也无意义）
• RateLimitError：请求频率超限，使用指数退避策略重试（等待时间逐次翻倍）
• APIConnectionError：网络连接失败，等待 1 秒后重试
• APIError：通用 API 错误，返回具体错误信息
• Exception：兜底捕获所有未知错误

指数退避（Exponential Backoff）是一种常用的重试策略：首次重试等待 1 秒，第二次等待 2 秒，第三次等待 4 秒……这样可以避免在高峰期持续发送请求加剧服务器压力。

2. 常见错误场景与排查方法

Chat Completions API 进阶使用

这一章我们会学习五个进阶能力：多轮对话、Function Calling、多模态输入、提示词工程、异步批处理。

这些能力是构建实用 AI 应用的关键。

• 多轮对话让 AI 能够记住上下文，实现连贯的交流；
• Function Calling 让 AI 能够调用外部工具，突破纯文本生成的限制；
• 多模态输入让 AI 能够"看图说话"，理解视觉信息；
• 提示词工程教你如何写出高质量的 Prompt，最大化模型能力；
• 异步批处理则通过并发调用大幅提升效率。

掌握这些技能后，你将能够构建真正实用的 AI 应用——从简单的聊天机器人，到能查询天气、搜索信息的智能助手，再到能分析图片、生成报告的多模态应用。让我们开始这段进阶之旅。

多轮对话：让 AI 记住上下文

实现多轮对话的核心思路是：将历史对话以 messages 列表的形式传递给 API。每次调用时，都把之前的所有消息（包括用户的提问和 AI 的回复）一起发送，这样模型就能"看到"完整的对话历史。

1. 基础多轮对话实现

# 初始化对话历史
conversation_history = [
    {"role": "system", "content": "你是一位友好的 AI 助手，擅长回答各种问题。"}
]

# 创建客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

# 第一轮对话
user_message_1 = "我叫张三，今年25岁"
conversation_history.append({"role": "user", "content": user_message_1})

# 调用 API 获取第一轮对话回复
response_1 = client.chat.completions.create(
    model="deepseek-chat",
    messages=conversation_history
)

# 提取并保存 AI 的回复内容
assistant_message_1 = response_1.choices[0].message.content

# 将 AI 的回复添加到对话历史中，以维持上下文连贯性
conversation_history.append({"role": "assistant", "content": assistant_message_1})

# 打印第一轮对话的用户输入和 AI 回复
print(f"用户: {user_message_1}")
print(f"AI: {assistant_message_1}\n")

# 第二轮对话（测试是否记住了用户信息）
user_message_2 = "我叫什么名字？"
conversation_history.append({"role": "user", "content": user_message_2})

# 调用 API 获取第二轮对话回复
response_2 = client.chat.completions.create(
    model="deepseek-chat",
    messages=conversation_history
)

# 提取并保存 AI 的回复内容
assistant_message_2 = response_2.choices[0].message.content
conversation_history.append({"role": "assistant", "content": assistant_message_2})

print(f"用户: {user_message_2}")
print(f"AI: {assistant_message_2}\n")

# 查看完整的对话历史
print("=" * 60)
print("完整对话历史：")
print("=" * 60)
for i, msg in enumerate(conversation_history):
    print(f"{i}. [{msg['role']}] {msg['content']}")

用户: 我叫张三，今年25岁
AI: 你好张三！很高兴认识你。25岁正是充满活力和无限可能的年纪呢！😊  
最近在忙些什么呢？工作、学习，还是有什么特别的计划或爱好吗？

用户: 我叫什么名字？
AI: 你刚才告诉我，你叫**张三**！😊  
需要我帮你记住什么其他信息吗？或者想聊聊名字相关的趣事？

============================================================
完整对话历史：
============================================================
0. [system] 你是一位友好的 AI 助手，擅长回答各种问题。
1. [user] 我叫张三，今年25岁
2. [assistant] 你好张三！很高兴认识你。25岁正是充满活力和无限可能的年纪呢！😊  
最近在忙些什么呢？工作、学习，还是有什么特别的计划或爱好吗？
3. [user] 我叫什么名字？
4. [assistant] 你刚才告诉我，你叫**张三**！😊  
需要我帮你记住什么其他信息吗？或者想聊聊名字相关的趣事？

这段代码展示了多轮对话的核心逻辑：

1. 初始化对话历史：创建一个列表 conversation_history，包含 system 消息
2. 每次用户提问前：将用户消息追加到 conversation_history
3. 调用 API：将完整的 conversation_history 传递给模型
4. 收到回复后：将 AI 的回复也追加到 conversation_history

这样，每次调用 API 时，模型都能"看到"完整的对话历史，从而实现上下文记忆。在第二轮对话中，AI 能够正确回答"你叫张三"，说明它成功记住了第一轮对话的内容。

2. 交互式多轮对话

在实际应用中，我们通常需要一个循环，让用户可以持续输入，AI 持续回复。下面是一个更实用的交互式多轮对话示例：

def chat_loop(system_prompt="你是一位友好的 AI 助手。", max_rounds=5):
    """
    交互式多轮对话函数
    
    Args:
        system_prompt: 系统提示词
        max_rounds: 最大对话轮数
    """
    # 初始化对话历史
    conversation_history = [
        {"role": "system", "content": system_prompt}
    ]
    
    client = OpenAI(
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com"
    )
    
    print("=" * 60)
    print("多轮对话开始（输入 'quit' 退出）")
    print("=" * 60)
    
    for round_num in range(1, max_rounds + 1):
        # 获取用户输入
        user_input = input(f"\n[轮次 {round_num}] 你: ").strip()
        
        # 检查是否退出
        if user_input.lower() in ['quit', 'exit', '退出']:
            print("\n对话已结束")
            break
        
        if not user_input:
            print("输入不能为空，请重新输入")
            continue
        
        # 添加用户消息到历史
        conversation_history.append({"role": "user", "content": user_input})
        
        # 调用 API
        try:
            response = client.chat.completions.create(
                model="deepseek-chat",
                messages=conversation_history,
                max_tokens=300
            )
            
            assistant_message = response.choices[0].message.content
            
            # 添加 AI 回复到历史
            conversation_history.append({"role": "assistant", "content": assistant_message})
            
            print(f"\nAI: {assistant_message}")
            print(f"\n[Token 消耗] 输入: {response.usage.prompt_tokens}, "
                  f"输出: {response.usage.completion_tokens}, "
                  f"总计: {response.usage.total_tokens}")
        
        except Exception as e:
            print(f"\n❌ 错误: {e}")
            # 移除刚才添加的用户消息
            conversation_history.pop()
            continue
    
    return conversation_history

# 运行对话（在 Jupyter Notebook 中可以交互）
# 注意：这个函数需要用户输入，如果在 Notebook 中运行，会弹出输入框
# history = chat_loop(max_rounds=3)

print("✅ 交互式对话函数已定义，可以调用 chat_loop() 开始对话")

✅ 交互式对话函数已定义，可以调用 chat_loop() 开始对话

history = chat_loop(max_rounds=3)

============================================================
多轮对话开始（输入 'quit' 退出）
============================================================

AI: 你好张三！很高兴认识你！😊

我是DeepSeek，一个由深度求索公司开发的AI助手。我是一个纯文本模型，可以帮你解答各种问题、进行对话交流、协助处理文本任务等等。

我支持文件上传功能，可以读取图像、txt、pdf、ppt、word、excel等文件中的文字信息来帮助你。虽然我不支持多模态识别，但能处理上传文件中的文字内容。另外我还有128K的上下文长度，可以进行较长的对话。

完全免费使用，你可以通过官方应用商店下载App，或者直接在网页上使用。有什么我可以帮助你的吗？

[Token 消耗] 输入: 17, 输出: 127, 总计: 144

AI: 哈哈，虽然作为AI我没有真实的“喜好”，但我的设计目标就是**特别喜欢帮助人们编程和学习编程**！😄

我可以帮你：
- 解释编程概念和算法
- 调试代码错误
- 编写代码示例
- 学习新的编程语言
- 优化代码性能
- 讨论技术架构设计

你最近在学什么编程语言或技术栈呢？或者有没有什么编程项目需要帮助？我很乐意和你一起探讨！

[Token 消耗] 输入: 157, 输出: 96, 总计: 253

AI: 你刚才告诉我你叫**张三**，而且你提到你喜欢**编程**！😊

我记得很清楚呢：
- **姓名**：张三
- **爱好**：编程

不过如果你有其他爱好想补充，或者想聊聊具体的编程方向（比如Python、Web开发、算法等等），我都很乐意听你分享！你最近在做什么编程项目吗？

[Token 消耗] 输入: 264, 输出: 76, 总计: 340

这个函数实现了一个完整的多轮对话循环，包括：

• 输入验证：检查用户输入是否为空
• 退出机制：用户输入 quit 或 exit 可以退出
• 错误处理：如果 API 调用失败，撤销刚才添加的消息
• Token 统计：每轮对话后显示 Token 消耗

使用这个函数，你可以快速构建一个简单的聊天机器人。只需调用 chat_loop()，然后在弹出的输入框中与 AI 对话即可。

3. 上下文长度管理：滑动窗口与摘要压缩

多轮对话虽然强大，但也带来了一个严重的问题：随着对话轮数增加，conversation_history 会越来越长，消耗的 Token 也会急剧增加。

假设每轮对话平均消耗 100 个 Token（输入 + 输出），那么：

• 第 1 轮：100 tokens
• 第 2 轮：200 tokens（包含第 1 轮的历史）
• 第 3 轮：300 tokens
• 第 10 轮：1000 tokens

可以看到，Token 消耗呈线性增长，成本也随之上升。更严重的是，当对话历史超过模型的上下文限制（如 128 K tokens），API 调用会直接失败。

解决这个问题有两种常用策略：

策略一：滑动窗口（Sliding Window）

只保留最近 N 轮对话，丢弃更早的历史。这是最简单的方法：

def manage_conversation_history(history, max_turns=5):
    """
    使用滑动窗口管理对话历史
    
    Args:
        history: 对话历史列表
        max_turns: 保留的最大对话轮数（不包括 system 消息）
    
    Returns:
        压缩后的对话历史
    """
    # 提取 system 消息（通常是第一条）
    system_messages = [msg for msg in history if msg["role"] == "system"]
    
    # 提取对话消息（user 和 assistant）
    dialog_messages = [msg for msg in history if msg["role"] != "system"]
    
    # 只保留最近 max_turns 轮对话（每轮包含 user + assistant）
    # 每轮 = 2 条消息，所以保留 max_turns * 2 条
    recent_messages = dialog_messages[-(max_turns * 2):]
    
    # 重新组合：system + 最近的对话
    return system_messages + recent_messages

# 示例：模拟一个很长的对话历史
long_history = [
    {"role": "system", "content": "你是 AI 助手"},
    {"role": "user", "content": "第1轮用户消息"},
    {"role": "assistant", "content": "第1轮AI回复"},
    {"role": "user", "content": "第2轮用户消息"},
    {"role": "assistant", "content": "第2轮AI回复"},
    {"role": "user", "content": "第3轮用户消息"},
    {"role": "assistant", "content": "第3轮AI回复"},
    {"role": "user", "content": "第4轮用户消息"},
    {"role": "assistant", "content": "第4轮AI回复"},
    {"role": "user", "content": "第5轮用户消息"},
    {"role": "assistant", "content": "第5轮AI回复"},
]

# 只保留最近 2 轮
compressed_history = manage_conversation_history(long_history, max_turns=2)

print("原始历史长度:", len(long_history))
print("压缩后长度:", len(compressed_history))
print("\n压缩后的内容:")
for msg in compressed_history:
    print(f"  [{msg['role']}] {msg['content']}")

原始历史长度: 11
压缩后长度: 5

压缩后的内容:
  [system] 你是 AI 助手
  [user] 第4轮用户消息
  [assistant] 第4轮AI回复
  [user] 第5轮用户消息
  [assistant] 第5轮AI回复

Function Calling：让 AI 调用工具

大模型虽然强大，但本质上只能生成文本，无法直接查询实时数据、执行计算、调用外部 API。例如，如果你问"北京现在的天气"，模型只能根据训练数据猜测，无法获取真实的天气信息。

Function Calling（函数调用）功能解决了这个问题。它让 AI 能够：

• 识别用户意图需要调用哪个工具
• 从用户输入中提取参数
• 返回一个"调用请求"（而不是直接调用）
• 由你的代码执行实际的函数调用
• 将结果返回给 AI，让它生成最终回复

这个过程是人机协作：AI 负责理解意图和提取参数，你的代码负责执行实际操作。通过这种方式，AI 可以查天气、搜索资料、操作数据库、调用任意 API。

1. Function Calling 的核心流程

Function Calling 的完整流程包括以下步骤：

1. 定义工具（tools）：告诉 AI 你有哪些函数可以调用，每个函数的参数是什么
2. 第一次调用 API：AI 分析用户输入，决定是否需要调用函数
3. 检查响应：如果 AI 返回了 tool_calls，说明它想调用函数
4. 执行函数：根据 AI 的请求，执行实际的函数调用
5. 第二次调用 API：将函数执行结果返回给 AI
6. AI 生成最终回复：结合函数结果，生成用户可读的回答

2. 完整示例：天气查询工具

首先，我们定义一个获取天气的函数（这里用 Mock 数据模拟真实 API）：

import json

def get_weather(city: str, unit: str = "celsius") -> str:
    """
    获取指定城市的天气信息（Mock 函数，实际应调用天气 API）
    
    Args:
        city: 城市名称
        unit: 温度单位（celsius 或 fahrenheit）
    
    Returns:
        天气信息的 JSON 字符串
    """
    # 模拟天气数据
    weather_data = {
        "北京": {"temperature": 15, "condition": "晴天", "humidity": 45},
        "上海": {"temperature": 20, "condition": "多云", "humidity": 60},
        "深圳": {"temperature": 28, "condition": "小雨", "humidity": 75},
    }
    
    # 检查城市是否存在于模拟数据中
    if city in weather_data:
        data = weather_data[city]
        # 如果单位为华氏度，则进行温度单位转换
        if unit == "fahrenheit":
            data["temperature"] = int(data["temperature"] * 9/5 + 32)
        
        # 返回包含详细天气信息的 JSON 字符串
        return json.dumps({
            "city": city,
            "temperature": data["temperature"],
            "unit": unit,
            "condition": data["condition"],
            "humidity": data["humidity"]
        }, ensure_ascii=False)
    else:
        # 若城市未在数据中定义，返回错误信息
        return json.dumps({"error": f"未找到 {city} 的天气数据"}, ensure_ascii=False)

# 测试函数
print("测试天气查询函数：")
print(get_weather("北京"))
print(get_weather("上海", "fahrenheit"))

测试天气查询函数：
{"city": "北京", "temperature": 15, "unit": "celsius", "condition": "晴天", "humidity": 45}
{"city": "上海", "temperature": 68, "unit": "fahrenheit", "condition": "多云", "humidity": 60}

接下来，我们需要定义工具的 schema（描述），告诉 AI 这个函数的作用、参数类型等信息：

# 定义工具 schema
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的实时天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如：北京、上海、深圳"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，celsius（摄氏度）或 fahrenheit（华氏度）"
                    }
                },
                "required": ["city"]  # city 是必填参数，unit 是可选参数
            }
        }
    }
]

print("✅ 工具 schema 已定义")

✅ 工具 schema 已定义，这个 schema 使用 JSON Schema 格式，包含：

• name：函数名称（必须与实际函数名一致）
• description：函数的作用描述（AI 根据这个描述判断是否调用）
• parameters：参数定义（类型、描述、是否必填）

现在，让我们完整实现 Function Calling 流程：

# 创建客户端
client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

# 用户提问
user_query = "北京现在的天气怎么样？"

# 初始化消息
messages = [
    {"role": "system", "content": "你是一个友好的天气助手，可以查询天气信息。"},
    {"role": "user", "content": user_query}
]

print(f"用户: {user_query}\n")

# 第一次调用：让 AI 决定是否需要调用工具
response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    tools=tools,  # 传递工具定义
    tool_choice="auto"  # auto: AI 自动决定是否调用；也可以设为 "none" 或强制调用某个工具
)

# 检查 AI 是否想调用函数
if response.choices[0].message.tool_calls:
    print("AI 决定调用工具：")
    
    # 提取工具调用信息
    tool_call = response.choices[0].message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    
    print(f"  函数名: {function_name}")
    print(f"  参数: {function_args}\n")
    
    # 执行实际的函数调用
    if function_name == "get_weather":
        function_result = get_weather(**function_args)
        print(f"函数执行结果: {function_result}\n")
        
        # 将函数结果添加到消息历史
        messages.append(response.choices[0].message)  # AI 的工具调用请求
        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": function_result
        })
        
        # 第二次调用：让 AI 根据函数结果生成最终回复
        final_response = client.chat.completions.create(
            model="deepseek-chat",
            messages=messages
        )
        
        final_answer = final_response.choices[0].message.content
        print(f"AI 最终回复: {final_answer}")
    
else:
    # AI 认为不需要调用工具，直接回复
    print(f"AI 直接回复: {response.choices[0].message.content}")

用户: 北京现在的天气怎么样？

AI 决定调用工具：
  函数名: get_weather
  参数: {'city': '北京', 'unit': 'celsius'}

函数执行结果: {"city": "北京", "temperature": 15, "unit": "celsius", "condition": "晴天", "humidity": 45}

AI 最终回复: 北京现在是晴天，气温15°C，湿度45%。天气不错，适合外出活动！

这段代码展示了完整的 Function Calling 流程：

1. 第一次 API 调用：传递 tools 参数，AI 分析用户意图
2. **检查 tool_calls **：如果存在，说明 AI 想调用函数
3. 提取参数：从 tool_call.function.arguments 中提取 JSON 格式的参数
4. 执行函数：调用实际的 Python 函数
5. 第二次 API 调用：将函数结果以 role="tool" 的消息返回给 AI
6. AI 生成回复：结合天气数据，生成自然语言回答

运行后，AI 会回复类似"北京现在的天气是晴天，温度 15℃，湿度 45%"这样的完整回答。

3. Function Calling 的实际应用

• 客服机器人：查询订单状态、修改地址、申请退款
• 数据分析助手：查询数据库、生成报表、发送邮件
• 开发助手：搜索文档、执行代码、部署应用

注意：并非所有模型都支持 Function Calling。虽然目前为止大部分的大模型都支持 Function Calling，但使用前还是需要查阅平台文档确认支持情况。

多模态输入：让 AI "看图说话"

多模态模型（Multimodal Model）可以同时处理文本和图像输入。目前支持视觉理解的主流模型包括：

• gpt-5 / GPT-5 nano：OpenAI 的多模态模型，图像理解能力强
• Claude 4.5 Sonnet/Opus：Anthropic 的多模态模型，代码图像分析出色
• Gemini 3.0 Pro：Google 的多模态模型，支持超长上下文
• Qwen-VL-Plus：阿里通义千问视觉版，国内可直接使用

1. 图片传递方式一：URL 链接

# 使用 OpenRouter 调用 gpt-5（支持视觉理解）
client = OpenAI(
    api_key=os.getenv("OPENROUTER_API_KEY"),
    base_url="https://openrouter.ai/api/v1"
)

# 一张公开的图片 URL（示例）
image_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"

# 构造包含图片的消息
response = client.chat.completions.create(
    model="openai/gpt-4o",  # 使用支持视觉的模型
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "这张图片里有什么？请详细描述。"},
                {
                    "type": "image_url",
                    "image_url": {"url": image_url}
                }
            ]
        }
    ],
    max_tokens=500
)

print("AI 对图片的描述：")
print(response.choices[0].message.content)

AI 对图片的描述：
这张图片展示了一条木制小路，它延伸通过一片广阔的草地。小路两侧是绿色的植物和灌木丛。在远处，可以看到一些低矮的树木。天空中蓝天白云交织，阳光明媚，营造出一种宁静自然的氛围。整幅图像给人一种开阔和平静的感觉。

关键点：

1. content 变成列表：不再是单纯的字符串，而是包含多个元素的列表
2. text 元素：{"type": "text", "text": "..."}，表示文本输入
3. image_url 元素：{"type": "image_url", "image_url": {"url": "..."}}，表示图片 URL

AI 会分析图片内容，然后用自然语言描述它看到的内容。这种方式适合图片已经托管在云存储、CDN 或公开网站上的场景。

2. 图片传递方式二：Base 64 编码

如果图片在本地，或者不方便通过 URL 访问，可以将图片编码为 Base 64 字符串后传递：

pip install Pillow

# 方法2：使用base64编码本地图片
print("=" * 60)
print("方法2：通过base64编码传递本地图片")
print("=" * 60)

from PIL import Image
import io
import base64

def compress_image(image_path, max_size=(800, 800)):
    """压缩图片到合适大小"""
    with Image.open(image_path) as img:
        # 保持宽高比缩放
        img.thumbnail(max_size)
        
        # 保存为JPEG并压缩
        buffer = io.BytesIO()
        img.save(buffer, format='JPEG', quality=85)
        
        # 编码为base64
        return base64.b64encode(buffer.getvalue()).decode('utf-8')

# 使用压缩后的图片
b64_image = compress_image("/Users/mac/大模型资料/大模型基础入门/images/zhipu_model_plaza.png")
print(f"压缩后大小: {len(b64_image)/1024:.2f} KB")  # 确保 <500KB

# ✅ 这样更有可能成功
messages=[{
    "role": "user",
    "content": [
        {"type": "text", "text": "描述图片"},
        {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64_image}"}}
    ]
}]

response = client.chat.completions.create(
    model="openai/gpt-4o",
    messages=messages,
    max_tokens=100
)

print(f"AI回复：{response.choices[0].message.content}")

============================================================
方法2：通过base64编码传递本地图片
============================================================
压缩后大小: 36.54 KB
AI回复：这是一张关于“BigModel”网站的截图，界面上显示了多种模型的分类。左侧的菜单中有“语音模型”、“多模态模型”、“音视频模型”、“其他模型”等选项。页面的右侧主要展示了几款模型的名称和简短描述，比如“GLM-4.7”、“GLM-4.7-FlashX”等。这些模型被标注为“新发布”或是“精确

Base 64 编码的要点：

1. 格式：必须以 data:image/jpeg;base64, 开头，然后跟 Base 64 字符串
2. 压缩：大图片会导致 Token 消耗激增，建议压缩到 1024 x 1024 以内
3. 适用场景：本地图片、用户上传的图片、临时图片

成本提示：图片输入会消耗大量 Token。以 gpt-5.2 为例，一张 1024 x 1024 的图片约消耗 765 tokens。因此，使用多模态功能时要特别注意成本控制。

提示词工程：写出高质量 Prompt

提示词工程（Prompt Engineering）是一门让 AI 更准确理解你意图的艺术

技巧 1：明确角色与行为规范

它的核心逻辑是：通过明确告诉模型"你是谁"以及"你应该怎么做"，来约束和引导模型的行为。

可以在 System Prompt 中定义模型的身份（比如"你是一位资深的 Python 技术导师"）、输出风格（比如"解释简洁易懂，避免术语堆砌"）、以及具体的行为规范（比如"必须提供可运行的代码示例"）。这些规则会在整个对话过程中持续生效，成为模型回答的"行为准则"。

好的角色定义通常包含三个要素：身份定位、专业领域、以及输出约束。这三者缺一不可，共同构成了一个清晰的"人设"。

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv(override=True)

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

# ❌ 糟糕的提示词
bad_prompt = {"role": "user", "content": "解释一下装饰器"}

# ✅ 优秀的提示词（明确角色）
good_system = """你是一个专业的Python技术导师。
特点：
- 解释简洁易懂，避免术语堆砌
- 提供可运行的代码示例
- 指出常见错误和注意事项
- 语气友好，鼓励学习"""

good_prompt = {"role": "user", "content": "请解释Python装饰器的原理"}

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": good_system},
        good_prompt
    ]
)

print(response.choices[0].message.content)

技巧 2：要求格式化输出

格式化输出是让大模型返回结构化数据的核心技巧。它的核心逻辑是：通过在 System Prompt 中明确指定输出格式（如 JSON、Markdown 表格等），让模型的回答变得可预测、可解析。

实现格式化输出的关键在于两点：第一是提供清晰的格式模板，让模型知道期望的结构；第二是给出一个具体的输出示例，这比纯文字描述更能让模型"理解"你的意图。

常见的格式包括 JSON（最适合程序解析）、Markdown（适合文档生成）、以及自定义分隔符格式（适合简单场景）。选择哪种格式取决于你的下游需求：如果要接入自动化流程，优先选 JSON；如果是给人看的报告，Markdown 更合适。

# 要求JSON格式输出
system_prompt = """请以JSON格式返回结果，严格遵循以下格式：
{
    "summary": "核心要点（一句话）",
    "steps": ["步骤1", "步骤2", "步骤3"],
    "code_example": "代码示例",
    "common_mistakes": ["常见错误1", "常见错误2"]
}"""

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": "如何使用Python读取CSV文件？"}
    ],
    temperature=0  # 确定性输出
)

print(response.choices[0].message.content)

技巧 3：少样本学习（Few-Shot Learning）

Few-Shot Learning（少样本学习）是一种通过在提示中嵌入少量示例，让大模型"临时学会"特定任务的技术。它的核心逻辑可以用"先示范、再提问"来概括。

示例的数量通常在 2-5 个之间，太少可能让模型"学不会"，太多则会消耗过多 Token 并增加成本。选择具有代表性、边界清晰的示例，是 Few-Shot 成功的关键。

Few-Shot 的核心逻辑

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv(override=True)

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

# ============ Few-Shot Learning 示例 ============
# 任务：情感分类（正面/负面/中性）

messages = [
    # 系统角色定义任务
    {"role": "system", "content": "你是一个情感分析助手。请根据用户输入的文本，判断情感倾向，只输出：正面、负面 或 中性。"},
    
    # ========== Few-Shot 示例开始 ==========
    # 示例 1：正面
    {"role": "user", "content": "这家餐厅的服务太棒了，菜品也很美味！"},
    {"role": "assistant", "content": "正面"},
    
    # 示例 2：负面
    {"role": "user", "content": "等了一个小时外卖还没到，客服态度也很差。"},
    {"role": "assistant", "content": "负面"},
    
    # 示例 3：中性
    {"role": "user", "content": "今天天气一般，不冷也不热。"},
    {"role": "assistant", "content": "中性"},
    # ========== Few-Shot 示例结束 ==========
    
    # 真正需要模型处理的新问题
    {"role": "user", "content": "这个产品质量不错，但是价格有点贵。"}
]

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=messages,
    temperature=0  # 降低随机性，让分类更稳定
)

print("情感分析结果：", response.choices[0].message.content)

技巧 4：链式思考（Chain of Thought）

链式思考（Chain of Thought，简称 CoT）是一种让大模型"逐步推理"的提示技术。它的核心逻辑是：通过要求模型先展示思考过程，再给出最终答案，从而提升复杂问题的回答准确率。

实现链式思考有两种方式：第一种是在提示词中直接加上"请一步一步思考"或 "Let's think step by step" 这样的指令；第二种是通过 Few-Shot 示例，给模型展示几个带有详细推理过程的样例，让它学会这种输出风格。

链式思考特别适合解决需要多步推理的问题，比如数学计算、逻辑推断、代码调试等。但对于简单问答类任务，使用 CoT 可能会增加不必要的 Token 消耗，因此需要根据具体场景权衡使用。

# 让AI展示推理过程
prompt = """请一步步分析以下问题：
问题：一个班级有30名学生，其中60%是女生。如果再加入5名男生，女生占比是多少？

请按以下格式作答：
1. 理解题意：...
2. 计算原始数据：...
3. 计算新数据：...
4. 得出结论：...
"""

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": prompt}],
    temperature=0
)

print(response.choices[0].message.content)

• 最佳实践总结：
  • system 消息定义角色和行为规范
  • 明确要求输出格式（JSON/Markdown/表格）
  • 使用 Few-Shot 示例帮助 AI 理解模式
  • 对于复杂任务，要求展示推理过程

链式思考适用场景

进阶技巧：可以在 Few-Shot 示例中同时展示"问题 + 逐步推理 + 答案"的完整流程，让 AI 学会这种思维模式。

异步批处理：并发提升效率

如果需要同时处理多个问题（如批量翻译、批量摘要），同步逐个调用会非常慢。使用 AsyncOpenAI 客户端配合 asyncio，可以并发执行多个请求，大幅提升吞吐量。

import os
import time
import asyncio
from openai import OpenAI, AsyncOpenAI
from dotenv import load_dotenv

load_dotenv(override=True)

# 同步客户端
sync_client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

# 异步客户端
async_client = AsyncOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com"
)

# 测试问题列表
questions = [
    "什么是Python？",
    "什么是JavaScript？",
    "什么是Go语言？",
    "什么是Rust？",
    "什么是TypeScript？"
]

# 方法1：同步调用（逐个执行）
def sync_batch():
    print("=" * 60)
    print("同步调用（逐个执行）")
    print("=" * 60)
    
    start_time = time.time()
    results = []
    
    for question in questions:
        response = sync_client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": question}],
            max_tokens=50
        )
        results.append(response.choices[0].message.content)
    
    elapsed = time.time() - start_time
    print(f"完成 {len(questions)} 个请求")
    print(f"耗时：{elapsed:.2f} 秒\n")
    
    return results, elapsed

# 方法2：异步并发调用
async def ask_question_async(question):
    """异步调用单个问题"""
    response = await async_client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": question}],
        max_tokens=50
    )
    return response.choices[0].message.content

async def async_batch():
    print("=" * 60)
    print("异步并发调用")
    print("=" * 60)
    
    start_time = time.time()
    
    # 使用 asyncio.gather 并发执行所有请求
    results = await asyncio.gather(*[ask_question_async(q) for q in questions])
    
    elapsed = time.time() - start_time
    print(f"完成 {len(questions)} 个请求")
    print(f"耗时：{elapsed:.2f} 秒\n")
    
    return results, elapsed

# 性能对比
print("开始性能测试...\n")

# 同步测试
sync_results, sync_time = sync_batch()

# 异步测试,Jupyter 专用
async_results, async_time = await async_batch()

# 普通python环境使用
# asyncio.run(async_batch())

# 性能提升
improvement = (sync_time - async_time) / sync_time * 100

print("=" * 60)
print("性能对比")
print("=" * 60)
print(f"同步调用耗时：{sync_time:.2f} 秒")
print(f"异步调用耗时：{async_time:.2f} 秒")
print(f"性能提升：{improvement:.1f}%")
print(f"\n异步调用使耗时减少了 {sync_time - async_time:.2f} 秒！")

适用场景：

• 批量翻译、批量摘要、批量分类

  • 多文档并发处理
  • Web 应用中的高并发请求

• 注意事项：

  • 仍需遵守速率限制（Rate Limit），不要无限并发
  • 建议配合信号量（asyncio.Semaphore）控制并发数

同步 vs 异步性能对比

通过掌握异步批处理，你的 AI 应用将能够高效处理大规模任务，从"一个一个慢慢来"进化到"批量并发快速完成"。

最佳实践：对于 100 个以上的大批量任务，建议分批执行（如每批 20 个），避免内存占用过高和网络不稳定的影响。

重要提示：虽然我们在本节演示了 Chat Completions API，但 OpenAI 在 2025年还推出了新的 Responses API，提供更多专属功能。不过对于跨平台开发，优先掌握通用的 Chat Completions API，这样你的代码才能在所有模型间自由迁移。

Responses API

Responses API 是 OpenAI 2025 年推出的新一代 agentic API，

• 主要差异点：
  1. 服务端状态管理 - 通过 previous_response_id 自动维护对话历史
  2. 内置工具支持 - Web 搜索、文件搜索、计算机操作等
  3. 事件驱动架构 - 更可预测的流式响应
  4. 简化的 agentic 工作流 - 专为 AI Agent 设计

from openai import OpenAI
from dotenv import load_dotenv
load_dotenv(override=True)

# 初始化客户端
client = OpenAI()  # 确保设置了 OPENAI_API_KEY 环境变量

# ============================================================
# 示例1: Chat Completions API (传统方式)
# 需要手动维护和传递完整的对话历史
# ============================================================
print("=" * 60)
print("【Chat Completions API - 传统方式】")
print("=" * 60)

# 第一轮对话
messages = [{"role": "user", "content": "我叫小明，请记住我的名字"}]
response1 = client.chat.completions.create(
    model="gpt-5-nano",
    messages=messages
)
print(f"用户: {messages[0]['content']}")
print(f"助手: {response1.choices[0].message.content}\n")

# 第二轮对话 - 必须手动维护历史
messages.append({"role": "assistant", "content": response1.choices[0].message.content})
messages.append({"role": "user", "content": "你还记得我叫什么名字吗？"})

response2 = client.chat.completions.create(
    model="gpt-5-nano",
    messages=messages  # 必须传入完整历史
)
print(f"用户: {messages[-1]['content']}")
print(f"助手: {response2.choices[0].message.content}")
print(f"\n⚠️  需手动管理的消息数: {len(messages)}")

Responses API vs Chat Completions API 核心差异

1. Chat Completions - 需手动维护 messages 数组
2. Responses API - 使用 store=True + previous_response_id 自动关联上下文
3. 内置工具 - 直接使用 tools=[{"type": "web_search_preview"}] 进行网络搜索