9장: 스트리밍과 실시간 처리 패턴

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
 
app = FastAPI()
 
@app.get("/chat/stream")
async def stream_chat(question: str):
    model = ChatOpenAI(model="gpt-4o", streaming=True)
 
    async def event_generator():
        async for chunk in model.astream(question):
            if chunk.content:
                yield f"data: {chunk.content}\n\n"
        yield "data: [DONE]\n\n"
 
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

from fastapi import FastAPI, WebSocket
 
app = FastAPI()
 
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
 
    while True:
        data = await websocket.receive_json()
        question = data["question"]
 
        model = ChatOpenAI(model="gpt-4o", streaming=True)
 
        async for chunk in model.astream(question):
            if chunk.content:
                await websocket.send_json({
                    "type": "token",
                    "content": chunk.content,
                })
 
        await websocket.send_json({"type": "done"})

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
 
model = ChatOpenAI(model="gpt-4o", streaming=True)
prompt = ChatPromptTemplate.from_template("{question}")
chain = prompt | model
 
# 동기 스트리밍
for chunk in chain.stream({"question": "Python의 장점은?"}):
    print(chunk.content, end="", flush=True)
 
# 비동기 스트리밍
async for chunk in chain.astream({"question": "Python의 장점은?"}):
    print(chunk.content, end="", flush=True)

config = {"configurable": {"thread_id": "user-123"}}
 
# 노드별 스트리밍
async for event in app.astream(
    {"messages": [HumanMessage("분석해줘")]},
    config=config,
    stream_mode="values",
):
    # 각 노드 완료 시 전체 상태 수신
    messages = event.get("messages", [])
    if messages:
        print(f"Latest: {messages[-1].content}")
 
# 토큰 단위 스트리밍
async for event in app.astream(
    {"messages": [HumanMessage("분석해줘")]},
    config=config,
    stream_mode="messages",
):
    # 개별 토큰 수신
    message, metadata = event
    if message.content:
        print(message.content, end="", flush=True)

from llama_index.core import VectorStoreIndex
 
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(streaming=True)
 
# 스트리밍 응답
streaming_response = query_engine.query("LlamaIndex의 특징은?")
 
for token in streaming_response.response_gen:
    print(token, end="", flush=True)

from haystack.components.generators import OpenAIGenerator
 
generator = OpenAIGenerator(
    model="gpt-4o",
    streaming_callback=lambda chunk: print(chunk.content, end=""),
)
 
# 스트리밍 콜백 방식
result = generator.run(prompt="Haystack의 특징은?")

async for event in chain.astream_events(
    {"question": "Python의 GIL에 대해 설명해줘"},
    version="v2",
):
    kind = event["event"]
    name = event["name"]
 
    if kind == "on_chat_model_start":
        print(f"\n--- 모델 호출 시작: {name} ---")
 
    elif kind == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="", flush=True)
 
    elif kind == "on_chat_model_end":
        usage = event["data"]["output"].usage_metadata
        print(f"\n--- 모델 호출 완료 ---")
        print(f"토큰 사용: 입력 {usage['input_tokens']}, "
              f"출력 {usage['output_tokens']}")
 
    elif kind == "on_tool_start":
        print(f"\n[도구 호출] {event['data']['input']}")
 
    elif kind == "on_tool_end":
        print(f"[도구 결과] {event['data']['output']}")

async for event in app.astream_events(
    {"messages": [HumanMessage("날씨를 확인하고 옷차림을 추천해줘")]},
    config=config,
    version="v2",
):
    if event["event"] == "on_chain_start":
        if event["metadata"].get("langgraph_node"):
            node = event["metadata"]["langgraph_node"]
            print(f"\n=== 노드 시작: {node} ===")
 
    elif event["event"] == "on_chat_model_stream":
        content = event["data"]["chunk"].content
        if content:
            print(content, end="", flush=True)

from langchain_openai import ChatOpenAI
from pydantic import BaseModel
 
class AnalysisResult(BaseModel):
    summary: str
    key_points: list[str]
    sentiment: str
    confidence: float
 
model = ChatOpenAI(model="gpt-4o")
structured_model = model.with_structured_output(
    AnalysisResult,
    method="json_schema",
)
 
# 구조화된 출력 스트리밍
# 각 청크가 부분적으로 채워진 Pydantic 모델로 전달됨
async for partial in structured_model.astream(
    "이 기사를 분석해주세요: ..."
):
    if partial.summary:
        print(f"요약 (진행 중): {partial.summary}")
    if partial.key_points:
        print(f"핵심 포인트: {len(partial.key_points)}개 발견")

import { useState, useCallback } from 'react'
 
interface StreamState {
  content: string
  isStreaming: boolean
  error: string | null
}
 
function useStreamChat() {
  const [state, setState] = useState<StreamState>({
    content: '',
    isStreaming: false,
    error: null,
  })
 
  const sendMessage = useCallback(async (question: string) => {
    setState({ content: '', isStreaming: true, error: null })
 
    try {
      const response = await fetch(
        `/api/chat/stream?question=${encodeURIComponent(question)}`
      )
 
      const reader = response.body?.getReader()
      const decoder = new TextDecoder()
 
      if (!reader) throw new Error('Stream not available')
 
      while (true) {
        const { done, value } = await reader.read()
        if (done) break
 
        const text = decoder.decode(value)
        const lines = text.split('\n')
 
        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = line.slice(6)
            if (data === '[DONE]') break
 
            setState(prev => ({
              ...prev,
              content: prev.content + data,
            }))
          }
        }
      }
    } catch (error) {
      setState(prev => ({
        ...prev,
        error: error instanceof Error ? error.message : 'Unknown error',
      }))
    } finally {
      setState(prev => ({ ...prev, isStreaming: false }))
    }
  }, [])
 
  return { ...state, sendMessage }
}

import { streamText } from 'ai'
import { openai } from '@ai-sdk/openai'
 
export async function POST(request: Request) {
  const { messages } = await request.json()
 
  const result = streamText({
    model: openai('gpt-4o'),
    messages,
  })
 
  return result.toDataStreamResponse()
}

'use client'
 
import { useChat } from 'ai/react'
 
export default function ChatPage() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } =
    useChat({
      api: '/api/chat',
    })
 
  return (
    <div>
      {messages.map(m => (
        <div key={m.id}>
          <strong>{m.role}:</strong> {m.content}
        </div>
      ))}
      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={handleInputChange}
          placeholder="메시지를 입력하세요..."
          disabled={isLoading}
        />
      </form>
    </div>
  )
}

import asyncio
 
async def buffered_stream(chain, input_data, buffer_size: int = 5):
    """토큰을 버퍼링하여 네트워크 효율성 개선"""
    buffer = []
 
    async for chunk in chain.astream(input_data):
        if chunk.content:
            buffer.append(chunk.content)
 
            if len(buffer) >= buffer_size:
                yield "".join(buffer)
                buffer = []
 
    # 남은 버퍼 전송
    if buffer:
        yield "".join(buffer)

async def stream_with_timeout(chain, input_data, timeout: float = 30.0):
    """스트리밍에 타임아웃 적용"""
    try:
        async with asyncio.timeout(timeout):
            async for chunk in chain.astream(input_data):
                yield chunk
    except asyncio.TimeoutError:
        yield {"type": "error", "message": "응답 시간이 초과되었습니다."}