[langgraph] 스트리밍

원문 출처: https://langchain-ai.github.io/langgraph/agents/streaming/

스트리밍은 반응성 높은 애플리케이션을 구축하는 데 핵심적인 기능입니다. 스트리밍할 수 있는 데이터 유형은 다음과 같습니다:

에이전트 진행 상황(Agent progress) — 에이전트 그래프의 각 노드가 실행될 때마다 업데이트를 받습니다.
LLM 토큰(LLM tokens) — 언어 모델이 토큰을 생성할 때마다 실시간으로 스트림합니다.
커스텀 업데이트(Custom updates) — 도구 실행 중 임의의 데이터를 스트림으로 내보낼 수 있습니다(예: "100개 중 10개 레코드 조회 완료").

여러 유형의 데이터를 동시에 스트리밍할 수도 있습니다.

Waiting is for pigeons.

에이전트 진행 상황(Agent progress)

에이전트 진행 상황을 스트리밍하려면 stream() 또는 astream() 메서드에 stream_mode="updates"를 지정하세요. 에이전트의 각 단계가 끝날 때마다 이벤트가 발생합니다.

예를 들어, 에이전트가 도구를 한 번 호출하는 경우 다음과 같은 업데이트를 볼 수 있습니다:

LLM 노드: 도구 호출 요청이 포함된 AI 메시지
도구 노드: 실행 결과가 담긴 도구 메시지
LLM 노드: 최종 AI 응답

동기 예시

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="updates"
):
    print(chunk)
    print("\n")

비동기 예시

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="updates"
):
    print(chunk)
    print("\n")

LLM 토큰(LLM tokens)

LLM이 생성하는 토큰을 실시간으로 스트리밍하려면 stream_mode="messages"를 사용하세요.

동기 예시

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
for token, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="messages"
):
    print("Token", token)
    print("Metadata", metadata)
    print("\n")

비동기 예시

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)
async for token, metadata in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="messages"
):
    print("Token", token)
    print("Metadata", metadata)
    print("\n")

도구 업데이트(Tool updates)

도구 실행 중 임의의 데이터를 스트림하려면 get_stream_writer를 사용할 수 있습니다.

동기 예시

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """주어진 도시의 날씨를 조회합니다."""
    writer = get_stream_writer()
    # 임의의 데이터 스트림
    writer(f"도시 데이터 조회 중: {city}")
    return f"{city}의 날씨는 항상 맑음!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="custom"
):
    print(chunk)
    print("\n")

비동기 예시

from langgraph.config import get_stream_writer

def get_weather(city: str) -> str:
    """주어진 도시의 날씨를 조회합니다."""
    writer = get_stream_writer()
    # 임의의 데이터 스트림
    writer(f"도시 데이터 조회 중: {city}")
    return f"{city}의 날씨는 항상 맑음!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

async for chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode="custom"
):
    print(chunk)
    print("\n")

참고: 도구 내부에 get_stream_writer를 추가하면 LangGraph 실행 컨텍스트 외부에서는 해당 도구를 호출할 수 없습니다.

다중 스트리밍 모드(Stream multiple modes)

여러 스트리밍 모드를 동시에 지정하려면 stream_mode에 리스트를 전달하세요: stream_mode=["updates", "messages", "custom"]

동기 예시

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

for stream_mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")

비동기 예시

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_weather],
)

async for stream_mode, chunk in agent.astream(
    {"messages": [{"role": "user", "content": "what is the weather in sf"}]},
    stream_mode=["updates", "messages", "custom"]
):
    print(chunk)
    print("\n")