<div id="readability-page-1" class="page"><div>
                <div id="SE-094b4140-d2ab-4781-ac7f-7db43cf611d6">
                                    <p id="SE-15fabfcd-11cb-4930-9b2a-a21ce95da755"><span id="SE-b4bf4d2c-ec52-4bde-b504-54dbae3e31c0">안녕하세요 멍개입니다.</span></p><p id="SE-85af0d75-250e-45e9-b2af-7c0df1fa521c"><span id="SE-99655d2d-9aa1-4910-b1df-de5ebd50ef9b">​</span></p><p id="SE-c557c152-e2f3-4107-88ce-e0694c4a2640"><span id="SE-9db81f52-9a1b-4390-b9bf-d765e5ab4ef0">1편부터 3편동안 LangGraph의 구성요소와 LangGraph를 이용하여 Agent를 만드는 방법을 알아보았습니다.</span></p><p id="SE-954737b7-72ac-43e4-8204-7f328e30a783"><span id="SE-30f0ac69-deca-4de3-93fe-f4137ac369a8">​</span></p><p id="SE-b69bbfb7-161e-44e4-b28b-043b92013a10"><span id="SE-e96ef65b-ce6a-4786-9c93-52b59759298e">이번 시간은 LangGraph에서 상태(기록)을 저장하고 관리하는 checkpoint에 대해서 알아보겠습니다.</span></p><p id="SE-3e44f9a7-58f0-40eb-91e5-308b9e442f75"><span id="SE-d40a222c-b4b8-454a-8667-10a4c9b843f6">​</span></p><p id="SE-7f2b417e-61d2-4509-bcaa-434b653d7910"><span id="SE-fe2bbd23-3ca3-4226-850b-436cb3df427d">checkpoint의 핵심 기능은 대화 컨텍스트 유지입니다.</span></p><p id="SE-332a4a5c-8808-4712-bf99-6da417ae6e43"><span id="SE-68c2f477-7e6e-4a67-8a6b-81cb8d7456a1">​</span></p><p id="SE-7afe94ae-cb64-4fb0-88cf-565c9448b16d"><span id="SE-5b450767-e530-4e44-a134-6496e5dd5c81"><b>· checkpoint 없을 때</b></span></p>
                            </div>                <div id="SE-640302c2-a9bd-4334-800b-06f3448bc4f0">
                                    <p>from langgraph.graph import StateGraph, MessagesState, START
from langchain_openai import ChatOpenAI

from dotenv import load_dotenv
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

model = ChatOpenAI(
    base_url="http://localhost:1234/v1",
    model_name="l3-8b-stheno-v3.1-iq-imatrix"
)

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

input_message = {"type": "user", "content": "안녕, 나의 이름은 홍길동이야"}
for chunk in graph.stream({"messages": [input_message]}, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

input_message = {"type": "user", "content": "내 이름이 뭐야?"}
for chunk in graph.stream({"messages": [input_message]}, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
</p>
                                </div>                <div id="SE-47fa326f-1696-4720-9932-be2c82b42987">
                                    <p>================================ Human Message =================================

안녕, 나의 이름은 홍길동이야
================================== Ai Message ==================================

안녕하세요! 저는 AI 기반의 언어 모델로 다양한 언어와 대화를 할 수 있습니다. 어떻게 도울 수 있는지 궁금하신가요? 제게 질문해주세요! 🤖👋
================================ Human Message =================================

내 이름이 뭐야?
================================== Ai Message ==================================

My name is LLaMA, a language model AI developed by Meta AI. I'm here to assist and provide information on various topics to the best of my ability. How can I help you today?</p>
                                </div>                <div id="SE-c5f52870-266a-4b27-8bc2-2d378127a8c8">
                                    <p id="SE-426e112c-8703-41d1-b85a-74980b138634"><span id="SE-c47e8bdb-e8dc-46ab-9852-c07d33af7712"><b>· checkpoint 있을 때</b></span></p><p id="SE-78f0bd12-5880-4fa2-908a-4133c7574e7f"><span id="SE-8e16b7f4-b394-4e3a-9185-722660e488b2">checkpoint는 graph를 빌드할 때 checkepointer를 전달하면 됩니다.</span></p>
                            </div>                <div id="SE-e9fa4f7c-b5f9-43fe-9bfa-f05055b06234">
                                    <p>from langgraph.graph import StateGraph, MessagesState, START
from langchain_openai import ChatOpenAI

from dotenv import load_dotenv
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

model = ChatOpenAI(
    base_url="http://localhost:1234/v1",
    model_name="l3-8b-stheno-v3.1-iq-imatrix"
)

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "1"}}

input_message = {"type": "user", "content": "안녕, 나의 이름은 홍길동이야"}
for chunk in graph.stream({"messages": [input_message]}, config=config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

input_message = {"type": "user", "content": "내 이름이 뭐야?"}
for chunk in graph.stream({"messages": [input_message]}, config=config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
</p>
                                </div>                <div id="SE-e29bc6f8-1155-4521-95ea-576caf39bfeb">
                                    <p>================================ Human Message =================================

안녕, 나의 이름은 홍길동이야
================================== Ai Message ==================================

안녕하세요! 제 이름도 홍길동입니다. 만나서 반가워요~! 😊👋 

몇살이에요? 그리고 어디에서 사세요? 🏠🌍

너무 좋아하는 hobby나 취미가 있나요? 🎨💻🏀

이제부터는 우리가 친구가 되기로 했어요! 같이 좋은 추억 만들어요~! 😄
================================ Human Message =================================

내 이름이 뭐야?
================================== Ai Message ==================================

안녕하세요! 제 이름은 Hong Gil-dong입니다. 만나서 반가워요~! 😊👋

몇살이에요? 그리고 어디에서 사세요? 🏠🌍

너무 좋아하는 hobby나 취미가 있나요? 🎨💻🏀

이제부터는 우리가 친구가 되기로 했어요! 같이 좋은 추억 만들어요~! 😄</p>
                                </div>                <div id="SE-a4e15019-fc73-41c2-b759-bb5497b485d6">
                                    <p id="SE-030a9274-64a3-49c9-9915-681f9c1666cb"><span id="SE-d5534855-27a8-494b-97c9-3d3fd3e620c7"><b>· thread_id로 관리되는 체크포인트</b></span></p><p id="SE-cdc0cdda-0f52-406c-a890-f8dc4c1bde64"><span id="SE-7de7109c-aba3-4b19-9f7e-b3af0a462ead">LangGraph에서 checkpoint는 thread_id 마다의 state를 관리합니다. 이는 get_state를 통해 읽거나 update_state를 통해 수정할 수 있습니다.</span></p>
                            </div>                <div id="SE-ede46313-d787-4077-baca-bfe3c73500f6">
                                    <p>from langgraph.graph import StateGraph, MessagesState, START
from langchain_openai import ChatOpenAI

from dotenv import load_dotenv
from langgraph.checkpoint.memory import MemorySaver

load_dotenv()

model = ChatOpenAI(
    base_url="http://localhost:1234/v1",
    model_name="l3-8b-stheno-v3.1-iq-imatrix"
)

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}


builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

config1 = {"configurable": {"thread_id": "1"}}
config2 = {"configurable": {"thread_id": "2"}}

input_message = {"type": "user", "content": "안녕, 나의 이름은 홍길동이야"}
for chunk in graph.stream({"messages": [input_message]}, config=config1, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

input_message = {"type": "user", "content": "내 이름이 뭐야?"}
for chunk in graph.stream({"messages": [input_message]}, config=config1, stream_mode="values"):
    chunk["messages"][-1].pretty_print()


input_message = {"type": "user", "content": "안녕, 나의 이름은 박정태야"}
for chunk in graph.stream({"messages": [input_message]}, config=config2, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

# state 가져오기
print(graph.get_state(config1))
print()
print(graph.get_state(config2))

# state 복구
graph.update_state(config1, graph.get_state(config1))

input_message = {"type": "user", "content": "내 이름이 뭐야?"}
for chunk in graph.stream({"messages": [input_message]}, config=config1, stream_mode="values"):
    chunk["messages"][-1].pretty_print()</p>
                                </div>                <div id="SE-c536df92-0f2f-4869-9bd0-525185ff23bd">
                                <p><a href="#" onclick="return false;" data-linktype="img" data-linkdata="{&quot;id&quot; : &quot;SE-c536df92-0f2f-4869-9bd0-525185ff23bd&quot;, &quot;src&quot; : &quot;https://mblogthumb-phinf.pstatic.net/MjAyNTA1MTFfMjYx/MDAxNzQ2OTQyNTc3NDUw.Sa3LixUSBZeymMp2nNk7WUtmrS6bJMoSeFDmBsllZs4g.L9xN8bNFdGp8GEQv_VY-09tWGK4GbRkEypVchLJ0z2Yg.PNG/Screenshot_2025-05-11_at_14.49.25.png&quot;, &quot;originalWidth&quot; : &quot;2642&quot;, &quot;originalHeight&quot; : &quot;1584&quot;, &quot;linkUse&quot; : &quot;false&quot;, &quot;link&quot; : &quot;&quot;}">
                                        <img src="https://mblogthumb-phinf.pstatic.net/MjAyNTA1MTFfMjYx/MDAxNzQ2OTQyNTc3NDUw.Sa3LixUSBZeymMp2nNk7WUtmrS6bJMoSeFDmBsllZs4g.L9xN8bNFdGp8GEQv_VY-09tWGK4GbRkEypVchLJ0z2Yg.PNG/Screenshot_2025-05-11_at_14.49.25.png?type=w800" data-width="693" data-height="415" alt="">
                                    </a>
                                </p>
                        </div>
                <div id="SE-e302e1e4-abe4-4636-b25e-67ae7ebdea00">
                            <p id="SE-b0d9829b-670b-40ff-9922-a3d62b792dbc"><span id="SE-f5ca7fe3-c185-4d7a-96e9-38c1febe24a7"><b>· functional api 기반의 체크포인트 설정</b></span></p>
                        </div>                <div id="SE-3bdd6f45-3686-4b74-bb72-1623da592aad">
                                    <p>from langchain_core.messages import AnyMessage
from langgraph.graph import add_messages
from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

model = ChatOpenAI(
    base_url="http://localhost:1234/v1",
    model_name="l3-8b-stheno-v3.1-iq-imatrix"
)

@task
def call_model(messages: list[AnyMessage]):
    response = model.invoke(messages)
    return response

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def workflow(inputs: list[AnyMessage], *, previous: list[AnyMessage]):
    if previous:
        inputs = add_messages(previous, inputs)

    response = call_model(inputs).result()
    return entrypoint.final(value=response, save=add_messages(inputs, response))

config = {
    "configurable": {
        "thread_id": "1"
    }
}

for chunk in workflow.stream(
    [{"role": "user", "content": "hi! I'm bob"}],
    config,
    stream_mode="values",
):
    chunk.pretty_print()

for chunk in workflow.stream(
    [{"role": "user", "content": "what's my name?"}],
    config,
    stream_mode="values",
):
    chunk.pretty_print()</p>
                                </div>                                <div id="SE-484bbe78-a84b-4341-a9c1-058164bb6c5d">
                                    <p>thread_id = "1"
checkpointer.delete_thread(thread_id)</p>
                                </div>                <div id="SE-9f2c9aff-176a-4b68-a151-622810022057">
                                    <p id="SE-50b8a00b-034d-42d6-98bf-8108d6767e62"><span id="SE-334992a3-c8ab-4dda-a382-ae1bd98ece4d">​</span></p><p id="SE-d3f958ae-e159-4b0c-bb65-d448c9ebbf4e"><span id="SE-87c795d4-de64-494c-ba41-e7383f2a4365"><b>● 체크포인트 사용 사례</b></span></p><p id="SE-9032d0eb-60bc-460c-8b35-0a152cd6b93f"><span id="SE-7dbf8c7e-1f40-4e43-b6fe-f7d8aff4a719">체크포인트는 여러 상황에서 사용가능합니다.</span></p><p id="SE-18c80b4e-7e89-4c5b-8af0-1ee0ab465a73"><span id="SE-e4860f71-705b-40e1-8b8f-f2eba16bdbe7">​</span></p><p id="SE-9d950c14-6083-440e-add7-c200c59c184e"><span id="SE-274cdc94-b683-4930-8969-31f9ca03793f"><b>· 오류 복구를 위한 체크포인트</b></span></p><p id="SE-350dc394-756f-4dde-9906-b5624bbe9868"><span id="SE-b4231eca-1641-4189-88fd-592b8832c921">exception이 발생하면 해당 config의 상태를 가져와서 update_state로 상태를 업데이트 합니다.</span></p>
                            </div>                <div id="SE-bdaa3835-386c-4c88-98e5-d688d3e00764">
                                    <p>try:
    result = graph.invoke(input_data, config=config)
except Exception as e:
    print(f"오류 발생: {e}")
    last_stable_state = graph.get_state(config)
    graph.update_state(config, last_stable_state.values)
    print("마지막 안정 상태로 복구됨")</p>
                                </div>                <div id="SE-b45875a2-f2a8-443b-89e5-d76f226f2b0e">
                                    <p id="SE-ed9268da-0d5a-43db-9dcb-ba64186dca5b"><span id="SE-8a40b589-61cf-4a99-9693-d88ca3338b8b">​</span></p><p id="SE-0999d3e1-c3be-4814-b1d5-7041f7b111c9"><span id="SE-c9cb21f8-847a-44d2-b9dd-6421aa1f1023"><b>· HITL</b></span></p><p id="SE-db8c2d42-17a4-4704-920b-0fed47a96d0c"><span id="SE-84c9a91e-ff2e-4c4c-9a14-a0788ce0486b">human_review_node가 실행되면 need_human_review에 따라 exception을 발생합니다. 이때 except 구문에서 config(thread_id)에 따른 update_state를 통해 상태를 변경해줍니다.</span></p>
                            </div>                <div id="SE-b96272dc-e6bd-468e-a8bc-ffa5120aec14">
                                    <p>def human_review_node(state):
    if need_human_review(state):
        raise NodeInterrupt("인간 검토 필요")
    return state

# 그래프 설정
workflow.add_node("human_review", human_review_node)
graph = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])

# 실행
try:
    # ... graph.invoke()
    # human_review_node 실행 전에 중단된 상태
    result = graph.invoke(input_data, config=config)
except NodeInterrupt as e:
    print("인간 검토를 위해 중단됨")
    # 인간 입력 처리 로직
    human_input = get_human_input()
    graph.update_state(config, {"human_feedback": human_input})
    result = graph.invoke(None, config=config)  # 실행 재개</p>
                                </div>                <div id="SE-7ee4d6dd-3e2f-444b-8c56-4d3254f74101">
                                    <p id="SE-edb1605d-052a-4a63-92c2-ff034ed745ea"><span id="SE-31beaa79-eba8-4dbc-829e-72f4408645e1"><b>· A/B 테스트</b></span></p><p id="SE-d4a4fc66-9177-4ab4-9375-247013a3684a"><span id="SE-f5fb1d24-0d9a-4b98-9c1a-e78015d11e58">서로 다른 config를 통해 A/B 테스트를 수행할 수 있습니다.</span></p>
                            </div>                <div id="SE-9dd2aafd-bba8-4111-9e16-f500b590f5dd">
                                    <p>def create_state_version(config, version_name):
    current_state = graph.get_state(config).values
    versioned_config = {**config, "configurable": {**config["configurable"], "version": version_name}}
    graph.update_state(versioned_config, current_state)
    return versioned_config

# 버전 생성
version_a_config = create_state_version(config, "version_a")
version_b_config = create_state_version(config, "version_b")

# 각 버전에서 실험
result_a = graph.invoke(input_data, config=version_a_config)
result_b = graph.invoke(input_data, config=version_b_config)</p>
                                </div>                <div id="SE-86483110-fdbf-4350-825b-2311350ad47a">
                                    <p id="SE-b8fa3f51-87a9-4e8d-bbce-4de8cc1f7ca7"><span id="SE-88e9514b-d341-4ae6-b5b7-98469b5f03bb"><b>· 기록 분석</b></span></p><p id="SE-b473d5b9-21b8-4c70-b9b2-d0f9d7aea9e3"><span id="SE-c44b1c92-63a5-47d4-9e62-495427cf1d28">LangGraph는 get_state_history를 통해 상태가 어떻게 변화했는지 리스트로 가지고 있습니다.</span></p>
                            </div>                <div id="SE-e7810562-c992-4ff0-9927-4ed78121daed">
                                    <p>history = list(graph.get_state_history(config))
for i, checkpoint in enumerate(history):
    print(f"Checkpoint {i}:")
    print(f"  Time: {checkpoint.created_at}")</p>
                                </div>                <div id="SE-ec509883-66a3-4b75-8ba8-d90a41ac1dbb">
                                    <p id="SE-d970ce7e-6d4e-499b-beb2-a9e7d1b7fc06"><span id="SE-64799a16-70a7-42e2-93dd-45d6634af348"><b>· 영속성</b></span></p><p id="SE-d46694ac-7b47-42ab-aa7a-59037e6e3c3a"><span id="SE-88459ab6-3b61-408b-b9d0-315afed93b68">LangGraph의 체크포인트는 인메모리가 아닌 postgres와 같은 데이터 베이스로 설정할 수 있습니다.</span></p>
                            </div>                <div id="SE-03282345-0719-46a9-a544-e19d33146456">
                                    <p>from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    builder = StateGraph(...)
    graph = builder.compile(checkpointer=checkpointer)</p>
                                </div>                    </div></div>

삽질하는 멍개씨

안녕하세요 멍개입니다. 1편부터 3편동안 LangGraph의 구성요소와 LangGraph를 이용하여 Agent를 만드는 방법을 알아보았습니다. 이번 시간은 LangGraph에서 상태(기록)을 저장하고 관리하는 checkpoint에 대해서 알아보겠습니다. checkpoint의 핵심 기능은 대화 컨텍스트 유지입니다. · checkpoint 없을 때 · checkpoint 있을 때 checkpoint는 graph를 빌드할 때 checkepointer를 전달하면 됩니다. · thread_id로 관리되는 체크포인트 LangGraph에서 checkpoint는 thread_id 마다의 state를 관리합니다. 이는 get_state를 통해 읽거나 update_state를 통해 수정할 수 있습니다. · functional api 기반의 체크포인트 설정 · 체크포인트 삭제 ● 체.......

[langgraph] 4편 - checkpoint를 이용한 히스토리 관리

댓글