Model dan platform AI
Panggilan API LLM Asinkron di Python: Panduan Lengkap
Tambahkan Unite.AI ke sumber pilihan Anda di Google
Sebagai pengembang dan ilmuwan data, kita sering menemukan diri kita memerlukan berinteraksi dengan model-model kuat ini melalui API. Namun, ketika aplikasi kita tumbuh dalam kompleksitas dan skala, kebutuhan akan interaksi API yang efisien dan performant menjadi sangat penting. Inilah di mana pemrograman asinkron bersinar, memungkinkan kita untuk memaksimalkan throughput dan meminimalkan latency saat bekerja dengan API LLM.
Dalam panduan lengkap ini, kita akan menjelajahi dunia panggilan API LLM asinkron di Python. Kita akan membahas semua hal, dari dasar-dasar pemrograman asinkron hingga teknik lanjutan untuk menangani alur kerja yang kompleks. Pada akhir artikel ini, Anda akan memiliki pemahaman yang solid tentang cara menggunakan pemrograman asinkron untuk meningkatkan aplikasi LLM Anda.
Sebelum kita memasuki spesifik panggilan API LLM asinkron, mari kita membangun fondasi yang kuat dalam konsep pemrograman asinkron.
Pemrograman asinkron memungkinkan beberapa operasi untuk dijalankan secara bersamaan tanpa memblokir thread eksekusi utama. Di Python, ini sebagian besar dicapai melalui modul asyncio, yang menyediakan kerangka kerja untuk menulis kode konkuren menggunakan korutin, loop acara, dan masa depan.
Konsep Kunci:
- Korutin: Fungsi yang didefinisikan dengan async def yang dapat dihentikan dan dilanjutkan.
- Loop Acara: Mekanisme eksekusi sentral yang mengelola dan menjalankan tugas asinkron.
- Objek yang Dapat Menunggu: Objek yang dapat digunakan dengan kata kunci menunggu (korutin, tugas, masa depan).
Berikut adalah contoh sederhana untuk mengilustrasikan konsep-konsep ini:
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # Simulasi operasi I/O
print(f"Halo, {name}!")</p>
<p>async def main():
await asyncio.gather(
greet("Alice"),
greet("Bob"),
greet("Charlie")
)</p>
asyncio.run(main())
Dalam contoh ini, kita mendefinisikan fungsi asinkron greet yang mensimulasikan operasi I/O dengan asyncio.sleep(). Fungsi main menggunakan asyncio.gather() untuk menjalankan beberapa sapaan secara bersamaan. Meskipun ada penundaan sleep, semua tiga sapaan akan dicetak setelah sekitar 1 detik, menunjukkan kekuatan eksekusi asinkron.
Keperluan Asinkron dalam Panggilan API LLM
Ketika bekerja dengan API LLM, kita sering menghadapi skenario di mana kita perlu membuat beberapa panggilan API, baik secara berurutan atau paralel. Kode sinkron tradisional dapat menyebabkan bottleneck kinerja yang signifikan, terutama ketika menangani operasi berlatensi tinggi seperti permintaan jaringan ke layanan LLM.
Pertimbangkan skenario di mana kita perlu menghasilkan ringkasan untuk 100 artikel yang berbeda menggunakan API LLM. Dengan pendekatan sinkron, setiap panggilan API akan memblokir sampai menerima respons, berpotensi memakan waktu beberapa menit untuk menyelesaikan semua permintaan. Pendekatan asinkron, di sisi lain, memungkinkan kita untuk memulai beberapa panggilan API secara bersamaan, secara dramatis mengurangi waktu eksekusi keseluruhan.
Mengatur Lingkungan Anda
Untuk memulai dengan panggilan API LLM asinkron, Anda perlu mengatur lingkungan Python Anda dengan perpustakaan yang diperlukan. Berikut adalah apa yang Anda butuhkan:
- Python 3.7 atau yang lebih tinggi (untuk dukungan asinkron asli)
- aiohttp: Perpustakaan klien HTTP asinkron
- openai: Klien Python OpenAI resmi (jika Anda menggunakan model GPT OpenAI)
- langchain: Kerangka kerja untuk membangun aplikasi dengan LLM (opsional, tetapi disarankan untuk alur kerja yang kompleks)
Anda dapat menginstal dependensi ini menggunakan pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Panggilan API LLM Asinkron Dasar dengan asyncio dan aiohttp
Mari kita mulai dengan membuat panggilan API LLM asinkron sederhana menggunakan aiohttp. Kita akan menggunakan API GPT-3.5 OpenAI sebagai contoh, tetapi konsepnya berlaku untuk API LLM lainnya juga.
import asyncio
import aiohttp
from openai import AsyncOpenAI
<p>async def generate_text(prompt, client):
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content</p>
<p>async def main():
prompts = [
"Jelaskan komputasi kuantum dalam istilah sederhana.",
"Tulis haiku tentang kecerdasan buatan.",
"Jelaskan proses fotosintesis."
]</p>
<p>async with AsyncOpenAI() as client:
tasks = [generate_text(prompt, client) for prompt in prompts]
results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in zip(prompts, results):
print(f"Prompt: {prompt}\nRespons: {result}\n")</p>
asyncio.run(main())
Dalam contoh ini, kita mendefinisikan fungsi asinkron generate_text yang membuat panggilan ke API OpenAI menggunakan klien AsyncOpenAI. Fungsi main membuat beberapa tugas untuk prompt yang berbeda dan menggunakan asyncio.gather() untuk menjalankannya secara bersamaan.
Pendekatan ini memungkinkan kita untuk mengirim beberapa permintaan ke API LLM secara bersamaan, secara signifikan mengurangi waktu total yang diperlukan untuk memproses semua prompt.
Teknik Lanjutan: Pengelompokan dan Kontrol Konkurensi
Sementara contoh sebelumnya mendemonstrasikan dasar-dasar panggilan API LLM asinkron, aplikasi dunia nyata sering memerlukan pendekatan yang lebih canggih. Mari kita jelajahi dua teknik penting: pengelompokan permintaan dan kontrol konkurensi.
Pengelompokan Permintaan: Ketika menangani sejumlah besar prompt, seringkali lebih efisien untuk mengelompokkan mereka menjadi grup daripada mengirim permintaan individu untuk setiap prompt. Ini mengurangi overhead beberapa panggilan API dan dapat menyebabkan kinerja yang lebih baik.
import asyncio
from openai import AsyncOpenAI
<p>async def process_batch(batch, client):
responses = await asyncio.gather(*[
client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
) for prompt in batch
])
return [response.choices[0].message.content for response in responses]</p>
<p>async def main():
prompts = [f"Beritahu saya sebuah fakta tentang angka {i}" for i in range(100)]
batch_size = 10</p>
<p>async with AsyncOpenAI() as client:
results = []
for i in range(0, len(prompts), batch_size):
batch = prompts[i:i+batch_size]
batch_results = await process_batch(batch, client)
results.extend(batch_results)</p>
<p>for prompt, result in zip(prompts, results):
print(f"Prompt: {prompt}\nRespons: {result}\n")</p>
asyncio.run(main())
Kontrol Konkurensi: Sementara pemrograman asinkron memungkinkan eksekusi konkuren, penting untuk mengontrol tingkat konkurensi untuk menghindari membebani server API atau melebihi batas tarif. Kita dapat menggunakan asyncio.Semaphore untuk tujuan ini.
import asyncio
from openai import AsyncOpenAI
<p>async def generate_text(prompt, client, semaphore):
async with semaphore:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content</p>
<p>async def main():
prompts = [f"Beritahu saya sebuah fakta tentang angka {i}" for i in range(100)]
max_concurrent_requests = 5
semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:
tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]
results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in zip(prompts, results):
print(f"Prompt: {prompt}\nRespons: {result}\n")</p>
asyncio.run(main())
Dalam contoh ini, kita menggunakan semaphore untuk membatasi jumlah permintaan konkuren menjadi 5, memastikan kita tidak membebani server API.
Pengelolaan Kesalahan dan Pengulangan dalam Panggilan LLM Asinkron
Ketika bekerja dengan API eksternal, sangat penting untuk mengimplementasikan pengelolaan kesalahan dan mekanisme pengulangan yang kuat. Mari kita tingkatkan kode kita untuk menangani kesalahan umum dan mengimplementasikan pengulangan eksponensial.
import asyncio
import random
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
class APIError(Exception):
pass
<p>@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def generate_text_with_retry(prompt, client):
try:
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
except Exception as e:
print(f"Kesalahan terjadi: {e}")
raise APIError("Gagal menghasilkan teks")</p>
<p>async def process_prompt(prompt, client, semaphore):
async with semaphore:
try:
result = await generate_text_with_retry(prompt, client)
return prompt, result
except APIError:
return prompt, "Gagal menghasilkan respons setelah beberapa upaya."</p>
<p>async def main():
prompts = [f"Beritahu saya sebuah fakta tentang angka {i}" for i in range(20)]
max_concurrent_requests = 5
semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:
tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]
results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in results:
print(f"Prompt: {prompt}\nRespons: {result}\n")</p>
asyncio.run(main())
Versi ini ditingkatkan termasuk:
- Penanganan kesalahan khusus untuk kesalahan API.
- Fungsi generate_text_with_retry yang dihiasi dengan @retry dari perpustakaan tenacity, mengimplementasikan pengulangan eksponensial.
- Penanganan kesalahan dalam fungsi process_prompt untuk menangani kegagalan.
Mengoptimalkan Kinerja: Mengalirkan Respons
Untuk konten panjang, mengalirkan respons dapat secara signifikan meningkatkan kinerja aplikasi Anda. Sebagai gantinya menunggu respons lengkap, Anda dapat memproses dan menampilkan potongan teks saat mereka tersedia.
import asyncio
from openai import AsyncOpenAI
<p>async def stream_text(prompt, client):
stream = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
stream=True
)</p>
<p>full_response = ""
async for chunk in stream:
if chunk.choices[0].delta.content is not None:
content = chunk.choices[0].delta.content
full_response += content
print(content, end='', flush=True)</p>
<p>print("\n")
return full_response</p>
<p>async def main():
prompt = "Tulis sebuah cerita pendek tentang seorang ilmuwan yang melakukan perjalanan waktu."</p>
<p>async with AsyncOpenAI() as client:
result = await stream_text(prompt, client)</p>
<p>print(f"Respons Lengkap:\n{result}")</p>
asyncio.run(main())
Contoh ini mendemonstrasikan cara mengalirkan respons dari API, mencetak setiap potongan saat tiba. Pendekatan ini sangat berguna untuk aplikasi obrolan atau skenario di mana Anda ingin memberikan umpan balik waktu nyata kepada pengguna.
Membangun Alur Kerja Asinkron dengan LangChain
Untuk aplikasi LLM yang lebih kompleks, kerangka kerja LangChain menyediakan abstraksi tingkat tinggi yang memudahkan proses menghubungkan beberapa panggilan LLM dan mengintegrasikan alat lain. Mari kita lihat contoh menggunakan LangChain dengan kemampuan asinkron:
Contoh ini menunjukkan bagaimana LangChain dapat digunakan untuk membuat alur kerja yang lebih kompleks dengan streaming dan eksekusi asinkron. AsyncCallbackManager dan StreamingStdOutCallbackHandler memungkinkan pengaliran respons yang dihasilkan secara waktu nyata.
import asyncio
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.callbacks.manager import AsyncCallbackManager
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
<p>async def generate_story(topic):
llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
prompt = PromptTemplate(
input_variables=["topic"],
template="Tulis sebuah cerita pendek tentang {topic}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>
<p>async def main():
topics = ["hutan ajaib", "kota futuristik", "peradaban bawah laut"]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):
print(f"\nTopik: {topic}\nCerita: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Melayani Aplikasi LLM Asinkron dengan FastAPI
Untuk membuat aplikasi LLM asinkron Anda tersedia sebagai layanan web, FastAPI adalah pilihan yang baik karena dukungan asinkron bawaannya. Berikut adalah contoh tentang cara membuat endpoint API sederhana untuk penghasilan teks:
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI
app = FastAPI()
client = AsyncOpenAI()
<p>class GenerationRequest(BaseModel):
prompt: str</p>
<p>class GenerationResponse(BaseModel):
generated_text: str</p>
<p>@app.post("/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": request.prompt}]
)
generated_text = response.choices[0].message.content</p>
<p># Simulasi beberapa pemrosesan latar belakang
background_tasks.add_task(log_generation, request.prompt, generated_text)</p>
<p>return GenerationResponse(generated_text=generated_text)</p>
<p>async def log_generation(prompt: str, generated_text: str):
# Simulasi logging atau pemrosesan tambahan
await asyncio.sleep(2)
print(f"Logged: Prompt '{prompt}' menghasilkan teks sepanjang {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Aplikasi FastAPI ini membuat endpoint /generate yang menerima prompt dan mengembalikan teks yang dihasilkan. Ini juga mendemonstrasikan cara menggunakan tugas latar belakang untuk pemrosesan tambahan tanpa memblokir respons.
Praktik Terbaik dan Kesalahan Umum
Saat bekerja dengan panggilan API LLM asinkron, pertahankan praktik terbaik berikut:
- Gunakan pengelompokan koneksi: Saat membuat beberapa permintaan, gunakan kembali koneksi untuk mengurangi overhead.
- Implementasikan penanganan kesalahan yang tepat: Selalu pertimbangkan masalah jaringan, kesalahan API, dan respons yang tidak terduga.
- Hormati batas tarif: Gunakan semafor atau mekanisme kontrol konkurensi lain untuk menghindari membebani server API.
- Pantau dan log: Implementasikan logging yang komprehensif untuk melacak kinerja dan mengidentifikasi masalah.
- Gunakan pengaliran untuk konten panjang: Ini meningkatkan pengalaman pengguna dan memungkinkan pemrosesan awal hasil parsial.
