पायथन में एसिंक्रोनस एलएलएम एपीआई कॉल: एक व्यापक गाइड

डेवलपर्स और dta वैज्ञानिकों के रूप में, हम अक्सर खुद को API के माध्यम से इन शक्तिशाली मॉडलों के साथ बातचीत करने की आवश्यकता महसूस करते हैं। हालाँकि, जैसे-जैसे हमारे अनुप्रयोग जटिलता और पैमाने में बढ़ते हैं, कुशल और प्रदर्शनकारी API इंटरैक्शन की आवश्यकता महत्वपूर्ण हो जाती है। यह वह जगह है जहाँ एसिंक्रोनस प्रोग्रामिंग चमकती है, जिससे हमें LLM API के साथ काम करते समय थ्रूपुट को अधिकतम करने और विलंबता को कम करने की अनुमति मिलती है।

इस विस्तृत गाइड में, हम पायथन में एसिंक्रोनस LLM API कॉल्स की दुनिया का अन्वेषण करेंगे। हम एसिंक्रोनस प्रोग्रामिंग की मूल बातों से लेकर जटिल वर्कफ़्लोज़ को संभालने की उन्नत तकनीकों तक, सब कुछ कवर करेंगे। इस लेख के अंत तक, आपको अपने LLM-संचालित अनुप्रयोगों को सुपरचार्ज करने के लिए एसिंक्रोनस प्रोग्रामिंग का लाभ उठाने की ठोस समझ हो जाएगी।

इससे पहले कि हम एसिंक्रोनस एलएलएम एपीआई कॉल की बारीकियों में उतरें, आइए एसिंक्रोनस प्रोग्रामिंग अवधारणाओं में एक ठोस आधार स्थापित करें।

एसिंक्रोनस प्रोग्रामिंग निष्पादन के मुख्य थ्रेड को ब्लॉक किए बिना कई ऑपरेशनों को एक साथ निष्पादित करने की अनुमति देता है। पायथन में, यह मुख्य रूप से इसके माध्यम से प्राप्त किया जाता है asyncio मॉड्यूल, जो कोरोउटिन, इवेंट लूप और फ्यूचर्स का उपयोग करके समवर्ती कोड लिखने के लिए एक ढांचा प्रदान करता है।

महत्वपूर्ण अवधारणाएं:

• Coroutines: के साथ परिभाषित फ़ंक्शन एसिंक्रोनस डेफ़ जिसे रोका और पुनः शुरू किया जा सकता है।
• इवेंट लूप: केंद्रीय निष्पादन तंत्र जो अतुल्यकालिक कार्यों का प्रबंधन और संचालन करता है।
• प्रतीक्षा योग्य: ऑब्जेक्ट्स जिनका उपयोग await कीवर्ड के साथ किया जा सकता है (कोरोउटीन, टास्क, फ्यूचर्स)।

इन अवधारणाओं को स्पष्ट करने के लिए यहां एक सरल उदाहरण दिया गया है:

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())

इस उदाहरण में, हम एक एसिंक्रोनस फ़ंक्शन परिभाषित करते हैं greet जो I/O ऑपरेशन का अनुकरण करता है asyncio.sleep()। main फ़ंक्शन का उपयोग करता है asyncio.gather() एक साथ कई ग्रीटिंग्स चलाने के लिए। स्लीप डिले के बावजूद, सभी तीन ग्रीटिंग्स लगभग 1 सेकंड के बाद प्रिंट हो जाएंगे, जो एसिंक्रोनस निष्पादन की शक्ति को प्रदर्शित करता है।

LLM API कॉल में Async की आवश्यकता

LLM API के साथ काम करते समय, हम अक्सर ऐसे परिदृश्यों का सामना करते हैं जहाँ हमें अनुक्रम में या समानांतर रूप से कई API कॉल करने की आवश्यकता होती है। पारंपरिक सिंक्रोनस कोड महत्वपूर्ण प्रदर्शन बाधाओं को जन्म दे सकता है, खासकर जब LLM सेवाओं के लिए नेटवर्क अनुरोधों जैसे उच्च-विलंबता संचालन से निपटना हो।

एक परिदृश्य पर विचार करें जहां हमें LLM API का उपयोग करके 100 अलग-अलग लेखों के लिए सारांश तैयार करने की आवश्यकता है। एक समकालिक दृष्टिकोण के साथ, प्रत्येक API कॉल तब तक अवरुद्ध रहेगा जब तक कि उसे कोई प्रतिक्रिया प्राप्त न हो जाए, संभावित रूप से सभी अनुरोधों को पूरा करने में कई मिनट लग सकते हैं। दूसरी ओर, एक अतुल्यकालिक दृष्टिकोण हमें एक साथ कई API कॉल आरंभ करने की अनुमति देता है, जिससे समग्र निष्पादन समय में नाटकीय रूप से कमी आती है।

अपना पर्यावरण स्थापित करना

एसिंक्रोनस LLM API कॉल शुरू करने के लिए, आपको अपने पायथन वातावरण को आवश्यक लाइब्रेरीज़ के साथ सेट अप करना होगा। आपको इन चीज़ों की ज़रूरत होगी:

• अजगर 3.7 या उच्चतर (देशी asyncio समर्थन के लिए)
• aiohttp: एक एसिंक्रोनस HTTP क्लाइंट लाइब्रेरी
• openai: आधिकारिक ओपनएआई पायथन क्लाइंट (यदि आप OpenAI के GPT मॉडल का उपयोग कर रहे हैं)
• लैंगचैन: एलएलएम के साथ अनुप्रयोग बनाने के लिए एक ढांचा (वैकल्पिक, लेकिन जटिल वर्कफ़्लो के लिए अनुशंसित)

आप pip का उपयोग करके इन निर्भरताओं को स्थापित कर सकते हैं:

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

asyncio और aiohttp के साथ बेसिक Async LLM API कॉल

आइए aiohttp का उपयोग करके एक LLM API पर एक सरल एसिंक्रोनस कॉल करके शुरुआत करें। हम उदाहरण के तौर पर OpenAI के GPT-3.5 API का उपयोग करेंगे, लेकिन ये अवधारणाएँ अन्य LLM API पर भी लागू होती हैं।

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())

इस उदाहरण में, हम एक एसिंक्रोनस फ़ंक्शन परिभाषित करते हैं generate_text जो AsyncOpenAI क्लाइंट का उपयोग करके OpenAI API को कॉल करता है। main फ़ंक्शन विभिन्न प्रॉम्प्ट और उपयोगों के लिए कई कार्य बनाता है asyncio.gather() उन्हें एक साथ चलाने के लिए।

यह दृष्टिकोण हमें LLM API को एक साथ कई अनुरोध भेजने की अनुमति देता है, जिससे सभी संकेतों को संसाधित करने के लिए आवश्यक कुल समय में काफी कमी आती है।

उन्नत तकनीकें: बैचिंग और समवर्ती नियंत्रण

हालाँकि पिछला उदाहरण एसिंक्रोनस LLM API कॉल की मूल बातें दर्शाता है, लेकिन वास्तविक अनुप्रयोगों में अक्सर अधिक परिष्कृत तरीकों की आवश्यकता होती है। आइए दो महत्वपूर्ण तकनीकों पर गौर करें: बैचिंग अनुरोध और समवर्तीता नियंत्रण।

बैचिंग अनुरोध: बड़ी संख्या में प्रॉम्प्ट से निपटने के लिए, प्रत्येक प्रॉम्प्ट के लिए अलग-अलग अनुरोध भेजने के बजाय, उन्हें समूहों में बैच करना अक्सर अधिक कुशल होता है। इससे कई API कॉल का ओवरहेड कम हो जाता है और बेहतर प्रदर्शन प्राप्त हो सकता है।

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())

समवर्ती नियंत्रण: हालाँकि अतुल्यकालिक प्रोग्रामिंग समवर्ती निष्पादन की अनुमति देती है, लेकिन API सर्वर पर अत्यधिक भार या गति सीमा से अधिक होने से बचने के लिए समवर्तीता के स्तर को नियंत्रित करना महत्वपूर्ण है। हम इस उद्देश्य के लिए asyncio.Semaphore का उपयोग कर सकते हैं।

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())

इस उदाहरण में, हम समवर्ती अनुरोधों की संख्या को 5 तक सीमित करने के लिए सेमाफोर का उपयोग करते हैं, जिससे यह सुनिश्चित होता है कि हम API सर्वर पर अधिक भार न डालें।

एसिंक्रोनस LLM कॉल में त्रुटि प्रबंधन और पुनः प्रयास

बाहरी API के साथ काम करते समय, मज़बूत त्रुटि प्रबंधन और पुनः प्रयास तंत्र लागू करना बेहद ज़रूरी है। आइए, सामान्य त्रुटियों को संभालने और पुनः प्रयासों के लिए घातीय बैकऑफ़ लागू करने के लिए अपने कोड को बेहतर बनाएँ।

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())

इस उन्नत संस्करण में शामिल हैं:

• एक प्रथा APIError एपीआई-संबंधित त्रुटियों के लिए अपवाद.
• A generate_text_with_retry समारोह की सजावट @retry टेनेसिटी लाइब्रेरी से, एक्सपोनेंशियल बैकऑफ़ को कार्यान्वित करना।
• त्रुटि प्रबंधन process_prompt विफलताओं को पकड़ने और रिपोर्ट करने का कार्य।

प्रदर्शन को अनुकूलित करना: स्ट्रीमिंग प्रतिक्रियाएँ

लंबे-फ़ॉर्म कंटेंट जनरेशन के लिए, स्ट्रीमिंग रिस्पॉन्स आपके एप्लिकेशन के कथित प्रदर्शन को काफ़ी हद तक बेहतर बना सकते हैं। संपूर्ण प्रतिक्रिया की प्रतीक्षा करने के बजाय, आप टेक्स्ट के कुछ हिस्सों को प्रोसेस करके उन्हें उपलब्ध होते ही प्रदर्शित कर सकते हैं।

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())

यह उदाहरण दर्शाता है कि API से प्रतिक्रिया को कैसे स्ट्रीम किया जाए, प्रत्येक खंड को प्रिंट करते समय। यह दृष्टिकोण चैट अनुप्रयोगों या किसी भी परिदृश्य के लिए विशेष रूप से उपयोगी है जहाँ आप उपयोगकर्ता को वास्तविक समय की प्रतिक्रिया प्रदान करना चाहते हैं।

LangChain के साथ Async वर्कफ़्लोज़ का निर्माण

अधिक जटिल एलएलएम-संचालित अनुप्रयोगों के लिए, लैंगचेन ढांचा एक उच्च-स्तरीय अमूर्तता प्रदान करता है जो कई LLM कॉल्स को श्रृंखलाबद्ध करने और अन्य उपकरणों को एकीकृत करने की प्रक्रिया को सरल बनाता है। आइए, async क्षमताओं के साथ LangChain के उपयोग का एक उदाहरण देखें:

यह उदाहरण दिखाता है कि स्ट्रीमिंग और एसिंक्रोनस निष्पादन के साथ अधिक जटिल वर्कफ़्लो बनाने के लिए लैंगचेन का उपयोग कैसे किया जा सकता है। AsyncCallbackManager और StreamingStdOutCallbackHandler उत्पन्न सामग्री की वास्तविक समय स्ट्रीमिंग सक्षम करें।

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 के साथ Async LLM अनुप्रयोगों की सेवा करना

अपने एसिंक्रोनस LLM एप्लिकेशन को वेब सेवा के रूप में उपलब्ध कराने के लिए, FastAPI एक बेहतरीन विकल्प है क्योंकि यह एसिंक्रोनस ऑपरेशनों के लिए मूल रूप से समर्थित है। टेक्स्ट जनरेशन के लिए एक सरल API एंडपॉइंट बनाने का एक उदाहरण यहां दिया गया है:

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)

यह FastAPI एप्लिकेशन एक एंडपॉइंट बनाता है /generate जो एक प्रॉम्प्ट स्वीकार करता है और जेनरेटेड टेक्स्ट लौटाता है। यह यह भी दर्शाता है कि प्रतिक्रिया को अवरुद्ध किए बिना अतिरिक्त प्रसंस्करण के लिए पृष्ठभूमि कार्यों का उपयोग कैसे किया जाए।

सर्वोत्तम प्रथाएँ और सामान्य ख़तरे

जब आप एसिंक्रोनस LLM API के साथ काम करते हैं, तो इन सर्वोत्तम प्रथाओं को ध्यान में रखें:

1. कनेक्शन पूलिंग का उपयोग करेंएकाधिक अनुरोध करते समय, ओवरहेड को कम करने के लिए कनेक्शन का पुनः उपयोग करें।
2. उचित त्रुटि प्रबंधन लागू करें: नेटवर्क समस्याओं, API त्रुटियों और अप्रत्याशित प्रतिक्रियाओं का हमेशा ध्यान रखें।
3. दर सीमा का सम्मान करें: API पर अत्यधिक दबाव से बचने के लिए सेमाफोर या अन्य समवर्ती नियंत्रण तंत्र का उपयोग करें।
4. मॉनिटर करें और लॉग करें: प्रदर्शन को ट्रैक करने और समस्याओं की पहचान करने के लिए व्यापक लॉगिंग लागू करें।
5. लंबे प्रारूप वाली सामग्री के लिए स्ट्रीमिंग का उपयोग करें: यह उपयोगकर्ता अनुभव को बेहतर बनाता है और आंशिक परिणामों के शीघ्र प्रसंस्करण की अनुमति देता है।