به‌عنوان توسعه‌دهندگان و دانشمندان dta، اغلب نیاز به تعامل با این مدل‌های قدرتمند از طریق APIها داریم. با این حال، با افزایش پیچیدگی و مقیاس برنامه های ما، نیاز به تعاملات API کارآمد و کارآمد بسیار مهم می شود. اینجاست که برنامه‌نویسی ناهمزمان می‌درخشد و به ما امکان می‌دهد هنگام کار با APIهای LLM، توان عملیاتی را به حداکثر برسانیم و تأخیر را به حداقل برسانیم.
در این راهنمای جامع، دنیای فراخوانی های ناهمزمان LLM API در پایتون را بررسی خواهیم کرد. ما همه چیز را از اصول برنامه‌نویسی ناهمزمان تا تکنیک‌های پیشرفته برای مدیریت گردش‌های کاری پیچیده پوشش خواهیم داد. در پایان این مقاله، شما درک کاملی از نحوه استفاده از برنامه‌نویسی ناهمزمان برای افزایش شارژ برنامه‌های مبتنی بر LLM خواهید داشت.
قبل از اینکه به جزئیات فراخوانی های async LLM API بپردازیم، اجازه دهید یک پایه محکم در مفاهیم برنامه نویسی ناهمزمان ایجاد کنیم.
برنامه نویسی ناهمزمان اجازه می دهد تا چندین عملیات به طور همزمان بدون مسدود کردن رشته اصلی اجرا اجرا شوند. در پایتون، این در درجه اول از طریق به دست می آید asyncio ماژول، که چارچوبی را برای نوشتن کدهای همزمان با استفاده از کوروتین ها، حلقه های رویداد و آتی فراهم می کند.
مفاهیم کلیدی:
- کوروتین ها: توابع تعریف شده با async def که می توان مکث کرد و از سر گرفت.
- حلقه رویداد: مکانیزم اجرایی مرکزی که وظایف ناهمزمان را مدیریت و اجرا می کند.
- قابل انتظار: اشیایی که می توان با کلمه کلیدی 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 ثانیه چاپ می‌شوند که قدرت اجرای ناهمزمان را نشان می‌دهد.
نیاز به Async در تماس های API LLM
هنگام کار با API های LLM، اغلب با سناریوهایی مواجه می شویم که در آن باید چندین تماس API را به صورت متوالی یا موازی انجام دهیم. کد سنکرون سنتی می‌تواند منجر به گلوگاه‌های عملکردی قابل توجهی شود، به‌ویژه زمانی که با عملیات‌هایی با تأخیر بالا مانند درخواست‌های شبکه برای سرویس‌های LLM سروکار داریم.
سناریویی را در نظر بگیرید که در آن باید با استفاده از یک API LLM، برای 100 مقاله مختلف خلاصه تولید کنیم. با یک رویکرد همزمان، هر تماس API تا زمانی که پاسخی را دریافت نکند مسدود می‌شود و به طور بالقوه چندین دقیقه طول می‌کشد تا تمام درخواست‌ها تکمیل شوند. از سوی دیگر، یک رویکرد ناهمزمان به ما امکان می‌دهد تا چندین تماس API را همزمان آغاز کنیم و زمان اجرای کلی را به‌طور چشمگیری کاهش دهیم.
تنظیم محیط
برای شروع تماس‌های async LLM API، باید محیط پایتون خود را با کتابخانه‌های لازم راه‌اندازی کنید. در اینجا چیزی است که شما نیاز دارید:
- پایتون 3.7 یا بالاتر (برای پشتیبانی بومی asyncio)
- aiohttp: یک کتابخانه سرویس گیرنده HTTP ناهمزمان
- openai: مسئول کلاینت OpenAI Python (اگر از مدل های GPT OpenAI استفاده می کنید)
- زنجیره ای: چارچوبی برای ساخت برنامه های کاربردی با LLM (اختیاری، اما برای گردش های کاری پیچیده توصیه می شود)
می توانید این وابستگی ها را با استفاده از pip نصب کنید:
pip install aiohttp openai langchainBasic Async LLM API Calls with asyncio and aiohttp
Let's start by making a simple asynchronous call to an LLM API using aiohttp. We'll use OpenAI's GPT-3.5 API as an example, but the concepts apply to other LLM APIs as well.
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()برای اجرای همزمان آنهااین رویکرد به ما امکان می دهد چندین درخواست را به طور همزمان به API LLM ارسال کنیم و کل زمان مورد نیاز برای پردازش همه درخواست ها را به میزان قابل توجهی کاهش دهیم.
تکنیک های پیشرفته: دسته بندی و کنترل همزمان
در حالی که مثال قبلی اصول اولیه فراخوانی های API LLM را نشان می دهد، برنامه های کاربردی دنیای واقعی اغلب به رویکردهای پیچیده تری نیاز دارند. بیایید دو تکنیک مهم را بررسی کنیم: دسته بندی درخواست ها و کنترل همزمانی.
درخواست‌های دسته‌ای: وقتی با تعداد زیادی درخواست سروکار داریم، اغلب کارآمدتر است که آنها را به گروه‌ها دسته بندی کنیم تا اینکه درخواست‌های جداگانه برای هر درخواست ارسال کنیم. این امر سربار تماس های متعدد 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 را تحت تأثیر قرار نمی‌دهیم.
مدیریت خطا و تلاش مجدد در تماس های Async 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استثنا برای خطاهای مربوط به API.- الف
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 پخش کرد و هر تکه را به محض رسیدن چاپ کرد. این رویکرد به ویژه برای برنامه های چت یا هر سناریویی که می خواهید بازخورد بلادرنگ را به کاربر ارائه دهید مفید است.
ایجاد گردش کار Async با LangChain
برای برنامه های پیچیده تر مبتنی بر LLM، چارچوب LangChain یک انتزاع سطح بالا را ارائه می دهد که فرآیند زنجیره ای کردن چندین تماس LLM و یکپارچه سازی ابزارهای دیگر را ساده می کند. بیایید به مثالی از استفاده از LangChain با قابلیت‌های 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())ارائه برنامه های Async LLM با FastAPI
برای اینکه برنامه async 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که یک اعلان را می پذیرد و متن تولید شده را برمی گرداند. همچنین نشان می دهد که چگونه از وظایف پس زمینه برای پردازش اضافی بدون مسدود کردن پاسخ استفاده کنید.بهترین شیوه ها و دام های رایج
همانطور که با API های async LLM کار می کنید، این بهترین شیوه ها را در نظر داشته باشید:
- از ادغام اتصال استفاده کنید: هنگام درخواست چندگانه، از اتصالات برای کاهش هزینه های اضافی استفاده مجدد کنید.
- مدیریت صحیح خطا را اجرا کنید: همیشه مشکلات شبکه، خطاهای API و پاسخ های غیرمنتظره را در نظر بگیرید.
- به محدودیت های نرخ احترام بگذارید: از سمافورها یا سایر مکانیسم‌های کنترل همزمانی برای جلوگیری از تحت تأثیر قرار دادن API استفاده کنید.
- نظارت و ثبت نام کنید: برای ردیابی عملکرد و شناسایی مشکلات، گزارش جامع را اجرا کنید.
- از استریم برای محتوای طولانی استفاده کنید: تجربه کاربر را بهبود می بخشد و امکان پردازش زودهنگام نتایج جزئی را فراهم می کند.
