AI 모델 및 플랫폼
파이썬에서 비동기 LLM API 호출: 종합 가이드
Unite.AI를 Google의 선호 소스에 추가
개발자와 데이터 과학자로서, 우리는 종종 이러한 강력한 모델을 API를 통해 상호 작용해야 합니다. 그러나 우리의 응용 프로그램이 복잡성과 규모가 커짐에 따라, 효율적이고 성능이 좋은 API 상호 작용의 필요성이 중요해집니다. 이것이 비동기 프로그래밍이 빛나는 곳입니다. 비동기 프로그래밍을 통해 우리는 LLM API에서 처리량을 최대화하고 지연 시간을 최소화할 수 있습니다.
이 종합 가이드에서, 우리는 파이썬에서 비동기 LLM API 호출의 세계를 탐험할 것입니다. 우리는 비동기 프로그래밍의 기본부터 복잡한 워크플로우를 처리하는 고급 기술까지 모든 것을 다룰 것입니다. 이 기사의 끝까지, 비동기 프로그래밍을 사용하여 LLM 기반 응용 프로그램을 강화하는 방법에 대한扎实한 이해를 갖게 될 것입니다.
비동기 LLM API 호출의 특정 사항에 뛰어들기 전에, 비동기 프로그래밍 개념에 대한 견고한 기초를 세워보겠습니다.
비동기 프로그래밍은 여러 작업을 동시에 실행할 수 있게 해주며, 주 실행 스레드를 차단하지 않습니다. 파이썬에서, 이것은 주로 asyncio 모듈을 통해 달성되며, 코루틴, 이벤트 루프, 퓨처를 사용하여 동시 코드를 작성하는 프레임워크를 제공합니다.
핵심 개념:
- 코루틴: async def로 정의된 함수로, 일시 중지 및 재개할 수 있습니다.
- 이벤트 루프: 비동기 작업을 관리하고 실행하는 중앙 실행 메커니즘입니다.
- 대기 가능 항목: await 키워드와 함께 사용할 수 있는 객체(코루틴, 작업, 퓨처).
이러한 개념을 설명하는 간단한 예는 다음과 같습니다.
import asyncio
<p>async def greet(name):</p>
<p>await asyncio.sleep(1) # I/O 작업을 시뮬레이트</p>
<p>print(f"안녕하세요, {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())
이 예에서, 우리는 비동기 함수 greet를 정의하며, I/O 작업을 시뮬레이트하기 위해 asyncio.sleep()를 사용합니다. main 함수는 asyncio.gather()를 사용하여 여러 인사를 동시에 실행합니다.
이 접근 방식은 모든 인사를 동시에 보낼 수 있게 해주며, 모든 인사를 처리하는 데 필요한 총 시간을 크게 줄입니다.
비동기 LLM API 호출의 필요성
LLM API와 작업할 때, 우리는 종종 여러 API 호출을 순차적으로 또는 병렬로 수행해야 하는 시나리오를 마주합니다. 전통적인 동기식 코드는 특히 네트워크 요청과 같은 고지연 작업을 처리할 때 상당한 성능 병목 현상을 일으킬 수 있습니다.
100개의 다른 기사에 대한 요약을 생성해야 하는 시나리오를 고려해 보십시오. 동기식 접근 방식에서는 각 API 호출이 응답을 받을 때까지 차단되므로 모든 요청을 처리하는 데 수 분이 걸릴 수 있습니다. 반면에, 비동기 접근 방식은 여러 API 호출을 동시에 시작할 수 있게 해주며, 전체 실행 시간을 크게 줄입니다.
환경 설정
비동기 LLM API 호출을 시작하려면, 필요한 라이브러리가 포함된 파이썬 환경을 설정해야 합니다. 필요한 것은 다음과 같습니다.
- 파이썬 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 호출
비동기 LLM API 호출의 기본을 보여주는 간단한 예를 살펴보겠습니다. 우리는 OpenAI의 GPT-3.5 API를 사용할 것입니다.
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를 정의하며, OpenAI API에 요청을 보냅니다. main 함수는 여러 질문을 동시에 실행합니다.
이 접근 방식은 여러 요청을 동시에 보낼 수 있게 해주며, 모든 질문을 처리하는 데 필요한 총 시간을 크게 줄입니다.
고급 기술: 배칭 및 동시성 제어
실제 응용 프로그램에서는, 더 복잡한 접근 방식이 필요할 수 있습니다. 배칭과 동시성 제어에 대해 살펴보겠습니다.
배칭 요청: 많은 질문을 처리할 때, 개별 요청을 보내는 것보다 배칭으로 처리하는 것이 더 효율적일 수 있습니다. 이렇게 하면 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())
이 예에서, 우리는 세마포어를 사용하여 동시 요청 수를 5개로 제한합니다.
비동기 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())
이 강화된 버전에는 다음이 포함됩니다.
- API 관련 오류를 위한 사용자 정의
APIError예외 - tenacity 라이브러리의
@retry로裝饰된generate_text_with_retry함수 - APIError를 잡고 보고하는
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 = "시간 여행하는 과학자에 대한 단편 소설을 작성해주세요."</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에서 응답을 스트리밍하는 방법을 보여줍니다. 이는 채팅 애플리케이션 또는 사용자에게 실시간 피드백을 제공해야 하는 모든 시나리오에서 특히 유용합니다.
LangChain을 사용한 비동기 워크플로우 구축
보다 복잡한 LLM 기반 응용 프로그램을 위해, LangChain 프레임워크는 LLM 호출을 체인하고 다른 도구를 통합하는 과정을 단순화하는 고수준 추상화를 제공합니다. LangChain을 사용하여 비동기 워크플로우를 생성하는 예를 살펴보겠습니다.
이 예는 LangChain을 사용하여 스트리밍 및 비동기 실행과 함께 더 복잡한 워크플로우를 생성하는 방법을 보여줍니다.
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="{}에 대한 단편 소설을 작성해주세요."</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())
FastAPI를 사용한 비동기 LLM 애플리케이션 제공
비동기 LLM 애플리케이션을 웹 서비스로 제공하려면, FastAPI는 비동기 작업을 네이티브로 지원하기 때문에 좋은 선택입니다. 텍스트 생성을 위한 간단한 API 엔드포인트를 생성하는 예를 살펴보겠습니다.
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># 일부 후처리를 백그라운드에서 시뮬레이트</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># 로깅 또는 추가 처리를 시뮬레이트</p>
<p>await asyncio.sleep(2)</p>
<p>print(f"로깅: 질문 '{prompt}'에 대한 텍스트 생성, 길이 {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>
이 FastAPI 애플리케이션은 /generate 엔드포인트를 생성하여 프롬프트를 받고 생성된 텍스트를 반환합니다. 또한 백그라운드 작업을 사용하여 추가 처리를 시뮬레이트합니다.
최선의 관행 및 일반적인 함정
비동기 LLM API와 작업할 때, 다음 최선의 관행을 기억하세요.
- 연결 풀링 사용: 여러 요청을 보낼 때, 연결을 재사용하여 오버헤드를 줄입니다.
- 適切한 오류 처리 구현: 네트워크 문제, API 오류 및 예기치 않은 응답에 대해 항상 계산합니다.
- 레이트 제한 존중: 세마포어 또는 다른 동시성 제어 메커니즘을 사용하여 API 서버를 과부화시키지 않습니다.
- 모니터링 및 로깅: 성능을 추적하고 문제를 식별하기 위한 완전한 로깅을 구현합니다.
- 긴 형식 콘텐츠에 대한 스트리밍 사용: 사용자 경험을 개선하고 부분 결과의 초기 처리를 허용합니다.
