Yapay zeka modelleri ve platformları
Python’da Eşzamansız LLM API Çağrıları: Kapsamlı Bir Rehber
Unite.AI sitesini Google'daki tercih ettiğiniz kaynaklara ekleyin
Geliştiriciler ve veri bilimcileri olarak, güçlü modellerle API’ler aracılığıyla etkileşim kurmamız souvent gerekebilir. Ancak, uygulamalarımız karmaşıklık ve ölçek açısından büyüdükçe, verimli ve yüksek performanslı API etkileşimlerinin gerekliliği ortaya çıkar. İşte burada eşzamansız programlama parlar, asyncio ve aiohttp ile LLM API’lerini maksimum düzeyde kullanmamızı sağlar.
Bu kapsamlı rehberde, Python’da eşzamansız LLM API çağrılarının dünyasını keşfedeceğiz. Eşzamansız programlamanın temellerinden advanced tekniklere kadar her şeyi kapsayacağız. Bu makaleyi tamamladığınızda, eşzamansız programlamanın LLM güçlendirilmiş uygulamalarınızı nasıl güçlendirebileceğini anlamış olacaksınız.
Öncelikle, eşzamansız programlama kavramlarına sağlam bir temel oluşturalım.
Eşzamansız programlama, ana yürütme işleminin engellenmeden birden fazla işlemi aynı anda gerçekleştirmesine olanak tanır. Python’da, bu principalmente asyncio modülü aracılığıyla gerçekleştirilir, eşzamansız kodu yazmak için coroutine’ler, event loop’lar ve gelecekler sağlar.
Ana Kavramlar:
- Coroutine’ler: async def ile tanımlanan fonksiyonlar, durdurulabilir ve yeniden başlatılabilir.
- Event Loop: Eşzamansız görevleri yöneten ve çalıştıran merkezi yürütme mekanizması.
- Beiwaitable’ler: await anahtar sözcüğü ile kullanılabilen nesneler (coroutine’ler, görevler, gelecekler).
Şimdi, bu kavramları gösteren basit bir örnek verelim:
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # Bir I/O işlemi simüle et
print(f"Merhaba, {name}!")</p>
<p>async def main():
await asyncio.gather(
greet("Alice"),
greet("Bob"),
greet("Charlie")
)</p>
asyncio.run(main())
Bu örnekte, async def ile tanımlanan bir greet fonksiyonu tanımladık, bu fonksiyon bir I/O işlemi simüle etmek için asyncio.sleep() kullanıyor. main fonksiyonu, asyncio.gather() kullanarak birden fazla selamlaşma işlemini aynı anda çalıştırıyor.
Bu yaklaşım, tüm selamlaşmaların yaklaşık 1 saniye sonra basılmasını sağlar, eşzamansız yürütmenin gücünü gösterir.
LLM API Çağrılarında Eşzamansızlık İhtiyacı
LLM API’leriyle çalışırken, genellikle birden fazla API çağrısı yapmamız gerekir, ya da sıralı ya da paralel olarak. Geleneksel senkronik kod, özellikle ağ istekleri gibi yüksek gecikme operasyonlarında önemli performans darboğazlarına neden olabilir.
Örneğin, 100 farklı makale için özetler oluşturmak için bir LLM API’sini kullanmak istediğimizi varsayalım. Senkronik bir yaklaşım, her API çağrısının bir önceki çağrının yanıtını alana kadar beklemesine neden olur, bu da tüm isteklerin tamamlanması birkaç dakika sürebilir. Eşzamansız bir yaklaşım ise, birden fazla API çağrısını aynı anda başlatmamızı sağlar, bu da toplam yürütme süresini önemli ölçüde azaltır.
Çevrenizi Ayarlama
Eşzamansız LLM API çağrıları ile çalışmaya başlamak için, Python ortamınızı gerekli kütüphanelerle ayarlaymanız gerekir. İşte ihtiyacınız olan şeyler:
- Python 3.7 veya daha yüksek (yerel asyncio desteği için)
- aiohttp: Eşzamansız bir HTTP istemci kütüphanesi
- openai: Resmi OpenAI Python istemcisi (OpenAI’nin GPT modellerini kullanıyorsanız)
- langchain: LLM’lerle uygulamalar oluşturmak için bir çerçeve (isteğe bağlı, ancak karmaşık iş akışları için önerilir)
Bu bağımlılıkları pip kullanarak kurabilirsiniz:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Asyncio ve aiohttp ile Temel Eşzamansız LLM API Çağrıları
Şimdi, aiohttp kullanarak bir LLM API’sine basit bir eşzamansız çağrı yapalım. OpenAI’nin GPT-3.5 API’sini bir örnek olarak kullanacağız, ancak kavramlar diğer LLM API’lerine de uygulanabilir.
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 = [
"Kuantum bilgisayarları basit terimlerle açıklar mısınız?",
"Yapay zeka hakkında bir haiku yazın.",
"Fotosentez sürecini açıklar mısınız?"
]</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"İstek: {prompt}\nYanıt: {result}\n")</p>
asyncio.run(main())
Bu örnekte, async def ile tanımlanan bir generate_text fonksiyonu tanımladık, bu fonksiyon OpenAI API’sini kullanarak bir metin oluşturur. main fonksiyonu, asyncio.gather() kullanarak birden fazla metin oluşturma işlemini aynı anda çalıştırır.
Bu yaklaşım, birden fazla isteği aynı anda API’ye göndermemizi sağlar, bu da toplam işlem süresini önemli ölçüde azaltır.
İleri Teknikler: Toplu İşlem ve Eşzamansızlık Kontrolü
Gerçek dünya uygulamalarında, daha gelişmiş yaklaşımlar genellikle gereklidir. İki önemli tekniği keşfedeceğiz: toplu işlem ve eşzamansızlık kontrolü.
Toplu İşlem: Birçok isteği işlerken, her bir isteği ayrı ayrı göndermek yerine toplu olarak göndermek daha verimlidir. Bu, birden fazla API çağrısının gereksiz yükünü azaltır ve daha iyi performans sağlar.
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"Sayı {i} hakkında bir gerçeği söyleyin." 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"İstek: {prompt}\nYanıt: {result}\n")</p>
asyncio.run(main())
Eşzamansızlık Kontrolü: Eşzamansız programlama, eşzamansız yürütme sağlar, ancak API sunucusunu veya hız sınırlarını aşmamaya dikkat etmek önemlidir. asyncio.Semaphore kullanarak eşzamansızlığı kontrol edebiliriz.
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"Sayı {i} hakkında bir gerçeği söyleyin." 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"İstek: {prompt}\nYanıt: {result}\n")</p>
asyncio.run(main())
Bu örnekte, semaphore kullanarak eşzamansızlığı kontrol ediyoruz, API sunucusunu veya hız sınırlarını aşmamaya dikkat ediyoruz.
Hata İşleme ve Yeniden Deneme Eşzamansız LLM Çağrıları
Dış API’lerle çalışırken, güçlü hata işleme ve yeniden deneme mekanizmaları uygulamak önemlidir. Kodumuzu geliştirelim ve ortak hataları işleyelim ve yeniden deneme için üssümsel geri çekme uygulayalım.
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"Hata oluştu: {e}")
raise APIError("Metin oluşturulamadı")</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, "Birden fazla deneme sonrasında yanıt oluşturulamadı."</p>
<p>async def main():
prompts = [f"Sayı {i} hakkında bir gerçeği söyleyin." 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"İstek: {prompt}\nYanıt: {result}\n")</p>
asyncio.run(main())
Bu geliştirilmiş sürüm, APIError adlı bir özel istisna sınıfı, generate_text_with_retry adlı bir fonksiyon ve process_prompt adlı bir fonksiyon içerir.
Performansı Optimizleme: Yanıt Akışı
Uzun metin oluşturma için, yanıt akışı uygulamanızın algılanan performansını önemli ölçüde iyileştirebilir. Tüm yanıtı beklemek yerine, metin parçalarını oluşturuldukça işleyebilir ve görüntüleyebilirsiniz.
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 = "Zaman yolculuğu yapan bir bilim insanı hakkında bir kısa hikaye yazın."</p>
<p>async with AsyncOpenAI() as client:
result = await stream_text(prompt, client)</p>
<p>print(f"Tüm yanıt:\n{result}")</p>
asyncio.run(main())
Bu örnekte, yanıt akışı kullanarak API’den gelen yanıtı parçalar halinde işliyoruz ve her parçayı oluşturuldukça basıyoruz.
LangChain ile Eşzamansız İş Akışları Oluşturma
Daha karmaşık LLM güçlendirilmiş uygulamalar için, LangChain çerçevesi yüksek düzeyde bir soyutlama sağlar ve birden fazla LLM çağrısını zincirleme ve diğer araçları entegre etmeyi basitleştirir. LangChain ile eşzamansız iş akışları oluşturmayı keşfedeceğiz:
Bu örnek, LangChain’in daha karmaşık iş akışları oluşturmak için nasıl kullanılabileceğini gösterir.
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="{topic} hakkında bir kısa hikaye yazın."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>
<p>async def main():
topics = ["bir büyülü orman", "bir gelecek şehri", "bir su altı uygarlığı"]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):
print(f"\nKonu: {topic}\nHikaye: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Eşzamansız LLM Uygulamalarını FastAPI ile Hizmet Verme
Eşzamansız LLM uygulamanızı bir web hizmeti olarak sunmak için, FastAPI eşzamansız operasyonları desteklediği için iyi bir seçimdir. İşte basit bir API endpoint’u oluşturmanın bir örneği:
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># Biraz post-işlem simüle edelim
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):
# Biraz logging veya ek işlemler simüle edelim
await asyncio.sleep(2)
print(f"Logged: Prompt '{prompt}' generated text of length {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Bu FastAPI uygulaması, bir endpoint oluşturur ve bir metin oluşturur, ayrıca arka plan görevleri için bir örnek verir.
En İyi Uygulamalar ve Sık Karşılaşılan Hatalar
Eşzamansız LLM API’leriyle çalışırken, aşağıdaki en iyi uygulamalara dikkat edin:
- Bağlantı havuzu kullanın: Birden fazla istek yaparken, bağlantıları yeniden kullanın.
- Doğru hata işleme uygulayın: Ağ sorunları, API hataları ve beklenmeyen yanıtları her zaman dikkate alın.
- Hız sınırlarına saygı gösterin: Eşzamansızlığı kontrol etmek için semaforlar veya diğer mekanizmalar kullanın.
- İzleme ve günlüğe kaydetme: Performansı izlemek ve sorunları tespit etmek için kapsamlı günlüğe kaydetme uygulayın.
- Uzun metin için akışı kullanın: Kullanıcı deneyimini iyileştirir ve kısmi sonuçların erken işlenmesine olanak tanır.
