스트리밍

스트리밍 (Streaming)

딥 에이전트 실행과 서브에이전트 실행에서 실시간 업데이트를 스트리밍하는 방법을 설명할게요. 새로운 애플리케이션이라면 Deep Agents v0.6에서 도입된 이벤트 스트리밍 — 타입 기반 프로젝션 API — 을 쓰는 걸 권장해요. 이벤트 스트리밍은 프로젝션별(서브에이전트, 메시지, 도구 호출, 값)로 분리된 이터레이터를 제공해서, stream_mode 청크를 분기 처리하는 대신 각각 독립적으로 소비할 수 있어요.

Deep Agents는 LangGraph의 스트리밍 인프라 위에 서브에이전트 스트림에 대한 일급 지원을 얹어요. 딥 에이전트가 서브에이전트에 작업을 위임하면 각 서브에이전트의 업데이트를 독립적으로 스트리밍해서 진행 상황, LLM 토큰, 도구 호출을 실시간으로 추적할 수 있어요.

딥 에이전트 스트리밍으로 가능한 것:

  • 서브에이전트 진행 스트리밍 — 병렬로 실행되는 각 서브에이전트의 실행을 추적
  • LLM 토큰 스트리밍 — 메인 에이전트와 각 서브에이전트의 토큰을 스트리밍
  • 도구 호출 스트리밍 — 서브에이전트 실행 내부의 도구 호출과 결과 확인
  • 커스텀 업데이트 스트리밍 — 서브에이전트 노드 내부에서 사용자 정의 신호를 emit

출처: 공식문서

서브그래프 스트리밍 활성화 (Enable subgraph streaming)

Deep Agents는 LangGraph의 서브그래프 스트리밍을 사용해 서브에이전트 실행의 이벤트를 표면화해요. 서브에이전트 이벤트를 받으려면 스트리밍할 때 stream_subgraphs를 활성화해야 해요.

from deepagents import create_deep_agent

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    system_prompt="You are a helpful research assistant",
    subagents=[
        {
            "name": "researcher",
            "description": "Researches a topic in depth",
            "system_prompt": "You are a thorough researcher.",
        },
    ],
)

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        if chunk["ns"]:
            # Subagent event - namespace identifies the source
            print(f"[subagent: {chunk['ns']}]")
        else:
            # Main agent event
            print("[main agent]")
        print(chunk["data"])

이 예제는 Google(google_genai:gemini-3.6-flash) 기준이며, OpenAI(openai:gpt-5.5), Anthropic(anthropic:claude-sonnet-4-6), OpenRouter(openrouter:z-ai/glm-5.2), Fireworks, Baseten, Ollama 등 다른 프로바이더도 model 값만 바꾸면 동일한 코드로 동작해요. chunk["ns"](네임스페이스)가 있으면 서브에이전트 이벤트, 비어 있으면 메인 에이전트 이벤트란 뜻이에요.

LLM 토큰 스트리밍 (LLM tokens)

stream_mode="messages"를 사용하면 메인 에이전트와 서브에이전트 양쪽의 개별 토큰을 스트리밍할 수 있어요. 각 메시지 이벤트에는 출처 에이전트를 식별하는 메타데이터가 포함돼요.

current_source = ""
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing advances"}]},
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        # Check if this event came from a subagent (namespace contains "tools:")
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
        if is_subagent:
            # Token from a subagent
            subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
            if subagent_ns != current_source:
                print(f"\n\n--- [subagent: {subagent_ns}] ---")
                current_source = subagent_ns
            if token.content:
                print(token.content, end="", flush=True)
        else:
            # Token from the main agent
            if "main" != current_source:
                print("\n\n--- [main agent] ---")
                current_source = "main"
            if token.content:
                print(token.content, end="", flush=True)
print()

도구 호출 스트리밍 (Tool calls)

서브에이전트가 도구를 쓰면 도구 호출 이벤트를 스트리밍해서 각 서브에이전트가 무엇을 하고 있는지 표시할 수 있어요. 도구 호출 청크는 messages 스트림 모드에 나타나요.

from langchain.messages import AIMessageChunk, ToolMessage

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research recent quantum computing advances"}]},
    stream_mode="messages",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        # Identify source: "main" or the subagent namespace segment
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
        source = next((s for s in chunk["ns"] if s.startswith("tools:")), "main") if is_subagent else "main"
        # Tool call chunks (streaming tool invocations)
        if isinstance(token, AIMessageChunk) and token.tool_call_chunks:
            for tc in token.tool_call_chunks:
                if tc.get("name"):
                    print(f"\n[{source}] Tool call: {tc['name']}")
                # Args stream in chunks - write them incrementally
                if tc.get("args"):
                    print(tc["args"], end="", flush=True)
        # Tool results
        if isinstance(token, ToolMessage):
            print(f"\n[{source}] Tool result [{token.name}]: {str(token.content)[:150]}")
        # Regular AI content (skip tool call messages)
        if (
            isinstance(token, AIMessageChunk)
            and token.content
            and not token.tool_call_chunks
        ):
            print(token.content, end="", flush=True)
print()

커스텀 업데이트 (Custom updates)

서브에이전트 도구 내부에서 get_stream_writer를 사용해 사용자 정의 진행 신호를 emit할 수 있어요. stream_mode="custom"으로 스트리밍할 때 받게 돼요. 아래는 분석 도구가 진행 상황을 emit하는 예제예요.

import time
from langchain.tools import tool
from langgraph.config import get_stream_writer
from deepagents import create_deep_agent

@tool
def analyze_data(topic: str) -> str:
    """Run a data analysis on a given topic.
    This tool performs the actual analysis and emits progress updates.
    You MUST call this tool for any analysis request.
    """
    writer = get_stream_writer()
    writer({"status": "starting", "topic": topic, "progress": 0})
    time.sleep(0.5)
    writer({"status": "analyzing", "progress": 50})
    time.sleep(0.5)
    writer({"status": "complete", "progress": 100})
    return (
        f'Analysis of "{topic}": Customer sentiment is 85% positive, '
        "driven by product quality and support response times."
    )

agent = create_deep_agent(
    model="baseten:zai-org/GLM-5.2",
    system_prompt=(
        "You are a coordinator. For any analysis request, you MUST delegate "
        "to the analyst subagent using the task tool. Never try to answer directly. "
        "After receiving the result, summarize it in one sentence."
    ),
    subagents=[
        {
            "name": "analyst",
            "description": "Performs data analysis with real-time progress tracking",
            "system_prompt": (
                "You are a data analyst. You MUST call the analyze_data tool "
                "for every analysis request. Do not use any other tools. "
                "After the analysis completes, report the result."
            ),
            "tools": [analyze_data],
        },
    ],
)

custom_event_count = 0
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze customer satisfaction trends"}]},
    stream_mode="custom",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "custom":
        custom_event_count += 1
        is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
        if is_subagent:
            subagent_ns = next(s for s in chunk["ns"] if s.startswith("tools:"))
            print(f"[{subagent_ns}]", chunk["data"])
        else:
            print("[main]", chunk["data"])

출력:

[tools:call_abc123] {'status': 'starting', 'topic': 'customer satisfaction trends', 'progress': 0}
[tools:call_abc123] {'status': 'analyzing', 'progress': 50}
[tools:call_abc123] {'status': 'complete', 'progress': 100}

여러 모드 스트리밍 (Stream multiple modes)

여러 스트림 모드를 결합하면 에이전트 실행의 완전한 그림을 얻을 수 있어요.

# Skip internal middleware steps - only show meaningful node names
INTERESTING_NODES = {"model", "tools"}
last_source = ""
mid_line = False  # True when we've written tokens without a trailing newline

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Analyze the impact of remote work on team productivity"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    is_subagent = any(s.startswith("tools:") for s in chunk["ns"])
    source = "subagent" if is_subagent else "main"
    if chunk["type"] == "updates":
        for node_name in chunk["data"]:
            if node_name not in INTERESTING_NODES:
                continue
            if mid_line:
                print()
                mid_line = False
            print(f"[{source}] step: {node_name}")
    elif chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            # Print a header when the source changes
            if source != last_source:
                if mid_line:
                    print()
                    mid_line = False
                print(f"\n[{source}] ", end="")
                last_source = source
            print(token.content, end="", flush=True)
            mid_line = True
    elif chunk["type"] == "custom":
        if mid_line:
            print()
            mid_line = False
        print(f"[{source}] custom event:", chunk["data"])
print()

일반 패턴 (Common patterns)

서브에이전트 라이프사이클 추적 (Track subagent lifecycle)

서브에이전트가 언제 시작하고, 실행하고, 완료되는지 모니터링해요.

active_subagents = {}
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research the latest AI safety developments"}]},
    stream_mode="updates",
    subgraphs=True,
    version="v2",
):
    if chunk["type"] == "updates":
        for node_name, data in chunk["data"].items():
            # ─── Phase 1: Detect subagent starting ────────────────────────
            # When the main agent's model node contains task tool calls,
            # a subagent has been spawned.
            if not chunk["ns"] and node_name == "model":
                for msg in data.get("messages", []):
                    for tc in getattr(msg, "tool_calls", []):
                        if tc["name"] == "task":
                            active_subagents[tc["id"]] = {
                                "type": tc["args"].get("subagent_type"),
                                "description": tc["args"].get("description", "")[:80],
                                "status": "pending",
                            }
                            print(
                                f'[lifecycle] PENDING  → subagent "{tc["args"].get("subagent_type")}" '
                                f'({tc["id"]})'
                            )
            # ─── Phase 2: Detect subagent running ─────────────────────────
            # When we receive events from a tools:UUID namespace, that
            # subagent is actively executing.
            if chunk["ns"] and chunk["ns"][0].startswith("tools:"):
                pregel_id = chunk["ns"][0].split(":")[1]
                # Check if any pending subagent needs to be marked running.
                # Note: the pregel task ID differs from the tool_call_id,
                # so we mark any pending subagent as running on first subagent event.
                for sub_id, sub in active_subagents.items():
                    if sub["status"] == "pending":
                        sub["status"] = "running"
                        print(
                            f'[lifecycle] RUNNING  → subagent "{sub["type"]}" '
                            f"(pregel: {pregel_id})"
                        )
                        break
            # ─── Phase 3: Detect subagent completing ──────────────────────
            # When the main agent's tools node returns a tool message,
            # the subagent has completed and returned its result.
            if not chunk["ns"] and node_name == "tools":
                for msg in data.get("messages", []):
                    if msg.type == "tool":
                        sub = active_subagents.get(msg.tool_call_id)
                        if sub:
                            sub["status"] = "complete"
                            print(
                                f'[lifecycle] COMPLETE → subagent "{sub["type"]}" '
                                f"({msg.tool_call_id})"
                            )
                            print(f"  Result preview: {str(msg.content)[:120]}...")
# Print final state
print("\n--- Final subagent states ---")
for sub_id, sub in active_subagents.items():
    print(f"  {sub['type']}: {sub['status']}")

v2 스트리밍 포맷

LangGraph 1.1 이상이 필요해요. 이 페이지의 모든 예제는 권장 방식인 v2 스트리밍 포맷(version="v2")을 사용해요. 모든 청크는 type, ns, data 키를 가진 StreamPart dict이며, 스트림 모드나 모드 수, 서브그래프 설정과 관계없이 동일한 형태예요.

v2 포맷은 중첩 튜플 언패킹을 제거해서 Deep Agents에서 서브그래프 스트리밍을 처리하기 쉬워요. 두 포맷을 비교하면:

# v2 (권장) — 통합된 포맷, 중첩 튜플 언패킹 없음
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
    version="v2",
):
    print(chunk["type"])  # "updates", "messages", or "custom"
    print(chunk["ns"])    # () for main agent, ("tools:<id>",) for subagent
    print(chunk["data"])  # payload

# v1 (legacy) — (namespace, (mode, data)) 중첩 튜플을 처리해야 함
for namespace, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Research quantum computing"}]},
    stream_mode=["updates", "messages", "custom"],
    subgraphs=True,
):
    mode, data = chunk[0], chunk[1]
    print(mode)       # "updates", "messages", or "custom"
    print(namespace)  # () for main agent, ("tools:<id>",) for subagent
    print(data)       # payload

v2 포맷의 타입 내로잉(Type narrowing), Pydantic/dataclass 강제 변환 등 자세한 내용은 LangGraph 스트리밍 문서를 참고해요.

더 알아보기 (Learn more)