Split View: 멀티턴 챗봇 대화 상태 관리와 컨텍스트 압축 전략 2026

from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI

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

# Automatically summarizes old messages when max_token_limit is exceeded
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,
    return_messages=True,
    memory_key="chat_history",
    human_prefix="Customer",
    ai_prefix="Agent",
)

# Save conversations
memory.save_context(
    {"input": "I'm considering buying a laptop and my budget is 1.5 million won"},
    {"output": "I'll recommend a great laptop within your 1.5 million won budget. What will be the primary use?"},
)
memory.save_context(
    {"input": "I'll be doing programming and light video editing"},
    {"output": "For development and video editing, I recommend specs with at least 16GB RAM and 512GB SSD."},
)
memory.save_context(
    {"input": "Which is better, the MacBook Air M4 or Lenovo ThinkPad?"},
    {"output": "Both are excellent products, but there are differences depending on use case. The MacBook Air M4 is ..."},
)

# Automatic summary + recent messages retained when token limit is exceeded
loaded = memory.load_memory_variables({})
print(loaded["chat_history"])
# SystemMessage: "The customer is looking for a laptop for programming and video editing with a 1.5M won budget..."
# + recent original messages

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,
    # Summary trigger token ratio (triggers when exceeding 70% of total limit)
    summarize_threshold=0.7,
)

# Connect memory to 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="You are a technical support specialist.",
)

response = chat_engine.chat("Tell me more about the error code I mentioned earlier")

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]:
    """Token count-based sliding window.
    Always keeps system messages, fills from recent messages in reverse order.
    """
    enc = tiktoken.encoding_for_model(model)
    result = []
    current_tokens = 0

    # Reserve system messages first
    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

    # Add from most recent messages in reverse order
    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", """You are a customer support conversation summarization expert.
Merge the existing summary with the new conversation to produce an updated summary.

Rules:
1. Always preserve the customer's core requirements, constraints, and preferences
2. Never omit confirmed facts (names, order numbers, dates, etc.)
3. Specify the current progress stage and next required action
4. Record resolved issues briefly, unresolved issues in detail
5. Keep within 200 characters"""),
    ("human", """Existing summary:
{existing_summary}

New conversation:
{new_messages}

Updated summary:"""),
])


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"Customer: {user_msg}")
        self.buffer.append(f"Agent: {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 "(none)",
            "new_messages": new_messages,
        })
        self.summary = result.content
        self.buffer = []  # Clear buffer

    def get_context(self) -> str:
        """Return context combining summary + recent buffer"""
        parts = []
        if self.summary:
            parts.append(f"[Conversation Summary]\n{self.summary}")
        if self.buffer:
            parts.append(f"[Recent Conversation]\n" + "\n".join(self.buffer))
        return "\n\n".join(parts)

from llmlingua import PromptCompressor

# Initialize LLMLingua-2
compressor = PromptCompressor(
    model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
    use_llmlingua2=True,
    device_map="cpu",  # Use "cuda" for GPU
)

# Compress long conversation history
conversation_history = """
Customer: Hello, I'd like to inquire about an order I placed last week.
Agent: Hello! Please provide your order number and I'll check for you.
Customer: The order number is ORD-2026-03-1234. My delivery hasn't arrived yet.
Agent: Let me check on that. Please wait a moment.
Agent: I've confirmed order number ORD-2026-03-1234. It's currently in transit and scheduled to arrive tomorrow.
Customer: Tomorrow? It was supposed to arrive yesterday. Why is it delayed?
Agent: It was delayed by one day due to logistics center circumstances. We apologize for the inconvenience.
Customer: Can I get a shipping fee refund?
Agent: Yes, a shipping fee refund is available due to the delivery delay. Shall I proceed with the refund?
Customer: Yes, please.
Agent: The shipping fee refund of 3,000 won has been processed. It will be refunded to the original payment method within 1-3 days.
"""

compressed = compressor.compress_prompt(
    conversation_history,
    rate=0.5,  # 50% compression rate
    force_tokens=["order number", "ORD-2026-03-1234", "refund"],  # Tokens to always keep
)

print(f"Original tokens: {compressed['origin_tokens']}")
print(f"Compressed tokens: {compressed['compressed_tokens']}")
print(f"Compression ratio: {compressed['ratio']:.1f}x")
print(f"Compressed result:\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. State schema definition
class OrderSupportState(TypedDict):
    messages: Annotated[list, add_messages]  # Accumulate messages with reducer
    order_id: str | None
    issue_type: str | None  # "shipping", "refund", "exchange", "other"
    step: str  # "greeting", "identify", "diagnose", "resolve", "close"
    collected_info: dict
    summary: str  # Previous conversation summary


# 2. Node function definitions
llm = ChatOpenAI(model="gpt-4o", temperature=0)


def greeting_node(state: OrderSupportState) -> dict:
    """Greeting and initial classification"""
    response = llm.invoke([
        SystemMessage(content="You are a customer support chatbot. Identify the customer's inquiry type."),
        *state["messages"],
    ])
    return {
        "messages": [response],
        "step": "identify",
    }


def identify_node(state: OrderSupportState) -> dict:
    """Identify order number and issue type"""
    context_parts = []
    if state.get("summary"):
        context_parts.append(f"Previous conversation summary: {state['summary']}")

    system_msg = f"""Identify the customer's order number and problem type.
Collected info: {state.get('collected_info', dict())}
{chr(10).join(context_parts)}"""

    response = llm.invoke([
        SystemMessage(content=system_msg),
        *state["messages"][-10:],  # Use only the last 10 messages
    ])

    # Extract order number from response (more sophisticated parsing needed in practice)
    return {
        "messages": [response],
        "step": "diagnose",
    }


def resolve_node(state: OrderSupportState) -> dict:
    """Propose problem resolution"""
    response = llm.invoke([
        SystemMessage(content=f"Issue type: {state.get('issue_type', 'unidentified')}. "
                              f"Order number: {state.get('order_id', 'unidentified')}. Propose a solution."),
        *state["messages"][-6:],
    ])
    return {
        "messages": [response],
        "step": "close",
    }


# 3. Routing function
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 construction
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. Persist state with checkpointer
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# 6. Execute (sessions distinguished by thread_id)
config = {"configurable": {"thread_id": "user-session-abc123"}}
result = app.invoke(
    {
        "messages": [HumanMessage(content="My ordered item hasn't arrived yet")],
        "step": "greeting",
        "collected_info": {},
        "summary": "",
    },
    config=config,
)

import json
import time
import redis
import tiktoken


class ChatSessionManager:
    """Redis-based multi-turn conversation session manager"""

    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        session_ttl: int = 3600,  # 1 hour
        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:
        """Create a new session"""
        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:
        """Add a message and manage tokens"""
        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)

        # Update metadata
        self.redis.hincrby(self._key(session_id, "meta"), "turn_count", 1)
        self.redis.hset(
            self._key(session_id, "meta"), "updated_at", str(time.time())
        )
        # Renew TTL
        self.redis.expire(self._key(session_id, "meta"), self.session_ttl)

    def get_context_messages(self, session_id: str) -> list[dict]:
        """Return context messages within the token budget"""
        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

        # Add summary first if available
        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"Previous conversation summary: {summary}"})

        # Add recent messages in reverse order within token budget
        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:
        """Update conversation summary"""
        self.redis.hset(self._key(session_id, "meta"), "summary", summary)

    def delete_session(self, session_id: str) -> None:
        """Delete session"""
        for suffix in ("meta", "history"):
            self.redis.delete(self._key(session_id, suffix))

import tiktoken
from dataclasses import dataclass


@dataclass
class TokenBudget:
    """Token budget allocation calculator"""
    model: str = "gpt-4o"
    max_context: int = 128000  # gpt-4o context window
    system_prompt_tokens: int = 500
    response_reserve: int = 4000  # Reserved for response
    rag_budget: int = 3000  # For RAG documents
    tool_result_budget: int = 2000  # For tool execution results

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

    @property
    def conversation_budget(self) -> int:
        """Number of tokens available for conversation history"""
        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:
        """Token usage status report for current messages"""
        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,
        }


# Usage example
budget = TokenBudget(model="gpt-4o", max_context=128000)
print(f"Available tokens for conversation history: {budget.conversation_budget:,}")

report = budget.allocate([
    {"role": "user", "content": "Please check my previous order status"},
    {"role": "assistant", "content": "Could you provide your order number?"},
])
print(f"Utilization: {report['utilization_pct']}%")
print(f"Compression needed: {report['needs_compression']}")