LangGraph 단계별 스트리밍 출력

StateGraph의 stream()과 출력 모드 정리

LangGraph의 stream() 메서드 동작과 주요 옵션(stream_mode, output_keys, interrupt_before/after, config)을 단계별 예제와 함께 설명한다. values, updates, debug 등 스트리밍 모드별 출력 형태와 output_keys 활용법, 체크포인터와 결합한 멀티턴 스트리밍 사용 사례 및 실무 팁을 포함한다.

Agent
LangGraph
Streaming
AI
Python
저자

Kwangmin Kim

공개

2025년 07월 15일

1 LangGraph 단계별 스트리밍 출력

  • LangGraph 의 stream() 은 LangChain의 Token Streaming과 다른 개념이다.
  • LangGraph 의 스트리밍 출력 함수는 그래프의 각 노드 단계 단위로 스트리밍하는 기능을 제공한다.
  • 특정 노드에서 token 단위 스트리밍이 필요한 경우, 해당 노드에서 LLM의 token streaming 기능을 활용하여 구현할 수 있다.

참고: 아래의 LangGraph 예제는 이전 섹션의 예제와 동일하다. (뉴스 검색 추가)

# API 키를 환경변수로 관리하기 위한 설정 파일
from dotenv import load_dotenv

# API 키 정보 로드
load_dotenv()
# LangSmith 추적을 설정합니다. https://smith.langchain.com
# !pip install -qU langchain-teddynote
from langchain_teddynote import logging

# 프로젝트 이름을 입력합니다.
logging.langsmith("CH17-LangGraph-Modules")
from typing import Annotated, List, Dict
from typing_extensions import TypedDict

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_teddynote.graphs import visualize_graph
from langchain_teddynote.tools import GoogleNews


########## 1. 상태 정의 ##########
# 상태 정의
class State(TypedDict):
    # 메시지 목록 주석 추가
    messages: Annotated[list, add_messages]
    dummy_data: Annotated[str, "dummy"]


########## 2. 도구 정의 및 바인딩 ##########
# 도구 초기화
# 키워드로 뉴스 검색하는 도구 생성
news_tool = GoogleNews()


@tool
def search_keyword(query: str) -> List[Dict[str, str]]:
    """Look up news by keyword"""
    news_tool = GoogleNews()
    return news_tool.search_by_keyword(query, k=5)


tools = [search_keyword]

# LLM 초기화
llm = ChatOpenAI(model="gpt-4o-mini")

# 도구와 LLM 결합
llm_with_tools = llm.bind_tools(tools)


########## 3. 노드 추가 ##########
# 챗봇 함수 정의
def chatbot(state: State):
    # 메시지 호출 및 반환
    return {
        "messages": [llm_with_tools.invoke(state["messages"])],
        "dummy_data": "[chatbot] 호출, dummy data",  # 테스트를 위하여 더미 데이터를 추가합니다.
    }


# 상태 그래프 생성
graph_builder = StateGraph(State)

# 챗봇 노드 추가
graph_builder.add_node("chatbot", chatbot)


# 도구 노드 생성 및 추가
tool_node = ToolNode(tools=tools)

# 도구 노드 추가
graph_builder.add_node("tools", tool_node)

# 조건부 엣지
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

########## 4. 엣지 추가 ##########

# tools > chatbot
graph_builder.add_edge("tools", "chatbot")

# START > chatbot
graph_builder.add_edge(START, "chatbot")

# chatbot > END
graph_builder.add_edge("chatbot", END)

########## 5. 그래프 컴파일 ##########

# 그래프 빌더 컴파일
graph = graph_builder.compile()

########## 6. 그래프 시각화 ##########
# 그래프 시각화
visualize_graph(graph)

1.1 StateGraph의 stream 메서드

stream 메서드는 단일 입력에 대한 그래프 단계를 스트리밍하는 기능을 제공한다.

매개변수 - input (Union[dict[str, Any], Any]): 그래프에 대한 입력 - config (Optional[RunnableConfig]): 실행 구성 - stream_mode (Optional[Union[StreamMode, list[StreamMode]]]): 출력 스트리밍 모드 - output_keys (Optional[Union[str, Sequence[str]]]): 스트리밍할 키 - interrupt_before (Optional[Union[All, Sequence[str]]]): 실행 전에 중단할 노드 - interrupt_after (Optional[Union[All, Sequence[str]]]): 실행 후에 중단할 노드 - debug (Optional[bool]): 디버그 정보 출력 여부 - subgraphs (bool): 하위 그래프 스트리밍 여부

반환값 - Iterator[Union[dict[str, Any], Any]]: 그래프의 각 단계 출력. 출력 형태는 stream_mode에 따라 다름

주요 기능 1. 입력된 설정에 따라 그래프 실행을 스트리밍 방식으로 처리 2. 다양한 스트리밍 모드 지원 (values, updates, debug) 3. 콜백 관리 및 오류 처리 4. 재귀 제한 및 중단 조건 처리

스트리밍 모드 - values: 각 단계의 현재 상태 값 출력 - updates: 각 단계의 상태 업데이트만 출력 - debug: 각 단계의 디버그 이벤트 출력

from langchain_core.runnables import RunnableConfig

# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 상태를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-tag"],  # Tag
)

config 를 설정하고 스트리밍 출력을 진행한다.

# 스트리밍의 기본 모드로 출력: 기본 모드의 경우 딕셔너리 형식이지만 모드에 따라 리스트같이 출력 형태가 달라질 수 있다.
for event in graph.stream(input=input, config=config):
    for key, value in event.items():
        print(f"\n[ {key} ]\n") # key 는 노드 이름, value 는 노드의 출력값 (state값)
        # value 에 messages 가 존재하는 경우
        if "messages" in value:
            messages = value["messages"]
            # 가장 최근 메시지 1개만 출력합니다.
            value["messages"][-1].pretty_print()
[ chatbot ]

================================== Ai Message ==================================
Tool Calls:
  search_keyword (call_MvThd5IASHA7pd6FL3I3zXse)
 Call ID: call_MvThd5IASHA7pd6FL3I3zXse
  Args:
    query: 2024 Nobel Prize in Literature

[ tools ]

================================= Tool Message =================================
Name: search_keyword

[{"url": "https://news.google.com/rss/articles/CBMiU0FVX3lxTE9sNE41R21QRkJzc25W", "content": "Translators share fond memories of working with Nobel winner Han Kang - 네이트 뉴스"}, {"url": "https://news.google.com/rss/articles/CBMidEFVX3lxTE5jd0IwTnpaQ1NkRjE1d3d2TjBkVk9JN", "content": "S. Korean author Han Kang wins Nobel Prize for literature - K-VIBE"}, {"url": "https://news.google.com/rss/articles/CBMiWkFVX3lxTFBqRFRqTld0NkozVEcxamNzcXkwSnJKc1FYYXdKS1l1N2dBQXhNZHhOSzlfcG5CS0M0QTlEMXd", "content": "[속보] 노벨위 \"한강, 역사적 트라우마에 맞선 강렬한 시적 산문\" - 파이낸셜뉴스"}, {"url": "https://news.google.com/rss/articles/CBMiU0FVX3lxTE1PQ1ZLSXNWTmpjamg3", "content": "Han Kang breaks silence on Nobel Prize: \"I need time to think about what this prize means\" - 네이트 뉴스"}, {"url": "https://news.google.com/rss/articles/CBMiRkFVX3lxTE00aXB2QjdRd3BOaGF", "content": "한글 탄생 578년을 품은 쾌거 한강 작가의 노벨문학상 수상 (1) - 브레이크뉴스"}]

[ chatbot ]

================================== Ai Message ==================================

2024년 노벨 문학상에 관한 최근 뉴스는 다음과 같습니다:

1. **한강 작가와의 회상**: 번역가들이 노벨 수상자 한강과 함께 작업했던 소중한 기억을 공유하는 내용입니다. [자세한 내용 보기](https://news.google.com/rss/articles/CBMiU0FVX3lxTE)

2. **한강, 노벨 문학상 수상**: 한국 작가 한강이 문학 분야에서 노벨상을 수상했다는 소식입니다. [자세한 내용 보기](https://news.google.com/rss/articles/CBMidEFVX3lxTE5jd0IwTnp)

3. **한강의 시적 산문**: 노벨 위원회는 한강의 작품을 "역사적 트라우마에 맞선 강렬한 시적 산문"이라고 평가했습니다. [자세한 내용 보기](https://news.google.com/rss/articles/CBMiWkFVX3lxT)

4. **노벨상 수상에 대한 한강의 반응**: 한강이 노벨상 수상에 대한 생각이 필요하다고 언급하며 침묵을 깨뜨린 내용입니다. [자세한 내용 보기](https://news.google.com/rss/articles/CBMiU0FVX3lxTE1P)

5. **한강의 한글 탄생 578년과의 연결**: 한강 작가의 노벨 문학상 수상이 한국어의 역사와 연결된 성과임을 강조하는 내용입니다. [자세한 내용 보기](https://news.google.com/rss/articles/CBMiRkFVX3lxTE00a)

이와 같은 뉴스들이 현재 2024년 노벨 문학상과 관련하여 보도되고 있습니다.

1.1.1 output_keys 옵션

  • output_keys 옵션은 스트리밍할 키를 지정하는 데 사용된다.
  • list 형식으로 지정할 수 있으며, channels 에 정의된 키 중 하나 여야 한다.

팁(tip)

  • 매 단계마다 출력되는 State key 가 많은 경우, 일부만 스트리밍하고 싶은 경우에 유용하다.
# channels 에 정의된 키 목록을 출력합니다.
print(list(graph.channels.keys()))
['messages', 'dummy_data', '__start__', 'chatbot', 'tools', 'start:chatbot', 'branch:chatbot:tools_condition:chatbot', 'branch:chatbot:tools_condition:tools']
# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    output_keys=["dummy_data"],  #["dummy_data","messages"] messages 추가
):
    for key, value in event.items():
        # key 는 노드 이름
        print(f"\n[ {key} ]\n")

        # dummy_data 가 존재하는 경우
        if value:
            # value 는 노드의 출력값
            print(value.keys())
            # dummy_data key 가 존재하는 경우
            if "dummy_data" in value:
                print(value["dummy_data"])
            # messages 가 존재하는 경우
            #if "messages" in value:
            #    print(value["messages"])
[ chatbot ]

dict_keys(['dummy_data'])
[chatbot] 호출, dummy data

[ tools ]


[ chatbot ]

dict_keys(['dummy_data'])
[chatbot] 호출, dummy data
# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    output_keys=["messages"],  # messages 만 출력
):
    for key, value in event.items():
        # messages 가 존재하는 경우
        if value and "messages" in value:
            # key 는 노드 이름
            print(f"\n[ {key} ]\n")
            # messages 의 마지막 요소의 content 를 출력합니다.
            print(value["messages"][-1].content)
[ chatbot ]



[ tools ]

[{"url": "https://news.google.com/rss/articles/CBMiU0FVX3lx", "content": "Translators share fond memories of working with Nobel winner Han Kang - 네이트 뉴스"}, {"url": "https://news.google.com/rss/articles/CBMidEFVX", "content": "S. Korean author Han Kang wins Nobel Prize for literature - K-VIBE"}, {"url": "https://news.google.com/rss/articles/CBMiWkFVX3lx", "content": "[속보] 노벨위 \"한강, 역사적 트라우마에 맞선 강렬한 시적 산문\" - 파이낸셜뉴스"}, {"url": "https://news.google.com/rss/articles/CBMiU0FVX3lx", "content": "Han Kang breaks silence on Nobel Prize: \"I need time to think about what this prize means\" - 네이트 뉴스"}, {"url": "https://news.google.com/rss/articles/CBMiRkFVX3lxTE00aX", "content": "한글 탄생 578년을 품은 쾌거 한강 작가의 노벨문학상 수상 (1) - 브레이크뉴스"}]

[ chatbot ]

2024년 노벨 문학상과 관련된 최근 뉴스는 다음과 같습니다:

1. [Translators share fond memories of working with Nobel winner Han Kang - 네이트 뉴스](https://news.google.com/rss/articles/CBMiU0FVX3lxT)

2. [S. Korean author Han Kang wins Nobel Prize for literature - K-VIBE](https://news.google.com/rss/articles/CBMidEFVX3lx)

3. [노벨위 "한강, 역사적 트라우마에 맞선 강렬한 시적 산문" - 파이낸셜뉴스](https://news.google.com/rss/articles/CBMiWkFVX3)

4. [Han Kang breaks silence on Nobel Prize: "I need time to think about what this prize means" - 네이트 뉴스](https://news.google.com/rss/articles/CBMiU0FVX3lxT)

5. [한글 탄생 578년을 품은 쾌거 한강 작가의 노벨문학상 수상 (1) - 브레이크뉴스](https://news.google.com/rss/articles/CBMiRkFVX3l)

이 뉴스들은 한강 작가가 노벨 문학상을 수상한 사실과 관련된 여러 측면을 다루고 있습니다.

1.1.2 stream_mode 옵션

stream_mode 옵션은 스트리밍 출력 모드를 지정하는 데 사용된다.
- values: 각 단계의 현재 상태 값 출력
- updates: 각 단계의 상태 업데이트만 출력 (기본값)

1.1.2.1 stream_mode = "default"

  • key: 노드명
  • value: 노드의 상태

1.1.2.2 stream_mode = "values"

  • stream_mode="values"일 때: event는 { state_key: current_value } 형태다.
    • 예: key="messages", value=[Message(...), ...] → 현재 상태의 전체 값(현재 메시지 목록)을 의미.
  • 반면 stream_mode="updates"일 때: event는 { node_name: { updated_key: updated_value, ... } } 형태다.
    • 예: key="chatbot", value={"messages": [...], "dummy_data": "..."} → 해당 노드가 업데이트한 부분만 들어있음.
  • values 모드는 각 단계의 현재 상태 값을 출력합니다.

참고

  • values 모드인 경우:
    • event.items()는 (state_key, current_value) 쌍을 반환한다. 예: ('messages', [Message,...]) — 여기서 value는 해당 state 키의 전체 현재 값이다.
  • updates 모드인 경우(참고로):
    • event.items()는 (node_name, {updated_key: updated_value, ...}) 쌍을 반환한다. 예: ('chatbot', {'messages': [...]} ) — 여기서 value는 해당 노드에서 새로 업데이트한 값들이다.
# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

# values 모드로 스트리밍 출력
for event in graph.stream(
    input=input,
    stream_mode="values",  # 기본값
):
    for key, value in event.items():
        # key 는 state 의 key 값
        print(f"\n[ {key} ]\n")
        if key == "messages":
            print(f"메시지 개수: {len(value)}")
            # print(value)
    print("===" * 10, " 단계 ", "===" * 10)
[ messages ]

메시지 개수: 1

[ dummy_data ]

==============================  단계  ==============================

[ messages ]

메시지 개수: 2

[ dummy_data ]

==============================  단계  ==============================

[ messages ]

메시지 개수: 3

[ dummy_data ]

==============================  단계  ==============================

[ messages ]

메시지 개수: 4

[ dummy_data ]

==============================  단계  ==============================

1.1.2.3 stream_mode = "updates"

  • updates 모드는 각 단계에 대한 업데이트된 State 만 내보낸다.
    • 출력은 노드 이름을 key 로, 업데이트된 값을 values 으로 하는 dictionary 이다
  • event.items()
    • key: 노드(Node) 의 이름
    • value: 해당 노드(Node) 단계에서의 출력 값(dictionary). 즉, 여러 개의 key-value 쌍을 가진 dictionary 이다.
# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

# updates 모드로 스트리밍 출력
for event in graph.stream(
    input=input,
    stream_mode="updates",  # 기본값
):
    for key, value in event.items():
        # key 는 노드 이름
        print(f"\n[ {key} ]\n")

        # value 는 노드의 출력값
        print(value.keys())

        # value 에는 state 가 dict 형태로 저장(values 의 key 값)
        if "messages" in value:
            print(f"메시지 개수: {len(value['messages'])}")
            # print(value["messages"])
    print("===" * 10, " 단계 ", "===" * 10)
[ chatbot ]

dict_keys(['messages', 'dummy_data'])
메시지 개수: 1
==============================  단계  ==============================

[ tools ]

dict_keys(['messages'])
메시지 개수: 1
==============================  단계  ==============================

[ chatbot ]

dict_keys(['messages', 'dummy_data'])
메시지 개수: 1
==============================  단계  ==============================

1.1.3 interrupt_before 와 interrupt_after 옵션 (강제 중단 옵션)

  • interrupt_before 와 interrupt_after 옵션은 스트리밍 중단 시점을 지정하는 데 사용된다.
    • interrupt_before: 지정된 노드 이전에 스트리밍 중단
    • interrupt_after: 지정된 노드 이후에 스트리밍 중단
  • 언제 필요한가:
    • 실시간 UI 중단 지점: 대화형 UI에서 중간 결과(도구 호출 결과 등)를 보여주고 사용자의 승인/입력 후 다음 단계 진행할 때.
    • 비용/시간 절감: 고비용/장시간 노드(외부 API 호출, 대규모 RAG 검색 등)를 불필요하게 실행하지 않도록 중단 또는 예약 실행할 때.
    • 부분 출력만 필요할 때: 초기 요약/목록 등 일부 결과만 빠르게 노출하고, 전체 생성은 나중에 수행할 때.
    • 안전/검토 루프: LLM이 민감한 동작(계좌이체, 수정 등)을 수행하기 전에 검증/승인 단계로 중단.
    • 타임박스/타임아웃: 응답 시간이 길어질 경우 특정 노드 이전/이후에서 강제 중단하여 타임아웃 동작을 구현할 때.
    • 병렬·분기 제어: 브랜치가 여러 개인 그래프에서 특정 분기 실행 전후로 중단해 외부 로직이나 모니터링을 삽입할 때.
    • 체크포인트/재개 트리거: interrupt_after로 체크포인트를 찍고, 필요 시 같은 thread_id로 재개(또는 롤백)할 때.
  • UI 통합 예 (SSE/웹소켓 흐름 요약)
    • 서버: graph.stream() 를 순회하면서 각 event를 JSON으로 클라이언트에 푸시.
    • 클라이언트: 특정 event(예: node == “tools” 또는 __interrupt__)를 수신하면 사용자에게 “계속/중지” 버튼을 노출.
    • 사용자가 ‘계속’ 선택 시 서버에 재요청하거나 동일 thread_id로 재개 처리.
  • FastAPI + Server-Sent Events 예시(개념)
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

def event_generator():
    for event in graph.stream(input=input, stream_mode="updates", interrupt_after=["tools"]):
        yield format_as_sse(event)
        # 클라이언트 응답(예: via websocket/HTTP)으로 재개 제어 가능

app = FastAPI()
@app.get("/stream")
def stream_endpoint():
    return StreamingResponse(event_generator(), media_type="text/event-stream")

구체적 응용 시나리오 (패턴)
- 사용자 승인 패턴: LLM이 행동(예: 주문/변경)을 하기 전 interrupt_before로 멈춰 요약을 보여주고 사용자가 승인하면 재개.
- 라이트-먼저(Preview then Full) 패턴: 먼저 interrupt_after로 도구 결과(검색 목록)를 받고 클라이언트가 원하는 항목을 선택하면, 선택을 인풋에 넣어 이후 노드(예: 상세 생성)를 실행.
- 비용 제어 패턴: heavy 노드(유료 API) 이전에 interrupt_before로 멈춰 사용자의 허가를 받음.
- 시간박스 패턴: 일정 시간 내 응답을 못 받으면 interrupt_after로 강제 중단해 대체 메시지 제공.
- 체크포인트 + 재시도: interrupt_after로 체크포인트를 찍고 실패 시 체크포인트에서 재시도 또는 다른 전략으로 분기.

핵심 구현·운영 팁 (주의사항)
- 이벤트 형태 파악: stream_mode="values"와 stream_mode="updates"의 이벤트 구조가 다르니 클라이언트에서 해석 로직을 분기 처리해야 함.
- 노드 이름 정확성: interrupt_*에 명시한 이름은 그래프의 노드(또는 브랜치 이름)과 정확히 매칭되어야 함. 브랜치/중복노드가 있으면 의도치 않은 지점에서 멈출 수 있음.
- 중단 이벤트 처리: 라이브러리에서 __interrupt__ 같은 특수 키를 반환할 수 있으므로 클라이언트는 이를 인지하고 UI 상태를 갱신해야 함.
- 부작용(idempotency): 중단/재개 시 외부 API가 중복 호출될 수 있으니 도구 호출은 멱등하게 설계하거나 호출 전 체크포인트/ 토큰 관리 필요.
- 출력량 제어: 중단 직전까지 많은 데이터를 스트리밍하면 네트워크/메모리 부담이 커짐 → output_keys로 필요한 키만 전송.
- 동시성/동기화: 여러 클라이언트가 같은 thread_id를 공유하면 충돌 가능. thread_id 설계(사용자별, 세션별)를 명확히 하라.
- 디버깅: stream_mode="debug"를 사용해 내부 이벤트를 관찰하고 적절한 중단 지점을 결정하라.
- 브랜치/하위그래프 주의: 하위그래프(subgraphs=True)를 스트리밍할 땐 interrupt_*가 의도한 하위 노드에 적용되는지 확인.

# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    stream_mode="updates",  # 기본값
    interrupt_before=["tools"],  # tools 노드 이전에 스트리밍 중단
):
    for key, value in event.items():
        # key 는 노드 이름
        print(f"\n[{key}]\n")

        # value 는 노드의 출력값
        if isinstance(value, dict):
            print(value.keys())
            if "messages" in value:
                print(value["messages"])

        # value 에는 state 가 dict 형태로 저장(values 의 key 값)
        if "messages" in value:
            print(f"메시지 개수: {len(value['messages'])}")
    print("===" * 10, " 단계 ", "===" * 10)
[chatbot]

dict_keys(['messages', 'dummy_data'])
[AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_PS0YB5Vmb1st8EL78Sy0Nh1X', 'function': {'arguments': '{"query":"2024년 노벨 문학상"}', 'name': 'search_keyword'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 21, 'prompt_tokens': 57, 'total_tokens': 78, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_f59a81427f', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-a5fa47c5-0aac-4782-88e0-fbdc2c3d58cd-0', tool_calls=[{'name': 'search_keyword', 'args': {'query': '2024년 노벨 문학상'}, 'id': 'call_PS0YB5Vmb1st8EL78Sy0Nh1X', 'type': 'tool_call'}], usage_metadata={'input_tokens': 57, 'output_tokens': 21, 'total_tokens': 78, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]
메시지 개수: 1
==============================  단계  ==============================

[__interrupt__]

==============================  단계  ==============================
# 질문
question = "2024년 노벨 문학상 관련 뉴스를 알려주세요."

# 초기 입력 State 를 정의
input = State(dummy_data="테스트 문자열", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # 최대 10개의 노드까지 방문. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    stream_mode="updates",
    interrupt_after=["tools"],  # tools 실행 후 interrupt
):
    for value in event.values():
        # key 는 노드 이름
        print(f"\n[{key}]\n")

        if isinstance(value, dict):
            # value 는 노드의 출력값
            print(value.keys())
            if "messages" in value:
                print(value["messages"])

        # value 에는 state 가 dict 형태로 저장(values 의 key 값)
        if "messages" in value:
            print(f"메시지 개수: {len(value['messages'])}")
[__interrupt__]

dict_keys(['messages', 'dummy_data'])
[AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_yOeS75Xf6bYIfy4Edx3Im3eA', 'function': {'arguments': '{"query":"2024 노벨 문학상"}', 'name': 'search_keyword'}, 'type': 'function'}], 'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 57, 'total_tokens': 77, 'completion_tokens_details': {'audio_tokens': None, 'reasoning_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-mini-2024-07-18', 'system_fingerprint': 'fp_f59a81427f', 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-984a77e7-6301-468b-b973-4fa49a7ccdaa-0', tool_calls=[{'name': 'search_keyword', 'args': {'query': '2024 노벨 문학상'}, 'id': 'call_yOeS75Xf6bYIfy4Edx3Im3eA', 'type': 'tool_call'}], usage_metadata={'input_tokens': 57, 'output_tokens': 20, 'total_tokens': 77, 'input_token_details': {'cache_read': 0}, 'output_token_details': {'reasoning': 0}})]
메시지 개수: 1

[__interrupt__]

dict_keys(['messages'])
[ToolMessage(content='[{"url": "https://news.google.com/rss/articles/CBMifEFVX3lxTE", "content": "한강의 노벨문학상 수상에 대하여 | UMNews.org - 연합감리교뉴스"}, {"url": "https://news.google.com/rss/articles/CBMia0FVX3lxTFBl", "content": "한강의 노벨 문학상보다 더 충격을 준 노벨 화학상 - 프레시안"}, {"url": "https://news.google.com/rss/articles/CBMiaEFVX3", "content": "역대 노벨문학상 수상작가 작품 보러가 볼까? - 대구일보"}, {"url": "https://news.google.com/rss/articles/CBMihwFBVV9", "content": "과학계에서 노벨상이 나오려면 - 전북도민일보"}, {"url": "https://news.google.com/rss/articles/CBMiygJBVV95c", "content": "2024년 노벨 문학상 수상자 ‘한강’의 작품 세계와 문학적 특성 - Colorado Times"}]', name='search_keyword', id='3497e144-2a86-4e61-92e6-07ce51670a78', tool_call_id='call_yOeS75Xf6bYIfy4Edx3Im3eA')]
메시지 개수: 1

[__interrupt__]

Subscribe

Enjoy this blog? Get notified of new posts by email: