写在前面

你有没有想过,为什么OpenAI的SDK能连接那么多不同的AI模型?是不是觉得每个模型都要学一套新的API很麻烦?别担心,这篇教程就是为你准备的!

我们会用最通俗的语言,把那些看起来高大上的概念掰开揉碎了讲给你听。同时,我们也会深入到底层原理,让你真正理解"为什么是这样"。


第一部分:核心概念——这些是你必须知道的

1.1 API Key:你的"通行证"

大白话解释: 想象API Key就像是一张门禁卡。每个AI服务商都有自己的大门,你想进去用他们的AI,就得有这张卡。没有卡?对不起,门都不给你开。

深入理解: API Key本质上是一个加密字符串,用于身份验证和授权。它通常包含两部分:

  • 前缀标识:如sk-表示OpenAI,ak-表示阿里云等
  • 加密密钥:一串随机字符,确保唯一性和安全性

安全提醒:

# ❌ 错误做法:直接写在代码里
api_key = "sk-xxxxxxxxxxxxx"

# ✅ 正确做法:从环境变量读取
import os
api_key = os.getenv("OPENAI_API_KEY")

1.2 Base URL:模型的"门牌号"

大白话解释: Base URL就是AI服务商的地址。就像你要去不同的商店,每个商店都有自己的地址。OpenAI有OpenAI的地址,阿里云有阿里云的地址。

深入理解: Base URL是一个HTTP端点(endpoint),它决定了你的请求会被发送到哪里。标准的OpenAI格式是:

https://api.openai.com/v1

但当你使用其他服务商时,这个地址会变:

  • Azure OpenAI: https://your-resource.openai.azure.com/openai/deployments/your-deployment
  • 通义千问: https://dashscope.aliyuncs.com/compatible-mode/v1
  • 智谱GLM: https://open.bigmodel.cn/api/paas/v4
  • DeepSeek: https://api.deepseek.com/v1

为什么这很重要? 因为OpenAI的SDK设计得很聪明,它允许你自定义Base URL。这意味着,只要你把Base URL改成其他服务商的地址,同时那个服务商又兼容OpenAI的接口格式,你就可以用同一套代码连接不同的模型!

1.3 模型名称:你要调用的"角色"

大白话解释: 每个AI服务商都有很多模型,就像一家餐厅有很多菜品。你得告诉服务员你要点什么菜,这就是模型名称的作用。

深入理解: 模型名称通常包含以下信息:

  • 模型类型:如gpt-4gpt-3.5-turbo
  • 版本信息:如gpt-4-0125-preview
  • 能力标识:如gpt-4-vision-preview表示支持视觉

不同服务商的命名规则不同:

# OpenAI
model = "gpt-4-turbo"

# 通义千问
model = "qwen-turbo"

# 智谱GLM
model = "glm-4"

# DeepSeek
model = "deepseek-chat"

1.4 请求和响应:你和AI的"对话流程"

大白话解释: 你发一个问题,AI回一个答案,这就是一次请求和响应。但在这个过程中,有很多细节需要注意。

深入理解:

请求结构:

{
    "model": "gpt-4",              # 用哪个模型
    "messages": [                   # 对话历史
        {"role": "system", "content": "你是一个助手"},
        {"role": "user", "content": "你好"}
    ],
    "temperature": 0.7,            # 创造性程度
    "max_tokens": 1000,            # 最大回复长度
    "stream": False                # 是否流式输出
}

响应结构:

{
    "id": "chatcmpl-xxxxx",
    "object": "chat.completion",
    "created": 1234567890,
    "model": "gpt-4",
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": "你好!有什么我可以帮助你的吗?"
            },
            "finish_reason": "stop"
        }
    ],
    "usage": {
        "prompt_tokens": 10,
        "completion_tokens": 15,
        "total_tokens": 25
    }
}

1.5 Token:AI世界的"货币"

大白话解释: Token可以理解为AI处理文本的基本单位。一个英文单词大约是1个token,一个汉字大约是2个token。你用AI,就是按token数量付费的。

深入理解: Token不仅仅是计费单位,它还关系到:

  • 上下文窗口:模型能记住多少内容
  • 输入限制:你一次能发多少内容
  • 输出限制:模型一次能回多少内容

不同模型的Token限制: | 模型 | 输入上限 | 输出上限 | |------|---------|---------| | GPT-3.5-Turbo | 16K tokens | 4K tokens | | GPT-4-Turbo | 128K tokens | 4K tokens | | GPT-4o | 128K tokens | 16K tokens | | Claude 3 Opus | 200K tokens | 4K tokens | | 通义千问-Max | 32K tokens | 8K tokens |

1.6 流式输出:像"打字机"一样逐字显示

大白话解释: 普通输出是等AI全部想好了再一次性给你,流式输出是AI想一个字给你一个字,就像打字机一样。

深入理解: 流式输出使用Server-Sent Events (SSE)技术,通过HTTP长连接实现实时数据推送。它的优势是:

  • 用户体验更好:不用等待,立即看到反馈
  • 降低超时风险:长时间生成不会因为超时而失败
  • 便于监控:可以实时看到生成进度

第二部分:使用方式的差异——不同服务商怎么连

2.1 OpenAI官方:标准范例

OpenAI是行业的标杆,它的API设计被广泛采用。

基本连接:

from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxxxxxxxxxxx"
)

response = client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[
        {"role": "user", "content": "介绍一下你自己"}
    ]
)

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

流式输出:

stream = client.chat.completions.create(
    model="gpt-4-turbo",
    messages=[
        {"role": "user", "content": "写一首诗"}
    ],
    stream=True
)

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

2.2 Azure OpenAI:企业级方案

微软Azure提供的OpenAI服务,更适合企业使用,但配置稍有不同。

关键差异:

  • 需要指定部署名称(deployment name)
  • 认证方式支持API Key和Azure AD
  • Base URL格式不同

连接方式:

from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint="https://your-resource.openai.azure.com",
    api_key="your-azure-key",
    api_version="2024-02-15-preview"
)

response = client.chat.completions.create(
    model="gpt-4-deployment",  # 这是你的部署名称
    messages=[
        {"role": "user", "content": "你好"}
    ]
)

2.3 国内大模型:兼容OpenAI接口

很多国内厂商为了降低迁移成本,都提供了兼容OpenAI的接口。你只需要改Base URL和API Key就行!

2.3.1 通义千问(阿里云)
from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxxxxxxxxxxx",  # 阿里云的API Key
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

response = client.chat.completions.create(
    model="qwen-turbo",
    messages=[
        {"role": "user", "content": "介绍一下通义千问"}
    ]
)
2.3.2 智谱GLM
from openai import OpenAI

client = OpenAI(
    api_key="xxxxxxxxxxxxx.xxxxxxxxxxxxx",  # GLM的API Key格式
    base_url="https://open.bigmodel.cn/api/paas/v4"
)

response = client.chat.completions.create(
    model="glm-4",
    messages=[
        {"role": "user", "content": "介绍一下智谱GLM"}
    ]
)
2.3.3 DeepSeek
from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxxxxxxxxxxx",
    base_url="https://api.deepseek.com/v1"
)

response = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "user", "content": "介绍一下DeepSeek"}
    ]
)
2.3.4 Moonshot(月之暗面)
from openai import OpenAI

client = OpenAI(
    api_key="sk-xxxxxxxxxxxxx",
    base_url="https://api.moonshot.cn/v1"
)

response = client.chat.completions.create(
    model="moonshot-v1-8k",
    messages=[
        {"role": "user", "content": "介绍一下Kimi"}
    ]
)

2.4 统一封装:一个类搞定所有模型

为了更方便地切换不同模型,我们可以封装一个统一的类:

from openai import OpenAI
from typing import Optional, Dict, List
import os

class UniversalLLM:
    """
    统一的大模型调用接口
    支持OpenAI、Azure、以及兼容OpenAI接口的国内模型
    """

    PROVIDERS = {
        "openai": {
            "base_url": "https://api.openai.com/v1",
            "default_model": "gpt-4-turbo"
        },
        "qwen": {
            "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
            "default_model": "qwen-turbo"
        },
        "glm": {
            "base_url": "https://open.bigmodel.cn/api/paas/v4",
            "default_model": "glm-4"
        },
        "deepseek": {
            "base_url": "https://api.deepseek.com/v1",
            "default_model": "deepseek-chat"
        },
        "moonshot": {
            "base_url": "https://api.moonshot.cn/v1",
            "default_model": "moonshot-v1-8k"
        }
    }

    def __init__(
        self,
        provider: str = "openai",
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        model: Optional[str] = None
    ):
        """
        初始化LLM客户端

        Args:
            provider: 服务商名称(openai/qwen/glm/deepseek/moonshot)
            api_key: API密钥,如果不提供会从环境变量读取
            base_url: 自定义Base URL,如果不提供使用默认配置
            model: 模型名称,如果不提供使用默认模型
        """
        if provider not in self.PROVIDERS:
            raise ValueError(f"不支持的provider: {provider}")

        self.provider = provider
        self.config = self.PROVIDERS[provider]

        self.api_key = api_key or os.getenv(f"{provider.upper()}_API_KEY")
        if not self.api_key:
            raise ValueError(f"请提供{provider}的API Key")

        self.base_url = base_url or self.config["base_url"]
        self.model = model or self.config["default_model"]

        self.client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )

    def chat(
        self,
        message: str,
        system_prompt: Optional[str] = None,
        history: Optional[List[Dict]] = None,
        stream: bool = False,
        **kwargs
    ) -> str:
        """
        发起对话

        Args:
            message: 用户消息
            system_prompt: 系统提示词
            history: 历史对话记录
            stream: 是否流式输出
            **kwargs: 其他参数(temperature、max_tokens等)

        Returns:
            模型的回复内容
        """
        messages = []

        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})

        if history:
            messages.extend(history)

        messages.append({"role": "user", "content": message})

        if stream:
            return self._stream_chat(messages, **kwargs)
        else:
            return self._sync_chat(messages, **kwargs)

    def _sync_chat(self, messages: List[Dict], **kwargs) -> str:
        """同步对话"""
        response = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            **kwargs
        )
        return response.choices[0].message.content

    def _stream_chat(self, messages: List[Dict], **kwargs):
        """流式对话"""
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            stream=True,
            **kwargs
        )

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

        print()
        return full_content

llm = UniversalLLM(provider="qwen", api_key="your-api-key")
response = llm.chat("你好,请介绍一下你自己", stream=True)

第三部分:不同模态大模型的连接——文本、图像、语音

3.1 文本模型:最基础也最重要

文本模型是最常用的,我们已经在前面的例子中展示了基本用法。这里补充一些高级技巧。

3.1.1 多轮对话管理
class ChatSession:
    """多轮对话管理器"""

    def __init__(self, llm: UniversalLLM, system_prompt: str = ""):
        self.llm = llm
        self.system_prompt = system_prompt
        self.history = []

    def send(self, message: str) -> str:
        """发送消息并保持历史记录"""
        response = self.llm.chat(
            message=message,
            system_prompt=self.system_prompt,
            history=self.history
        )

        self.history.append({"role": "user", "content": message})
        self.history.append({"role": "assistant", "content": response})

        return response

    def clear(self):
        """清空历史记录"""
        self.history = []

session = ChatSession(llm, system_prompt="你是一个友好的AI助手")
print(session.send("你好"))
print(session.send("我刚才说了什么?"))
3.1.2 Function Calling(函数调用)

这是让AI能够执行具体操作的能力,非常强大!

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称,如:北京"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

def get_weather(city: str) -> dict:
    """模拟天气查询函数"""
    weather_data = {
        "北京": {"temperature": "25°C", "weather": "晴"},
        "上海": {"temperature": "28°C", "weather": "多云"},
        "广州": {"temperature": "32°C", "weather": "雷阵雨"}
    }
    return weather_data.get(city, {"error": "未找到该城市"})

response = llm.client.chat.completions.create(
    model=llm.model,
    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":
        result = get_weather(function_args["city"])
        print(f"天气查询结果:{result}")

3.2 图像模型:让AI学会"画画"

图像模型主要有两类:

  • 图像生成:根据文本描述生成图像(如DALL-E、Stable Diffusion)
  • 图像理解:理解图像内容并回答问题(如GPT-4V、Gemini Vision)
3.2.1 图像生成(DALL-E)
def generate_image(
    prompt: str,
    model: str = "dall-e-3",
    size: str = "1024x1024",
    quality: str = "standard"
):
    """
    使用DALL-E生成图像

    Args:
        prompt: 图像描述
        model: 模型名称(dall-e-2 或 dall-e-3)
        size: 图像尺寸
        quality: 图像质量(standard 或 hd)
    """
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    response = client.images.generate(
        model=model,
        prompt=prompt,
        size=size,
        quality=quality,
        n=1
    )

    image_url = response.data[0].url
    print(f"生成的图像地址:{image_url}")
    return image_url

image_url = generate_image(
    prompt="一只穿着宇航服的猫咪在月球上漫步,背景是地球",
    quality="hd"
)
3.2.2 图像理解(GPT-4V)
import base64

def analyze_image(
    image_path: str,
    question: str,
    model: str = "gpt-4-vision-preview"
):
    """
    使用GPT-4V分析图像

    Args:
        image_path: 图像文件路径
        question: 要问的问题
        model: 模型名称
    """
    with open(image_path, "rb") as f:
        image_data = base64.b64encode(f.read()).decode("utf-8")

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    response = client.chat.completions.create(
        model=model,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": question
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{image_data}"
                        }
                    }
                ]
            }
        ],
        max_tokens=1000
    )

    return response.choices[0].message.content

result = analyze_image(
    image_path="photo.jpg",
    question="这张图片里有什么?请详细描述。"
)
print(result)
3.2.3 使用图像URL进行分析
def analyze_image_url(image_url: str, question: str):
    """使用图像URL进行分析"""
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    response = client.chat.completions.create(
        model="gpt-4-vision-preview",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": question},
                    {
                        "type": "image_url",
                        "image_url": {"url": image_url}
                    }
                ]
            }
        ]
    )

    return response.choices[0].message.content

result = analyze_image_url(
    image_url="https://example.com/image.jpg",
    question="描述这张图片的内容"
)

3.3 语音模型:让AI"听"和"说"

语音模型包括:

  • 语音转文字(STT):如Whisper
  • 文字转语音(TTS):如OpenAI TTS
3.3.1 语音转文字(Whisper)
def transcribe_audio(
    audio_path: str,
    model: str = "whisper-1",
    language: str = "zh"
):
    """
    使用Whisper转录音频

    Args:
        audio_path: 音频文件路径
        model: 模型名称
        language: 音频语言(zh表示中文)
    """
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    with open(audio_path, "rb") as audio_file:
        transcript = client.audio.transcriptions.create(
            model=model,
            file=audio_file,
            language=language
        )

    return transcript.text

text = transcribe_audio("recording.mp3")
print(f"转录结果:{text}")
3.3.2 文字转语音(TTS)
def text_to_speech(
    text: str,
    output_path: str = "output.mp3",
    model: str = "tts-1",
    voice: str = "alloy"
):
    """
    将文字转换为语音

    Args:
        text: 要转换的文字
        output_path: 输出文件路径
        model: 模型名称(tts-1 或 tts-1-hd)
        voice: 声音类型(alloy/echo/fable/onyx/nova/shimmer)
    """
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    response = client.audio.speech.create(
        model=model,
        voice=voice,
        input=text
    )

    with open(output_path, "wb") as f:
        f.write(response.content)

    print(f"语音已保存到:{output_path}")

text_to_speech(
    text="你好,这是一个文字转语音的示例。",
    voice="nova"
)

3.4 多模态模型:文本+图像的融合

最新的模型如GPT-4o、Gemini等支持多模态输入,可以同时处理文本和图像。

def multimodal_chat(
    text: str,
    image_paths: list = None,
    image_urls: list = None,
    model: str = "gpt-4o"
):
    """
    多模态对话

    Args:
        text: 文本内容
        image_paths: 本地图像路径列表
        image_urls: 图像URL列表
        model: 模型名称
    """
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    content = [{"type": "text", "text": text}]

    if image_paths:
        for path in image_paths:
            with open(path, "rb") as f:
                image_data = base64.b64encode(f.read()).decode("utf-8")
            content.append({
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{image_data}"
                }
            })

    if image_urls:
        for url in image_urls:
            content.append({
                "type": "image_url",
                "image_url": {"url": url}
            })

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": content}]
    )

    return response.choices[0].message.content

result = multimodal_chat(
    text="比较这两张图片的不同之处",
    image_paths=["image1.jpg", "image2.jpg"]
)
print(result)

第四部分:高级技巧与最佳实践

4.1 错误处理和重试机制

调用API时,网络问题、限流等都可能导致失败,需要做好错误处理。

import time
from openai import APIError, RateLimitError, APIConnectionError

def robust_chat(
    client: OpenAI,
    model: str,
    messages: list,
    max_retries: int = 3,
    retry_delay: float = 1.0
):
    """
    带重试机制的对话函数
    """
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages
            )
            return response.choices[0].message.content

        except RateLimitError:
            print(f"触发限流,等待{retry_delay}秒后重试...")
            time.sleep(retry_delay)
            retry_delay *= 2

        except APIConnectionError:
            print(f"网络连接失败,尝试第{attempt + 1}次重试...")
            time.sleep(retry_delay)

        except APIError as e:
            print(f"API错误:{e}")
            if attempt == max_retries - 1:
                raise

    raise Exception("超过最大重试次数")

4.2 成本控制

Token消耗直接关系到成本,需要合理控制。

def estimate_tokens(text: str) -> int:
    """粗略估算token数量"""
    chinese_chars = sum(1 for char in text if '\u4e00' <= char <= '\u9fff')
    english_words = len(text.split())
    return chinese_chars * 2 + english_words

def cost_aware_chat(
    client: OpenAI,
    model: str,
    messages: list,
    max_cost_usd: float = 0.1
):
    """
    带成本控制的对话函数
    """
    total_tokens = sum(estimate_tokens(msg["content"]) for msg in messages)

    price_per_1k_tokens = {
        "gpt-4-turbo": 0.01,
        "gpt-3.5-turbo": 0.0005,
        "gpt-4o": 0.005
    }

    estimated_cost = (total_tokens / 1000) * price_per_1k_tokens.get(model, 0.01)

    if estimated_cost > max_cost_usd:
        raise ValueError(f"预估成本${estimated_cost:.4f}超过限制${max_cost_usd}")

    print(f"预估成本:${estimated_cost:.4f}")

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

    actual_tokens = response.usage.total_tokens
    actual_cost = (actual_tokens / 1000) * price_per_1k_tokens.get(model, 0.01)
    print(f"实际成本:${actual_cost:.4f}")

    return response.choices[0].message.content

4.3 异步调用

对于批量请求,异步调用可以大幅提升效率。

import asyncio
from openai import AsyncOpenAI

async def async_chat(client: AsyncOpenAI, model: str, messages: list):
    """异步对话"""
    response = await client.chat.completions.create(
        model=model,
        messages=messages
    )
    return response.choices[0].message.content

async def batch_chat(prompts: list, model: str = "gpt-3.5-turbo"):
    """批量异步对话"""
    client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    tasks = [
        async_chat(
            client,
            model,
            [{"role": "user", "content": prompt}]
        )
        for prompt in prompts
    ]

    results = await asyncio.gather(*tasks)
    return results

async def main():
    prompts = [
        "什么是人工智能?",
        "什么是机器学习?",
        "什么是深度学习?"
    ]

    results = await batch_chat(prompts)
    for prompt, result in zip(prompts, results):
        print(f"问题:{prompt}")
        print(f"回答:{result}\n")

asyncio.run(main())

4.4 使用LangChain简化开发

LangChain是一个流行的框架,可以简化LLM应用开发。

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

def langchain_example():
    """使用LangChain调用模型"""
    llm = ChatOpenAI(
        model="gpt-4-turbo",
        temperature=0.7
    )

    messages = [
        SystemMessage(content="你是一个专业的Python编程导师"),
        HumanMessage(content="如何学习Python?")
    ]

    response = llm.invoke(messages)
    print(response.content)

langchain_example()

第五部分:常见问题与解决方案

Q1: 如何选择合适的模型?

A: 根据任务需求选择:

  • 简单任务:GPT-3.5-Turbo(快速、便宜)
  • 复杂推理:GPT-4-Turbo(能力强)
  • 多模态任务:GPT-4o、GPT-4V
  • 长文本处理:Claude 3(200K上下文)
  • 中文场景:通义千问、智谱GLM

Q2: 如何处理超长文本?

A: 几种方案:

  1. 分段处理:将长文本分成多个部分
  2. 摘要压缩:先用模型生成摘要,再处理
  3. 向量检索:使用RAG技术,只检索相关部分
def chunk_text(text: str, chunk_size: int = 4000):
    """将长文本分段"""
    chunks = []
    for i in range(0, len(text), chunk_size):
        chunks.append(text[i:i + chunk_size])
    return chunks

def process_long_text(client: OpenAI, text: str, task: str):
    """处理长文本"""
    chunks = chunk_text(text)
    results = []

    for chunk in chunks:
        response = client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "user", "content": f"{task}\n\n文本:{chunk}"}
            ]
        )
        results.append(response.choices[0].message.content)

    return "\n\n".join(results)

Q3: 如何提高响应速度?

A: 几个技巧:

  1. 使用流式输出:用户能更快看到反馈
  2. 选择更快的模型:GPT-3.5比GPT-4快
  3. 减少输出长度:设置max_tokens
  4. 使用缓存:相同问题不重复请求
import hashlib

class ResponseCache:
    """简单的响应缓存"""

    def __init__(self):
        self.cache = {}

    def _get_key(self, messages: list) -> str:
        content = "".join([msg["content"] for msg in messages])
        return hashlib.md5(content.encode()).hexdigest()

    def get(self, messages: list):
        key = self._get_key(messages)
        return self.cache.get(key)

    def set(self, messages: list, response: str):
        key = self._get_key(messages)
        self.cache[key] = response

cache = ResponseCache()

def cached_chat(client: OpenAI, model: str, messages: list):
    """带缓存的对话"""
    cached = cache.get(messages)
    if cached:
        print("使用缓存结果")
        return cached

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

    result = response.choices[0].message.content
    cache.set(messages, result)
    return result

Q4: 如何保护API Key安全?

A: 最佳实践:

  1. 使用环境变量:不要硬编码
  2. 使用密钥管理服务:如AWS Secrets Manager
  3. 定期轮换:定期更换API Key
  4. 最小权限:只给必要的权限
import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")

if not api_key:
    raise ValueError("请在.env文件中设置OPENAI_API_KEY")

第六部分:总结与展望

核心要点回顾

  1. 统一接口是关键:OpenAI的API设计已成为行业标准,掌握它就能连接大多数模型
  2. 理解底层概念:API Key、Base URL、Token等概念是基础中的基础
  3. 多模态是趋势:文本、图像、语音的融合将越来越重要
  4. 成本和性能要平衡:根据实际需求选择合适的模型

未来发展方向

  1. 更强的多模态能力:视频理解、3D生成等
  2. 更长的上下文:无限上下文窗口
  3. 更低的成本:模型优化带来价格下降
  4. 更快的速度:实时交互成为可能

学习建议

  1. 动手实践:光看不练假把式,多写代码
  2. 关注官方文档:API更新很快,要保持学习
  3. 加入社区:和其他开发者交流经验
  4. 关注前沿:AI领域发展迅速,要保持敏感

附录:完整代码示例

完整的多模态AI助手

import os
import base64
from openai import OpenAI
from typing import Optional, List, Dict
import json

class AIAssistant:
    """
    多模态AI助手
    支持文本对话、图像理解、语音处理
    """

    def __init__(
        self,
        provider: str = "openai",
        api_key: Optional[str] = None,
        model: str = "gpt-4o"
    ):
        providers = {
            "openai": "https://api.openai.com/v1",
            "qwen": "https://dashscope.aliyuncs.com/compatible-mode/v1",
            "glm": "https://open.bigmodel.cn/api/paas/v4",
            "deepseek": "https://api.deepseek.com/v1"
        }

        self.api_key = api_key or os.getenv(f"{provider.upper()}_API_KEY")
        self.base_url = providers.get(provider)
        self.model = model

        self.client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )

        self.conversation_history = []

    def chat(
        self,
        message: str,
        system_prompt: Optional[str] = None,
        stream: bool = False
    ) -> str:
        """文本对话"""
        if system_prompt and not self.conversation_history:
            self.conversation_history.append({
                "role": "system",
                "content": system_prompt
            })

        self.conversation_history.append({
            "role": "user",
            "content": message
        })

        if stream:
            return self._stream_response()
        else:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.conversation_history
            )

            assistant_message = response.choices[0].message.content
            self.conversation_history.append({
                "role": "assistant",
                "content": assistant_message
            })

            return assistant_message

    def _stream_response(self):
        """流式响应"""
        stream = self.client.chat.completions.create(
            model=self.model,
            messages=self.conversation_history,
            stream=True
        )

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

        print()

        self.conversation_history.append({
            "role": "assistant",
            "content": full_response
        })

        return full_response

    def analyze_image(
        self,
        image_path: Optional[str] = None,
        image_url: Optional[str] = None,
        question: str = "描述这张图片"
    ) -> str:
        """图像理解"""
        content = [{"type": "text", "text": question}]

        if image_path:
            with open(image_path, "rb") as f:
                image_data = base64.b64encode(f.read()).decode("utf-8")
            content.append({
                "type": "image_url",
                "image_url": {
                    "url": f"data:image/jpeg;base64,{image_data}"
                }
            })
        elif image_url:
            content.append({
                "type": "image_url",
                "image_url": {"url": image_url}
            })

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": content}]
        )

        return response.choices[0].message.content

    def generate_image(
        self,
        prompt: str,
        size: str = "1024x1024"
    ) -> str:
        """图像生成"""
        response = self.client.images.generate(
            model="dall-e-3",
            prompt=prompt,
            size=size,
            n=1
        )

        return response.data[0].url

    def transcribe_audio(
        self,
        audio_path: str,
        language: str = "zh"
    ) -> str:
        """语音转文字"""
        with open(audio_path, "rb") as audio_file:
            transcript = self.client.audio.transcriptions.create(
                model="whisper-1",
                file=audio_file,
                language=language
            )

        return transcript.text

    def text_to_speech(
        self,
        text: str,
        output_path: str = "output.mp3",
        voice: str = "alloy"
    ):
        """文字转语音"""
        response = self.client.audio.speech.create(
            model="tts-1",
            voice=voice,
            input=text
        )

        with open(output_path, "wb") as f:
            f.write(response.content)

        return output_path

    def clear_history(self):
        """清空对话历史"""
        self.conversation_history = []

    def export_history(self) -> str:
        """导出对话历史"""
        return json.dumps(self.conversation_history, ensure_ascii=False, indent=2)

if __name__ == "__main__":
    assistant = AIAssistant(provider="openai", model="gpt-4o")

    print("=== 文本对话测试 ===")
    response = assistant.chat(
        "你好,请介绍一下你自己",
        system_prompt="你是一个友好的AI助手",
        stream=True
    )

    print("\n=== 图像理解测试 ===")
    image_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/dd/Gfp-wisconsin-madison-the-nature-boardwalk.jpg/2560px-Gfp-wisconsin-madison-the-nature-boardwalk.jpg"
    description = assistant.analyze_image(
        image_url=image_url,
        question="详细描述这张图片的景色"
    )
    print(f"图片描述:{description}")

    print("\n=== 对话历史 ===")
    print(assistant.export_history())

结语

恭喜你读完了这篇教程!现在你应该已经掌握了:

✅ OpenAI API的核心概念 ✅ 如何连接不同的AI服务商 ✅ 如何处理文本、图像、语音等多种模态 ✅ 实际开发中的最佳实践

记住,学习AI开发最重要的是动手实践。不要只是看,要自己写代码、调试、解决问题。只有这样,你才能真正掌握这些技能。

如果你在学习过程中遇到问题,可以:

  • 查阅官方文档
  • 在GitHub上找开源项目学习
  • 加入开发者社区交流

祝你学习愉快!🚀


参考资源:

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐