LiteLLMとLangGraphのCheckpointでスレッド単位の会話履歴を永続化してみる

Python 3.13
litellm 1.83.14
langgraph 1.1.10
langgraph-checkpoint-sqlite 2.0.6
langchain-litellm 0.6.4
langchain-core 1.3.2

$ uv pip install litellm langgraph langgraph-checkpoint-sqlite langchain-litellm langchain-core

"""InMemorySaver で会話履歴を保持する最小サンプル。

thread_id ごとに会話履歴が分離される。プロセスが終了すると履歴は消える。
"""

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver

@tool
def get_weather(city: str) -> str:
    """都市の天気を返します。"""
    return f"{city}: 晴れ、24°C"

llm = ChatLiteLLM(model="openai/gpt-5-mini")
checkpointer = InMemorySaver()
agent = create_agent(model=llm, tools=[get_weather], checkpointer=checkpointer)

# thread_id で会話セッションを識別
config_a = {"configurable": {"thread_id": "user-a"}}
config_b = {"configurable": {"thread_id": "user-b"}}

# user-a が東京の天気を聞く
print("=== user-a: 東京の天気 ===")
agent.invoke(
    {"messages": [{"role": "user", "content": "東京の天気は？"}]},
    config=config_a,
)

# user-b が大阪の天気を聞く（別スレッド）
print("=== user-b: 大阪の天気 ===")
agent.invoke(
    {"messages": [{"role": "user", "content": "大阪の天気は？"}]},
    config=config_b,
)

# user-a が「さっきの都市は？」と聞く -> 東京と覚えている
print("\n=== user-a: 文脈を引き継いで質問 ===")
result = agent.invoke(
    {"messages": [{"role": "user", "content": "さっき聞いた都市の名前は？"}]},
    config=config_a,
)
print(result["messages"][-1].content)

# user-b の同じ質問 -> 大阪と覚えている
print("\n=== user-b: 文脈を引き継いで質問 ===")
result = agent.invoke(
    {"messages": [{"role": "user", "content": "さっき聞いた都市の名前は？"}]},
    config=config_b,
)
print(result["messages"][-1].content)

$ python inmemory_checkpoint.py
=== user-a: 東京の天気 ===
=== user-b: 大阪の天気 ===

=== user-a: 文脈を引き継いで質問 ===
先ほどお尋ねした都市名は「東京」です。

=== user-b: 文脈を引き継いで質問 ===
さっき聞いた都市は「大阪」です。ほかの都市の天気も知りたいですか？

"""SqliteSaver でディスクに会話履歴を永続化する。

プロセスを再起動しても thread_id ごとの履歴が復元できる。
"""

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.sqlite import SqliteSaver

@tool
def remember_fact(fact: str) -> str:
    """ユーザーに関する情報を記憶します（実際にはstateに残ることでLLMが参照する）。"""
    return f"記憶しました: {fact}"

llm = ChatLiteLLM(model="openai/gpt-5-mini")

# context manager で SQLite ファイルを開く
with SqliteSaver.from_conn_string("./checkpoints.db") as checkpointer:
    agent = create_agent(model=llm, tools=[remember_fact], checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "session-001"}}

    # 1回目の起動: 名前を伝える
    print("=== 1回目: 名前を伝える ===")
    agent.invoke(
        {"messages": [{"role": "user", "content": "私の名前はkobayashiです"}]},
        config=config,
    )

# 上の with を抜けたところで一度終了したと仮定し、新たに開き直す
print("\n--- プロセス再起動を模した再オープン ---\n")

with SqliteSaver.from_conn_string("./checkpoints.db") as checkpointer:
    agent = create_agent(model=llm, tools=[remember_fact], checkpointer=checkpointer)
    config = {"configurable": {"thread_id": "session-001"}}

    print("=== 2回目: 名前を聞く ===")
    result = agent.invoke(
        {"messages": [{"role": "user", "content": "私の名前を覚えていますか？"}]},
        config=config,
    )
    print(result["messages"][-1].content)

$ python sqlite_checkpoint.py
=== 1回目: 名前を伝える ===

--- プロセス再起動を模した再オープン ---

=== 2回目: 名前を聞く ===
はい、覚えています。kobayashiさんです。今後もその名前でお呼びしてよろしいですか？別の呼び方が良ければ教えてください。

"""長くなった会話履歴を要約・トリミングする戦略。

会話を続けるとメッセージが膨らみ、コンテキストウィンドウを圧迫する。
ここでは「N件を超えたら要約に置き換える」シンプルな戦略を実装する。
"""

import litellm
from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
    RemoveMessage,
    SystemMessage,
)
from langchain_litellm import ChatLiteLLM
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph

# Anthropic（summarizer）はツール無し呼出時に modify_params=True が必要
litellm.modify_params = True

llm = ChatLiteLLM(model="openai/gpt-5-mini")
# 要約は会話文脈の品質に直結するため Claude を割当
summarizer_llm = ChatLiteLLM(model="anthropic/claude-sonnet-4-6")

# 履歴を要約に置き換えるしきい値
SUMMARY_THRESHOLD = 6

class State(MessagesState):
    summary: str

def chat_node(state: State) -> dict:
    """要約があればシステムプロンプトに含めて応答する。"""
    summary = state.get("summary") or ""
    sys = (
        f"これまでの会話の要約:\n{summary}"
        if summary
        else "通常の会話を続けてください。"
    )
    response = llm.invoke([SystemMessage(content=sys), *state["messages"]])
    return {"messages": [response]}

def summarize_node(state: State) -> dict:
    """しきい値を超えたら古いメッセージを要約に置き換える。"""
    messages: list[BaseMessage] = state["messages"]
    existing_summary = state.get("summary") or ""
    target = messages[:-2]  # 直近2件は残す
    summary_input = "\n".join(
        f"{m.type}: {m.content}" for m in target if isinstance(m.content, str)
    )
    prompt = (
        f"これまでの要約:\n{existing_summary}\n\n"
        f"新しい会話:\n{summary_input}\n\n"
        "上記を統合し、ユーザーに関する重要な情報を保持しつつ短く要約してください。"
    )
    response = summarizer_llm.invoke([HumanMessage(content=prompt)])

    # 古いメッセージは削除（直近2件のみ残す）
    delete_msgs = [RemoveMessage(id=m.id) for m in target if m.id]
    print(f"  [Summarize] {len(delete_msgs)} 件を要約に圧縮")
    return {"summary": str(response.content), "messages": delete_msgs}

def should_summarize(state: State) -> str:
    if len(state["messages"]) >= SUMMARY_THRESHOLD:
        return "summarize"
    return "end"

graph = StateGraph(State)
graph.add_node("chat", chat_node)
graph.add_node("summarize", summarize_node)

graph.add_edge(START, "chat")
graph.add_conditional_edges(
    "chat",
    should_summarize,
    {"summarize": "summarize", "end": END},
)
graph.add_edge("summarize", END)

app = graph.compile(checkpointer=InMemorySaver())

# 8 ターンの会話を投げて要約発火を観察
config = {"configurable": {"thread_id": "demo"}}
turns = [
    "こんにちは、私はkobayashiです。",
    "趣味は釣りです。",
    "好きな魚はマグロです。",
    "東京に住んでいます。",
    "猫を2匹飼っています。",
    "コーヒーが好きです。",
    "今日はとても疲れました。",
    "ここまでで覚えていること、私の趣味は何でしたか？",
]

for i, t in enumerate(turns, 1):
    print(f"\n[Turn {i}] USER: {t}")
    result = app.invoke({"messages": [HumanMessage(content=t)]}, config=config)
    last: AIMessage = result["messages"][-1]
    print(f"AI: {str(last.content)[:120]}")
    if result.get("summary"):
        print(f"(要約: {result['summary'][:120]})")

$ python history_management.py

[Turn 1] USER: こんにちは、私はkobayashiです。
AI: こんにちは、kobayashiさん。よろしくお願いします。お元気ですか？

[Turn 2] USER: 趣味は釣りです。
AI: いいですね、kobayashiさん。釣りがお好きなんですね！どんな釣りをよくされますか？...

[Turn 3] USER: 好きな魚はマグロです。
  [Summarize] 4 件を要約に圧縮
AI: いいですね、マグロが好きなんですね！...
(要約: **ユーザー情報まとめ：**
- 名前：kobayashi
- 趣味：釣り（詳細な種類・スタイルは未回答）)

[Turn 4] USER: 東京に住んでいます。
AI: 東京にお住まいですね。情報ありがとうございます。マグロ狙い（釣る／食べる）どちらが目的かでおすすめが変わりますが、まず釣り前提で東京からアクセスしやすい選択肢をまとめます。
....

[Turn 5] USER: 猫を2匹飼っています。
  [Summarize] 4 件を要約に圧縮
AI: 猫ちゃんが2匹いらっしゃるんですね...
(要約: **ユーザー情報まとめ：**
- 名前：kobayashi
- 居住地：東京
- 趣味：釣り
- 好きな魚：マグロ
- 現在の関心：マグロ釣り・食べ方どちらに興味があるか未回答)

[Turn 6] USER: コーヒーが好きです。
AI: コーヒーがお好きなんですね！情報をプロフィールに追加しました。
...

[Turn 7] USER: 今日はとても疲れました。
  [Summarize] 4 件を要約に圧縮
AI: それはお疲れさまでした...

[Turn 8] USER: ここまでで覚えていること、私の趣味は何でしたか？
AI: 覚えていることを手短に言うと、趣味は「釣り」です。
（補足で覚えている点：好きな魚はマグロ、好きな飲み物はコーヒー、東京在住で猫が2匹います。）