マルチターンチャットボットの会話状態管理とコンテキスト圧縮戦略 2026

from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# max_token_limitを超えると古いメッセージを自動要約
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,
    return_messages=True,
    memory_key="chat_history",
    human_prefix="顧客",
    ai_prefix="担当者",
)

# 会話の保存
memory.save_context(
    {"input": "ノートPC購入を検討中で予算は150万ウォンです"},
    {"output": "150万ウォンの予算で良いノートPCをお勧めいたします。主な用途は何ですか？"},
)
memory.save_context(
    {"input": "プログラミングと軽い動画編集をします"},
    {"output": "開発と動画編集用であれば、RAM 16GB以上、SSD 512GB以上のスペックをお勧めします。"},
)
memory.save_context(
    {"input": "MacBook Air M4とLenovo ThinkPadではどちらが良いですか？"},
    {"output": "どちらも優れた製品ですが、用途によって違いがあります。MacBook Air M4は..."},
)

# トークン制限超過時に自動要約 + 直近メッセージ保持
loaded = memory.load_memory_variables({})
print(loaded["chat_history"])
# SystemMessage: "顧客は150万ウォンの予算でプログラミングと動画編集用のノートPCを探しており..."
# + 直近の元のメッセージ

from llama_index.core.memory import ChatSummaryMemoryBuffer
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o", temperature=0)

memory = ChatSummaryMemoryBuffer.from_defaults(
    llm=llm,
    token_limit=3000,
    # 要約トリガーのトークン比率（全体上限の70%超過時）
    summarize_threshold=0.7,
)

# Chat Engineにメモリを接続
from llama_index.core.chat_engine import CondensePlusContextChatEngine

chat_engine = CondensePlusContextChatEngine.from_defaults(
    retriever=index.as_retriever(similarity_top_k=3),
    memory=memory,
    llm=llm,
    system_prompt="あなたは技術サポートの専門家です。",
)

response = chat_engine.chat("先ほど説明したエラーコードについてもっと詳しく教えてください")

import tiktoken


def sliding_window_by_tokens(
    messages: list[dict],
    max_tokens: int = 4000,
    model: str = "gpt-4o",
    always_keep_system: bool = True,
) -> list[dict]:
    """トークン数ベースのスライディングウィンドウ。
    システムメッセージは常に保持し、直近メッセージから逆順に埋める。
    """
    enc = tiktoken.encoding_for_model(model)
    result = []
    current_tokens = 0

    # システムメッセージを優先確保
    system_messages = [m for m in messages if m["role"] == "system"]
    non_system = [m for m in messages if m["role"] != "system"]

    if always_keep_system:
        for sm in system_messages:
            sm_tokens = len(enc.encode(sm["content"]))
            result.append(sm)
            current_tokens += sm_tokens

    # 直近メッセージから逆順に追加
    selected = []
    for msg in reversed(non_system):
        msg_tokens = len(enc.encode(msg["content"]))
        if current_tokens + msg_tokens > max_tokens:
            break
        selected.append(msg)
        current_tokens += msg_tokens

    result.extend(reversed(selected))
    return result

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

PROGRESSIVE_SUMMARY_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """あなたはカスタマーサポート会話の要約専門家です。
既存の要約と新しい会話をマージして更新された要約を生成してください。

ルール:
1. 顧客の核心的な要件、制約条件、好みを必ず維持
2. 確認された事実（名前、注文番号、日付など）を絶対に漏らさない
3. 現在の進行段階と次に必要なアクションを明示
4. 解決済みの問題は簡潔に、未解決の問題は詳細に記録
5. 200文字以内で作成"""),
    ("human", """既存の要約:
{existing_summary}

新しい会話:
{new_messages}

更新された要約:"""),
])


class ProgressiveSummarizer:
    def __init__(self, llm, summary_interval: int = 5):
        self.llm = llm
        self.chain = PROGRESSIVE_SUMMARY_PROMPT | llm
        self.summary = ""
        self.buffer = []
        self.summary_interval = summary_interval
        self.turn_count = 0

    def add_turn(self, user_msg: str, assistant_msg: str):
        self.buffer.append(f"顧客: {user_msg}")
        self.buffer.append(f"担当者: {assistant_msg}")
        self.turn_count += 1

        if self.turn_count % self.summary_interval == 0:
            self._update_summary()

    def _update_summary(self):
        new_messages = "\n".join(self.buffer)
        result = self.chain.invoke({
            "existing_summary": self.summary or "（なし）",
            "new_messages": new_messages,
        })
        self.summary = result.content
        self.buffer = []  # バッファクリア

    def get_context(self) -> str:
        """要約 + 直近バッファを結合したコンテキストを返す"""
        parts = []
        if self.summary:
            parts.append(f"[会話要約]\n{self.summary}")
        if self.buffer:
            parts.append(f"[最近の会話]\n" + "\n".join(self.buffer))
        return "\n\n".join(parts)

from llmlingua import PromptCompressor

# LLMLingua-2の初期化
compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
    use_llmlingua2=True,
    device_map="cpu",  # GPU使用時は"cuda"
)

# 長い会話履歴の圧縮
conversation_history = """
顧客: こんにちは、先週注文した商品について問い合わせます。
担当者: こんにちは！注文番号をお知らせいただければ確認いたします。
顧客: 注文番号はORD-2026-03-1234です。配送がまだ届いていません。
担当者: 確認いたします。少々お待ちください。
担当者: 注文番号ORD-2026-03-1234を確認しました。現在配送中で明日到着予定です。
顧客: 明日ですか？元々昨日届くはずだったのに、なぜ遅れているのですか？
担当者: 物流センターの事情で1日遅延しました。ご不便をおかけして申し訳ございません。
顧客: 送料の返金は可能ですか？
担当者: はい、配送遅延による送料返金が可能です。返金処理を進めてよろしいですか？
顧客: はい、お願いします。
担当者: 送料3,000ウォンの返金処理を完了しました。1-3日以内に元のお支払い方法に返金されます。
"""

compressed = compressor.compress_prompt(
    conversation_history,
    rate=0.5,  # 50%圧縮率
    force_tokens=["注文番号", "ORD-2026-03-1234", "返金"],  # 必ず保持するトークン
)

print(f"元のトークン: {compressed['origin_tokens']}")
print(f"圧縮後のトークン: {compressed['compressed_tokens']}")
print(f"圧縮率: {compressed['ratio']:.1f}x")
print(f"圧縮結果:\n{compressed['compressed_prompt']}")

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage


# 1. 状態スキーマの定義
class OrderSupportState(TypedDict):
    messages: Annotated[list, add_messages]  # リデューサーでメッセージを累積
    order_id: str | None
    issue_type: str | None  # "配送", "返金", "交換", "その他"
    step: str  # "greeting", "identify", "diagnose", "resolve", "close"
    collected_info: dict
    summary: str  # 以前の会話要約


# 2. ノード関数の定義
llm = ChatOpenAI(model="gpt-4o", temperature=0)


def greeting_node(state: OrderSupportState) -> dict:
    """挨拶と初期分類"""
    response = llm.invoke([
        SystemMessage(content="カスタマーサポートチャットボットです。顧客の問い合わせ種類を把握してください。"),
        *state["messages"],
    ])
    return {
        "messages": [response],
        "step": "identify",
    }


def identify_node(state: OrderSupportState) -> dict:
    """注文番号と問題タイプの特定"""
    context_parts = []
    if state.get("summary"):
        context_parts.append(f"以前の会話要約: {state['summary']}")

    system_msg = f"""顧客の注文番号と問題の種類を把握してください。
収集済み情報: {state.get('collected_info', dict())}
{chr(10).join(context_parts)}"""

    response = llm.invoke([
        SystemMessage(content=system_msg),
        *state["messages"][-10:],  # 直近10件のメッセージのみ使用
    ])

    # 応答から注文番号を抽出（実際にはより精緻なパースが必要）
    return {
        "messages": [response],
        "step": "diagnose",
    }


def resolve_node(state: OrderSupportState) -> dict:
    """問題解決の提案"""
    response = llm.invoke([
        SystemMessage(content=f"問題タイプ: {state.get('issue_type', '未確認')}。 "
                              f"注文番号: {state.get('order_id', '未確認')}。 解決策を提示してください。"),
        *state["messages"][-6:],
    ])
    return {
        "messages": [response],
        "step": "close",
    }


# 3. ルーティング関数
def route_by_step(state: OrderSupportState) -> str:
    step = state.get("step", "greeting")
    if step == "greeting":
        return "greeting"
    elif step == "identify":
        return "identify"
    elif step in ("diagnose", "resolve"):
        return "resolve"
    else:
        return END


# 4. グラフの構築
graph = StateGraph(OrderSupportState)
graph.add_node("greeting", greeting_node)
graph.add_node("identify", identify_node)
graph.add_node("resolve", resolve_node)

graph.add_conditional_edges(START, route_by_step)
graph.add_conditional_edges("greeting", route_by_step)
graph.add_conditional_edges("identify", route_by_step)
graph.add_conditional_edges("resolve", route_by_step)

# 5. チェックポインターで状態を永続化
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 6. 実行（thread_idでセッションを区別）
config = {"configurable": {"thread_id": "user-session-abc123"}}
result = app.invoke(
    {
        "messages": [HumanMessage(content="注文した商品がまだ届きません")],
        "step": "greeting",
        "collected_info": {},
        "summary": "",
    },
    config=config,
)

import json
import time
import redis
import tiktoken


class ChatSessionManager:
    """Redisベースのマルチターン会話セッションマネージャー"""

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        session_ttl: int = 3600,  # 1時間
        max_history_tokens: int = 4000,
    ):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.session_ttl = session_ttl
        self.max_history_tokens = max_history_tokens
        self.encoder = tiktoken.encoding_for_model("gpt-4o")

    def _key(self, session_id: str, suffix: str) -> str:
        return f"chat:session:{session_id}:{suffix}"

    def create_session(self, session_id: str, metadata: dict | None = None) -> dict:
        """新しいセッションの作成"""
        session_data = {
            "session_id": session_id,
            "created_at": time.time(),
            "updated_at": time.time(),
            "turn_count": 0,
            "total_tokens": 0,
            "metadata": json.dumps(metadata or {}),
            "summary": "",
        }
        self.redis.hset(self._key(session_id, "meta"), mapping=session_data)
        self.redis.expire(self._key(session_id, "meta"), self.session_ttl)
        return session_data

    def add_message(self, session_id: str, role: str, content: str) -> None:
        """メッセージの追加とトークン管理"""
        msg = json.dumps({
            "role": role,
            "content": content,
            "timestamp": time.time(),
            "tokens": len(self.encoder.encode(content)),
        })
        history_key = self._key(session_id, "history")
        self.redis.rpush(history_key, msg)
        self.redis.expire(history_key, self.session_ttl)

        # メタデータの更新
        self.redis.hincrby(self._key(session_id, "meta"), "turn_count", 1)
        self.redis.hset(
            self._key(session_id, "meta"), "updated_at", str(time.time())
        )
        # TTL更新
        self.redis.expire(self._key(session_id, "meta"), self.session_ttl)

    def get_context_messages(self, session_id: str) -> list[dict]:
        """トークン予算内でコンテキストメッセージを返す"""
        history_key = self._key(session_id, "history")
        all_messages = self.redis.lrange(history_key, 0, -1)

        if not all_messages:
            return []

        parsed = [json.loads(m) for m in all_messages]
        result = []
        token_count = 0

        # 要約があれば先に追加
        summary = self.redis.hget(self._key(session_id, "meta"), "summary")
        if summary:
            summary_tokens = len(self.encoder.encode(summary))
            token_count += summary_tokens
            result.append({"role": "system", "content": f"以前の会話要約: {summary}"})

        # 直近メッセージから逆順にトークン予算内で追加
        selected = []
        for msg in reversed(parsed):
            msg_tokens = msg.get("tokens", 0)
            if token_count + msg_tokens > self.max_history_tokens:
                break
            selected.append({"role": msg["role"], "content": msg["content"]})
            token_count += msg_tokens

        result.extend(reversed(selected))
        return result

    def update_summary(self, session_id: str, summary: str) -> None:
        """会話要約の更新"""
        self.redis.hset(self._key(session_id, "meta"), "summary", summary)

    def delete_session(self, session_id: str) -> None:
        """セッションの削除"""
        for suffix in ("meta", "history"):
            self.redis.delete(self._key(session_id, suffix))

import tiktoken
from dataclasses import dataclass


@dataclass
class TokenBudget:
    """トークン予算配分計算機"""
    model: str = "gpt-4o"
    max_context: int = 128000  # gpt-4o context window
    system_prompt_tokens: int = 500
    response_reserve: int = 4000  # 応答用予約
    rag_budget: int = 3000  # RAGドキュメント用
    tool_result_budget: int = 2000  # ツール実行結果用

    def __post_init__(self):
        self.encoder = tiktoken.encoding_for_model(self.model)

    @property
    def conversation_budget(self) -> int:
        """会話履歴に割り当て可能なトークン数"""
        reserved = (
            self.system_prompt_tokens
            + self.response_reserve
            + self.rag_budget
            + self.tool_result_budget
        )
        return self.max_context - reserved

    def count_tokens(self, text: str) -> int:
        return len(self.encoder.encode(text))

    def allocate(self, messages: list[dict]) -> dict:
        """現在のメッセージに対するトークン使用状況レポート"""
        msg_tokens = sum(
            self.count_tokens(m.get("content", "")) for m in messages
        )
        budget = self.conversation_budget
        return {
            "total_context": self.max_context,
            "system_prompt": self.system_prompt_tokens,
            "response_reserve": self.response_reserve,
            "rag_budget": self.rag_budget,
            "tool_result_budget": self.tool_result_budget,
            "conversation_budget": budget,
            "conversation_used": msg_tokens,
            "conversation_remaining": budget - msg_tokens,
            "utilization_pct": round(msg_tokens / budget * 100, 1),
            "needs_compression": msg_tokens > budget * 0.8,
        }


# 使用例
budget = TokenBudget(model="gpt-4o", max_context=128000)
print(f"会話履歴の利用可能トークン: {budget.conversation_budget:,}")

report = budget.allocate([
    {"role": "user", "content": "以前の注文状況を確認してください"},
    {"role": "assistant", "content": "注文番号を教えていただけますか？"},
])
print(f"使用率: {report['utilization_pct']}%")
print(f"圧縮が必要: {report['needs_compression']}")