[LangChain 함수 3] Streaming을 통해 결과 반환반기 (stream)

왜 Streaming이 중요한가

---

 

대화형 챗봇이나 고객 응대 Agent를 만들다 보면, “응답이 너무 늦다”는 문제가 가장 먼저 드러난다.
모델이 모든 출력을 다 생성한 뒤 한 번에 결과를 반환하는 방식은, 사용자 입장에서는 항상 느리게 느껴진다.

 

LangChain은 이런 문제를 해결하기 위해 Streaming을 기본 기능으로 제공한다.

Streaming이란, 최종 결과가 완성되기 전에 생성되는 데이터를 사용자에게 즉시 전달하는 방식이다.

특히 다음과 같은 경우에 매우 유용하다.

• 챗봇, 고객 상담 Agent처럼 상호작용성이 중요한 서비스
• 모델 응답이 길어질 가능성이 있는 경우
• Agent가 내부적으로 여러 step(추론, 도구 호출 등)을 거치는 경우

LangChain의 Agent는 내부적으로 loop 구조를 가지기 때문에,
Streaming을 사용하면 이 loop의 중간 과정까지 사용자에게 그대로 노출할 수 있다.

 

 

 

Streaming 없이 호출하기: invoke

---

가장 기본적인 Agent 호출 방식은 .invoke 방식이다.

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Tell me a joke"}]}
)
print(result["messages"][-1].content)

 

이 방식의 특징은 다음과 같다.

• Agent의 전체 실행이 끝난 후 결과를 한 번에 반환한다
• 토큰 단위 출력이나 중간 단계는 볼 수 없다
• 구현은 단순하지만, 지연 시간이 가장 크다

Streaming은 이 invoke 방식의 대안이라고 볼 수 있다.

 

 

 

Streaming 모드 1 : values 스트리밍 (step 단위 출력)

---

values 스트리밍 모드는 Agent loop의 각 step이 끝날 때마다 결과를 반환한다.
즉, 모델 호출 후, Tool 호출 후 등 주요 단계마다 상태를 확인할 수 있다.

for step in agent.stream(
    {"messages": [{"role": "user", "content": "Tell me a Dad joke"}]},
    stream_mode="values",
):
    step["messages"][-1].pretty_print()

 

이 모드의 특징은 다음과 같다.

• step 단위로 Message 상태를 확인할 수 있다
• HumanMessage → AIMessage → ToolMessage 흐름을 그대로 관찰할 수 있다
• 디버깅이나 Agent 내부 동작 이해에 매우 유용하다

 

 

Streaming 모드 2 : messsages 스트리밍 (토큰 단위 출력)

---

messages 스트리밍은 가장 낮은 지연 시간을 제공하는 방식이다.
모델이 토큰을 생성하는 즉시 사용자에게 전달된다.

for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "Write me a family friendly poem."}]},
    stream_mode="messages",
):
    print(token.content, end="")

 

이 방식은 다음과 같은 특징을 가진다.

• 토큰 단위로 출력이 즉시 전달된다
• 사용자 입장에서 “생각하며 답하는 것처럼” 보인다
• 챗봇, 실시간 인터랙션 UI에 가장 적합하다

ChatGPT 스타일의 실시간 응답을 구현하고 싶다면, 이 모드가 사실상 표준 선택이다.

 

 

 

Streamingg 모드 3 : custom 모드 (Tool에서도 streaming하기)

---

 

LangChain의 강력한 점 중 하나는 Tool에서도 Streaming이 가능하다는 점이다.
Tool 내부에서 아래 예제와 같이 get_stream_writer()를 사용하면, Tool 실행 중 임의의 데이터를 스트리밍할 수 있다.

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    writer = get_stream_writer()
    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")
    return f"It's always sunny in {city}!"

 

이 Tool을 Agent에 연결하고 실행하면 다음과 같이 사용할 수 있다.

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "What is the weather in SF?"}]},
    stream_mode=["custom"],
):
    print(chunk)