OpenAI 落地实践

文章摘要 FakeGPT

加载中...|

此内容根据文章生成，并经过人工审核，仅用于文章内容的解释与总结投诉

概述

OpenAI 是目前最领先的 LLM 提供商之一，其 GPT 系列模型能力强大、生态完善。本文将深入介绍 OpenAI API 的使用技巧和最佳实践，帮助你在实际项目中高效应用 OpenAI 技术。

OpenAI 产品家族

模型系列

text

┌─────────────────────────────────────────────────────────┐
│                   OpenAI 模型家族                         │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  GPT-4o 系列                                             │
│  ├── GPT-4o: 最新旗舰，多模态，快速                       │
│  ├── GPT-4o-mini: 轻量版，低成本                         │
│  └── GPT-4o-2024-05-13: 旧版本                           │
│                                                         │
│  GPT-4 Turbo 系列                                        │
│  ├── GPT-4-turbo: 高性能，128K 上下文                    │
│  └── GPT-4-turbo-preview: 预览版                         │
│                                                         │
│  GPT-3.5 系列                                            │
│  ├── GPT-3.5-turbo: 性价比高                            │
│  └── GPT-3.5-turbo-instruct: 指令微调版本                │
│                                                         │
│  嵌入模型                                                │
│  ├── text-embedding-3-small: 快速，1536 维               │
│  ├── text-embedding-3-large: 高质量，3072 维             │
│  └── text-embedding-ada-002: 旧版本                      │
│                                                         │
│  图像模型                                                │
│  ├── DALL-E 3: 最强图像生成                              │
│  └── DALL-E 2: 基础版本                                  │
│                                                         │
│  语音模型                                                │
│  ├── Whisper-3: 语音转文字                               │
│  ├── TTS-1: 文字转语音                                   │
│  └── TTS-1-hd: 高质量 TTS                                │
│                                                         │
└─────────────────────────────────────────────────────────┘

模型选择建议

使用场景	推荐模型	理由
复杂推理	gpt-4o	最强推理能力
多模态任务	gpt-4o	原生视觉能力
高并发聊天	gpt-4o-mini	低成本，快响应
简单任务	gpt-3.5-turbo	性价比最高
代码生成	gpt-4o	代码能力强
长文本处理	gpt-4-turbo	128K 上下文
嵌入向量	text-embedding-3-small	快速便宜
高质量嵌入	text-embedding-3-large	更好精度

Chat Completions API

基础用法

python

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个Python专家。"},
        {"role": "user", "content": "用Python写一个快排算法"}
    ]
)

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

参数详解

python

response = client.chat.completions.create(
    # 模型选择
    model="gpt-4o",

    # 消息列表
    messages=[
        {"role": "system", "content": "系统提示词"},
        {"role": "user", "content": "用户消息"},
        {"role": "assistant", "content": "助手回复"},  # 可选：历史对话
        {"role": "user", "content": "新的问题"}
    ],

    # 采样参数
    temperature=0.7,      # 0-2，越高越随机
    top_p=0.9,            # 0-1，核采样
    n=1,                  # 生成数量

    # 输出控制
    max_tokens=1000,      # 最大输出 token 数
    presence_penalty=0,   # -2 到 2，鼓励新话题
    frequency_penalty=0,  # -2 到 2，减少重复

    # 其他选项
    stream=False,         # 是否流式输出
    stop=["\n", "END"],   # 停止序列
    seed=42,              # 随机种子（确定性输出）
    user="user-123"       # 用户标识（监控）
)

流式输出

python

# 流式输出
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "讲一个短故事"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        print(chunk.choices[0].delta.content, end="", flush=True)

# 收集完整响应
full_response = ""
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "讲一个短故事"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        full_response += chunk.choices[0].delta.content

JSON 模式

python

# 强制 JSON 输出
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "分析这句话的情感：'我太开心了！'"}
    ],
    response_format={"type": "json_object"}
)

result = json.loads(response.choices[0].message.content)
print(result)

结构化输出（最新功能）

python

from pydantic import BaseModel

class SentimentAnalysis(BaseModel):
    """情感分析结果"""
    sentiment: str  # positive/negative/neutral
    confidence: float  # 0-1
    keywords: list[str]  # 关键词

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "分析这句话的情感：'我太开心了！'"}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "sentiment_analysis",
            "strict": True,
            "schema": SentimentAnalysis.model_json_schema()
        }
    }
)

result = SentimentAnalysis.model_validate_json(
    response.choices[0].message.content
)
print(result)

Function Calling 最佳实践

定义高质量工具

python

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_product_info",
            "description": """获取产品信息，包括价格、库存、规格等。
            可以根据产品ID或产品名称搜索。""",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {
                        "type": "string",
                        "description": "产品ID，如：P12345"
                    },
                    "product_name": {
                        "type": "string",
                        "description": "产品名称，如：iPhone 15 Pro"
                    }
                },
                "oneOf": [  # 只需要提供其中一个
                    {"required": ["product_id"]},
                    {"required": ["product_name"]}
                ]
            }
        }
    }
]

处理工具调用

python

def process_tool_calls(response):
    """处理工具调用的完整流程"""
    messages = [{"role": "user", "content": "iPhone 15 Pro 多少钱？"}]

    # 第一次调用
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )

    messages.append(response.choices[0].message)

    # 处理工具调用
    tool_calls = response.choices[0].message.tool_calls

    for tool_call in tool_calls:
        function_name = tool_call.function.name
        function_args = json.loads(tool_call.function.arguments)

        # 执行函数
        if function_name == "get_product_info":
            result = get_product_info(**function_args)

            # 添加工具结果
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

    # 获取最终回复
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )

    return final_response.choices[0].message.content

并行函数调用

python

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "查北京天气、纽约时间、汇率"}],
    tools=[weather_tool, time_tool, currency_tool],
    parallel_tool_calls=True  # 允许并行调用
)

# 并行处理所有工具调用
tool_results = []
for tool_call in response.choices[0].message.tool_calls:
    result = execute_tool(tool_call)
    tool_results.append(result)

Assistant API

创建 Assistant

python

from openai import OpenAI

client = OpenAI()

# 创建 Assistant
assistant = client.beta.assistants.create(
    name="客服助手",
    instructions="你是一个专业的客服，使用提供的工具回答用户问题。",
    model="gpt-4o",
    tools=[
        {"type": "code_interpreter"},  # 代码解释器
        {
            "type": "function",
            "function": {
                "name": "get_order_status",
                "description": "查询订单状态",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {"type": "string"}
                    },
                    "required": ["order_id"]
                }
            }
        }
    ]
)

创建 Thread 和消息

python

# 创建 Thread
thread = client.beta.threads.create()

# 添加消息
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="我的订单12345状态是什么？"
)

运行 Assistant

python

# 创建 Run
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

# 等待完成
import time

while run.status in ["queued", "in_progress", "requires_action"]:
    time.sleep(1)
    run = client.beta.threads.runs.retrieve(
        thread_id=thread.id,
        run_id=run.id
    )

    # 处理需要调用工具的情况
    if run.status == "requires_action":
        tool_outputs = []
        for tool_call in run.required_action.submit_tool_outputs.tool_calls:
            if tool_call.function.name == "get_order_status":
                result = get_order_status(
                    **json.loads(tool_call.function.arguments)
                )
                tool_outputs.append({
                    "tool_call_id": tool_call.id,
                    "output": json.dumps(result)
                )

        # 提交工具输出
        run = client.beta.threads.runs.submit_tool_outputs(
            thread_id=thread.id,
            run_id=run.id,
            tool_outputs=tool_outputs
        )

# 获取结果
messages = client.beta.threads.messages.list(thread.id)
print(messages.data[0].content)

代码解释器

python

# 支持文件上传和代码分析的 Assistant
assistant = client.beta.assistants.create(
    name="数据分析助手",
    instructions="使用代码解释器分析上传的数据文件",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}]
)

# 上传文件
file = client.files.create(
    file=open("data.csv", "rb"),
    purpose="assistants"
)

# 创建 Thread 并附加文件
thread = client.beta.threads.create(
    messages=[{
        "role": "user",
        "content": "分析这个CSV文件的统计信息",
        "attachments": [{
            "file_id": file.id,
            "tools": [{"type": "code_interpreter"}]
        }]
    }]
)

# 运行
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

成本优化策略

Token 使用优化

python

# 1. 使用更小的模型处理简单任务
def classify_task(query: str) -> str:
    """先分类任务，再用合适的模型"""
    classification = client.chat.completions.create(
        model="gpt-3.5-turbo",  # 用小模型分类
        messages=[{
            "role": "user",
            "content": f"分类这个任务：{query}\nA.简单 B.中等 C.复杂"
        }]
    )

    result = classification.choices[0].message.content

    if "A" in result:
        return "gpt-3.5-turbo"
    elif "B" in result:
        return "gpt-4o-mini"
    else:
        return "gpt-4o"

# 2. 系统提示词优化
system_prompt = """
你是客服助手。规则：
1. 简洁回答，不超过50字
2. 只回答与产品相关的问题
3. 不知道就说不知道
"""  # 精简提示词节省 token

# 3. 使用 max_tokens 限制输出
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    max_tokens=150  # 限制输出长度
)

缓存策略

python

import hashlib
import json

class CachedOpenAI:
    def __init__(self, cache_file="cache.json"):
        self.cache_file = cache_file
        self.cache = self._load_cache()
        self.client = OpenAI()

    def _load_cache(self):
        try:
            with open(self.cache_file) as f:
                return json.load(f)
        except:
            return {}

    def _save_cache(self):
        with open(self.cache_file, 'w') as f:
            json.dump(self.cache, f)

    def _get_cache_key(self, model, messages):
        content = json.dumps({"model": model, "messages": messages})
        return hashlib.md5(content.encode()).hexdigest()

    def chat(self, model, messages, **kwargs):
        cache_key = self._get_cache_key(model, messages)

        if cache_key in self.cache:
            print("使用缓存")
            return self.cache[cache_key]

        response = self.client.chat.completions.create(
            model=model,
            messages=messages,
            **kwargs
        )

        self.cache[cache_key] = response
        self._save_cache()
        return response

# 使用
cached_client = CachedOpenAI()
response = cached_client.chat("gpt-4o", messages)

批量处理

python

# 批量 API（更便宜）
from openai import OpenAI

client = OpenAI()

# 创建批处理文件
batch_file = client.files.create(
    file=open("batch_requests.jsonl", "rb"),
    purpose="batch"
)

# 创建批处理
batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h"
)

# 检查状态
batch = client.batches.retrieve(batch.id)
print(f"状态: {batch.status}")
print(f"请求数: {batch.request_counts}")
print(f"总费用: ${batch.total_tokens:.2f}")

成本监控

python

# Token 计费
from tiktoken import encoding_for_model

def estimate_cost(model, messages):
    """估算 API 调用成本"""
    enc = encoding_for_model(model)

    # 计算 input tokens
    input_tokens = sum(
        len(enc.encode(msg.get("content", "")))
        for msg in messages
    )

    # 价格（美元/1K tokens）
    prices = {
        "gpt-4o": {"input": 0.005, "output": 0.015},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015}
    }

    model_prices = prices.get(model, prices["gpt-4o"])
    input_cost = (input_tokens / 1000) * model_prices["input"]

    # 假设输出与输入相同的 token 数
    output_cost = (input_tokens / 1000) * model_prices["output"]

    return {
        "input_tokens": input_tokens,
        "estimated_input_cost": input_cost,
        "estimated_total_cost": input_cost + output_cost
    }

cost = estimate_cost("gpt-4o", messages)
print(f"估算成本: ${cost['estimated_total_cost']:.4f}")

企业级部署

配置管理

python

# config.py
from pydantic import BaseModel
from typing import Optional

class OpenAIConfig(BaseModel):
    api_key: str
    model: str = "gpt-4o"
    temperature: float = 0.7
    max_tokens: int = 1000
    timeout: int = 30
    max_retries: int = 3

# 生产环境
class OpenAIClient:
    def __init__(self, config: OpenAIConfig):
        self.config = config
        self.client = OpenAI(
            api_key=config.api_key,
            timeout=config.timeout,
            max_retries=config.max_retries
        )

    def chat(self, messages, **kwargs):
        return self.client.chat.completions.create(
            model=self.config.model,
            messages=messages,
            temperature=kwargs.get("temperature", self.config.temperature),
            max_tokens=kwargs.get("max_tokens", self.config.max_tokens)
        )

# 使用
config = OpenAIConfig(
    api_key=os.getenv("OPENAI_API_KEY"),
    model="gpt-4o",
    temperature=0.5
)

client = OpenAIClient(config)

错误处理

python

from openai import OpenAI, APIError, RateLimitError, APITimeoutError

class RobustOpenAIClient:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)

    def chat_with_retry(self, model, messages, max_retries=3):
        """带重试的聊天"""
        for attempt in range(max_retries):
            try:
                return self.client.chat.completions.create(
                    model=model,
                    messages=messages
                )
            except RateLimitError as e:
                if attempt < max_retries - 1:
                    wait_time = 2 ** attempt  # 指数退避
                    print(f"速率限制，等待 {wait_time} 秒...")
                    time.sleep(wait_time)
                else:
                    raise
            except APITimeoutError as e:
                if attempt < max_retries - 1:
                    print("超时，重试...")
                    time.sleep(1)
                else:
                    raise
            except APIError as e:
                print(f"API 错误: {e}")
                raise

# 使用
client = RobustOpenAIClient(api_key="your-key")
response = client.chat_with_retry("gpt-4o", messages)

安全最佳实践

python

# 1. API Key 管理
import os
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载
api_key = os.getenv("OPENAI_API_KEY")

# 2. 内容审核
from openai import OpenAI

client = OpenAI()

def moderate_content(content: str) -> bool:
    """检查内容是否安全"""
    response = client.moderations.create(input=content)
    return response.results[0].flagged

# 3. 用户数据保护
def sanitize_input(text: str) -> str:
    """清理用户输入"""
    # 移除敏感信息（简化示例）
    import re

    # 移除身份证号
    text = re.sub(r'\d{15}|\d{18}', '[ID_REDACTED]', text)

    # 移除手机号
    text = re.sub(r'1[3-9]\d{9}', '[PHONE_REDACTED]', text)

    return text

# 4. 请求追踪
from contextvars import ContextVar

user_id_var = ContextVar("user_id", default=None)

def chat_with_tracking(messages):
    user_id = user_id_var.get()

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        user=str(user_id)  # OpenAI 用于监控
    )

    # 记录使用情况
    log_usage(user_id, response.usage)

    return response

小结

OpenAI 提供了完整的 AI 应用开发平台：

核心要点

模型选择
- GPT-4o：最强能力，多模态
- GPT-4o-mini：性价比高
- GPT-3.5-turbo：最经济
API 使用
- Chat Completions 是核心
- 流式输出提升体验
- 结构化输出保证格式
高级功能
- Function Calling 连接外部世界
- Assistant API 简化应用开发
- 代码解释器处理数据分析
成本优化
- 选择合适模型
- 使用缓存减少调用
- 批量处理降低单价
生产部署
- 完善的错误处理
- 内容审核和安全
- 请求追踪和监控

下一篇文章将介绍 Anthropic Claude 的落地实践。

OpenAI 落地实践https://indulgeback.github.io/posts/AI%E4%B8%8ELLM/10%E3%80%81OpenAI%20%E8%90%BD%E5%9C%B0%E5%AE%9E%E8%B7%B5

作者LeviLiu

发布于4天前

更新于2天前

许可协议 CC BY-NC-SA 4.0

署名-非商业性使用-相同方式共享 4.0 国际

LLM OpenAI GPT 实战

反馈与投诉

赞赏博主

评论隐私政策

概述 ​

OpenAI 产品家族 ​

模型系列 ​

模型选择建议 ​

Chat Completions API ​

基础用法 ​

参数详解 ​

流式输出 ​

JSON 模式 ​

结构化输出（最新功能） ​

Function Calling 最佳实践 ​

定义高质量工具 ​

处理工具调用 ​

并行函数调用 ​

Assistant API ​

创建 Assistant ​

创建 Thread 和消息 ​

运行 Assistant ​

代码解释器 ​

成本优化策略 ​

Token 使用优化 ​

缓存策略 ​

批量处理 ​

成本监控 ​

企业级部署 ​

配置管理 ​

错误处理 ​

安全最佳实践 ​

小结 ​

核心要点 ​

概述

OpenAI 产品家族

模型系列

模型选择建议

Chat Completions API

基础用法

参数详解

流式输出

JSON 模式

结构化输出（最新功能）

Function Calling 最佳实践

定义高质量工具

处理工具调用

并行函数调用

Assistant API

创建 Assistant

创建 Thread 和消息

运行 Assistant

代码解释器

成本优化策略

Token 使用优化

缓存策略

批量处理

成本监控

企业级部署

配置管理

错误处理

安全最佳实践

小结

核心要点