Python OpenAI

openai ist eine leistungsstarke Python-Bibliothek zur Interaktion mit der Suite von Modellen und Diensten von OpenAI.

openai kapselt alle RESTful-API-Aufrufe und ermöglicht Entwicklern, leistungsstarke KI-Funktionen – wie Verarbeitung natürlicher Sprache, Bildgenerierung und Spracherkennung – einfach in ihre Python-Anwendungen zu integrieren.

Hauptmerkmale:

  • Textgenerierung: Verwenden Sie Modelle wie GPT-4 oder GPT-5, um Artikel, Code, Zusammenfassungen, Gespräche und mehr zu generieren.
  • Bildgenerierung: Erstellen Sie Bilder aus Textbeschreibungen mit DALL-E-Modellen.
  • Embeddings: Konvertieren Sie Text in Vektordarstellungen, die häufig für Aufgaben wie semantische Suche, Textklassifikation und Clustering verwendet werden.
  • Sprache-zu-Text: Verwenden Sie das Whisper-Modell, um Audiodateien in Text zu transkribieren.
  • Fine-Tuning: Trainiere ein zielgerichteteres Modell, indem du deinen eigenen Datensatz bereitstellst.
  • Assistants API: Erstelle komplexe Anwendungen, die Kontext verstehen, Tools aufrufen und langfristige Interaktionen ausführen können.

openai Open-Source-Repository:https://github.com/openai/openai-python


Wie verwendet man es?

Requirements:

  • Python-Version: 3.9 oder höher.
  • Abhängigkeiten: httpx (Standard), aiohttp (optional), websockets (erforderlich für die Realtime API).

Zuerst musst du die openai-Bibliothek mit pip installieren:

pip install openai

oder 

pip3 install openai

Gehe dann zur offiziellen OpenAI-Websitehttps://platform.openai.com/ , um ein Konto zu registrieren und auf der API-Keys-Seite einen API-Schlüssel zu generieren.

Überprüfe die installierte Version:

import openai
print(openai.__version__)  # 输出当前 SDK 版本

Beispiel

import os
from openai import OpenAI

client = OpenAI(
    # Dies ist der Standard und kann weggelassen werden
    api_key="Der API-Schlüssel, den du beantragt hast",
)

response = client.responses.create(
    model="gpt-4o",
    instructions="Du bist ein Coding-Assistent, der wie ein Pirat spricht.",
    input="Wie prüfe ich, ob ein Python-Objekt eine Instanz einer Klasse ist?",
)

print(response.output_text)

Parameter description:

Parameter Erforderlich Type Beschreibung
api_key Ja str Der OpenAI-Schlüssel, den du beantragt hast
model Ja str Gibt das verwendete OpenAI-Modell an und bestimmt damit Leistungsfähigkeit, Reasoning-Level und Kosten
instructions nein str System-Prompt, der die Identität des Modells, Verhaltensnormen und den Ausdrucksstil definiert und Vorrang hat vorinput
input Ja str / list Inhalt der Benutzereingabe, der die konkrete Frage oder Aufgabe beschreibt

Modelle von Drittanbietern

Der Zugriff auf OpenAI aus China ist nach wie vor etwas problematisch. Viele inländische Anbieter unterstützen ebenfalls OpenAI-Kompatibilität, wie DeepSeek, Qwen und GLM.

DeepSeek

Die DeepSeek-API ist vollständig mit dem API-Format von OpenAI kompatibel. Sie müssen nur ein paar Konfigurationseinstellungen ändern, um das OpenAI SDK oder kompatible Tools direkt für den Zugriff auf die DeepSeek-API zu verwenden.

Parameter Wert / Beschreibung
base_url Erforderlich, fester Wert:https://api.deepseek.com(Sie können auchhttps://api.deepseek.com/v1, nur für OpenAI-Kompatibilität; v1 steht nicht im Zusammenhang mit Modellversionen)
api_key Erforderlich. Sie müssen zuerst einen dedizierten API-Schlüssel auf der offiziellen DeepSeek-Website beantragen (Antrags-URL:https://platform.deepseek.com/)
model

Erforderlich. Das festzulegende Modell:

  • deepseek-v4-flash: Entspricht DeepSeeksNicht-Denkmodus, mit schneller Antwortgeschwindigkeit, geeignet für Routine-Fragen und -Antworten.
  • deepseek-v4-pro: Entspricht DeepSeeksDenkmodus, mit stärkerer Schlussfolgerungsfähigkeit, geeignet zum Lösen komplexer Probleme.

Beispiel

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get('DEEPSEEK_API_KEY'),   # Es wird empfohlen, sie über Umgebungsvariablen zu konfigurieren; Sie können sie auch fest im Code hinterlegen (nicht empfohlen)
    base_url="https://api.deepseek.com")

response = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {"role": "system", "content": "Du bist ein hilfreicher Assistent"},
        {"role": "user", "content": "Hallo"},
    ],
    stream=False    # stream=False Nicht-Streaming (auf einmal zurückgegeben), stream=True Streaming (in Echtzeit zurückgegeben)
)

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

Nutzung

Als Nächstes verwenden wir das OpenAI SDK, um auf die Qwen-Modelle im Bailian-Dienst zuzugreifen.

Beispiel für einen Nicht-Streaming-Aufruf

Beispiel

from openai import OpenAI
import os

def get_response():
    client = OpenAI(
        api_key="sk-xxx",  # Verwenden Sie Ihren Alibaba Cloud Bailian API Key
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # Geben Sie die base_url des DashScope SDK ein
    )
    completion = client.chat.completions.create(
        model="qwen-plus",  # Hier verwenden wir qwen-plus als Beispiel; Sie können den Modellnamen nach Bedarf ändern. Modellliste: https://help.aliyun.com/zh/model-studio/getting-started/models
        messages=[{'role': 'system', 'content': 'Sie sind ein hilfreicher Assistent.'},
                  {'role': 'user', 'content': 'Wer sind Sie?'}]
        )
    # JSON-Daten
    #print(completion.model_dump_json())
    print(completion.choices[0].message.content)

if __name__ == '__main__':
    get_response()

Das Ausführen des Codes liefert das folgende Ergebnis:

I am Qwen, an ultra-large-scale language model independently developed by the Tongyi Laboratory under Alibaba Group. I can help you answer questions and create text, such as writing stories, official documents, emails, scripts, logical reasoning, programming, and more. I can also express opinions and play games. If you have any questions or need help, feel free to tell me at any time!

Beispiel für einen Streaming-Aufruf

Beispiel

from openai import OpenAI

def get_response():
    client = OpenAI(
        api_key="sk-xxx",
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )

    completion = client.chat.completions.create(
        model="qwen-plus",
        messages=[
            {'role': 'system', 'content': 'Sie sind ein hilfreicher Assistent.'},
            {'role': 'user', 'content': 'Wer sind Sie?'}
        ],
        stream=True,
        stream_options={"include_usage": True}
    )

    for chunk in completion:
        # chunk enthält möglicherweise keine choices oder delta
        if hasattr(chunk, "choices") and len(chunk.choices) > 0:
            choice = chunk.choices[0]
            if hasattr(choice, "delta") and hasattr(choice.delta, "content"):
                print(choice.delta.content, end='', flush=True)

if __name__ == '__main__':
    get_response()

Das Ausführen des Codes liefert das folgende Ergebnis:

I am Qwen, an ultra-large-scale language model independently developed by the Tongyi Laboratory under Alibaba Group. I can help you answer questions and create text, such as writing stories, official documents, emails, scripts, logical reasoning, programming, and more. I can also express opinions and play games. If you have any questions or need help, feel free to tell me at any time!

Vorgestellte Funktionen

Vision-Fähigkeit

Unterstützt Bildeingabe (URL oder Base64-kodiert), um multimodale Interaktion zu ermöglichen.

Bild-URL-Eingabe:

Beispiel

prompt = "Was ist auf diesem Bild zu sehen?"
img_url = "https://upload.wikimedia.org/wikipedia/commons/thumb/d/d5/2023_06_08_Raccoon1.jpg/1599px-2023_06_08_Raccoon1.jpg"

response = client.responses.create(
    model="gpt-5.2",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": prompt},
                {"type": "input_image", "image_url": img_url},
            ],
        }
    ],
)
print(response.output_text)

Base64-kodierte Bildeingabe:

Beispiel

import base64
from openai import OpenAI

client = OpenAI()

prompt = "Was ist auf diesem Bild zu sehen?"
# Ein lokales Bild einlesen und als Base64 kodieren
with open("path/to/image.png", "rb") as image_file:
    b64_image = base64.b64encode(image_file.read()).decode("utf-8")

response = client.responses.create(
    model="gpt-5.2",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": prompt},
                {"type": "input_image", "image_url": f"data:image/png;base64,{b64_image}"},
            ],
        }
    ],
)

Asynchrone Verwendung (Async usage)

Ersetze OpenAI durch AsyncOpenAI und rufe die API mit await auf:

Einfaches Async-Beispiel:

Beispiel

import os
import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key=os.environ.get("OPENAI_API_KEY"),
)

async def main() -> None:
    response = await client.responses.create(
        model="gpt-5.2",
        input="Erkläre Disestablishmentarianismus einem klugen fünfjährigen Kind."
    )
    print(response.output_text)

asyncio.run(main())

Für die Optimierung asynchroner Szenarien (aiohttp-Backend) installiere die erweiterte Version:

pip install openai[aiohttp]

Optimierung des aiohttp-Backends:

Beispiel

import os
import asyncio
from openai import DefaultAioHttpClient, AsyncOpenAI

async def main() -> None:
    # Kontextmanager stellt sicher, dass Ressourcen freigegeben werden
    async with AsyncOpenAI(
        api_key=os.environ.get("OPENAI_API_KEY"),
        http_client=DefaultAioHttpClient(),  # Aktiviere das aiohttp-Backend
    ) as client:
        chat_completion = await client.chat.completions.create(
            messages=[{"role": "user", "content": "Sag, dass dies ein Test ist"}],
            model="gpt-5.2",
        )

asyncio.run(main())

Streaming-Antworten

Basierend auf Server Side Events (SSE), um Antworten in Echtzeit zu erhalten; synchrone und asynchrone Schnittstellen sind konsistent.

Synchrones Streaming-Beispiel:

Beispiel

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="gpt-5.2",
    input="Schreibe eine Gutenachtgeschichte über ein Einhorn, die aus einem Satz besteht.",
    stream=True,  # Streaming-Ausgabe aktivieren
)

for event in stream:
    print(event)  # Gib die Antwort Segment für Segment aus

Beispiel für asynchrones Streaming:

Beispiel

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    stream = await client.responses.create(
        model="gpt-5.2",
        input="Schreibe eine Gutenachtgeschichte über ein Einhorn, die aus einem Satz besteht.",
        stream=True,
    )
    async for event in stream:
        print(event)

asyncio.run(main())

Realtime-API

Multimodale Konversation mit niedriger Latenz (Text / Audio), implementiert über WebSocket, erfordert die websockets-Bibliothek.

Einfaches Textbeispiel:

Beispiel

import asyncio
from openai import AsyncOpenAI

async def main():
    client = AsyncOpenAI()
    # Echtzeitverbindung herstellen
    async with client.realtime.connect(model="gpt-realtime") as connection:
        # Sitzungskonfiguration aktualisieren (nur Textausgabe)
        await connection.session.update(
            session={"type": "realtime", "output_modalities": ["text"]}
        )
        # Sende eine Benutzernachricht
        await connection.conversation.item.create(
            item={
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Sag Hallo!"}],
            }
        )
        # Modellantwort auslösen
        await connection.response.create()
        # Auf Echtzeit-Ereignisse lauschen
        async for event in connection:
            if event.type == "response.output_text.delta":
                print(event.delta, flush=True, end="")  # Textfragmente in Echtzeit ausgeben
            elif event.type == "response.output_text.done":
                print()  # Zeilenumbruch hinzufügen, wenn die Antwort endet
            elif event.type == "response.done":
                break  # Zuhören beenden

asyncio.run(main())

Fehlerbehandlung der Realtime-API:

Beispiel

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    async with client.realtime.connect(model="gpt-realtime") as connection:
        async for event in connection:
            if event.type == 'error':
                # Fehlerereignisse abfangen und behandeln (die Verbindung wird nicht getrennt)
                print(f"Fehlertyp: {event.error.type}")
                print(f"Fehlercode: {event.error.code}")
                print(f"Fehlermeldung: {event.error.message}")

asyncio.run(main())

Referenzhandbuch

Die folgende Tabelle enthält:

  • Kerninitialisierung: für synchrone Nutzung,OpenAIfür asynchrone Nutzung,AsyncOpenAIfür Azure-Nutzung,AzureOpenAIEs wird empfohlen, den API Key über Umgebungsvariablen zu konfigurieren.
  • Textgenerierung: bevorzugen Sie die neue Version,responses.create()für klassische Szenarien verwenden Siechat.completions.create()Streaming-Ausgabe erfordert das Festlegen vonstream=True。
  • Erweiterte Funktionen: Unterstützung für Multimodalität (Bildeingabe), Realtime-API, Datei-Upload/Fine-Tuning; Fehlerbehandlung muss catchAPIErrorUnterklassen und Timeout/Retry können flexibel konfiguriert werden.

Installation und Import

Installieren Sie das offizielle SDK:

pip install --upgrade openai

Kernklassen importieren:

from openai import OpenAI, AsyncOpenAI, AzureOpenAI

Einen Client erstellen

Einen standardmäßigen OpenAI-Client erstellen:

from openai import OpenAI

client = OpenAI(api_key="YOUR_API_KEY")

Einen Client über Umgebungsvariablen erstellen:

import os
from openai import OpenAI

os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
client = OpenAI()

Anfrage-Timeout und Anzahl der Wiederholungsversuche festlegen:

client = OpenAI(
    timeout=30,
    max_retries=2
)

Textgenerierung (Responses API)

Die einfachste Textgenerierungsanfrage:

response = client.responses.create(
    model="gpt-5.2",
    input="用一句话解释什么是 Python"
)

print(response.output_text)

Modellverhalten über Anweisungen einschränken:

response = client.responses.create(
    model="gpt-5.2",
    instructions="你是一个严谨的 Python 教程作者",
    input="解释什么是列表推导式"
)

print(response.output_text)

Mehrrundige Konversation

Mit dem input-Array eine kontextbezogene Konversation aufbauen:

response = client.responses.create(
    model="gpt-5.2",
    input=[
        {"role": "user", "content": "Python 是什么?"},
        {"role": "assistant", "content": "Python 是一门高级编程语言。"},
        {"role": "user", "content": "它适合做什么?"}
    ]
)

print(response.output_text)

Strukturierte Ausgabe (JSON Schema)

Bitten Sie das Modell, JSON-Daten zurückzugeben, die dem Schema entsprechen:

response = client.responses.create(
    model="gpt-5.2",
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "tech_summary",
            "schema": {
                "type": "object",
                "properties": {
                    "title": {"type": "string"},
                    "summary": {"type": "string"}
                },
                "required": ["title", "summary"]
            }
        }
    },
    input="介绍 asyncio"
)

print(response.output_parsed)

Tool-Aufruf (Function Calling)

Tools definieren und das Modell automatisch entscheiden lassen, ob es sie aufruft:

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取城市天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"}
                },
                "required": ["city"]
            }
        }
    }
]

response = client.responses.create(
    model="gpt-5.2",
    tools=tools,
    input="北京今天天气怎么样?"
)

print(response.output)

Streaming-Ausgabe (Streaming)

Synchrone Streaming-Ausgabe:

with client.responses.stream(
    model="gpt-5.2",
    input="写一段 Python 示例代码"
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="")

Asynchrone Streaming-Ausgabe:

stream = await client.responses.create(
    model="gpt-5.2",
    input="解释什么是生成式 AI",
    stream=True
)

async for event in stream:
    print(event)

Asynchrone Aufrufe (AsyncOpenAI)

Verwenden Sie den asynchronen Client in asyncio:

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    response = await client.responses.create(
        model="gpt-5.2",
        input="解释什么是事件循环"
    )
    print(response.output_text)

asyncio.run(main())

Realtime-API (Realtime)

Verwenden Sie WebSocket für Echtzeit-Konversationen mit niedriger Latenz:

async with client.realtime.connect(model="gpt-realtime") as conn:
    await conn.session.update(
        session={"type": "realtime", "output_modalities": ["text"]}
    )

    await conn.conversation.item.create(
        item={
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "你好"}]
        }
    )

    async for event in conn:
        print(event)

Bilderstellung

Ein Bild aus Text generieren:

image = client.images.generate(
    model="gpt-image-1",
    prompt="一只在写代码的猫",
    size="1024x1024"
)

print(image.data[0].url)

Embeddings-Generierung

Text-Vektor-Embeddings generieren:

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

print(embedding.data[0].embedding)

Sprache zu Text

Eine Audiodatei in Text umwandeln:

audio_file = open("speech.mp3", "rb")

transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=audio_file
)

print(transcript.text)

Text zu Sprache

Text in eine Sprachdatei umwandeln:

speech = client.audio.speech.create(
    model="gpt-4o-mini-tts",
    voice="alloy",
    input="你好,欢迎使用 OpenAI"
)

with open("output.mp3", "wb") as f:
    f.write(speech)

Dateiverwaltung (Files API)

Eine Datei hochladen:

from pathlib import Path

client.files.create(
    file=Path("data.jsonl"),
    purpose="fine-tune"
)

Dateien auflisten:

files = client.files.list()
for f in files:
    print(f.id, f.filename)

Eine Datei löschen:

client.files.delete(file_id="file-xxx")

Modell-Fine-Tuning (Fine-Tuning-Jobs)

Einen Fine-Tuning-Job erstellen:

job = client.fine_tuning.jobs.create(
    training_file="file-xxx",
    model="gpt-3.5-turbo"
)

print(job.id, job.status)

Webhook-Verifizierung

Webhook-Anfrage verifizieren und parsen:

event = client.webhooks.unwrap(request_body, request_headers)

Nur die Signatur verifizieren:

client.webhooks.verify_signature(request_body, request_headers)

Fehlerbehandlung

Vom SDK ausgelöste Ausnahmen abfangen:

import openai

try:
    client.responses.create(
        model="gpt-5.2",
        input="test"
    )
except openai.RateLimitError as e:
    print("触发速率限制:", e)
except openai.APIConnectionError as e:
    print("连接失败:", e)

Die Request-ID einer fehlgeschlagenen Anfrage abrufen:

try:
    client.responses.create(model="gpt-5.2", input="test")
except openai.APIStatusError as exc:
    print(exc.request_id)

Azure OpenAI-Kompatibilität

Azure OpenAI-Client initialisieren:

from openai import AzureOpenAI

client = AzureOpenAI(
    api_version="2023-07-01-preview",
    azure_endpoint="https://xxx.openai.azure.com"
)

Versionen und Changelog

SDK-Version prüfen:

import openai
print(openai.__version__)

SDK-Debug-Logging aktivieren:

export OPENAI_LOG=debug

Weitere API-Referenzen:https://github.com/openai/openai-python/blob/main/api.md

Andere Erweiterungen