멀티턴 챗봇 대화 상태 관리와 컨텍스트 압축 전략 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": "노트북 구매를 고려 중인데 예산이 150만 원이에요"},
    {"output": "150만 원 예산으로 좋은 노트북을 추천드리겠습니다. 주 용도가 어떻게 되시나요?"},
)
memory.save_context(
    {"input": "프로그래밍과 가벼운 영상 편집을 할 거예요"},
    {"output": "개발과 영상 편집용이시라면 RAM 16GB 이상, SSD 512GB 이상의 사양을 추천드립니다."},
)
memory.save_context(
    {"input": "맥북 에어 M4와 레노버 씽크패드 중에 뭐가 나을까요?"},
    {"output": "두 제품 모두 훌륭하지만 용도에 따라 차이가 있습니다. 맥북 에어 M4는 ..."},
)

# 토큰 한계 초과 시 자동 요약 + 최근 메시지 유지
loaded = memory.load_memory_variables({})
print(loaded["chat_history"])
# SystemMessage: "고객은 150만 원 예산으로 프로그래밍과 영상 편집용 노트북을 찾고 있으며..."
# + 최근 원본 메시지들

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 확인했습니다. 현재 배송 중이며 내일 도착 예정입니다.
고객: 내일이요? 원래 어제 도착했어야 하는데 왜 늦어진 건가요?
상담원: 물류센터 사정으로 하루 지연되었습니다. 불편을 드려 죄송합니다.
고객: 배송비 환불은 가능한가요?
상담원: 네, 배송 지연으로 인한 배송비 환불이 가능합니다. 환불 처리를 진행할까요?
고객: 네, 부탁드립니다.
상담원: 배송비 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']}")