多 Agent Token 優化系列 pt.3：Context 壓縮與摘要 — 打造可無限對話的 AI 系統

  1import anthropic
  2from dataclasses import dataclass, field
  3from typing import Optional
  4
  5client = anthropic.Anthropic()
  6
  7@dataclass
  8class Message:
  9    role: str  # "user" or "assistant"
 10    content: str
 11
 12    def to_dict(self) -> dict:
 13        return {"role": self.role, "content": self.content}
 14
 15@dataclass
 16class SlidingWindowConfig:
 17    """滑動視窗配置"""
 18    max_turns: int = 10          # 最多保留幾輪對話
 19    max_tokens: int = 8000       # 最大 tokens 數（估算）
 20    tokens_per_char: float = 0.4 # 中文約 0.5，英文約 0.25
 21
 22
 23class SlidingWindowManager:
 24    """
 25    滑動視窗 Context 管理器
 26
 27    策略：只保留最近 N 輪對話，超過則丟棄最舊的
 28
 29    優點：實作簡單，延遲可預測
 30    缺點：完全丟失早期對話資訊
 31    """
 32
 33    def __init__(self, config: Optional[SlidingWindowConfig] = None):
 34        self.config = config or SlidingWindowConfig()
 35        self.messages: list[Message] = []
 36        self._stats = {
 37            "total_messages": 0,
 38            "dropped_messages": 0
 39        }
 40
 41    def _estimate_tokens(self, text: str) -> int:
 42        """估算文字的 token 數量"""
 43        return int(len(text) * self.config.tokens_per_char)
 44
 45    def _get_total_tokens(self) -> int:
 46        """計算當前所有訊息的 tokens"""
 47        return sum(self._estimate_tokens(m.content) for m in self.messages)
 48
 49    def add_message(self, role: str, content: str):
 50        """添加新訊息"""
 51        self.messages.append(Message(role=role, content=content))
 52        self._stats["total_messages"] += 1
 53        self._trim_if_needed()
 54
 55    def _trim_if_needed(self):
 56        """如果超過限制，移除最舊的訊息"""
 57        # 按輪數限制（一輪 = user + assistant）
 58        while len(self.messages) > self.config.max_turns * 2:
 59            self.messages.pop(0)
 60            self._stats["dropped_messages"] += 1
 61
 62        # 按 token 數限制
 63        while self._get_total_tokens() > self.config.max_tokens and len(self.messages) > 2:
 64            self.messages.pop(0)
 65            self._stats["dropped_messages"] += 1
 66
 67    def get_messages(self) -> list[dict]:
 68        """取得格式化的訊息列表"""
 69        return [m.to_dict() for m in self.messages]
 70
 71    def get_stats(self) -> dict:
 72        return {
 73            **self._stats,
 74            "current_messages": len(self.messages),
 75            "current_tokens": self._get_total_tokens()
 76        }
 77
 78
 79class SlidingWindowChatbot:
 80    """使用滑動視窗的聊天機器人"""
 81
 82    def __init__(
 83        self,
 84        system_prompt: str,
 85        window_config: Optional[SlidingWindowConfig] = None
 86    ):
 87        self.system_prompt = system_prompt
 88        self.window = SlidingWindowManager(window_config)
 89
 90    def chat(self, user_input: str) -> str:
 91        # 添加使用者訊息
 92        self.window.add_message("user", user_input)
 93
 94        # 呼叫 API
 95        response = client.messages.create(
 96            model="claude-sonnet-4-6",
 97            max_tokens=2048,
 98            system=self.system_prompt,
 99            messages=self.window.get_messages()
100        )
101
102        assistant_response = response.content[0].text
103
104        # 添加助手回應
105        self.window.add_message("assistant", assistant_response)
106
107        return assistant_response
108
109
110# 使用範例
111if __name__ == "__main__":
112    bot = SlidingWindowChatbot(
113        system_prompt="你是一個友善的助手。",
114        window_config=SlidingWindowConfig(max_turns=5, max_tokens=4000)
115    )
116
117    # 模擬多輪對話
118    conversations = [
119        "你好，我叫小明",
120        "我喜歡程式設計",
121        "特別是 Python",
122        "我正在學習機器學習",
123        "你能推薦一些資源嗎？",
124        "謝謝你的建議",
125        "對了，你還記得我叫什麼名字嗎？"  # 測試是否記得早期資訊
126    ]
127
128    for user_msg in conversations:
129        print(f"\n👤 User: {user_msg}")
130        response = bot.chat(user_msg)
131        print(f"🤖 Bot: {response[:200]}...")
132        print(f"   [視窗狀態: {bot.window.get_stats()}]")

  1import anthropic
  2from dataclasses import dataclass, field
  3from typing import Optional
  4from enum import Enum
  5
  6client = anthropic.Anthropic()
  7
  8
  9class SummarizationModel(Enum):
 10    """摘要使用的模型"""
 11    HAIKU = "claude-haiku-4-5-20251001"    # 便宜快速
 12    SONNET = "claude-sonnet-4-6"           # 平衡
 13
 14
 15@dataclass
 16class ProgressiveSummaryConfig:
 17    """漸進式摘要配置"""
 18    # 視窗設定
 19    recent_turns_to_keep: int = 5       # 保留最近幾輪完整對話
 20
 21    # 摘要觸發條件
 22    summarize_threshold: int = 10        # 超過幾輪時觸發摘要
 23
 24    # 摘要設定
 25    summary_model: SummarizationModel = SummarizationModel.HAIKU
 26    max_summary_tokens: int = 1000       # 摘要最大長度
 27
 28    # 摘要內容指引
 29    summary_focus: list[str] = field(default_factory=lambda: [
 30        "使用者身份和背景",
 31        "主要任務或目標",
 32        "關鍵決策和結論",
 33        "重要的技術細節",
 34        "待解決的問題",
 35        "使用者偏好"
 36    ])
 37
 38
 39@dataclass
 40class ConversationState:
 41    """對話狀態"""
 42    summary: Optional[str] = None           # 歷史摘要
 43    recent_messages: list[dict] = field(default_factory=list)  # 最近的完整訊息
 44    total_turns: int = 0                    # 總對話輪數
 45    last_summary_turn: int = 0              # 上次摘要的輪數
 46
 47
 48class ProgressiveSummarizer:
 49    """
 50    漸進式摘要管理器
 51
 52    核心邏輯：
 53    1. 保留最近 N 輪的完整對話
 54    2. 超過閾值時，將舊對話壓縮為摘要
 55    3. 摘要會隨著對話推進而更新（包含新資訊）
 56    """
 57
 58    def __init__(self, config: Optional[ProgressiveSummaryConfig] = None):
 59        self.config = config or ProgressiveSummaryConfig()
 60        self.state = ConversationState()
 61        self._stats = {
 62            "summarization_count": 0,
 63            "tokens_before_summary": 0,
 64            "tokens_after_summary": 0
 65        }
 66
 67    def _format_messages_for_summary(self, messages: list[dict]) -> str:
 68        """將訊息格式化為摘要輸入"""
 69        formatted = []
 70        for i, msg in enumerate(messages):
 71            role = "使用者" if msg["role"] == "user" else "助手"
 72            formatted.append(f"[{role}]: {msg['content']}")
 73        return "\n\n".join(formatted)
 74
 75    def _create_summary_prompt(self, messages_text: str, existing_summary: Optional[str]) -> str:
 76        """建立摘要提示"""
 77        focus_points = "\n".join(f"- {f}" for f in self.config.summary_focus)
 78
 79        if existing_summary:
 80            return f"""請更新以下對話摘要，整合新的對話內容。
 81
 82現有摘要：
 83{existing_summary}
 84
 85新增對話：
 86{messages_text}
 87
 88請產生更新後的摘要，重點保留：
 89{focus_points}
 90
 91要求：
 921. 保持簡潔，控制在 {self.config.max_summary_tokens} tokens 以內
 932. 使用條列式整理關鍵資訊
 943. 標註重要的變化或新發現
 954. 移除已解決或不再相關的資訊
 96
 97更新後的摘要："""
 98        else:
 99            return f"""請為以下對話產生摘要。
100
101對話內容：
102{messages_text}
103
104請產生摘要，重點保留：
105{focus_points}
106
107要求：
1081. 保持簡潔，控制在 {self.config.max_summary_tokens} tokens 以內
1092. 使用條列式整理關鍵資訊
1103. 突出關鍵決策和結論
111
112摘要："""
113
114    def _generate_summary(self, messages: list[dict]) -> str:
115        """使用 LLM 生成摘要"""
116        messages_text = self._format_messages_for_summary(messages)
117        prompt = self._create_summary_prompt(messages_text, self.state.summary)
118
119        response = client.messages.create(
120            model=self.config.summary_model.value,
121            max_tokens=self.config.max_summary_tokens,
122            messages=[{"role": "user", "content": prompt}]
123        )
124
125        self._stats["summarization_count"] += 1
126        return response.content[0].text
127
128    def add_turn(self, user_message: str, assistant_message: str):
129        """添加一輪對話"""
130        self.state.recent_messages.append({"role": "user", "content": user_message})
131        self.state.recent_messages.append({"role": "assistant", "content": assistant_message})
132        self.state.total_turns += 1
133
134        # 檢查是否需要摘要
135        self._check_and_summarize()
136
137    def _check_and_summarize(self):
138        """檢查並執行摘要（如果需要）"""
139        turns_since_summary = self.state.total_turns - self.state.last_summary_turn
140
141        if turns_since_summary >= self.config.summarize_threshold:
142            self._perform_summarization()
143
144    def _perform_summarization(self):
145        """執行摘要"""
146        # 計算要摘要的訊息數量
147        messages_to_summarize_count = len(self.state.recent_messages) - (self.config.recent_turns_to_keep * 2)
148
149        if messages_to_summarize_count <= 0:
150            return
151
152        # 分離要摘要的訊息和要保留的訊息
153        messages_to_summarize = self.state.recent_messages[:messages_to_summarize_count]
154        messages_to_keep = self.state.recent_messages[messages_to_summarize_count:]
155
156        # 記錄壓縮前的 tokens
157        before_tokens = sum(len(m["content"]) for m in self.state.recent_messages) * 0.4
158        self._stats["tokens_before_summary"] += before_tokens
159
160        # 生成摘要
161        print(f"📝 正在壓縮 {len(messages_to_summarize)} 條訊息...")
162        self.state.summary = self._generate_summary(messages_to_summarize)
163
164        # 更新狀態
165        self.state.recent_messages = messages_to_keep
166        self.state.last_summary_turn = self.state.total_turns
167
168        # 記錄壓縮後的 tokens
169        after_tokens = (len(self.state.summary) + sum(len(m["content"]) for m in messages_to_keep)) * 0.4
170        self._stats["tokens_after_summary"] += after_tokens
171
172        print(f"✅ 壓縮完成：{int(before_tokens)} → {int(after_tokens)} tokens")
173
174    def get_context_messages(self) -> list[dict]:
175        """取得要發送給 API 的訊息"""
176        messages = []
177
178        # 如果有摘要，作為第一條訊息注入
179        if self.state.summary:
180            messages.append({
181                "role": "user",
182                "content": f"[對話歷史摘要]\n{self.state.summary}\n\n請基於以上背景繼續對話。"
183            })
184            messages.append({
185                "role": "assistant",
186                "content": "我已了解之前的對話背景，請繼續。"
187            })
188
189        # 加入最近的完整訊息
190        messages.extend(self.state.recent_messages)
191
192        return messages
193
194    def get_stats(self) -> dict:
195        """取得統計資訊"""
196        compression_ratio = 0
197        if self._stats["tokens_before_summary"] > 0:
198            compression_ratio = (
199                1 - self._stats["tokens_after_summary"] / self._stats["tokens_before_summary"]
200            ) * 100
201
202        return {
203            **self._stats,
204            "total_turns": self.state.total_turns,
205            "has_summary": self.state.summary is not None,
206            "recent_messages_count": len(self.state.recent_messages),
207            "compression_ratio": f"{compression_ratio:.1f}%"
208        }
209
210
211class ProgressiveSummaryChatbot:
212    """使用漸進式摘要的聊天機器人"""
213
214    def __init__(
215        self,
216        system_prompt: str,
217        summary_config: Optional[ProgressiveSummaryConfig] = None
218    ):
219        self.system_prompt = system_prompt
220        self.summarizer = ProgressiveSummarizer(summary_config)
221
222    def chat(self, user_input: str) -> str:
223        # 取得當前 context
224        messages = self.summarizer.get_context_messages()
225        messages.append({"role": "user", "content": user_input})
226
227        # 呼叫 API
228        response = client.messages.create(
229            model="claude-sonnet-4-6",
230            max_tokens=2048,
231            system=self.system_prompt,
232            messages=messages
233        )
234
235        assistant_response = response.content[0].text
236
237        # 更新摘要器
238        self.summarizer.add_turn(user_input, assistant_response)
239
240        return assistant_response
241
242    def get_current_summary(self) -> Optional[str]:
243        """取得當前摘要"""
244        return self.summarizer.state.summary
245
246
247# 使用範例
248if __name__ == "__main__":
249    config = ProgressiveSummaryConfig(
250        recent_turns_to_keep=3,
251        summarize_threshold=5,
252        summary_focus=[
253            "使用者的身份和角色",
254            "討論的主題和目標",
255            "已做出的決定",
256            "關鍵的技術細節"
257        ]
258    )
259
260    bot = ProgressiveSummaryChatbot(
261        system_prompt="你是一個專業的技術顧問，幫助使用者解決程式問題。",
262        summary_config=config
263    )
264
265    # 模擬長對話
266    conversations = [
267        "你好，我是小明，是一個後端工程師",
268        "我在開發一個電商平台",
269        "使用 Python 和 FastAPI 框架",
270        "資料庫是 PostgreSQL",
271        "現在遇到一個效能問題",
272        "商品列表的查詢很慢",
273        "大概要 5 秒才能返回",
274        "我已經加了索引",
275        "但還是很慢",
276        "你覺得還有什麼優化方向？",
277        "對了，你還記得我用的是什麼框架嗎？"  # 測試摘要是否保留資訊
278    ]
279
280    for i, user_msg in enumerate(conversations, 1):
281        print(f"\n{'='*50}")
282        print(f"Turn {i}")
283        print(f"👤 User: {user_msg}")
284        response = bot.chat(user_msg)
285        print(f"🤖 Bot: {response[:300]}...")
286
287        stats = bot.summarizer.get_stats()
288        print(f"\n📊 Stats: {stats}")
289
290        if bot.get_current_summary():
291            print(f"\n📋 Current Summary:\n{bot.get_current_summary()[:500]}...")

  1import anthropic
  2import json
  3from dataclasses import dataclass, field
  4from typing import Optional
  5from datetime import datetime
  6from pathlib import Path
  7
  8client = anthropic.Anthropic()
  9
 10
 11@dataclass
 12class HierarchicalSummaryConfig:
 13    """階層式摘要配置"""
 14    # Level 1: 最近完整對話
 15    recent_turns: int = 5
 16
 17    # Level 2: Session 摘要
 18    session_summary_threshold: int = 10  # 幾輪觸發 Session 摘要更新
 19    session_summary_max_tokens: int = 800
 20
 21    # Level 3: 長期記憶
 22    long_term_memory_file: Optional[str] = None  # 持久化檔案路徑
 23    long_term_summary_max_tokens: int = 300
 24
 25    # 摘要模型
 26    summary_model: str = "claude-haiku-4-5-20251001"
 27
 28
 29@dataclass
 30class UserProfile:
 31    """使用者輪廓（長期記憶）"""
 32    name: Optional[str] = None
 33    role: Optional[str] = None
 34    preferences: list[str] = field(default_factory=list)
 35    expertise: list[str] = field(default_factory=list)
 36    ongoing_projects: list[str] = field(default_factory=list)
 37    interaction_style: Optional[str] = None
 38    last_updated: Optional[str] = None
 39
 40    def to_summary(self) -> str:
 41        """轉換為摘要文字"""
 42        parts = []
 43        if self.name:
 44            parts.append(f"使用者：{self.name}")
 45        if self.role:
 46            parts.append(f"角色：{self.role}")
 47        if self.preferences:
 48            parts.append(f"偏好：{', '.join(self.preferences)}")
 49        if self.expertise:
 50            parts.append(f"專長：{', '.join(self.expertise)}")
 51        if self.ongoing_projects:
 52            parts.append(f"進行中專案：{', '.join(self.ongoing_projects)}")
 53        if self.interaction_style:
 54            parts.append(f"互動風格：{self.interaction_style}")
 55        return "\n".join(parts) if parts else "（尚無使用者資料）"
 56
 57    def to_dict(self) -> dict:
 58        return {
 59            "name": self.name,
 60            "role": self.role,
 61            "preferences": self.preferences,
 62            "expertise": self.expertise,
 63            "ongoing_projects": self.ongoing_projects,
 64            "interaction_style": self.interaction_style,
 65            "last_updated": self.last_updated
 66        }
 67
 68    @classmethod
 69    def from_dict(cls, data: dict) -> "UserProfile":
 70        return cls(**data)
 71
 72
 73@dataclass
 74class SessionState:
 75    """Session 狀態"""
 76    session_id: str
 77    started_at: str
 78    topic: Optional[str] = None
 79    summary: Optional[str] = None
 80    key_decisions: list[str] = field(default_factory=list)
 81    pending_issues: list[str] = field(default_factory=list)
 82    recent_messages: list[dict] = field(default_factory=list)
 83    turn_count: int = 0
 84
 85
 86class HierarchicalMemoryManager:
 87    """
 88    階層式記憶管理器
 89
 90    三層結構：
 91    - Level 3 (Long-term): 使用者輪廓，跨 Session 持久化
 92    - Level 2 (Session): 當前 Session 的摘要
 93    - Level 1 (Recent): 最近幾輪的完整對話
 94    """
 95
 96    def __init__(self, config: Optional[HierarchicalSummaryConfig] = None):
 97        self.config = config or HierarchicalSummaryConfig()
 98
 99        # Level 3: 長期記憶
100        self.user_profile = self._load_user_profile()
101
102        # Level 2: Session 狀態
103        self.session = SessionState(
104            session_id=datetime.now().strftime("%Y%m%d_%H%M%S"),
105            started_at=datetime.now().isoformat()
106        )
107
108        self._stats = {
109            "session_summaries": 0,
110            "profile_updates": 0
111        }
112
113    def _load_user_profile(self) -> UserProfile:
114        """載入持久化的使用者輪廓"""
115        if self.config.long_term_memory_file:
116            path = Path(self.config.long_term_memory_file)
117            if path.exists():
118                with open(path, "r", encoding="utf-8") as f:
119                    data = json.load(f)
120                    return UserProfile.from_dict(data)
121        return UserProfile()
122
123    def _save_user_profile(self):
124        """儲存使用者輪廓"""
125        if self.config.long_term_memory_file:
126            path = Path(self.config.long_term_memory_file)
127            path.parent.mkdir(parents=True, exist_ok=True)
128            with open(path, "w", encoding="utf-8") as f:
129                json.dump(self.user_profile.to_dict(), f, ensure_ascii=False, indent=2)
130
131    def _update_user_profile(self, messages: list[dict]):
132        """從對話中更新使用者輪廓"""
133        if not messages:
134            return
135
136        messages_text = "\n".join(
137            f"[{'使用者' if m['role'] == 'user' else '助手'}]: {m['content']}"
138            for m in messages
139        )
140
141        current_profile = self.user_profile.to_summary()
142
143        prompt = f"""根據以下對話，更新使用者輪廓。只提取明確提到的資訊，不要推測。
144
145現有輪廓：
146{current_profile}
147
148最近對話：
149{messages_text}
150
151請以 JSON 格式返回更新後的輪廓，只包含有變化的欄位：
152{{
153    "name": "使用者名稱（如果提到）",
154    "role": "職業或角色（如果提到）",
155    "preferences": ["偏好1", "偏好2"],
156    "expertise": ["專長1", "專長2"],
157    "ongoing_projects": ["專案1", "專案2"],
158    "interaction_style": "互動風格描述"
159}}
160
161如果沒有新資訊，返回空物件 {{}}"""
162
163        response = client.messages.create(
164            model=self.config.summary_model,
165            max_tokens=500,
166            messages=[{"role": "user", "content": prompt}]
167        )
168
169        try:
170            # 嘗試解析 JSON
171            response_text = response.content[0].text
172            # 處理可能的 markdown 程式碼區塊
173            if "```json" in response_text:
174                response_text = response_text.split("```json")[1].split("```")[0]
175            elif "```" in response_text:
176                response_text = response_text.split("```")[1].split("```")[0]
177
178            updates = json.loads(response_text.strip())
179
180            if updates:
181                # 更新輪廓
182                for key, value in updates.items():
183                    if value and hasattr(self.user_profile, key):
184                        if isinstance(value, list):
185                            # 合併列表，去重
186                            existing = getattr(self.user_profile, key) or []
187                            setattr(self.user_profile, key, list(set(existing + value)))
188                        else:
189                            setattr(self.user_profile, key, value)
190
191                self.user_profile.last_updated = datetime.now().isoformat()
192                self._save_user_profile()
193                self._stats["profile_updates"] += 1
194                print(f"✅ 已更新使用者輪廓")
195
196        except (json.JSONDecodeError, IndexError) as e:
197            print(f"⚠️ 輪廓更新解析失敗: {e}")
198
199    def _update_session_summary(self):
200        """更新 Session 摘要"""
201        if not self.session.recent_messages:
202            return
203
204        messages_text = "\n".join(
205            f"[{'使用者' if m['role'] == 'user' else '助手'}]: {m['content']}"
206            for m in self.session.recent_messages
207        )
208
209        existing_summary = self.session.summary or "（這是新對話的開始）"
210
211        prompt = f"""請更新對話 Session 摘要。
212
213現有摘要：
214{existing_summary}
215
216新增對話：
217{messages_text}
218
219請產生更新後的摘要，包含：
2201. 主要討論主題
2212. 關鍵決策和結論
2223. 待解決的問題
2234. 重要的技術細節
224
225控制在 {self.config.session_summary_max_tokens} tokens 以內。
226
227更新後的摘要："""
228
229        response = client.messages.create(
230            model=self.config.summary_model,
231            max_tokens=self.config.session_summary_max_tokens,
232            messages=[{"role": "user", "content": prompt}]
233        )
234
235        self.session.summary = response.content[0].text
236        self._stats["session_summaries"] += 1
237        print(f"✅ 已更新 Session 摘要")
238
239    def add_turn(self, user_message: str, assistant_message: str):
240        """添加一輪對話"""
241        self.session.recent_messages.append({"role": "user", "content": user_message})
242        self.session.recent_messages.append({"role": "assistant", "content": assistant_message})
243        self.session.turn_count += 1
244
245        # 檢查是否需要更新
246        if self.session.turn_count % self.config.session_summary_threshold == 0:
247            # 更新 Session 摘要
248            old_messages = self.session.recent_messages[:-self.config.recent_turns * 2]
249            if old_messages:
250                self._update_session_summary()
251                self._update_user_profile(old_messages)
252                # 只保留最近的訊息
253                self.session.recent_messages = self.session.recent_messages[-self.config.recent_turns * 2:]
254
255    def get_context(self) -> str:
256        """
257        取得階層式 Context
258
259        返回格式：
260        [長期記憶] + [Session 摘要] + [最近對話]
261        """
262        context_parts = []
263
264        # Level 3: 長期記憶
265        profile_summary = self.user_profile.to_summary()
266        if profile_summary != "（尚無使用者資料）":
267            context_parts.append(f"[使用者背景]\n{profile_summary}")
268
269        # Level 2: Session 摘要
270        if self.session.summary:
271            context_parts.append(f"[本次對話摘要]\n{self.session.summary}")
272
273        return "\n\n".join(context_parts)
274
275    def get_messages(self) -> list[dict]:
276        """取得要發送給 API 的訊息"""
277        messages = []
278
279        context = self.get_context()
280        if context:
281            messages.append({
282                "role": "user",
283                "content": f"{context}\n\n請基於以上背景繼續對話。"
284            })
285            messages.append({
286                "role": "assistant",
287                "content": "我已了解背景資訊，請繼續。"
288            })
289
290        messages.extend(self.session.recent_messages)
291        return messages
292
293    def get_stats(self) -> dict:
294        return {
295            **self._stats,
296            "session_turn_count": self.session.turn_count,
297            "recent_messages": len(self.session.recent_messages),
298            "has_session_summary": self.session.summary is not None,
299            "has_user_profile": self.user_profile.name is not None
300        }
301
302
303class HierarchicalMemoryChatbot:
304    """使用階層式記憶的聊天機器人"""
305
306    def __init__(
307        self,
308        system_prompt: str,
309        config: Optional[HierarchicalSummaryConfig] = None
310    ):
311        self.system_prompt = system_prompt
312        self.memory = HierarchicalMemoryManager(config)
313
314    def chat(self, user_input: str) -> str:
315        messages = self.memory.get_messages()
316        messages.append({"role": "user", "content": user_input})
317
318        response = client.messages.create(
319            model="claude-sonnet-4-6",
320            max_tokens=2048,
321            system=self.system_prompt,
322            messages=messages
323        )
324
325        assistant_response = response.content[0].text
326        self.memory.add_turn(user_input, assistant_response)
327
328        return assistant_response
329
330    def get_user_profile(self) -> UserProfile:
331        return self.memory.user_profile
332
333    def get_session_summary(self) -> Optional[str]:
334        return self.memory.session.summary
335
336
337# 使用範例
338if __name__ == "__main__":
339    config = HierarchicalSummaryConfig(
340        recent_turns=3,
341        session_summary_threshold=5,
342        long_term_memory_file="./user_memory.json"
343    )
344
345    bot = HierarchicalMemoryChatbot(
346        system_prompt="你是一個專業的技術顧問。",
347        config=config
348    )
349
350    # 模擬多輪對話
351    test_conversations = [
352        "你好，我是小明",
353        "我是一個 Python 開發者",
354        "主要做後端開發",
355        "最近在學 Rust",
356        "覺得 Rust 的所有權系統很有趣",
357        "想把它用在一個高效能的服務上",
358        # ... 更多對話
359    ]
360
361    for msg in test_conversations:
362        print(f"\n👤 User: {msg}")
363        response = bot.chat(msg)
364        print(f"🤖 Bot: {response[:200]}...")
365
366    print("\n" + "="*50)
367    print("使用者輪廓：")
368    print(bot.get_user_profile().to_summary())
369
370    print("\nSession 摘要：")
371    print(bot.get_session_summary() or "（尚未產生）")

  1import anthropic
  2import json
  3from dataclasses import dataclass, field
  4from typing import Optional, Any
  5from enum import Enum
  6
  7client = anthropic.Anthropic()
  8
  9
 10class ImportanceLevel(Enum):
 11    CRITICAL = "critical"    # 必須保留
 12    HIGH = "high"           # 高度重要
 13    MEDIUM = "medium"       # 中等重要
 14    LOW = "low"             # 可以捨棄
 15
 16
 17@dataclass
 18class SemanticUnit:
 19    """語意單元"""
 20    content: str
 21    unit_type: str  # "fact", "decision", "question", "preference", "context"
 22    importance: ImportanceLevel
 23    turn_created: int
 24    related_entities: list[str] = field(default_factory=list)
 25
 26    def to_dict(self) -> dict:
 27        return {
 28            "content": self.content,
 29            "type": self.unit_type,
 30            "importance": self.importance.value,
 31            "turn": self.turn_created,
 32            "entities": self.related_entities
 33        }
 34
 35
 36@dataclass
 37class SemanticCompressionConfig:
 38    """語意壓縮配置"""
 39    # 提取設定
 40    extraction_model: str = "claude-haiku-4-5-20251001"
 41
 42    # 保留策略
 43    max_semantic_units: int = 50           # 最多保留幾個語意單元
 44    max_context_tokens: int = 3000         # 重建 context 的最大 tokens
 45
 46    # 重要性衰減
 47    importance_decay_rate: float = 0.1     # 每輪重要性衰減率
 48    min_importance_to_keep: float = 0.3    # 低於此值的單元會被移除
 49
 50    # 最近對話
 51    recent_turns: int = 3
 52
 53
 54class SemanticMemoryStore:
 55    """語意記憶儲存"""
 56
 57    def __init__(self, config: SemanticCompressionConfig):
 58        self.config = config
 59        self.units: list[SemanticUnit] = []
 60        self.entity_index: dict[str, list[int]] = {}  # entity -> unit indices
 61        self.current_turn: int = 0
 62
 63    def add_units(self, units: list[SemanticUnit]):
 64        """添加語意單元"""
 65        for unit in units:
 66            idx = len(self.units)
 67            self.units.append(unit)
 68
 69            # 更新實體索引
 70            for entity in unit.related_entities:
 71                if entity not in self.entity_index:
 72                    self.entity_index[entity] = []
 73                self.entity_index[entity].append(idx)
 74
 75    def decay_importance(self):
 76        """衰減舊單元的重要性"""
 77        for unit in self.units:
 78            age = self.current_turn - unit.turn_created
 79            if age > 0:
 80                # 根據年齡降低重要性
 81                decay = 1.0 - (self.config.importance_decay_rate * age)
 82                decay = max(decay, 0.1)  # 最低保留 10%
 83
 84                # 更新重要性等級
 85                if unit.importance == ImportanceLevel.CRITICAL:
 86                    pass  # CRITICAL 不衰減
 87                elif decay < 0.3:
 88                    unit.importance = ImportanceLevel.LOW
 89                elif decay < 0.6:
 90                    unit.importance = ImportanceLevel.MEDIUM
 91
 92    def prune(self):
 93        """移除低重要性的單元"""
 94        # 按重要性排序，保留最重要的
 95        importance_order = {
 96            ImportanceLevel.CRITICAL: 4,
 97            ImportanceLevel.HIGH: 3,
 98            ImportanceLevel.MEDIUM: 2,
 99            ImportanceLevel.LOW: 1
100        }
101
102        sorted_units = sorted(
103            self.units,
104            key=lambda u: (importance_order[u.importance], -u.turn_created),
105            reverse=True
106        )
107
108        # 保留前 N 個
109        self.units = sorted_units[:self.config.max_semantic_units]
110
111        # 重建實體索引
112        self.entity_index = {}
113        for idx, unit in enumerate(self.units):
114            for entity in unit.related_entities:
115                if entity not in self.entity_index:
116                    self.entity_index[entity] = []
117                self.entity_index[entity].append(idx)
118
119    def get_relevant_units(self, query: str, entities: list[str]) -> list[SemanticUnit]:
120        """取得與查詢相關的語意單元"""
121        relevant_indices = set()
122
123        # 根據實體匹配
124        for entity in entities:
125            if entity in self.entity_index:
126                relevant_indices.update(self.entity_index[entity])
127
128        # 取得相關單元
129        relevant_units = [self.units[i] for i in relevant_indices]
130
131        # 加入所有 CRITICAL 單元
132        for unit in self.units:
133            if unit.importance == ImportanceLevel.CRITICAL and unit not in relevant_units:
134                relevant_units.append(unit)
135
136        return relevant_units
137
138    def to_context_string(self, units: Optional[list[SemanticUnit]] = None) -> str:
139        """將語意單元重建為 context 字串"""
140        units = units or self.units
141
142        if not units:
143            return ""
144
145        # 按類型分組
146        grouped = {
147            "fact": [],
148            "decision": [],
149            "question": [],
150            "preference": [],
151            "context": []
152        }
153
154        for unit in units:
155            if unit.unit_type in grouped:
156                grouped[unit.unit_type].append(unit.content)
157
158        # 格式化輸出
159        parts = []
160
161        if grouped["context"]:
162            parts.append("背景資訊：\n" + "\n".join(f"- {c}" for c in grouped["context"]))
163
164        if grouped["fact"]:
165            parts.append("已確認的事實：\n" + "\n".join(f"- {f}" for f in grouped["fact"]))
166
167        if grouped["decision"]:
168            parts.append("已做的決定：\n" + "\n".join(f"- {d}" for d in grouped["decision"]))
169
170        if grouped["preference"]:
171            parts.append("使用者偏好：\n" + "\n".join(f"- {p}" for p in grouped["preference"]))
172
173        if grouped["question"]:
174            parts.append("待解答的問題：\n" + "\n".join(f"- {q}" for q in grouped["question"]))
175
176        return "\n\n".join(parts)
177
178
179class SemanticCompressor:
180    """語意壓縮器"""
181
182    def __init__(self, config: Optional[SemanticCompressionConfig] = None):
183        self.config = config or SemanticCompressionConfig()
184        self.memory = SemanticMemoryStore(self.config)
185        self.recent_messages: list[dict] = []
186
187    def _extract_semantic_units(self, messages: list[dict]) -> list[SemanticUnit]:
188        """從對話中提取語意單元"""
189        messages_text = "\n".join(
190            f"[{'使用者' if m['role'] == 'user' else '助手'}]: {m['content']}"
191            for m in messages
192        )
193
194        prompt = f"""分析以下對話，提取關鍵語意單元。
195
196對話內容：
197{messages_text}
198
199請以 JSON 格式返回提取的語意單元列表：
200[
201    {{
202        "content": "語意內容",
203        "type": "fact|decision|question|preference|context",
204        "importance": "critical|high|medium|low",
205        "entities": ["相關實體1", "相關實體2"]
206    }}
207]
208
209類型說明：
210- fact: 已確認的事實或資訊
211- decision: 已做出的決定或選擇
212- question: 待解答的問題
213- preference: 使用者的偏好或習慣
214- context: 背景資訊
215
216重要性說明：
217- critical: 核心身份、關鍵決策（必須保留）
218- high: 重要技術細節、主要目標
219- medium: 一般資訊
220- low: 次要細節
221
222只返回 JSON，不要其他說明。"""
223
224        response = client.messages.create(
225            model=self.config.extraction_model,
226            max_tokens=2000,
227            messages=[{"role": "user", "content": prompt}]
228        )
229
230        try:
231            response_text = response.content[0].text
232            # 處理可能的 markdown 程式碼區塊
233            if "```json" in response_text:
234                response_text = response_text.split("```json")[1].split("```")[0]
235            elif "```" in response_text:
236                response_text = response_text.split("```")[1].split("```")[0]
237
238            units_data = json.loads(response_text.strip())
239
240            units = []
241            for data in units_data:
242                unit = SemanticUnit(
243                    content=data["content"],
244                    unit_type=data["type"],
245                    importance=ImportanceLevel(data["importance"]),
246                    turn_created=self.memory.current_turn,
247                    related_entities=data.get("entities", [])
248                )
249                units.append(unit)
250
251            return units
252
253        except (json.JSONDecodeError, KeyError) as e:
254            print(f"⚠️ 語意提取解析失敗: {e}")
255            return []
256
257    def add_turn(self, user_message: str, assistant_message: str):
258        """添加一輪對話"""
259        self.recent_messages.append({"role": "user", "content": user_message})
260        self.recent_messages.append({"role": "assistant", "content": assistant_message})
261        self.memory.current_turn += 1
262
263        # 如果最近訊息超過保留數量，進行語意提取
264        if len(self.recent_messages) > self.config.recent_turns * 2:
265            # 取出要壓縮的訊息
266            messages_to_compress = self.recent_messages[:-self.config.recent_turns * 2]
267            self.recent_messages = self.recent_messages[-self.config.recent_turns * 2:]
268
269            # 提取語意單元
270            print(f"🔍 正在提取語意單元...")
271            units = self._extract_semantic_units(messages_to_compress)
272
273            if units:
274                self.memory.add_units(units)
275                print(f"✅ 提取了 {len(units)} 個語意單元")
276
277            # 衰減和修剪
278            self.memory.decay_importance()
279            self.memory.prune()
280
281    def get_context(self) -> str:
282        """取得語意 context"""
283        return self.memory.to_context_string()
284
285    def get_messages(self) -> list[dict]:
286        """取得要發送給 API 的訊息"""
287        messages = []
288
289        context = self.get_context()
290        if context:
291            messages.append({
292                "role": "user",
293                "content": f"[對話記憶]\n{context}\n\n請基於以上背景繼續對話。"
294            })
295            messages.append({
296                "role": "assistant",
297                "content": "我已了解背景資訊，請繼續。"
298            })
299
300        messages.extend(self.recent_messages)
301        return messages
302
303    def get_stats(self) -> dict:
304        return {
305            "total_semantic_units": len(self.memory.units),
306            "current_turn": self.memory.current_turn,
307            "recent_messages": len(self.recent_messages),
308            "entities_tracked": len(self.memory.entity_index)
309        }
310
311
312class SemanticCompressionChatbot:
313    """使用語意壓縮的聊天機器人"""
314
315    def __init__(
316        self,
317        system_prompt: str,
318        config: Optional[SemanticCompressionConfig] = None
319    ):
320        self.system_prompt = system_prompt
321        self.compressor = SemanticCompressor(config)
322
323    def chat(self, user_input: str) -> str:
324        messages = self.compressor.get_messages()
325        messages.append({"role": "user", "content": user_input})
326
327        response = client.messages.create(
328            model="claude-sonnet-4-6",
329            max_tokens=2048,
330            system=self.system_prompt,
331            messages=messages
332        )
333
334        assistant_response = response.content[0].text
335        self.compressor.add_turn(user_input, assistant_response)
336
337        return assistant_response
338
339    def get_memory_snapshot(self) -> list[dict]:
340        """取得記憶快照"""
341        return [unit.to_dict() for unit in self.compressor.memory.units]
342
343
344# 使用範例
345if __name__ == "__main__":
346    config = SemanticCompressionConfig(
347        recent_turns=2,
348        max_semantic_units=30
349    )
350
351    bot = SemanticCompressionChatbot(
352        system_prompt="你是一個專業的技術顧問。",
353        config=config
354    )
355
356    conversations = [
357        "你好，我叫小明，是後端工程師",
358        "我擅長 Python 和 Go",
359        "最近在開發一個即時通訊系統",
360        "使用 WebSocket 進行雙向通訊",
361        "遇到了連線數擴展的問題",
362        "目前單機只能支撐 10000 連線",
363        "想要擴展到 100000 連線",
364        "你建議用什麼架構？",
365        # 更多對話...
366    ]
367
368    for msg in conversations:
369        print(f"\n👤 User: {msg}")
370        response = bot.chat(msg)
371        print(f"🤖 Bot: {response[:200]}...")
372
373    print("\n" + "="*50)
374    print("記憶快照：")
375    for unit in bot.get_memory_snapshot():
376        print(f"  [{unit['importance']}] ({unit['type']}) {unit['content'][:50]}...")