1 上下文压缩

大体上分为三部分：（1）模型响应内容的压缩 、（2）工具结果的压缩 、（3）工具注入（System_Prompt）的压缩

# 导入类型注解：列表、字典
from typing import List, Dict

# 上下文管理器：统一管理【上下文压缩】 + 【技能懒加载】
class ContextManager:
    # 初始化：创建上下文压缩器 + 技能加载器
    def __init__(self):
        # 上下文压缩器（你原来写的，用来压缩长对话）
        self.compressor = ContextCompressor()
        # 技能加载器（用来加载工具/技能的元数据、详情）
        self.skill_loader = SkillLoader()

    # 对外提供：压缩消息列表
    def compress_messages(self, messages: List[Dict]) -> List[Dict]:
        # 调用压缩器，返回压缩后的消息列表
        return self.compressor.compress(messages)

    # ====================== 🔥 正确懒加载 ======================
    # 获取技能（只加载元数据，不加载完整参数）
    def get_skills_lazy(self, skill_names: List[str] = None) -> List[Dict]:
        """
        第一层懒加载：只返回工具的【名称 + 描述】
        作用：给 LLM 看，让它知道有什么工具可以用，不占 token
        :param skill_names: 指定要加载的技能列表，不传则加载全部
        :return: 工具元数据列表
        """
        # 如果没有指定技能名 → 加载所有技能的元数据
        if not skill_names:
            metas = self.skill_loader.get_all_metas()
        # 如果指定了技能名 → 只加载指定的几个技能元数据
        else:
            metas = []
            for name in skill_names:
                # 从技能加载器获取单个技能的元数据
                m = self.skill_loader.get_meta(name)
                if m:  # 如果获取到了，就加入列表
                    metas.append(m)
        return metas

    # 获取工具【完整信息】（包含参数、实现等）
    def get_skill_full(self, skill_name: str) -> Dict:
        """
        第二层懒加载：
        🔥 只有在【真正要调用这个工具时】，才加载完整参数！
        平时绝不加载，节省大量 token
        """
        # 加载并返回某个技能的完整详情
        return self.skill_loader.load_skill_detail(skill_name)

    # 获取上下文压缩统计信息（压缩了多少、节省多少token等）
    def get_compression_stats(self):
        return self.compressor.get_compression_stats()

2（1）模型响应内容的压缩 compress_messages()

    def compress(self, messages: List[Dict], force_summary: bool = False) -> List[Dict]:
        """🗜️ 压缩消息历史"""
        if len(messages) <= self.max_messages and not force_summary:
            return messages
        
        # 1. 分离系统消息（始终保留）
        system_msgs = [m for m in messages if m["role"] == "system"]
        # 2. 保留最近 N 条非系统消息
        recent_msgs = [m for m in messages if m["role"] != "system"][-self.max_messages:]
        
        # 3. 生成历史摘要（如果触发）
        if force_summary or len(messages) % self.summary_interval == 0:
            old_msgs = [m for m in messages if m["role"] != "system"][:-self.max_messages]
            if old_msgs:
                summary = self._generate_summary(old_msgs)
                self._summaries.append(summary)
        
        # 4. 合并：系统消息 + 摘要 + 最近消息
        compressed = system_msgs.copy()
        if self._summaries:
            compressed.append({
                "role": "system",
                "content": f"📚 Conversation Summary (earlier):\n{self._summaries[-1]['content']}"
            })
        compressed.extend(recent_msgs)
        
        # 5. 替换大工具结果为占位符
        return self._replace_large_results(compressed)

• 第一步：保留最近几条非 System 的内容： 响应内容很多都是重复的，尤其在子任务的多轮解答中，所以我们只需要保留最近几条即可，一般显示只会显示最近一条；

			  # 1. 分离系统消息（始终保留）
        system_msgs = [m for m in messages if m["role"] == "system"]
        # 2. 保留最近 N 条非系统消息
        recent_msgs = [m for m in messages if m["role"] != "system"][-self.max_messages:]

• 第二步：如果 token 数量达到阈值，则对输出进行摘要化： 为了保证我们的上下文能够足够长，能够进行多论问答。所以需要对响应内容进行摘要化。

        # 3. 生成历史摘要（如果触发）
        if force_summary or len(messages) % self.summary_interval == 0:
            old_msgs = [m for m in messages if m["role"] != "system"][:-self.max_messages]
            if old_msgs:
                summary = self._generate_summary(old_msgs)
                self._summaries.append(summary)

• 第三步：工具调用的结果进行占位符替换 工具的结果一般是 action 的操作，我们直接替换即可；
  注意替换也是有边界条件的，比如 tool_result 超过 500 个token的；

import hashlib  # 代码里用到了，需要导入
def _replace_large_results(self, messages: List[Dict], threshold: int = 500) -> List[Dict]:
    """
    🔖 核心功能：将超长的工具调用结果替换成简短占位符
    目的：减少 LLM 上下文长度、节省 Token、防止上下文溢出
    只会替换 role = tool 的消息（工具返回的结果）
    """
    # 存储压缩后的消息列表
    compressed = []

    # 遍历每一条上下文消息
    for m in messages:
        # 判断：是否是【工具返回的消息】 + 内容长度超过【阈值500字符】
        if m.get("role") == "tool" and len(m.get("content", "")) > threshold:
            
            # 生成唯一占位符 ID：TR_ + 内容MD5哈希前6位（保证不重复）
            placeholder_id = f"TR_{hashlib.md5(m['content'].encode()).hexdigest()[:6]}"
            
            # 把原始的大段工具结果存起来（用占位符ID做key）
            self._result_placeholders[placeholder_id] = m["content"]
            
            # 替换原消息：只保留占位符，不再携带大段文本
            compressed.append({
                **m,  # 保留原消息的其他字段（如 role、tool_call_id 等）
                # 内容替换成简短提示
                "content": f"[TOOL_RESULT:{placeholder_id}] (output truncated, {len(m['content'])} chars)"
            })
        else:
            # 不符合条件的消息（普通消息/短工具结果）直接保留
            compressed.append(m)

    # 返回压缩后的消息
    return compressed

3（2）工具的压缩 SkillLoader 类

主要实现方式： 通过 Skill 的方式，将主要信息比如 name、desc、参数 注入到 System_Prompt 中，在真正需要某个 tool 的时候再进行全局加载。
以下是 Skill 加载的方式： 包含 meta 数据的加载，和详细数据的加载，以及缓存的实现；

# core/skill_loader.py
import os, re, json
from pathlib import Path
from typing import Dict, Optional, List
import yaml  # pip install pyyaml

class SkillLoader:
    """🔧 技能懒加载器 - 从.md 文件动态加载工具元信息"""
    # 1. 初始化：只扫目录、只加载 meta
    def __init__(self, tools_dir: str = "tools"):
        self.tools_dir = Path(tools_dir)
        self._meta_cache: Dict[str, Dict] = {}   # 元信息缓存（常驻内存）
        self._detail_cache: Dict[str, str] = {}  # 详细内容缓存（懒加载）
        self.tools_dir.mkdir(parents=True, exist_ok=True)
        self._scan_meta() # 只扫 meta，不读详情
    
    # 2. 扫描 MD 文件：扫描 tools 目录下所有的 .md 文件
    def _scan_meta(self):
        """扫描 tools 目录，预加载技能元信息"""
        for md_file in self.tools_dir.glob("*.md"):
            meta = self._parse_frontmatter(md_file)
            if meta and "name" in meta:
            		 # 按键值对的形式进行存储，包含 meta 数据和文件路径，详情不加载。
                self._meta_cache[meta["name"]] = {
                    **meta,
                    "_file": str(md_file),
                    "_loaded_detail": False
                }
    
    # 3. 根据分割符--- 严格匹配，取出其中内容，返回 dict（name/description/parameters）
    def _parse_frontmatter(self, file_path: Path) -> Optional[Dict]:
        """解析 Markdown 文件头部的 YAML frontmatter"""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        match = re.match(r'^---\s*\n(.*?)\n---\s*\n', content, re.DOTALL)
        if match:
            try:
            		 # 把 YAML 文本（文本字符串） 转换成了 Python 字典
                return yaml.safe_load(match.group(1))
            except:
                return None
        return None
    
    def get_meta(self, skill_name: str) -> Optional[Dict]:
        """获取技能元信息（始终在内存）"""
        return self._meta_cache.get(skill_name)
    
    def get_all_metas(self) -> List[Dict]:
        """获取所有技能元信息列表（用于构建 tools 参数）"""
        metas = []
        for name, meta in self._meta_cache.items():
            # 转换为 OpenAI function calling 格式
            func_def = {
                "type": "function",
                "function": {
                    "name": meta["name"],
                    "description": meta.get("description", ""),
                    "parameters": {
                        "type": "object",
                        "properties": meta.get("parameters", {}),
                        "required": [k for k, v in meta.get("parameters", {}).items() 
                                   if isinstance(v, dict) and v.get("required")]
                    }
                }
            }
            metas.append(func_def)
            # 返回字典集合
        return metas
    
    def load_detail(self, skill_name: str) -> Optional[str]:
        """🐌 懒加载：首次使用时加载技能详细内容"""
        if skill_name not in self._meta_cache:
            return None
        meta = self._meta_cache[skill_name]
        if meta.get("_loaded_detail"): # 肯定是 False
            return self._detail_cache.get(skill_name)
        
        # 读取详细内容（去除 frontmatter）
        with open(meta["_file"], 'r', encoding='utf-8') as f:
            content = f.read()
        # 移除 YAML frontmatter
        detail = re.sub(r'^---\s*\n.*?\n---\s*\n', '', content, flags=re.DOTALL)
        # 所有内容
        self._detail_cache[skill_name] = detail.strip()
        meta["_loaded_detail"] = True
        return detail
    
    def get_skill_prompt(self, skill_name: str) -> str:
        """生成技能调用提示（元信息 + 详细内容）"""
        meta = self.get_meta(skill_name)
        detail = self.load_detail(skill_name)
        if not meta or not detail:
            return f"⚠️ Skill '{skill_name}' not found"
        
        return f"""## Tool: {meta['name']}
📋 Description: {meta.get('description', 'No description')}

📦 Parameters:
{json.dumps(meta.get('parameters', {}), indent=2, ensure_ascii=False)}

📚 Usage Guide:
{detail}
"""

懒加载的具体实现： 将 meta 数据封装到 metas 列表中；

def get_skills_lazy(self, skill_names: List[str] = None) -> List[Dict]:
        """懒加载技能：按需获取技能元数据，而非全部加载
            :param skill_names: 指定需要的技能名称，为空则返回所有技能元数据
        """
        if not skill_names:
            return self.skill_loader.get_all_metas()
        metas = []
        for name in skill_names:
            # 根据名称获取单个技能元数据
            meta = self.skill_loader.get_meta(name)
            if meta:
                # 组装为大模型支持的函数调用格式
                metas.append({"type": "function", "function": {"name": meta["name"], "description": meta.get("description", ""), "parameters": meta.get("parameters", {})}})
        return metas

4 示例

def demo_lazy_skill_loading():
    """演示 5: 技能懒加载"""
    print("\n=== 演示 5: 技能懒加载 ===")
    ctx_mgr = ContextManager()

    # 获取所有技能元信息（不加载详细内容）
    skills = ctx_mgr.get_skills_lazy()
    print(f"✅ 加载了 {len(skills)} 个技能元信息")

    # 获取特定技能
    specific = ctx_mgr.get_skills_lazy(["execute_bash", "read_file"])
    print(f"✅ 加载了 {len(specific)} 个特定技能")
    for skill in specific:
        print(f"  - {skill['function']['name']}: {skill['function']['description']}")

• 第一次加载： 可以看到代码 skills = ctx_mgr.get_skills_lazy() ，输入不包含任何参数，这是 meta 数据提取的方式；
• 第二次加载： 代码 .get_skills_lazy(["execute_bash", "read_file"]) 表示详细数据的提取方式；