المكالمات المتزامنة لبروتوكول LLM API في بايثون: دليل شامل

كمتطويرين وعلوميي البيانات ، غالبًا ما نجد أنفسنا في حاجة إلى التفاعل مع هذه النماذج القوية من خلال واجهات برمجة التطبيقات. ومع ذلك ،随ما تزداد تطبيقاتنا تعقيدًا وتنمو ، يصبح الحاجة إلى تفاعلات واجهة برمجة التطبيقات الكفؤة والمؤثرة حاسمة. यह هو المكان الذي يبرز فيه البرمجة المتزامنة ، مما يسمح لنا بتعظيم الإنتاجية وتقليل التأخير عند العمل مع واجهات برمجة التطبيقات LLM.

في هذا الدليل الشامل ، سنستكشف عالم المكالمات المتزامنة لبروتوكول LLM API في بايثون. سنغطي كل شيء من أساسيات البرمجة المتزامنة إلى تقنيات متقدمة لتعامل السلاسل العاملة المعقدة. بنهاية هذا المقال ، سيكون لديك فهم صلب ل كيفية استغلال البرمجة المتزامنة لتعزيز تطبيقاتك التي تعمل باللمس LLM.

قبل أن ندخل في تفاصيل المكالمات المتزامنة لبروتوكول LLM API ، دعونا نؤسس أساسًا صلبًا في مفاهيم البرمجة المتزامنة.

البرمجة المتزامنة تسمح بتنفيذ عمليات متعددة بشكل متوازي دون حجب الخيط الرئيسي للتنفيذ. في بايثون ، يتم تحقيق ذلك بشكل رئيسي من خلال وحدة asyncio ، التي توفر إطارًا لكتابة رمز متوازي باستخدام الروتينات ، حلقة الأحداث ، والمستقبل.

المفاهيم الرئيسية:

• الروتينات: الوظائف المحددة ب async def يمكن إيقافها واستئنافها.
• حلقة الأحداث: الآلية الرئيسية للتنفيذ التي تدير وتشغل المهام المتزامنة.
• النواتج: الكائنات التي يمكن استخدامها مع كلمة await (الروتينات ، المهام ، المستقبل).

هنا مثال بسيط لتوضيح هذه المفاهيم:

import asyncio

<p>async def greet(name):
await asyncio.sleep(1) #模拟 عملية I/O
print(f"Hello, {name}!")</p>

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

asyncio.run(main())

في هذا المثال ، نحدد دالة متزامنة greet التي ت模يل عملية I/O باستخدام asyncio.sleep(). الدالة main تستخدم asyncio.gather() لتشغيل تحيات متعددة بشكل متوازي.

هذا النهج يسمح لنا بإرسال عدة طلبات إلى واجهة برمجة التطبيقات LLM في نفس الوقت ، مما يقلل بشكل كبير من الوقت الإجمالي المطلوب لمعالجة جميع التحفيزات.

الحاجة إلى المكالمات المتزامنة لبروتوكول LLM API

عند العمل مع واجهات برمجة التطبيقات LLM ، غالبًا ما نواجه سيناريوهات حيث نحتاج إلى إجراء مكالمات متعددة ، إما بشكل متتالي أو متوازي. يمكن أن يؤدي الرمز المتزامن التقليدي إلى عرقلة كبيرة في الأداء ، خاصة عند التعامل مع عمليات منخفضة الكفاءة مثل طلبات الشبكة إلى خدمات LLM.

افترض أننا بحاجة إلى إنشاء ملخصات ل 100 مقال مختلف باستخدام واجهة برمجة التطبيقات LLM. باستخدام نهج متزامن ، سوف تحجز كل مكالمة واجهة برمجة التطبيقات حتى تتلقى استجابة ، مما قد يستغرق بضع دقائق لإكمال جميع الطلبات. من ناحية أخرى ، يسمح النهج المتزامن لنا ببدء مكالمات واجهة برمجة التطبيقات المتعددة بشكل متوازي ، مما يقلل بشكل كبير من وقت التنفيذ الإجمالي.

إعداد بيئتك

لبدء العمل مع مكالمات واجهة برمجة التطبيقات LLM المتزامنة ، ستحتاج إلى إعداد بيئة بايثون الخاصة بك مع المكتبات الضرورية. هنا ما ستحتاج إليه:

• بايثون 3.7 أو أعلى (لدعم متزامن أصلي)
• aiohttp: عميل HTTP متزامن
• openai: عميل OpenAI الرسمي (إذا كنت تستخدم نماذج GPT من OpenAI)
• langchain: إطار لإنشاء تطبيقات مع LLMs (اختياري ، ولكن موصى به لسلاسل العمل المعقدة)

يمكنك تثبيت هذه الإعتماديات باستخدام pip:

<p>pip install aiohttp openai langchain
&lt;div class=&quot;relative flex flex-col rounded-lg&quot;&gt;

مكالمات واجهة برمجة التطبيقات LLM الأساسية المتزامنة مع asyncio و aiohttp

دعونا نبدأ بإنشاء مكالمة واجهة برمجة التطبيقات LLM متزامنة بسيطة باستخدام aiohttp. سنستخدم واجهة برمجة التطبيقات GPT-3.5 من OpenAI كمثال ، ولكن المفاهيم تنطبق على واجهات برمجة التطبيقات LLM الأخرى أيضًا.

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 = [
"اشرح الحوسبة الكمومية ببساطة.",
"اكتب قصيدة عن الذكاء الاصطناعي.",
"أصف عملية Photosynthesis."
]</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"Prompt: {prompt}\nResponse: {result}\n")</p>

asyncio.run(main())

في هذا المثال ، نحدد دالة متزامنة generate_text التي تقوم بإجراء مكالمة إلى واجهة برمجة التطبيقات OpenAI باستخدام عميل AsyncOpenAI. الدالة main tạo مهام متعددة لتحفيزات مختلفة و تستخدم asyncio.gather() لتشغيلها بشكل متوازي.

هذا النهج يسمح لنا بإرسال عدة طلبات إلى واجهة برمجة التطبيقات LLM في نفس الوقت ، مما يقلل بشكل كبير من الوقت الإجمالي المطلوب لمعالجة جميع التحفيزات.

تقنيات متقدمة: تعبئة الطلبات ومراقبة التزامن

في حين أن المثال السابق يظهر أساسيات مكالمات واجهة برمجة التطبيقات LLM المتزامنة ، فإن التطبيقات الفعلية غالبًا ما تتطلب نهجًا أكثر تعقيدًا. دعونا نستكشف تقنيتين مهمتين: تعبئة الطلبات ومراقبة التزامن.

تعبئة الطلبات: عند التعامل مع عدد كبير من التحفيزات ، غالبًا ما يكون من الأفضل تعبئةها في مجموعات بدلاً من إرسال طلبات فردية لكل تحفيز. هذا يقلل من عبء عمل مكالمات واجهة برمجة التطبيقات المتعددة ويمكن أن يؤدي إلى أداء أفضل.

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"Tell me a fact about number {i}" 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"Prompt: {prompt}\nResponse: {result}\n")</p>

asyncio.run(main())

مراقبة التزامن: في حين أن البرمجة المتزامنة تسمح بالتنفيذ المتوازي ، من المهم مراقبة مستوى التزامن لتجنب إفراط واجهة برمجة التطبيقات أو تجاوز الحدود. يمكننا استخدام asyncio.Semaphore لهذا الغرض.

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"Tell me a fact about number {i}" 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"Prompt: {prompt}\nResponse: {result}\n")</p>

asyncio.run(main())

في هذا المثال ، نستخدم semaphore لتحديد عدد الطلبات المتزامنة إلى 5 ، مما يضمن ألا نفرط في واجهة برمجة التطبيقات.

تعامل الأخطاء ومحاولة إعادة التشغيل في مكالمات واجهة برمجة التطبيقات LLM المتزامنة

عند العمل مع واجهات برمجة التطبيقات الخارجية ، من المهم تنفيذ تعامل أخطاء قوي وآليات إعادة التشغيل. دعونا نُحسن رمزنا لتعامل الأخطاء الشائعة وتنفيذ إعادة التشغيل مع تأخير متسارع.

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"Error occurred: {e}")
raise APIError("Failed to generate text")</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, "Failed to generate response after multiple attempts."</p>

<p>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)</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"Prompt: {prompt}\nResponse: {result}\n")</p>

asyncio.run(main())

هذا الإصدار المحسن يتضمن:

• استثناء مخصص APIError ل الأخطاء المتعلقة بواجهة برمجة التطبيقات.
• دالة generate_text_with_retry المزخرفة ب @retry من مكتبة tenacity ، مع إعادة تشغيل متسارع.
• تعامل الأخطاء في دالة process_prompt لالتقاط وتسجيل الأخطاء.

تحسين الأداء: استجابات التدفق

لإنشاء المحتوى الطويل ، يمكن أن يؤدي استجابات التدفق إلى تحسين كبير في أداء تطبيقك. بدلاً من انتظار الاستجابة الكاملة ، يمكنك معالجة وعرض قطع من النص كما يصبح متاحًا.

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 = "Write a short story about a time-traveling scientist."</p>

<p>async with AsyncOpenAI() as client:
result = await stream_text(prompt, client)</p>

<p>print(f"Full response:\n{result}")</p>

asyncio.run(main())

هذا المثال يظهر كيفية استجابة التدفق من واجهة برمجة التطبيقات ، وطباعة كل قطعة كما يصل.

بناء سلاسل العمل المتزامنة مع LangChain

للتطبيقات الأكثر تعقيدًا التي تعمل باللمس LLM ، يوفر إطار LangChain تجريدًا عاليًا يبسط عملية ربط مكالمات متعددة واجهة برمجة التطبيقات LLM وتكامل أدوات أخرى. دعونا نلقي نظرة على مثال لاستخدام LangChain مع القدرات المتزامنة:

هذا المثال يظهر كيف يمكن استخدام LangChain لإنشاء سلاسل عمل أكثر تعقيدًا مع التنفيذ المتزامن والتدفق.

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="Write a short story about {topic}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>

<p>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)</p>

<p>for topic, story in zip(topics, stories):
print(f"\nTopic: {topic}\nStory: {story}\n{'='*50}\n")</p>

asyncio.run(main())

تقديم تطبيقات واجهة برمجة التطبيقات LLM المتزامنة مع FastAPI

لجعل تطبيق واجهة برمجة التطبيقات LLM المتزامن متاحًا كخدمة ويب ، FastAPI هو خيار جيد بسبب دعمه الأصلي للعمليات المتزامنة. هنا مثال لإنشاء نقطة نهاية بسيطة لإنشاء نص:

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># Simulate some post-processing in the background
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):
# Simulate logging or additional processing
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)

تطبيق FastAPI هذا يخلق نقطة نهاية /generate التي تقبل تحفيزًا وترجع نصًا تم إنشاؤه. كما يظهر كيفية استخدام مهام الخلفية لمعالجة إضافية دون حجب الاستجابة.

أفضل الممارسات والخطوط العريضة الشائعة

عند العمل مع مكالمات واجهة برمجة التطبيقات LLM المتزامنة ، احرص على اتباع هذه أفضل الممارسات:

1. استخدم مجموعة الاتصالات: عند إجراء عدة طلبات ، أعد استخدام الاتصالات لتقليل العبء.
2. تنفيذ تعامل أخطاء مناسب: احسب دائمًا لمشاكل الشبكة وأخطاء واجهة برمجة التطبيقات و الاستجابات غير المتوقعة.
3. احترام حدود المعدل: استخدم سمافور أو آليات مراقبة التزامن الأخرى لتجنب إفراط واجهة برمجة التطبيقات.
4. مراقبة وتسجيل: تنفيذ تسجيل شامل لتتبع الأداء وتحديد المشاكل.
5. استخدام التدفق لإنشاء المحتوى الطويل: يحسن تجربة المستخدم ويسمح بمعالجة مبكرة للنتائج الجزئية.