LangGraphを使用してAIアプリケーションを構築する際、最も頭を悩ませる問題の一つが「会話コンテキストの保持」です。ユーザーが会話を中断した後再開したとき、previousの文脈が消えていてはまるで別の人に話しかけているようで用户体验は台無しです。
本稿では、HolySheep AIを活用したLangGraphの状態管理について、プログラミング経験がない完全な初心者でも理解できる説明を尽くします。スクリーンショットの代わりにテキストヒント隨所に記載しますので、一緒に手を動かしながら學習しましょう。
前提條件：LangGraphとは
LangGraphは、複雑なAIワークフローを「グラフ」として定義できるライブラリです。ノード（處理步骤）とエッジ（遷移條件）で構成され、会話のような狀態を持つアプリケーションに最適です。LangChainファミリーの一員として、 HolySheep AI やOpenAI等多种APIと組み合わせ可能です。
状態管理の基本概念
なぜ状態管理が重要か
従来のAPI呼び出しでは、各リクエストは獨立しています。前の質問で「日本の首都は？」と聞いた翌日に「それは哪里？」と聞いても、AIは返答できません。しかし実際の应用中、ユーザーは以下を期待します：
- 中断した場所からの續行
- 複数のセッション間の文脈保持
- システム故障時の自動復元
- スケールアウト時の狀態共有
LangGraphの状態構造
LangGraphでは、StateGraphというクラスを使って狀態を管理します。以下のコードは最もシンプルな例です：
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
状態の型定義
class ConversationState(TypedDict):
messages: list # 会話履歴
user_id: str # ユーザー識別子
session_id: str # セッション識別子
グラフの構築
graph = StateGraph(ConversationState)
def process_message(state: ConversationState, new_message: str) -> ConversationState:
"""新しいメッセージを狀態に追加"""
return {
"messages": state["messages"] + [{"role": "user", "content": new_message}],
"user_id": state["user_id"],
"session_id": state["session_id"]
}
ノードとエッジの追加
graph.add_node("chat", process_message)
graph.add_edge("chat", END)
コンパイル
app = graph.compile()
永続化の三つの柱：チェックポイント、恢复、狀態永続化
1. チェックポイント方式（推奨）
LangGraphのMemorySaverは、最もシンプルな永続化解决方案です。以下の點がポイントです：
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END
from typing import TypedDict
class ChatState(TypedDict):
messages: list
context_summary: str
チェックポインターの設定
checkpointer = MemorySaver()
graph = StateGraph(ChatState)
def chat_node(state: ChatState, input_data: dict) -> ChatState:
""" HolySheep AI APIを呼び出して応答生成 """
import requests
# APIリクエスト構築
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4o",
"messages": state["messages"],
"temperature": 0.7
}
)
result = response.json()
assistant_message = result["choices"][0]["message"]
return {
"messages": state["messages"] + [assistant_message],
"context_summary": f"会話{len(state['messages'])}件の履歴を保持中"
}
graph.add_node("chat", chat_node)
graph.add_edge("chat", END)
app = graph.compile(checkpointer=checkpointer)
最初の会話
config = {"configurable": {"thread_id": "user123-session1"}}
result1 = app.invoke(
{"messages": [{"role": "user", "content": "日本の四季について教えて"}],
"context_summary": ""},
config
)
同じthread_idで再開（コンテキスト保持）
result2 = app.invoke(
{"messages": [{"role": "user", "content": "特に春有何特徴？"}],
"context_summary": result1["context_summary"]},
config
)
print("復元された文脈:", result2["context_summary"])
ポイント：thread_idを固定にすることで、同一スレッド内の狀態が 자동으로復元されます。result1에서戻ったcontext_summary를 result2에 전달하는 점이重要です。
2. PostgreSQL永続化（本番環境推奨）
實際のアプリケーションでは、メモリだけでなくデータベースへの永続化が 필수です。以下の構成で信頼性の高い狀態管理が実現できます：
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import StateGraph
import psycopg2
データベース接続設定
DB_CONFIG = {
"host": "your-db-host",
"port": 5432,
"dbname": "langgraph_state",
"user": "your_user",
"password": "your_password"
}
class ProductionState(TypedDict):
messages: list
user_id: str
session_id: str
metadata: dict
PostgresSaverの初期化
checkpointer = PostgresSaver.from_conn_string(
"postgresql://your_user:your_password@your-db-host:5432/langgraph_state"
)
checkpointer.setup() # テーブル自動作成
graph = StateGraph(ProductionState)
def ai_chat(state: ProductionState) -> ProductionState:
"""HolySheep AI APIで文脈を考慮した応答生成"""
import requests
# 会話履歴 الكاملةをAPIに送信
full_context = state["messages"]
api_response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "claude-sonnet-4-20250514", # HolySheepの低価格モデル
"messages": full_context,
"max_tokens": 1000
},
timeout=30
).json()
assistant_reply = api_response["choices"][0]["message"]
return {
"messages": state["messages"] + [assistant_reply],
"user_id": state["user_id"],
"session_id": state["session_id"],
"metadata": {"last_updated": "2024-01-15T10:30:00Z"}
}
graph.add_node("chat", ai_chat)
app = graph.compile(checkpointer=checkpointer)
スレッド単位での狀態復元
config = {
"configurable": {
"thread_id": "production-user-456",
"checkpoint_ns": "production"
}
}
特定チェックポイントからの復元
checkpoint_id = "1beab6e9-4f4a-42c3-9b5e-8d7c3a1f9e2b"
restore_config = {
"configurable": {
"thread_id": "production-user-456",
"checkpoint_id": checkpoint_id
}
}
実践的な狀態管理パターン
パターン1：自動摘要生成
会話が長期化すると、コンテキストウィンドウの制約やコストが増大します。以下のパターンは、重要な情報を定期的に摘要して狀態を維持します：
class SummarizedState(TypedDict):