AI 模型与平台
异步LLM API调用在Python中的应用：综合指南
作为开发人员和数据科学家，我们经常需要通过API与这些强大的模型进行交互。然而，当我们的应用程序变得更加复杂和庞大时，高效和高性能的API交互变得至关重要。这就是异步编程的用处，它允许我们最大化吞吐量和最小化延迟，当我们与LLM API合作时。
在这份综合指南中，我们将探索Python中异步LLM API调用的世界。我们将涵盖从异步编程基础到处理复杂工作流的高级技术的一切。到本文结束时，您将对如何利用异步编程来增强您的LLM驱动应用程序有一个扎实的理解。
在我们深入异步LLM API调用的具体细节之前，让我们先建立一个异步编程概念的坚实基础。
异步编程允许多个操作同时执行，而不会阻塞主执行线程。在Python中，这主要是通过asyncio模块实现的，asyncio模块提供了一个框架，用于使用协程、事件循环和未来来编写并发代码。
关键概念：
- 协程：使用async def定义的函数，可以暂停和恢复。
- 事件循环：管理和运行异步任务的中心执行机制。
- 可等待对象：可以使用await关键字的对象（协程、任务、未来）。
以下是一个简单的示例来说明这些概念：
import asyncio
<p>async def greet(name):
await asyncio.sleep(1) # 模拟I/O操作
print(f"你好，{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()同时运行多个问候。尽管有延迟，所有三个问候都会在大约1秒后打印出来，展示了异步执行的力量。
异步LLM API调用中的需求
当我们与LLM API合作时，我们经常遇到需要进行多个API调用的场景，无论是顺序还是并行。传统的同步代码可能会导致显著的性能瓶颈，尤其是在处理高延迟操作（如网络请求到LLM服务）时。
考虑一个场景，我们需要使用LLM API为100个不同的文章生成摘要。使用同步方法，每个API调用都会阻塞，直到它收到响应，可能需要几分钟才能完成所有请求。另一方面，异步方法允许我们同时启动多个API调用，大大减少了总执行时间。
设置您的环境
要开始使用异步LLM API调用，您需要设置Python环境，并安装必要的库。您需要以下内容：
- Python 3.7 或更高版本（用于本地asyncio支持）
- aiohttp：一个异步HTTP客户端库
- openai：官方的OpenAI Python客户端（如果您使用OpenAI的GPT模型）
- langchain：一个用于构建LLM应用程序的框架（可选，但建议用于复杂工作流）
您可以使用pip安装这些依赖项：
<p>pip install aiohttp openai langchain <div class="relative flex flex-col rounded-lg">
使用asyncio和aiohttp的基本异步LLM API调用
让我们从使用aiohttp对LLM API进行简单的异步调用开始。我们将使用OpenAI的GPT-3.5 API作为示例，但这些概念也适用于其他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，它使用AsyncOpenAI客户端对OpenAI API进行调用。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())
在这个例子中，我们使用一个信号量来限制并发请求的数量，确保我们不会让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函数，使用tenacity库的@retry装饰器，实现指数退避重试。 - 在
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"\n主题: {topic}\n故事: {story}\n{'='*50}\n")</p>
asyncio.run(main())
使用FastAPI提供异步LLM应用程序
要将您的异步LLM应用程序作为Web服务提供，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服务器不堪重负。
- 监控和记录：实施全面记录以跟踪性能并识别问题。
- 使用流式传输进行长篇内容：它改善了用户体验，并允许实时处理部分结果。
