Моделі та платформи ШІ
Асинхронні виклики LLM API в Python: Комплексний посібник
Додайте Unite.AI до бажаних джерел у Google
Як розробники та вчені-дані, нам часто потрібно взаємодіяти з цими потужними моделями через API. Однак, коли наші додатки зростають у складності та масштабі, потреба в ефективних та продуктивних взаємодіях з API стає критичною. Саме тут асинхронне програмування сяє, дозволяючи нам максимізувати пропускну здатність та мінімізувати затримку при роботі з LLM API.
У цьому комплексному посібнику ми дослідимо світ асинхронних викликів LLM API в Python. Ми покриємо все, від основ асинхронного програмування до просунутих технік для обробки складних робочих процесів. Після закінчення цієї статті у вас буде солідне розуміння того, як використовувати асинхронне програмування для підвищення продуктивності ваших додатків, що працюють з LLM.
Перед тим, як ми перейдемо до конкретики асинхронних викликів LLM API, встановимо солідну основу в концепціях асинхронного програмування.
Асинхронне програмування дозволяє виконувати кілька операцій одночасно без блокування основної нитки виконання. У Python це в основному досягається за допомогою модуля asyncio, який забезпечує.framework для написання конкуруючого коду за допомогою корутин, циклів подій та майбутнього.
Ключові концепції:
- Корутини: Функції, визначені з async def, які можна призупинити та відновити.
- Цикл подій: Центральний механізм виконання, який керує та запускає асинхронні завдання.
- Чекаючі об’єкти: Об’єкти, які можна використовувати з ключовим словом await (корутини, завдання, майбутнє).
Ось простий приклад, який ілюструє ці концепції:
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # Симулюємо операцію вводу/виводу
print(f"Привіт, {name}!")</p>
<p>async def main():
await asyncio.gather(
greet("Аліса"),
greet("Боб"),
greet("Чарлі")
)</p>
asyncio.run(main())
У цьому прикладі ми визначаємо асинхронну функцію greet, яка симулює операцію вводу/виводу за допомогою asyncio.sleep(). Функція main використовує asyncio.gather(), щоб виконувати кілька привітань одночасно. Навіть з затримкою, всі три привітання будуть надруковані після приблизно 1 секунди, демонструючи силу асинхронного виконання.
Потрібність асинхронності в викликах LLM API
При роботі з LLM API нам часто зустрічаються сценарії, коли потрібно зробити кілька викликів API, або послідовно, або паралельно. Традиційний синхронний код може привести до суттєвих проблем з продуктивністю, особливо при роботі з високозатримковими операціями, такими як мережеві запити до сервісів LLM.
Розгляньте сценарій, коли нам потрібно згенерувати резюме для 100 різних статей за допомогою LLM API. З синхронним підходом кожен виклик API блокуватиме виконання до отримання відповіді, потенційно займаючи кілька хвилин для обробки всіх запитів. Асинхронний підхід, з іншого боку, дозволяє нам ініціювати кілька викликів API одночасно, суттєво скорочуючи загальний час виконання.
Налаштування середовища
Щоб почати роботу з асинхронними викликами LLM API, вам потрібно налаштувати середовище Python з необхідними бібліотеками. Ось що вам потрібно:
- Python 3.7 або вище (для підтримки асинхронного програмування)
- aiohttp: Асинхронна бібліотека клієнта HTTP
- openai: Офіційна бібліотека клієнта OpenAI для Python (якщо ви використовуєте моделі GPT від OpenAI)
- langchain: Фреймворк для побудови додатків з LLM (опціонально, але рекомендується для складних робочих процесів)
Ви можете встановити ці залежності за допомогою pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Базові асинхронні виклики LLM API з asyncio та aiohttp
Давайте почнемо з простого асинхронного виклику до LLM API за допомогою aiohttp. Ми використовуємо API GPT-3.5 від OpenAI як приклад, але ці концепції застосовуються до інших LLM API.
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 = [
"Опишіть квантові обчислення у простих термінах.",
"Напишіть хайку про штучний інтелект.",
"Опишіть процес фотосинтезу."
]</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}\nВідповідь: {result}\n")</p>
asyncio.run(main())
У цьому прикладі ми визначаємо асинхронну функцію generate_text, яка робить виклик до API OpenAI за допомогою клієнта AsyncOpenAI. Функція main створює кілька завдань для різних промптів і використовує asyncio.gather(), щоб виконувати їх одночасно.
Цей підхід дозволяє нам надсилати кілька запитів до LLM API одночасно, суттєво скорочуючи загальний час обробки всіх промптів.
Просунуті техніки: Батчінг та контроль конкуренції
Хоча попередній приклад демонструє основи асинхронних викликів LLM API, реальні додатки часто потребують більш просунутих підходів. Давайте дослідимо дві важливі техніки: батчінг запитів та контроль конкуренції.
Батчінг запитів: Коли ви працюєте з великою кількістю промптів, часто більш ефективно групувати їх у пакети, а не надсилати окремі запити для кожного промпту. Це зменшує накладні витрати кількох викликів API і може привести до кращої продуктивності.
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"Розкажіть про число {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}\nВідповідь: {result}\n")</p>
asyncio.run(main())
Контроль конкуренції: Хоча асинхронне програмування дозволяє виконувати кілька операцій одночасно, важливо контролювати рівень конкуренції, щоб не перевантажувати сервер API або перевищувати ліміти швидкості. Ми можемо використовувати 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"Розкажіть про число {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}\nВідповідь: {result}\n")</p>
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
<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"Помилка відбулася: {e}")
raise APIError("Не вдалося згенерувати текст")</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, "Не вдалося згенерувати відповідь після кількох спроб."</p>
<p>async def main():
prompts = [f"Розкажіть про число {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}\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):
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 = "Напишіть коротку історію про мандрівника у часі."</p>
<p>async with AsyncOpenAI() as client:
result = await stream_text(prompt, client)</p>
<p>print(f"Повна відповідь:\n{result}")</p>
asyncio.run(main())
У цьому прикладі ми демонструємо, як потоково обробляти відповідь від API, друкуючи фрагменти тексту по мірі їх появи. Це підхід особливо корисний для чат-додатків або будь-якої ситуації, коли ви хочете надавати користувачеві зворотний зв’язок у режимі реального часу.
Будування асинхронних робочих процесів з 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):
llm = OpenAI(temperature=0.7, streaming=True, callback_manager=AsyncCallbackManager([StreamingStdOutCallbackHandler()]))
prompt = PromptTemplate(
input_variables=["topic"],
template="Напишіть коротку історію про {topic}."
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>
<p>async def main():
topics = ["магічний ліс", "футуристичне місто", "підводна цивілізація"]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>
<p>for topic, story in zip(topics, stories):
print(f"Тема: {topic}\nІсторія: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Побудова асинхронних додатків LLM з FastAPI
Щоб зробити ваш асинхронний додаток LLM доступним як веб-сервіс, FastAPI є чудовим вибором завдяки його вбудованій підтримці асинхронних операцій. Давайте розглянемо приклад створення простого API-ендпоінта для генерації тексту:
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># Симулюємо деяку обробку у фоні
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):
# Симулюємо логування або додаткову обробку
await asyncio.sleep(2)
print(f"Залоговано: Промпт '{prompt}' згенерував текст довжиною {len(generated_text)}")</p>
<p>if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
У цьому прикладі ми створюємо додаток FastAPI з ендпоінтом /generate, який приймає промпт і повертає згенерований текст. Ми також демонструємо, як використовувати завдання у фоні для додаткової обробки без блокування відповіді.
Найкращі практики та поширені помилки
При роботі з асинхронними викликами LLM API дотримуйтесь цих найкращих практик:
- Використовуйте пулі підключень: Коли робите кілька запитів, повторно використовуйте підключення, щоб зменшити накладні витрати.
- Реалізуйте належну обробку помилок: Завжди передбачайте мережеві проблеми, помилки API та непередбачувані відповіді.
- Відповідайте ліміти швидкості: Використовуйте семафори або інші механізми контролю конкуренції, щоб не перевантажувати сервер API.
- Моніторьте та логуйте: Реалізуйте повне логування, щоб відстежувати продуктивність та виявляти проблеми.
- Використовуйте потокові відповіді для довгих текстів: Це покращує досвід користувача та дозволяє обробляти часткові результати.
