文档
OpenAI 应用开发实战 —— 构建生产级 AI 应用
本章目标
- 掌握 Token 管理与成本控制
- 实现可靠的错误处理与重试
- 学会 Prompt Engineering 最佳实践
- 了解 Assistants API
1. Token 管理与成本优化
1.1 Token 是什么？
Token 是 LLM 处理的文本基本单位，约等于英文的 0.75 个词或中文的 0.5 个字。
import tiktoken
enc = tiktoken.get_encoding("cl100k_base")
# 中文 vs 英文 token 效率
text_en = "Hello, how are you today?"
text_zh = "你好，今天过得怎么样？"
print(f"英文 '{text_en}' → {len(enc.encode(text_en))} tokens")
print(f"中文 '{text_zh}' → {len(enc.encode(text_zh))} tokens")
# 中文 token 消耗约为英文的 2-3 倍
1.2 成本优化策略
# ① 使用 gpt-4o-mini 替代 gpt-4o（成本降低 90%+）
# gpt-4o: $2.50/$10.00 per 1M tokens
# gpt-4o-mini: $0.15/$0.60 per 1M tokens
# ② 限制输出长度
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
max_tokens=200, # 不要设为过大的值
)
# ③ 缓存常见回答（用 Embeddings 做语义缓存）
# ④ 分类任务用 gpt-4o-mini，复杂推理才用 gpt-4o
2. 错误处理与重试
import time
from openai import (
OpenAI, RateLimitError, APIError, APITimeoutError,
APIConnectionError, AuthenticationError,
)
MAX_RETRIES = 3
def call_with_retry(messages, **kwargs):
"""带重试的 API 调用"""
for attempt in range(MAX_RETRIES):
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
**kwargs,
)
return response
except RateLimitError:
wait = 2 ** (attempt + 1)
print(f"⏳ 速率限制，第 {attempt+1} 次重试，等待 {wait}s")
time.sleep(wait)
except (APIError, APITimeoutError, APIConnectionError) as e:
wait = 2 ** attempt
print(f"⚠️ API 错误: {e}，第 {attempt+1} 次重试")
time.sleep(wait)
except AuthenticationError:
print("❌ API Key 无效！请检查 OPENAI_API_KEY")
raise
raise Exception(f"重试 {MAX_RETRIES} 次后仍失败")
3. Prompt Engineering 最佳实践
3.1 结构化 Prompt 模板
def build_prompt(template: str, **kwargs) -> list[dict]:
"""构建结构化 Prompt"""
system = f"""你是 {kwargs.get('role', '一个有用的助手')}。
规则：
1. 始终用中文回复
2. 如果不知道答案，诚实说不知道
3. 保持回答简洁
输出格式要求：
{kwargs.get('output_format', '纯文本')}
"""
return [
{"role": "system", "content": system},
{"role": "user", "content": template.format(**kwargs)},
]
# 使用
messages = build_prompt(
"分析以下代码的复杂度: {code}",
role="资深软件工程师",
output_format="JSON",
code="def fib(n): return n if n<=1 else fib(n-1)+fib(n-2)",
)
3.2 Few-Shot Prompting
few_shot_messages = [
{"role": "system", "content": "将用户输入分类为：正面、负面、中性"},
{"role": "user", "content": "这个产品太棒了！"},
{"role": "assistant", "content": "分类: 正面"},
{"role": "user", "content": "一般般，没什么特别的"},
{"role": "assistant", "content": "分类: 中性"},
{"role": "user", "content": "物流慢得要命，投诉！"},
{"role": "assistant", "content": "分类: 负面"},
# 新的输入
{"role": "user", "content": "还行吧，凑合用"},
]
3.3 Chain of Thought（思维链）
cot_prompt = """请一步步求解以下数学题。
步骤1: 列出已知条件
步骤2: 列出需要求的未知数
步骤3: 选择合适的公式
步骤4: 代入计算
步骤5: 验证结果
问题：一个班级有 60% 的男生和 40% 的女生。
男生平均分 85，女生平均分 92。求全班平均分。"""
messages = [{"role": "user", "content": cot_prompt}]
4. Assistants API（托管式 Agent）
# 创建 Assistant
assistant = client.beta.assistants.create(
name="代码审查助手",
instructions="你是一个代码审查专家，帮助发现代码中的 Bug 和优化机会。",
model="gpt-4o",
tools=[{"type": "code_interpreter"}], # 内置代码执行
)
# 创建 Thread（对话线程）
thread = client.beta.threads.create()
# 添加消息
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="分析以下代码的时间复杂度：def fib(n): ...",
)
# 运行 Assistant
run = client.beta.threads.runs.create_and_poll(
thread_id=thread.id,
assistant_id=assistant.id,
)
# 获取回复
messages = client.beta.threads.messages.list(thread_id=thread.id)
for msg in messages.data:
if msg.role == "assistant":
print(msg.content[0].text.value)
5. 安全注意事项
# ❌ 不要这样做！
api_key = "sk-abc123..." # 硬编码在代码中
# 会被 git 提交到版本库！
# ✅ 正确做法
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
# ✅ 生产环境：使用密钥管理服务（AWS Secrets Manager / Vault）
思考题
- Function Calling 和传统的 if-else 路由有何本质不同？
- RAG 系统中 Embedding 模型的选择如何影响检索质量？
- 如何评估一个 Prompt 的好坏？有哪些量化指标？
- Assistants API 与自行实现 Agent 循环各有什么优劣势？