एआई मॉडल और प्लेटफ़ॉर्म
पाइथन में एसिंक्रोनस एलएलएम एपीआई कॉल: एक व्यापक गाइड
Unite.AI को Google पर अपने पसंदीदा स्रोतों में जोड़ें
विकासकर्ताओं और डेटा वैज्ञानिकों के रूप में, हम अक्सर इन शक्तिशाली मॉडलों के साथ एपीआई के माध्यम से बातचीत करने की आवश्यकता महसूस करते हैं। हालांकि, जब हमारे अनुप्रयोग जटिलता और पैमाने में बढ़ते हैं, तो कुशल और प्रदर्शनकारी एपीआई इंटरैक्शन की आवश्यकता महत्वपूर्ण हो जाती है। यहीं पर एसिंक्रोनस प्रोग्रामिंग चमकती है, जो हमें एलएलएम एपीआई के साथ काम करते समय थ्रूपुट को अधिकतम करने और लेटेंसी को कम करने की अनुमति देती है।
इस व्यापक गाइड में, हम पाइथन में एसिंक्रोनस एलएलएम एपीआई कॉल की दुनिया का अन्वेषण करेंगे। हम एसिंक्रोनस प्रोग्रामिंग की मूल बातों से लेकर जटिल कार्यों को संभालने के लिए उन्नत तकनीकों तक सब कुछ कवर करेंगे। इस लेख के अंत तक, आपको एसिंक्रोनस प्रोग्रामिंग का लाभ उठाकर अपने एलएलएम-संचालित अनुप्रयोगों को बढ़ावा देने की एक ठोस समझ होगी।
एसिंक्रोनस एलएलएम एपीआई कॉल के विशिष्ट विवरण में गोता लगाने से पहले, एसिंक्रोनस प्रोग्रामिंग की अवधारणाओं की एक ठोस नींव स्थापित करने दें।
एसिंक्रोनस प्रोग्रामिंग कई ऑपरेशनों को निष्पादित करने की अनुमति देती है जो निष्पादन के मुख्य धागे को ब्लॉक किए बिना समानांतर में चल सकते हैं। पाइथन में, यह मुख्य रूप से asyncio मॉड्यूल के माध्यम से प्राप्त किया जाता है, जो कोराउटीन, इवेंट लूप और फ्यूचर का उपयोग करके समांतर कोड लिखने के लिए एक फ्रेमवर्क प्रदान करता है।
मुख्य अवधारणाएं:
- कोराउटीन: async def के साथ परिभाषित किए गए फ़ंक्शन जो रोके और फिर से शुरू किए जा सकते हैं।
- इवेंट लूप: एसिंक्रोनस टास्क को प्रबंधित और चलाने के लिए केंद्रीय कार्यकारी तंत्र।
- प्रतीक्षा करने योग्य: await कीवर्ड के साथ उपयोग किए जाने वाले वस्तुएं (कोराउटीन, टास्क, फ्यूचर्स)
यहाँ इन अवधारणाओं को दर्शाने के लिए एक सरल उदाहरण दिया गया है:
import asyncio
<p>async def greet(name):</p>
<p>await asyncio.sleep(1) # एक आई/ओ ऑपरेशन का अनुकरण करें</p>
<p>print(f"नमस्ते, {name}!")</p>
<p>async def main():</p>
<p>await asyncio.gather(</p>
<p>greet("अलिस"),</p>
<p>greet("बॉब"),</p>
<p>greet("चार्ली")</p>
<p>)</p>
asyncio.run(main())
इस उदाहरण में, हम एक एसिंक्रोनस फ़ंक्शन greet परिभाषित करते हैं जो एक आई/ओ ऑपरेशन का अनुकरण करने के लिए asyncio.sleep() का उपयोग करता है। main फ़ंक्शन asyncio.gather() का उपयोग करके तीन नमस्कार समानांतर में चलाता है।尽管 प्रत्येक नमस्कार में 1 सेकंड की देरी है, सभी तीन नमस्कार लगभग 1 सेकंड के बाद ही प्रिंट हो जाएंगे, एसिंक्रोनस निष्पादन की शक्ति का प्रदर्शन करते हैं।
एलएलएम एपीआई कॉल में एसिंक्रोनस की आवश्यकता
जब हम एलएलएम एपीआई के साथ काम करते हैं, तो हम अक्सर ऐसे दृश्यों का सामना करते हैं जहां हमें एक ही समय में एक से अधिक एपीआई कॉल करने की आवश्यकता होती है, या तो क्रमिक रूप से या समानांतर में। पारंपरिक सिंक्रोनस कोड इस तरह के उच्च-लेटेंसी ऑपरेशन के साथ महत्वपूर्ण प्रदर्शन बोतलनेक का कारण बन सकता है, खासकर जब नेटवर्क अनुरोध जैसे एलएलएम सेवाओं के लिए जाते हैं।
एक दृश्य को ध्यान में रखें जहां हमें 100 अलग-अलग लेखों के लिए सारांश उत्पन्न करने के लिए एक एलएलएम एपीआई का उपयोग करने की आवश्यकता है। सिंक्रोनस दृष्टिकोण के साथ, प्रत्येक एपीआई कॉल पिछले एक के पूरा होने तक ब्लॉक करेगी, संभावित रूप से कई मिनट लगेगा। दूसरी ओर, एसिंक्रोनस दृष्टिकोण हमें एक ही समय में कई एपीआई कॉल शुरू करने की अनुमति देता है, जिससे कुल निष्पादन समय में काफी कमी आती है।
अपने वातावरण की स्थापना
एसिंक्रोनस एलएलएम एपीआई कॉल के साथ शुरू करने के लिए, आपको अपने पाइथन वातावरण को आवश्यक लाइब्रेरी के साथ सेट अप करने की आवश्यकता होगी। आपको निम्नलिखित की आवश्यकता होगी:
- पाइथन 3.7 या उच्चतर (मूल asyncio समर्थन के लिए)
- aiohttp: एक एसिंक्रोनस HTTP क्लाइंट लाइब्रेरी
- openai: आधिकारिक ओपनएआई पाइथन क्लाइंट (यदि आप ओपनएआई के जीपीटी मॉडल का उपयोग कर रहे हैं)
- langchain: एलएलएम के साथ अनुप्रयोग बनाने के लिए एक फ्रेमवर्क (वैकल्पिक, लेकिन जटिल कार्यों के लिए अनुशंसित)
आप पाइप का उपयोग करके इन निर्भरताओं को स्थापित कर सकते हैं:
pip install aiohttp openai langchain
asyncio और aiohttp के साथ बुनियादी एसिंक्रोनस एलएलएम एपीआई कॉल
आइए एक सरल एसिंक्रोनस कॉल के साथ शुरू करें एलएलएम एपीआई के लिए aiohttp का उपयोग करके। हम ओपनएआई के जीपीटी-3.5 एपीआई का उपयोग एक उदाहरण के रूप में करेंगे, लेकिन अवधारणाएं अन्य एलएलएम एपीआई पर भी लागू होती हैं।
import asyncio
import aiohttp
from openai import AsyncOpenAI
<p>async def generate_text(prompt, client):</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>async def main():</p>
<p>prompts = [</p>
<p>"सरल शब्दों में क्वांटम कंप्यूटिंग की व्याख्या करें।",</p>
<p>"कृत्रिम बुद्धिमत्ता के बारे में एक हाइकु लिखें।",</p>
<p>"प्रकाश संश्लेषण की प्रक्रिया का वर्णन करें।"</p>
<p>]</p>
<p>async with AsyncOpenAI() as client:</p>
<p>tasks = [generate_text(prompt, client) for prompt in prompts]</p>
<p>results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in zip(prompts, results):</p>
<p>print(f"प्रॉम्प्ट: {prompt}\nउत्तर: {result}\n")</p>
asyncio.run(main())
इस उदाहरण में, हम एक एसिंक्रोनस फ़ंक्शन generate_text परिभाषित करते हैं जो ओपनएआई एपीआई के लिए AsyncOpenAI क्लाइंट का उपयोग करके एक प्रॉम्प्ट के लिए पाठ उत्पन्न करता है। main फ़ंक्शन विभिन्न प्रॉम्प्ट के लिए कार्य बनाता है और उन्हें समानांतर में चलाने के लिए asyncio.gather() का उपयोग करता है।
यह दृष्टिकोण हमें एक ही समय में कई अनुरोध भेजने की अनुमति देता है, जिससे सभी प्रॉम्प्ट को संसाधित करने में कुल समय काफी कम हो जाता है।
उन्नत तकनीकें: बैचिंग और समांतर नियंत्रण
जबकि पिछला उदाहरण एसिंक्रोनस एलएलएम एपीआई कॉल की मूल बातों को प्रदर्शित करता है, वास्तविक दुनिया के अनुप्रयोग अक्सर अधिक परिष्कृत दृष्टिकोणों की आवश्यकता होती है। आइए बैचिंग और समांतर नियंत्रण जैसी दो महत्वपूर्ण तकनीकों का अन्वेषण करें।
बैचिंग अनुरोध: जब एक बड़ी संख्या में प्रॉम्प्ट के साथ काम किया जाता है, तो उन्हें समूहों में बैच करना अक्सर अधिक कुशल होता है, प्रत्येक प्रॉम्प्ट के लिए व्यक्तिगत अनुरोध भेजने के बजाय। यह कई एपीआई कॉल के ओवरहेड को कम करता है और बेहतर प्रदर्शन का कारण बन सकता है।
import asyncio
from openai import AsyncOpenAI
<p>async def process_batch(batch, client):</p>
<p>responses = await asyncio.gather(*[</p>
<p>client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>) for prompt in batch</p>
<p>])</p>
<p>return [response.choices[0].message.content for response in responses]</p>
<p>async def main():</p>
<p>prompts = [f"नंबर {i} के बारे में एक तथ्य बताएं।" for i in range(100)]</p>
<p>batch_size = 10</p>
<p>async with AsyncOpenAI() as client:</p>
<p>results = []</p>
<p>for i in range(0, len(prompts), batch_size):</p>
<p>batch = prompts[i:i+batch_size]</p>
<p>batch_results = await process_batch(batch, client)</p>
<p>results.extend(batch_results)</p>
<p>for prompt, result in zip(prompts, results):</p>
<p>print(f"प्रॉम्प्ट: {prompt}\nउत्तर: {result}\n")</p>
asyncio.run(main())
समांतर नियंत्रण: जबकि एसिंक्रोनस प्रोग्रामिंग समानांतर निष्पादन की अनुमति देती है, एपीआई सर्वर को अभिभूत करने या दर सीमा से अधिक नहीं जाने के लिए समांतरता के स्तर को नियंत्रित करना महत्वपूर्ण है। हम asyncio.Semaphore का उपयोग इस उद्देश्य के लिए कर सकते हैं।
import asyncio
from openai import AsyncOpenAI
<p>async def generate_text(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>async def main():</p>
<p>prompts = [f"नंबर {i} के बारे में एक तथ्य बताएं।" for i in range(100)]</p>
<p>max_concurrent_requests = 5</p>
<p>semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:</p>
<p>tasks = [generate_text(prompt, client, semaphore) for prompt in prompts]</p>
<p>results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in zip(prompts, results):</p>
<p>print(f"प्रॉम्प्ट: {prompt}\nउत्तर: {result}\n")</p>
asyncio.run(main())
इस उदाहरण में, हम एक सेमाफोर का उपयोग 5 तक समांतर अनुरोधों की संख्या को सीमित करने के लिए करते हैं, सुनिश्चित करते हुए कि हम एपीआई सर्वर को अभिभूत नहीं करते हैं।
एसिंक्रोनस एलएलएम कॉल में त्रुटि हैंडलिंग और पुनः प्रयास
बाहरी एपीआई के साथ काम करते समय, त्रुटि हैंडलिंग और पुनः प्रयास तंत्र को लागू करना महत्वपूर्ण है। आइए अपने कोड को सामान्य त्रुटियों को संभालने और पुनः प्रयास के लिए एक्सपोनेंशियल बैकऑफ़ को लागू करने के लिए बढ़ाएं।
import asyncio
import random
from openai import AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
<p>class APIError(Exception):</p>
<p>pass</p>
<p>@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))</p>
<p>async def generate_text_with_retry(prompt, client):</p>
<p>try:</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}]</p>
<p>)</p>
<p>return response.choices[0].message.content</p>
<p>except Exception as e:</p>
<p>print(f"त्रुटि हुई: {e}")</p>
<p>raise APIError("पाठ उत्पन्न करने में विफल")</p>
<p>async def process_prompt(prompt, client, semaphore):</p>
<p>async with semaphore:</p>
<p>try:</p>
<p>result = await generate_text_with_retry(prompt, client)</p>
<p>return prompt, result</p>
<p>except APIError:</p>
<p>return prompt, "पुनः प्रयास के बाद भी उत्तर उत्पन्न करने में विफल।"</p>
<p>async def main():</p>
<p>prompts = [f"नंबर {i} के बारे में एक तथ्य बताएं।" for i in range(20)]</p>
<p>max_concurrent_requests = 5</p>
<p>semaphore = asyncio.Semaphore(max_concurrent_requests)</p>
<p>async with AsyncOpenAI() as client:</p>
<p>tasks = [process_prompt(prompt, client, semaphore) for prompt in prompts]</p>
<p>results = await asyncio.gather(*tasks)</p>
<p>for prompt, result in results:</p>
<p>print(f"प्रॉम्प्ट: {prompt}\nउत्तर: {result}\n")</p>
asyncio.run(main())
इस बढ़ाए गए संस्करण में शामिल हैं:
- एक कस्टम
APIErrorअपवाद एपीआई से संबंधित त्रुटियों के लिए। - एक
generate_text_with_retryफ़ंक्शन जो tenacity लाइब्रेरी से@retryके साथ अलंकृत है, जो पुनः प्रयास के लिए एक्सपोनेंशियल बैकऑफ़ लागू करता है। - त्रुटि हैंडलिंग
process_promptफ़ंक्शन में विफलताओं को पकड़ने और रिपोर्ट करने के लिए।
प्रदर्शन का अनुकूलन: स्ट्रीमिंग प्रतिक्रियाएं
लंबे फॉर्म सामग्री उत्पादन के लिए, प्रतिक्रियाओं को स्ट्रीमिंग करना आपके अनुप्रयोग के प्रतिपादन प्रदर्शन में काफी सुधार कर सकता है। पूरी प्रतिक्रिया की प्रतीक्षा किए बिना, आप प्राप्त होते ही पाठ के टुकड़ों को संसाधित और प्रदर्शित कर सकते हैं।
import asyncio
from openai import AsyncOpenAI
<p>async def stream_text(prompt, client):</p>
<p>stream = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": prompt}],</p>
<p>stream=True</p>
<p>)</p>
<p>full_response = ""</p>
<p>async for chunk in stream:</p>
<p>if chunk.choices[0].delta.content is not None:</p>
<p>content = chunk.choices[0].delta.content</p>
<p>full_response += content</p>
<p>print(content, end="", flush=True)</p>
<p>print("\n")</p>
<p>return full_response</p>
<p>async def main():</p>
<p>prompt = "एक समय-यात्रा करने वाले वैज्ञानिक के बारे में एक छोटी कहानी लिखें।"</p>
<p>async with AsyncOpenAI() as client:</p>
<p>result = await stream_text(prompt, client)</p>
<p>print(f"पूरी प्रतिक्रिया:\n{result}")</p>
asyncio.run(main())
इस उदाहरण में, हम एपीआई से प्रतिक्रिया को स्ट्रीमिंग करने के लिए दिखाते हैं, जैसे ही प्रत्येक टुकड़ा आता है उसे प्रिंट करते हैं। यह दृष्टिकोण विशेष रूप से चैट अनुप्रयोगों या किसी भी दृश्य के लिए उपयुक्त है जहां आप उपयोगकर्ता को वास्तविक समय में प्रतिक्रिया प्रदान करना चाहते हैं।
LangChain के साथ एसिंक्रोनस वर्कफ़्लो बनाना
अधिक जटिल एलएलएम-संचालित अनुप्रयोगों के लिए, LangChain फ्रेमवर्क एक उच्च-स्तरीय अभिव्यक्ति प्रदान करता है जो एलएलएम कॉल को चेन करने और अन्य टूल को एकीकृत करने की प्रक्रिया को सरल बनाता है। आइए LangChain के साथ एसिंक्रोनस क्षमताओं का उपयोग करने का एक उदाहरण देखें:
यह उदाहरण दिखाता है कि कैसे 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
<p>async def generate_story(topic):</p>
<p>llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))</p>
<p>prompt = PromptTemplate(</p>
<p>input_variables=["topic"],</p>
<p>template="विषय {topic} के बारे में एक छोटी कहानी लिखें।"</p>
<p>)</p>
<p>chain = LLMChain(llm=llm, prompt=prompt)</p>
<p>return await chain.arun(topic=topic)</p>
<p>async def main():</p>
<p>topics = ["एक जादुई वन", "एक भविष्यवाणी शहर", "एक जलमग्न सभ्यता"]</p>
<p>tasks = [generate_story(topic) for topic in topics]</p>
<p>stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):</p>
<p>print(f"\nविषय: {topic}\nकहानी: {story}\n{'='*50}\n")</p>
asyncio.run(main())
फास्टएपीआई के साथ एसिंक्रोनस एलएलएम एप्लिकेशन की सेवा
अपने एसिंक्रोनस एलएलएम एप्लिकेशन को वेब सेवा के रूप में उपलब्ध कराने के लिए, फास्टएपीआई एक उत्कृष्ट विकल्प है क्योंकि यह एसिंक्रोनस ऑपरेशन के लिए मूल समर्थन प्रदान करता है। आइए एक सरल एपीआई एंडपॉइंट बनाने का एक उदाहरण देखें पाठ उत्पादन के लिए:
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from openai import AsyncOpenAI
app = FastAPI()
client = AsyncOpenAI()
<p>class GenerationRequest(BaseModel):</p>
<p>prompt: str</p>
<p>class GenerationResponse(BaseModel):</p>
<p>generated_text: str</p>
<p>@app.post("/generate", response_model=GenerationResponse)</p>
<p>async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):</p>
<p>response = await client.chat.completions.create(</p>
<p>model="gpt-3.5-turbo",</p>
<p>messages=[{"role": "user", "content": request.prompt}]</p>
<p>)</p>
<p>generated_text = response.choices[0].message.content</p>
<p># पोस्ट-प्रोसेसिंग का अनुकरण करने के लिए पृष्ठभूमि में कुछ कार्य जोड़ें</p>
<p>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):</p>
<p># लॉगिंग या अतिरिक्त प्रसंस्करण का अनुकरण करें</p>
<p>await asyncio.sleep(2)</p>
<p>print(f"लॉग किया गया: प्रॉम्प्ट '{prompt}' ने {len(generated_text)} लंबाई का पाठ उत्पन्न किया।")</p>
<p>if __name__ == "__main__":</p>
<p>import uvicorn</p>
<p>uvicorn.run(app, host="0.0.0.0", port=8000)</p>
यह फास्टएपीआई अनुप्रयोग /generate नामक एक एंडपॉइंट बनाता है जो एक प्रॉम्प्ट स्वीकार करता है और उत्पन्न पाठ लौटाता है। यह पृष्ठभूमि कार्यों का भी प्रदर्शन करता है, जो अतिरिक्त प्रसंस्करण के लिए उपयोगी हो सकते हैं बिना प्रतिक्रिया को ब्लॉक किए।
सर्वोत्तम अभ्यास और सामान्य गलतियाँ
एसिंक्रोनस एलएलएम एपीआई के साथ काम करते समय, इन सर्वोत्तम अभ्यासों को ध्यान में रखें:
- कनेक्शन पूलिंग का उपयोग करें: एक से अधिक अनुरोध करते समय, कनेक्शन को पुन: उपयोग करने से ओवरहेड कम होता है।
- उचित त्रुटि हैंडलिंग लागू करें: नेटवर्क मुद्दों, एपीआई त्रुटियों और अप्रत्याशित प्रतिक्रियाओं के लिए खाता बनाएं।
- दर सीमा का सम्मान करें: सेमाफोर या अन्य समांतर नियंत्रण तंत्र का उपयोग करके एपीआई सर्वर को अभिभूत करने से बचें।
- निगरानी और लॉगिंग करें: प्रदर्शन को ट्रैक करने और मुद्दों की पहचान करने के लिए व्यापक लॉगिंग लागू करें।
- लंबे फॉर्म सामग्री के लिए स्ट्रीमिंग का उपयोग करें: यह उपयोगकर्ता अनुभव में सुधार करता है और आंशिक परिणामों की प्रारंभिक प्रसंस्करण की अनुमति देता है।
