การเรียก API LLM แบบ Asynchronous ใน Python: คู่มือที่ครอบคลุม

import asyncio
import aiohttp
from openai import AsyncOpenAI

<p>async def generate_text(prompt, client):
response = await client.chat.completions.create(
model=&quot;gpt-3.5-turbo&quot;,
messages=[{&quot;role&quot;: &quot;user&quot;, &quot;content&quot;: prompt}]
)
return response.choices[0].message.content</p>

<p>async def main():
prompts = [
&quot;อธิบายการคำนวณแบบควอนตัมในคำพูดที่เรียบง่าย.&quot;,
&quot;เขียนฮะอิกูเกี่ยวกับปัญญาประดิษฐ์.&quot;,
&quot;อธิบายกระบวนการของการ фотอสินเทซิส.&quot;
]</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&quot;คำถาม: {prompt}\nคำตอบ: {result}\n&quot;)</p>

asyncio.run(main())

import asyncio
from openai import AsyncOpenAI

<p>async def process_batch(batch, client):
responses = await asyncio.gather(*[
client.chat.completions.create(
model=&quot;gpt-3.5-turbo&quot;,
messages=[{&quot;role&quot;: &quot;user&quot;, &quot;content&quot;: prompt}]
) for prompt in batch
])
return [response.choices[0].message.content for response in responses]</p>

<p>async def main():
prompts = [f&quot;บอกข้อมูลเกี่ยวกับตัวเลข {i}&quot; 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&quot;คำถาม: {prompt}\nคำตอบ: {result}\n&quot;)</p>

asyncio.run(main())

import asyncio
from openai import AsyncOpenAI

<p>async def generate_text(prompt, client, semaphore):
async with semaphore:
response = await client.chat.completions.create(
model=&quot;gpt-3.5-turbo&quot;,
messages=[{&quot;role&quot;: &quot;user&quot;, &quot;content&quot;: prompt}]
)
return response.choices[0].message.content</p>

<p>async def main():
prompts = [f&quot;บอกข้อมูลเกี่ยวกับตัวเลข {i}&quot; 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&quot;คำถาม: {prompt}\nคำตอบ: {result}\n&quot;)</p>

asyncio.run(main())

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=&quot;gpt-3.5-turbo&quot;,
messages=[{&quot;role&quot;: &quot;user&quot;, &quot;content&quot;: prompt}]
)
return response.choices[0].message.content
except Exception as e:
print(f&quot;ข้อผิดพลาดเกิดขึ้น: {e}&quot;)
raise APIError(&quot;ไม่สามารถสร้างข้อความได้&quot;)</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, &quot;ไม่สามารถสร้างคำตอบหลังจากหลายครั้ง&quot;</p>

<p>async def main():
prompts = [f&quot;บอกข้อมูลเกี่ยวกับตัวเลข {i}&quot; 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&quot;คำถาม: {prompt}\nคำตอบ: {result}\n&quot;)</p>

asyncio.run(main())

import asyncio
from openai import AsyncOpenAI

<p>async def stream_text(prompt, client):
stream = await client.chat.completions.create(
model=&quot;gpt-3.5-turbo&quot;,
messages=[{&quot;role&quot;: &quot;user&quot;, &quot;content&quot;: prompt}],
stream=True
)</p>

<p>full_response = &quot;&quot;
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=&#039;&#039;, flush=True)</p>

<p>print(&quot;\n&quot;)
return full_response</p>

<p>async def main():
prompt = &quot;เขียนเรื่องสั้นเกี่ยวกับนักวิทยาศาสตร์ที่เดินทางข้ามเวลา&quot;</p>

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

<p>print(f&quot;คำตอบทั้งหมด:\n{result}&quot;)</p>

asyncio.run(main())

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=[&quot;topic&quot;],
template=&quot;เขียนเรื่องสั้นเกี่ยวกับ {topic}&quot;
)
chain = LLMChain(llm=llm, prompt=prompt)
return await chain.arun(topic=topic)</p>

<p>async def main():
topics = [&quot;ป่ามหัศจรรย์&quot;, &quot;เมืองในอนาคต&quot;, &quot;อารยธรรมใต้น้ำ&quot;]
tasks = [generate_story(topic) for topic in topics]
stories = await asyncio.gather(*tasks)</p>

<p>for topic, story in zip(topics, stories):
print(f&quot;\nหัวข้อ: {topic}\nเรื่องราว: {story}\n{&#039;=&#039;*50}\n&quot;)</p>

asyncio.run(main())

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(&quot;/generate&quot;, response_model=GenerationResponse)
async def generate_text(request: GenerationRequest, background_tasks: BackgroundTasks):
response = await client.chat.completions.create(
model=&quot;gpt-3.5-turbo&quot;,
messages=[{&quot;role&quot;: &quot;user&quot;, &quot;content&quot;: 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&quot;บันทึก: คำถาม &#039;{prompt}&#039; สร้างข้อความที่มีความยาว {len(generated_text)}&quot;)</p>

<p>if __name__ == &quot;__main__&quot;:
import uvicorn
uvicorn.run(app, host=&quot;0.0.0.0&quot;, port=8000)