LiteLLM에서 스트리밍과 비동기 호출하기

LLM 응답은 생성이 끝나야 통째로 돌아오는 게 기본이에요. 그런데 채팅 UI나 실시간 상황에서 '몇 초 기다렸다가 한꺼번에 받는' 경험은 답답하죠. 이번엔 LiteLLM이 어떻게 토큰을 하나씩 흘려보내는 스트리밍을 지원하고, 비동기(Async)로는 어떻게 호출하는지 살펴볼게요. 같은 completion 인터페이스에서 stream=True만 붙이면 거의 모든 공급자에서 동일하게 동작하니, 한 번 익혀두면 어떤 모델로 바꿔도 그대로 쓸 수 있어요.

출처: 공식문서

스트리밍 응답

completion 함수에 stream=True를 넘기면 모델 응답을 스트리밍으로 받을 수 있어요. 결과는 한 번에 내려오는 대신, 생성되는 대로 조각(chunk)들이 반복자(iterator)로 흘러나오죠.

from litellm import completion

messages = [{"role": "user", "content": "Hey, how's it going?"}]

response = completion(model="gpt-5.6-luna", messages=messages, stream=True)

for part in response:
    print(part.choices[0].delta.content or "")

스트리밍 동안에는 part.choices[0].delta.content에 지금 막 생성된 내용 조각이 들어 있어요. 문자열이 비어 있는 경우도 있으니 or ""로 처리해 자연스럽게 이어 붙이면 돼요.

스트리밍 조각을 하나의 응답으로 다시 조립하기

stream=True가 아니라 스트리밍을 하면서도 나중에 전체 응답을 한 번에 얻고 싶을 때가 있어요. LiteLLM은 흘러나온 조각들을 모아 완성된 응답으로 되돌려 주는 헬퍼 함수 stream_chunk_builder를 제공해요.

from litellm import completion

messages = [{"role": "user", "content": "Hey, how's it going?"}]

response = completion(model="gpt-5.6-luna", messages=messages, stream=True)

for chunk in response:
    chunks.append(chunk)

print(litellm.stream_chunk_builder(chunks, messages=messages))

지금까지 모은 chunks 리스트를 넘기면, 마치 스트리밍을 안 한 것처럼 완성된 응답 객체를 반환해 줘요.

비동기 completion — `acompletion`

LiteLLM은 completion의 비동기 버전인 acompletion도 제공해요. 대용량 요청을 동시에 여러 개 처리할 때 비동기 호출은 I/O 대기 시간을 크게 줄여 주죠. 사용법은 거의 똑같고 await만 붙이면 돼요.

from litellm import acompletion
import asyncio

async def test_get_response():
    user_message = "Hello, how are you?"
    messages = [{"content": user_message, "role": "user"}]
    response = await acompletion(model="gpt-5.6-luna", messages=messages)
    return response

response = asyncio.run(test_get_response())
print(response)

비동기 스트리밍

비동기와 스트리밍은 함께 쓸 수 있어요. LiteLLM은 스트리밍 객체에 __anext__() 함수를 구현해 두어서, 반환된 스트리밍 객체를 async for로 순회할 수 있어요.

from litellm import acompletion
import asyncio, os, traceback

async def completion_call():
    try:
        print("test acompletion + streaming")
        response = await acompletion(
            model="gpt-5.6-luna",
            messages=[{"content": "Hello, how are you?", "role": "user"}],
            stream=True
        )
        print(f"response: {response}")
        async for chunk in response:
            print(chunk)
    except:
        print(f"error occurred: {traceback.format_exc()}")
        pass

asyncio.run(completion_call())

stream=True를 함께 넘기면 비동기 스트리밍이 되고, async for로 조각을 하나씩 받아요.

무한 루프 방지 처리

가끔 모델이 같은 조각만 계속 반복하며 무한 루프에 빠지는 경우가 있어요. LiteLLM은 같은 조각이 'n'번(기본값 100) 반복되면 이를 감지해 litellm.InternalServerError를 던져서 재시도 로직이 동작할 수 있게 해줘요.

import litellm
import os

litellm.set_verbose = False
loop_amount = litellm.REPEATED_STREAMING_CHUNK_LIMIT + 1
chunks = [
    litellm.ModelResponse(**{
    "id": "chatcmpl-123",
    "object": "chat.completion.chunk",
    "created": 1694268190,
    "model": "gpt-5.6-luna",
    "system_fingerprint": "fp_44709d6fcb",
    "choices": [
        {"index": 0, "delta": {"content": "How are you?"}, "finish_reason": "stop"}
    ],
}, stream=True)
] * loop_amount
completion_stream = litellm.ModelResponseListIterator(model_responses=chunks)

response = litellm.CustomStreamWrapper(
    completion_stream=completion_stream,
    model="gpt-5.6-luna",
    custom_llm_provider="cached_response",
    logging_obj=litellm.L
)

한계값은 litellm.REPEATED_STREAMING_CHUNK_LIMIT = 100처럼 직접 조절할 수 있어요. 기본값을 높게 잡아 둔 이유는 정상 응답이 오탐(false positive)되지 않도록 하기 위해서예요.

더 알아보기

채팅 completion의 입력 파라미터와 사용량(usage) 처리 방식이 궁금하다면 선택한 페이지들을 함께 봐요.
스트리밍 응답에 사용량 정보를 함께 담는 방법은 completion 사용량 페이지에서 확인할 수 있어요.

LiteLLM에서 스트리밍과 비동기 호출하기