Python'da Asenkron LLM API Çağrıları: Kapsamlı Bir Kılavuz

Geliştiriciler ve dta bilim insanları olarak, kendimizi sıklıkla API'ler aracılığıyla bu güçlü modellerle etkileşime girme ihtiyacı içinde buluyoruz. Ancak, uygulamalarımız karmaşıklık ve ölçek açısından büyüdükçe, verimli ve performanslı API etkileşimlerine duyulan ihtiyaç kritik hale geliyor. İşte asenkron programlamanın parladığı yer burasıdır ve LLM API'leriyle çalışırken verimi en üst düzeye çıkarmamızı ve gecikmeyi en aza indirmemizi sağlar.

Bu kapsamlı kılavuzda, Python'da eşzamansız LLM API çağrılarının dünyasını keşfedeceğiz. Eşzamansız programlamanın temellerinden karmaşık iş akışlarını yönetmeye yönelik gelişmiş tekniklere kadar her şeyi ele alacağız. Bu makalenin sonunda, LLM destekli uygulamalarınızı güçlendirmek için eşzamansız programlamayı nasıl kullanacağınız konusunda sağlam bir anlayışa sahip olacaksınız.

Asenkron LLM API çağrılarının ayrıntılarına dalmadan önce, asenkron programlama kavramlarında sağlam bir temel oluşturalım.

Eşzamansız programlama, yürütmenin ana iş parçacığını engellemeden birden fazla işlemin eş zamanlı olarak yürütülmesine olanak tanır. Python'da, bu öncelikle şu şekilde elde edilir: Asyncio coroutine'leri, olay döngülerini ve gelecekleri kullanarak eş zamanlı kod yazmak için bir çerçeve sağlayan modül.

Anahtar kavramlar:

• Eşyordamlar: ile tanımlanan işlevler asenkron tanım Duraklatılıp tekrar başlatılabilen.
• Olay Döngüsü: Asenkron görevleri yöneten ve çalıştıran merkezi yürütme mekanizması.
• Beklenenler: Await anahtar sözcüğü ile kullanılabilen nesneler (coroutines, tasks, futures).

Bu kavramları açıklamak için basit bir örnek verelim:

import asyncio

async def greet(name):
    await asyncio.sleep(1)  # Simulate an I/O operation
    print(f"Hello, {name}!")

async def main():
    await asyncio.gather(
        greet("Alice"),
        greet("Bob"),
        greet("Charlie")
    )

asyncio.run(main())

Bu örnekte, asenkron bir fonksiyon tanımlıyoruz greet bir G/Ç işlemini simüle eden asyncio.sleep(). main fonksiyon kullanır asyncio.gather() birden fazla selamlamayı aynı anda çalıştırmak için. Uyku gecikmesine rağmen, üç selamlama da yaklaşık 1 saniye sonra yazdırılacak ve bu da eş zamanlı yürütmenin gücünü gösterecektir.

LLM API Çağrılarında Async'e İhtiyaç

LLM API'leriyle çalışırken, sıklıkla sırayla veya paralel olarak birden fazla API çağrısı yapmamız gereken senaryolarla karşılaşırız. Geleneksel senkron kod, özellikle LLM hizmetlerine yönelik ağ istekleri gibi yüksek gecikmeli işlemlerle uğraşırken önemli performans darboğazlarına yol açabilir.

Bir LLM API kullanarak 100 farklı makale için özetler üretmemiz gereken bir senaryoyu düşünün. Eşzamanlı bir yaklaşımla, her API çağrısı bir yanıt alana kadar bloke olur ve tüm istekleri tamamlamak potansiyel olarak birkaç dakika sürer. Öte yandan, eşzamansız bir yaklaşım, birden fazla API çağrısını eş zamanlı olarak başlatmamızı sağlayarak genel yürütme süresini önemli ölçüde azaltır.

Ortamınızı Ayarlama

Asenkron LLM API çağrılarına başlamak için Python ortamınızı gerekli kütüphanelerle kurmanız gerekir. İhtiyacınız olanlar şunlardır:

• Python 3.7 veya daha yüksek (yerel asyncio desteği için)
• aiohttp: Eşzamansız bir HTTP istemci kitaplığı
• openai: Resmi OpenAI Python istemcisi (OpenAI'nin GPT modellerini kullanıyorsanız)
• dil zinciri: LLM'ler ile uygulama oluşturmaya yönelik bir çerçeve (isteğe bağlı, ancak karmaşık iş akışları için önerilir)

Bu bağımlılıkları pip kullanarak kurabilirsiniz:

pip install aiohttp openai langchain
<div class="relative flex flex-col rounded-lg">

asyncio ve aiohttp ile Temel Async LLM API Çağrıları

aiohttp kullanarak bir LLM API'sine basit bir eşzamansız çağrı yaparak başlayalım. Örnek olarak OpenAI'nin GPT-3.5 API'sini kullanacağız, ancak kavramlar diğer LLM API'leri için de geçerlidir.

import asyncio
import aiohttp
from openai import AsyncOpenAI

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

async def main():
    prompts = [
        "Explain quantum computing in simple terms.",
        "Write a haiku about artificial intelligence.",
        "Describe the process of photosynthesis."
    ]
    
    async with AsyncOpenAI() as client:
        tasks = [generate_text(prompt, client) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    for prompt, result in zip(prompts, results):
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

Bu örnekte, asenkron bir fonksiyon tanımlıyoruz generate_text AsyncOpenAI istemcisini kullanarak OpenAI API'sine bir çağrı yapan main fonksiyon farklı istemler ve kullanımlar için birden fazla görev oluşturur asyncio.gather() bunları eş zamanlı olarak çalıştırmak için.

Bu yaklaşım, LLM API'sine aynı anda birden fazla istek göndermemize olanak tanır ve bu sayede tüm istemleri işlemek için gereken toplam süreyi önemli ölçüde azaltır.

Gelişmiş Teknikler: Toplu İşlem ve Eşzamanlılık Kontrolü

Önceki örnek, asenkron LLM API çağrılarının temellerini gösterse de, gerçek dünya uygulamaları genellikle daha karmaşık yaklaşımlar gerektirir. İki önemli tekniği inceleyelim: toplu istekler ve eşzamanlılığı kontrol etme.

Toplu İstekler: Çok sayıda istekle uğraşırken, her bir istek için ayrı ayrı istekler göndermek yerine, bunları gruplar halinde toplamak genellikle daha verimlidir. Bu, birden fazla API çağrısının getirdiği ek yükü azaltır ve daha iyi performansa yol açabilir.

import asyncio
from openai import AsyncOpenAI

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]

async def main():
    prompts = [f"Tell me a fact about number {i}" for i in range(100)]
    batch_size = 10
    
    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)
    
    for prompt, result in zip(prompts, results):
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

Eşzamanlılık Kontrolü: Eşzamansız programlama eş zamanlı yürütmeye olanak sağlasa da, API sunucusunun aşırı yüklenmesini veya hız sınırlarının aşılmasını önlemek için eşzamanlılık düzeyini kontrol etmek önemlidir. Bu amaçla asyncio.Semaphore'u kullanabiliriz.

import asyncio
from openai import AsyncOpenAI

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

async def main():
    prompts = [f"Tell me a fact about number {i}" for i in range(100)]
    max_concurrent_requests = 5
    semaphore = asyncio.Semaphore(max_concurrent_requests)
    
    async with AsyncOpenAI() as client:
        tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    for prompt, result in zip(prompts, results):
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

Bu örnekte, API sunucusunu aşırı yüklememek için eş zamanlı istek sayısını 5 ile sınırlamak amacıyla bir semafor kullanıyoruz.

Async LLM Çağrılarında Hata İşleme ve Yeniden Denemeler

Harici API'lerle çalışırken, güçlü hata işleme ve yeniden deneme mekanizmaları uygulamak çok önemlidir. Kodumuzu yaygın hataları işleyecek ve yeniden denemeler için üstel geri çekilme uygulayacak şekilde geliştirelim.

import asyncio
import random
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential

class APIError(Exception):
    pass

@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"Error occurred: {e}")
        raise APIError("Failed to generate text")

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, "Failed to generate response after multiple attempts."

async def main():
    prompts = [f"Tell me a fact about number {i}" for i in range(20)]
    max_concurrent_requests = 5
    semaphore = asyncio.Semaphore(max_concurrent_requests)
    
    async with AsyncOpenAI() as client:
        tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]
        results = await asyncio.gather(*tasks)
    
    for prompt, result in results:
        print(f"Prompt: {prompt}\nResponse: {result}\n")

asyncio.run(main())

Bu geliştirilmiş sürüm şunları içerir:

• Bir gelenek APIError API ile ilgili hatalar için istisna.
• A generate_text_with_retry işlevi ile dekore edilmiş @retry inatçılık kütüphanesinden, üstel geri çekilmeyi uygulayarak.
• Hata işleme process_prompt Arızaları yakalama ve raporlama fonksiyonu.

Performansı Optimize Etme: Akış Yanıtları

Uzun biçimli içerik üretimi için, akış yanıtları uygulamanızın algılanan performansını önemli ölçüde iyileştirebilir. Tüm yanıtı beklemek yerine, kullanılabilir hale geldikçe metin parçalarını işleyebilir ve görüntüleyebilirsiniz.

import asyncio
from openai import AsyncOpenAI

async def stream_text(prompt, client):
    stream = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    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)
    
    print("\n")
    return full_response

async def main():
    prompt = "Write a short story about a time-traveling scientist."
    
    async with AsyncOpenAI() as client:
        result = await stream_text(prompt, client)
    
    print(f"Full response:\n{result}")

asyncio.run(main())

Bu örnek, API'den gelen yanıtın nasıl akıtılacağını ve her parçanın geldiği gibi nasıl yazdırılacağını gösterir. Bu yaklaşım, özellikle sohbet uygulamaları veya kullanıcıya gerçek zamanlı geri bildirim sağlamak istediğiniz herhangi bir senaryo için faydalıdır.

LangChain ile Asenkron İş Akışları Oluşturma

Daha karmaşık LLM destekli uygulamalar için, LangChain çerçevesi Birden fazla LLM çağrısını zincirleme ve diğer araçları entegre etme sürecini basitleştiren üst düzey bir soyutlama sağlar. Eşzamansız yeteneklere sahip LangChain kullanımına bir örnek verelim:

Bu örnek, LangChain'in akış ve eşzamansız yürütme ile daha karmaşık iş akışları oluşturmak için nasıl kullanılabileceğini gösterir. AsyncCallbackManager ve StreamingStdOutCallbackHandler Üretilen içeriğin gerçek zamanlı akışını sağlamak.

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

async def generate_story(topic):
    llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
    prompt = PromptTemplate(
        input_variables=["topic"],
        template="Write a short story about {topic}."
    )
    chain = LLMChain(llm=llm, prompt=prompt)
    return await chain.arun(topic=topic)

async def main():
    topics = ["a magical forest", "a futuristic city", "an underwater civilization"]
    tasks = [generate_story(topic) for topic in topics]
    stories = await asyncio.gather(*tasks)
    
    for topic, story in zip(topics, stories):
        print(f"\nTopic: {topic}\nStory: {story}\n{'='*50}\n")

asyncio.run(main())

FastAPI ile Async LLM Uygulamalarına Hizmet Verme

Eşzamansız LLM uygulamanızı bir web servisi olarak kullanıma sunmak için FastAPI, eşzamansız işlemlere yönelik yerel desteği sayesinde harika bir seçimdir. İşte metin üretimi için basit bir API uç noktası oluşturma örneği:

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

class GenerationRequest(BaseModel):
    prompt: str

class GenerationResponse(BaseModel):
    generated_text: str

@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
    
    # Simulate some post-processing in the background
    background_tasks.add_task(log_generation, request.prompt, generated_text)
    
    return GenerationResponse(generated_text=generated_text)

async def log_generation(prompt: str, generated_text: str):
    # Simulate logging or additional processing
    await asyncio.sleep(2)
    print(f"Logged: Prompt '{prompt}' generated text of length {len(generated_text)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Bu FastAPI uygulaması bir uç nokta oluşturur /generate bir istemi kabul eden ve üretilen metni döndüren. Ayrıca, yanıtı engellemeden ek işleme için arka plan görevlerinin nasıl kullanılacağını da gösterir.

En İyi Uygulamalar ve Yaygın Tuzaklar

Asenkron LLM API'leriyle çalışırken şu en iyi uygulamaları aklınızda bulundurun:

1. Bağlantı havuzunu kullan: Birden fazla istekte bulunurken, yükü azaltmak için bağlantıları yeniden kullanın.
2. Uygun hata işlemeyi uygulayın: Ağ sorunlarını, API hatalarını ve beklenmeyen yanıtları her zaman hesaba katın.
3. Oran sınırlarına saygı gösterin: API'yi aşırı yüklememek için semaforları veya diğer eşzamanlılık kontrol mekanizmalarını kullanın.
4. İzle ve kaydet: Performansı izlemek ve sorunları belirlemek için kapsamlı günlük kaydı uygulayın.
5. Uzun biçimli içerik için akışı kullanın:Kullanıcı deneyimini iyileştirir ve kısmi sonuçların erken işlenmesine olanak tanır.