Upload folder using huggingface_hub

README.md

@@ -0,0 +1,128 @@
+---
+language:
+  - zh
+  - en
+license: gpl-3.0
+tags:
+  - agent
+  - kv-cache
+  - compression
+  - memory-management
+  - deepseek
+  - autonomous-agent
+---
+
+# AgentFrame v2 — 自主记忆 Agent 框架
+
+**脑 = DeepSeek · 手 = 工具执行 · 记忆 = 模型自主管理**
+
+> v2 核心升级: **MemoryDirector 自主记忆** — 模型自己决定记什么、忘什么
+> 存储层仍保持 35.6x KV 压缩 (吸收式 MLA + INT4, L40S 实测)
+
+---
+
+## 🆕 v2 新特性: 模型自主记忆 (MemoryDirector)
+
+传统 KV 压缩靠**数值启发式**驱逐 (L2范数/Δh/注意力入度) — 全部被反例证伪, 因为它们不懂语义。
+v2 让 **LLM 自己当驱逐决策器**: 每轮对话后分析"什么值得记, 什么该撇了"。
+
+### 实测效果
+
+| 输入 | 模型决策 | 正确性 |
+|------|---------|--------|
+| 域名+密码+路径+闲聊 | ✅ 记域名/路径, **拒绝记密码**, 忘天气 | 100% |
+| 重申已有信息 | ✅ 去重, 0 条重复入库 | 100% |
+| 纯闲聊 | ✅ 全忘, 不浪费缓存 | 100% |
+
+**惊喜**: 模型自主涌现安全意识 — "密码不宜存储", 拒绝把密码写入记忆 (非硬编码规则)。
+
+### 压缩比 (三层叠加)
+
+```
+输入对话 67 tokens (4条关键信息 + 2条闲聊)
+  ↓ ① 语义蒸馏 (自主记忆): 3.2x   模型剔除闲聊, 只留关键
+  ↓ ② KV 物理压缩 (MLA+INT4): 35.6x  270KB→7.6KB/token (L40S实测)
+  ↓ ③ 总压缩: ~113x (17.7MB → 159KB)
+```
+
+## 🏗️ 架构
+
+```
+┌─────────────────────────────────────────────┐
+│             ContextEngine 引擎               │
+│  ┌────────┐ ┌──────────┐ ┌────────┐ ┌─────┐ │
+│  │L1 认知  │→│L2 路由    │→│L3 存储  │→│L4 物理│ │
+│  │MetaCog │ │Landmark  │ │Absorbed│ │Pager│ │
+│  │任务分解 │ │检索top-k  │ │MLA压缩  │ │换页  │ │
+│  └────────┘ └──────────┘ └────────┘ └─────┘ │
+│         ↕           ↕           ↕        ↕  │
+│  ┌───────────────────────────────────────┐   │
+│  │  MemoryDirector (v2): 模型自主记/忘     │   │
+│  │  LLM Provider (DeepSeek, thinking+工具) │   │
+│  │  Embedding (哈希/API, 文本→向量)         │   │
+│  └───────────────────────────────────────┘   │
+└─────────────────────────────────────────────┘
+```
+
+## 🚀 快速开始
+
+```bash
+pip install -e agentframe/
+export AGENTFRAME_API_KEY="sk-xxx"
+```
+
+### CLI
+```bash
+python3 -m agentframe.cli demo                  # 离线演示
+python3 -m agentframe.cli ingest "知识" --tags kv
+python3 -m agentframe.cli ask "问题"
+```
+
+### Python
+```python
+from agentframe.config import AgentFrameConfig
+from agentframe.core.engine import ContextEngine
+
+eng = ContextEngine(AgentFrameConfig.from_env())
+r = eng.ask_autopilot("我的服务器 IP 是 98.142.241.130, SSH 端口 44123")
+print(r.meta["memory_decision"])   # 自主记忆决策: 记住/遗忘/重要性
+```
+
+### REST API
+```bash
+python3 -m agentframe.api.server 8090
+# POST /v1/sessions
+# POST /v1/sessions/<sid>/ask         查询
+# POST /v1/sessions/<sid>/autopilot   自主记忆决策
+# POST /v1/sessions/<sid>/ingest      摄入
+# POST /v1/sessions/<sid>/forget      遗忘
+```
+
+## 📁 结构
+
+```
+agentframe/
+├── config.py          # 配置 (env/JSON)
+├── core/
+│   ├── quad.py        # 四层核心 (KV压缩/分层/路由/遗忘)
+│   └── engine.py      # ContextEngine + MemoryDirector
+├── llm/               # DeepSeek Provider (thinking+工具循环)
+├── embed/             # 哈希/API 嵌入
+├── memory/            # JSON 快照持久化
+├── api/               # REST API v1
+└── cli.py             # 命令行
+```
+
+## 🔬 底层技术 (实测)
+
+1. **吸收式 MLA**: 576维潜在向量, 270KB→7.6KB/token = **35.6x** (L40S)
+2. **Top-K 精度保护**: 关键块16bit+其余4bit = **0/100翻转**
+3. **Sector-Block-Module**: 16/256/1024 分层硬件对齐
+4. **Aura 遗忘曲线**: S(t)=I·2^(-t/τ) + log2(access+1)×0.1
+5. **MemoryDirector**: LLM 语义判断驱逐 (替代被证伪的数值启发式)
+
+## 📜 License
+
+GPL-3.0 © Cloud LTE Studio
+
+*AgentFrame v2 · 模型自主记忆 · 上下文永不丢失*

__init__.py

@@ -0,0 +1,41 @@
+"""
+AgentFrame — Agent 专用上下文保持框架
+======================================
+脑 = DeepSeek (LLM Provider)
+手 = 工具执行 (Function Calling)
+记忆 = 四层上下文保持 (认知 × 路由 × 存储 × 物理)
+
+版本: 1.0.0 (成熟框架版)
+"""
+__version__ = "1.0.0"
+
+from .core.quad import (
+    CompressedKV,
+    HierarchicalKV,
+    ReversibleQuantizer,
+    AbsorbedMLA,
+    ChunkSelection,
+    LandmarkRouter,
+    ForgettingCurve,
+    KVPager,
+    SubTask,
+    RetrievalDirective,
+    MetaCog,
+    QuadLayerAgent,
+)
+
+__all__ = [
+    "__version__",
+    "CompressedKV",
+    "HierarchicalKV",
+    "ReversibleQuantizer",
+    "AbsorbedMLA",
+    "ChunkSelection",
+    "LandmarkRouter",
+    "ForgettingCurve",
+    "KVPager",
+    "SubTask",
+    "RetrievalDirective",
+    "MetaCog",
+    "QuadLayerAgent",
+]

api/__init__.py

@@ -0,0 +1,4 @@
+"""API 模块"""
+from .server import AgentFrameAPI, create_app
+
+__all__ = ["AgentFrameAPI", "create_app"]

api/server.py

@@ -0,0 +1,239 @@
+"""
+AgentFrame REST API v1
+=======================
+多会话管理: 每个 session 独立 ContextEngine 实例
+
+端点:
+  GET    /health                         健康检查
+  POST   /v1/sessions                    创建会话
+  GET    /v1/sessions                    列出会话
+  DELETE /v1/sessions/<sid>              删除会话
+  POST   /v1/sessions/<sid>/ingest       摄入知识
+  POST   /v1/sessions/<sid>/ingest_batch 批量摄入
+  POST   /v1/sessions/<sid>/ask          查询+生成
+  POST   /v1/sessions/<sid>/ask_hands    查询+生成(带工具)
+  GET    /v1/sessions/<sid>/stats        会话状态
+  POST   /v1/sessions/<sid>/forget       触发遗忘
+  POST   /v1/sessions/<sid>/save         保存快照
+  POST   /v1/sessions/<sid>/load         恢复快照
+"""
+import os
+import uuid
+
+from flask import Flask, request, jsonify
+
+from ..config import AgentFrameConfig
+from ..core.engine import ContextEngine
+
+
+class AgentFrameAPI:
+    """Flask 应用工厂"""
+
+    def __init__(self, config: AgentFrameConfig = None, state_dir: str = None):
+        self.config = config or AgentFrameConfig.from_env()
+        self.state_dir = state_dir or os.environ.get(
+            "AGENTFRAME_STATE_DIR", "/var/lib/agentframe")
+        os.makedirs(self.state_dir, exist_ok=True)
+
+        self.sessions = {}   # sid -> ContextEngine
+
+        self.app = Flask("agentframe")
+        self._register_routes()
+
+    # ============ 会话管理 ============
+
+    def _get_session(self, sid: str) -> ContextEngine:
+        if sid not in self.sessions:
+            raise KeyError(f"会话不存在: {sid}")
+        return self.sessions[sid]
+
+    def _create_session(self) -> str:
+        sid = uuid.uuid4().hex[:12]
+        self.sessions[sid] = ContextEngine(self.config)
+        return sid
+
+    def _restore_sessions(self):
+        """启动时自动恢复已保存的会话快照"""
+        if not os.path.isdir(self.state_dir):
+            return
+        for fname in sorted(os.listdir(self.state_dir)):
+            if not fname.endswith(".json"):
+                continue
+            sid = fname[:-5]
+            if sid in self.sessions:
+                continue
+            eng = ContextEngine(self.config)
+            try:
+                if eng.load(os.path.join(self.state_dir, fname)):
+                    self.sessions[sid] = eng
+                    print(f"   ↻ 恢复会话 {sid}")
+            except Exception as e:
+                print(f"   ⚠️ 恢复失败 {sid}: {e}")
+
+    # ============ 路由 ============
+
+    def _register_routes(self):
+        app = self.app
+        self._restore_sessions()
+
+        @app.get("/health")
+        def health():
+            return jsonify({"status": "ok", "version": "1.0.0",
+                            "sessions": len(self.sessions),
+                            "system": "AgentFrame Context API"})
+
+        @app.post("/v1/sessions")
+        def create_session():
+            sid = self._create_session()
+            return jsonify({"session_id": sid, "ok": True}), 201
+
+        @app.get("/v1/sessions")
+        def list_sessions():
+            return jsonify({
+                "sessions": [
+                    {"id": sid, "chunks": len(eng.agent.chunk_meta),
+                     "turns": len(eng.chat_history) // 2}
+                    for sid, eng in self.sessions.items()
+                ],
+                "total": len(self.sessions),
+            })
+
+        @app.delete("/v1/sessions/<sid>")
+        def delete_session(sid):
+            if sid in self.sessions:
+                del self.sessions[sid]
+                return jsonify({"ok": True, "deleted": sid})
+            return jsonify({"error": "会话不存在"}), 404
+
+        @app.post("/v1/sessions/<sid>/ingest")
+        def ingest(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            data = request.get_json(force=True, silent=True) or {}
+            text = data.get("text", "")
+            if not text:
+                return jsonify({"error": "缺少 text 字段"}), 400
+            cid = eng.ingest(text, data.get("tags", []), data.get("source"))
+            return jsonify({"ok": True, "chunk_id": cid, "state": eng.stats()})
+
+        @app.post("/v1/sessions/<sid>/ingest_batch")
+        def ingest_batch(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            data = request.get_json(force=True, silent=True) or {}
+            chunks = data.get("chunks", [])
+            if not chunks:
+                return jsonify({"error": "缺少 chunks 字段"}), 400
+            ids = eng.ingest_batch(chunks)
+            return jsonify({"ok": True, "chunk_ids": ids})
+
+        @app.post("/v1/sessions/<sid>/ask")
+        def ask(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            data = request.get_json(force=True, silent=True) or {}
+            q = data.get("query", "")
+            if not q:
+                return jsonify({"error": "缺少 query 字段"}), 400
+            result = eng.ask(q, chat=data.get("chat", True),
+                             use_fast=data.get("use_fast", False))
+            return jsonify({"ok": True, **result.to_dict()})
+
+        @app.post("/v1/sessions/<sid>/ask_hands")
+        def ask_hands(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            data = request.get_json(force=True, silent=True) or {}
+            q = data.get("query", "")
+            if not q:
+                return jsonify({"error": "缺少 query 字段"}), 400
+            result = eng.ask_with_hands(q)
+            return jsonify({"ok": True, **result.to_dict()})
+
+        @app.post("/v1/sessions/<sid>/autopilot")
+        def autopilot(sid):
+            """自主记忆管理: 分析对话, 自动记/忘"""
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            data = request.get_json(force=True, silent=True) or {}
+            q = data.get("query", "")
+            if not q:
+                return jsonify({"error": "缺少 query 字段"}), 400
+            # 用最近一次回答做记忆分析
+            answer = ""
+            if eng.chat_history and eng.chat_history[-1].get("role") == "assistant":
+                answer = eng.chat_history[-1]["content"]
+            decision = eng.autopilot_memory(q, answer)
+            return jsonify({"ok": True, **decision})
+
+        @app.get("/v1/sessions/<sid>/stats")
+        def stats(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            return jsonify(eng.stats())
+
+        @app.post("/v1/sessions/<sid>/forget")
+        def forget(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            data = request.get_json(force=True, silent=True) or {}
+            threshold = float(data.get("threshold", 0.15))
+            victims = eng.forget(threshold)
+            return jsonify({"ok": True, "forgotten": victims,
+                            "remaining": len(eng.agent.chunk_meta)})
+
+        @app.post("/v1/sessions/<sid>/save")
+        def save(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            path = os.path.join(self.state_dir, f"{sid}.json")
+            eng.save(path)
+            return jsonify({"ok": True, "path": path})
+
+        @app.post("/v1/sessions/<sid>/load")
+        def load(sid):
+            try:
+                eng = self._get_session(sid)
+            except KeyError as e:
+                return jsonify({"error": str(e)}), 404
+            path = os.path.join(self.state_dir, f"{sid}.json")
+            ok = eng.load(path)
+            return jsonify({"ok": ok, "path": path})
+
+
+def create_app(config: AgentFrameConfig = None, state_dir: str = None) -> Flask:
+    """应用工厂 (供 gunicorn/flask run 使用)"""
+    return AgentFrameAPI(config, state_dir).app
+
+
+def main():
+    """直接运行: python -m agentframe.api.server [port]"""
+    import sys
+    config = AgentFrameConfig.from_env()
+    port = int(sys.argv[1]) if len(sys.argv) > 1 else config.api.port
+    api = AgentFrameAPI(config)
+    print(f"🚀 AgentFrame API v1.0.0")
+    print(f"   四层上下文保持: 认知×路由×存储×物理")
+    print(f"   会话: POST /v1/sessions | 摄入: /ingest | 查询: /ask")
+    print(f"   监听: {config.api.host}:{port}")
+    api.app.run(host=config.api.host, port=port, debug=config.api.debug)
+
+
+if __name__ == "__main__":
+    main()

cli.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+AgentFrame CLI — 命令行工具
+============================
+用法:
+  agentframe serve [--port 8090] [--config config.json]
+  agentframe ingest "文本" [--tags a,b] [--session <sid>] [--state <path>]
+  agentframe ask "问题" [--no-chat] [--hands] [--state <path>]
+  agentframe stats [--state <path>]
+  agentframe forget [--threshold 0.15] [--state <path>]
+  agentframe config [--output config.json]
+  agentframe demo
+"""
+import argparse
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from agentframe.config import AgentFrameConfig
+from agentframe.core.engine import ContextEngine
+
+DEFAULT_STATE = "/var/lib/agentframe/cli.json"
+
+
+def load_engine(config_path: str = None, state_path: str = None) -> ContextEngine:
+    config = AgentFrameConfig.from_file(config_path) if config_path else AgentFrameConfig.from_env()
+    eng = ContextEngine(config)
+    sp = state_path or DEFAULT_STATE
+    if os.path.exists(sp):
+        eng.load(sp)
+    return eng
+
+
+def cmd_serve(args):
+    from agentframe.api.server import AgentFrameAPI
+    config = AgentFrameConfig.from_file(args.config) if args.config else AgentFrameConfig.from_env()
+    api = AgentFrameAPI(config)
+    print(f"🚀 AgentFrame API — http://{config.api.host}:{args.port or config.api.port}")
+    api.app.run(host=config.api.host, port=args.port or config.api.port,
+                debug=config.api.debug)
+
+
+def cmd_ingest(args):
+    eng = load_engine(args.config, args.state)
+    tags = args.tags.split(",") if args.tags else []
+    cid = eng.ingest(args.text, tags)
+    sp = args.state or DEFAULT_STATE
+    os.makedirs(os.path.dirname(sp), exist_ok=True)
+    eng.save(sp)
+    print(f"✅ 摄入 chunk_{cid}: {args.text[:50]}...")
+    print(f"   标签: {tags} | 已存: {sp}")
+
+
+def cmd_ask(args):
+    eng = load_engine(args.config, args.state)
+    if args.hands:
+        result = eng.ask_with_hands(args.query)
+    else:
+        result = eng.ask(args.query, chat=not args.no_chat)
+    print(f"\n🧠 检索到 {len(result.retrieved)} 块:")
+    for cid, prev in result.retrieved:
+        print(f"   chunk_{cid}: {prev}...")
+    if result.answer:
+        print(f"\n💬 回答:\n{result.answer}")
+    sp = args.state or DEFAULT_STATE
+    eng.save(sp)
+
+
+def cmd_stats(args):
+    eng = load_engine(args.config, args.state)
+    for k, v in eng.stats().items():
+        print(f"  {k}: {v}")
+
+
+def cmd_forget(args):
+    eng = load_engine(args.config, args.state)
+    victims = eng.forget(args.threshold)
+    sp = args.state or DEFAULT_STATE
+    eng.save(sp)
+    print(f"🧹 遗忘 {len(victims)} 块: {victims}")
+    print(f"   剩余: {len(eng.agent.chunk_meta)} 块")
+
+
+def cmd_config(args):
+    config = AgentFrameConfig.from_env()
+    out = args.output or "agentframe.config.json"
+    config.save(out)
+    print(f"✅ 配置已写入: {out}")
+    print(f"   LLM: {config.llm.provider} / {config.llm.model}")
+    print(f"   Memory: {config.memory.n_layers}层 INT{config.memory.quant_bits} top-k={config.memory.top_k}")
+    print(f"   API: {config.api.host}:{config.api.port}")
+
+
+def cmd_demo(args):
+    """离线演示: 摄入 → 查询 → 遗忘 (无 API key 也可跑)"""
+    config = AgentFrameConfig.from_env()
+    config.llm.provider = "mock"
+    eng = ContextEngine(config)
+
+    print("=" * 56)
+    print("AgentFrame 离线演示 (Mock LLM)")
+    print("=" * 56)
+
+    chunks = [
+        {"text": "KV 缓存 270KB/token 展开存储", "tags": ["data"]},
+        {"text": "吸收式 MLA 缓存 576 维潜在向量", "tags": ["method"]},
+        {"text": "L40S 实测 35.6x 压缩", "tags": ["data", "result"]},
+        {"text": "注意力分数不等于任务重要性", "tags": ["agent"]},
+    ]
+    for c in chunks:
+        cid = eng.ingest(c["text"], c["tags"])
+        print(f"  📥 摄入 chunk_{cid}: {c['text']}")
+
+    print(f"\n  [存储] {eng.stats()['layers']}")
+    print(f"  [压缩] {eng.stats()['bytes_per_token']} B/token")
+
+    r = eng.ask("KV压缩能到多少倍？", chat=True)
+    print(f"\n  🧠 检索: {r.retrieved}")
+    print(f"  💬 回答: {r.answer}")
+
+    v = eng.forget(0.3)
+    print(f"\n  🧹 遗忘: {len(v)} 块 → 剩余 {len(eng.agent.chunk_meta)} 块")
+    print("\n✅ 演示完成")
+
+
+def main():
+    p = argparse.ArgumentParser(prog="agentframe", description="AgentFrame 上下文保持框架")
+    sub = p.add_subparsers(dest="cmd")
+
+    pserve = sub.add_parser("serve", help="启动 API 服务")
+    pserve.add_argument("--port", type=int, default=None)
+    pserve.add_argument("--config", default=None)
+
+    pi = sub.add_parser("ingest", help="摄入知识")
+    pi.add_argument("text")
+    pi.add_argument("--tags", default=None)
+    pi.add_argument("--config", default=None)
+    pi.add_argument("--state", default=None)
+
+    pa = sub.add_parser("ask", help="查询")
+    pa.add_argument("query")
+    pa.add_argument("--no-chat", action="store_true")
+    pa.add_argument("--hands", action="store_true")
+    pa.add_argument("--config", default=None)
+    pa.add_argument("--state", default=None)
+
+    pst = sub.add_parser("stats", help="状态")
+    pst.add_argument("--config", default=None)
+    pst.add_argument("--state", default=None)
+
+    pf = sub.add_parser("forget", help="遗忘")
+    pf.add_argument("--threshold", type=float, default=0.15)
+    pf.add_argument("--config", default=None)
+    pf.add_argument("--state", default=None)
+
+    pc = sub.add_parser("config", help="生成配置")
+    pc.add_argument("--output", default=None)
+
+    sub.add_parser("demo", help="离线演示")
+
+    args = p.parse_args()
+    if args.cmd == "serve":
+        cmd_serve(args)
+    elif args.cmd == "ingest":
+        cmd_ingest(args)
+    elif args.cmd == "ask":
+        cmd_ask(args)
+    elif args.cmd == "stats":
+        cmd_stats(args)
+    elif args.cmd == "forget":
+        cmd_forget(args)
+    elif args.cmd == "config":
+        cmd_config(args)
+    elif args.cmd == "demo":
+        cmd_demo(args)
+    else:
+        p.print_help()
+
+
+if __name__ == "__main__":
+    main()

config.py

@@ -0,0 +1,92 @@
+"""AgentFrame 配置系统: 环境变量 + JSON 配置文件 + 默认值"""
+import json
+import os
+from dataclasses import dataclass, field, asdict
+
+
+@dataclass
+class LLMConfig:
+    """LLM Provider 配置"""
+    provider: str = "deepseek"              # deepseek | openai | mock
+    api_key: str = ""                       # 从环境变量 AGENTFRAME_API_KEY 或 DEEPSEEK_API_KEY
+    base_url: str = "https://api.deepseek.com/v1"
+    model: str = "deepseek-v4-pro"          # 主模型
+    fast_model: str = "deepseek-v4-flash"   # 快速模型 (检测/摘要)
+    max_tokens: int = 2000
+    temperature: float = 0.8
+    timeout: int = 120
+
+
+@dataclass
+class MemoryConfig:
+    """上下文保持核心配置"""
+    n_layers: int = 27                      # 模型层数 (DeepSeek-V2 27 层)
+    quant_bits: int = 4                     # 量化位宽 (4 = INT4, 35.6x)
+    top_k: int = 32                         # 路由检索 top-k
+    vram_limit_mb: int = 10240              # 显存层上限
+    ram_limit_mb: int = 32768               # 内存层上限
+    seed: int = 42
+    reversible: bool = False                # 可逆量化开关
+
+
+@dataclass
+class APIConfig:
+    """API 服务配置"""
+    host: str = "0.0.0.0"
+    port: int = 8090
+    debug: bool = False
+    max_history_turns: int = 12             # 对话历史保留轮数
+
+
+@dataclass
+class AgentFrameConfig:
+    """总配置"""
+    llm: LLMConfig = field(default_factory=LLMConfig)
+    memory: MemoryConfig = field(default_factory=MemoryConfig)
+    api: APIConfig = field(default_factory=APIConfig)
+
+    @classmethod
+    def from_env(cls) -> "AgentFrameConfig":
+        """从环境变量加载 (最高优先级)"""
+        cfg = cls()
+        cfg.llm.api_key = (
+            os.environ.get("AGENTFRAME_API_KEY")
+            or os.environ.get("DEEPSEEK_API_KEY")
+            or cfg.llm.api_key
+        )
+        cfg.llm.base_url = os.environ.get("AGENTFRAME_BASE_URL", cfg.llm.base_url)
+        cfg.llm.model = os.environ.get("AGENTFRAME_MODEL", cfg.llm.model)
+        cfg.llm.fast_model = os.environ.get("AGENTFRAME_FAST_MODEL", cfg.llm.fast_model)
+        cfg.api.port = int(os.environ.get("AGENTFRAME_PORT", cfg.api.port))
+        return cfg
+
+    @classmethod
+    def from_file(cls, path: str) -> "AgentFrameConfig":
+        """从 JSON 配置文件加载"""
+        with open(path) as f:
+            data = json.load(f)
+        cfg = cls.from_env()
+        # 覆盖: 文件 < 环境变量
+        if "llm" in data:
+            for k, v in data["llm"].items():
+                setattr(cfg.llm, k, v)
+        if "memory" in data:
+            for k, v in data["memory"].items():
+                setattr(cfg.memory, k, v)
+        if "api" in data:
+            for k, v in data["api"].items():
+                setattr(cfg.api, k, v)
+        return cfg
+
+    def to_dict(self) -> dict:
+        """导出为 dict (序列化用)"""
+        return {
+            "llm": asdict(self.llm),
+            "memory": asdict(self.memory),
+            "api": asdict(self.api),
+        }
+
+    def save(self, path: str):
+        """保存配置到 JSON"""
+        with open(path, "w") as f:
+            json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)

core/__init__.py

@@ -0,0 +1,14 @@
+"""核心模块"""
+from .quad import (
+    CompressedKV, HierarchicalKV, ReversibleQuantizer, AbsorbedMLA,
+    ChunkSelection, LandmarkRouter, ForgettingCurve, KVPager,
+    SubTask, RetrievalDirective, MetaCog, QuadLayerAgent,
+)
+from .engine import ContextEngine, AnswerResult
+
+__all__ = [
+    "CompressedKV", "HierarchicalKV", "ReversibleQuantizer", "AbsorbedMLA",
+    "ChunkSelection", "LandmarkRouter", "ForgettingCurve", "KVPager",
+    "SubTask", "RetrievalDirective", "MetaCog", "QuadLayerAgent",
+    "ContextEngine", "AnswerResult",
+]

core/engine.py

@@ -0,0 +1,427 @@
+"""
+ContextEngine — AgentFrame 核心引擎
+====================================
+整合四层上下文保持 + LLM + Embedding:
+
+  L1 MetaCog       认知层: 任务分解 / 信息缺口 / 检索指令
+  L2 LandmarkRouter 路由层: landmark 摘要检索 top-k
+  L3 AbsorbedMLA   存储层: 吸收式 MLA 压缩 (35.6x)
+  L4 KVPager       物理层: 三级换页 + Aura 遗忘曲线
+
+对外接口 (引擎级, 与 API/CLI 解耦):
+  ingest(text, tags)        → chunk_id
+  ask(query, chat=True)     → AnswerResult
+  stats()                   → 状态 dict
+  forget(threshold)         → 遗忘列表
+  save(path) / load(path)   → 持久化
+"""
+import json
+import numpy as np
+
+from ..config import AgentFrameConfig, MemoryConfig
+from ..llm import create_llm, LLMProvider
+from ..embed import create_embedding, EmbeddingProvider
+from ..memory.store import StateStore
+from .quad import QuadLayerAgent
+
+# DeepSeek 工具定义 (给 Agent 的"手")
+RUN_PYTHON_TOOL = [{
+    "type": "function",
+    "function": {
+        "name": "run_python",
+        "description": "执行一段 Python 代码并返回 stdout/stderr。写代码前先调用此工具验证。",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "code": {"type": "string", "description": "要执行的 Python 代码"}
+            },
+            "required": ["code"]
+        }
+    }
+}]
+
+
+class AnswerResult:
+    """ask() 的返回"""
+    def __init__(self, answer: str, retrieved: list, scores: list,
+                 context_dim: int, directive=None, meta: dict = None):
+        self.answer = answer
+        self.retrieved = retrieved      # [(chunk_id, text_preview)]
+        self.scores = scores
+        self.context_dim = context_dim
+        self.directive = directive
+        self.meta = meta or {}
+
+    def to_dict(self) -> dict:
+        return {
+            "answer": self.answer,
+            "retrieved": self.retrieved,
+            "scores": self.scores,
+            "context_dim": self.context_dim,
+            "meta": self.meta,
+        }
+
+
+MEMORY_DIRECTOR_SYSTEM = (
+    "你是记忆主管 (MemoryDirector)。分析一段对话, 只输出 JSON 记忆决策:\n"
+    "{\"remember\": [\"值得长期记住的新信息(具体、可复用)\", ...], "
+    "\"forget\": [\"已过时/无关紧要/可丢弃的信息\", ...], "
+    "\"promote_importance\": 0-1, \"reason\": \"一句话理由\"}\n"
+    "规则: remember 只收对未来有复用价值的事实(偏好/身份/参数/数据); "
+    "forget 收闲聊/一次性内容/已被替代信息; 宁可少记不可乱记; "
+    "promote_importance 表示这段对话整体重要性。"
+)
+
+
+class ContextEngine:
+    """四层上下文保持引擎"""
+
+    def __init__(self, config: AgentFrameConfig = None, llm: LLMProvider = None,
+                 embedder: EmbeddingProvider = None):
+        self.config = config or AgentFrameConfig.from_env()
+        m: MemoryConfig = self.config.memory
+
+        # 四层核心
+        self.agent = QuadLayerAgent(
+            n_layers=m.n_layers, quant_bits=m.quant_bits,
+            top_k=m.top_k, seed=m.seed, reversible=m.reversible)
+
+        # 外部能力
+        self.llm = llm or create_llm(
+            provider=self.config.llm.provider,
+            api_key=self.config.llm.api_key,
+            base_url=self.config.llm.base_url,
+            model=self.config.llm.model,
+            fast_model=self.config.llm.fast_model,
+            timeout=self.config.llm.timeout)
+        self.embedder = embedder or create_embedding("hash", dim=m.n_layers * 0 or 576)
+
+        # 会话状态
+        self.chat_history = []
+        self.max_history_turns = self.config.api.max_history_turns
+        self.now = 0.0
+        self._chunk_counter = 0
+
+    # ============ 摄入 ============
+
+    def ingest(self, text: str, tags: list = None, source: str = None) -> int:
+        """摄入知识块 → 压缩进 KV 缓存"""
+        tags = tags or []
+        # 文本 → 向量 (真实嵌入)
+        vec = self.embedder.embed(text)
+        # 四层摄取: 用向量作为 hidden states 输入
+        kv = self.agent.store.encode(vec)
+        kv.heat = 0.9
+        self.agent.pager.place(kv, self.now)
+        # 建 landmark 摘要
+        k_prime, bias = self.agent.router.build_summary(kv.latent.reshape(1, -1))
+        cid = kv.chunk_id
+        self.agent.summaries[cid] = (k_prime, bias)
+        self.agent.chunk_meta[cid] = {
+            "tags": tags, "text": text, "source": source or "",
+            "created_at": self.now,
+        }
+        self._chunk_counter += 1
+        return cid
+
+    def ingest_batch(self, chunks: list) -> list:
+        """批量摄入: [{"text":..., "tags":[...]}, ...]"""
+        return [self.ingest(c["text"], c.get("tags", []), c.get("source"))
+                for c in chunks]
+
+    # ============ 查询 ============
+
+    def _exec_tool(self, code: str, timeout: int = 15) -> str:
+        """执行 run_python 工具"""
+        import subprocess
+        import sys
+        try:
+            r = subprocess.run([sys.executable, "-c", code],
+                               capture_output=True, text=True, timeout=timeout,
+                               cwd="/tmp")
+            out = (r.stdout or "") + (("\n[stderr] " + r.stderr) if r.stderr else "")
+            if r.returncode != 0:
+                out = f"[退出码 {r.returncode}] " + out
+            return (out or "(无输出) 执行成功")[:800]
+        except subprocess.TimeoutExpired:
+            return f"[超时 >{timeout}s]"
+        except Exception as e:
+            return f"[执行异常] {e}"
+
+    def _llm_with_tools(self, system: str, user: str, max_rounds: int = 4,
+                        thinking: dict = None, use_fast: bool = False) -> str:
+        """带工具循环的 LLM 调用 (脑+手)"""
+        messages = [{"role": "system", "content": system},
+                    {"role": "user", "content": user}]
+        for _ in range(max_rounds + 1):
+            resp = self.llm.chat(messages, max_tokens=2000, temperature=0.8,
+                                 thinking=thinking, tools=RUN_PYTHON_TOOL,
+                                 use_fast=use_fast)
+            tool_calls = resp.get("tool_calls")
+            if not tool_calls:
+                return resp.get("content") or ""
+            # 执行工具并回传
+            messages.append({"role": "assistant", "content": resp.get("content") or "",
+                             "tool_calls": tool_calls})
+            for tc in tool_calls:
+                fn = tc["function"]
+                try:
+                    args = json.loads(fn.get("arguments", "{}"))
+                except Exception:
+                    args = {}
+                result = self._exec_tool(args.get("code", ""))
+                messages.append({"role": "tool", "tool_call_id": tc.get("id", ""),
+                                 "content": result})
+        return messages[-1].get("content", "") if messages else ""
+
+    def ask(self, query: str, chat: bool = True, use_fast: bool = False,
+            thinking: dict = None) -> AnswerResult:
+        """查询 + 生成"""
+        # 1. 认知层: 分解 + 检索指令
+        subtasks = self.agent.metacog.decompose(query)
+        directive = self.agent.metacog.build_directive(subtasks[0], self.agent.chunk_meta)
+
+        # 2. 路由层: landmark 检索
+        q_vec = self.embedder.embed(query)
+        selection = self.agent.router.route(q_vec, q_vec, self.agent.summaries)
+
+        # 3. 物理层: 访问更新 (遗忘曲线刷新)
+        for cid, score in zip(selection.chunk_ids, selection.scores):
+            self.agent.pager.access(cid, abs(score), self.now)
+            self.agent.metacog.promote(cid, abs(score))
+        self.now += 1
+
+        # 4. 存储层: 聚合检索结果
+        gathered, hit_ids = [], []
+        for cid in selection.chunk_ids:
+            kv = (self.agent.pager.vram.get(cid)
+                  or self.agent.pager.ram.get(cid)
+                  or self.agent.pager.disk.get(cid))
+            if kv is not None:
+                gathered.append(kv.latent)
+                meta = self.agent.chunk_meta.get(cid, {})
+                hit_ids.append((cid, meta.get("text", "")[:40]))
+        context_dim = len(gathered[0]) if gathered else 0
+
+        # 5. 生成回答
+        answer = ""
+        if chat:
+            ctx_texts = []
+            for cid in selection.chunk_ids:
+                meta = self.agent.chunk_meta.get(cid, {})
+                if meta.get("text"):
+                    ctx_texts.append(f"[知识块 {cid}] {meta['text']}")
+            ctx_block = "\n".join(ctx_texts) if ctx_texts else "(无检索到相关知识块)"
+
+            system = (
+                "你是 AgentFrame 上下文保持系统。下面是从压缩缓存中检索到的知识块，"
+                "回答时优先使用这些知识。如果知识不足就如实说明。"
+                f"\n\n【检索到的知识】\n{ctx_block}")
+            messages = [{"role": "system", "content": system}]
+            messages.extend(self.chat_history[-self.max_history_turns * 2:])
+            messages.append({"role": "user", "content": query})
+
+            resp = self.llm.chat(messages, max_tokens=self.config.llm.max_tokens,
+                                 temperature=self.config.llm.temperature,
+                                 use_fast=use_fast)
+            answer = resp.get("content") or ""
+
+            # 更新会话历史 (老对话由压缩系统兜底)
+            self.chat_history.append({"role": "user", "content": query})
+            self.chat_history.append({"role": "assistant", "content": answer})
+            if len(self.chat_history) > self.max_history_turns * 2:
+                self.chat_history = self.chat_history[-self.max_history_turns * 2:]
+
+        return AnswerResult(
+            answer=answer,
+            retrieved=hit_ids,
+            scores=[round(float(s), 4) for s in selection.scores[:len(hit_ids)]],
+            context_dim=context_dim,
+            directive=directive,
+            meta={"now": self.now, "chunks": len(self.agent.chunk_meta)})
+
+    # ============ 工具增强版 ask ============
+
+    def ask_with_hands(self, query: str, thinking: dict = None) -> AnswerResult:
+        """带工具循环的查询 (Agent 可自行执行代码验证)"""
+        result = self.ask(query, chat=False)
+        ctx_texts = []
+        for cid in result.retrieved:
+            meta = self.agent.chunk_meta.get(cid[0], {})
+            if meta.get("text"):
+                ctx_texts.append(f"[知识块 {cid[0]}] {meta['text']}")
+        ctx_block = "\n".join(ctx_texts) if ctx_texts else "(无检索到相关知识块)"
+        system = (
+            "你是 AgentFrame 上下文保持系统。下面是检索到的知识块。"
+            "回答时优先使用这些知识，知识不足就如实说明。"
+            "你有 run_python 工具可以执行代码验证你的方案。"
+            f"\n\n【检索到的知识】\n{ctx_block}")
+        answer = self._llm_with_tools(system, query, thinking=thinking)
+        return AnswerResult(answer=answer, retrieved=result.retrieved,
+                            scores=result.scores, context_dim=result.context_dim,
+                            meta={"hands": True, "chunks": len(self.agent.chunk_meta)})
+
+    # ============ 自主记忆管理 (MemoryDirector) ============
+
+    def _parse_memory_decision(self, raw: str) -> dict:
+        """解析 LLM 输出的记忆决策 JSON (容错处理)"""
+        import re
+        m = re.search(r'\{.*\}', raw, re.DOTALL)
+        if not m:
+            return {"remember": [], "forget": [], "promote_importance": 0.5, "reason": ""}
+        try:
+            return json.loads(m.group(0))
+        except Exception:
+            return {"remember": [], "forget": [], "promote_importance": 0.5, "reason": ""}
+
+    def autopilot_memory(self, query: str, answer: str) -> dict:
+        """
+        模型自主记忆管理: 分析一轮对话, 自动决定记什么/忘什么.
+        返回决策摘要.
+        """
+        conversation = f"用户: {query}\n助手: {answer[:800]}"
+        resp = self.llm.chat(
+            [{"role": "system", "content": MEMORY_DIRECTOR_SYSTEM},
+             {"role": "user", "content": f"对话:\n{conversation}"}],
+            max_tokens=600, temperature=0.3, use_fast=False)
+        decision = self._parse_memory_decision(resp.get("content") or "")
+
+        # 执行决策
+        remembered = []
+        forgotten = []
+
+        # 1. remember → 摄入新知识块 (先去重: 与已有块相似则跳过)
+        for item in decision.get("remember", []):
+            if not (isinstance(item, str) and item.strip()):
+                continue
+            # 去重: 检查是否已有高相似度块
+            dup = False
+            new_vec = self.embedder.embed(item.strip())
+            for cid, meta in self.agent.chunk_meta.items():
+                old_vec = self.embedder.embed(meta.get("text", ""))
+                sim = float(new_vec @ old_vec)
+                if sim > 0.8:  # 余弦相似度阈值
+                    dup = True
+                    break
+            if dup:
+                continue
+            cid = self.ingest(item.strip(), tags=["auto"])
+            remembered.append((cid, item.strip()))
+
+        # 2. forget → 按文本匹配降低重要性/删除
+        forget_list = decision.get("forget", [])
+        if forget_list:
+            for cid, meta in list(self.agent.chunk_meta.items()):
+                text = meta.get("text", "")
+                for f_item in forget_list:
+                    if isinstance(f_item, str) and f_item and \
+                       (f_item in text or text in f_item):
+                        kv = (self.agent.pager.vram.get(cid)
+                              or self.agent.pager.ram.get(cid)
+                              or self.agent.pager.disk.get(cid))
+                        if kv:
+                            kv.importance *= 0.3  # 大幅降权
+                            kv.heat = 0.1
+                            forgotten.append((cid, f_item))
+
+        # 3. promote_importance → 记录整体重要性
+        imp = float(decision.get("promote_importance", 0.5))
+        self._last_importance = imp
+
+        return {
+            "remembered": remembered,
+            "forgotten": forgotten,
+            "importance": imp,
+            "reason": decision.get("reason", ""),
+        }
+
+    def ask_autopilot(self, query: str) -> AnswerResult:
+        """查询 + 自主记忆管理 (一步到位)"""
+        result = self.ask(query, chat=True)
+        decision = self.autopilot_memory(query, result.answer)
+        result.meta["memory_decision"] = decision
+        return result
+
+    # ============ 遗忘 / 状态 / 持久化 ============
+
+    def forget(self, threshold: float = 0.15) -> list:
+        """按遗忘曲线清理弱记忆"""
+        victims = []
+        for layer_name in ["disk", "ram", "vram"]:
+            pool = getattr(self.agent.pager, layer_name)
+            for cid, kv in list(pool.items()):
+                strength = self.agent.pager.curve.strength(kv, self.now)
+                if strength < threshold:
+                    victims.append(cid)
+                    del pool[cid]
+                    self.agent.chunk_meta.pop(cid, None)
+                    self.agent.summaries.pop(cid, None)
+        return victims
+
+    def stats(self) -> dict:
+        return {
+            "chunks_total": len(self.agent.chunk_meta),
+            "layers": self.agent.pager.stats(),
+            "bytes_per_token": round(self.agent.store.bytes_per_token(), 2),
+            "chat_history_turns": len(self.chat_history) // 2,
+            "now": self.now,
+            "compression": "吸收式MLA + INT4 = 35.6x (L40S 实测)",
+        }
+
+    def save(self, path: str):
+        """保存引擎状态"""
+        store = StateStore(path)
+        state = {
+            "chunk_meta": self.agent.chunk_meta,
+            "summaries": self.agent.summaries,
+            "chat_history": self.chat_history,
+            "now": self.now,
+            "counter": self._chunk_counter,
+            # pager 状态
+            "vram": self.agent.pager.vram,
+            "ram": self.agent.pager.ram,
+            "disk": self.agent.pager.disk,
+        }
+        store.save(state)
+
+    def load(self, path: str):
+        """恢复引擎状态"""
+        store = StateStore(path)
+        state = store.load()
+        if not state:
+            return False
+        self.agent.chunk_meta = {int(k): v for k, v in state.get("chunk_meta", {}).items()}
+        self.agent.summaries = {int(k): v for k, v in state.get("summaries", {}).items()}
+        self.chat_history = state.get("chat_history", [])
+        self.now = state.get("now", 0.0)
+        self._chunk_counter = state.get("counter", 0)
+        # 反序列化: dict → CompressedKV
+        from .quad import CompressedKV
+        for layer_name in ["vram", "ram", "disk"]:
+            raw = state.get(layer_name, {})
+            pool = {}
+            for cid, d in raw.items():
+                if isinstance(d, CompressedKV):
+                    pool[int(cid)] = d
+                elif isinstance(d, dict):
+                    kv = CompressedKV(
+                        chunk_id=int(d.get("chunk_id", int(cid))),
+                        latent=np.asarray(d.get("latent", [0.0]), dtype=np.float32),
+                        quant_bits=int(d.get("quant_bits", 4)),
+                        heat=float(d.get("heat", 0.5)),
+                        last_access=float(d.get("last_access", 0.0)),
+                        size_bytes=int(d.get("size_bytes", 0)),
+                        importance=float(d.get("importance", 0.5)),
+                        access_count=int(d.get("access_count", 0)),
+                        protected=bool(d.get("protected", False)),
+                        reversible=bool(d.get("reversible", False)),
+                        sector_id=int(d.get("sector_id", -1)),
+                        block_id=int(d.get("block_id", -1)),
+                        module_id=int(d.get("module_id", -1)),
+                    )
+                    vh = d.get("value_hist")
+                    if isinstance(vh, np.ndarray):
+                        kv.value_hist = vh
+                    pool[int(cid)] = kv
+            setattr(self.agent.pager, layer_name, pool)
+        return True

core/quad.py

@@ -0,0 +1,690 @@
+"""
+AgentFrame 四层融合实现
+=======================
+认知层 × 路由层 × 存储层 × 物理层
+
+设计: agentframe_four_layer_blueprint.md
+实现: numpy (可 CPU 验证), 接口兼容 torch 迁移
+
+层:
+  L1 MetaCog      — 认知层: 任务分解/置信度/信息缺口
+  L2 LandmarkRouter — 路由层: landmark 摘要 + 分层软max + Q-Cal
+  L3 AbsorbedMLA  — 存储层: 吸收式 MLA + 分层量化 (35.6×)
+  L4 KVPager      — 物理层: 热/温/冷三级换页 + 预测性驱逐
+"""
+import numpy as np
+import math
+from dataclasses import dataclass, field
+from typing import Optional, Literal
+
+
+# ============================================================
+# L3 存储层: 吸收式 MLA + 分层量化 (核心, 自研)
+# ============================================================
+@dataclass
+class CompressedKV:
+    """压缩后的 KV 块 (存储层输出, 物理层/路由层操作对象)"""
+    chunk_id: int
+    latent: np.ndarray          # [576] 潜在向量 (kv_lora_rank 512 + k_pe 64)
+    quant_bits: int             # 16/8/4
+    heat: float = 0.5           # 热度 (物理层)
+    last_access: float = 0.0    # 最后访问时间
+    size_bytes: int = 0         # 实际字节数
+    importance: float = 0.5     # 重要性 (遗忘曲线用, 0-1)
+    access_count: int = 0       # 访问次数 (遗忘曲线用)
+    # ===== 自适应 Top-K 精度保护 (3-Agent 讨论室产出 + 实测迭代) =====
+    protected: bool = False                   # 是否为 Top-K 高精度块 (路由层标记)
+    reversible: bool = False                  # 启用保护机制
+    # ===== Sector-Block-Module 分层组织 (用户设计 + Pro 讨论完善) =====
+    sector_id: int = -1          # 所在区 (16 token)
+    block_id: int = -1           # 所在大块 (256 token)
+    module_id: int = -1          # 所在模组 (1024 token)
+    value_hist: Optional[np.ndarray] = None   # 块内价值直方图 (两阶段驱逐用)
+
+
+class HierarchicalKV:
+    """
+    Sector-Block-Module 分层 KV 组织 (用户设计 + Pro 讨论完善)
+    ============================================================
+    结构: 16 token(带位置) = 1 Sector
+          16 Sector      = 1 Block  (256 token)
+          4  Block       = 1 Module (1024 token)
+
+    用途 (Pro 讨论结论):
+      ✅ 存储骨架: 定位/索引/并行计算 (硬件对齐)
+      ✅ 驱逐决策: 两阶段价值分配 (每块分位数直方图 + 全局配额 + 块内排序)
+      ✅ 量化粒度: per-Sector (对齐 GPU warp)
+      ✅ 换页单元: per-Block (热/温/冷)
+    """
+    SECTOR_SIZE = 16    # token 数
+    BLOCK_SECTORS = 16  # 每 Block 的 Sector 数
+    MODULE_BLOCKS = 4   # 每 Module 的 Block 数
+    BLOCK_SIZE = SECTOR_SIZE * BLOCK_SECTORS        # 256
+    MODULE_SIZE = BLOCK_SIZE * MODULE_BLOCKS        # 1024
+
+    @staticmethod
+    def addr(token_idx: int) -> tuple:
+        """token 地址 → (module_id, block_id, sector_id, offset)"""
+        module_id = token_idx // HierarchicalKV.MODULE_SIZE
+        rem = token_idx % HierarchicalKV.MODULE_SIZE
+        block_id = rem // HierarchicalKV.BLOCK_SIZE
+        rem2 = rem % HierarchicalKV.BLOCK_SIZE
+        sector_id = rem2 // HierarchicalKV.SECTOR_SIZE
+        offset = rem2 % HierarchicalKV.SECTOR_SIZE
+        return module_id, block_id, sector_id, offset
+
+    @staticmethod
+    def addr_to_token(module_id, block_id, sector_id, offset) -> int:
+        return (module_id * HierarchicalKV.MODULE_SIZE
+                + block_id * HierarchicalKV.BLOCK_SIZE
+                + sector_id * HierarchicalKV.SECTOR_SIZE
+                + offset)
+
+    @staticmethod
+    def module_range(module_id: int) -> tuple:
+        """Module 的 token 范围"""
+        start = module_id * HierarchicalKV.MODULE_SIZE
+        return start, start + HierarchicalKV.MODULE_SIZE
+
+    @staticmethod
+    def build_value_hist(latents: list, bins: int = 8) -> np.ndarray:
+        """构建块价值直方图 (两阶段驱逐用, Pro 讨论结论)"""
+        if not latents:
+            return np.zeros(bins)
+        # 用 latent 范数作为价值代理 (可替换为注意力分数)
+        vals = np.array([np.linalg.norm(l) for l in latents])
+        hist, _ = np.histogram(vals, bins=bins, range=(0, max(vals.max(), 1e-8)))
+        return hist / max(hist.sum(), 1)
+
+    @staticmethod
+    def two_stage_evict(blocks: dict, total_keep: int = None):
+        """
+        两阶段驱逐 (Pro 终极方案):
+          1. 每块分位数直方图 → 全局价值分布代理
+          2. 按价值密度分配各块保留配额
+          3. 块内按分数排序精确选幸存
+          4. 相邻块配额微调消除边界效应
+        """
+        block_ids = list(blocks.keys())
+        if not block_ids:
+            return {}
+
+        # 1. 每块价值摘要 (直方图熵加权)
+        summaries = {}
+        for bid in block_ids:
+            h = blocks[bid].get('value_hist')
+            if h is None:
+                h = np.ones(8) / 8
+            # 价值密度 = 直方图高价值区间占比 (后50%区间加权)
+            density = float(np.sum(h[len(h)//2:]) + 0.1 * np.sum(h[:len(h)//2]))
+            summaries[bid] = density
+
+        # 2. 全局配额分配
+        total_density = sum(summaries.values()) or 1.0
+        if total_keep is None:
+            total_keep = sum(len(b.get('latents', [])) for b in blocks.values())
+        quotas = {bid: int(summaries[bid] / total_density * total_keep)
+                  for bid in block_ids}
+
+        # 3. 块内排序选幸存
+        survivors = {}
+        for bid in block_ids:
+            latents = blocks[bid].get('latents', [])
+            scores = blocks[bid].get('scores', [])
+            q = min(quotas[bid], len(latents))
+            if scores:
+                idx = np.argsort(scores)[-q:] if q > 0 else []
+                survivors[bid] = {
+                    'latents': [latents[i] for i in idx],
+                    'scores': [scores[i] for i in idx],
+                }
+            else:
+                survivors[bid] = {'latents': latents[-q:], 'scores': []}
+
+        return survivors
+
+
+class ReversibleQuantizer:
+    """
+    自适应 Top-K 精度保护 (3-Agent 讨论室产出 → 实测迭代出真方案)
+    =================================================================
+    讨论室方案: 1bit 误差符号补偿 → 实测无效 (Top-1 翻转 34%)
+    实测根因: 99.9% 翻转发生在分数差<0.1 的接近竞争, 补偿无法消除噪声
+    真解法: 路由层已知 Top-K → 对 Top-K 块保留 16bit, 其余 4bit
+    → 翻转率 0/100 (完美), 存储压缩大部分保留
+
+    核心原则: 精度预算花在"可能参与 Top-K 竞争"的块上
+    """
+    @staticmethod
+    def quantize(latent: np.ndarray, quant_bits: int, n_ch: int = 32,
+                 reversible: bool = False):
+        """
+        per-channel 非对称量化 + 可选 Top-K 保护标记
+        返回: (量化后值, 保护标记, 大小字节)
+        """
+        d = latent.shape[-1]
+        ch = d // n_ch
+        tc = latent.reshape(n_ch, ch)
+
+        if quant_bits < 16:
+            max_val = 2 ** quant_bits - 1
+            tmin = tc.min(axis=-1, keepdims=True)
+            tmax = tc.max(axis=-1, keepdims=True)
+            scale = (tmax - tmin) / max_val
+            q = np.clip(np.round((tc - tmin) / (scale + 1e-8)), 0, max_val)
+            deq = q * scale + tmin
+            bytes_per = quant_bits / 8
+        else:
+            deq = tc
+            bytes_per = 2
+
+        size = int(d * bytes_per)
+        return deq.reshape(-1), None, size
+
+    @staticmethod
+    def protect_topk(kv: CompressedKV, is_topk: bool):
+        """
+        Top-K 保护: 路由层确认该块参与 Top-K 竞争时, 标记为高精度块
+        回滚/关键推理时, 这些块用原始精度 (补偿时跳过量化误差)
+        """
+        kv.protected = is_topk
+        return kv
+
+    @staticmethod
+    def compensate(kv: CompressedKV) -> np.ndarray:
+        """
+        回滚补偿: 对受保护的 Top-K 块, 返回"需重读原始值"标记
+        实际由存储层决定: protected 块走高精度路径, 其余走量化路径
+        """
+        if kv.protected:
+            return kv.latent  # 高精度块: 量化误差可忽略
+        return kv.latent
+
+
+class AbsorbedMLA:
+    """
+    吸收式 MLA: 只缓存 576 维潜在向量, 不展开 KV
+    270KB → 30.4KB (8.9×) → INT8 15.2KB → INT4 7.6KB (35.6×)
+    """
+    KV_LORA_RANK = 512
+    K_ROPE = 64
+    DIM = KV_LORA_RANK + K_ROPE  # 576
+
+    def __init__(self, n_layers=27, quant_bits=4, n_ch=32, reversible=False):
+        self.n_layers = n_layers
+        self.quant_bits = quant_bits
+        self.n_ch = n_ch
+        self.reversible = reversible  # 1bit 可逆量化开关
+        self.chunks = {}  # chunk_id -> CompressedKV
+
+    def encode(self, hidden_states: np.ndarray) -> CompressedKV:
+        """压缩: [seq, d] → 潜在向量 [576]"""
+        # 模拟 kv_a_proj_with_mqa: 压缩到 512 + 64
+        if hidden_states.ndim == 2:
+            h = hidden_states.mean(axis=0)
+        else:
+            h = hidden_states
+        # 投影到 576 维 (实际是学习矩阵, 这里用确定性映射)
+        latent = np.tanh(h[:self.DIM] if len(h) >= self.DIM else np.pad(h, (0, self.DIM - len(h))))
+        return self._quantize(latent)
+
+    def _quantize(self, latent: np.ndarray) -> CompressedKV:
+        """per-channel 非对称量化 + Top-K 保护标记"""
+        deq, _, size = ReversibleQuantizer.quantize(
+            latent, self.quant_bits, self.n_ch, self.reversible)
+        kv = CompressedKV(
+            chunk_id=len(self.chunks),
+            latent=deq,
+            quant_bits=self.quant_bits,
+            size_bytes=size,
+            reversible=self.reversible,
+        )
+        self.chunks[kv.chunk_id] = kv
+        return kv
+
+    def protect_topk(self, chunk_id: int):
+        """路由层调用: 标记 Top-K 块为高精度保护"""
+        kv = self.chunks.get(chunk_id)
+        if kv:
+            ReversibleQuantizer.protect_topk(kv, True)
+
+    def rollback_compensate(self, chunk_id: int) -> np.ndarray:
+        """回滚补偿: 受保护块走高精度路径"""
+        kv = self.chunks.get(chunk_id)
+        if kv is None:
+            return None
+        return ReversibleQuantizer.compensate(kv)
+
+    def bytes_per_token(self) -> float:
+        """每 token 每层字节 (27 层总)"""
+        per_layer = self.DIM * (self.quant_bits / 8) if self.quant_bits < 16 else self.DIM * 2
+        return per_layer * self.n_layers
+
+    def memory_bytes(self, seq_len: int) -> int:
+        """N 层总占用"""
+        return self.bytes_per_token() * seq_len * self.n_layers
+
+
+# ============================================================
+# L2 路由层: Landmark 摘要 + 分层软max + Q-Cal
+# ============================================================
+@dataclass
+class ChunkSelection:
+    """路由层输出"""
+    chunk_ids: list
+    scores: list
+    weights: list  # 分层软max 权重
+
+
+class LandmarkRouter:
+    """
+    HiLS 思路: 块摘要 = Attn(q'c, Kc, Kc) 加权和 + 熵偏置
+    分数 = (q̂·k'c)/√d + b'c, q̂ = q + W_up W_down h (Q-Cal)
+    """
+    def __init__(self, chunk_size=64, top_k=32, d_model=576, rank=16, seed=42):
+        self.chunk_size = chunk_size
+        self.top_k = top_k
+        self.d_model = d_model
+        rng = np.random.default_rng(seed)
+        # Q-Cal 低秩校准 (仅 0.6% 参数)
+        self.W_up = rng.normal(0, 0.02, (d_model, rank)) * 0.1
+        self.W_down = rng.normal(0, 0.02, (rank, d_model)) * 0.1
+        # landmark 可学习 query (每层共享一个, 简化)
+        self.landmark_q = rng.normal(0, 0.1, d_model)
+
+    def build_summary(self, chunk_latents: np.ndarray) -> tuple:
+        """
+        块摘要: k'c = Σ p_j k_j (注意力加权和)
+                b'c = -Σ p_j log p_j (熵偏置)
+        """
+        # 用 landmark query 对块内 latent 打分
+        scores = self.landmark_q @ chunk_latents.T / np.sqrt(self.d_model)
+        p = np.exp(scores - scores.max())
+        p /= p.sum()
+        k_prime = p @ chunk_latents          # 加权和 → 块摘要
+        entropy = -(p * np.log(p + 1e-9)).sum()  # 熵偏置
+        return k_prime, entropy
+
+    def q_calibrate(self, query: np.ndarray, hidden: np.ndarray) -> np.ndarray:
+        """低秩校准: q̂ = q + W_up W_down h"""
+        delta = self.W_up @ (self.W_down @ hidden)
+        return query + delta
+
+    def route(self, query: np.ndarray, hidden: np.ndarray,
+              chunk_summaries: dict) -> ChunkSelection:
+        """
+        打分 + top-K:
+        ŝ_i,c = (q̂ᵢ·k'c)/√d + b'c
+        """
+        q_hat = self.q_calibrate(query, hidden)
+        scores = {}
+        for cid, (k_prime, bias) in chunk_summaries.items():
+            s = q_hat @ k_prime / np.sqrt(self.d_model) + bias
+            scores[cid] = s
+
+        # top-K
+        ranked = sorted(scores.items(), key=lambda x: -x[1])
+        selected = ranked[:self.top_k]
+
+        # 分层软max: 分数标准化防饱和
+        raw = np.array([s for _, s in selected])
+        # z-score 标准化 + 温度
+        if raw.std() > 1e-8:
+            norm = (raw - raw.mean()) / (raw.std() + 1e-8)
+        else:
+            norm = raw - raw.mean()
+        exp_s = np.exp(norm)
+        weights = exp_s / exp_s.sum()
+
+        return ChunkSelection(
+            chunk_ids=[cid for cid, _ in selected],
+            scores=[s for _, s in selected],
+            weights=weights.tolist(),
+        )
+
+    def gqa_group_select(self, group_scores: np.ndarray) -> list:
+        """GQA 组内 max 聚合 (任一头重要即选中)"""
+        group_max = np.max(group_scores, axis=0)
+        return np.argsort(group_max)[-self.top_k:].tolist()
+
+
+# ============================================================
+# L4 物理层: KV 换页 + 预测性驱逐
+# ============================================================
+# ============================================================
+# L4 物理层: KV 换页 + 预测性驱逐 (含遗忘曲线)
+# ============================================================
+class ForgettingCurve:
+    """
+    遗忘曲线 (源自 Project-Aura): 决定 KV 块的热度衰减
+
+    公式: S(t) = I · 2^(-t / τ)
+    其中:
+      S(t) = t 时刻的记忆强度 (热度)
+      I = 初始重要性
+      τ = 半衰期 (高频访问的块半衰期延长)
+    访问增强: boost = log2(access_count+1) × 0.1
+    """
+    def __init__(self, default_half_life: float = 100.0):
+        # KV 场景时间尺度小 (轮次而非天), 默认 100 轮
+        self.default_half_life = default_half_life
+
+    def compute_half_life(self, kv: CompressedKV) -> float:
+        """根据访问频率动态调整半衰期"""
+        base = self.default_half_life
+        if kv.access_count > 10:
+            base *= 7                           # 高频访问延至 7×
+        elif kv.access_count > 5:
+            base *= 3                           # 中频延至 3×
+        base *= (0.5 + kv.importance)           # 高重要性更持久
+        return base
+
+    def strength(self, kv: CompressedKV, now: float) -> float:
+        """记忆强度: 指数衰减 + 访问增强 (替代简单线性热度)"""
+        elapsed = max(now - kv.last_access, 0.0)
+        half_life = self.compute_half_life(kv)
+        decay = math.pow(2, -elapsed / half_life)
+        boost = math.log2(kv.access_count + 1) * 0.1
+        return min(1.0, kv.importance * decay + boost)
+
+    def should_forget(self, kv: CompressedKV, now: float, threshold: float = 0.15) -> bool:
+        """强度低于阈值 → 该换出"""
+        return self.strength(kv, now) < threshold
+
+
+class KVPager:
+    """
+    三级存储: 显存(热) / 内存(温) / 磁盘(冷, mmap)
+    预测性驱逐: 遗忘曲线强度 + 注意力衰减
+    """
+    def __init__(self, vram_limit_mb=10240, ram_limit_mb=32768):
+        self.vram_limit = vram_limit_mb * 1024 * 1024
+        self.ram_limit = ram_limit_mb * 1024 * 1024
+        self.vram = {}   # chunk_id -> CompressedKV
+        self.ram = {}
+        self.disk = {}   # 模拟 mmap
+        self.vram_used = 0
+        self.ram_used = 0
+        self.access_log = []  # (chunk_id, time, attention_score)
+        self.curve = ForgettingCurve()  # 遗忘曲线 (Aura 移植)
+
+    def place(self, kv: CompressedKV, now: float = 0.0):
+        """放置: 按遗忘曲线强度分层"""
+        strength = self.curve.strength(kv, now)
+        kv.heat = strength
+        if strength >= 0.6:
+            if self.vram_used + kv.size_bytes <= self.vram_limit:
+                self.vram[kv.chunk_id] = kv
+                self.vram_used += kv.size_bytes
+                return
+        if strength >= 0.3:
+            if self.ram_used + kv.size_bytes <= self.ram_limit:
+                self.ram[kv.chunk_id] = kv
+                self.ram_used += kv.size_bytes
+                return
+        self.disk[kv.chunk_id] = kv
+
+    def access(self, chunk_id: int, attention_score: float, now: float):
+        """访问: 更新访问计数 + 重要性 + 日志"""
+        self.access_log.append((chunk_id, now, attention_score))
+        kv = self.vram.get(chunk_id) or self.ram.get(chunk_id) or self.disk.get(chunk_id)
+        if kv:
+            kv.access_count += 1
+            kv.importance = min(1.0, kv.importance + abs(attention_score) * 0.05)
+            kv.last_access = now
+            # 重新计算强度并分层
+            strength = self.curve.strength(kv, now)
+            kv.heat = strength
+            # 磁盘 → 提升
+            if chunk_id in self.disk and strength >= 0.5:
+                self.disk.pop(chunk_id)
+                self.place(kv, now)
+
+    def eviction_score(self, kv: CompressedKV, now: float) -> float:
+        """预测性驱逐分数 (越高越该驱逐): 遗忘曲线 + 注意力衰减"""
+        # 遗忘曲线: 强度越低越该走 (主因子)
+        forget_factor = 1.0 - self.curve.strength(kv, now)
+        # 注意力衰减: 最近分数越低越该走
+        logs = [a for cid, t, a in self.access_log if cid == kv.chunk_id]
+        attn_decay = 1.0 - np.mean(logs[-5:]) if logs else 0.5
+        # 时间衰减: 越久没访问越该走
+        time_decay = min((now - kv.last_access) / self.curve.compute_half_life(kv), 1.0)
+        return 0.5 * forget_factor + 0.3 * attn_decay + 0.2 * time_decay
+
+    def evict(self, now: float, target_layer: str = "vram"):
+        """驱逐: 选分数最高的换出"""
+        pool = self.vram if target_layer == "vram" else self.ram
+        if not pool:
+            return None
+        victim_id = max(pool, key=lambda cid: self.eviction_score(pool[cid], now))
+        victim = pool.pop(victim_id)
+        if target_layer == "vram":
+            self.vram_used -= victim.size_bytes
+            self.place(victim, now)  # 降级到 RAM/DISK (按遗忘曲线自动分)
+        else:
+            self.ram_used -= victim.size_bytes
+            self.disk[victim_id] = victim
+        return victim_id
+
+    def stats(self) -> dict:
+        return {
+            "vram": f"{len(self.vram)} 块 / {self.vram_used/1024/1024:.1f}MB",
+            "ram": f"{len(self.ram)} 块 / {self.ram_used/1024/1024:.1f}MB",
+            "disk": f"{len(self.disk)} 块",
+        }
+
+
+# ============================================================
+# L1 认知层: 元认知控制器
+# ============================================================
+@dataclass
+class SubTask:
+    """子任务"""
+    id: int
+    name: str
+    info_needs: list          # 需要的信息类型
+    confidence: float = 0.0   # 完成置信度
+
+
+@dataclass
+class RetrievalDirective:
+    """认知层 → 物理层指令"""
+    required_chunks: list
+    priority: Literal["hot", "warm", "cold"]
+    reason: str
+    confidence: float
+
+
+class MetaCog:
+    """
+    认知层: 任务分解 + 置信度追踪 + 信息缺口分析 + 记忆提升
+    """
+    def __init__(self, confidence_threshold=0.7):
+        self.confidence_threshold = confidence_threshold
+        self.task_stack = []
+        self.long_term = {}      # 长期记忆 (key -> 摘要)
+        self.episodic = []       # 情景记忆
+        self._task_id = 0
+
+    def decompose(self, task: str) -> list[SubTask]:
+        """任务分解: 规则+关键词 (生产可换 LLM)"""
+        # 简单规则: 按句子/分号切分
+        parts = [p.strip() for p in task.replace('；', ';').replace('。', ';').split(';') if p.strip()]
+        subs = []
+        for p in parts:
+            self._task_id += 1
+            # 信息需求推断: 含"分析/对比/总结"等需要历史信息
+            needs = []
+            for kw, info in [("对比", "comparison"), ("分析", "method"),
+                             ("历史", "history"), ("数据", "data"),
+                             ("代码", "code"), ("总结", "result"),
+                             ("稀疏", "sparse")]:
+                if kw in p:
+                    needs.append(info)
+            if not needs:
+                needs = ["data"]  # 默认需要数据
+            subs.append(SubTask(id=self._task_id, name=p, info_needs=needs))
+        self.task_stack = subs
+        return subs
+
+    def track_confidence(self, response_likelihood: float) -> float:
+        """置信度追踪 (0-1)"""
+        return response_likelihood
+
+    def analyze_gap(self, subtask: SubTask, context_keys: set) -> list:
+        """信息缺口: 需要但上下文没有的信息"""
+        gaps = [need for need in subtask.info_needs if need not in context_keys]
+        return gaps
+
+    def build_directive(self, subtask: SubTask, chunk_map: dict) -> RetrievalDirective:
+        """生成检索指令 (认知层 → 路由/物理层)"""
+        needed = []
+        for need in subtask.info_needs:
+            for cid, meta in chunk_map.items():
+                if need in meta.get("tags", []):
+                    needed.append(cid)
+        return RetrievalDirective(
+            required_chunks=needed,
+            priority="hot" if len(needed) <= 3 else "warm",
+            reason=f"子任务 '{subtask.name}' 需要 {subtask.info_needs}",
+            confidence=subtask.confidence,
+        )
+
+    def promote(self, chunk_id: int, importance: float):
+        """记忆提升: 高频+重要 → 长期"""
+        if importance > 0.8:
+            self.long_term[chunk_id] = {"importance": importance, "promoted_at": len(self.episodic)}
+        self.episodic.append({"chunk": chunk_id, "importance": importance})
+
+
+# ============================================================
+# 四层编排 (端到端)
+# ============================================================
+class QuadLayerAgent:
+    """四层融合 Agent"""
+    def __init__(self, n_layers=27, quant_bits=4, top_k=32, seed=42, reversible=False):
+        self.metacog = MetaCog()
+        self.router = LandmarkRouter(top_k=top_k, seed=seed)
+        self.store = AbsorbedMLA(n_layers=n_layers, quant_bits=quant_bits, reversible=reversible)
+        self.pager = KVPager()
+        self.chunk_meta = {}  # chunk_id -> {tags, ...}
+        self.summaries = {}   # chunk_id -> (k_prime, bias)
+        self.now = 0.0
+
+    def ingest(self, text_chunks: list[dict]):
+        """
+        摄取知识: text_chunks = [{"text": "...", "tags": ["data"]}, ...]
+        每块 → 压缩为 CompressedKV → 建摘要 → 放置
+        """
+        for i, chunk in enumerate(text_chunks):
+            # 模拟 hidden states
+            h = np.random.default_rng(i).normal(0, 1, self.store.DIM)
+            kv = self.store.encode(h)
+            kv.heat = 0.9  # 新知识 = 热
+            self.pager.place(kv, self.now)
+            # 建 landmark 摘要
+            k_prime, bias = self.router.build_summary(kv.latent.reshape(1, -1))
+            self.summaries[kv.chunk_id] = (k_prime, bias)
+            self.chunk_meta[kv.chunk_id] = {"tags": chunk.get("tags", []), "text": chunk.get("text", "")}
+
+    def run_task(self, task: str, query: np.ndarray):
+        """执行任务: 认知 → 路由 → 存储 → 输出"""
+        print(f"\n{'='*56}")
+        print(f"任务: {task}")
+        print(f"{'='*56}")
+
+        # L1 认知: 分解
+        subtasks = self.metacog.decompose(task)
+        print(f"[认知层] 分解为 {len(subtasks)} 个子任务")
+
+        # L1 认知: 信息缺口 → 检索指令
+        directive = self.metacog.build_directive(subtasks[0], self.chunk_meta)
+        print(f"[认知层] 检索指令: 需要 {directive.required_chunks} ({directive.reason})")
+
+        # L4 物理: 预取
+        for cid in directive.required_chunks:
+            if cid in self.pager.disk:
+                kv = self.pager.disk.pop(cid)
+                kv.heat = 0.6
+                self.pager.place(kv, self.now)
+        print(f"[物理层] 预取完成 → {self.pager.stats()}")
+
+        # L2 路由: landmark 检索
+        selection = self.router.route(query, query, self.summaries)
+        print(f"[路由层] top-{len(selection.chunk_ids)} 块: {selection.chunk_ids[:5]}...")
+        print(f"[路由层] 权重分布: {[f'{w:.2f}' for w in selection.weights[:5]]}...")
+
+        # L4 物理: 访问更新
+        for cid, score in zip(selection.chunk_ids, selection.scores):
+            self.pager.access(cid, abs(score), self.now)
+            self.metacog.promote(cid, abs(score))
+
+        # L3 存储: 汇总被选中块的压缩表示
+        gathered = []
+        for cid in selection.chunk_ids:
+            kv = (self.pager.vram.get(cid) or self.pager.ram.get(cid) or self.pager.disk.get(cid))
+            if kv:
+                gathered.append(kv.latent)
+        context = np.mean(gathered, axis=0) if gathered else np.zeros(self.store.DIM)
+        print(f"[存储层] 聚合 {len(gathered)} 块压缩表示 (每块 {self.store.bytes_per_token():.1f}B)")
+
+        # 置信度
+        conf = self.metacog.track_confidence(0.85)
+        print(f"[认知层] 置信度: {conf:.2f}")
+
+        # 驱逐维护
+        if self.pager.vram_used > self.pager.vram_limit * 0.9:
+            victim = self.pager.evict(self.now, "vram")
+            print(f"[物理层] 驱逐 {victim} → 降级")
+
+        self.now += 1
+        return context, selection
+
+
+# ============================================================
+# 验证
+# ============================================================
+if __name__ == "__main__":
+    print("=" * 56)
+    print("AgentFrame 四层融合 — 端到端验证")
+    print("=" * 56)
+
+    # 创建 Agent
+    agent = QuadLayerAgent(n_layers=27, quant_bits=4, top_k=8)
+
+    # 注入知识库 (12 块, 带标签)
+    knowledge = [
+        {"text": "KV 缓存 270KB/token 展开存储", "tags": ["data", "comparison"]},
+        {"text": "吸收式 MLA 缓存 576 维潜在向量", "tags": ["data", "method"]},
+        {"text": "INT8 思考链误差 0.011", "tags": ["data", "comparison"]},
+        {"text": "INT4 工具结果误差 0.079", "tags": ["data"]},
+        {"text": "per-channel 非对称量化 27 倍精度", "tags": ["method"]},
+        {"text": "L40S 实测 138 万 token 上下文", "tags": ["data", "result", "comparison"]},
+        {"text": "HiLS 分层软max 端到端块选择", "tags": ["method", "sparse"]},
+        {"text": "landmark token 块摘要检索", "tags": ["method", "sparse"]},
+        {"text": "Q-Cal 低秩校准 0.6% 参数", "tags": ["method"]},
+        {"text": "GQA 组内 max 聚合", "tags": ["method", "sparse"]},
+        {"text": "KV 换页 mmap 冷存储", "tags": ["method", "paging"]},
+        {"text": "预测性驱逐注意力信号", "tags": ["method", "paging"]},
+    ]
+    agent.ingest(knowledge)
+    print(f"[存储层] 注入 {len(knowledge)} 块知识")
+    print(f"[存储层] 每 token 缓存: {agent.store.bytes_per_token():.1f}B (原 270KB → 35.6×)")
+    print(f"[物理层] 初始: {agent.pager.stats()}")
+
+    # 执行任务
+    rng = np.random.default_rng(0)
+    query = rng.normal(0, 1, 576)
+
+    agent.run_task("对比压缩方案；分析稀疏路由；总结实测数据", query)
+
+    # 容量对比
+    seq_len = 100_000
+    original = 270 * 1024 * seq_len * 27
+    compressed = agent.store.memory_bytes(seq_len)
+    print(f"\n{'='*56}")
+    print(f"容量对比 (100K token × 27 层):")
+    print(f"  原始展开: {original/1024**3:.1f} GB")
+    print(f"  四层后:   {compressed/1024**3:.3f} GB")
+    print(f"  压缩比:   {original/compressed:.1f}×")
+    print(f"{'='*56}")

embed/__init__.py

@@ -0,0 +1,4 @@
+"""Embedding 模块"""
+from .provider import EmbeddingProvider, HashEmbedding, ApiEmbedding, create_embedding
+
+__all__ = ["EmbeddingProvider", "HashEmbedding", "ApiEmbedding", "create_embedding"]

embed/provider.py

@@ -0,0 +1,112 @@
+"""Embedding Provider: 文本 → 向量 (可插拔)"""
+import hashlib
+import numpy as np
+from abc import ABC, abstractmethod
+
+
+class EmbeddingProvider(ABC):
+    """文本嵌入抽象接口"""
+
+    @abstractmethod
+    def embed(self, text: str) -> np.ndarray:
+        """文本 → 向量"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def dim(self) -> int:
+        raise NotImplementedError
+
+
+class HashEmbedding(EmbeddingProvider):
+    """
+    本地哈希嵌入 (零依赖, 默认实现)
+    =================================
+    基于字符 n-gram 特征哈希: 确定性、可复现、无语义但可区分相似文本.
+    生产环境可替换为真实 embedding API (见 ApiEmbedding).
+
+    特点:
+      - 相同文本 → 相同向量
+      - 相似文本 (共享 n-gram) → 相似向量
+      - 维度固定 (默认 576, 与 MLA latent 对齐)
+    """
+
+    def __init__(self, dim: int = 576, ngram: int = 3, seed: int = 42):
+        self._dim = dim
+        self.ngram = ngram
+        rng = np.random.default_rng(seed)
+        # 每维一个随机投影方向, 用于特征哈希
+        self._proj = rng.normal(0, 1, (dim,)).astype(np.float32)
+
+    def dim(self) -> int:
+        return self._dim
+
+    def _ngrams(self, text: str) -> list:
+        """提取字符 n-gram (含 unicode 感知)"""
+        if not text:
+            return []
+        grams = []
+        for i in range(len(text) - self.ngram + 1):
+            grams.append(text[i:i + self.ngram])
+        return grams
+
+    def embed(self, text: str) -> np.ndarray:
+        vec = np.zeros(self._dim, dtype=np.float32)
+        grams = self._ngrams(text)
+        if not grams:
+            return vec
+        for g in grams:
+            # 特征哈希: 确定性映射到维度 + 符号
+            h = int(hashlib.md5(g.encode("utf-8")).hexdigest(), 16)
+            idx = h % self._dim
+            sign = 1.0 if (h >> 32) % 2 == 0 else -1.0
+            vec[idx] += sign
+        # 归一化 (单位向量, 余弦相似度)
+        norm = np.linalg.norm(vec)
+        if norm > 1e-8:
+            vec /= norm
+        return vec
+
+
+class ApiEmbedding(EmbeddingProvider):
+    """
+    远程 API 嵌入 (OpenAI 兼容 /embeddings 端点)
+    需配置: base_url + api_key + model
+    """
+
+    def __init__(self, base_url: str, api_key: str,
+                 model: str = "text-embedding-3-small", dim: int = 1536):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.model = model
+        self._dim = dim
+
+    def dim(self) -> int:
+        return self._dim
+
+    def embed(self, text: str) -> np.ndarray:
+        import json
+        import urllib.request
+        data = json.dumps({
+            "model": self.model,
+            "input": text,
+        }).encode()
+        req = urllib.request.Request(
+            f"{self.base_url}/embeddings", data=data, headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {self.api_key}",
+            })
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                r = json.loads(resp.read())
+            emb = r["data"][0]["embedding"]
+            return np.asarray(emb, dtype=np.float32)
+        except Exception:
+            # API 失败时回退哈希嵌入
+            return HashEmbedding(self._dim).embed(text)
+
+
+def create_embedding(kind: str = "hash", **kwargs) -> EmbeddingProvider:
+    """工厂: hash | api"""
+    if kind == "api":
+        return ApiEmbedding(**kwargs)
+    return HashEmbedding(**kwargs)

llm/__init__.py

@@ -0,0 +1,6 @@
+"""LLM 模块"""
+from .base import LLMProvider, create_llm
+from .deepseek import DeepSeekProvider
+from .mock import MockProvider
+
+__all__ = ["LLMProvider", "create_llm", "DeepSeekProvider", "MockProvider"]

llm/base.py

@@ -0,0 +1,34 @@
+"""LLM Provider: 抽象基类 + 工厂"""
+from abc import ABC, abstractmethod
+from typing import Optional
+
+
+class LLMProvider(ABC):
+    """LLM Provider 抽象接口"""
+
+    @abstractmethod
+    def chat(self, messages: list, max_tokens: int = 2000,
+             temperature: float = 0.8, thinking: Optional[dict] = None,
+             tools: Optional[list] = None) -> dict:
+        """
+        对话生成.
+        返回: {"content": str, "tool_calls": [...]|None}
+        """
+        raise NotImplementedError
+
+    @abstractmethod
+    def name(self) -> str:
+        raise NotImplementedError
+
+
+def create_llm(provider: str, api_key: str, base_url: str,
+               model: str, fast_model: str, **kwargs) -> LLMProvider:
+    """工厂: 按名称创建 Provider"""
+    if provider == "deepseek":
+        from .deepseek import DeepSeekProvider
+        return DeepSeekProvider(api_key=api_key, base_url=base_url,
+                                model=model, fast_model=fast_model)
+    if provider == "mock":
+        from .mock import MockProvider
+        return MockProvider()
+    raise ValueError(f"未知 LLM Provider: {provider}")

llm/deepseek.py

@@ -0,0 +1,59 @@
+"""DeepSeek LLM Provider (支持 thinking + function calling)"""
+import json
+import urllib.request
+from typing import Optional
+
+from .base import LLMProvider
+
+
+class DeepSeekProvider(LLMProvider):
+    """DeepSeek API Provider"""
+
+    def __init__(self, api_key: str, base_url: str = "https://api.deepseek.com/v1",
+                 model: str = "deepseek-v4-pro",
+                 fast_model: str = "deepseek-v4-flash",
+                 timeout: int = 120):
+        self.api_key = api_key
+        self.base_url = base_url.rstrip("/")
+        self.model = model
+        self.fast_model = fast_model
+        self.timeout = timeout
+
+    def name(self) -> str:
+        return "deepseek"
+
+    def chat(self, messages: list, max_tokens: int = 2000,
+             temperature: float = 0.8, thinking: Optional[dict] = None,
+             tools: Optional[list] = None,
+             use_fast: bool = False) -> dict:
+        """调用 DeepSeek API, 支持思考强度与工具调用"""
+        model = self.fast_model if use_fast else self.model
+        payload = {
+            "model": model,
+            "messages": messages,
+            "max_tokens": max_tokens,
+            "temperature": temperature,
+        }
+        if thinking is not None:
+            payload["thinking"] = thinking
+        if tools:
+            payload["tools"] = tools
+            payload["tool_choice"] = "auto"
+
+        data = json.dumps(payload).encode()
+        req = urllib.request.Request(
+            f"{self.base_url}/chat/completions", data=data, headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {self.api_key}",
+            })
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                r = json.loads(resp.read())
+            msg = r["choices"][0]["message"]
+            return {
+                "content": (msg.get("content") or "").strip(),
+                "tool_calls": msg.get("tool_calls"),
+                "finish_reason": r["choices"][0].get("finish_reason"),
+            }
+        except Exception as e:
+            return {"content": f"[API错误: {e}]", "tool_calls": None, "finish_reason": "error"}

llm/mock.py

@@ -0,0 +1,24 @@
+"""Mock LLM Provider: 无 API 时的离线回退 (测试/演示用)"""
+from typing import Optional
+
+from .base import LLMProvider
+
+
+class MockProvider(LLMProvider):
+    """离线回退 Provider: 返回固定结构, 便于无网络测试"""
+
+    def name(self) -> str:
+        return "mock"
+
+    def chat(self, messages: list, max_tokens: int = 2000,
+             temperature: float = 0.8, thinking: Optional[dict] = None,
+             tools: Optional[list] = None,
+             use_fast: bool = False) -> dict:
+        # 提取最后一条用户消息
+        user_msgs = [m for m in messages if m.get("role") == "user"]
+        last = user_msgs[-1]["content"][:100] if user_msgs else ""
+        return {
+            "content": f"[Mock] 已收到: {last}",
+            "tool_calls": None,
+            "finish_reason": "stop",
+        }

memory/__init__.py

@@ -0,0 +1,4 @@
+"""记忆模块"""
+from .store import StateStore
+
+__all__ = ["StateStore"]

memory/store.py

@@ -0,0 +1,53 @@
+"""持久化: 状态保存/加载 (JSON 快照)"""
+import json
+import os
+import numpy as np
+
+
+def _serialize(obj):
+    """递归序列化 (处理 numpy 类型)"""
+    if isinstance(obj, np.ndarray):
+        return {"__ndarray__": True, "shape": list(obj.shape),
+                "dtype": str(obj.dtype), "data": obj.tobytes().hex()}
+    if isinstance(obj, np.generic):
+        return obj.item()
+    if isinstance(obj, dict):
+        return {str(k): _serialize(v) for k, v in obj.items()}
+    if isinstance(obj, (list, tuple)):
+        return [_serialize(v) for v in obj]
+    if hasattr(obj, "__dict__"):
+        return _serialize(obj.__dict__)
+    return obj
+
+
+def _deserialize(obj):
+    """递归反序列化"""
+    if isinstance(obj, dict):
+        if obj.get("__ndarray__"):
+            arr = np.frombuffer(bytes.fromhex(obj["data"]), dtype=obj["dtype"])
+            return arr.reshape(obj["shape"])
+        return {k: _deserialize(v) for k, v in obj.items()}
+    if isinstance(obj, list):
+        return [_deserialize(v) for v in obj]
+    return obj
+
+
+class StateStore:
+    """JSON 快照存储: 保存引擎状态到文件, 可恢复"""
+
+    def __init__(self, path: str):
+        self.path = path
+
+    def save(self, state: dict):
+        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
+        with open(self.path, "w") as f:
+            json.dump(_serialize(state), f, ensure_ascii=False)
+
+    def load(self) -> dict:
+        if not os.path.exists(self.path):
+            return {}
+        with open(self.path) as f:
+            return _deserialize(json.load(f))
+
+    def exists(self) -> bool:
+        return os.path.exists(self.path)

pyproject.toml

@@ -0,0 +1,28 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "agentframe"
+version = "1.0.0"
+description = "Agent 专用 DeepSeek 推理框架 — 脑=LLM + 手=工具 + 四层上下文保持 (35.6x KV 压缩)"
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "GPL-3.0" }
+authors = [{ name = "Cloud LTE Studio" }]
+dependencies = [
+    "numpy>=1.24",
+    "flask>=3.0",
+]
+
+[project.optional-dependencies]
+api = ["flask>=3.0"]
+dev = ["pytest"]
+
+[project.scripts]
+agentframe = "agentframe.cli:main"
+agentframe-api = "agentframe.api.server:main"
+
+[tool.setuptools]
+packages = ["agentframe", "agentframe.core", "agentframe.llm",
+            "agentframe.embed", "agentframe.memory", "agentframe.api"]