OpenAI API详解

概述

OpenAI API是目前最成熟、功能最丰富的大模型API之一。本章将详细介绍OpenAI的核心API接口，包括Chat Completions、Embeddings、Assistants API等，帮助你快速掌握OpenAI API的使用方法。

环境准备

安装SDK

pip install openai

配置API Key

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
    base_url="https://api.openai.com/v1"
)

环境变量配置：

export OPENAI_API_KEY="sk-xxxxxxxxxxxxx"

Chat Completions API

Chat Completions是OpenAI最核心的API，用于生成对话响应。

基础调用

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个专业的Python开发助手。"},
        {"role": "user", "content": "如何实现一个简单的装饰器？"}
    ]
)

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

参数详解

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "你是一个AI助手。"},
        {"role": "user", "content": "介绍一下机器学习。"}
    ],
    temperature=0.7,
    top_p=0.9,
    max_tokens=1000,
    presence_penalty=0.6,
    frequency_penalty=0.5,
    n=1,
    stream=False,
    stop=["\n\n", "END"]
)

参数说明：

参数	类型	说明
model	string	模型名称，如 gpt-4o、gpt-4o-mini
messages	array	对话消息列表
temperature	float	控制随机性，0-2，默认1
top_p	float	核采样，0-1，默认1
max_tokens	int	最大生成Token数
presence_penalty	float	存在惩罚，-2.0到2.0
frequency_penalty	float	频率惩罚，-2.0到2.0
n	int	生成几个响应
stream	bool	是否流式输出
stop	array	停止序列

流式输出

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)

多轮对话

messages = [
    {"role": "system", "content": "你是一个友好的AI助手。"}
]

while True:
    user_input = input("你: ")
    if user_input.lower() in ["exit", "quit", "q"]:
        break
    
    messages.append({"role": "user", "content": user_input})
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    
    assistant_message = response.choices[0].message.content
    messages.append({"role": "assistant", "content": assistant_message})
    
    print(f"AI: {assistant_message}\n")

视觉能力（GPT-4 Vision）

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "这张图片里有什么？"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/image.jpg"
                    }
                }
            ]
        }
    ],
    max_tokens=300
)

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

Base64图片：

import base64

with open("image.png", "rb") as image_file:
    base64_image = base64.b64encode(image_file.read()).decode('utf-8')

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "描述这张图片"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{base64_image}"
                    }
                }
            ]
        }
    ]
)

Function Calling

Function Calling允许模型调用外部函数，实现更强大的功能。

定义函数

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如：北京"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

调用流程

import json

def get_weather(city, unit="celsius"):
    weather_data = {
        "北京": {"temperature": 22, "condition": "晴"},
        "上海": {"temperature": 25, "condition": "多云"},
        "深圳": {"temperature": 28, "condition": "晴"}
    }
    
    if city in weather_data:
        data = weather_data[city]
        return json.dumps({
            "city": city,
            "temperature": data["temperature"],
            "unit": unit,
            "condition": data["condition"]
        })
    return json.dumps({"error": "未找到该城市"})

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "北京今天天气怎么样？"}
    ],
    tools=tools,
    tool_choice="auto"
)

message = response.choices[0].message

if message.tool_calls:
    tool_call = message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    
    if function_name == "get_weather":
        function_response = get_weather(
            city=function_args.get("city"),
            unit=function_args.get("unit", "celsius")
        )
        
        messages = [
            {"role": "user", "content": "北京今天天气怎么样？"},
            message,
            {
                "tool_call_id": tool_call.id,
                "role": "tool",
                "name": function_name,
                "content": function_response
            }
        ]
        
        second_response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages
        )
        
        print(second_response.choices[0].message.content)

并行调用

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_current_weather",
            "description": "获取当前天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_forecast",
            "description": "获取天气预报",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                    "days": {"type": "integer"}
                },
                "required": ["location"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "北京现在的天气和未来3天的预报是什么？"}
    ],
    tools=tools
)

for tool_call in response.choices[0].message.tool_calls:
    print(f"Function: {tool_call.function.name}")
    print(f"Arguments: {tool_call.function.arguments}")

Embeddings API

Embeddings用于将文本转换为向量表示，常用于语义搜索、聚类等场景。

基础使用

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="你好，世界！"
)

embedding = response.data[0].embedding
print(f"向量维度: {len(embedding)}")
print(f"前10个值: {embedding[:10]}")

批量处理

texts = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络",
    "自然语言处理处理文本数据"
]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts
)

for i, item in enumerate(response.data):
    print(f"文本 {i+1} 向量维度: {len(item.embedding)}")

相似度计算

import numpy as np

def cosine_similarity(vec1, vec2):
    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))

text1 = "我喜欢编程"
text2 = "我热爱写代码"
text3 = "今天天气很好"

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=[text1, text2, text3]
)

emb1 = response.data[0].embedding
emb2 = response.data[1].embedding
emb3 = response.data[2].embedding

print(f"'{text1}' vs '{text2}': {cosine_similarity(emb1, emb2):.4f}")
print(f"'{text1}' vs '{text3}': {cosine_similarity(emb1, emb3):.4f}")

语义搜索示例

documents = [
    "Python是一种高级编程语言",
    "JavaScript主要用于网页开发",
    "机器学习是AI的核心技术",
    "数据库用于存储和管理数据",
    "云计算提供按需计算服务"
]

query = "编程语言有哪些？"

doc_embeddings = client.embeddings.create(
    model="text-embedding-3-small",
    input=documents
)

query_embedding = client.embeddings.create(
    model="text-embedding-3-small",
    input=query
)

query_vec = query_embedding.data[0].embedding

similarities = []
for i, doc_emb in enumerate(doc_embeddings.data):
    sim = cosine_similarity(query_vec, doc_emb.embedding)
    similarities.append((i, sim, documents[i]))

similarities.sort(key=lambda x: x[1], reverse=True)

print("搜索结果：")
for i, sim, doc in similarities[:3]:
    print(f"相似度: {sim:.4f} - {doc}")

Assistants API

Assistants API提供了更高级的AI助手功能，支持持久化对话、文件处理等。

创建助手

assistant = client.beta.assistants.create(
    name="Python编程助手",
    instructions="你是一个专业的Python编程助手，帮助用户解决编程问题。",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}]
)

print(f"助手ID: {assistant.id}")

创建线程和消息

thread = client.beta.threads.create()

message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="如何实现快速排序算法？"
)

运行助手

run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id
)

if run.status == "completed":
    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)

文件上传

file = client.files.create(
    file=open("data.csv", "rb"),
    purpose='assistants'
)

assistant = client.beta.assistants.create(
    name="数据分析助手",
    instructions="你是一个数据分析专家，帮助用户分析CSV文件。",
    model="gpt-4o",
    tools=[{"type": "code_interpreter"}],
    tool_resources={
        "code_interpreter": {
            "file_ids": [file.id]
        }
    }
)

模型选择指南

模型	上下文窗口	特点	适用场景
gpt-4o	128K	多模态、速度快	通用场景、视觉任务
gpt-4o-mini	128K	成本低、速度快	简单任务、批量处理
gpt-4-turbo	128K	性能强大	复杂推理任务
gpt-3.5-turbo	16K	成本最低	简单对话

成本优化

Token计算

import tiktoken

def count_tokens(text, model="gpt-4o"):
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

text = "这是一段测试文本，用于计算Token数量。"
token_count = count_tokens(text)
print(f"Token数量: {token_count}")

成本估算

def estimate_cost(input_tokens, output_tokens, model="gpt-4o"):
    prices = {
        "gpt-4o": {"input": 2.5 / 1_000_000, "output": 10 / 1_000_000},
        "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.6 / 1_000_000},
        "gpt-3.5-turbo": {"input": 0.5 / 1_000_000, "output": 1.5 / 1_000_000}
    }
    
    price = prices.get(model, prices["gpt-4o"])
    cost = input_tokens * price["input"] + output_tokens * price["output"]
    return cost

input_tokens = 1000
output_tokens = 500
cost = estimate_cost(input_tokens, output_tokens, "gpt-4o")
print(f"预估成本: ${cost:.6f}")

最佳实践

1. 合理设置Temperature

creative_task = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "写一首诗"}],
    temperature=0.9
)

precise_task = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "计算123*456"}],
    temperature=0
)

2. 使用System Prompt

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": """你是一个专业的技术文档撰写助手。
            - 使用清晰简洁的语言
            - 提供代码示例
            - 包含必要的解释"""
        },
        {"role": "user", "content": "介绍Python的列表推导式"}
    ]
)

3. 控制输出格式

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "system",
            "content": "以JSON格式输出结果。"
        },
        {
            "role": "user",
            "content": "列出三种编程语言及其特点"
        }
    ],
    response_format={"type": "json_object"}
)

import json
result = json.loads(response.choices[0].message.content)
print(json.dumps(result, indent=2, ensure_ascii=False))

小结

本章详细介绍了OpenAI API的核心功能：

1. Chat Completions - 最常用的对话API，支持多轮对话和流式输出
2. Function Calling - 实现工具调用，扩展AI能力
3. Embeddings - 文本向量化，用于语义搜索和相似度计算
4. Assistants API - 高级助手功能，支持文件处理和持久化

掌握这些API后，你可以构建各种AI应用，从简单的聊天机器人到复杂的智能助手系统。下一章将介绍Claude API的使用方法。

参考资源

• OpenAI官方文档
• OpenAI API参考
• OpenAI Cookbook