Streaming API

LangGraph SDK를 사용하면 LangGraph Server API에서 출력을 스트리밍할 수 있습니다.

LangGraph SDK와 LangGraph Server는 LangSmith의 일부입니다.

기본 사용법

기본 사용 예제:

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# create a streaming run
async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input=inputs,
    stream_mode="updates"
):
    print(chunk.data)

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";

// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// create a streaming run
const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input,
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

cURL

스레드를 생성합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

스트리밍 실행을 생성합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--header 'x-api-key: <API_KEY>'
--data "{
  \"assistant_id\": \"agent\",
  \"input\": <inputs>,
  \"stream_mode\": \"updates\"
}"

확장된 예제: 업데이트 스트리밍

다음은 LangGraph API 서버에서 실행할 수 있는 예제 그래프입니다. 자세한 내용은 LangSmith 빠른 시작을 참조하세요.

# graph.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    topic: str
    joke: str

def refine_topic(state: State):
    return {"topic": state["topic"] + " and cats"}

def generate_joke(state: State):
    return {"joke": f"This is a joke about {state['topic']}"}

graph = (
    StateGraph(State)
    .add_node(refine_topic)
    .add_node(generate_joke)
    .add_edge(START, "refine_topic")
    .add_edge("refine_topic", "generate_joke")
    .add_edge("generate_joke", END)
    .compile()
)

실행 중인 LangGraph API 서버가 있으면, LangGraph SDK를 사용하여 상호작용할 수 있습니다.

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

# create a streaming run
async for chunk in client.runs.stream(  # (1)!
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="updates"  # (2)!
):
    print(chunk.data)

1. client.runs.stream() 메서드는 스트리밍 출력을 생성하는 이터레이터를 반환합니다. 2. stream_mode="updates"를 설정하면 각 노드 실행 후 그래프 상태의 업데이트만 스트리밍합니다. 다른 스트림 모드도 사용할 수 있습니다. 자세한 내용은 지원되는 스트림 모드를 참조하세요.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";

// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// create a streaming run
const streamResponse = client.runs.stream(  // (1)!
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "updates"  // (2)!
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

1. client.runs.stream() 메서드는 스트리밍 출력을 생성하는 이터레이터를 반환합니다.
2. streamMode: "updates"를 설정하면 각 노드 실행 후 그래프 상태의 업데이트만 스트리밍합니다. 다른 스트림 모드도 사용할 수 있습니다. 자세한 내용은 지원되는 스트림 모드를 참조하세요.

cURL

스레드를 생성합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

스트리밍 실행을 생성합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"updates\"
}"

{'run_id': '1f02c2b3-3cef-68de-b720-eec2a4a8e920', 'attempt': 1}
{'refine_topic': {'topic': 'ice cream and cats'}}
{'generate_joke': {'joke': 'This is a joke about ice cream and cats'}}

지원되는 스트림 모드

모드	설명	LangGraph 라이브러리 메서드
values	각 슈퍼 스텝 후 전체 그래프 상태를 스트리밍합니다.	stream_mode="values"와 함께 .stream() / .astream()
updates	그래프의 각 단계 후 상태 업데이트를 스트리밍합니다. 동일한 단계에서 여러 업데이트가 발생하는 경우(예: 여러 노드 실행), 해당 업데이트는 개별적으로 스트리밍됩니다.	stream_mode="updates"와 함께 .stream() / .astream()
messages-tuple	LLM이 호출되는 그래프 노드에 대한 LLM 토큰 및 메타데이터를 스트리밍합니다(채팅 앱에 유용함).	stream_mode="messages"와 함께 .stream() / .astream()
debug	그래프 실행 전반에 걸쳐 가능한 한 많은 정보를 스트리밍합니다.	stream_mode="debug"와 함께 .stream() / .astream()
custom	그래프 내부에서 사용자 정의 데이터를 스트리밍합니다.	stream_mode="custom"와 함께 .stream() / .astream()
events	그래프 상태를 포함한 모든 이벤트를 스트리밍합니다. 주로 대규모 LCEL 앱을 마이그레이션할 때 유용합니다.	.astream_events()

여러 모드 스트리밍

stream_mode 매개변수에 리스트를 전달하여 여러 모드를 동시에 스트리밍할 수 있습니다.

스트리밍된 출력은 (mode, chunk) 형태의 튜플로 제공되며, 여기서 mode는 스트림 모드의 이름이고 chunk는 해당 모드에서 스트리밍된 데이터입니다.

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input=inputs,
    stream_mode=["updates", "custom"]
):
    print(chunk)

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input,
    streamMode: ["updates", "custom"]
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk);
}

cURL

curl --request POST \
 --url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
 --header 'Content-Type: application/json' \
 --data "{
   \"assistant_id\": \"agent\",
   \"input\": <inputs>,
   \"stream_mode\": [
     \"updates\"
     \"custom\"
   ]
 }"

그래프 상태 스트리밍

스트림 모드 updates와 values를 사용하여 그래프가 실행될 때 상태를 스트리밍합니다.

• updates는 그래프의 각 단계 후 상태의 업데이트를 스트리밍합니다.
• values는 그래프의 각 단계 후 상태의 전체 값을 스트리밍합니다.

예제 그래프

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
  topic: str
  joke: str

def refine_topic(state: State):
    return {"topic": state["topic"] + " and cats"}

def generate_joke(state: State):
    return {"joke": f"This is a joke about {state['topic']}"}

graph = (
  StateGraph(State)
  .add_node(refine_topic)
  .add_node(generate_joke)
  .add_edge(START, "refine_topic")
  .add_edge("refine_topic", "generate_joke")
  .add_edge("generate_joke", END)
  .compile()
)

상태 유지 실행 아래 예제는 스트리밍 실행의 출력을 유지하려는 경우를 가정하며, 체크포인터 DB에 저장하고 스레드를 생성합니다. 스레드를 생성하려면:

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"
# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";
// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"]

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

실행 출력을 유지할 필요가 없는 경우, 스트리밍할 때 thread_id 대신 None을 전달할 수 있습니다.

스트림 모드: updates

각 단계 후 노드가 반환한 상태 업데이트만 스트리밍하려면 이 모드를 사용합니다. 스트리밍된 출력에는 노드 이름과 업데이트가 포함됩니다.

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="updates"
):
    print(chunk.data)

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"updates\"
}"

스트림 모드: values

각 단계 후 그래프의 전체 상태를 스트리밍하려면 이 모드를 사용합니다.

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="values"
):
    print(chunk.data)

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "values"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"values\"
}"

서브그래프

스트리밍 출력에 서브그래프의 출력을 포함하려면, 상위 그래프의 .stream() 메서드에서 subgraphs=True를 설정할 수 있습니다. 이렇게 하면 상위 그래프와 모든 서브그래프의 출력을 스트리밍합니다.

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"foo": "foo"},
    stream_subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)

1. 서브그래프의 출력을 스트리밍하려면 stream_subgraphs=True를 설정합니다.

확장된 예제: 서브그래프에서 스트리밍

다음은 LangGraph API 서버에서 실행할 수 있는 예제 그래프입니다. 자세한 내용은 LangSmith 빠른 시작을 참조하세요.

# graph.py
from langgraph.graph import START, StateGraph
from typing import TypedDict

# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}

subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()

# Define parent graph
class ParentState(TypedDict):
    foo: str

def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

실행 중인 LangGraph API 서버가 있으면, LangGraph SDK를 사용하여 상호작용할 수 있습니다.

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>)

# Using the graph deployed with the name "agent"
assistant_id = "agent"

# create a thread
thread = await client.threads.create()
thread_id = thread["thread_id"]

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"foo": "foo"},
    stream_subgraphs=True, # (1)!
    stream_mode="updates",
):
    print(chunk)

1. 서브그래프의 출력을 스트리밍하려면 stream_subgraphs=True를 설정합니다.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL> });

// Using the graph deployed with the name "agent"
const assistantID = "agent";

// create a thread
const thread = await client.threads.create();
const threadID = thread["thread_id"];

// create a streaming run
const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { foo: "foo" },
    streamSubgraphs: true,  // (1)!
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk);
}

1. 서브그래프의 출력을 스트리밍하려면 streamSubgraphs: true를 설정합니다.

cURL

스레드를 생성합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads \
--header 'Content-Type: application/json' \
--data '{}'

스트리밍 실행을 생성합니다:

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"foo\": \"foo\"},
  \"stream_subgraphs\": true,
  \"stream_mode\": [
    \"updates\"
  ]
}"

노드 업데이트뿐만 아니라 어떤 그래프(또는 서브그래프)에서 스트리밍되는지를 알려주는 네임스페이스도 수신한다는 점에 유의하세요.

디버깅

debug 스트리밍 모드를 사용하여 그래프 실행 전반에 걸쳐 가능한 한 많은 정보를 스트리밍합니다. 스트리밍된 출력에는 노드 이름과 전체 상태가 포함됩니다.

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="debug"
):
    print(chunk.data)

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "debug"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"debug\"
}"

LLM 토큰

messages-tuple 스트리밍 모드를 사용하여 노드, 도구, 서브그래프 또는 태스크를 포함한 그래프의 모든 부분에서 대규모 언어 모델(LLM) 출력을 토큰 단위로 스트리밍합니다.

messages-tuple 모드에서 스트리밍된 출력은 (message_chunk, metadata) 튜플입니다:

• message_chunk: LLM의 토큰 또는 메시지 세그먼트입니다.
• metadata: 그래프 노드 및 LLM 호출에 대한 세부 정보를 포함하는 딕셔너리입니다.

예제 그래프

from dataclasses import dataclass

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START

@dataclass
class MyState:
    topic: str
    joke: str = ""

model = init_chat_model(model="openai:gpt-4o-mini")

def call_model(state: MyState):
    """Call the LLM to generate a joke about a topic"""
    model_response = model.invoke( # (1)!
        [
            {"role": "user", "content": f"Generate a joke about {state.topic}"}
        ]
    )
    return {"joke": model_response.content}

graph = (
    StateGraph(MyState)
    .add_node(call_model)
    .add_edge(START, "call_model")
    .compile()
)

1. LLM이 stream이 아닌 invoke를 사용하여 실행되는 경우에도 메시지 이벤트가 발생한다는 점에 유의하세요.

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="messages-tuple",
):
    if chunk.event != "messages":
        continue

    message_chunk, metadata = chunk.data  # (1)!
    if message_chunk["content"]:
        print(message_chunk["content"], end="|", flush=True)

1. “messages-tuple” 스트림 모드는 (message_chunk, metadata) 튜플의 이터레이터를 반환합니다. 여기서 message_chunk는 LLM이 스트리밍한 토큰이고 metadata는 LLM이 호출된 그래프 노드와 기타 정보에 대한 정보가 포함된 딕셔너리입니다.

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "messages-tuple"
  }
);
for await (const chunk of streamResponse) {
  if (chunk.event !== "messages") {
    continue;
  }
  console.log(chunk.data[0]["content"]);  // (1)!
}

1. “messages-tuple” 스트림 모드는 (message_chunk, metadata) 튜플의 이터레이터를 반환합니다. 여기서 message_chunk는 LLM이 스트리밍한 토큰이고 metadata는 LLM이 호출된 그래프 노드와 기타 정보에 대한 정보가 포함된 딕셔너리입니다.

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"messages-tuple\"
}"

LLM 토큰 필터링

• LLM 호출별로 스트리밍된 토큰을 필터링하려면 LLM 호출과 tags를 연결할 수 있습니다.
• 특정 노드에서만 토큰을 스트리밍하려면 stream_mode="messages"를 사용하고 스트리밍된 메타데이터의 langgraph_node 필드로 출력을 필터링합니다.

사용자 정의 데이터 스트리밍

사용자 정의 데이터를 전송하려면:

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"query": "example"},
    stream_mode="custom"
):
    print(chunk.data)

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { query: "example" },
    streamMode: "custom"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"query\": \"example\"},
  \"stream_mode\": \"custom\"
}"

이벤트 스트리밍

그래프 상태를 포함한 모든 이벤트를 스트리밍하려면:

Python

async for chunk in client.runs.stream(
    thread_id,
    assistant_id,
    input={"topic": "ice cream"},
    stream_mode="events"
):
    print(chunk.data)

JavaScript

const streamResponse = client.runs.stream(
  threadID,
  assistantID,
  {
    input: { topic: "ice cream" },
    streamMode: "events"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/stream \
--header 'Content-Type: application/json' \
--data "{
  \"assistant_id\": \"agent\",
  \"input\": {\"topic\": \"ice cream\"},
  \"stream_mode\": \"events\"
}"

상태 비저장 실행

스트리밍 실행의 출력을 유지하지 않고 체크포인터 DB에 저장하지 않으려면 스레드를 생성하지 않고 상태 비저장 실행을 만들 수 있습니다:

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

async for chunk in client.runs.stream(
    None,  # (1)!
    assistant_id,
    input=inputs,
    stream_mode="updates"
):
    print(chunk.data)

1. thread_id UUID 대신 None을 전달하고 있습니다.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

// create a streaming run
const streamResponse = client.runs.stream(
  null,  // (1)!
  assistantID,
  {
    input,
    streamMode: "updates"
  }
);
for await (const chunk of streamResponse) {
  console.log(chunk.data);
}

1. thread_id UUID 대신 None을 전달하고 있습니다.

cURL

curl --request POST \
--url <DEPLOYMENT_URL>/runs/stream \
--header 'Content-Type: application/json' \
--header 'x-api-key: <API_KEY>'
--data "{
  \"assistant_id\": \"agent\",
  \"input\": <inputs>,
  \"stream_mode\": \"updates\"
}"

참여 및 스트리밍

LangSmith를 사용하면 활성 백그라운드 실행에 참여하고 출력을 스트리밍할 수 있습니다. 이를 위해 LangGraph SDK의 client.runs.join_stream 메서드를 사용할 수 있습니다:

Python

from langgraph_sdk import get_client
client = get_client(url=<DEPLOYMENT_URL>, api_key=<API_KEY>)

async for chunk in client.runs.join_stream(
    thread_id,
    run_id,  # (1)!
):
    print(chunk)

1. 이것은 참여하려는 기존 실행의 run_id입니다.

JavaScript

import { Client } from "@langchain/langgraph-sdk";
const client = new Client({ apiUrl: <DEPLOYMENT_URL>, apiKey: <API_KEY> });

const streamResponse = client.runs.joinStream(
  threadID,
  runId  // (1)!
);
for await (const chunk of streamResponse) {
  console.log(chunk);
}

1. 이것은 참여하려는 기존 실행의 run_id입니다.

cURL

curl --request GET \
--url <DEPLOYMENT_URL>/threads/<THREAD_ID>/runs/<RUN_ID>/stream \
--header 'Content-Type: application/json' \
--header 'x-api-key: <API_KEY>'

출력이 버퍼링되지 않음 .join_stream을 사용하면 출력이 버퍼링되지 않으므로, 참여하기 전에 생성된 출력은 수신되지 않습니다.

API 레퍼런스

API 사용법 및 구현에 대해서는 API 레퍼런스를 참조하세요.

---

Edit the source of this page on GitHub.

Connect these docs programmatically to Claude, VSCode, and more via MCP for real-time answers.