05. Node's step-by-step streaming output

LangGraph streaming output

This time LangGrpah stream() Proceed with a little more detailed explanation of the output function.

LangGraph's streaming output function provides the ability to stream each step of the graph.

Note: The LangGraph example below is the same as the example in the previous section.

# Configuration file for managing API keys as environment variables
from dotenv import load_dotenv

# Load API key information
load_dotenv()

 True 

# Set up LangSmith tracking. https://smith.langchain.com
# !pip install -qU langchain-teddynote
from langchain_teddynote import logging

# Enter a project name.
logging.langsmith("CH17-LangGraph-Modules")

 Start tracking LangSmith. 
[Project name] 
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. state definition ##########
# state definition
class State(TypedDict):
    # Add a message list comment
    messages: Annotated[list, add_messages]
    dummy_data: Annotated[str, "dummy"]


########## 2. Tool definition and binding ##########
# Initialize tool
# Create a tool to search news by keyword
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 Initialization
llm = ChatOpenAI(model="gpt-4o-mini")

# Combining tools and LLM
llm_with_tools = llm.bind_tools(tools)


########## 3. add note ##########
# Defining a chatbot function
def chatbot(state: State):
    # Calling and returning messages
    return {
        "messages": [llm_with_tools.invoke(state["messages"])],
        "dummy_data": "[chatbot] call, dummy data", # Add dummy data for testing.
    }


# Create a state graph
graph_builder = StateGraph(State)

# Add a chatbot node
graph_builder.add_node("chatbot", chatbot)


# Creating and adding tool nodes
tool_node = ToolNode(tools=tools)

# 
graph_builder.add_node("tools", tool_node)

# Conditional Edge
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,
)

########## 4. add edge ##########

# 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. Compile the graph ##########

# Compile the graph builder
graph = graph_builder.compile()

########## 6. Graph visualization ##########
# Graph visualization
visualize_graph(graph)

StateGraph stream method

stream The method provides the ability to stream graph steps for a single input.

parameter

• input (Union[dict[str, Any], Any]): input to the graph

• config (Optional[RunnableConfig]): Execution configuration

• stream_mode (Optional[Union[StreamMode, list[StreamMode]]]): Output streaming mode

• output_keys (Optional[Union[str, Sequence[str]]]): Key to stream

• interrupt_before (Optional[Union[All, Sequence[str]]]): Nodes to stop before running

• interrupt_after (Optional[Union[All, Sequence[str]]]): Nodes to stop after execution

• debug (Optional[bool]): Whether debug information is output

• subgraphs (bool): Whether subgraph streaming

return value

• Iterator[Union[dict[str, Any], Any]]: Output of each step of the graph. Output form stream_mode Depends on

Main features

1. Stream graph execution according to entered settings

2. Support for various streaming modes ( values , updates , debug )

3. Callback management and error handling

4. Treatment of recursive restrictions and suspension conditions

Streaming mode

• values : Output the current state value for each step

• updates : Output only status updates for each step

• debug : Output of debug events at each stage

from langchain_core.runnables import RunnableConfig

# question
question = "Please tell me the news related to the 2024 Nobel Prize in Literature."

# Define the initial input state
input = State(dummy_data="test string", messages=[("user", question)])

# config settings
config = RunnableConfig(
    recursion_limit=10,  # Visit up to 10 nodes. Anything more than that will result in a RecursionError
    configurable={"thread_id": "1"},  # Thread ID setting
    tags=["my-tag"],  # Tag
)

config Set up and proceed to streaming output.

for event in graph.stream(input=input, config=config):
    for key, value in event.items():
        print(f"\n[ {key} ]\n")
        # If messages exist in value
        if "messages" in value:
            messages = value["messages"]
            # Print only the most recent message.
            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 -Nate News" }, {"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": "[Breaking News] \" Han River, intense poetic prose against historical trauma\"- Financial News"},<  I needle time to think about what this prize means\" -Nate News" }, {"url": "https://news.google.com/rss/articles/CBMiRkFVX3lxTE00aXB2QjdRd3BOaGF", "content": "Hanger Han Kang's Nobel Prize for the 578-year-old birth (1) -Brake News" }] 

[ chatbot] 

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

Here is the latest news about the 2024 Nobel Prize for Literature: 

One. **Recall with Han Kang Writer**: This is what translators share the precious memories they worked with Nobel laureate Han Kang. [See details] (https://news.google.com/rss/articles/CBMiU0FVX3lxTE) 

2. ** Han Kang, Nobel Prize for Literature **: This is the news that Korean writer Han Kang won the Nobel Prize in literature. [See details] (https://news.google.com/rss/articles/CBMidEFVX3lxTE5jd0IwTnp) 

3. ** Poetic prose of the Han River**: The Nobel Committee rated Han Kang's work as "an intense poetic prose against a historical trauma." [See details] (https://news.google.com/rss/articles/CBMiWkFVX3lxT) 

4. ** Han Kang's response to the Nobel Prize **: Han Kang said he needed an idea to win the Nobel Prize and broke the silence. [See details] (https://news.google.com/rss/articles/CBMiU0FVX3lxTE1P) 

5. ** Han Kang's birth of Hangeul's connection to 578 years **: Emphasizing that Han Kang's Nobel Prize for Literature is a achievement linked to Korean history. [See details] (https://news.google.com/rss/articles/CBMiRkFVX3lxTE00a) 

News like this is currently being reported in connection with the 2024 Nobel Prize for Literature. 

output_keys option

output_keys The option is used to specify the key to stream.

can be specified in list format, One of the keys defined in the channels Should be.

Tip

• This is useful if you have a lot of State key outputs every step, and you only want to stream some.

# Prints a list of keys defined in 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
question = "Please tell me the news related to the 2024 Nobel Prize in Literature."

# Define the initial input State
input = State(dummy_data="test string", messages=[("user", question)])

# Define the initial input State
config = RunnableConfig(
    recursion_limit=10,  # Visit up to 10 nodes, more than that will result in RecursionError
    configurable={"thread_id": "1"},  # Thread ID setting
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    output_keys=["dummy_data"],  # Try adding messages!
):
    for key, value in event.items():
        # key is the node name
        print(f"\n[ {key} ]\n")

        # If dummy_data exists
        if value:
            # value is the output value of the node
            print(value.keys())
            # If dummy_data key exists
            if "dummy_data" in value:
                print(value["dummy_data"])

 [ chatbot] 

dict_keys (['dummy_data']) 
[chatbot] call, dummy data 

[tools] 


[ chatbot] 

dict_keys (['dummy_data']) 
[chatbot] call, dummy data 

# question
question = "Please tell me the news related to the 2024 Nobel Prize in Literature."

# Define the initial input State
input = State(dummy_data="test string", messages=[("user", question)])

# config 설정
config = RunnableConfig(
    recursion_limit=10,  # w. 그 이상은 RecursionError 발생
    configurable={"thread_id": "1"},  # 스레드 ID 설정
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,w
    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 -Nate News" }, {"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": "[Break] Nobel \" Han River, intense poetic prose against historical trauma \"- Financial News"},<  com/rss/articles/CBMiRkFVX3lxTE00aX", "content": "Hanger Han Kang's Nobel Prize for the Nobel Prize for the 578-year-old birth (1)-Brake News" }] 

[ chatbot] 

Here are some recent news related to the 2024 Nobel Prize for Literature: 

One. [Translators share fond memories of working with Nobel winner Han Kang -Nate News] (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. [Nobelwi "Intense Poetic Prose Against Han River, Historical Trauma"-Financial News] (https://news.google.com/rss/articles/CBMiWkFVX3) 

4. [Han Kang breaks silence on Nobel Prize: "I needle time to think about what this prize means" -Nate News] (https://news.google.com/rss/articles/CBMiU0FVX3lxT) 

5. [Hanger Han Kang's Nobel Prize in Literature for the 578th year of Hangeul's birth (1)-Brake News] (https://news.google.com/rss/articles/CBMiRkFVX3l) 

These news cover many aspects related to the fact that Han River writer won the Nobel Prize for Literature. 

stream_mode option

stream_mode The option is used to specify the streaming output mode.

• values : Output the current state value for each step

• updates : Output only status updates for each step (default)

stream_mode = "values"

values Mode outputs the current state value for each step.

Reference

event.items()

• key : State key value

• value : Value for State's key

# 질문
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] 

Number of messages: 1 

[dummy_data] 

============================== Step ============================== 

[messages] 

Number of messages: 2 

[dummy_data] 

============================== Step ============================== 

[messages] 

Number of messages: 3 

[dummy_data] 

============================== Step ============================== 

[messages] 

Number of messages: 4 

[dummy_data] 

============================== Step ============================== 

stream_mode = "updates"

updates Mode only exports updated State for each step.

• The output is the node name as key, and the updated value is values dictionary is.

Reference

event.items()

• key : Name of Node

• value : Output value (dictionary) at that node stage. That is, it is a dictionary with multiple key-value pairs.

# 질문
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']) 
Number of messages: 1 
============================== Step ============================== 

[tools] 

dict_keys (['messages']) 
Number of messages: 1 
============================== Step ============================== 

[ chatbot] 

dict_keys (['messages','dummy_data']) 
Number of messages: 1 
============================== Step ============================== 

interrupt_before Wow interrupt_after option

interrupt_before Wow interrupt_after Options are used to specify when streaming stops.

• interrupt_before : Stop streaming before the specified node

• interrupt_after : Stop streaming after the specified node

# question
question = "Please tell me the news related to the 2024 Nobel Prize in Literature."

# Define the initial input State
input = State(dummy_data="test string", messages=[("user", question)])

# config settings
config = RunnableConfig(
    recursion_limit=10,  # Visit up to 10 nodes. Anything more than that will result in a RecursionError
    configurable={"thread_id": "1"},  # Thread ID setting
    tags=["my-rag"],  # Tag
)

for event in graph.stream(
    input=input,
    config=config,
    stream_mode="updates",  # default
    interrupt_before=["tools"],  # Stop streaming before tools node
):
    for key, value in event.items():
        # key is the node name
        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':'TAG1gpt-4o-mini-2024-07-18','system_fingerprint':'fp_f59a81427f','finish_reason':'tool_calls','logprobs': None}, id='  lun-a5fa47c5-0aac-4782-88e0-fbdc2c3d58cd-0', tool_calls={','args': {'  {'cache_read': 0},'output_token_details': {'reasoning': 0}})]  {'cache_read': 0},'output_token_details': {'reasoning': 0}})] 
Number of messages: 1 
============================== Step ============================== 

[__interrupt__] 

============================== Step ============================== 

# 질문
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':'TAG1>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={','args': {'  {'cache_read': 0},'output_token_details': {'reasoning': 0}})]  {'cache_read': 0},'output_token_details': {'reasoning': 0}})] 
Number of messages: 1 

[__interrupt__] 

dict_keys (['messages']) 
[ToolMessage (content=' [{"url": "https://news.google.com/rss/articles/CBMifEFVX3lxTE", "content": "To see the Nobel Prize in Korea | UMNews.org -Un Methodist News"} "url": "{2024 Nobel Prize for Literature ‘Han Kang ’'s work world and literary characteristics - Colorado Times" }]', name='search_keyword', id='  3497e144-2a86-4e61-92e6-07ce51670a78', tool_call_id='call_yOeS75Xf6bYIfy4Edx3Im3eA')] 
Number of messages: 1 

[__interrupt__]