LangChain StreamingCallbackHandler로 실시간 스트리밍 출력 구현하기

대규모 언어 모델(LLM) 애플리케이션에서 사용자에게 즉각적인 피드백을 제공하는 것은用户体验의 핵심입니다. 이번 튜토리얼에서는 HolySheep AI와 함께 LangChain의 StreamingCallbackHandler를 활용하여 실시간 토큰 스트리밍을 구현하는 방법을 자세히 알아보겠습니다.

HolySheep AI vs 공식 API vs 기타 릴레이 서비스 비교

비교 항목	HolySheep AI	공식 OpenAI API	기타 릴레이 서비스
결제 방식	로컬 결제 지원 (신용카드 불필요)	해외 신용카드 필수	제한적 결제 옵션
스트리밍 지연	평균 120-180ms	평균 150-200ms	200-350ms
GPT-4.1 가격	$8/MTok	$8/MTok	$10-15/MTok
모델 통합	단일 API 키로 10+ 모델	단일 모델	제한적
단일 API 키	✅ 지원	❌	부분 지원
무료 크레딧	✅ 가입 시 제공	❌	불규칙적

왜 스트리밍이 중요한가?

저는 실제 프로덕션 환경에서 스트리밍 미지원 시 사용자가 체감하는 대기 시간이 3-5초에 달하는 것을 경험했습니다. 스트리밍을 통해 토큰이 생성되는 즉시 출력하면 사용자는:

• 응답 시작까지의 대기 시간 단축
• 전체 응답 완료 예상 시간 파악 가능
• 대화형 인터페이스에서 더 자연스러운 상호작용

필수 설치 및 환경 설정

# 필요한 패키지 설치
pip install langchain langchain-openai langchain-core python-dotenv

프로젝트 디렉토리 생성
mkdir langchain-streaming && cd langchain-streaming
touch .env

StreamHandler 클래스 구현

먼저 LangChain의 CallbackHandler를 상속하는 커스텀 핸들러를 만들어 보겠습니다. 이 핸들러는 각 토큰이 생성될 때마다 실시간으로 출력합니다.

import sys
from typing import Any, List, Union
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import Generation, ChatGeneration, LLMResult

class StreamHandler(BaseCallbackHandler):
    """
    LangChain 스트리밍 출력을 위한 커스텀 핸들러
    실시간 토큰 출력 및 타이핑 효과 구현
    """
    
    def __init__(self, container, initial_text=""):
        self.container = container
        self.text = initial_text
        self.first_token = True
        
    def on_llm_new_token(
        self, 
        token: str, 
        **kwargs: Any
    ) -> None:
        """새 토큰이 생성될 때마다 호출되는 콜백"""
        if self.first_token:
            # 첫 토큰은 즉시 출력
            print(f"\n🚀 응답 시작: {token}", end="", flush=True)
            self.first_token = False
        else:
            # 이후 토큰은 실시간 출력
            print(token, end="", flush=True)
        self.text += token
        
    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """LLM 응답 완료 시 호출"""
        print("\n\n✅ 응답 완료")
        self.first_token = True
        
    def on_llm_error(
        self, 
        error: Union[Exception, BaseException], 
        **kwargs: Any
    ) -> None:
        """오류 발생 시 호출"""
        print(f"\n❌ 오류 발생: {str(error)}")
        self.first_token = True
        
    def get_text(self) -> str:
        """현재까지 누적된 텍스트 반환"""
        return self.text
        
    def reset(self) -> None:
        """텍스트 및 상태 초기화"""
        self.text = ""
        self.first_token = True

HolySheep AI와 LangChain 스트리밍 통합

이제 HolySheep AI의 API를 LangChain과 연결하여 스트리밍을 구현하겠습니다. HolySheep AI는 120-180ms의 낮은 지연 시간으로 스트리밍 응답을 제공합니다.

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

.env 파일에서 API 키 로드
load_dotenv()

HolySheep AI 설정
base_url: https://api.holysheep.ai/v1
API 키는 HolySheep 대시보드에서 생성
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")

HolySheep AI를 사용하는 ChatOpenAI 인스턴스 생성
llm = ChatOpenAI(
    model="gpt-4.1",  # 또는 claude-3-5-sonnet, gemini-2.5-flash 등
    temperature=0.7,
    streaming=True,  # 스트리밍 활성화 필수
    api_key=HOLYSHEEP_API_KEY,
    base_url="https://api.holysheep.ai/v1",  # HolySheep AI 엔드포인트
    max_tokens=2048,
)

시스템 및 사용자 메시지 설정
messages = [
    SystemMessage(content="당신은 도움이 되는 AI 어시스턴트입니다. 한국어로 답변해주세요."),
    HumanMessage(content="파이썬에서 리스트 컴프리헨션을 사용하는 3가지 예제를 코드와 함께 설명해주세요.")
]

스트리밍 응답 생성
print("=" * 60)
print("HolySheep AI 스트리밍 응답 테스트")
print("=" * 60)

StreamHandler 인스턴스 생성 및 LLM 호출
stream_handler = StreamHandler(container=None)
response = llm.invoke(messages, callbacks=[stream_handler])

print("\n" + "=" * 60)
print("최종 응답 텍스트:")
print("=" * 60)
print(stream_handler.get_text())

비동기 스트리밍 구현

비동기 환경(AIOhttp, FastAPI)에서 스트리밍을 사용하려면 AsyncCallbackHandler를 구현해야 합니다.

import asyncio
from typing import Any, Awaitable, List, Union
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import Generation, ChatGeneration, LLMResult

class AsyncStreamHandler(BaseCallbackHandler):
    """
    비동기 환경용 스트리밍 핸들러
    FastAPI 또는 aiohttp와 함께 사용
    """
    
    def __init__(self, queue: asyncio.Queue):
        self.queue = queue
        self.first_token = True
        
    async def on_llm_new_token(
        self, 
        token: str, 
        **kwargs: Any
    ) -> None:
        """비동기 새 토큰 핸들러"""
        if self.first_token:
            await self.queue.put(f"\n🚀 응답 시작: {token}")
            self.first_token = False
        else:
            await self.queue.put(token)
            
    async def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """응답 완료 시 토큰 끝 표시"""
        await self.queue.put("\n\n✅ 응답 완료")
        await self.queue.put(None)  # 스트리밍 종료 신호
        self.first_token = True

async def stream_with_holysheep():
    """비동기 스트리밍 예제 함수"""
    from langchain_openai import ChatOpenAI
    
    llm = ChatOpenAI(
        model="gpt-4.1",
        streaming=True,
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1",
    )
    
    messages = [
        HumanMessage(content="인공지능의 미래에 대해 3문장으로 설명해주세요.")
    ]
    
    queue = asyncio.Queue()
    handler = AsyncStreamHandler(queue)
    
    # 백그라운드에서 LLM 실행
    async def run_llm():
        await llm.agenerate([messages], callbacks=[handler])
    
    # 병렬로 LLM 실행 및 출력
    await asyncio.gather(
        run_llm(),
        stream_output(queue)
    )

async def stream_output(queue: asyncio.Queue):
    """큐에서 토큰을 읽어 실시간 출력"""
    while True:
        token = await queue.get()
        if token is None:
            break
        print(token, end="", flush=True)

실행
if __name__ == "__main__":
    asyncio.run(stream_with_holysheep())

FastAPI 통합 예제

실제 웹 애플리케이션에서 SSE(Server-Sent Events)를 통한 스트리밍 응답을 구현하는 방법입니다.

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
from typing import AsyncGenerator

app = FastAPI(title="HolySheep AI Streaming API")

HolySheep AI 클라이언트 설정
def get_llm():
    return ChatOpenAI(
        model="gpt-4.1",
        streaming=True,
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1",
    )

class SSEStreamHandler:
    """SSE 포맷으로 변환하는 핸들러"""
    
    def __init__(self):
        self.queue = asyncio.Queue()
        
    async def on_llm_new_token(self, token: str, **kwargs) -> None:
        await self.queue.put(f"data: {json.dumps({'token': token})}\n\n")
        
    async def on_llm_end(self, response, **kwargs) -> None:
        await self.queue.put(f"data: {json.dumps({'done': True})}\n\n")
        await self.queue.put(None)
        
    async def stream(self) -> AsyncGenerator[str, None]:
        while True:
            token = await self.queue.get()
            if token is None:
                break
            yield token

@app.post("/stream")
async def stream_chat(request: Request):
    """
    POST /stream
    요청 본문: {"message": "질문 내용"}
    """
    body = await request.json()
    user_message = body.get("message", "")
    
    llm = get_llm()
    handler = SSEStreamHandler()
    
    messages = [HumanMessage(content=user_message)]
    
    # LLM 실행 (실제 구현에서는 asyncio.to_thread 사용 권장)
    async def run():
        # 동기 호출을 비동기 컨텍스트에서 실행
        for chunk in llm.stream(messages):
            await handler.on_llm_new_token(chunk.content)
        await handler.on_llm_end(None)
    
    asyncio.create_task(run())
    
    return StreamingResponse(
        handler.stream(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

성능 최적화 팁

• 버퍼 크기 조정: 네트워크 상황에 따라 max_tokens 설정값을 조절하여 첫 토큰 응답 시간을 개선하세요
• 연결 재사용: HTTP Keep-Alive를 활성화하여 연결 오버헤드를 줄입니다
• 비동기 처리: 다중 사용자 환경에서는 AsyncCallbackHandler 사용으로 동시성 성능을 향상시킵니다
• 에러 재시도 로직: 네트워크 불안정 시 자동 재시도机制 구현으로 안정성을 확보합니다

자주 발생하는 오류와 해결책

오류 1: StreamingCallbackHandler가 작동하지 않음

문제: callbacks 인자를 전달해도 스트리밍이 적용되지 않음

# ❌ 잘못된 사용법 - callbacks 전달 방식 오류
response = llm.invoke(messages)
response = llm.invoke(messages, callbacks=[handler])  # 이 방식은 동기 방식에서만 동작

✅ 올바른 사용법 - streaming=True와 함께 callbacks 전달
llm = ChatOpenAI(
    model="gpt-4.1",
    streaming=True,  # 반드시 True로 설정
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1",
)
response = llm.invoke(messages, callbacks=[stream_handler])  # 콜백 직접 전달

또는 bind() 메서드 사용
llm_with_callback = llm.bind(callbacks=[stream_handler])
response = llm_with_callback.invoke(messages)

오류 2: HolySheep AI 연결 실패 - Invalid API Key

문제: API 키 인증 오류로 연결이 거부됨

# ❌ 잘못된 설정
llm = ChatOpenAI(
    api_key="sk-...",  # HolySheep API 키 형식이 다름
    base_url="https://api.holysheep.ai/v1",
)

✅ 올바른 설정 - HolySheep 대시보드에서 생성한 키 사용
import os

환경 변수에서 안전하게 API 키 로드
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY")
if not HOLYSHEEP_API_KEY:
    raise ValueError(
        "HOLYSHEEP_API_KEY가 설정되지 않았습니다. "
        "https://www.holysheep.ai/register에서 API 키를 생성하세요."
    )

llm = ChatOpenAI(
    model="gpt-4.1",
    streaming=True,
    api_key=HOLYSHEEP_API_KEY,
    base_url="https://api.holysheep.ai/v1",
)

연결 테스트
try:
    test_response = llm.invoke([HumanMessage(content="test")])
    print("✅ HolySheep AI 연결 성공")
except Exception as e:
    print(f"❌ 연결 실패: {e}")

오류 3: async/await 환경에서 스트리밍 핸들러 미작동

문제: 비동기 함수에서 콜백 핸들러가 토큰을 캡처하지 못함

# ❌ 잘못된 비동기 구현
async def wrong_async_stream():
    llm = ChatOpenAI(
        streaming=True,
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1",
    )
    # agenerate가 없거나 콜백이 제대로 전달되지 않음
    response = llm.generate([messages])

✅ 올바른 비동기 구현 - agenerate 메서드 사용
from langchain_core.callbacks import BaseCallbackHandler
from typing import AsyncIterator

class AsyncTokenHandler(BaseCallbackHandler):
    def __init__(self):
        self.tokens = []
        
    async def on_llm_new_token(self, token: str, **kwargs) -> None:
        self.tokens.append(token)
        # 즉시 출력하지 않고 큐에 전달
        print(token, end="", flush=True)

async def correct_async_stream():
    llm = ChatOpenAI(
        model="gpt-4.1",
        streaming=True,
        api_key="YOUR_HOLYSHEEP_API_KEY",
        base_url="https://api.holysheep.ai/v1",
    )
    
    handler = AsyncTokenHandler()
    messages = [HumanMessage(content="인공지능에 대해 설명해주세요.")]
    
    # agenerate를 사용하여 비동기 스트리밍
    response = await llm.agenerate(
        [messages],
        callbacks=[handler]
    )
    
    return "".join(handler.tokens)

실행
result = asyncio.run(correct_async_stream())
print(f"\n최종 결과: {result}")

오류 4: Rate Limit 초과

문제: 요청过多导致速率限制

import time
from functools import wraps

class RateLimitedHandler:
    """요청 간 지연 시간을 자동 적용하는 핸들러"""
    
    def __init__(self, min_interval: float = 0.1):
        self.min_interval = min_interval
        self.last_call = 0
        
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        now = time.time()
        elapsed = now - self.last_call
        
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
            
        self.last_call = time.time()
        print(token, end="", flush=True)

또는 HolySheep AI 대시보드에서 RPM/RPD 제한 확인
기본 제한: RPM 60, RPD 1000 (플랜에 따라 상이)

실전 활용 사례

저는 실제 서비스 개발에서 HolySheep AI의 스트리밍 기능을 다음과 같은 시나리오에 적용했습니다:

• 실시간 챗봇: 사용자가 입력하는 동안 AI가 응답을 생성하여 마치 대화하는 느낌 제공
• 코드 작성 어시스턴트: 코드 조각이 실시간으로 출력되어 개발 생산성 향상
• 긴 형식 콘텐츠 생성: 블로그 포스트나 문서 작성 시 중간 결과를 확인하여 불필요한 대기 시간 제거

결론

LangChain의 StreamingCallbackHandler를 활용하면 HolySheep AI의 저지연 스트리밍을 쉽게 구현할 수 있습니다. HolySheep AI는:

• 로컬 결제 지원으로 해외 신용카드 없이 즉시 시작 가능
• 단일 API 키로 GPT-4.1, Claude, Gemini, DeepSeek 등 다중 모델 통합
• 경쟁력 있는 가격 ($8~$15/MTok) 및 무료 크레딧 제공

구현有问题이나 추가 질문이 있으시면 HolySheep AI의 기술 문서를 참고하세요.

---

👉 HolySheep AI 가입하고 무료 크레딧 받기