# 硬截斷：保留 System + 最近 N 條消息
from langchain_core.messages import SystemMessage, BaseMessage
def hard_truncate(messages: list[BaseMessage], max_messages: int = 20) -> list[BaseMessage]:
    """保留 SystemMessage 不動 + 最近 max_messages 條"""
    system = [m for m in messages if isinstance(m, SystemMessage)]
    others = [m for m in messages if not isinstance(m, SystemMessage)]
    return system + others[-max_messages:]

# 滑動窗口：保留最近 K 輪（user+assistant 算一輪）
def sliding_window(messages: list[BaseMessage], window_size: int = 10) -> list[BaseMessage]:
    """以"輪"為單位保留，user+assistant 算 1 輪"""
    system = [m for m in messages if isinstance(m, SystemMessage)]
    others = [m for m in messages if not isinstance(m, SystemMessage)]
    # 按 2 條消息一輪倒推
    kept = others[-(window_size * 2):] if len(others) > window_size * 2 else others
    return system + kept

# 摘要壓縮：Anchored Iterative Summarization（Anthropic 推薦結(jié)構(gòu)）
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, RemoveMessage
SUMMARY_PROMPT = """請將以下對話壓縮成結(jié)構(gòu)化摘要，每個 section 一行，不要超過 250 字：
- INTENT: 用戶核心目標(biāo)
- DECISIONS: 已做出的關(guān)鍵決定
- FACTS: 重要的實體/數(shù)字/事實
- OPEN: 未解決的問題
對話：
{messages}
"""
def summarize_compress(messages: list[BaseMessage], threshold: int = 12) -> dict:
    if len(messages) <= threshold:
        return {"messages": messages, "summary": ""}
    # 保留 System + 最近 2 輪，其余送 LLM 摘要
    system = [m for m in messages if isinstance(m, SystemMessage)]
    to_compress = [m for m in messages if not isinstance(m, SystemMessage)][:-2]
    recent = [m for m in messages if not isinstance(m, SystemMessage)][-2:]
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    text_blob = "\n".join(f"{m.type}: {m.content[:300]}" for m in to_compress)
    summary = llm.invoke(SUMMARY_PROMPT.format(messages=text_blob)).content
    # 顯式刪除被壓縮的消息
    return {
        "messages": system + recent + [SystemMessage(content=f"歷史摘要：{summary}")]
                              + [RemoveMessage(id=m.id) for m in to_compress if hasattr(m, "id")],
        "summary": summary,
    }

# 關(guān)鍵提?。褐槐Ａ魧嶓w/數(shù)字/決策的骨架
import re
from typing import TypedDict
class ExtractedFacts(TypedDict):
    entities: list[str]      # 人名/地名/產(chǎn)品名
    numbers: list[str]       # 金額/日期/百分比
    decisions: list[str]     # 用戶做出的決定
def extract_key_facts(messages: list[BaseMessage]) -> ExtractedFacts:
    """規(guī)則版：正則抽實體/數(shù)字/LLM 抽決策"""
    text = "\n".join(m.content for m in messages if hasattr(m, "content"))
    # 實體：簡單正則（生產(chǎn)換 GLiNER / 小模型）
    entities = list(set(re.findall(r"[A-Z][a-zA-Z]{2,}", text)))[:20]
    # 數(shù)字：金額/日期
    numbers = re.findall(r"\d{1,3}(?:,\d{3})*(?:\.\d+)?|\d{4}-\d{2}-\d{2}", text)[:20]
    # 決策：用小模型判斷
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    decisions_prompt = (
        f"從下面對話中提取用戶做出的關(guān)鍵決定，每條一行，無決定返回'NONE'：\n{text[:2000]}"
    )
    decisions_text = llm.invoke(decisions_prompt).content
    decisions = [d.strip() for d in decisions_text.split("\n") if d.strip() and d.strip() != "NONE"]
    return {"entities": entities, "numbers": numbers, "decisions": decisions}
# 用法：把 facts 拼到 system prompt 里
def build_prompt_with_facts(messages, facts: ExtractedFacts) -> str:
    fact_str = (
        f"已知實體：{', '.join(facts['entities'][:10])}\n"
        f"關(guān)鍵數(shù)字：{', '.join(facts['numbers'][:10])}\n"
        f"用戶決定：{'; '.join(facts['decisions'][:5])}"
    )
    return fact_str

# 神經(jīng)壓縮：Microsoft LLMLingua-2，逐 token 分類
# pip install llmlingua
from llmlingua import PromptCompressor
def neural_compress(prompt: str, target_ratio: float = 0.4) -> str:
    """target=0.4 表示壓縮到原文 40%"""
    compressor = PromptCompressor(
        model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank",
        device_map="cpu",  # GPU 改 "cuda"
    )
    result = compressor.compress_prompt(
        prompt,
        rate=target_ratio,
        force_context_tokens=["IMPORTANT", "DECISION", "USER_PREF"],
    )
    return result["compressed_prompt"]
# 用法
raw_prompt = "用戶在前 20 輪的對話內(nèi)容..." * 50
compressed = neural_compress(raw_prompt, target_ratio=0.3)
print(f"原 {len(raw_prompt)} → 壓 {len(compressed)} chars")

# 外部化：大塊內(nèi)容寫向量庫，prompt 里只留檢索片段
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
import uuid
class ContextOffloader:
    """把長內(nèi)容外置，prompt 只引用 ref_id"""
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.store = FAISS.from_texts(["__init__"], self.embeddings)
        self.ref_map: dict[str, str] = {"__init__": "__init__"}
    def offload(self, content: str, metadata: dict | None = None) -> str:
        """返回 ref_id，content 寫進(jìn)向量庫"""
        ref_id = f"ctx_{uuid.uuid4().hex[:8]}"
        self.store.add_texts(
            [content],
            metadatas=[{**({"ref_id": ref_id}), **(metadata or {})}],
        )
        self.ref_map[ref_id] = content
        return ref_id
    def retrieve(self, ref_id: str, query: str, k: int = 3) -> str:
        """按 query 檢索回最相關(guān)片段（不是原文）"""
        if ref_id not in self.ref_map:
            return ""
        full = self.ref_map[ref_id]
        # 把引用內(nèi)容加進(jìn)去檢索
        tmp_store = FAISS.from_texts([full], self.embeddings)
        docs = tmp_store.similarity_search(query, k=k)
        return "\n".join(d.page_content for d in docs)
# 用法
offloader = ContextOffloader()
ref = offloader.offload("（此處是 10K 字的會議記錄...）")
# 后續(xù)調(diào)用 LLM 時 prompt 里只寫：[REF:ref_id] 查"用戶決策"
# LLM 觸發(fā)工具調(diào)用時再 retrieve()

你的上下文有多長？
├─ < 4K  → 不用壓縮（裸跑）
├─ 4K-32K → 滑動窗口（最穩(wěn)）
├─ 32K-100K → 摘要壓縮（5 輪一次）
├─ 100K-500K → 神經(jīng)壓縮（LLMLingua）
└─ > 500K → 外部化 + RAG 檢索
你的內(nèi)容能丟嗎？
├─ 閑聊可丟 → 硬截斷 / 滑窗
├─ 決策不能丟 → 關(guān)鍵提取 + 摘要
└─ 合規(guī)不能錯 → 關(guān)鍵提取 + 外部化