1. OpenAI API 基础
1.1 安装与配置
在使用 OpenAI API 之前，需要先安装官方 SDK 并配置 API Key。推荐使用 python-dotenv 管理环境变量，避免将密钥硬编码到代码中。
pip install openai python-dotenv
import os
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 基础调用
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个 helpful 助手"},
{"role": "user", "content": "解释 Python 的 GIL 是什么"}
],
temperature=0.7,
max_tokens=500
)
print(response.choices[0].message.content)
temperature 参数控制输出的随机性，取值范围 0 到 2。值为 0 时输出最确定性，适合代码生成和结构化数据提取；值为 0.7 到 1 时适合创意写作和头脑风暴。max_tokens 限制单次响应的最大 token 数，可防止意外产生超长输出导致费用激增。
1.2 国产大模型适配
除 OpenAI 外，国内主流大模型平台均提供 OpenAI SDK 兼容接口，只需替换 base_url 和 API Key 即可无缝切换。
from openai import OpenAI
import os
# 阿里云通义千问
qwen = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# 百度文心一言
wenxin = OpenAI(
api_key=os.getenv("QIANFAN_API_KEY"),
base_url="https://qianfan.baidubce.com/v2"
)
# 智谱 GLM-4
glm = OpenAI(
api_key=os.getenv("ZHIPU_API_KEY"),
base_url="https://open.bigmodel.cn/api/paas/v4/"
)
# 统一调用方式
resp = qwen.chat.completions.create(
model="qwen-max",
messages=[{"role": "user", "content": "你好"}]
)
print(resp.choices[0].message.content)
2. OpenAI API 进阶
2.1 Function Calling 工具调用
Function Calling 允许大模型根据用户意图自动选择并调用外部函数，将自然语言转化为结构化操作。这是构建 AI Agent 的核心能力之一。
import json
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def get_weather(city: str) -> str:
"""查询城市天气（模拟）"""
weather_data = {
"北京": "晴天 25°C",
"上海": "多云 22°C",
"深圳": "小雨 28°C"
}
return weather_data.get(city, "未知城市")
def search_database(query: str) -> str:
"""模拟数据库查询"""
db = {
"Python": "Python 是 Guido van Rossum 于 1991 年发布的高级编程语言",
"FastAPI": "FastAPI 是 Sebastián Ramírez 开发的高性能 Python Web 框架"
}
return db.get(query, "未找到相关信息")
# 定义工具列表
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，如北京、上海"}
},
"required": ["city"]
}
}
},
{
"type": "function",
"function": {
"name": "search_database",
"description": "搜索内部知识库",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
}
]
def chat_with_tools(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
# 处理工具调用
if message.tool_calls:
messages.append(message)
for tool_call in message.tool_calls:
function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
if function_name == "get_weather":
result = get_weather(**arguments)
elif function_name == "search_database":
result = search_database(**arguments)
else:
result = "未知函数"
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
final = client.chat.completions.create(
model="gpt-4o",
messages=messages
)
return final.choices[0].message.content
return message.content
# 运行
print(chat_with_tools("北京今天天气怎么样？"))
print(chat_with_tools("FastAPI 是谁开发的？"))
2.2 JSON Mode 结构化输出
当需要模型返回严格符合 JSON Schema 的数据时，可以启用 JSON Mode，确保输出可直接被程序解析，无需额外的正则提取或容错处理。
import json
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
system_prompt = """你是一个信息提取助手。请从用户输入中提取实体信息，
并以 JSON 格式输出，包含以下字段：name（姓名）、age（年龄）、skills（技能列表）、city（城市）。"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": "张三今年28岁，住在深圳，精通 Python、Docker 和 Kubernetes。"}
],
response_format={"type": "json_object"},
temperature=0.1
)
result = json.loads(response.choices[0].message.content)
print(json.dumps(result, ensure_ascii=False, indent=2))
# 输出:
# {
# "name": "张三",
# "age": 28,
# "skills": ["Python", "Docker", "Kubernetes"],
# "city": "深圳"
# }
2.3 Assistants API
Assistants API 是 OpenAI 提供的高阶接口，内置线程管理、文件检索、代码解释器和函数调用能力，适合构建持久化对话的 AI 助手应用。
from openai import OpenAI
import os
import time
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 创建 Assistant
assistant = client.beta.assistants.create(
name="Python 导师",
instructions="你是一位 Python 编程专家，帮助用户解答 Python 相关问题。",
model="gpt-4o"
)
# 创建对话线程
thread = client.beta.threads.create()
# 添加用户消息
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="请解释 Python 中的装饰器是什么，并给出一个示例。"
)
# 运行 Assistant
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)
# 轮询等待完成
while True:
run_status = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
if run_status.status == "completed":
break
time.sleep(1)
# 获取回复
messages = client.beta.threads.messages.list(thread_id=thread.id)
for msg in messages.data:
if msg.role == "assistant":
for content in msg.content:
if content.type == "text":
print(content.text.value)
# 清理
client.beta.assistants.delete(assistant.id)
3. 提示工程（Prompt Engineering）
提示工程是通过优化输入文本来引导大模型产生高质量输出的技术。掌握提示工程可以显著提升模型在特定任务上的表现，无需进行昂贵的微调。
3.1 Few-shot 少样本提示
Few-shot 提示通过在问题前提供若干输入输出示例，让模型快速理解任务模式和输出格式，特别适用于分类、抽取和格式转换任务。
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
few_shot_prompt = """请将以下文本按情感分类为：正面、负面、中性。
示例 1：
文本：这部电影太精彩了，情节紧凑，演员演技在线。
分类：正面
示例 2：
文本：排队等了三小时，结果餐厅关门了，真的无语。
分类：负面
示例 3：
文本：今天天气晴朗，气温 25 度。
分类：中性
现在请分类：
文本：新发布的 Python 3.13 性能提升明显，解释器启动速度更快了。
分类："""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": few_shot_prompt}],
temperature=0.1,
max_tokens=10
)
print(response.choices[0].message.content.strip())
# 输出: 正面
3.2 Chain-of-Thought 思维链
Chain-of-Thought（CoT）提示引导模型在给出最终答案前先展示推理过程。这种方法在数学计算、逻辑推理和复杂问答任务中效果显著，能大幅减少模型幻觉。
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
cot_prompt = """请逐步思考并回答以下问题。
问题：一个篮子里有 5 个苹果和 3 个橙子。小明拿走了 2 个苹果和 1 个橙子，
然后小红又放进来了 4 个苹果。现在篮子里分别有多少个苹果和橙子？
请按以下步骤回答：
1. 初始数量
2. 小明拿走后的数量
3. 小红放入后的最终数量
4. 总结答案"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": cot_prompt}],
temperature=0.2
)
print(response.choices[0].message.content)
3.3 ReAct 推理与行动
ReAct（Reasoning + Acting）是一种将 LLM 的推理能力与外部工具调用相结合的提示框架。模型通过交替进行 Thought（思考）、Action（行动）、Observation（观察）三个步骤来逐步解决问题。
from openai import OpenAI
import os
import json
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def calculate(expression: str) -> str:
"""安全计算数学表达式"""
try:
allowed_names = {"abs": abs, "max": max, "min": min, "sum": sum, "len": len}
return str(eval(expression, {"__builtins__": {}}, allowed_names))
except Exception as e:
return f"计算错误: {e}"
def search_wiki(term: str) -> str:
"""模拟百科搜索"""
knowledge = {
"爱因斯坦": "阿尔伯特·爱因斯坦（1879-1955），德国出生的理论物理学家，相对论创立者。",
"Python": "Python 是一种高级通用编程语言，由 Guido van Rossum 于 1991 年创建。"
}
return knowledge.get(term, f"未找到关于 '{term}' 的信息")
react_prompt = """你是一个智能助手，需要通过推理和调用工具来回答问题。
可用工具：
- calculate(expression): 计算数学表达式
- search_wiki(term): 搜索百科知识
请按照以下格式回答：
Thought: [你的思考过程]
Action: [工具名称(参数)]
Observation: [工具返回结果]
...（重复 Thought/Action/Observation 直到问题解决）
Thought: [最终思考]
Answer: [最终答案]
问题：爱因斯坦出生在哪一年？他的儿子汉斯·爱因斯坦在 1973 年时多少岁？
"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": react_prompt}],
temperature=0.2
)
print(response.choices[0].message.content)
4. 向量数据库与 RAG 实战
4.1 Embedding 与语义搜索
Embedding 将文本映射到高维向量空间，语义相近的文本在向量空间中距离更近。这是 RAG 检索增强生成的核心技术基础。
from openai import OpenAI
import numpy as np
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def get_embedding(text: str) -> list[float]:
resp = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return resp.data[0].embedding
def cosine_similarity(a: list[float], b: list[float]) -> float:
a_vec = np.array(a)
b_vec = np.array(b)
return float(np.dot(a_vec, b_vec) / (np.linalg.norm(a_vec) * np.linalg.norm(b_vec)))
# 示例文档
documents = [
"Python 3.12 于 2023 年 10 月发布，引入了改进的错误消息和 f-string 调试语法",
"FastAPI 是一个现代 Python Web 框架，基于 Starlette 和 Pydantic，支持异步",
"Docker 容器化技术可以将 Python 应用打包成镜像，实现环境一致性"
]
# 计算所有文档的 Embedding
doc_embeddings = [get_embedding(doc) for doc in documents]
# 查询
query = "什么是 FastAPI？"
query_embedding = get_embedding(query)
# 计算相似度并排序
results = []
for i, doc_emb in enumerate(doc_embeddings):
sim = cosine_similarity(query_embedding, doc_emb)
results.append((sim, documents[i]))
results.sort(reverse=True)
print(f"查询: {query}")
for sim, doc in results:
print(f"[相似度 {sim:.4f}] {doc}")
4.2 Chroma 向量数据库
Chroma 是一个开源的轻量级向量数据库，专为 AI 应用设计，提供简单的 Python API 和持久化存储能力，非常适合中小型 RAG 项目。
pip install chromadb
from openai import OpenAI
import chromadb
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 创建 Chroma 客户端
chroma_client = chromadb.PersistentClient(path="./chroma_db")
# 创建集合
collection = chroma_client.get_or_create_collection(name="tech_docs")
# 准备文档
documents = [
"Python 3.12 引入了改进的错误消息，让调试更加高效",
"FastAPI 基于 Starlette 和 Pydantic，性能接近 Node.js 和 Go",
"Docker Compose 可以定义和运行多容器应用",
"Kubernetes 是 Google 开源的容器编排平台"
]
# 使用 OpenAI Embedding 添加到 Chroma
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction
embedding_function = OpenAIEmbeddingFunction(
api_key=os.getenv("OPENAI_API_KEY"),
model_name="text-embedding-3-small"
)
# 清空并重新添加
collection.delete(where={})
for i, doc in enumerate(documents):
collection.add(
documents=[doc],
ids=[f"doc_{i}"],
metadatas=[{"source": "tech_blog", "index": i}]
)
# 查询
results = collection.query(
query_texts=["Python 新版本有什么改进？"],
n_results=2,
include=["documents", "distances", "metadatas"]
)
for doc, dist, meta in zip(results["documents"][0], results["distances"][0], results["metadatas"][0]):
print(f"[距离: {dist:.4f}] {doc} (来源: {meta['source']})")
4.3 FAISS 高效相似度搜索
FAISS（Facebook AI Similarity Search）是 Meta 开发的高性能向量检索库，支持 GPU 加速和海量数据检索，适合对性能要求极高的生产环境。
pip install faiss-cpu # 或使用 faiss-gpu
import faiss
import numpy as np
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 获取 Embedding 的辅助函数
def get_embedding(text: str) -> np.ndarray:
resp = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return np.array(resp.data[0].embedding, dtype="float32")
# 文档库
documents = [
"Python 3.12 改进了错误消息和 f-string 性能",
"FastAPI 使用 Pydantic 进行数据验证",
"Django 是一个功能完备的 Python Web 框架",
"Flask 是轻量级 Web 框架，适合小型应用"
]
# 构建向量矩阵
dimension = 1536 # text-embedding-3-small 维度
embeddings = np.array([get_embedding(doc) for doc in documents], dtype="float32")
# 创建 FAISS 索引
index = faiss.IndexFlatIP(dimension) # 内积索引（余弦相似度需先归一化）
faiss.normalize_L2(embeddings) # L2 归一化
index.add(embeddings)
# 查询
query = "Python 新版本特性"
query_vec = get_embedding(query)
faiss.normalize_L2(query_vec.reshape(1, -1))
k = 2
distances, indices = index.search(query_vec.reshape(1, -1), k)
print(f"查询: {query}")
for i, (idx, dist) in enumerate(zip(indices[0], distances[0])):
print(f"#{i+1} [相似度: {dist:.4f}] {documents[idx]}")
5. 流式响应与 SSE
5.1 流式输出基础
流式响应（Streaming）让 LLM 的输出实时逐字返回，显著改善用户感知延迟，是聊天机器人和实时交互应用的标准实现方式。
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "写一首关于 AI 未来的短诗"}],
stream=True
)
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
print(content, end="", flush=True)
5.2 FastAPI SSE 实时推送
Server-Sent Events（SSE）是浏览器与服务器之间实现单向实时通信的标准协议。结合 FastAPI 和 OpenAI 流式 API，可以构建企业级实时 AI 问答服务。
pip install fastapi uvicorn
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
import os
app = FastAPI()
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
@app.get("/chat")
async def chat_stream(q: str):
async def event_generator():
stream = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个 helpful 助手"},
{"role": "user", "content": q}
],
stream=True
)
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
# SSE 格式: data: <内容>\n\n
yield f"data: {content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream"
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
启动服务后，使用 curl 测试：
curl -N "http://localhost:8000/chat?q=解释Python装饰器" \
-H "Accept: text/event-stream"
6. LangChain 框架集成
LangChain 是构建 LLM 应用最流行的 Python 框架，提供了模块化组件，包括模型接口、提示模板、文档加载器、向量存储和链式编排。
6.1 Chains 链式调用
Chain 是 LangChain 的核心概念，将多个组件按顺序连接成一个处理管道。RetrievalQA 链将文档检索和问答结合，是 RAG 的标准实现。
pip install langchain langchain-openai langchain-community chromadb
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os
# 初始化 LLM
llm = ChatOpenAI(
model="gpt-4o",
temperature=0.3,
api_key=os.getenv("OPENAI_API_KEY")
)
# 长文本文档
text = """Python 是一种解释型、高级、通用编程语言。Python 的设计哲学强调代码的可读性和简洁的语法，
其语言结构以及面向对象的方法旨在帮助程序员为小型和大型项目编写清晰、合乎逻辑的代码。
Python 由 Guido van Rossum 于 1991 年创建，目前由 Python 软件基金会维护。
Python 拥有庞大的标准库，支持多种编程范式，包括面向对象、函数式和结构化编程。"""
# 文本切分
splitter = RecursiveCharacterTextSplitter(
chunk_size=100,
chunk_overlap=20,
separators=["\n\n", "\n", "。", " "]
)
docs = splitter.create_documents([text])
# 向量化存储
embeddings = OpenAIEmbeddings(api_key=os.getenv("OPENAI_API_KEY"))
vectorstore = Chroma.from_documents(docs, embeddings)
# 构建 RAG 链
qa = RetrievalQA.from_chain_type(
llm=llm,
retriever=vectorstore.as_retriever(search_kwargs={"k": 2}),
chain_type="stuff", # 将检索到的文档直接拼接进提示
return_source_documents=True
)
result = qa.invoke({"query": "Python 由谁创建？它强调什么设计哲学？"})
print("答案:", result["result"])
print("\n引用文档:")
for i, doc in enumerate(result["source_documents"]):
print(f" [{i+1}] {doc.page_content[:80]}...")
6.2 Agents 智能代理
Agent 是 LangChain 的高级功能，赋予 LLM 自主决策能力，让它根据任务目标自动选择合适的工具并按序执行，直到完成目标或达到最大迭代次数。
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool
import os
llm = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))
# 定义工具
def multiply(a: str) -> str:
"""乘法计算，如 '3,4' → 12"""
try:
x, y = map(float, a.split(','))
return str(x * y)
except:
return "请提供两个数字，用逗号分隔"
def get_current_weather(city: str) -> str:
"""模拟天气查询"""
weather = {"北京": "晴天 25°C", "上海": "多云 22°C", "深圳": "小雨 28°C"}
return weather.get(city, "未知城市")
tools = [
Tool(name="multiply", func=multiply, description="乘法计算，输入格式：'数字1,数字2'"),
Tool(name="weather", func=get_current_weather, description="查询城市天气")
]
# 初始化 Agent
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.OPENAI_FUNCTIONS,
verbose=True, # 打印思考过程
max_iterations=3
)
result = agent.run("北京今天多少度？然后计算 25 乘以 4 是多少？")
print("最终结果:", result)
6.3 RAG with LangChain
以下展示一个完整的 LangChain RAG 流程，包括文档加载、切分、向量化、检索和生成，适合快速构建企业知识库问答系统。
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.text_splitter import CharacterTextSplitter
import os
llm = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))
# 模拟企业文档
raw_docs = [
"公司休假政策：员工每年享有 15 天带薪年假，工龄满 5 年增至 20 天。",
"技术栈规范：后端使用 Python + FastAPI，数据库使用 PostgreSQL。",
"代码评审流程：所有代码提交必须通过至少一名同事的 Review 才能合并。"
]
# 切分文档
text_splitter = CharacterTextSplitter(chunk_size=200, chunk_overlap=30)
documents = []
for text in raw_docs:
documents.extend(text_splitter.create_documents([text]))
# 构建向量存储
embeddings = OpenAIEmbeddings(api_key=os.getenv("OPENAI_API_KEY"))
vectordb = Chroma.from_documents(documents, embeddings)
# 对话记忆
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 构建对话式 RAG 链
qa_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=vectordb.as_retriever(search_kwargs={"k": 2}),
memory=memory,
verbose=True
)
# 多轮对话
q1 = "我们公司的年假有多少天？"
print(f"用户: {q1}")
print(f"AI: {qa_chain.invoke({'question': q1})['answer']}\n")
q2 = "工龄满 5 年呢？"
print(f"用户: {q2}")
print(f"AI: {qa_chain.invoke({'question': q2})['answer']}\n")
q3 = "我们后端用什么技术？"
print(f"用户: {q3}")
print(f"AI: {qa_chain.invoke({'question': q3})['answer']}\n")
7. 模型评估与监控
在生产环境中部署 LLM 应用后，需要建立持续的评估和监控机制，确保输出质量稳定并及时发现异常。
7.1 自动评估指标
使用 LLM-as-a-Judge 模式让能力更强的模型评估当前模型的输出质量，同时结合传统的 ROUGE 和 BLEU 指标进行量化评估。
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def evaluate_answer(question: str, answer: str, reference: str) -> dict:
"""使用 LLM 评估回答质量"""
eval_prompt = f"""你是一个严格的评估专家。请评估以下回答的质量。
问题: {question}
参考答案: {reference}
模型回答: {answer}
请从以下维度评分（1-5 分），并给出理由：
1. 准确性（回答是否正确）
2. 完整性（是否覆盖问题的所有方面）
3. 简洁性（是否简洁不冗余）
请以 JSON 格式输出: {{"accuracy": X, "completeness": Y, "conciseness": Z, "reason": "..."}}"""
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": eval_prompt}],
response_format={"type": "json_object"},
temperature=0.1
)
import json
return json.loads(response.choices[0].message.content)
# 测试
question = "Python 的 GIL 是什么？"
answer = "GIL 是全局解释器锁，它确保同一时间只有一个线程执行 Python 字节码。"
reference = "全局解释器锁（GIL）是 CPython 实现中的互斥锁，防止多线程同时执行字节码，简化了内存管理但限制了多核 CPU 利用率。"
result = evaluate_answer(question, answer, reference)
print(f"评估结果: {result}")
7.2 请求日志与成本监控
通过包装客户端调用，可以记录每次请求的 token 消耗、延迟和成本，便于进行用量分析和预算控制。
from openai import OpenAI
import time
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# 模型定价（每 1K tokens，单位：美元）
PRICING = {
"gpt-4o": {"input": 0.0025, "output": 0.010},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006}
}
class TrackedClient:
"""带监控的 OpenAI 客户端包装器"""
def __init__(self, client: OpenAI):
self.client = client
self.total_requests = 0
self.total_cost = 0.0
self.total_input_tokens = 0
self.total_output_tokens = 0
def chat(self, model: str, messages: list, **kwargs):
start = time.time()
response = self.client.chat.completions.create(
model=model,
messages=messages,
**kwargs
)
latency = time.time() - start
usage = response.usage
input_tokens = usage.prompt_tokens
output_tokens = usage.completion_tokens
price = PRICING.get(model, {"input": 0, "output": 0})
cost = (input_tokens / 1000 * price["input"] +
output_tokens / 1000 * price["output"])
self.total_requests += 1
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
self.total_cost += cost
print(f"[请求 #{self.total_requests}] 模型: {model}")
print(f" Token: 输入 {input_tokens} + 输出 {output_tokens} = {input_tokens + output_tokens}")
print(f" 成本: ${cost:.6f} | 延迟: {latency:.2f}s")
print(f" 累计成本: ${self.total_cost:.6f}")
return response
# 使用
tracked = TrackedClient(client)
resp = tracked.chat("gpt-4o-mini", [
{"role": "user", "content": "解释 REST API 是什么"}
])
print("\n回答:", resp.choices[0].message.content)
7.3 响应质量守护
通过设置输出长度限制、敏感词过滤和重试机制，构建基础的质量保障层，防止模型产生异常或有害内容。
from openai import OpenAI
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
def safe_chat(user_message: str, max_tokens: int = 500, max_retries: int = 2) -> dict:
"""带安全保护的聊天调用"""
# 简单敏感词检查
blocked_words = {"密码", "密钥", "api_key", "secret", "xxx"} # 实际应使用更完善的方案
lower_msg = user_message.lower()
for word in blocked_words:
if word in lower_msg and "环境变量" not in lower_msg:
return {"success": False, "error": "请求包含敏感内容", "content": None}
for attempt in range(max_retries):
try:
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "你是一个 helpful 的安全助手。"},
{"role": "user", "content": user_message}
],
max_tokens=max_tokens,
temperature=0.3
)
content = response.choices[0].message.content
# 输出长度检查
if len(content) > max_tokens * 4: # 粗略估算
return {"success": False, "error": "输出过长，可能异常", "content": None}
return {
"success": True,
"content": content,
"tokens": response.usage.total_tokens
}
except Exception as e:
if attempt == max_retries - 1:
return {"success": False, "error": str(e), "content": None}
return {"success": False, "error": "未知错误", "content": None}
# 测试
print(safe_chat("什么是面向对象编程？"))
延伸阅读
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。