LangGraph Agent: 챗봇에서 Tool·Memory·HITL·Time Travel까지

from langchain.chat_models import init_chat_model

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

llm.invoke([{"role": "user", "content": "Hello!"}])
# → AIMessage(content='Hello! How can I assist you today?', ...)

from langchain_core.messages import AnyMessage
import operator
from typing import Annotated


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

from langgraph.graph.message import add_messages

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

def chatbot(state: State):
    # state 안의 messages를 그대로 모델에 전달
    response = llm.invoke(state["messages"])
    # LLM이 반환한 response를 messages에 넣어 반환 → add_messages가 누적
    return {"messages": [response]}


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

graph = graph_builder.compile()

graph.invoke(
    {"messages": [{"role": "user", "content": "how are you?"}]}
)
# → {'messages': [HumanMessage(content='how are you?', ...),
#                 AIMessage(content="I'm just a computer program, ...", ...)]}

from langgraph.prebuilt import ToolNode, tools_condition

tool_node = ToolNode(
    tools=[],
)

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)

graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges("chatbot", tools_condition)
graph_builder.add_edge("tools", "chatbot")

from langchain_core.tools import tool


@tool
def get_weather(city: str):
    """Gets weather in city"""
    return f"The weather in {city} is sunny"


# AI 모델이 이 tool을 알고 있어야 한다 → LLM을 tool과 바인딩(결합)
llm_with_tools = llm.bind_tools(tools=[get_weather])


def chatbot(state: State):
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}


# node가 호출할 tool을 등록
tool_node = ToolNode(
    tools=[get_weather],
)

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

# SQLite connection
conn = sqlite3.connect("memory.db", check_same_thread=False)

# SqliteSaver를 compile 시 checkpointer로 넘긴다
graph = graph_builder.compile(
    checkpointer=SqliteSaver(conn),
)

graph.invoke(
    {"messages": [{"role": "user", "content": "how are you?"}]},
    config={
        "configurable": {
            "thread_id": "1"
        },
        # "recursion_limit": 2  # 그래프가 몇 step까지 허용되는지
    },
)

# state 이력 탐색 — 어느 대화의 history인지 알아야 하므로 config를 넘긴다
for state in graph.get_state_history(
    {"configurable": {"thread_id": "1"}}
):
    print(state.next)

# 결과 (아래에서 위로 진행된 순서)
# ()
# ('chatbot',)
# ('__start__',)

print("hello")
print("world")
input()      # ← 여기서 대기, 입력 전까지 blocking
print("bye")

from langgraph.types import interrupt


@tool
def get_human_feedback(poem: str):
    """
    Asks the user for feedback on the poem.
    Use this before returning the final response.
    """
    # interrupt 실행 → 그래프 중단. 재개 시 응답이 feedback으로 들어온다
    feedback = interrupt(f"Here is the poem, tell me what you think\n{poem}")
    return feedback


# interrupt 발생 여부 확인
result = graph.invoke(
    {"messages": [{"role": "user", "content": "Please make a poem about Python Code."}]},
    config=config,
)

for message in result["messages"]:
    # pretty_print — 메시지를 사람이 보기 좋게 변환해 주는 langchain 메서드
    message.pretty_print()

from langgraph.types import Command

# interrupt 질문에 대한 응답을 Command로 제공
response = Command(
    # resume 값이 곧 interrupt의 반환값이 된다.
    # 문자열·딕셔너리 등 모든 형태 가능. 단, interrupt 쪽에서 받을 준비가 되어 있어야 한다.
    resume="It looks good!!"
)

# 이 응답으로 그래프를 다시 호출 → 재개
# state 대신 Command를 전달하면 그래프가 재개된다.
# config가 있어야 어떤 thread_id로 재개할지 알 수 있다.
result = graph.invoke(response, config=config)

# config를 넘겨야 thread_id로 어떤 대화를 가리키는지 알 수 있다
state_history = graph.get_state_history(config)

for state_snapshot in list(state_history):
    print(state_snapshot.next)              # 각 단계에서 실행할 노드 이름
    print(state_snapshot.values["messages"])  # 메시지만 출력
    print("=========\n")

# 포크할 위치 선택 (기록의 -5번째라고 가정)
# a = [1, 2, 3, 4, 5] → a[-1] = 5
state_history = graph.get_state_history(config)
to_fork = list(state_history)[-5]
to_fork.values["messages"]

from langchain_core.messages import HumanMessage

# 그래프 상태 업데이트 (fork)
graph.update_state(
    # 기본 config가 아니라, "포크할 위치"를 가리키는 config (checkpoint_id 포함).
    # 이걸 넘기면 LangGraph는 이 시점의 state를 업데이트하려는 것으로 인식한다.
    to_fork.config,
    # 무엇을 — 기존 메시지 ID를 그대로 넘기면 그 메시지를 수정한다
    {
        "messages": HumanMessage(
            content="I live in Japan",
            id="8eac45a4-688c-4c2c-870d-29673e964d5f",
        )
    },
)

# update_state 후 받은 checkpoint_id로 변경 확인
forked_state = graph.get_state_history(
    {"configurable": {
        "thread_id": "1",
        "checkpoint_ns": "",
        "checkpoint_id": "1f17f873-caef-612e-8004-a02fab4321a6",
    }}
)
list(forked_state)

# 그 지점에서 그래프를 다시 호출 (state 대신 None → 저장된 state로 재개)
result = graph.invoke(
    None,
    {"configurable": {
        "thread_id": "1",
        "checkpoint_ns": "",
        "checkpoint_id": "1f17f873-caef-612e-8004-a02fab4321a6",
    }},
)

for message in result["messages"]:
    message.pretty_print()

{
  "dependencies": ["langchain_openai", "./main.py"],
  "env": "./.env",
  "graphs": {
    "mr_poet": "./main.py:graph"
  }
}

langgraph dev