스트리밍

스트리밍 (Streaming)

LLM 기반 앱을 만들다 보면 답이 완전히 나올 때까지 사용자가 하염없이 기다리게 만들고 싶지 않을 거예요. 스트리밍(streaming)은 그 문제를 해결해 줍니다. LangChain은 실시간 업데이트를 표면화하는 스트리밍 시스템을 구현해요. 완전한 응답이 준비되기 전에도 출력을 점진적으로 표시하면서, 특히 LLM의 지연 시간이 길 때 사용자 경험(UX)을 크게 개선하죠.

출처: LangChain 공식 문서 — Streaming

새 애플리케이션에는 이벤트 스트리밍(event streaming)을 권장해요. 이것은 LangChain v1.3에서 도입된 타입 기반 프로젝션 API로, 프로젝션(메시지, 값, 도구 호출, 서브그래프)별로 별도의 이터레이터를 주기 때문에 stream_mode 청크로 분기하는 대신 각각을 독립적으로 소비할 수 있어요.

개요 (Overview)

LangChain의 스트리밍 시스템은 에이전트 실행의 라이브 피드백을 애플리케이션에 표면화할 수 있게 해줍니다. LangChain 스트리밍으로 가능한 것들:

추가적인 end-to-end 예시는 아래 일반적인 패턴(common patterns) 섹션을 참고하세요.

지원되는 스트림 모드 (Supported stream modes)

stream 또는 astream 메서드에 다음 스트림 모드 중 하나 이상을 리스트로 전달해요.

모드 설명
updates 각 에이전트 단계 후 state 업데이트를 스트리밍. 한 단계에서 여러 업데이트(예: 여러 노드 실행)가 있으면 각각 별도로 스트리밍.
messages LLM이 호출되는 그래프 노드에서 (token, metadata) 튜플을 스트리밍.
custom 스트림 라이터를 사용해 그래프 노드 안에서 커스텀 데이터를 스트리밍.

에이전트 진행 (Agent progress)

에이전트 진행을 스트리밍하려면 stream_mode="updates"와 함께 stream 또는 astream 메서드를 씁니다. 이러면 매 에이전트 단계 후 이벤트가 방출돼요. 예를 들어 도구를 한 번 호출하는 에이전트라면 다음 업데이트를 볼 수 있어요.

  • LLM 노드: 도구 호출 요청이 담긴 AIMessage
  • 도구 노드: 실행 결과가 담긴 ToolMessage
  • LLM 노드: 최종 AI 응답

config를 통해 thread_id를 전달하면 대화가 체크포인트되고 후속 턴이 같은 기록을 이어갈 수 있어요. thread_id는 stream_mode와 독립적이며, 그 옆에 도구가 runtime.context에서 읽는 실행별 데이터를 위한 context도 함께 전달할 수 있어요.

Google 모델 기준 예시:

from langchain.agents import create_agent
from langchain_core.utils.uuid import uuid7
from langgraph.checkpoint.memory import InMemorySaver

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

agent = create_agent(
    model="google_genai:gemini-3.6-flash",
    tools=[get_weather],
    checkpointer=InMemorySaver()
)
config = {"configurable": {"thread_id": str(uuid7())}}
stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    config=config,
    version="v3",
)
for kind, item in stream.interleave("messages", "tool_calls"):
    if kind == "messages":
        for token in item.text:
            print(token, end="", flush=True)
    elif kind == "tool_calls":
        print(f"\nTool call: {item.tool_name}({item.input})")
        for delta in item.output_deltas:
            print(delta, end="", flush=True)
        print(f"\nTool result: {item.output}")

final_state = stream.output

OpenAI, Anthropic, OpenRouter, Fireworks, Baseten, Ollama 등 다른 제공자도 모델 문자열만 다르고 동일한 패턴을 사용합니다. stream_events와 version="v3", interleave("messages", "tool_calls") 조합으로 메시지 토큰과 도구 호출을 별도로 처리하면서 실시간으로 출력하죠.

LLM 토큰 스트리밍 (Stream LLM tokens)

stream_mode="messages"를 쓰면 LLM이 생성하는 토큰을 실시간으로 스트리밍할 수 있어요. (token, metadata) 튜플이 나오고, metadata에는 메시지 타입, 모델, 토큰 수 등이 담겨요. 이를 통해 답변이 완성되기 전에도 사용자에게 글자가 타이핑되는 것처럼 보여줄 수 있습니다.

요약 (Summary)

  • updates 모드는 에이전트 진행(상태 전이)을 보고 싶을 때, messages 모드는 LLM 토큰을 실시간으로 보여주고 싶을 때, custom 모드는 앱이 정의한 진행 상황(예: "Fetched 10/100 records")을 스트리밍할 때 써요.
  • stream_events는 interleave로 메시지와 도구 호출을 함께 소비할 수 있는데, 이 페이지 예시들이 이 방식을 사용해요.
  • thread_id를 넘기면 대화가 체크포인트되어 후속 턴을 재개할 수 있어요.

모델별 스트리밍 비활성화 (Disable streaming)

어떤 모델은 스트리밍이 필요 없거나 지원하지 않아요. LangSmith에 배포할 때 클라이언트로 스트리밍하고 싶지 않은 모델의 출력에는 streaming=False를 설정하세요. 이는 배포 전에 그래프 코드에서 구성합니다.

모든 채팅 모델 통합이 streaming 파라미터를 지원하는 건 아니에요. 지원하지 않는 모델이라면 disable_streaming=True를 대신 쓰세요. 이 파라미터는 기본 클래스를 통해 모든 채팅 모델에서 사용할 수 있어요.

자세한 내용은 LangGraph 스트리밍 가이드를 참고하세요.

v2 스트리밍 형식 (v2 streaming format)

LangGraph >= 1.1이 필요해요.

통합된 출력 형식을 얻으려면 stream() 또는 astream()에 version="v2"를 전달하세요. 모든 청크는 type, ns, data 키를 가진 StreamPart dict로, 스트림 모드나 모드 수와 관계없이 동일한 형태예요.

v2(신규):

# Unified format — no more tuple unpacking
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["updates", "custom"],
    version="v2",
):
    print(chunk["type"])  # "updates" or "custom"
    print(chunk["data"])  # payload

v1(현재 기본값):

# Must unpack (mode, data) tuples
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["updates", "custom"],
):
    print(mode)   # "updates" or "custom"
    print(chunk)  # payload

v2 형식은 invoke()도 개선해요. .value와 .interrupts 속성을 가진 GraphOutput 객체를 반환하면서 state와 interrupt 메타데이터를 깔끔하게 분리하죠.

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}]},
    version="v2",
)
print(result.value)       # state (dict, Pydantic model, or dataclass)
print(result.interrupts)  # tuple of Interrupt objects (empty if none)

v2 형식에 대한 더 자세한 내용(타입 좁히기, Pydantic/dataclass 강제 변환, 서브그래프 스트리밍 포함)은 LangGraph 스트리밍 문서를 참고하세요.

더 알아보기 (Learn more)