OpenAI API 实战指南
本文是作者学习OpenAI API的实践总结，从Prompt设计到API实现的完整指南。文章详细介绍了Chat Completions API、流式输出(SSE)、Function Calling等核心功能，并提供了可运行的代码示例。内容涵盖环境配置、消息角色体系、Token计算、错误处理、多轮对话管理等关键技术点，最后通过一个AI对话应用项目验证所学。作者分享了从理论到实践的完整学习路径，特别推
·
OpenAI API 实战指南
本文是我学习 OpenAI API 过程中的总结与梳理。在写完《Prompt 工程指南》后，我发现一个很实际的问题：Prompt 设计得再好，不知道怎么通过代码发给 LLM，不知道怎么处理返回结果，那终究是纸上谈兵。于是我开始啃 OpenAI 的官方文档，边学边写代码，把 Chat Completions、流式输出（SSE）、Function Calling、Token 计算、错误处理、多轮对话管理这些核心能力一个个摸了一遍。这篇文章就是我的学习笔记，每种技术都附上了我实际跑通的代码，还顺手搭了一个功能完整的 AI 对话应用来验证所学。
声明：本文为作者在学习 过程中的总结与梳理，仅供学习参考。由于作者水平有限，文中可能存在表述不准确或遗漏之处，欢迎读者提出指正与交流。配套示例项目 github: ai-chat-app/ gitee: ai-chat-app 中找到并直接运行。
目录
- 引言：从 Prompt 到 API —— 理论到实践的桥梁
- 环境准备与认证配置
- Chat Completions API 深度解析
- 流式输出：原理与实现
- 4.1 SSE 协议原理
- 4.2 Python 流式客户端实现
- 4.3 前端流式渲染
- Function Calling：让 LLM 使用工具
- 5.1 Function Calling 原理
- 5.2 工具定义与注册
- 5.3 完整调用流程
- 5.4 与 Prompt 工程的结合
- Token 计算与成本控制
- 6.1 Token 计算原理
- 6.2 成本估算与优化
- 错误处理与重试策略
- 多轮对话管理
- 结构化输出
- 生产环境最佳实践
- 配套项目：AI 对话应用
- 总结与建议
1. 引言：从 Prompt 到 API —— 理论到实践的桥梁
1.1 为什么要写这篇文章
在写完《Prompt 工程指南》之后，我掌握了五要素结构、五大推理技术、分层 Prompt 架构这些设计层面的知识。但很快我就发现一个问题：我知道怎么写一个好的 Prompt，但我不知道怎么通过代码把它发给 LLM，不知道怎么处理返回结果，遇到报错也不知道该怎么办。
于是我开始动手写代码，从最简单的 Chat Completions 调用开始，一步步摸索流式输出、Function Calling、Token 计算、错误处理……这篇文章记录的就是我这一路踩坑和收获的过程。
下面这张图是我理解的两层关系——Prompt 工程是"设计层"，OpenAI API 是"实现层"：
Prompt 工程（设计层） OpenAI API（实现层）
│ │
├─ 五要素结构 → ├─ messages 参数构建
├─ Few-shot / CoT → ├─ 多轮对话中的 Prompt 注入
├─ 结构化输出 → ├─ response_format / Function Calling
├─ 分层 Prompt 架构 → ├─ System / User / Assistant 消息分层
├─ A/B 测试 → ├─ 多模型并行调用对比
└─ 安全防护 → └─ 输入过滤 + 输出审核
1.2 我学到了什么
经过这段时间的学习和实践，我掌握了以下能力：
- 独立搭建一个支持流式输出的 AI 对话应用
- 实现 Function Calling，让 LLM 调用外部 API 和工具
- 精确计算 Token 消耗，控制 API 成本
- 处理各种 API 错误（限流、超时、内容过滤等）
- 管理多轮对话的上下文窗口
- 将 Prompt 工程的最佳实践落地到生产代码中
1.3 我的学习体会
回头看这段学习经历，有几个感受想分享：
- 流式输出是体验的分水岭，加了之后整个应用的感觉完全不一样了，从"等结果"变成了"看结果"
- Function Calling 是最让我兴奋的部分，它让 LLM 从"聊天机器人"变成了真正能"干活"的助手
- 错误处理一开始觉得繁琐，但实际跑起来才发现，网络波动、限流、超时这些情况太常见了，不处理根本没法用
- 如果你是刚开始学，建议按顺序来，每节的代码我都验证过可以直接跑；如果已经有基础，可以直接跳到流式输出和 Function Calling 这两节，这是我觉得最有意思的部分
2. 环境准备与认证配置
2.1 安装依赖
pip install openai tiktoken streamlit python-dotenv
2.2 API Key 配置
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
class APIConfig:
"""OpenAI API 配置"""
# 从环境变量读取，避免硬编码
API_KEY = os.getenv("OPENAI_API_KEY")
BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
# 模型配置
DEFAULT_MODEL = "gpt-4o-mini" # 性价比最高的默认模型
REASONING_MODEL = "gpt-4o" # 复杂推理任务
FAST_MODEL = "gpt-4o-mini" # 简单快速任务
# 请求配置
DEFAULT_TEMPERATURE = 0.7
DEFAULT_MAX_TOKENS = 2048
REQUEST_TIMEOUT = 60 # 请求超时（秒）
# 重试配置
MAX_RETRIES = 3
RETRY_BACKOFF_FACTOR = 2 # 指数退避因子
RETRY_INITIAL_DELAY = 1 # 初始延迟（秒）
# 限流配置
MAX_REQUESTS_PER_MINUTE = 500 # RPM 限制（取决于你的 Tier）
MAX_TOKENS_PER_MINUTE = 200000 # TPM 限制
@classmethod
def validate(cls):
"""验证配置是否完整"""
if not cls.API_KEY:
raise ValueError(
"未找到 OPENAI_API_KEY，请在 .env 文件中设置或设置环境变量"
)
if cls.API_KEY == "sk-your-api-key-here":
raise ValueError("请将 OPENAI_API_KEY 替换为你的真实 API Key")
2.3 .env 文件
# .env
OPENAI_API_KEY=sk-your-api-key-here
OPENAI_BASE_URL=https://api.openai.com/v1
2.4 客户端初始化
# llm_client.py - 基础客户端
from openai import OpenAI
from config import APIConfig
def create_client() -> OpenAI:
"""创建 OpenAI 客户端实例"""
APIConfig.validate()
return OpenAI(
api_key=APIConfig.API_KEY,
base_url=APIConfig.BASE_URL,
timeout=APIConfig.REQUEST_TIMEOUT,
max_retries=0, # 我们自己管理重试逻辑
)
3. Chat Completions API 深度解析
3.1 消息角色体系
OpenAI Chat API 用结构化的消息列表来组织对话，这是我刚开始接触时觉得最巧妙的设计。三种消息角色各司其职：
┌─────────────────────────────────────────────────────────┐
│ 消息角色体系 │
│ │
│ system : 系统级指令，定义 AI 的行为边界和角色 │
│ → 对应 Prompt 工程中的"角色设定"+"约束条件" │
│ → 权重最高，AI 会优先遵循 │
│ │
│ user : 用户输入，代表人类的问题或指令 │
│ → 对应 Prompt 工程中的"任务指令"+"上下文" │
│ │
│ assistant : AI 的历史回复，用于维持对话连贯性 │
│ → 在多轮对话中，需要把历史 assistant 消息传回 │
│ │
│ tool : 工具调用的结果，Function Calling 专用 │
│ → AI 调用工具后，结果以 tool 角色传回 │
└─────────────────────────────────────────────────────────┘
from openai import OpenAI
client = create_client()
# 基础调用示例
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "你是一位Python技术专家。回答要包含可运行的代码示例。"
},
{
"role": "user",
"content": "如何用Python读取CSV文件并计算某一列的平均值？"
}
],
temperature=0.7,
max_tokens=1000,
)
# 提取回复内容
answer = response.choices[0].message.content
print(answer)
# 查看 Token 使用情况
print(f"输入 Token: {response.usage.prompt_tokens}")
print(f"输出 Token: {response.usage.completion_tokens}")
print(f"总计 Token: {response.usage.total_tokens}")
3.2 核心参数详解
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
model |
str | 必填 | 模型名称，如 gpt-4o、gpt-4o-mini |
messages |
list | 必填 | 消息列表，按对话顺序排列 |
temperature |
float | 1.0 | 控制随机性，0=确定，2=最大随机 |
max_tokens |
int | 无限制 | 输出 Token 上限（不含输入） |
top_p |
float | 1.0 | 核采样，与 temperature 二选一即可 |
n |
int | 1 | 生成多少个候选回复 |
stream |
bool | False | 是否启用流式输出 |
stop |
str/list | None | 停止生成的标记 |
presence_penalty |
float | 0 | -2.0~2.0，正值鼓励谈论新话题 |
frequency_penalty |
float | 0 | -2.0~2.0，正值减少重复 |
response_format |
dict | None | {"type": "json_object"} 强制 JSON |
tools |
list | None | Function Calling 工具定义 |
tool_choice |
str/dict | “auto” | 工具选择策略 |
Temperature 选择指南
temperature 值的选择取决于任务类型：
0.0 - 0.2 → 事实性问答、代码生成、数学计算、JSON 输出
特点：输出稳定、可复现，但缺乏创造性
0.3 - 0.5 → 技术写作、翻译、摘要、数据分析
特点：在准确性和灵活性之间取得平衡
0.6 - 0.8 → 创意写作、头脑风暴、对话、角色扮演
特点：输出多样、有创意，但可能不够精确
0.9 - 1.5 → 诗歌创作、故事生成、探索性任务
特点：高度随机，输出不可预测
3.3 响应结构解析
# 完整的响应对象结构
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "你好"}],
)
# response 对象的关键属性
print(f"响应ID: {response.id}")
print(f"模型: {response.model}")
print(f"创建时间: {response.created}")
# choices 列表（n=1 时通常只有一个）
choice = response.choices[0]
print(f"结束原因: {choice.finish_reason}")
# finish_reason 可能的值：
# - "stop": 正常结束
# - "length": 达到 max_tokens 限制
# - "content_filter": 内容被过滤
# - "tool_calls": 模型请求调用工具
# - "function_call": 模型请求调用函数（旧版）
print(f"消息角色: {choice.message.role}")
print(f"回复内容: {choice.message.content}")
print(f"工具调用: {choice.message.tool_calls}")
# usage 信息
print(f"输入Token: {response.usage.prompt_tokens}")
print(f"输出Token: {response.usage.completion_tokens}")
print(f"总计Token: {response.usage.total_tokens}")
4. 流式输出：原理与实现
4.1 SSE 协议原理
流式输出是我觉得最能提升用户体验的一个特性。它基于 SSE（Server-Sent Events） 协议，原理其实不复杂：
传统请求-响应模式：
客户端 ──请求──→ 服务器 ──等待LLM生成完毕──→ 客户端
(等待 3-10 秒，用户看到空白)
流式输出模式：
客户端 ──请求──→ 服务器 ──token1──→ 客户端（显示"你好"）
──token2──→ 客户端（显示"你好，我"）
──token3──→ 客户端（显示"你好，我是"）
──token4──→ 客户端（显示"你好，我是AI"）
...
──[DONE]──→ 客户端（完成）
(用户看到文字逐字出现，体验流畅)
SSE 数据格式：
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"你好"}}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"，我"}}]}
data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"是"}}]}
data: [DONE]
4.2 Python 流式客户端实现
# llm_client.py - 流式客户端
from openai import OpenAI
from openai.types.chat import ChatCompletionChunk
from typing import Generator
class StreamingClient:
"""支持流式输出的 OpenAI 客户端"""
def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
self.client = client
self.model = model
def chat_stream(
self,
messages: list[dict],
temperature: float = 0.7,
max_tokens: int = 2048,
) -> Generator[str, None, None]:
"""
流式对话，逐 token 返回
Yields:
str: 每次返回一个增量文本片段
"""
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=True,
stream_options={"include_usage": True}, # 获取最终 usage 信息
)
for chunk in stream:
# 检查是否有内容
if chunk.choices and len(chunk.choices) > 0:
delta = chunk.choices[0].delta
if delta.content:
yield delta.content
def chat_stream_with_usage(
self,
messages: list[dict],
temperature: float = 0.7,
max_tokens: int = 2048,
) -> Generator[dict, None, None]:
"""
流式对话，返回内容和 usage 信息
Yields:
dict: {"type": "content", "data": "文本"} 或
{"type": "usage", "data": {...}}
"""
stream = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
stream=True,
stream_options={"include_usage": True},
)
for chunk in stream:
if chunk.choices and len(chunk.choices) > 0:
delta = chunk.choices[0].delta
if delta.content:
yield {"type": "content", "data": delta.content}
# 最后一个 chunk 包含 usage 信息
if hasattr(chunk, "usage") and chunk.usage:
yield {
"type": "usage",
"data": {
"prompt_tokens": chunk.usage.prompt_tokens,
"completion_tokens": chunk.usage.completion_tokens,
"total_tokens": chunk.usage.total_tokens,
}
}
# 使用示例
if __name__ == "__main__":
client = create_client()
stream_client = StreamingClient(client)
messages = [
{"role": "system", "content": "你是一位Python技术专家。"},
{"role": "user", "content": "用一句话解释什么是装饰器。"},
]
print("AI 回复: ", end="", flush=True)
for chunk in stream_client.chat_stream(messages):
print(chunk, end="", flush=True)
print()
4.3 前端流式渲染
前端接收流式输出，我了解到通常用 EventSource 或 Fetch API + ReadableStream：
// 前端流式接收示例（JavaScript）
async function chatWithStream(userMessage) {
const response = await fetch('/api/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ message: userMessage }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// 解析 SSE 数据
const lines = buffer.split('\n');
buffer = lines.pop() || ''; // 保留不完整的行
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = line.slice(6);
if (data === '[DONE]') return;
try {
const parsed = JSON.parse(data);
const content = parsed.choices?.[0]?.delta?.content;
if (content) {
// 逐字追加到页面
appendToChat(content);
}
} catch (e) {
// 忽略解析错误
}
}
}
}
}
5. Function Calling：让 LLM 使用工具
5.1 Function Calling 原理
Function Calling 是我学习过程中最兴奋的发现。它让 LLM 能够"意识到"自己需要外部工具，并生成结构化的工具调用请求。刚开始我以为 LLM 真的会去"执行"函数，后来才搞明白：
核心流程：
用户: "今天北京天气怎么样？"
│
▼
┌──────────────────┐
│ LLM 分析意图 │ → "我需要查询天气，但我没有实时数据"
│ 决定调用工具 │ → "应该调用 get_weather 函数"
└──────┬───────────┘
│ 返回 tool_calls（不是文本回复）
▼
┌──────────────────┐
│ 你的代码执行 │ → 调用天气 API，获取真实数据
│ 获取工具结果 │ → {"temperature": 22, "condition": "晴"}
└──────┬───────────┘
│ 将结果以 tool 角色传回
▼
┌──────────────────┐
│ LLM 整合信息 │ → "北京今天晴天，气温22°C，适合户外活动"
│ 生成最终回复 │
└──────────────────┘
我的理解：Function Calling 不是 LLM 真的"执行"了函数，而是 LLM 告诉你"我想调用哪个函数、传什么参数"，由你的代码来实际执行。这个设计很巧妙——LLM 负责"决策"，代码负责"执行"，各司其职。
5.2 工具定义与注册
# function_tools.py
import json
import math
from typing import Any
from datetime import datetime
# ============================================================
# 工具定义（JSON Schema 格式）
# ============================================================
TOOL_DEFINITIONS = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "获取指定城市的当前天气信息。当用户询问天气相关问题时使用。",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'、'Tokyo'",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位，celsius（摄氏度）或 fahrenheit（华氏度）",
},
},
"required": ["city"],
},
},
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算。支持基本运算、三角函数、对数等。当用户需要进行数学计算时使用。",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如'2+3*4'、'sqrt(16)'、'sin(pi/2)'",
},
},
"required": ["expression"],
},
},
},
{
"type": "function",
"function": {
"name": "search_knowledge_base",
"description": "在内部知识库中搜索信息。当用户询问需要专业知识或内部文档的问题时使用。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索查询词",
},
"top_k": {
"type": "integer",
"description": "返回结果数量，默认3",
"default": 3,
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前日期和时间。当用户询问当前时间、日期时使用。",
"parameters": {
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，如'Asia/Shanghai'、'America/New_York'，默认为'Asia/Shanghai'",
},
},
},
},
},
]
# ============================================================
# 工具实现
# ============================================================
class ToolExecutor:
"""工具执行器"""
def __init__(self):
# 注册工具名称到实现函数的映射
self._tools = {
"get_current_weather": self._get_weather,
"calculate": self._calculate,
"search_knowledge_base": self._search_kb,
"get_current_time": self._get_time,
}
def execute(self, tool_name: str, arguments: dict) -> str:
"""执行指定的工具"""
if tool_name not in self._tools:
return json.dumps({"error": f"未知工具: {tool_name}"})
try:
result = self._tools[tool_name](**arguments)
return json.dumps(result, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": str(e)}, ensure_ascii=False)
def _get_weather(self, city: str, unit: str = "celsius") -> dict:
"""获取天气（模拟数据，实际应调用天气API）"""
# 实际项目中替换为真实天气 API 调用
# 例如：requests.get(f"https://api.weather.com/v1/current?city={city}")
weather_data = {
"北京": {"temperature": 22, "condition": "晴", "humidity": 45},
"上海": {"temperature": 26, "condition": "多云", "humidity": 65},
"广州": {"temperature": 30, "condition": "阵雨", "humidity": 80},
"深圳": {"temperature": 29, "condition": "阴", "humidity": 75},
"Tokyo": {"temperature": 20, "condition": "晴", "humidity": 50},
}
default = {"temperature": 18, "condition": "未知", "humidity": 60}
weather = weather_data.get(city, default)
if unit == "fahrenheit":
weather["temperature"] = round(weather["temperature"] * 9 / 5 + 32, 1)
return {
"city": city,
"unit": unit,
**weather,
"timestamp": datetime.now().isoformat(),
}
def _calculate(self, expression: str) -> dict:
"""安全地执行数学计算"""
# 白名单：只允许安全的数学函数
allowed_names = {
"abs": abs, "round": round, "min": min, "max": max,
"sum": sum, "pow": pow,
"sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
"tan": math.tan, "log": math.log, "log10": math.log10,
"pi": math.pi, "e": math.e,
}
try:
# 编译表达式（安全沙箱）
code = compile(expression, "<calc>", "eval")
# 检查是否使用了未允许的名称
for name in code.co_names:
if name not in allowed_names:
raise ValueError(f"不允许的函数或变量: {name}")
result = eval(code, {"__builtins__": {}}, allowed_names)
return {"expression": expression, "result": result}
except Exception as e:
return {"expression": expression, "error": str(e)}
def _search_kb(self, query: str, top_k: int = 3) -> dict:
"""搜索知识库（模拟数据）"""
# 实际项目中替换为向量数据库搜索
kb = {
"python": "Python 是一种解释型、面向对象的高级编程语言，由 Guido van Rossum 于 1991 年发布。",
"openai": "OpenAI 是一家人工智能研究公司，开发了 GPT 系列大语言模型。其 API 提供了文本生成、图像生成、语音识别等服务。",
"rag": "RAG（检索增强生成）是一种结合信息检索和文本生成的 AI 技术架构，通过先检索相关文档再生成回答来提高准确性。",
"prompt": "Prompt 工程是设计和优化输入给大语言模型的提示文本的技术，目的是获得更准确、更有用的输出。",
}
results = []
for key, value in kb.items():
if query.lower() in key.lower() or query.lower() in value.lower():
results.append({"source": key, "content": value})
return {
"query": query,
"results": results[:top_k],
"total_found": len(results),
}
def _get_time(self, timezone: str = "Asia/Shanghai") -> dict:
"""获取当前时间"""
now = datetime.now()
return {
"datetime": now.isoformat(),
"date": now.strftime("%Y-%m-%d"),
"time": now.strftime("%H:%M:%S"),
"weekday": ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][now.weekday()],
"timezone": timezone,
}
5.3 完整调用流程
# llm_client.py - Function Calling 客户端
import json
from openai import OpenAI
from function_tools import TOOL_DEFINITIONS, ToolExecutor
class FunctionCallingClient:
"""支持 Function Calling 的客户端"""
def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
self.client = client
self.model = model
self.tool_executor = ToolExecutor()
def chat_with_tools(
self,
messages: list[dict],
tools: list[dict] = None,
temperature: float = 0.7,
max_iterations: int = 5,
) -> dict:
"""
支持工具调用的对话
会自动处理多轮工具调用，直到 LLM 不再请求工具或达到最大迭代次数。
Returns:
dict: {
"answer": "最终回复文本",
"tool_calls_made": [...], # 所有工具调用记录
"total_tokens": 1234,
}
"""
if tools is None:
tools = TOOL_DEFINITIONS
tool_calls_made = []
total_tokens = 0
for iteration in range(max_iterations):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools,
tool_choice="auto", # 让模型自己决定是否调用工具
temperature=temperature,
)
choice = response.choices[0]
total_tokens += response.usage.total_tokens
# 情况1：模型想调用工具
if choice.finish_reason == "tool_calls" and choice.message.tool_calls:
# 将模型的工具调用请求加入消息历史
messages.append({
"role": "assistant",
"content": choice.message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments,
},
}
for tc in choice.message.tool_calls
],
})
# 执行每个工具调用
for tc in choice.message.tool_calls:
tool_name = tc.function.name
arguments = json.loads(tc.function.arguments)
# 执行工具
result = self.tool_executor.execute(tool_name, arguments)
tool_calls_made.append({
"tool": tool_name,
"arguments": arguments,
"result": result,
})
# 将工具结果加入消息历史
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
# 继续循环，让模型处理工具结果
continue
# 情况2：模型直接回复文本
if choice.message.content:
return {
"answer": choice.message.content,
"tool_calls_made": tool_calls_made,
"total_tokens": total_tokens,
}
# 情况3：异常情况
return {
"answer": "抱歉，我无法处理这个请求。",
"tool_calls_made": tool_calls_made,
"total_tokens": total_tokens,
}
# 达到最大迭代次数
return {
"answer": "处理超时，请简化您的问题后重试。",
"tool_calls_made": tool_calls_made,
"total_tokens": total_tokens,
}
# 使用示例
if __name__ == "__main__":
client = create_client()
fc_client = FunctionCallingClient(client)
messages = [
{"role": "system", "content": "你是一个有用的助手，可以使用工具来回答问题。"},
{"role": "user", "content": "北京今天天气怎么样？顺便帮我算一下 123 * 456 等于多少。"},
]
result = fc_client.chat_with_tools(messages)
print(f"回复: {result['answer']}")
print(f"工具调用: {json.dumps(result['tool_calls_made'], ensure_ascii=False, indent=2)}")
print(f"Token 消耗: {result['total_tokens']}")
5.4 与 Prompt 工程的结合
Function Calling 不是 Prompt 工程的替代，而是增强。回顾《Prompt 工程指南》中学到的内容，我尝试把两者结合起来：
# prompt_templates.py - 结合 Prompt 工程的 Function Calling
from function_tools import TOOL_DEFINITIONS
class PromptEnhancedToolCalling:
"""结合 Prompt 工程最佳实践的 Function Calling"""
@staticmethod
def build_system_prompt_with_tools(
role: str,
task: str,
rules: list[str] = None,
tool_usage_guide: str = None,
) -> str:
"""
构建包含工具使用指导的 System Prompt
对应 Prompt 工程中的"分层 Prompt 架构"：
Layer 1（系统层）：角色 + 规则 + 工具使用指导
"""
prompt = f"你是一位{role}。\n\n"
prompt += f"【核心任务】\n{task}\n\n"
if rules:
prompt += "【行为规则】\n"
for i, rule in enumerate(rules, 1):
prompt += f"{i}. {rule}\n"
prompt += "\n"
if tool_usage_guide:
prompt += f"【工具使用指导】\n{tool_usage_guide}\n\n"
else:
prompt += """【工具使用指导】
- 当需要实时数据（天气、时间）或精确计算时，使用工具获取
- 一次可以调用多个工具，工具调用是并行的
- 获取工具结果后，用自然语言整合信息回复用户
- 如果工具返回错误，告知用户并尝试其他方式
"""
return prompt
@staticmethod
def build_few_shot_tool_examples() -> list[dict]:
"""
构建 Few-shot 工具调用示例
对应 Prompt 工程中的"少样本提示（Few-shot Prompting）"
通过示例教会模型何时使用工具、如何使用
"""
return [
{"role": "user", "content": "上海今天热吗？"},
{
"role": "assistant",
"content": None,
"tool_calls": [{
"id": "call_example_1",
"type": "function",
"function": {
"name": "get_current_weather",
"arguments": '{"city": "上海", "unit": "celsius"}',
},
}],
},
{
"role": "tool",
"tool_call_id": "call_example_1",
"content": '{"city": "上海", "temperature": 26, "condition": "多云", "humidity": 65}',
},
{
"role": "assistant",
"content": "上海今天多云，气温26°C，湿度65%。体感偏热，建议穿短袖，出门注意防晒。",
},
]
@staticmethod
def build_cot_tool_prompt(question: str) -> str:
"""
构建 CoT + 工具调用的 Prompt
对应 Prompt 工程中的"思维链提示（Chain-of-Thought）"
引导模型先思考需要什么信息，再决定调用什么工具
"""
return f"""请按以下步骤思考并回答：
步骤1：分析问题，确定需要哪些信息
步骤2：判断哪些信息可以通过工具获取
步骤3：调用相应的工具
步骤4：整合工具返回的信息，给出最终答案
问题：{question}
请开始分析："""
6. Token 计算与成本控制
6.1 Token 计算原理
Token 是 LLM 处理文本的基本单位。刚开始我以为是按字符或单词算的，后来才发现完全不是那么回事。搞清楚 Token 怎么算，是控制成本的第一步：
Token 不等于字符，也不等于单词：
英文：1 token ≈ 0.75 个单词 ≈ 4 个字符
"Hello, world!" → 4 tokens
中文：1 token ≈ 0.5-1 个汉字
"你好世界" → 4-6 tokens（取决于模型的分词器）
代码：1 token ≈ 3-4 个字符
"def foo():" → 4 tokens
# token_tracker.py
import tiktoken
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class TokenUsage:
"""Token 使用记录"""
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
def __add__(self, other: "TokenUsage") -> "TokenUsage":
return TokenUsage(
prompt_tokens=self.prompt_tokens + other.prompt_tokens,
completion_tokens=self.completion_tokens + other.completion_tokens,
total_tokens=self.total_tokens + other.total_tokens,
)
class TokenCalculator:
"""Token 计算器"""
# 各模型的价格（每百万 Token，美元）
# 价格可能变动，请以 OpenAI 官网为准
PRICING = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"gpt-4-turbo": {"input": 10.00, "output": 30.00},
"gpt-4": {"input": 30.00, "output": 60.00},
"gpt-3.5-turbo": {"input": 0.50, "output": 1.50},
}
def __init__(self, model: str = "gpt-4o-mini"):
self.model = model
try:
self.encoding = tiktoken.encoding_for_model(model)
except KeyError:
self.encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(self, text: str) -> int:
"""计算文本的 Token 数量"""
return len(self.encoding.encode(text))
def count_messages_tokens(self, messages: list[dict]) -> int:
"""
计算消息列表的 Token 数量
参考 OpenAI 的计数规则：
- 每条消息有固定的开销（约 3-4 tokens）
- 角色名称计入 token
- 工具定义也计入 token
"""
tokens_per_message = 3 # 每条消息的固定开销
num_tokens = 0
for message in messages:
num_tokens += tokens_per_message
for key, value in message.items():
if key == "content" and value:
num_tokens += self.count_tokens(str(value))
elif key == "tool_calls":
# 工具调用也计入 token
num_tokens += self.count_tokens(str(value))
elif key == "name":
num_tokens += self.count_tokens(str(value))
num_tokens += 3 # 每次回复的固定开销
return num_tokens
def estimate_cost(
self,
prompt_tokens: int,
completion_tokens: int,
model: str = None,
) -> dict:
"""估算 API 调用成本"""
model = model or self.model
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
output_cost = (completion_tokens / 1_000_000) * pricing["output"]
total_cost = input_cost + output_cost
return {
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 6),
"total_cost_usd": round(total_cost, 6),
"total_cost_cny": round(total_cost * 7.2, 6), # 粗略汇率
}
class TokenTracker:
"""Token 使用追踪器（用于统计整个会话的成本）"""
def __init__(self, model: str = "gpt-4o-mini"):
self.model = model
self.calculator = TokenCalculator(model)
self.usage = TokenUsage()
self.call_history: list[dict] = []
def record(self, prompt_tokens: int, completion_tokens: int):
"""记录一次 API 调用的 Token 使用"""
self.usage.prompt_tokens += prompt_tokens
self.usage.completion_tokens += completion_tokens
self.usage.total_tokens += prompt_tokens + completion_tokens
cost = self.calculator.estimate_cost(prompt_tokens, completion_tokens)
self.call_history.append(cost)
def get_summary(self) -> dict:
"""获取使用摘要"""
total_cost = self.calculator.estimate_cost(
self.usage.prompt_tokens,
self.usage.completion_tokens,
)
return {
"model": self.model,
"total_calls": len(self.call_history),
"total_prompt_tokens": self.usage.prompt_tokens,
"total_completion_tokens": self.usage.completion_tokens,
"total_tokens": self.usage.total_tokens,
"total_cost_usd": total_cost["total_cost_usd"],
"total_cost_cny": total_cost["total_cost_cny"],
"call_history": self.call_history,
}
# 使用示例
if __name__ == "__main__":
calc = TokenCalculator("gpt-4o-mini")
text = "Prompt 工程是设计和优化输入给大语言模型的提示文本的技术。"
tokens = calc.count_tokens(text)
print(f"文本: {text}")
print(f"Token 数: {tokens}")
messages = [
{"role": "system", "content": "你是一个有用的助手。"},
{"role": "user", "content": "解释什么是 RAG。"},
]
msg_tokens = calc.count_messages_tokens(messages)
print(f"消息 Token 数: {msg_tokens}")
cost = calc.estimate_cost(500, 200)
print(f"预估成本: ${cost['total_cost_usd']} (约 {cost['total_cost_cny']} 元)")
6.2 成本估算与优化
# 成本优化策略汇总
class CostOptimizer:
"""API 成本优化器"""
@staticmethod
def select_model(task_complexity: str) -> str:
"""
根据任务复杂度选择模型
简单任务用便宜模型，复杂任务用强模型
"""
model_map = {
"simple": "gpt-4o-mini", # 简单问答、分类 → $0.15/M input
"moderate": "gpt-4o-mini", # 摘要、翻译 → $0.15/M input
"complex": "gpt-4o", # 复杂推理、代码生成 → $2.50/M input
"critical": "gpt-4o", # 关键业务 → $2.50/M input
}
return model_map.get(task_complexity, "gpt-4o-mini")
@staticmethod
def optimize_messages(
messages: list[dict],
max_context_tokens: int = 4000,
) -> list[dict]:
"""
优化消息列表，控制上下文长度
策略：
1. System Prompt 始终保留
2. 保留最近的 N 轮对话
3. 超出部分用摘要替代
"""
if len(messages) <= 3:
return messages
calc = TokenCalculator()
system_msgs = [m for m in messages if m["role"] == "system"]
other_msgs = [m for m in messages if m["role"] != "system"]
# 从后往前保留，直到接近 token 限制
kept = []
current_tokens = sum(calc.count_tokens(m.get("content", "")) for m in system_msgs)
for msg in reversed(other_msgs):
msg_tokens = calc.count_tokens(msg.get("content", ""))
if current_tokens + msg_tokens > max_context_tokens:
break
kept.insert(0, msg)
current_tokens += msg_tokens
return system_msgs + kept
@staticmethod
def estimate_conversation_cost(
avg_input_tokens: int,
avg_output_tokens: int,
conversations_per_day: int,
model: str = "gpt-4o-mini",
) -> dict:
"""估算每日/每月成本"""
calc = TokenCalculator(model)
daily_input = avg_input_tokens * conversations_per_day
daily_output = avg_output_tokens * conversations_per_day
daily_cost = calc.estimate_cost(daily_input, daily_output)
monthly_cost = calc.estimate_cost(daily_input * 30, daily_output * 30)
return {
"model": model,
"conversations_per_day": conversations_per_day,
"daily_cost_usd": daily_cost["total_cost_usd"],
"daily_cost_cny": daily_cost["total_cost_cny"],
"monthly_cost_usd": monthly_cost["total_cost_usd"],
"monthly_cost_cny": monthly_cost["total_cost_cny"],
}
# 使用示例
optimizer = CostOptimizer()
cost_estimate = optimizer.estimate_conversation_cost(
avg_input_tokens=500,
avg_output_tokens=300,
conversations_per_day=100,
model="gpt-4o-mini",
)
print(f"使用 gpt-4o-mini，每天 100 次对话：")
print(f" 每日成本: ${cost_estimate['daily_cost_usd']:.4f} (约 {cost_estimate['daily_cost_cny']:.2f} 元)")
print(f" 每月成本: ${cost_estimate['monthly_cost_usd']:.4f} (约 {cost_estimate['monthly_cost_cny']:.2f} 元)")
7. 错误处理与重试策略
7.1 错误类型分类
实际跑起来之后，我发现 API 调用出错的概率比想象中高得多。网络波动、限流、超时、内容过滤……各种情况都遇到过。我把常见的错误整理成了下面这张表：
| 错误类型 | HTTP 状态码 | 典型场景 | 是否可重试 |
|---|---|---|---|
RateLimitError |
429 | 超过 RPM/TPM 限制 | 是（等待后重试） |
APIConnectionError |
- | 网络问题、DNS 解析失败 | 是 |
APITimeoutError |
- | 请求超时 | 是 |
InternalServerError |
500 | OpenAI 服务器内部错误 | 是 |
ServiceUnavailableError |
503 | 服务暂时不可用 | 是 |
AuthenticationError |
401 | API Key 无效或过期 | 否 |
PermissionDeniedError |
403 | 无权限访问该资源 | 否 |
InvalidRequestError |
400 | 请求参数错误 | 否（修正参数后重试） |
ContentFilterError |
400 | 内容被安全过滤 | 否 |
7.2 指数退避重试
# llm_client.py - 带重试的客户端
import time
import random
from functools import wraps
from openai import (
OpenAI,
RateLimitError,
APIConnectionError,
APITimeoutError,
InternalServerError,
APIStatusError,
)
class RetryConfig:
"""重试配置"""
def __init__(
self,
max_retries: int = 3,
initial_delay: float = 1.0,
backoff_factor: float = 2.0,
max_delay: float = 60.0,
jitter: bool = True,
retryable_errors: tuple = (
RateLimitError,
APIConnectionError,
APITimeoutError,
InternalServerError,
),
):
self.max_retries = max_retries
self.initial_delay = initial_delay
self.backoff_factor = backoff_factor
self.max_delay = max_delay
self.jitter = jitter
self.retryable_errors = retryable_errors
def get_delay(self, attempt: int) -> float:
"""计算第 N 次重试的延迟时间（指数退避）"""
delay = self.initial_delay * (self.backoff_factor ** attempt)
delay = min(delay, self.max_delay)
if self.jitter:
# 添加随机抖动，避免"惊群效应"
delay = delay * (0.5 + random.random())
return delay
def with_retry(retry_config: RetryConfig = None):
"""重试装饰器"""
if retry_config is None:
retry_config = RetryConfig()
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(retry_config.max_retries + 1):
try:
return func(*args, **kwargs)
except retry_config.retryable_errors as e:
last_error = e
if attempt == retry_config.max_retries:
# 达到最大重试次数，抛出最终错误
raise RuntimeError(
f"API 调用失败，已重试 {retry_config.max_retries} 次。"
f"最后错误: {type(e).__name__}: {str(e)[:200]}"
) from e
delay = retry_config.get_delay(attempt)
print(
f"[重试 {attempt + 1}/{retry_config.max_retries}] "
f"{type(e).__name__}，等待 {delay:.1f} 秒..."
)
time.sleep(delay)
except APIStatusError as e:
# 非可重试错误，直接抛出
raise RuntimeError(
f"API 返回错误 (HTTP {e.status_code}): {str(e)[:500]}"
) from e
# 理论上不会到这里
raise last_error
return wrapper
return decorator
class RobustClient:
"""带重试和错误处理的 OpenAI 客户端"""
def __init__(
self,
client: OpenAI,
model: str = "gpt-4o-mini",
retry_config: RetryConfig = None,
):
self.client = client
self.model = model
self.retry_config = retry_config or RetryConfig()
def chat(
self,
messages: list[dict],
temperature: float = 0.7,
max_tokens: int = 2048,
**kwargs,
) -> dict:
"""
带重试的对话接口
Returns:
dict: {
"success": True/False,
"content": "回复内容" or None,
"error": "错误信息" or None,
"usage": {...} or None,
"retries": 0,
}
"""
last_error = None
for attempt in range(self.retry_config.max_retries + 1):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs,
)
return {
"success": True,
"content": response.choices[0].message.content,
"error": None,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
"total_tokens": response.usage.total_tokens,
},
"retries": attempt,
"finish_reason": response.choices[0].finish_reason,
}
except RateLimitError as e:
last_error = e
if attempt < self.retry_config.max_retries:
delay = self.retry_config.get_delay(attempt)
print(f"[限流] 等待 {delay:.1f} 秒后重试...")
time.sleep(delay)
except (APIConnectionError, APITimeoutError, InternalServerError) as e:
last_error = e
if attempt < self.retry_config.max_retries:
delay = self.retry_config.get_delay(attempt)
print(f"[{type(e).__name__}] 等待 {delay:.1f} 秒后重试...")
time.sleep(delay)
except Exception as e:
# 不可重试的错误
return {
"success": False,
"content": None,
"error": f"{type(e).__name__}: {str(e)[:300]}",
"usage": None,
"retries": attempt,
"finish_reason": None,
}
# 所有重试都失败了
return {
"success": False,
"content": None,
"error": f"重试 {self.retry_config.max_retries} 次后仍然失败: {type(last_error).__name__}",
"usage": None,
"retries": self.retry_config.max_retries,
"finish_reason": None,
}
7.3 限流处理
# rate_limiter.py
import time
import threading
from collections import deque
from dataclasses import dataclass
@dataclass
class RateLimitConfig:
"""限流配置"""
max_requests_per_minute: int = 500
max_tokens_per_minute: int = 200000
# 安全余量：实际限制设为官方限制的 80%
safety_margin: float = 0.8
class TokenBucketRateLimiter:
"""令牌桶限流器"""
def __init__(self, config: RateLimitConfig = None):
self.config = config or RateLimitConfig()
self.lock = threading.Lock()
# 计算带安全余量的实际限制
self.max_rpm = int(self.config.max_requests_per_minute * self.config.safety_margin)
self.max_tpm = int(self.config.max_tokens_per_minute * self.config.safety_margin)
# 请求记录（滑动窗口）
self.request_times: deque = deque()
self.token_usage: deque = deque() # (timestamp, tokens)
def _clean_old_records(self):
"""清理超过 1 分钟的旧记录"""
now = time.time()
one_minute_ago = now - 60
while self.request_times and self.request_times[0] < one_minute_ago:
self.request_times.popleft()
while self.token_usage and self.token_usage[0][0] < one_minute_ago:
self.token_usage.popleft()
def acquire(self, estimated_tokens: int = 0) -> bool:
"""
尝试获取请求许可
Args:
estimated_tokens: 预估的 Token 消耗
Returns:
bool: 是否允许发送请求
"""
with self.lock:
self._clean_old_records()
# 检查请求频率
if len(self.request_times) >= self.max_rpm:
return False
# 检查 Token 消耗
if estimated_tokens > 0:
current_tokens = sum(t[1] for t in self.token_usage)
if current_tokens + estimated_tokens > self.max_tpm:
return False
# 记录本次请求
now = time.time()
self.request_times.append(now)
if estimated_tokens > 0:
self.token_usage.append((now, estimated_tokens))
return True
def wait_and_acquire(self, estimated_tokens: int = 0, timeout: float = 30.0) -> bool:
"""等待直到获取许可或超时"""
start = time.time()
while time.time() - start < timeout:
if self.acquire(estimated_tokens):
return True
time.sleep(0.1)
return False
def get_stats(self) -> dict:
"""获取当前限流统计"""
with self.lock:
self._clean_old_records()
return {
"requests_this_minute": len(self.request_times),
"requests_limit": self.max_rpm,
"requests_usage_pct": round(len(self.request_times) / self.max_rpm * 100, 1),
"tokens_this_minute": sum(t[1] for t in self.token_usage),
"tokens_limit": self.max_tpm,
"tokens_usage_pct": round(
sum(t[1] for t in self.token_usage) / self.max_tpm * 100, 1
),
}
8. 多轮对话管理
多轮对话看起来简单，就是把历史消息传回去嘛。但实际做的时候发现有两个棘手的问题：一是上下文越来越长，Token 消耗越来越大；二是超过模型上下文窗口后，前面的对话就丢了。下面是我摸索出来的处理方案。
8.1 对话上下文窗口管理
# conversation_manager.py
from token_tracker import TokenCalculator
class ConversationManager:
"""多轮对话管理器"""
def __init__(
self,
system_prompt: str,
model: str = "gpt-4o-mini",
max_context_tokens: int = 8000,
max_turns: int = 20,
):
self.system_prompt = system_prompt
self.model = model
self.max_context_tokens = max_context_tokens
self.max_turns = max_turns
self.token_calc = TokenCalculator(model)
# 对话历史
self.messages: list[dict] = [
{"role": "system", "content": system_prompt}
]
def add_user_message(self, content: str):
"""添加用户消息"""
self.messages.append({"role": "user", "content": content})
def add_assistant_message(self, content: str):
"""添加助手回复"""
self.messages.append({"role": "assistant", "content": content})
def get_messages(self) -> list[dict]:
"""获取当前消息列表（自动裁剪）"""
self._trim_if_needed()
return self.messages
def _trim_if_needed(self):
"""如果上下文过长，自动裁剪"""
total_tokens = self.token_calc.count_messages_tokens(self.messages)
if total_tokens <= self.max_context_tokens:
return
# 策略：保留 system prompt + 最近的对话
# 被裁剪的旧对话用摘要替代
system_msg = self.messages[0]
conversation = self.messages[1:]
# 从后往前保留
kept = []
current_tokens = self.token_calc.count_tokens(system_msg["content"])
for msg in reversed(conversation):
msg_tokens = self.token_calc.count_tokens(msg.get("content", ""))
if current_tokens + msg_tokens > self.max_context_tokens:
break
kept.insert(0, msg)
current_tokens += msg_tokens
# 如果有被裁剪的消息，在开头插入摘要提示
if len(kept) < len(conversation):
trimmed_count = len(conversation) - len(kept)
summary_msg = {
"role": "system",
"content": f"[注意：由于上下文长度限制，已省略较早的 {trimmed_count} 轮对话。请基于当前可见的对话继续。]",
}
self.messages = [system_msg, summary_msg] + kept
else:
self.messages = [system_msg] + kept
def get_conversation_summary(self) -> dict:
"""获取对话摘要"""
user_messages = [m for m in self.messages if m["role"] == "user"]
assistant_messages = [m for m in self.messages if m["role"] == "assistant"]
total_tokens = self.token_calc.count_messages_tokens(self.messages)
return {
"total_messages": len(self.messages),
"user_messages": len(user_messages),
"assistant_messages": len(assistant_messages),
"total_turns": len(user_messages),
"total_tokens": total_tokens,
"context_usage_pct": round(
total_tokens / self.max_context_tokens * 100, 1
),
}
def clear(self):
"""清空对话历史（保留 system prompt）"""
self.messages = [
{"role": "system", "content": self.system_prompt}
]
8.2 对话摘要与压缩
class ConversationCompressor:
"""对话压缩器 —— 将长对话压缩为摘要"""
def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
self.client = client
self.model = model
def summarize_history(
self,
messages: list[dict],
keep_last_n: int = 3,
) -> list[dict]:
"""
压缩对话历史
策略：
1. 保留最近 N 轮对话不变
2. 将更早的对话压缩为一段摘要
3. 摘要作为 system 消息的一部分注入
"""
if len(messages) <= keep_last_n * 2 + 2:
return messages # 不需要压缩
system_msg = messages[0]
recent = messages[-(keep_last_n * 2):] # 保留最近 N 轮（每轮 user+assistant）
old = messages[1:-(keep_last_n * 2)] # 需要压缩的旧对话
# 构建摘要请求
old_conversation = "\n".join([
f"{'用户' if m['role'] == 'user' else '助手'}: {m.get('content', '')[:200]}"
for m in old
if m["role"] in ("user", "assistant")
])
summary_prompt = f"""请将以下对话历史压缩为一段简洁的摘要（不超过200字）。
只保留关键信息：用户的主要问题、助手给出的重要结论、任何需要记住的上下文。
对话历史：
{old_conversation}
摘要："""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": summary_prompt}],
temperature=0.3,
max_tokens=300,
)
summary = response.choices[0].message.content
# 构建新的消息列表
compressed = [
system_msg,
{
"role": "system",
"content": f"[历史对话摘要] {summary}",
},
] + recent
return compressed
9. 结构化输出
让 LLM 输出 JSON 格式在实际开发中太常用了——对接前端、存入数据库、调用下游服务，都需要结构化的数据。我摸索了两种方式：
# structured_output.py
import json
from openai import OpenAI
class StructuredOutputClient:
"""结构化输出客户端"""
def __init__(self, client: OpenAI, model: str = "gpt-4o-mini"):
self.client = client
self.model = model
def json_output(
self,
prompt: str,
system_prompt: str = None,
temperature: float = 0.1,
) -> dict:
"""
获取 JSON 格式的输出
使用 response_format 参数强制 JSON 输出
"""
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.append({"role": "user", "content": prompt})
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=temperature,
response_format={"type": "json_object"},
)
content = response.choices[0].message.content
return json.loads(content)
def json_with_schema(
self,
prompt: str,
json_schema: dict,
system_prompt: str = None,
) -> dict:
"""
获取符合 JSON Schema 的输出
通过在 Prompt 中嵌入 Schema 来约束输出结构
"""
schema_text = json.dumps(json_schema, ensure_ascii=False, indent=2)
full_prompt = f"""{prompt}
【输出格式要求】
请严格按照以下 JSON Schema 输出，确保所有必填字段都存在：
{schema_text}
只输出 JSON，不要包含其他文字。"""
return self.json_output(full_prompt, system_prompt)
# 使用示例
if __name__ == "__main__":
client = create_client()
so_client = StructuredOutputClient(client)
# 定义输出 Schema
analysis_schema = {
"type": "object",
"properties": {
"sentiment": {
"type": "string",
"enum": ["positive", "negative", "neutral"],
"description": "情感倾向",
},
"confidence": {
"type": "number",
"minimum": 0,
"maximum": 100,
"description": "置信度",
},
"keywords": {
"type": "array",
"items": {"type": "string"},
"description": "关键词列表",
},
"summary": {
"type": "string",
"maxLength": 100,
"description": "一句话摘要",
},
},
"required": ["sentiment", "confidence", "keywords", "summary"],
}
result = so_client.json_with_schema(
prompt="分析以下评论：'产品质量很好，但物流太慢了，等了一周才到。'",
json_schema=analysis_schema,
system_prompt="你是一个专业的文本分析助手。",
)
print(json.dumps(result, ensure_ascii=False, indent=2))
10. 生产环境最佳实践
把代码从"能跑"变成"能上线"，中间还有不少坑要填。下面是我整理的一些实践经验。
10.1 架构设计
┌─────────────────────────────────────────────────────────┐
│ 生产环境 API 调用架构 │
│ │
│ 应用层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 对话管理 │ │ Prompt │ │ 结果缓存 │ │
│ │ │ │ 模板引擎 │ │ │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ └─────────────┼─────────────┘ │
│ │ │
│ ┌─────┴─────┐ │
│ │ API 网关 │ ← 统一入口 │
│ └─────┬─────┘ │
│ │ │
│ 中间层 ┌──┴──┐ │
│ │ 限流器 │ ← Token Bucket │
│ └──┬──┘ │
│ ┌──┴──┐ │
│ │ 重试 │ ← 指数退避 │
│ └──┬──┘ │
│ ┌──┴──┐ │
│ │ 降级 │ ← 主模型不可用时切换备用模型 │
│ └──┬──┘ │
│ │ │
│ API 层 ┌───┴───┐ │
│ │ OpenAI │ │
│ │ API │ │
│ └───────┘ │
└─────────────────────────────────────────────────────────┘
10.2 我的实践清单
| 实践 | 说明 | 优先级 |
|---|---|---|
| API Key 环境变量化 | 永远不要硬编码 API Key | 必须 |
| 请求日志记录 | 记录每次调用的模型、Token、延迟、成本 | 必须 |
| 指数退避重试 | 处理网络波动和限流 | 必须 |
| 限流保护 | 避免超出 API 配额 | 必须 |
| 超时设置 | 避免请求无限等待 | 必须 |
| 内容安全过滤 | 输入和输出都需要安全检查 | 必须 |
| 模型降级 | 主模型不可用时切换备用模型 | 推荐 |
| 响应缓存 | 相同问题缓存结果，减少 API 调用 | 推荐 |
| 异步调用 | 使用异步客户端提升吞吐量 | 推荐 |
| 健康检查 | 定期检查 API 可用性 | 推荐 |
11. 配套项目：AI 对话应用
学完这些之后，我动手搭了一个基于 Streamlit 的 AI 对话应用，把上面讲的所有技术都集成进去了。跑通这个项目，基本就掌握了 OpenAI API 的核心用法。
项目结构
ai-chat-app/
├── app.py # Streamlit 主应用
├── config.py # 配置管理
├── llm_client.py # OpenAI API 客户端封装
├── function_tools.py # Function Calling 工具定义与实现
├── prompt_templates.py # Prompt 模板（结合 Prompt 工程）
├── token_tracker.py # Token 计算与成本追踪
├── conversation_manager.py # 多轮对话管理
├── requirements.txt # 项目依赖
└── .env.example # 环境变量模板
应用功能
- 流式对话：AI 回复逐字显示，体验流畅
- Function Calling：支持天气查询、数学计算、知识库搜索、时间查询
- Token 追踪：实时显示 Token 消耗和成本估算
- Prompt 模板切换：内置标准、Few-shot、CoT、ReAct、代码助手、翻译助手六种 Prompt 模式
- 错误处理：自动重试、限流保护、优雅降级
- 对话导出：支持导出对话记录
项目文件详见配套代码目录 d:\aiagent\学习资料\ai-chat-app\
12. 总结与回顾
我的学习历程
回头看这段学习经历，大概经历了四个阶段：
第一阶段：基础调用
→ 跑通第一个 Chat Completions API 调用
→ 搞懂消息角色体系（system/user/assistant）
→ 摸清 temperature、max_tokens 这些参数的作用
第二阶段：流式与工具
→ 实现流式输出，体验瞬间提升
→ 啃下 Function Calling，让 LLM 能"干活"了
→ 尝试把 Prompt 工程的最佳实践融入 API 调用
第三阶段：工程化
→ 学会计算 Token，开始关注成本
→ 加上错误处理和重试，应用不再动不动就崩
→ 处理多轮对话的上下文管理
第四阶段：打磨
→ 加上限流保护
→ 实现对话压缩
→ 搭了一个完整的 Streamlit 应用来验证
几点心得
- API Key 是敏感信息：永远不要提交到版本控制，用
.env管理 - 流式输出是体验的分水岭：加了之后用户感知的等待时间从 5 秒降到几乎为 0
- Function Calling 是能力放大器：让 LLM 从"聊天机器人"变成"智能助手"
- Token 即成本：每次 API 调用都在花钱，养成计算 Token 的习惯
- 错误处理不是可选项：生产环境中网络波动、限流、服务故障是常态，不处理根本没法用
- Prompt 工程 + API 工程 = 完整的 AI 应用能力：两者缺一不可
与 Prompt 工程文档的关系
本文是《Prompt 工程指南》的实践篇。我个人是按以下顺序学习的，供参考：
- 先读《Prompt 工程指南》，理解 Prompt 设计的理论基础
- 再读本文，学习如何通过 API 将 Prompt 设计落地
- 跑通配套项目，在实践中巩固所学
我参考的资料
- OpenAI API 官方文档：https://platform.openai.com/docs/api-reference
- OpenAI Cookbook：https://cookbook.openai.com/
- tiktoken 仓库：https://github.com/openai/tiktoken
- Streamlit 文档：https://docs.streamlit.io/
文档版本：v1.0 | 更新时间：2026-05-10 | 适用读者：AI 应用开发者 / LLM 工程师 / 对 OpenAI API 感兴趣的学习者
本文为原创技术文档，配套项目代码位于
https://github.com/yxy008/ai-prompt.git。所有代码示例均可直接运行。如需转载，请注明出处。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
13
0- 0
已为社区贡献2条内容
所有评论(0)