이벤트 스트리밍 (Event Streaming)

이벤트 스트리밍 (Event Streaming)

LangChain 에이전트는 LangGraph 위에서 만들어지기 때문에, 메시지·도구 호출·상태·커스텀 업데이트를 위한 에이전트 중심 프로젝션을 갖춘 동일한 스트리밍 스택을 그대로 지원해요. 대부분의 애플리케이션·프론트엔드 사용 사례에서는 stream_events(..., version="v3")를 쓰는 **이벤트 스트리밍(Event Streaming)**을 권장합니다. 이벤트 스트리밍은 타입이 지정된 프로젝션을 가진 run 객체를 반환해서, 스트림 모드 튜플을 직접 파싱하는 대신 각 프로젝션을 독립적으로 소비할 수 있어요.

from langchain.agents import create_agent

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

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
)

stream = agent.stream_events({
    "messages": [{"role": "user", "content": "What is the weather in SF?"}],
}, version="v3")

for message in stream.messages:
    for delta in message.text:
        print(delta, end="", flush=True)

final_state = stream.output

스트리밍할 수 있는 것들

stream.messages는 ChatModelStream 객체를 생성해요. 각 메시지 스트림은 .text, .reasoning, .tool_calls, .output을 노출합니다. 동기 프로젝션은 라이브 델타를 위해 반복(iterable)할 수 있고, 최종 값을 위해 끝까지 소진(drain)할 수도 있어요. 최종 텍스트는 str(message.text)로, 확정된 도구 호출은 message.tool_calls.get()으로 가져옵니다.

에이전트 메시지

각 LLM 호출의 모델 출력이 필요할 때 stream.messages를 사용하세요.

stream = agent.stream_events(input, version="v3")

for message in stream.messages:
    print(f"[{message.node}] ", end="")
    for delta in message.text:
        print(delta, end="", flush=True)

    full_message = message.output
    usage = full_message.usage_metadata
    if usage:
        print(usage)

message.output은 공급자별 콘텐츠 블록을 포함한 확정된 AI 메시지를 줍니다. TypeScript에서는 토큰 수나 다른 사용량 메타데이터만 필요할 때 message.usage를 쓰고, Python에서는 message.output.usage_metadata에서 사용량을 읽어요.

추론 콘텐츠 (Reasoning content)

추론 콘텐츠는 텍스트 콘텐츠와 같은 형태를 갖지만, 선택한 모델이 추론 블록을 내보낼 때만 사용할 수 있어요.

stream = agent.stream_events(input, version="v3")

for message in stream.messages:
    for delta in message.reasoning:
        print(f"[thinking] {delta}", end="", flush=True)

    for delta in message.text:
        print(delta, end="", flush=True)

모델 구성 세부 사항은 reasoning 가이드와 공급자 통합 페이지를 참고하세요.

도구 호출

유용한 도구 호출 프로젝션은 두 가지가 있어요.

• message.tool_calls는 모델이 도구 호출을 만들고 있는 동안 도구 호출 인자 청크를 스트리밍합니다.
• stream.tool_calls는 도구 호출이 시작된 후 도구 실행의 생명주기를 스트리밍해요.

stream = agent.stream_events(input, version="v3")

for message in stream.messages:
    for chunk in message.tool_calls:
        print(f"tool call chunk: {chunk}")

    finalized = message.tool_calls.get()
    if finalized:
        print(f"finalized tool calls: {finalized}")

for call in stream.tool_calls:
    print(f"{call.tool_name}({call.input})")
    for delta in call.output_deltas:
        print(delta, end="", flush=True)
    print(call.output, call.error)

서브 에이전트 스트리밍

create_agent 호출이 다른 이름 붙은 create_agent를 (보통 래핑 도구를 통해) 호출하면, 안쪽 에이전트의 이벤트가 중첩된 네임스페이스로 흐릅니다. create_agent에 넘기는 name=이 그 안쪽 에이전트를 스트림 안에서 식별해 주므로, 에이전트별로 필터링하고 라벨을 붙일 수 있어요. 이름 붙은 서브 에이전트는 전용 stream.subagents 프로젝션에 나타납니다. 각 핸들은 안쪽 에이전트 자신의 .messages, .values, .tool_calls, .output과 함께 .name(넘겨준 name= 값), .cause(서브 에이전트를 파견한 도구 호출)를 노출해요. 이름 붙은 create_agent 실행만 여기에 나타나기 때문에, 일반 서브그래프를 필터링할 필요가 없습니다.

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

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

weather_agent = create_agent(
    model=init_chat_model("openai:gpt-5.5"),
    tools=[get_weather],
    name="weather_agent",
)

def call_weather(query: str) -> str:
    """Query the weather agent."""
    result = weather_agent.invoke({"messages": [{"role": "user", "content": query}]})
    return result["messages"][-1].text

supervisor = create_agent(
    model=init_chat_model("openai:gpt-5.5"),
    tools=[call_weather],
    name="supervisor",
)

stream = supervisor.stream_events(
    {"messages": [{"role": "user", "content": "What's the weather in Boston?"}]},
    version="v3",
)

for subagent in stream.subagents:
    print(f"{subagent.name}: ", end="")
    for message in subagent.messages:
        for token in message.text:
            print(token, end="", flush=True)
    print()

도구에서 호출되는 일반 StateGraph 서브그래프도 stream.subgraphs에 나타납니다. .compile(name=...)에 name=을 설정하면 subagent.graph_name에 라벨이 붙어요. stream.subagents는 이름 붙은 create_agent 서브 에이전트를 보기에 좋은 시점이고, stream.subgraphs는 모든 중첩 그래프를 다룹니다. UI에 맞는 쪽을 쓰면 돼요.

상태와 최종 출력

상태 스냅샷에는 stream.values를, 최종 에이전트 상태에는 stream.output을 사용하세요.

stream = agent.stream_events(input, version="v3")

for snapshot in stream.values:
    print(snapshot)

final_state = stream.output

여러 프로젝션

비동기 코드에서 여러 프로젝션을 동시에 소비하려면 astream_events를 asyncio.gather와 함께 쓰세요.

import asyncio

stream = await agent.astream_events(input, version="v3")

async def consume_messages():
    async for message in stream.messages:
        print(await message.text)

async def consume_tool_calls():
    async for call in stream.tool_calls:
        print(call.tool_name, call.input)

await asyncio.gather(consume_messages(), consume_tool_calls())

동기 코드에서는 대신 stream.interleave(...)를 씁니다.

stream = agent.stream_events(input, version="v3")

for name, item in stream.interleave("messages", "tool_calls", "values"):
    if name == "messages":
        print(item.text)
    elif name == "tool_calls":
        print(item.tool_name, item.input)
    elif name == "values":
        print(item)

타입이 지정된 프로젝션으로 노출되지 않는 채널에 접근하거나 전체 이벤트 봉투를 확인해야 한다면, 원시 프로토콜 이벤트를 반복하면 됩니다.

for event in stream:
    print(event["method"], event["params"]["namespace"], event["params"]["data"])

커스텀 업데이트

내장되지 않은 프로젝션이 필요할 때, 예를 들어 검색 진행 상황이나 아티팩트, 도메인 특화 이벤트를 다룰 때는 커스텀 스트림 트랜스포머를 사용하세요.

stream = agent.stream_events(
    input,
    version="v3",
    transformers=[ToolActivityTransformer],
)

for activity in stream.extensions["tool_activity"]:
    print(activity)

미들웨어에 트랜스포머 등록하기

미들웨어에 등록된 트랜스포머는 langchain>=1.3.2가 필요합니다.

미들웨어는 훅과 도구와 함께 스트림 트랜스포머 팩토리를 선언할 수 있어요. 팩토리 형태는 언어별로 다릅니다. AgentMiddleware 하위 클래스에 transformers 속성을 팩토리 시퀀스로 설정합니다. 각 팩토리는 Callable[[tuple[str, ...]], StreamTransformer] 형태를 가지며 factory(scope)로 호출되는데, 여기서 scope는 미니-멀티플렉서(mux) 범위 튜플입니다(루트 mux는 (), 서브그래프는 비어 있지 않은 값). 호출마다 새 트랜스포머를 반환하면 각 서브그래프를 격리된 상태로 유지할 수 있어요.

from langchain.agents import create_agent
from langchain.agents.middleware import AgentMiddleware

class ToolActivityMiddleware(AgentMiddleware):
    transformers = (ToolActivityTransformer,)

agent = create_agent(
    model="gpt-5-nano",
    tools=[get_weather],
    middleware=[ToolActivityMiddleware()],
)

컴파일 타임에 create_agent는 미들웨어에 등록된 팩토리와 자신의 transformers= 인자로 넘어온 것을 합칩니다. 컴파일된 그래프에서 최종 순서는 다음과 같습니다.

1. 내장 ToolCallTransformer
2. 미들웨어 순서대로 등록된 미들웨어 팩토리
3. create_agent의 transformers=로 호출자가 넘긴 것

이렇게 하면 내장 도구 호출 프로젝션이 소비자 트랜스포머보다 앞에 오고, 호출자가 넘긴 항목이 마지막 결정권을 가져요. 내장 PIIMiddleware는 이 훅을 사용해 스트리밍된 통신 출력에서 PII를 가립니다. apply_to_output=True를 설정하면 등록된 트랜스포머가 텍스트 델타, 도구 호출 인자, 도구 출력, 상태 스냅샷에서 감지된 PII를 run에서 나가기 전에 지워서, after_model 상태 단계의 가림만으로는 원시 PII가 stream_events(version="v3")의 실시간 독자에게 새어 나갈 수 있는 틈을 막아줍니다.

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware

agent = create_agent(
    model="gpt-5-nano",
    tools=[],
    middleware=[
        PIIMiddleware("email", strategy="redact", apply_to_output=True),
    ],
)

전체 구성 표면은 PII 검출을, 트랜스포머 계약은 나만의 프로젝션 만들기를 참고하세요.

관련 자료

• Streaming — 저수준 Pregel 스트림 모드를 다룹니다.
• Build your own projection — 애플리케이션 특화 프로젝션 작성법을 다룹니다.
• Frontend streaming patterns — 스트리밍된 상태를 활용한 UI 사용 사례를 보여줍니다.