Parallelismus und Konfunktionalität
Auf dieser Seite
Gemeinsames Warten ist kein gemeinsames Rechnen
Ein API-Aufruf bei OpenAI ist eine I/O-lastige Operation: Ihr Programm rechnet keine Zahlen, es wartet auf eine Netzwerkantwort. Diese Natur diktiert das richtige Werkzeug. Parallelismus im Sinne von Multiprozessing verteilt Berechnungen über mehrere Kerne; es kostet einen vollständigen Prozess pro Arbeitseinheit, und das Ausführen von hundert Prozessen, die alle auf das Netzwerk warten, ist reiner Verschwendung. Konfunktionalität, wie sie von asyncio bereitgestellt wird, ermöglicht es einem einzelnen Prozess, hunderte gleichzeitige Warteoperationen mit vernachlässigbarem Aufwand zu bewältigen. Dies ist der Schlüssel zur Maximierung Ihres Aufrufdurchsatzes.
Der Unterschied ist mehr als nur akademisch: Ein Team, das seine LLM-Aufrufe mit einem Prozesspool parallelisiert, verbraucht mehrere Gigabyte Arbeitsspeicher für keinen Gewinn im Vergleich zu wenigen Zeilen asyncio.
Starten von konfunktionalen Aufrufen
Die SDK-Bibliothek stellt einen asynchronen Client, AsyncOpenAI, bereit, dessen Aufrufe mit await abgewartet werden. Das Sammeln mehrerer Koroutinen in einem asyncio.gather lässt sie alle gleichzeitig fortschreiten, wobei die Gesamtdauer derjenigen des langsamsten Aufrufs entspricht, nicht der Summe ihrer Einzelzeiten.
import asyncio
import openai
client = openai.AsyncOpenAI()
async def single_call(prompt: str, model: str = "gpt-5.6-terra") -> str:
"""Ein einzelner asynchroner API-Aufruf."""
response = await client.responses.create(
model=model,
input=prompt,
)
return response.output_text
async def concurrent_calls(prompts: list[str]) -> list[str]:
"""Startet alle Aufrufe konfunktional."""
tasks = [single_call(prompt) for prompt in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"Fehler bei Prompt {i}: {result}")
return [r for r in results if not isinstance(r, Exception)]Das Flag return_exceptions=True ist das, was eine Prototypenlösung von einer produktionsreifen Verarbeitung unterscheidet. Ohne dieses Flag unterbricht die erste Ausnahme den gather, und Sie verlieren die bereits erzielten neunundneunzig Ergebnisse. Mit diesem Flag werden Fehler zu Werten, die Sie inspizieren, protokollieren und in Ihrem eigenen Tempo erneut ausführen können.
Diese rohe Version hat jedoch einen sofortigen Mangel: Das gleichzeitige Starten von tausend Aufrufen würde Ihre Rate-Limits in Sekunden überschreiten. Ein Semaphore löst dies, indem er nur eine festgelegte Anzahl gleichzeitiger Aufrufe zulässt, während der Rest geduldig auf seine Reihe wartet. Zwanzig ist ein vernünftiger Ausgangspunkt, der je nach den tatsächlichen Quoten Ihrer Organisation angepasst werden sollte.
async def calls_with_limit(
prompts: list[str],
max_concurrency: int = 20,
) -> list[str]:
"""Begrenzt die Anzahl der gleichzeitigen Aufrufe."""
semaphore = asyncio.Semaphore(max_concurrency)
async def limited_call(prompt: str) -> str:
async with semaphore:
return await single_call(prompt)
tasks = [limited_call(prompt) for prompt in prompts]
return await asyncio.gather(*tasks, return_exceptions=True)Das Map-Reduce-Muster angewendet auf Text
Ein zweihundertseitiges Dokument kann nicht in einem einzigen Aufruf zusammengefasst werden, und eine sequenzielle Aufteilung würde Minuten in Anspruch nehmen. Das Map-Reduce-Muster lässt sich hier natürlich anwenden: Aufteilen in Abschnitte, jeden Abschnitt konfunktional zusammenfassen und dann das Modell um eine Synthese der Zusammenfassungen bitten. Die Map-Phase profitiert voll und ganz von der Konfunktionalität, während die Reduce-Phase ein einziger Aufruf auf Text ist, der nun kurz geworden ist.
async def map_reduce_summary(
document: str,
chunk_size: int = 3000,
) -> str:
"""Zusammenfasst ein langes Dokument über paralleles Map-Reduce."""
# Phase MAP: Aufteilen und parallel zusammenfassen
chunks = [
document[i:i + chunk_size]
for i in range(0, len(document), chunk_size)
]
map_prompts = [
f"Zusammenfassen Sie diesen Abschnitt in 3 Schlüsselpunkten:\n\n{chunk}"
for chunk in chunks
]
partial_summaries = await calls_with_limit(map_prompts, max_concurrency=10)
# Phase REDUCE: Zusammenfassungen kombinieren
summary_text = "\n\n---\n\n".join(
r for r in partial_summaries if isinstance(r, str)
)
final_summary = await single_call(
f"Synthetisieren Sie diese Teilzusammenfassungen zu einer kohärenten Zusammenfassung "
f"von 5 Schlüsselpunkten:\n\n{summary_text}"
)
return final_summaryDas Filtern if isinstance(r, str) ist sowohl eine produktspezifische Entscheidung als auch eine technische Vorsichtsmaßnahme: Es bedeutet, dass die finale Synthese auch dann produziert wird, wenn drei Abschnitte fehlgeschlagen sind, und zwar stumm. Je nach Kontext ist dies genau das, was Sie wollen, oder es ist inakzeptabel. Entscheiden Sie sich explizit und protokollieren Sie in jedem Fall die Anzahl der verlorenen Abschnitte.
Wann Multiprozessing wieder relevant wird
Es bleiben seltene Situationen, in denen die Berechnung wieder die Kontrolle übernimmt: aufwändige Nachverarbeitung von Antworten, Entitätsextraktion, Bewertung, strukturiertes Parsen. Diese Arbeit ist CPU-lastig und bringt von asyncio keinen Gewinn. Die elegante Kombination besteht darin, die beiden Regime in einer einzigen Pipeline zu verketten: Netzwerkaufrufe in Konfunktionalität, gefolgt von einer Berechnung, die über einen Prozesspool via run_in_executor verteilt wird.
import asyncio
from concurrent.futures import ProcessPoolExecutor
def heavy_post_processing(text: str) -> dict:
"""CPU-lastige Verarbeitung des Ergebnisses (NLP, Parsen usw.)."""
# Beispiel: Entitätsextraktion, Bewertung usw.
words = text.split()
return {
"word_count": len(words),
"unique_words": len(set(words)),
}
async def complete_pipeline(prompts: list[str]) -> list[dict]:
"""Pipeline: asynchrone LLM-Aufrufe + parallele CPU-Nachverarbeitung."""
# Schritt 1: konfunktionale LLM-Aufrufe
llm_results = await calls_with_limit(prompts)
# Schritt 2: parallele Nachverarbeitung (CPU)
loop = asyncio.get_event_loop()
with ProcessPoolExecutor(max_workers=4) as pool:
tasks = [
loop.run_in_executor(pool, heavy_post_processing, r)
for r in llm_results
if isinstance(r, str)
]
final_results = await asyncio.gather(*tasks)
return final_resultsStellen Sie sicher, dass die Berechnung den Pool wirklich rechtfertigt, bevor Sie ihn einführen: Das Serialisieren von Daten in einen anderen Prozess hat einen Kostenfaktor, und für Behandlungen, die nur wenige Millisekunden dauern, übersteigt dieser Kostenfaktor den Gewinn.
Überprüfung des realen Durchsatzes
Das Einstellen von max_concurrency ist keine Vermutung. Zu niedrig, und Sie lassen Durchsatz auf der Strecke; zu hoch, und Sie lösen Rate-Limits aus, die Ihnen mehr Zeit kosten, als sie sparen. Messen Sie unter Ihrer tatsächlichen Last, variieren Sie den Wert und betrachten Sie sowohl den Durchsatz als auch die Erfolgsanzahl – eine sehr schnelle Pipeline, die bei jedem fünften Aufruf fehlschlägt, ist nicht schnell.
import time
async def benchmark_throughput(num_requests: int = 100):
"""Misst den realen Durchsatz Ihrer Pipeline."""
prompts = [
f"Erteilen Sie eine interessante Tatsache zur Nummer {i}."
for i in range(num_requests)
]
start = time.perf_counter()
results = await calls_with_limit(prompts, max_concurrency=20)
duration = time.perf_counter() - start
successes = sum(1 for r in results if isinstance(r, str))
print(f"Anfragen: {num_requests}")
print(f"Erfolge: {successes}")
print(f"Dauer: {duration:.1f}s")
print(f"Durchsatz: {successes / duration:.1f} req/s")Kernaussagen
- Verwenden Sie
AsyncOpenAIundasyncio.gatherfür I/O-Konfunktionalität - Begrenzen Sie die Konfunktionalität mit einem Semaphore, um Quoten zu respektieren
- Das Map-Reduce-Muster parallelisiert die Verarbeitung langer Dokumente
- Reservieren Sie
ProcessPoolExecutorfür CPU-intensiven Nachverarbeitung