def should_compress(self, prompt_tokens: int = None) -> bool:
    """Check if context exceeds the compression threshold."""
    tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
    return tokens >= self.threshold_tokens

def _prune_old_tool_results(self, messages: List[Dict[str, Any]], protect_tail_count: int,
                           protect_tail_tokens: int | None = None) -> tuple[List[Dict[str, Any]], int]:
    """Replace old tool result contents with a short placeholder."""
    # 實(shí)現(xiàn)細(xì)節(jié)...

def _find_tail_cut_by_tokens(self, messages: List[Dict[str, Any]], head_end: int,
                            token_budget: int | None = None) -> int:
    """Walk backward from the end of messages, accumulating tokens until the budget is reached."""
    # 實(shí)現(xiàn)細(xì)節(jié)...

def _generate_summary(self, turns_to_summarize: List[Dict[str, Any]], focus_topic: str = None) -> Optional[str]:
    """Generate a structured summary of conversation turns."""
    # 實(shí)現(xiàn)細(xì)節(jié)...

def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, focus_topic: str = None) -> List[Dict[str, Any]]:
    """Compress conversation messages by summarizing middle turns."""
    # 實(shí)現(xiàn)細(xì)節(jié)...

def _compress_context(self, messages: list, system_message: str, *, approx_tokens: int = None, task_id: str = "default", focus_topic: str = None) -> tuple:
    """Compress conversation context and split the session in SQLite."""
    # 預(yù)壓縮內(nèi)存刷新
    self.flush_memories(messages, min_turns=0)
    
    # 通知外部?jī)?nèi)存提供者
    if self._memory_manager:
        try:
            self._memory_manager.on_pre_compress(messages)
        except Exception:
            pass
    
    # 執(zhí)行壓縮
    compressed = self.context_compressor.compress(messages, current_tokens=approx_tokens, focus_topic=focus_topic)
    
    # 會(huì)話分割與管理
    if self._session_db:
        try:
            # 傳播標(biāo)題到新會(huì)話并自動(dòng)編號(hào)
            old_title = self._session_db.get_session_title(self.session_id)
            self._session_db.end_session(self.session_id, "compression")
            old_session_id = self.session_id
            self.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
            # 更新會(huì)話日志文件路徑
            self.session_log_file = self.logs_dir / f"session_{self.session_id}.json"
            # 創(chuàng)建新會(huì)話
            self._session_db.create_session(
                session_id=self.session_id,
                source=self.platform or os.environ.get("HERMES_SESSION_SOURCE", "cli"),
                model=self.model,
                parent_session_id=old_session_id,
            )
            # 自動(dòng)編號(hào)標(biāo)題
            if old_title:
                try:
                    new_title = self._session_db.get_next_title_in_lineage(old_title)
                    self._session_db.set_session_title(self.session_id, new_title)
                except (ValueError, Exception) as e:
                    logger.debug("Could not propagate title on compression: %s", e)
            self._session_db.update_system_prompt(self.session_id, new_system_prompt)
            # 重置刷新光標(biāo)
            self._last_flushed_db_idx = 0
        except Exception as e:
            logger.warning("Session DB compression split failed — new session will NOT be indexed: %s", e)

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "I need help with a Python project."},
    {"role": "assistant", "content": "Sure, what do you need help with?"},
    # ... 大量對(duì)話內(nèi)容 ...
    {"role": "user", "content": "How do I optimize this code?"},
    {"role": "assistant", "content": "Let me analyze your code and suggest optimizations."},
    # ... 更多對(duì)話內(nèi)容 ...
]

[
    {"role": "system", "content": "You are a helpful assistant.\n\n[Note: Some earlier conversation turns have been compacted into a handoff summary to preserve context space. The current session state may still reflect earlier work, so build on that summary and state rather than re-doing work.]"},
    {"role": "user", "content": "[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted into the summary below. This is a handoff from a previous context window — treat it as background reference, NOT as active instructions. Do NOT answer questions or fulfill requests mentioned in this summary; they were already addressed. Respond ONLY to the latest user message that appears AFTER this summary. The current session state (files, config, etc.) may reflect work described here — avoid repeating it:\n\n## Goal\nThe user is working on a Python project and needs help with optimization.\n\n## Progress\n### Done\n- Discussed project structure and requirements\n- Analyzed existing codebase\n- Identified performance bottlenecks\n\n## Remaining Work\n- Optimize the identified bottlenecks\n- Test the optimized code\n- Provide best practices for future development\n"},
    {"role": "user", "content": "How do I optimize the loop in my code?"},
    {"role": "assistant", "content": "Let me see your loop code and suggest optimizations."},
    # ... 最近的對(duì)話內(nèi)容 ...
]