랭그래프 기초

현서·2025년 9월 9일

AI Agent

목록 보기

3/8

1. 그래프의 상태 업데이트

HumanMessage : 사용자(사람)의 메시지
AIMessage : AI(LLM)의 메시지
AnyMessage : HumanMessage, AIMessage를 포함하는 메시지

!pip install langgraph

from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict

class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int

from langchain_core.messages import AIMessage

def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("안녕하세요! 무엇을 도와드릴까요?")

    # return {"messages": new_message, "extra_field": 10}
    return {"messages": messages + [new_message], "extra_field": 10}

from langgraph.graph import StateGraph

graph_builder = StateGraph(State)
graph_builder.add_node("node", node)
# set_entry_point : 그래프의 시작 노드를 지정하는 엣지 (START -> "node")
graph_builder.set_entry_point("node")
graph = graph_builder.compile()

graph

from langchain_core.messages import HumanMessage

result = graph.invoke({"messages": [HumanMessage("안녕")]}) # invoke : 랭그래프의 워크플로우가 끝날 때까지 기다리라
result

{'messages': [HumanMessage(content='안녕', additional_kwargs={}, response_metadata={}),
  AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={})],
 'extra_field': 10}

result["messages"]

[HumanMessage(content='안녕', additional_kwargs={}, response_metadata={}),
 AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={})]

2. 대화메시지 상태 누적 업데이트

add_messages 는 기존 메시지에서 추가 메시지를 병합하는 데 사용하는 함수로, 새로 들어온 메시지를 추가할 때 사용할 수 있는 리듀서 역할

from typing_extensions import Annotated
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int

def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("안녕하세요! 무엇을 도와드릴까요?")

    return {"messages": new_message, "extra_field": 10}

from langgraph.graph import StateGraph

graph_builder = StateGraph(State)
graph_builder.add_node("node", node)
graph_builder.set_entry_point("node")
graph = graph_builder.compile()

graph

input_message = {"role": "user", "content": "안녕하세요."}

result = graph.invoke({"messages": [input_message]})
result

{'messages': [HumanMessage(content='안녕하세요.', additional_kwargs={}, response_metadata={}, id='16fd3c5d-0947-4134-b42c-43443e94de0e'),
  AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}, id='891b730b-4ae2-48a9-bbfa-5c5cf608eda6')],
 'extra_field': 10}

for message in result["messages"]:
# pretty_print()는 데이터나 객체를 보기 좋게(Pretty) 정리해서 출력하는 함수
    message.pretty_print()

================================ Human Message =================================

안녕하세요.
================================== Ai Message ==================================

안녕하세요! 무엇을 도와드릴까요?

result["messages"]

[HumanMessage(content='안녕하세요.', additional_kwargs={}, response_metadata={}, id='16fd3c5d-0947-4134-b42c-43443e94de0e'),
 AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}, id='891b730b-4ae2-48a9-bbfa-5c5cf608eda6')]

# invoke : 하나의 요청에 대한 결과를 받을 때 까지 코드 실행 멈춤. 한번에 하나의 요청을 처리함
graph.invoke({"messages": [input_message]})

{'messages': [HumanMessage(content='안녕하세요.', additional_kwargs={}, response_metadata={}, id='f7d7d047-740c-4b52-9976-114a8c708db1'),
  AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}, id='9a45f403-adae-47f0-af8e-0fed8d6afb32')],
 'extra_field': 10}

# ainvoke : 비동기 처리로 여러 요청을 동시에 보낼 수 있음
await graph.ainvoke({"messages": [input_message]})

{'messages': [HumanMessage(content='안녕하세요.', additional_kwargs={}, response_metadata={}, id='91dcce3d-45bb-4975-bec9-03b6f315c1ed'),
  AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}, id='d50f6a4a-2948-4d96-bec1-9af9e73a5845')],
 'extra_field': 10}

# stream : 중간 결과를 실시간으로 반환함
# stream_mode="values" 각 단계의 현재 상태 값 출력
# Default) stream_mode="updates" 각 단계의 상태 업데이트만 출력
# stream_mode="messages" 각 단계의 메시지 출력

for chunk in graph.stream({"messages": [input_message]}, stream_mode="values"):
    print(chunk)
    for state_key, state_value in chunk.items():
      if state_key == "messages":
        state_value[-1].pretty_print()

{'messages': [HumanMessage(content='안녕하세요.', additional_kwargs={}, response_metadata={}, id='83eca88d-a9d5-4b78-91ec-34f0a700f0f7')]}
================================ Human Message =================================

안녕하세요.
{'messages': [HumanMessage(content='안녕하세요.', additional_kwargs={}, response_metadata={}, id='83eca88d-a9d5-4b78-91ec-34f0a700f0f7'), AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}, id='071e4550-6c8e-45c7-b9b8-47a74f55a752')], 'extra_field': 10}
================================== Ai Message ==================================

안녕하세요! 무엇을 도와드릴까요?

for chunk in graph.stream({"messages": [input_message]}, stream_mode="updates"):
    print(chunk)
    for node, value in chunk.items():
        if node:
            print(node)
        if "messages" in value:
            print(value['messages'].content)

{'node': {'messages': AIMessage(content='안녕하세요! 무엇을 도와드릴까요?', additional_kwargs={}, response_metadata={}, id='24f6a799-1201-48fc-8edb-d24d197ffe10'), 'extra_field': 10}}
node
안녕하세요! 무엇을 도와드릴까요?

for chunk_msg, metadata in graph.stream({"messages": [input_message]}, stream_mode="messages"):
    print(chunk_msg)
    print(chunk_msg.content)
    print(metadata)
    print(metadata["langgraph_node"])

content='안녕하세요! 무엇을 도와드릴까요?' additional_kwargs={} response_metadata={} id='7be20061-0bf9-4463-b181-f5c31450a579'
안녕하세요! 무엇을 도와드릴까요?
{'langgraph_step': 1, 'langgraph_node': 'node', 'langgraph_triggers': ('branch:to:node',), 'langgraph_path': ('__pregel_pull', 'node'), 'langgraph_checkpoint_ns': 'node:a8a3c176-dede-0ce6-2b4b-960862af96f9'}
node

# astream : 비동기 방식으로 스트리밍 처리
async for chunk_msg, metadata in graph.astream({"messages": [input_message]}, stream_mode="messages"):
    print(chunk_msg)
    print(chunk_msg.content)
    print(metadata)
    print(metadata["langgraph_node"])

content='안녕하세요! 무엇을 도와드릴까요?' additional_kwargs={} response_metadata={} id='2146537b-db0b-41ca-a801-286857b10415'
안녕하세요! 무엇을 도와드릴까요?
{'langgraph_step': 1, 'langgraph_node': 'node', 'langgraph_triggers': ('branch:to:node',), 'langgraph_path': ('__pregel_pull', 'node'), 'langgraph_checkpoint_ns': 'node:7fad5b0b-e174-2fd7-5a31-f270219c7d19'}
node

3. 노드와 엣지 연결

from typing_extensions import TypedDict

class State(TypedDict):
    value_1: str
    value_2: int

def step_1(state: State):
    return {"value_1": state["value_1"]}

def step_2(state: State):
    current_value_1 = state["value_1"]
    return {"value_1": f"{current_value_1} b"}

def step_3(state: State):
    return {"value_2": 10}

from langgraph.graph import START, StateGraph

graph_builder = StateGraph(State)

# 노드 추가
graph_builder.add_node(step_1)
graph_builder.add_node(step_2)
graph_builder.add_node(step_3)

# 엣지 추가
graph_builder.add_edge(START, "step_1") # START ->1
graph_builder.add_edge("step_1", "step_2") # 1-> 2
graph_builder.add_edge("step_2", "step_3") # 2->3

<langgraph.graph.state.StateGraph at 0x79ead7def3e0>

graph = graph_builder.compile()
graph

graph.invoke({"value_1": "apple"})

{'value_1': 'apple b', 'value_2': 10}

4. 노드와 엣지를 한번에 연결

graph_builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
graph_builder.add_edge(START, "step_1")
graph = graph_builder.compile()

graph.invoke({"value_1": "c"})

{'value_1': 'c b', 'value_2': 10}

5. 병렬로 연결

import operator
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add] # 업데이트 값이 뒤에 추가되도록 하는 operator.add 리듀서

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

graph_builder = StateGraph(State)

# 노드 추가
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)

# graph_builder.add_edge(START, "a")
# graph_builder.add_edge("a", "b")
# graph_builder.add_edge("b", "c")
# graph_builder.add_edge("c", "d")
# graph_builder.add_edge("d", END)
# graph = graph_builder.compile()

# graph

<langgraph.graph.state.StateGraph at 0x79ead6e2ffb0>

# 엣지 추가
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b") # a -> b
graph_builder.add_edge("a", "c") # a -> c
graph_builder.add_edge("b", "d") # b -> d
graph_builder.add_edge("c", "d") # c -> d
graph_builder.add_edge("d", END)
graph = graph_builder.compile()

graph

graph.invoke({"aggregate":[]})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']
{'aggregate': ['A', 'B', 'C', 'D']}

6. 조건부 엣지 병렬 연결

import operator
from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    which: str

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

def e(state: State):
    print(f'Adding "E" to {state["aggregate"]}')
    return {"aggregate": ["E"]}

graph_builder = StateGraph(State)
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)
graph_builder.add_node(e)
graph_builder.add_edge(START, "a")

<langgraph.graph.state.StateGraph at 0x79ead7dcf740>

# bc 혹은 cd 로 라우트를 결정하는 함수
def route_bc_or_cd(state: State) -> Sequence[str]:
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]

intermediates = ["b", "c", "d"]
graph_builder.add_conditional_edges(
    "a",
    route_bc_or_cd,
    intermediates,
)

<langgraph.graph.state.StateGraph at 0x79ead7dcf740>

for node in intermediates:
    graph_builder.add_edge(node, "e")

graph_builder.add_edge("e", END)
graph = graph_builder.compile()
graph

graph.invoke({"aggregate": [], "which": "bc"})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "E" to ['A', 'B', 'C']
{'aggregate': ['A', 'B', 'C', 'E'], 'which': 'bc'}

graph.invoke({"aggregate": [], "which": "cd"})

Adding "A" to []
Adding "C" to ['A']
Adding "D" to ['A']
Adding "E" to ['A', 'C', 'D']
{'aggregate': ['A', 'C', 'D', 'E'], 'which': 'cd'}

7. 조건과 반복

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Node A 처리 중 현재 상태값 : {state["aggregate"]}')
    return {"aggregate": ["A"]}


def b(state: State):
    print(f'Node B 처리 중 현재 상태값 : {state["aggregate"]}')
    return {"aggregate": ["B"]}


graph_builder = StateGraph(State)
graph_builder.add_node(a)
graph_builder.add_node(b)

<langgraph.graph.state.StateGraph at 0x79ead6e73740>

def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

graph_builder.add_edge(START, "a")
graph_builder.add_conditional_edges("a", route)
graph_builder.add_edge("b", "a")

graph = graph_builder.compile()
graph

import requests, zlib, base64
from IPython.display import Image

# Mermaid 코드 추출
code = graph.get_graph().draw_mermaid()

# 압축·인코딩 후 Kroki 요청
encoded = base64.urlsafe_b64encode(zlib.compress(code.encode())).decode()
url = f"https://kroki.io/mermaid/png/{encoded}"
resp = requests.get(url)
display(Image(resp.content))

graph.invoke({"aggregate": []})

Node A 처리 중 현재 상태값 : []
Node B 처리 중 현재 상태값 : ['A']
Node A 처리 중 현재 상태값 : ['A', 'B']
Node B 처리 중 현재 상태값 : ['A', 'B', 'A']
Node A 처리 중 현재 상태값 : ['A', 'B', 'A', 'B']
Node B 처리 중 현재 상태값 : ['A', 'B', 'A', 'B', 'A']
Node A 처리 중 현재 상태값 : ['A', 'B', 'A', 'B', 'A', 'B']
{'aggregate': ['A', 'B', 'A', 'B', 'A', 'B', 'A']}

from langgraph.errors import GraphRecursionError
# GraphRecursionError 로 에러를 반환하는 방법
try:
    graph.invoke({"aggregate": []}, config={"recursion_limit": 4})
except GraphRecursionError: # 반복 종료 조건에 도달할 수 없는 경우
    print("Recursion Error")

Node A 처리 중 현재 상태값 : []
Node B 처리 중 현재 상태값 : ['A']
Node A 처리 중 현재 상태값 : ['A', 'B']
Node B 처리 중 현재 상태값 : ['A', 'B', 'A']
Recursion Error

8. 조건에 따른 반복 처리하기

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Node A 처리 중 현재 상태값 : {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B 처리 중 현재 상태값 : {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Node C 처리 중 현재 상태값 : {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Node D 처리 중 현재 상태값 : {state["aggregate"]}')
    return {"aggregate": ["D"]}

graph_builder = StateGraph(State)
graph_builder.add_node(a)
graph_builder.add_node(b)
graph_builder.add_node(c)
graph_builder.add_node(d)

<langgraph.graph.state.StateGraph at 0x79eabc58f740>

def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END

graph_builder.add_edge(START, "a")
graph_builder.add_conditional_edges("a", route)
graph_builder.add_edge("b", "c")
graph_builder.add_edge("b", "d")
graph_builder.add_edge(["c", "d"], "a")
graph = graph_builder.compile()
graph

result = graph.invoke({"aggregate": []})

Node A 처리 중 현재 상태값 : []
Node B 처리 중 현재 상태값 : ['A']
Node C 처리 중 현재 상태값 : ['A', 'B']
Node D 처리 중 현재 상태값 : ['A', 'B']
Node A 처리 중 현재 상태값 : ['A', 'B', 'C', 'D']
Node B 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A']
Node C 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A', 'B']
Node D 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A', 'B']
Node A 처리 중 현재 상태값 : ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']

9. 사용자 입력에 따른 반복 조건 설정

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph.message import add_messages

class State(TypedDict):
    human_messages: Annotated[list[HumanMessage], add_messages]
    ai_messages: Annotated[list[AIMessage], add_messages]
    retry_num : int

def chatbot(state:State):
    retry_num = state["retry_num"]
    user_input = input(f"(현재 {retry_num}번째 답변) 사용자 입력: ")
    ai_message = AIMessage(f"{retry_num}번째 답변중!")

    return {"human_messages": [HumanMessage(content=user_input)], "ai_messages": [ai_message]}

def retry(state: State):
    return {"retry_num" : state["retry_num"] + 1}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("retry", retry)

<langgraph.graph.state.StateGraph at 0x79eabc3a33b0>

def route(state: State) -> Literal["retry", END]:
    if "반복" in state["human_messages"][-1].content:
        return "retry"
    else:
        return END

graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", route)
graph_builder.add_edge("retry", "chatbot")
graph = graph_builder.compile()
graph

for chunk in graph.stream({"human_messages" : "반복", "retry_num": 0}, stream_mode="updates"):
    print(chunk)
    for node, value in chunk.items():
        if node:
            print(node)
        if "messages" in value:
            print(value['messages'].content)

(현재 0번째 답변) 사용자 입력: 반복
{'chatbot': {'human_messages': [HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='25936af0-dd1d-4e6e-8592-97766a2a0ac5')], 'ai_messages': [AIMessage(content='0번째 답변중!', additional_kwargs={}, response_metadata={}, id='784bfaca-131e-476c-b1a7-aac769c94251')]}}
chatbot
{'retry': {'retry_num': 1}}
retry
(현재 1번째 답변) 사용자 입력: 반복
{'chatbot': {'human_messages': [HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='20f221b0-e7bd-4ddb-9a2e-e053b0af40e7')], 'ai_messages': [AIMessage(content='1번째 답변중!', additional_kwargs={}, response_metadata={}, id='cb9bed83-0d81-4608-ab32-710833fb8287')]}}
chatbot
{'retry': {'retry_num': 2}}
retry
(현재 2번째 답변) 사용자 입력: 반복
{'chatbot': {'human_messages': [HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='4c61cd7b-4981-4836-a96f-4c15937e5b72')], 'ai_messages': [AIMessage(content='2번째 답변중!', additional_kwargs={}, response_metadata={}, id='34653647-2f0b-4c76-a564-6f3c7e0604a2')]}}
chatbot
{'retry': {'retry_num': 3}}
retry
(현재 3번째 답변) 사용자 입력: 안녕
{'chatbot': {'human_messages': [HumanMessage(content='안녕', additional_kwargs={}, response_metadata={}, id='df4af519-7bbe-4a61-9733-16de7f18809a')], 'ai_messages': [AIMessage(content='3번째 답변중!', additional_kwargs={}, response_metadata={}, id='8ef9df84-c6e3-4343-8f4f-3e9f856eeb5c')]}}
chatbot

graph.invoke({"human_messages" : "반복", "retry_num": 0})

(현재 0번째 답변) 사용자 입력: 반복
(현재 1번째 답변) 사용자 입력: 반복
(현재 2번째 답변) 사용자 입력: 반복
(현재 3번째 답변) 사용자 입력: 안녕
{'human_messages': [HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='d6abae3d-13cb-4953-8c51-387329c4a95a'),
  HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='81457744-b58d-4542-be3c-67d7473d44d6'),
  HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='c4245ba8-1293-4e09-9751-0ce21c29a189'),
  HumanMessage(content='반복', additional_kwargs={}, response_metadata={}, id='608994ba-ef93-40dc-9c83-836f2c564f2c'),
  HumanMessage(content='안녕', additional_kwargs={}, response_metadata={}, id='3a3538c4-c0ca-4dc1-9bac-8359c9165faa')],
 'ai_messages': [AIMessage(content='0번째 답변중!', additional_kwargs={}, response_metadata={}, id='e6ce5d43-564b-4a4c-be71-2cf02a76cffc'),
  AIMessage(content='1번째 답변중!', additional_kwargs={}, response_metadata={}, id='a59efbb6-10c1-45bc-b067-f25332ae4855'),
  AIMessage(content='2번째 답변중!', additional_kwargs={}, response_metadata={}, id='aaf41175-86af-4048-ad9e-2054920e321e'),
  AIMessage(content='3번째 답변중!', additional_kwargs={}, response_metadata={}, id='ee0b4a59-0a83-4ae0-9325-dc7c0e88fff1')],
 'retry_num': 3}

10. 기본 챗봇 만들기

import getpass
import os

def _set_env(var: str):
  if not os.environ.get(var):
      os.environ[var] = getpass.getpass(f'{var}: ')

_set_env("OPENAI_API_KEY")

!pip install langgraph

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
  messages: Annotated[list, add_messages]

graph_builder = StateGraph(State)

!pip install langchain_openai

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model='gpt-5-nano')

def chatbot(state: State):
  return {"messages": [llm.invoke(state['messages'])]}

graph_builder.add_node('chatbot', chatbot)
graph_builder.add_edge(START, 'chatbot')
graph_builder.add_edge('chatbot', END)

graph = graph_builder.compile()
graph

def stream_graph_updates(user_input: str):
  for event in graph.stream({'messages': [{'role': 'user', 'content': user_input}]}):
    for value in event.values():
      print('Assistant:', value['messages'][-1].content)

while True:
  user_input = input('User: ')
  if user_input.lower() in ['quit', 'exit', 'q']:
    print('프로그램을 종료합니다.')
    break
  stream_graph_updates(user_input)

Multi-Turn 챗봇 메모리 추가

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

graph = graph_builder.compile(checkpointer=memory)
config = {'configurable':{'thread_id':'1'}}

def stream_graph_updates(user_input: str):
  for event in graph.stream({'messages': [{'role': 'user', 'content': user_input}]}, config=config):
    for value in event.values():
      print('Assistant:', value['messages'][-1].content)

while True:
  user_input = input('User: ')
  if user_input.lower() in ['quit', 'exit', 'q']:
    print('프로그램을 종료합니다.')
    break
  stream_graph_updates(user_input)