Mô hình và nền tảng AI
Các Cuộc Gọi LLM API Không Đồng Bộ trong Python: Hướng Dẫn Toàn Diện
Thêm Unite.AI vào các nguồn ưu tiên của bạn trên Google
Là các nhà phát triển và nhà khoa học dữ liệu, chúng tôi thường phải tương tác với các mô hình mạnh mẽ này thông qua các API. Tuy nhiên, khi các ứng dụng của chúng tôi phát triển về phức tạp và quy mô, nhu cầu về các tương tác API hiệu quả và hiệu suất trở nên quan trọng. Đây là nơi lập trình không đồng bộ tỏa sáng, cho phép chúng tôi tối đa hóa thông lượng và giảm thiểu độ trễ khi làm việc với các API LLM.
Trong hướng dẫn toàn diện này, chúng tôi sẽ khám phá thế giới của các cuộc gọi LLM API không đồng bộ trong Python. Chúng tôi sẽ bao gồm mọi thứ từ các khái niệm cơ bản về lập trình không đồng bộ đến các kỹ thuật tiên tiến để xử lý các công việc phức tạp. Sau khi đọc xong bài viết này, bạn sẽ có một hiểu biết vững chắc về cách tận dụng lập trình không đồng bộ để tăng cường các ứng dụng LLM của mình.
Trước khi chúng tôi đi vào các chi tiết cụ thể của các cuộc gọi LLM API không đồng bộ, hãy thiết lập một nền tảng vững chắc trong các khái niệm lập trình không đồng bộ.
Lập trình không đồng bộ cho phép nhiều hoạt động được thực hiện đồng thời mà không chặn luồng thực hiện chính. Trong Python, điều này được thực hiện chủ yếu thông qua mô-đun asyncio, cung cấp một khuôn khổ để viết mã đồng thời bằng cách sử dụng các coroutine, vòng lặp sự kiện và tương lai.
Khái Niệm Chính:
- Coroutine: Các hàm được định nghĩa với async def có thể được tạm dừng và tiếp tục.
- Vòng Lặp Sự Kiện: Cơ chế thực hiện trung tâm quản lý và chạy các nhiệm vụ không đồng bộ.
- Đối Tượng Có Thể Chờ: Các đối tượng có thể được sử dụng với từ khóa await (coroutine, task, tương lai).
Dưới đây là một ví dụ đơn giản để minh họa các khái niệm này:
import asyncio
<p>async def greet(name):</p>
<p>await asyncio.sleep(1) # Simulate một hoạt động I/O</p>
<p>print(f"Xin chào, {name}!")</p>
<p>async def main():</p>
<p>await asyncio.gather(</p>
<p>greet("Alice"),</p>
<p>greet("Bob"),</p>
<p>greet("Charlie")</p>
<p>)</p>
asyncio.run(main())
Trong ví dụ này, chúng tôi định nghĩa một hàm không đồng bộ greet mô phỏng một hoạt động I/O bằng cách sử dụng asyncio.sleep(). Hàm main sử dụng asyncio.gather() để chạy nhiều lời chào đồng thời. Mặc dù có độ trễ do giấc ngủ, tất cả ba lời chào sẽ được in sau khoảng 1 giây, chứng tỏ sức mạnh của việc thực hiện không đồng bộ.
Sự Cần Thiết Của Async Trong Các Cuộc Gọi LLM API
Khi làm việc với các API LLM, chúng tôi thường gặp các tình huống mà chúng tôi cần thực hiện nhiều cuộc gọi API, có thể là tuần tự hoặc song song. Mã đồng bộ truyền thống có thể dẫn đến các nút thắt hiệu suất đáng kể, đặc biệt là khi đối phó với các hoạt động có độ trễ cao như yêu cầu mạng đến các dịch vụ LLM.
Hãy xem xét một tình huống mà chúng tôi cần tạo tóm tắt cho 100 bài viết khác nhau bằng cách sử dụng một API LLM. Với cách tiếp cận đồng bộ, mỗi cuộc gọi API sẽ chặn cho đến khi nó nhận được phản hồi, có thể mất vài phút để hoàn thành tất cả các yêu cầu. Mặt khác, cách tiếp cận không đồng bộ cho phép chúng tôi khởi tạo nhiều cuộc gọi API đồng thời, giảm đáng kể thời gian thực hiện tổng thể.
Cài Đặt Môi Trường Của Bạn
Để bắt đầu với các cuộc gọi LLM API không đồng bộ, bạn sẽ cần cài đặt môi trường Python của mình với các thư viện cần thiết. Dưới đây là những gì bạn sẽ cần:
- Python 3.7 hoặc cao hơn (để hỗ trợ không đồng bộ native)
- aiohttp: Một thư viện khách hàng HTTP không đồng bộ
- openai: Thư viện khách hàng Python chính thức của OpenAI (nếu bạn đang sử dụng mô hình GPT của OpenAI)
- langchain: Một khuôn khổ để xây dựng các ứng dụng với LLM (không bắt buộc, nhưng được khuyến nghị cho các công việc phức tạp)
Bạn có thể cài đặt các phụ thuộc này bằng cách sử dụng pip:
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
Các Cuộc Gọi LLM API Cơ Bản với Asyncio và Aiohttp
Hãy bắt đầu bằng cách thực hiện một cuộc gọi LLM API không đồng bộ đơn giản bằng cách sử dụng aiohttp. Chúng tôi sẽ sử dụng API GPT-3.5 của OpenAI làm ví dụ, nhưng các khái niệm này áp dụng cho các API LLM khác.
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>"Giải thích tính toán lượng tử bằng ngôn ngữ đơn giản.",</p>
<p>"Viết một bài haiku về trí tuệ nhân tạo.",</p>
<p>"Mô tả quá trình quang hợ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: {prompt}\nPhản Hồi: {result}\n")</p>
asyncio.run(main())
Trong ví dụ này, chúng tôi định nghĩa một hàm không đồng bộ generate_text thực hiện một cuộc gọi đến API OpenAI bằng cách sử dụng khách hàng AsyncOpenAI. Hàm main tạo các nhiệm vụ cho các lời nhắc khác nhau và sử dụng asyncio.gather() để chạy chúng đồng thời.
Cách tiếp cận này cho phép chúng tôi gửi nhiều yêu cầu đến API LLM đồng thời, giảm đáng kể thời gian tổng thể cần thiết để xử lý tất cả các lời nhắc.
Kỹ Thuật Tiên Tiến: Batching và Kiểm Soát Đồng Thời
Mặc dù ví dụ trước đó chứng tỏ các nguyên tắc cơ bản của các cuộc gọi LLM API không đồng bộ, nhưng các ứng dụng thực tế thường yêu cầu các phương pháp tinh vi hơn. Hãy khám phá hai kỹ thuật quan trọng: batching yêu cầu và kiểm soát đồng thời.
Batching Yêu Cầu: Khi đối phó với một số lượng lớn lời nhắc, thường hiệu quả hơn khi nhóm chúng thành các lô thay vì gửi từng yêu cầu riêng lẻ. Điều này giảm thiểu overhead của nhiều cuộc gọi API và có thể dẫn đến hiệu suất tốt hơn.
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"Hãy cho tôi biết một sự kiện về số {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: {prompt}\nPhản Hồi: {result}\n")</p>
asyncio.run(main())
Kiểm Soát Đồng Thời: Mặc dù lập trình không đồng bộ cho phép thực hiện đồng thời, nhưng điều quan trọng là phải kiểm soát mức độ đồng thời để tránh làm quá tải máy chủ API hoặc vượt quá giới hạn tốc độ. Chúng tôi có thể sử dụng asyncio.Semaphore để thực hiện việc này.
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"Hãy cho tôi biết một sự kiện về số {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: {prompt}\nPhản Hồi: {result}\n")</p>
asyncio.run(main())
Trong ví dụ này, chúng tôi sử dụng một semaphore để giới hạn số lượng yêu cầu đồng thời lên 5, đảm bảo rằng chúng tôi không làm quá tải máy chủ API.
Xử Lý Lỗi và Thử Lại Trong Các Cuộc Gọi LLM Không Đồng Bộ
Khi làm việc với các API bên ngoài, việc triển khai xử lý lỗi mạnh mẽ và cơ chế thử lại là rất quan trọng. Hãy nâng cấp mã của chúng tôi để xử lý các lỗi phổ biến và triển khai thời gian chờ lại theo cấp số nhân.
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"Lỗi xảy ra: {e}")</p>
<p>raise APIError("Không thể tạo văn bản")</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, "Không thể tạo phản hồi sau nhiều lần thử."</p>
<p>async def main():</p>
<p>prompts = [f"Hãy cho tôi biết một sự kiện về số {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: {prompt}\nPhản Hồi: {result}\n")</p>
asyncio.run(main())
Phiên bản nâng cấp này bao gồm:
- Một ngoại lệ
APIErrortùy chỉnh cho các lỗi API. - Một hàm
generate_text_with_retryđược trang trí với@retrytừ thư viện tenacity, triển khai thời gian chờ lại theo cấp số nhân. - Xử lý lỗi trong hàm
process_promptđể bắt và báo cáo các lỗi.
Tối Ưu Hóa Hiệu Suất: Phản Hồi Dòng
Đối với việc tạo nội dung dài, phản hồi dòng có thể cải thiện đáng kể hiệu suất nhận thức của ứng dụng của bạn. Thay vì chờ toàn bộ phản hồi, bạn có thể xử lý và hiển thị các phần của văn bản khi chúng có sẵn.
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 = "Viết một câu chuyện ngắn về một nhà khoa học du hành thời gian."</p>
<p>async with AsyncOpenAI() as client:</p>
<p>result = await stream_text(prompt, client)</p>
<p>print(f"Phản Hồi Toàn Bộ:\n{result}")</p>
asyncio.run(main())
Ví dụ này chứng tỏ cách thức phản hồi dòng từ API, in từng phần của văn bản khi chúng đến. Cách tiếp cận này đặc biệt hữu ích cho các ứng dụng trò chuyện hoặc bất kỳ tình huống nào bạn muốn cung cấp phản hồi thời gian thực cho người dùng.
Xây Dựng Các Công Việc Không Đồng Bộ với LangChain
Đối với các ứng dụng LLM phức tạp, khuôn khổ LangChain cung cấp một mức độ trừu tượng cao giúp đơn giản hóa quá trình tạo chuỗi các cuộc gọi LLM và tích hợp các công cụ khác. Hãy xem một ví dụ về việc sử dụng LangChain với khả năng không đồng bộ:
Ví dụ này cho thấy cách LangChain có thể được sử dụng để tạo các công việc phức tạp hơn với thực hiện không đồng bộ và phản hồi dòng. Các AsyncCallbackManager và StreamingStdOutCallbackHandler cho phép phản hồi dòng thời gian thực.
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="Viết một câu chuyện ngắn về {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 = ["một khu rừng ma thuật", "một thành phố tương lai", "một nền văn minh dưới nước"]</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"\nChủ Đề: {topic}\nCâu Chuyện: {story}\n{'='*50}\n")</p>
asyncio.run(main())
Phục Vụ Các Ứng Dụng LLM Không Đồng Bộ với FastAPI
Để làm cho ứng dụng LLM không đồng bộ của bạn có sẵn như một dịch vụ web, FastAPI là một lựa chọn tuyệt vời do hỗ trợ không đồng bộ native. Dưới đây là một ví dụ về cách tạo một điểm cuối API đơn giản cho việc tạo văn bản:
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># Mô phỏng một số xử lý hậu trường</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># Mô phỏng việc ghi nhật ký hoặc xử lý thêm</p>
<p>await asyncio.sleep(2)</p>
<p>print(f"Đã Ghi: Prompt '{prompt}' tạo văn bản có độ dài {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>
Ứng dụng FastAPI này tạo một điểm cuối /generate chấp nhận một lời nhắc và trả về văn bản được tạo. Nó cũng chứng tỏ cách sử dụng các nhiệm vụ hậu trường cho xử lý thêm mà không chặn phản hồi.
Nguyên Tắc Tốt Nhất và Cái Bẫy Phổ Biến
Khi làm việc với các cuộc gọi LLM API không đồng bộ, hãy nhớ những nguyên tắc tốt nhất sau:
- Sử Dụng Hồ Kết Nối: Khi thực hiện nhiều yêu cầu, hãy tái sử dụng kết nối để giảm overhead.
- Triển Khai Xử Lý Lỗi Chính Thức: Luôn tính đến các vấn đề về mạng, lỗi API và phản hồi không mong đợi.
- Tôn Trọng Giới Hạn Tốc Độ: Sử dụng các semaphore hoặc cơ chế kiểm soát đồng thời khác để tránh làm quá tải máy chủ API.
- Giám Sát và Ghi Nhật Ký: Triển khai ghi nhật ký toàn diện để theo dõi hiệu suất và xác định vấn đề.
- Sử Dụng Phản Hồi Dòng Cho Nội Dung Dài: Nó cải thiện trải nghiệm người dùng và cho phép xử lý sớm các kết quả một phần.
