Python 调用 OpenAI API 深度实战——Embeddings、Fine-tuning、Batch API
大约 11 分钟
Python 调用 OpenAI API 深度实战——Embeddings、Fine-tuning、Batch API
适读人群：已入门 OpenAI API、想深入使用高级功能的工程师 | 阅读时长：约18分钟 | 核心价值：掌握 Embeddings 语义搜索、Fine-tuning 定制模型、Batch API 大批量处理三大进阶技能
老刘在一家电商公司做技术，去年接了一个项目：给客服系统加 AI 能力。最开始很顺利，调几个接口，客服机器人就能回答用户的常见问题了。领导很开心，让他继续深挖，做得更智能一点。
问题就从这里开始了。
他想做商品的语义搜索——用户输入"防晒霜"，能搜到"防晒乳液"、"SPF50防护霜"这类语义相关的商品。用关键词搜索肯定不行，他开始研究 Embeddings。结果发现把几十万个商品描述向量化，每天还要更新，光 API 费用就不是小数字。
然后他想做 Fine-tuning，让模型学会公司特有的产品术语和回复风格，结果数据格式弄了半天，训练完发现效果还不如基础模型。
最后他发现每月要对十几万条用户反馈做情感分析，一条条调 API 太慢，也很贵。
他找到我时说："老张，感觉 OpenAI API 越用越复杂，有很多功能我根本不知道该怎么用对。"
我说：你踩的这三个坑，正好是 OpenAI API 最值得学的三个深水区。今天我们逐一拆解。
一、Embeddings——语义向量化完整实践
1.1 基础 Embeddings 调用
import os
from openai import OpenAI
import numpy as np
from typing import List
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
def get_embedding(text: str, model: str = "text-embedding-3-small") -> List[float]:
"""
获取文本的向量表示
text-embedding-3-small：1536维，$0.02/1M tokens
text-embedding-3-large：3072维，$0.13/1M tokens（效果更好但贵）
"""
# 清理文本（去掉多余空白，减少 token 消耗）
text = text.replace("\n", " ").strip()
response = client.embeddings.create(
input=text,
model=model
)
return response.data[0].embedding
def get_embeddings_batch(texts: List[str], model: str = "text-embedding-3-small") -> List[List[float]]:
"""
批量获取向量，比逐条调用效率高很多
单次请求最多 2048 条文本（注意总 token 数限制在 8191）
"""
# 清理文本
texts = [t.replace("\n", " ").strip() for t in texts]
# 如果超过批次限制，分批处理
batch_size = 100
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
response = client.embeddings.create(
input=batch,
model=model
)
# 按 index 排序确保顺序正确
sorted_data = sorted(response.data, key=lambda x: x.index)
all_embeddings.extend([item.embedding for item in sorted_data])
print(f"已处理 {min(i+batch_size, len(texts))}/{len(texts)} 条")
return all_embeddings
def cosine_similarity(vec1: List[float], vec2: List[float]) -> float:
"""计算余弦相似度"""
v1 = np.array(vec1)
v2 = np.array(vec2)
return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))1.2 商品语义搜索实现
import json
from pathlib import Path
class ProductSemanticSearch:
"""
基于 OpenAI Embeddings 的商品语义搜索
"""
def __init__(self, cache_file: str = "./product_embeddings.json"):
self.cache_file = cache_file
self.products = []
self.embeddings = []
def build_index(self, products: List[dict]):
"""
构建商品向量索引
products: [{"id": "001", "name": "防晒霜", "description": "SPF50防护..."}]
"""
# 检查缓存
if Path(self.cache_file).exists():
print("从缓存加载向量索引...")
with open(self.cache_file, "r") as f:
cache = json.load(f)
self.products = cache["products"]
self.embeddings = cache["embeddings"]
return
print(f"开始向量化 {len(products)} 个商品...")
self.products = products
# 构建搜索文本（结合商品名和描述）
search_texts = [
f"{p['name']} {p.get('description', '')} {p.get('category', '')}"
for p in products
]
self.embeddings = get_embeddings_batch(search_texts)
# 保存缓存
with open(self.cache_file, "w") as f:
json.dump({
"products": self.products,
"embeddings": self.embeddings
}, f, ensure_ascii=False)
print("索引构建完成并已缓存")
def search(self, query: str, top_k: int = 5) -> List[dict]:
"""
语义搜索
返回最相关的 top_k 个商品
"""
query_embedding = get_embedding(query)
# 计算所有商品的相似度
similarities = [
cosine_similarity(query_embedding, emb)
for emb in self.embeddings
]
# 排序取 top_k
top_indices = np.argsort(similarities)[::-1][:top_k]
results = []
for idx in top_indices:
product = self.products[idx].copy()
product["similarity_score"] = round(float(similarities[idx]), 4)
results.append(product)
return results
# 测试
searcher = ProductSemanticSearch()
# 模拟商品数据
products = [
{"id": "001", "name": "日系防晒霜", "description": "SPF50+ PA++++，清爽不油腻", "category": "护肤"},
{"id": "002", "name": "保湿面霜", "description": "深层补水，适合干性肌肤", "category": "护肤"},
{"id": "003", "name": "防晒喷雾", "description": "便携防晒，户外补涂方便", "category": "护肤"},
{"id": "004", "name": "美白精华液", "description": "烟酰胺成分，淡斑提亮", "category": "护肤"},
{"id": "005", "name": "防晒口罩", "description": "UPF50冰丝面料，遮挡紫外线", "category": "防护"},
]
searcher.build_index(products)
# 语义搜索测试
results = searcher.search("夏天出门防紫外线", top_k=3)
print("搜索'夏天出门防紫外线'的结果：")
for r in results:
print(f" {r['name']}（相似度：{r['similarity_score']}）- {r['description']}")
# 实测：会正确找到防晒霜、防晒喷雾、防晒口罩，而不是按关键词匹配二、Fine-tuning——定制你自己的模型
2.1 什么时候该用 Fine-tuning
我的经验：先别用 Fine-tuning，大多数时候更好的 Prompt 就能解决问题，而且便宜得多。
真正适合 Fine-tuning 的场景：
- 需要输出特定格式（比如只输出 JSON，不要解释文字）
- 需要学习特有的语气/风格（比如模仿公司品牌口吻）
- 需要记住大量专有术语（比如内部产品名、行业术语）
- 需要大幅减少 Prompt 长度（把 few-shot 训练成模型能力）
2.2 准备训练数据
import json
def prepare_fine_tuning_data(examples: List[dict], output_file: str):
"""
准备 Fine-tuning 所需的 JSONL 格式数据
examples: [{"input": "用户消息", "output": "期望的回复", "system": "可选的系统提示"}]
"""
with open(output_file, "w", encoding="utf-8") as f:
for example in examples:
messages = []
# 系统提示（可选）
if "system" in example:
messages.append({
"role": "system",
"content": example["system"]
})
# 用户输入
messages.append({
"role": "user",
"content": example["input"]
})
# 期望输出
messages.append({
"role": "assistant",
"content": example["output"]
})
line = json.dumps({"messages": messages}, ensure_ascii=False)
f.write(line + "\n")
print(f"已生成 {len(examples)} 条训练数据：{output_file}")
# 示例：训练客服回复风格
training_examples = [
{
"system": "你是XX公司的客服助手，回复要简洁、友好、专业",
"input": "我的订单什么时候发货？",
"output": "您好！订单一般在付款后48小时内发货，发货后会短信通知您快递单号。如有特殊情况导致延迟，我们会提前告知。请问您的订单号是多少，我来帮您确认一下具体情况？"
},
{
"system": "你是XX公司的客服助手，回复要简洁、友好、专业",
"input": "产品坏了能退吗？",
"output": "当然可以！7天内支持无理由退换，质量问题15天内可退。请您先拍几张照片，通过订单页面申请售后，我们会在24小时内处理。"
},
# 建议准备 50-100 条高质量样本
]
prepare_fine_tuning_data(training_examples, "training_data.jsonl")2.3 提交 Fine-tuning 任务
def start_fine_tuning(training_file_path: str, model: str = "gpt-3.5-turbo") -> str:
"""
上传数据并启动 Fine-tuning
返回 job_id
"""
# 1. 上传训练文件
print("上传训练数据...")
with open(training_file_path, "rb") as f:
file_response = client.files.create(
file=f,
purpose="fine-tune"
)
file_id = file_response.id
print(f"文件上传成功，ID：{file_id}")
# 2. 创建 Fine-tuning 任务
print("创建 Fine-tuning 任务...")
job = client.fine_tuning.jobs.create(
training_file=file_id,
model=model,
hyperparameters={
"n_epochs": 3, # 训练轮次，通常3-5轮
"batch_size": "auto",
"learning_rate_multiplier": "auto"
},
suffix="customer-service-v1" # 模型名称后缀，方便识别
)
print(f"Fine-tuning 任务已创建，Job ID：{job.id}")
print(f"预计费用：$0.008/1K tokens（gpt-3.5-turbo）")
return job.id
def check_fine_tuning_status(job_id: str):
"""检查 Fine-tuning 进度"""
job = client.fine_tuning.jobs.retrieve(job_id)
print(f"状态：{job.status}")
print(f"训练 Token 数：{job.trained_tokens}")
if job.status == "succeeded":
print(f"训练完成！模型 ID：{job.fine_tuned_model}")
return job.fine_tuned_model
elif job.status == "failed":
print(f"训练失败：{job.error}")
return None
# 使用训练好的模型
def use_fine_tuned_model(model_id: str, user_message: str) -> str:
response = client.chat.completions.create(
model=model_id, # 使用 Fine-tuned 模型 ID
messages=[
{"role": "system", "content": "你是XX公司的客服助手"},
{"role": "user", "content": user_message}
]
)
return response.choices[0].message.content2.4 踩坑实录一：Fine-tuning 效果不如预期
现象：训练完后，模型在训练集上效果很好，但实际用户问的问题效果一般，有时还不如基础模型。
原因：训练数据质量不高，或数量太少（不到50条）；同时训练数据和实际问题分布差异太大。
解法：
# 训练数据质量检查
def validate_training_data(file_path: str):
"""检查训练数据质量"""
issues = []
examples = []
with open(file_path, "r") as f:
for i, line in enumerate(f):
try:
data = json.loads(line.strip())
examples.append(data)
except json.JSONDecodeError:
issues.append(f"第{i+1}行：JSON 格式错误")
# 统计
total = len(examples)
avg_input_len = np.mean([len(e["messages"][1]["content"]) for e in examples])
avg_output_len = np.mean([len(e["messages"][-1]["content"]) for e in examples])
print(f"总样本数：{total}（建议至少50条）")
print(f"平均输入长度：{avg_input_len:.0f} 字符")
print(f"平均输出长度：{avg_output_len:.0f} 字符")
if total < 50:
issues.append(f"训练数据不足（{total}条），建议50条以上")
if avg_output_len < 20:
issues.append("输出过短，模型可能学不到足够的风格特征")
if issues:
print("\n发现问题：")
for issue in issues:
print(f" - {issue}")
else:
print("\n数据格式检查通过")
return len(issues) == 0三、Batch API——大批量处理省50%成本
OpenAI 的 Batch API 是一个很被低估的功能：把任务打包提交，在24小时内处理完，价格只有标准 API 的一半。适合不需要实时响应的批量任务。
3.1 批量情感分析实践
import json
import time
from pathlib import Path
def create_batch_sentiment_analysis(texts: List[str], output_file: str) -> str:
"""
创建批量情感分析任务
返回 batch_id
"""
# 1. 构建批量请求文件（JSONL 格式）
requests_file = output_file + ".requests.jsonl"
with open(requests_file, "w", encoding="utf-8") as f:
for i, text in enumerate(texts):
request = {
"custom_id": f"request-{i:06d}", # 自定义 ID，用于后续匹配结果
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-3.5-turbo",
"messages": [
{
"role": "system",
"content": "判断用户评论的情感，只输出 JSON：{\"sentiment\": \"positive/negative/neutral\", \"score\": 0-10}"
},
{"role": "user", "content": text}
],
"max_tokens": 50,
"temperature": 0
}
}
f.write(json.dumps(request, ensure_ascii=False) + "\n")
# 2. 上传批量请求文件
with open(requests_file, "rb") as f:
batch_file = client.files.create(file=f, purpose="batch")
# 3. 创建 Batch 任务
batch = client.batches.create(
input_file_id=batch_file.id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
print(f"Batch 任务已创建：{batch.id}")
print(f"预计完成时间：24小时内")
print(f"成本（相比标准API节省50%）：约 ${len(texts) * 0.001:.3f}")
return batch.id
def wait_and_get_batch_results(batch_id: str, output_file: str) -> List[dict]:
"""
等待 Batch 完成并获取结果
"""
print(f"等待 Batch {batch_id} 完成...")
while True:
batch = client.batches.retrieve(batch_id)
status = batch.status
print(f"状态：{status} | 完成：{batch.request_counts.completed}/{batch.request_counts.total}")
if status == "completed":
break
elif status in ["failed", "expired", "cancelled"]:
raise Exception(f"Batch 失败：{status}")
time.sleep(30) # 每30秒检查一次
# 下载结果
result_content = client.files.content(batch.output_file_id)
results = {}
for line in result_content.text.strip().split("\n"):
data = json.loads(line)
custom_id = data["custom_id"]
if data.get("error"):
results[custom_id] = {"error": data["error"]}
else:
content = data["response"]["body"]["choices"][0]["message"]["content"]
try:
results[custom_id] = json.loads(content)
except json.JSONDecodeError:
results[custom_id] = {"raw": content}
# 按原始顺序返回
sorted_results = [results.get(f"request-{i:06d}", {}) for i in range(len(results))]
# 保存结果
with open(output_file, "w", encoding="utf-8") as f:
json.dump(sorted_results, f, ensure_ascii=False, indent=2)
return sorted_results
# 使用示例
user_reviews = [
"产品质量很好，快递也很快，满意",
"收到货发现有破损，客服也不理人，差评",
"还行吧，没有特别好也没有特别差",
# ... 可以有几万条
]
batch_id = create_batch_sentiment_analysis(user_reviews, "sentiment_results.json")
# 等待处理（最多24小时）
results = wait_and_get_batch_results(batch_id, "sentiment_results.json")
for review, result in zip(user_reviews[:3], results[:3]):
print(f"评论：{review}")
print(f"情感：{result}")
print()3.2 踩坑实录二：Batch 任务失败但没有错误提示
现象：Batch 任务显示 completed，但结果文件里有大量 {"error": ...} 条目。
原因：Batch API 是"尽力处理"模式，单条请求失败不会导致整个 Batch 失败，失败的条目会在结果里标记 error。
解法：
def analyze_batch_errors(results: List[dict]) -> dict:
"""分析 Batch 结果中的错误情况"""
success_count = 0
error_count = 0
error_types = {}
for result in results:
if "error" in result:
error_count += 1
error_type = result["error"].get("code", "unknown")
error_types[error_type] = error_types.get(error_type, 0) + 1
else:
success_count += 1
print(f"成功：{success_count}，失败：{error_count}")
if error_types:
print("错误类型分布：")
for err_type, count in error_types.items():
print(f" {err_type}: {count} 次")
# 常见错误：
# content_filter：内容被安全过滤器拦截（需要审查原始文本）
# rate_limit：虽然是批量但仍有限制，需要减小批次大小
# context_length_exceeded：输入太长，需要截断
return {"success": success_count, "error": error_count, "error_types": error_types}四、踩坑实录三：Embeddings 向量维度不一致
现象：系统上线半年后，更换了 Embedding 模型（从 ada-002 换到 text-embedding-3-small），发现新模型的向量和老向量的余弦相似度计算结果完全错乱。
原因：不同模型的向量空间完全不同，维度也不同（ada-002 是 1536 维，3-small 也是 1536 维，但向量空间不同）。不同模型的向量不能直接比较。
解法：
# 在向量库的元数据中记录使用的模型版本
def store_embedding_with_metadata(text: str, document_id: str):
"""存储向量时附带模型版本信息"""
CURRENT_EMBED_MODEL = "text-embedding-3-small"
embedding = get_embedding(text, model=CURRENT_EMBED_MODEL)
return {
"id": document_id,
"embedding": embedding,
"embed_model": CURRENT_EMBED_MODEL, # 记录版本
"created_at": time.time()
}
# 切换模型时，必须重新向量化所有历史数据
def migrate_embeddings(old_model: str, new_model: str, all_texts: List[str]):
"""迁移向量库到新模型"""
print(f"开始迁移：{old_model} → {new_model}")
print(f"共 {len(all_texts)} 条数据需要重新向量化")
# 批量重新计算
new_embeddings = get_embeddings_batch(all_texts, model=new_model)
print("迁移完成，请重新构建向量数据库索引")
return new_embeddings五、成本对比总结
| 功能 | 模型 | 价格 | 适用场景 |
|---|---|---|---|
| Embeddings | text-embedding-3-small | $0.02/1M tokens | 通用语义搜索，首选 |
| Embeddings | text-embedding-3-large | $0.13/1M tokens | 精度要求高 |
| Fine-tuning 训练 | gpt-3.5-turbo | $8/1M tokens | 一次性训练投入 |
| Fine-tuned 推理 | gpt-3.5-turbo-ft | $12/1M tokens | 高于基础模型，需权衡 |
| Batch API | gpt-3.5-turbo | $0.75/1M tokens | 非实时批量任务省50% |
我的经验：Embeddings 是性价比最高的功能，语义搜索场景几乎必用；Fine-tuning 要慎用，提示词工程往往能达到80%的效果；Batch API 用于夜间离线处理任务，省钱明显。