Ασύγχρονες Κλήσεις API LLM σε Python: Ένας Ολοκληρωμένος Οδηγός

Ως προγραμματιστές και επιστήμονες δεδομένων, συχνά ourselves χρειαζόμαστε να αλληλεπιδρούμε με αυτά τα ισχυρά μοντέλα μέσω API. Ωστόσο, καθώς οι εφαρμογές μας αυξάνονται σε πολυπλοκότητα και κλίμακα, η ανάγκη για αποτελεσματικές και αποδοτικές αλληλεπιδράσεις API γίνεται κρίσιμη. Αυτό είναι όπου η ασύγχρονη προγραμματισμός λάμπει, επιτρέποντας μας να μεγιστοποιούμε την απόδοση και να ελαττώνουμε την καθυστέρηση όταν εργαζόμαστε με API LLM.

Σε αυτόν τον ολοκληρωμένο οδηγό, θα εξερευνήσουμε τον κόσμο των ασύγχρονων κλήσεων API LLM σε Python. Θα καλύψουμε όλα, από τις βασικές έννοιες της ασύγχρονης προγραμματισμού έως τις προηγμένες τεχνικές για την αντιμετώπιση σύνθετων ροών εργασίας. Μέχρι το τέλος αυτού του άρθρου, θα έχετε μια στερεή κατανόηση του πώς να εκμεταλλευτείτε την ασύγχρονη προγραμματισμό για να ενισχύσετε τις εφαρμογές LLM που σας δρουν.

Πριν ξεκινήσουμε τις λεπτομέρειες των ασύγχρονων κλήσεων API LLM, ας καθορίσουμε μια στερεή βάση στις έννοιες της ασύγχρονης προγραμματισμού.

Η ασύγχρονη προγραμματισμός επιτρέπει πολλές λειτουργίες να εκτελούνται ταυτόχρονα χωρίς να μπλοκάρουν το κύριο νήμα εκτέλεσης. Σε Python, αυτό επιτυγχάνεται κυρίως μέσω του模块 asyncio, ο οποίος παρέχει ένα πλαίσιο για την γραφή συγχρονισμένου κώδικα χρησιμοποιώντας coroutines, event loops και futures.

Κλειδιά Έννοιες:

• Coroutines: Συναρτήσεις που ορίζονται με async def που μπορούν να pauset και να επαναλαμβάνονται.
• Event Loop: Το κεντρικό μηχανισμό εκτέλεσης που διαχειρίζεται και εκτελεί ασύγχρονες εργασίες.
• Awaitables: Αντικείμενα που μπορούν να χρησιμοποιηθούν με το await keyword (coroutines, tasks, futures).

Εδώ είναι ένα απλό παράδειγμα για να εικονογραφήσουμε αυτές τις έννοιες:

import asyncio

<p>async def greet(name):</p>
<p>await asyncio.sleep(1) # Simulate an I/O operation</p>
<p>print(f"Γεια σας, {name}!")</p>

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

asyncio.run(main())

Σε αυτό το παράδειγμα, ορίζουμε μια ασύγχρονη συνάρτηση greet που προσομοιώνει μια λειτουργία Εισόδου/Εξόδου με asyncio.sleep(). Η συνάρτηση main χρησιμοποιεί asyncio.gather() για να εκτελέσει πολλαπλές χαιρετίσματα ταυτόχρονα.尽管 υπάρχει καθυστέρηση sleep, όλα τα τρία χαιρετίσματα θα εκτυπωθούν μετά από περίπου 1 δευτερόλεπτο, εικονογραφώντας τη δύναμη της ασύγχρονης εκτέλεσης.

Η Ανάγκη για Async σε Κλήσεις API LLM

Όταν εργαζόμαστε με API LLM, συχνά συναντούμε σενάρια όπου πρέπει να κάνουμε πολλές κλήσεις API, είτε σε σειρά είτε παράλληλα. Ο παραδοσιακός συγχρονισμένος κώδικας μπορεί να οδηγήσει σε σημαντικά προβλήματα απόδοσης, ιδιαίτερα όταν αντιμετωπίζουμε λειτουργίες υψηλής καθυστέρησης όπως οι αιτήσεις δικτύου προς υπηρεσίες LLM.

Σκεφτείτε ένα σενάριο όπου πρέπει να δημιουργήσουμε περίληψη για 100 διαφορετικά άρθρα χρησιμοποιώντας ένα API LLM. Με μια συγχρονισμένη προσέγγιση, κάθε κλήση API θα μπλοκάρει μέχρι να λάβει απάντηση, потенτικά παίρνοντας αρκετά λεπτά για να ολοκληρώσει όλες τις αιτήσεις. Μια ασύγχρονη προσέγγιση, από την άλλη πλευρά, μας επιτρέπει να ξεκινήσουμε πολλές κλήσεις API ταυτόχρονα, μειώνοντας δραματικά τον συνολικό χρόνο εκτέλεσης.

Ρύθμιση του Περιβάλλοντος σας

Για να ξεκινήσετε με τις ασύγχρονες κλήσεις API LLM, θα χρειαστεί να ρυθμίσετε το περιβάλλον Python σας με τις απαραίτητες βιβλιοθήκες. Εδώ είναι ό,τι χρειάζεστε:

• Python 3.7 ή υψηλότερη (για εγγενή υποστήριξη asyncio)
• aiohttp: Μια ασύγχρονη βιβλιοθήκη πελάτη HTTP
• openai: Η επίσημη βιβλιοθήκη πελάτη Python της OpenAI (αν χρησιμοποιείτε τα μοντέλα GPT της OpenAI)
• langchain: Ένα πλαίσιο για την κατασκευή εφαρμογών με LLM (προαιρετικό, αλλά συνιστάται για σύνθετες ροές εργασίας)

Μπορείτε να εγκαταστήσετε αυτές τις εξαρτήσεις χρησιμοποιώντας pip:

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

Βασικές Ασύγχρονες Κλήσεις API LLM με asyncio και aiohttp

Ας ξεκινήσουμε με μια απλή ασύγχρονη κλήση σε ένα API LLM χρησιμοποιώντας aiohttp. Θα χρησιμοποιήσουμε το API GPT-3.5 της OpenAI ως παράδειγμα, αλλά οι έννοιες ισχύουν και για άλλα API LLM.

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 που κάνει μια κλήση στο API της OpenAI χρησιμοποιώντας τον πελάτη AsyncOpenAI. Η συνάρτηση main δημιουργεί πολλές εργασίες για διαφορετικές προτάσεις και χρησιμοποιεί asyncio.gather() για να εκτελέσει ταυτόχρονα.

Αυτή η προσέγγιση μας επιτρέπει να στείλουμε πολλές αιτήσεις στο API LLM ταυτόχρονα, μειώνοντας δραματικά τον συνολικό χρόνο εκτέλεσης.

Προηγμένες Τεχνικές: Συσσώρευση και Έλεγχος Συγκριτικής Từξης

Ενώ το προηγούμενο παράδειγμα δείχνει τις βασικές έννοιες των ασύγχρονων κλήσεων API LLM, οι πραγματικές εφαρμογές συχνά απαιτούν πιο σύνθετες προσεγγίσεις. Ας εξερευνήσουμε δύο σημαντικές τεχνικές: συσσώρευση αιτήσεων και έλεγχος συγκριτικής απόδοσης.

Συσσώρευση Αιτήσεων: Όταν αντιμετωπίζουμε ένα μεγάλο αριθμό προτάσεων, είναι συχνά πιο αποτελεσματικό να τις συσσωρεύσουμε σε ομάδες αντί να στέλνουμε ατομικές αιτήσεις για κάθε πρόταση. Αυτό μειώνει την επικεφαλίδευση πολλών κλήσεων API και μπορεί να οδηγήσει σε καλύτερη απόδοση.

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

Έλεγχος Συγκριτικής Απόδοσης: ενώ η ασύγχρονη προγραμματισμός επιτρέπει την ταυτόχρονη εκτέλεση, είναι σημαντικό να ελέγξουμε το επίπεδο συγκριτικής απόδοσης για να αποφευχθεί η υπερφόρτωση του διακομιστή API ή η υπέρβαση των ορίων συχνότητας. Μπορούμε να χρησιμοποιήσουμε 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())

Σε αυτό το παράδειγμα, χρησιμοποιούμε ένα semaphore για να περιορίσουμε τον αριθμό των ταυτόχρονων αιτήσεων σε 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

<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 για σφάλματα που σχετίζονται με το API.
• Μια συνάρτηση generate_text_with_retry που διακοσμείται με @retry από τη βιβλιοθήκη tenacity, υλοποιώντας εκθετική αναμονή.
• Χειρισμό σφαλμάτων στη συνάρτηση 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 = "Γράψτε μια σύντομη ιστορία για einen επιστήμονα που ταξίδευε στο χρόνο."</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())

Αυτό το παράδειγμα δείχνει πώς να ροήσετε την απάντηση από το API, εκτυπώνοντας κάθε τμήμα καθώς φθάνει. Αυτή η προσέγγιση είναι ιδιαίτερα χρήσιμη για εφαρμογές chat ή οποιοδήποτε σενάριο όπου θέλετε να παρέχετε άμεση ανάδραση στον χρήστη.

Δομή Ασύγχρονων Ροών Εργασίας με LangChain

Για πιο σύνθετες εφαρμογές που βασίζονται σε LLM, το πλαίσιο LangChain παρέχει μια υψηλού επιπέδου αφαίρεση που απλοποιεί τη διαδικασία αλυσίδωσης πολλών κλήσεων LLM και την ενσωμάτωση άλλων εργαλείων. Ας δούμε ένα παράδειγμα χρήσης του 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())

Παρουσίαση Ασύγχρονων Εφαρμογών LLM με FastAPI

Για να κάνετε την ασύγχρονη εφαρμογή LLM σας διαθέσιμη ως υπηρεσία web, το FastAPI είναι μια εξαιρετική επιλογή λόγω της εγγενής υποστήριξής του για ασύγχρονες λειτουργίες. Εδώ είναι ένα παράδειγμα του πώς να δημιουργήσετε ένα απλό σημείο API για γεννήτρια κειμένου:

Αυτή η εφαρμογή FastAPI δημιουργεί ένα σημείο API /generate που δέχεται μια πρόταση και επιστρέφει γεννημένο κείμενο. Επίσης, δείχνει πώς να χρησιμοποιήσετε εργασίες στο παρασκήνιο για πρόσθετη επεξεργασία χωρίς να μπλοκάρει την απάντηση.

Καλές Πρακτικές και Συνηθισμένα Λάθη

Καθώς εργάζεστε με ασύγχρονες κλήσεις API LLM, κρατήστε αυτές τις καλές πρακτικές στο μυαλό σας:

1. Χρησιμοποιήστε πισίνα συνδέσεων: Όταν κάνετε πολλές αιτήσεις, επαναχρησιμοποιήστε συνδέσεις για να μειώσετε την επικεφαλίδευση.
2. Υλοποιήστε σωστό χειρισμό σφαλμάτων: Πάντα να λαμβάνετε υπόψη τα προβλήματα δικτύου, σφάλματα API και απρόσμενες απαντήσεις.
3. Σεβαστείτε τα όρια συχνότητας: Χρησιμοποιήστε semaphores ή άλλα μηχανισμούς ελέγχου συγκριτικής απόδοσης για να αποφευχθεί η υπερφόρτωση του διακομιστή API.
4. Παρακολουθήστε και καταγράψτε: Υλοποιήστε μια πλήρη καταγραφή για να παρακολουθήσετε την απόδοση και να αναγνωρίσετε ζητήματα.
5. Χρησιμοποιήστε ροή για περιεχόμενο μεγάλης κλίμακας: Βελτιστοποιεί την εμπειρία του χρήστη και επιτρέπει την πρώιμη επεξεργασία των μερικών αποτελεσμάτων.