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
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:
|
Beispiel
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
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
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
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
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 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 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
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
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
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
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 catch
APIErrorUnterklassen 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
Andere ErweiterungenWeitere API-Referenzen:https://github.com/openai/openai-python/blob/main/api.md