onlyonepice

---
name: openviking-context
description: "OpenViking context database for AI agents — layered context loading (L0/L1/L2), semantic search, file-system memory management. Use when setting up OpenViking, managing agent memory/resources, performing semantic search, browsing context filesystem, or comparing token consumption. Triggers on: 'openviking', 'context database', 'viking memory', 'layered context', 'token saving', 'L0/L1/L2', 'viking://', 'memsearch', 'memread', 'context setup'."
---

# OpenViking Context Database

字节跳动开源的 Agent 上下文数据库。通过 `viking://` 文件系统协议统一管理记忆、资源和技能，L0/L1/L2 三层按需加载，token 消耗降低 83-96%。

| 能力 | 说明 |
|---|---|
| 文件系统协议 | `viking://` 统一管理 resources/user/agent 三类上下文 |
| L0/L1/L2 分层 | 摘要(~100 tokens) / 概览(~2k tokens) / 全文，按需加载 |
| 语义检索 | 目录递归检索 + 向量匹配 |
| 会话记忆 | 自动提取长期记忆，跨会话保持 |
| Token 节省 | 对比全量加载，输入 token 降低 83%~96% |

## 安装到 OpenClaw

```bash
bash scripts/install-skill.sh
```

脚本会将 skill 复制到 OpenClaw 的 skills 目录（自动检测路径），然后在 OpenClaw 中说 "refresh skills" 即可发现。

## 安装 OpenViking 依赖

skill 安装完成后，运行以下命令安装 OpenViking 本体：

```bash
bash scripts/install.sh
```

自动检测 Python >= 3.10，安装 `openviking` 包，创建工作目录，可选安装 Rust CLI (`ov`)。

## 配置

```bash
bash scripts/setup-config.sh
```

支持的模型提供商：

| 提供商 | VLM 模型 | Embedding 模型 |
|---|---|---|
| `openai` | gpt-4o | text-embedding-3-large (dim=3072) |
| `volcengine` | doubao-seed-2-0-pro-260215 | doubao-embedding-vision-250615 (dim=1024) |
| `litellm` | claude-3-5-sonnet / deepseek-chat | — |
| NVIDIA NIM | meta/llama-3.3-70b-instruct | nvidia/nv-embed-v1 (dim=4096) |

> **注意**：避免使用推理模型 (kimi-k2.5, deepseek-r1)，它们的 `reasoning` 字段与 OpenViking 不兼容。

## 启动服务器

```bash
openviking-server
# 或后台运行：
nohup openviking-server > ~/.openviking/server.log 2>&1 &
```

## 核心操作

通过 `scripts/viking.py` 与 OpenViking 交互：

```bash
python3 scripts/viking.py <command> [args]
```

| 命令 | 功能 | 示例 |
|---|---|---|
| `add <path_or_url>` | 添加资源（文件/URL/目录） | `viking.py add ./docs/` |
| `search <query>` | 语义搜索 | `viking.py search "认证逻辑"` |
| `ls [uri]` | 浏览资源目录 | `viking.py ls viking://resources/` |
| `tree [uri]` | 树形展示 | `viking.py tree viking://resources/ -L 2` |
| `abstract <uri>` | L0 摘要 (~100 tokens) | `viking.py abstract viking://resources/proj` |
| `overview <uri>` | L1 概览 (~2k tokens) | `viking.py overview viking://resources/proj` |
| `read <uri>` | L2 全文 | `viking.py read viking://resources/proj/api.md` |
| `info` | 检查服务状态 | `viking.py info` |
| `commit` | 提取当前会话记忆 | `viking.py commit` |
| `stats` | 查看 token 消耗统计 | `viking.py stats` |
| `stats --reset` | 重置统计数据 | `viking.py stats --reset` |

## Token 消耗追踪

每次调用 `search`、`abstract`、`overview`、`read` 时自动追踪：

- **实际消耗**：本次分层加载实际使用的 token 数
- **全量假设**：如果用传统方式全量加载同一资源需要的 token 数
- **节省量**：两者差值和百分比

每次命令结尾自动输出一行会话累计摘要：

```
📊 会话累计 | 实际: 2,300 tokens | 全量: 48,000 tokens | 节省: 45,700 (95.2%)
```

使用 `stats` 命令查看完整的逐操作明细表：

```bash
python3 scripts/viking.py stats
```

输出示例：

```
═══ Token 消耗统计 ═══
  会话开始: 2026-03-19 19:30:00
  操作次数: 4

  #    时间       操作       层级  实际     全量     节省     URI
  ──── ────────── ────────── ───── ──────── ──────── ──────── ──────────────────
  1    19:30:05   search     L0        300   48,000   47,700  用户认证 鉴权
  2    19:30:12   overview   L1      1,800   15,000   13,200  viking://resources/auth
  3    19:30:18   abstract   L0         80    8,000    7,920  viking://resources/db
  4    19:30:25   read       L2      3,200    3,200        0  viking://resources/auth/jwt

  ┌─────────────────────────────────────┐
  │  全量加载 (传统方式):     74,200 tokens │
  │  实际消耗 (分层加载):      5,380 tokens │
  │  节省 token 数量:        68,820 tokens │
  │  节省比例:                  92.8%        │
  └─────────────────────────────────────┘
```

统计数据持久化在 `~/.openviking/session_stats.json`，跨命令调用累积。新会话可用 `stats --reset` 重置。

## 分层加载工作流

收到开发需求（如"帮我写一个用户认证模块"）时：

**Step 1 — L0 快速扫描**（~300 tokens）

```bash
python3 scripts/viking.py search "用户认证 鉴权 登录"
```

用 L0 摘要判断哪些资源相关，过滤无关内容。

**Step 2 — L1 概览决策**（~2k tokens/资源）

```bash
python3 scripts/viking.py overview viking://resources/auth-docs
```

理解架构和技术选型，制定实现计划。

**Step 3 — L2 按需深读**（仅必要文件）

```bash
python3 scripts/viking.py read viking://resources/auth-docs/jwt-config.md
```

只加载写代码需要的具体文件。

### Token 对比演示

```bash
python3 scripts/demo-token-compare.py ./your-project-docs/
```

| 方案 | Token 消耗 | 说明 |
|---|---|---|
| 全量加载 (传统 RAG) | ~50,000 | 所有文档塞进 prompt |
| L0 扫描 + L1 概览 | ~3,000 | 分层按需，仅摘要和概览 |
| L0 + L1 + L2 按需 | ~8,000 | 最终只深读 2-3 个必要文件 |
| **节省比例** | **84%~94%** | 相比全量加载 |

## 故障排查

| 问题 | 原因 | 解决 |
|---|---|---|
| `Dense vector dimension mismatch` | embedding 维度配置错误 | 检查 ov.conf 中的 dimension 与模型匹配 |
| `NoneType is not subscriptable` | 使用了推理模型 | 换用 gpt-4o 或 llama-3.3-70b |
| `input_type required` | 使用了非对称 embedding | 换用对称模型如 nvidia/nv-embed-v1 |
| 搜索无结果 | 语义处理未完成 | 添加资源后等待：`viking.py add --wait` |
| 服务连接失败 | 服务器未启动 | 运行 `openviking-server` |

## 参考

- [OpenViking GitHub](https://github.com/volcengine/OpenViking)
- [OpenViking 官网](https://www.openviking.ai)
- [LiteLLM 提供商文档](https://docs.litellm.ai/docs/providers)
- [NVIDIA NIM API](https://build.nvidia.com/)

FILE:clawhub.json
{
  "name": "openviking-context",
  "tagline": "L0/L1/L2 分层上下文加载，token 消耗降低 83-96%",
  "description": "OpenViking 上下文数据库 Skill — 字节跳动开源的 Agent 上下文管理方案。通过 viking:// 文件系统协议统一管理记忆、资源和技能，利用 L0(摘要)/L1(概览)/L2(全文) 三层按需加载显著降低 token 消耗。内置会话级 token 追踪，每次操作自动对比分层加载 vs 传统全量加载的节省效果。",
  "version": "1.0.0",
  "license": "MIT",
  "category": "Developer Tools",
  "tags": ["rag", "context-database", "vector-search", "token-optimization", "memory", "semantic-search", "knowledge-base"],
  "author": {
    "name": "guoweisheng"
  },
  "pricing": "free",
  "requirements": {
    "python": ">=3.10",
    "packages": ["openviking"]
  },
  "triggers": [
    "openviking",
    "context database",
    "viking memory",
    "layered context",
    "token saving",
    "L0/L1/L2",
    "viking://",
    "memsearch",
    "memread",
    "context setup"
  ]
}

FILE:README.md
# OpenViking Context Database — OpenClaw Skill

L0/L1/L2 分层上下文加载，token 消耗降低 83-96%。

## 这个 Skill 做什么

基于字节跳动开源的 [OpenViking](https://github.com/volcengine/OpenViking) 上下文数据库，为 OpenClaw Agent 提供：

- **分层加载** — L0 摘要(~100 tokens) → L1 概览(~2k tokens) → L2 全文，按需逐层深入
- **语义检索** — 向量匹配 + 目录递归，精准定位相关上下文
- **会话记忆** — 自动提取长期记忆，Agent 越用越聪明
- **Token 追踪** — 每次操作实时展示节省了多少 token

## 安装

### 方式 1：从 ClawHub 安装

```bash
clawhub install openviking-context
```

### 方式 2：从源码安装

```bash
git clone https://github.com/guoweisheng/openviking-skill.git
cd openviking-skill
bash scripts/install-skill.sh
```

安装脚本自动检测本机 npm 安装 / Docker 安装路径。也可手动指定：

```bash
OPENCLAW_SKILLS_DIR=/your/path bash scripts/install-skill.sh
```

### 安装 OpenViking 依赖

```bash
bash scripts/install.sh      # 安装 openviking Python 包
bash scripts/setup-config.sh  # 交互式配置 API Key 和模型
openviking-server             # 启动服务
```

## 环境要求

| 依赖 | 版本 |
|---|---|
| Python | >= 3.10 |
| openviking (pip) | latest |
| OpenClaw | 2026.3+ |

需要配置一个模型提供商的 API Key（支持 OpenAI / 火山引擎 / NVIDIA NIM / LiteLLM）。

## 使用方式

在 OpenClaw 中直接对话即可触发，例如：

- "帮我用 openviking 搜索认证相关的文档"
- "查看 openviking 的 token 消耗统计"
- "把这个项目的文档导入 openviking"

也可以通过命令行使用：

```bash
# 添加资源
python3 scripts/viking.py add ./my-project-docs/

# 语义搜索（L0 摘要级别）
python3 scripts/viking.py search "用户认证"

# L1 概览
python3 scripts/viking.py overview viking://resources/auth-docs

# L2 全文（仅在需要时）
python3 scripts/viking.py read viking://resources/auth-docs/jwt-config.md

# 查看 token 消耗统计
python3 scripts/viking.py stats
```

## Token 节省效果

每次操作自动追踪并输出：

```
📊 会话累计 | 实际: 2,300 tokens | 全量: 48,000 tokens | 节省: 45,700 (95.2%)
```

`stats` 命令展示完整明细：

```
═══ Token 消耗统计 ═══
  ┌─────────────────────────────────────┐
  │  全量加载 (传统方式):     74,200 tokens │
  │  实际消耗 (分层加载):      5,380 tokens │
  │  节省 token 数量:        68,820 tokens │
  │  节省比例:                  92.8%        │
  └─────────────────────────────────────┘
```

## 目录结构

```
openviking-context/
├── SKILL.md                  # OpenClaw skill 入口
├── clawhub.json              # ClawHub 元数据
├── README.md                 # 本文件
└── scripts/
    ├── install-skill.sh      # 一键安装到 OpenClaw（本机/Docker）
    ├── install.sh            # 安装 OpenViking 依赖
    ├── setup-config.sh       # 交互式配置向导
    ├── viking.py             # CLI wrapper + token 追踪
    └── demo-token-compare.py # Token 对比演示
```

## 参考

- [OpenViking GitHub](https://github.com/volcengine/OpenViking)
- [OpenViking 官网](https://www.openviking.ai)
- [OpenClaw 技能文档](https://docs.openclaw.ai/tools/creating-skills)
- [ClawHub](https://clawhub.ai)

FILE:scripts/install-skill.sh
#!/usr/bin/env bash
set -euo pipefail

BOLD='\033[1m'
GREEN='\033[0;32m'
YELLOW='\033[0;33m'
RED='\033[0;31m'
CYAN='\033[0;36m'
NC='\033[0m'

SKILL_NAME="openviking-context"
SOURCE_DIR="$(cd "$(dirname "$0")/.." && pwd)"

echo -e "BOLD══════════════════════════════════════════════NC"
echo -e "BOLD  OpenViking Skill → OpenClaw 安装NC"
echo -e "BOLD══════════════════════════════════════════════NC"
echo ""

# ─── 检测 OpenClaw skills 目录 ───────────────────────────────────────

detect_skills_dir() {
    # 优先使用用户指定的环境变量
    if [ -n "-" ]; then
        echo "$OPENCLAW_SKILLS_DIR"
        return
    fi

    # 1) 本机安装：homebrew (Apple Silicon)
    local d="/opt/homebrew/lib/node_modules/openclaw/skills"
    [ -d "$d" ] && echo "$d" && return

    # 2) 本机安装：homebrew (Intel Mac) / Linux npm global
    d="/usr/local/lib/node_modules/openclaw/skills"
    [ -d "$d" ] && echo "$d" && return

    # 3) 本机安装：npm root -g
    local npm_root
    npm_root="$(npm root -g 2>/dev/null || true)"
    if [ -n "$npm_root" ] && [ -d "$npm_root/openclaw/skills" ]; then
        echo "$npm_root/openclaw/skills"
        return
    fi

    # 4) Docker：当前目录下的 skills/ (docker-compose 挂载)
    if [ -d "./skills" ] && [ -f "./docker-compose.yml" ] || [ -f "./docker-compose.yaml" ]; then
        echo "$(pwd)/skills"
        return
    fi

    # 5) Docker：常见 docker-compose 项目位置
    for p in "$HOME/openclaw/skills" "$HOME/.openclaw/skills" "$HOME/docker/openclaw/skills"; do
        [ -d "$p" ] && echo "$p" && return
    done

    # 未找到
    echo ""
}

TARGET_BASE=$(detect_skills_dir)

if [ -z "$TARGET_BASE" ]; then
    echo -e "YELLOW[!] 未自动检测到 OpenClaw skills 目录NC"
    echo ""
    echo "  请选择你的安装方式："
    echo ""
    echo "  1) 本机安装 (npm global) — 手动输入路径"
    echo "  2) Docker 安装 — 输入 docker-compose 项目中的 skills 目录"
    echo "  3) 自定义路径"
    echo ""
    echo -en "CYAN  请选择 [1-3]: NC"
    read -r CHOICE

    case "$CHOICE" in
        1)
            echo -en "CYAN  npm global skills 路径: NC"
            read -r TARGET_BASE
            ;;
        2)
            echo -en "CYAN  docker-compose 项目中的 skills 目录路径: NC"
            read -r TARGET_BASE
            ;;
        3)
            echo -en "CYAN  自定义路径: NC"
            read -r TARGET_BASE
            ;;
        *)
            echo -e "RED[✗] 无效选择NC"
            exit 1
            ;;
    esac

    if [ -z "$TARGET_BASE" ]; then
        echo -e "RED[✗] 未提供路径NC"
        exit 1
    fi
fi

TARGET_DIR="$TARGET_BASE/$SKILL_NAME"

echo -e "  检测到 skills 目录: BOLD$TARGET_BASENC"
echo -e "  安装到: BOLD$TARGET_DIRNC"
echo ""

# ─── 安装 ────────────────────────────────────────────────────────────

if [ -d "$TARGET_DIR" ]; then
    echo -e "YELLOW[!] 已存在: $TARGET_DIRNC"
    echo -en "    覆盖安装? [y/N]: "
    read -r OVERWRITE
    if [[ ! "$OVERWRITE" =~ ^[Yy]$ ]]; then
        echo -e "RED[✗] 已取消NC"
        exit 0
    fi
    rm -rf "$TARGET_DIR"
fi

mkdir -p "$TARGET_DIR/scripts"

cp "$SOURCE_DIR/SKILL.md" "$TARGET_DIR/SKILL.md"
cp "$SOURCE_DIR/scripts/install.sh" "$TARGET_DIR/scripts/install.sh"
cp "$SOURCE_DIR/scripts/setup-config.sh" "$TARGET_DIR/scripts/setup-config.sh"
cp "$SOURCE_DIR/scripts/viking.py" "$TARGET_DIR/scripts/viking.py"
cp "$SOURCE_DIR/scripts/demo-token-compare.py" "$TARGET_DIR/scripts/demo-token-compare.py"

chmod +x "$TARGET_DIR/scripts/"*.sh 2>/dev/null || true

echo ""
echo -e "GREEN[✓] Skill 已安装到: $TARGET_DIRNC"
echo ""
echo -e "  目录结构:"
echo -e "  $TARGET_DIR/"
echo -e "  ├── SKILL.md"
echo -e "  └── scripts/"
echo -e "      ├── install.sh"
echo -e "      ├── setup-config.sh"
echo -e "      ├── viking.py"
echo -e "      └── demo-token-compare.py"
echo ""
echo -e "BOLD下一步：NC"
echo -e "  1. 在 OpenClaw 中说 BOLD\"refresh skills\"NC"
echo -e "  2. 然后说 BOLD\"帮我安装 openviking\"NC"
echo -e "  3. 或手动:  bash $TARGET_DIR/scripts/install.sh"
echo ""
echo -e "YELLOW提示：NCDocker 用户安装后需要重启容器让 OpenClaw 发现新 skill："
echo -e "  docker compose restart openclaw"
echo ""
echo -e "也可以通过环境变量跳过自动检测："
echo -e "  OPENCLAW_SKILLS_DIR=/your/path bash scripts/install-skill.sh"
echo ""

FILE:scripts/install.sh
#!/usr/bin/env bash
set -euo pipefail

BOLD='\033[1m'
GREEN='\033[0;32m'
YELLOW='\033[0;33m'
RED='\033[0;31m'
NC='\033[0m'

log_info()  { echo -e "GREEN[✓]NC $1"; }
log_warn()  { echo -e "YELLOW[!]NC $1"; }
log_error() { echo -e "RED[✗]NC $1"; }
log_step()  { echo -e "\nBOLD▸ $1NC"; }

WORKSPACE_DIR="-$HOME/openviking_workspace"
CONFIG_DIR="$HOME/.openviking"

log_step "检查 Python 版本"
if ! command -v python3 &>/dev/null; then
    log_error "未找到 python3，请先安装 Python 3.10+"
    exit 1
fi

PY_VERSION=$(python3 -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')
PY_MAJOR=$(echo "$PY_VERSION" | cut -d. -f1)
PY_MINOR=$(echo "$PY_VERSION" | cut -d. -f2)

if [ "$PY_MAJOR" -lt 3 ] || ([ "$PY_MAJOR" -eq 3 ] && [ "$PY_MINOR" -lt 10 ]); then
    log_error "Python 版本 $PY_VERSION 不满足要求，需要 >= 3.10"
    exit 1
fi
log_info "Python $PY_VERSION ✓"

log_step "安装 OpenViking Python 包"
pip install openviking --upgrade --force-reinstall 2>&1 | tail -5
log_info "openviking 安装完成"

log_step "创建工作目录"
mkdir -p "$WORKSPACE_DIR"
mkdir -p "$CONFIG_DIR"
log_info "工作目录: $WORKSPACE_DIR"
log_info "配置目录: $CONFIG_DIR"

log_step "是否安装 Rust CLI (ov)? [y/N]"
read -r INSTALL_CLI
if [[ "$INSTALL_CLI" =~ ^[Yy]$ ]]; then
    if command -v cargo &>/dev/null; then
        log_info "通过 cargo 安装 ov_cli..."
        cargo install --git https://github.com/volcengine/OpenViking ov_cli
    else
        log_info "通过安装脚本安装 ov..."
        curl -fsSL https://raw.githubusercontent.com/volcengine/OpenViking/main/crates/ov_cli/install.sh | bash
    fi
    log_info "Rust CLI 安装完成"
else
    log_info "跳过 Rust CLI 安装"
fi

log_step "设置环境变量"
SHELL_RC=""
if [ -f "$HOME/.zshrc" ]; then
    SHELL_RC="$HOME/.zshrc"
elif [ -f "$HOME/.bashrc" ]; then
    SHELL_RC="$HOME/.bashrc"
fi

if [ -n "$SHELL_RC" ]; then
    if ! grep -q "OPENVIKING_CONFIG_FILE" "$SHELL_RC" 2>/dev/null; then
        {
            echo ""
            echo "# OpenViking"
            echo "export OPENVIKING_CONFIG_FILE=$CONFIG_DIR/ov.conf"
            echo "export OPENVIKING_CLI_CONFIG_FILE=$CONFIG_DIR/ovcli.conf"
        } >> "$SHELL_RC"
        log_info "环境变量已写入 $SHELL_RC"
    else
        log_warn "环境变量已存在于 $SHELL_RC，跳过"
    fi
fi

export OPENVIKING_CONFIG_FILE="$CONFIG_DIR/ov.conf"
export OPENVIKING_CLI_CONFIG_FILE="$CONFIG_DIR/ovcli.conf"

log_step "验证安装"
if python3 -c "import openviking; print(f'openviking {openviking.__version__}')" 2>/dev/null; then
    log_info "OpenViking 导入成功"
else
    log_warn "无法导入 openviking，可能需要重新安装"
fi

echo ""
echo -e "GREEN════════════════════════════════════════NC"
echo -e "GREEN  OpenViking 安装完成！NC"
echo -e "GREEN════════════════════════════════════════NC"
echo ""
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"

echo "下一步："
echo "  1. 运行配置脚本:  bash $SCRIPT_DIR/setup-config.sh"
echo "  2. 启动服务器:    openviking-server"
echo "  3. 测试连接:      python3 $SCRIPT_DIR/viking.py info"
echo ""

FILE:scripts/demo-token-compare.py
#!/usr/bin/env python3
"""
Token 消耗对比演示 — 展示 OpenViking L0/L1/L2 分层加载相比传统全量加载的 token 节省效果。

用法:
    python3 demo-token-compare.py <docs_directory>
    python3 demo-token-compare.py ./my-project-docs/

模拟场景: Agent 收到 "帮我写一个用户认证模块" 的自然语言开发需求。
对比传统全量塞入 prompt vs OpenViking 分层按需加载的 token 消耗差异。
"""

import glob
import json
import os
import sys
import time


def estimate_tokens(text: str) -> int:
    """粗略估算 token 数 — 中文按 1 token/字，英文按 4 字符/token"""
    if not text:
        return 0
    cn = sum(1 for c in text if "\u4e00" <= c <= "\u9fff")
    en = len(text) - cn
    return int(cn * 1.0 + en / 4.0)


def read_all_docs(directory: str) -> list[dict]:
    """读取目录下所有文档文件"""
    extensions = ("*.md", "*.txt", "*.py", "*.js", "*.ts", "*.json", "*.yaml", "*.yml", "*.html", "*.css")
    files = []
    for ext in extensions:
        files.extend(glob.glob(os.path.join(directory, "**", ext), recursive=True))

    docs = []
    for f in sorted(set(files)):
        if os.path.isfile(f):
            try:
                with open(f, encoding="utf-8", errors="ignore") as fh:
                    content = fh.read()
                docs.append({"path": f, "content": content, "tokens": estimate_tokens(content)})
            except Exception:
                pass
    return docs


def simulate_l0_abstract(content: str) -> str:
    """模拟 L0 摘要 — 取首行或前 100 字符"""
    lines = content.strip().splitlines()
    first_meaningful = ""
    for line in lines:
        stripped = line.strip().lstrip("#").strip()
        if stripped and not stripped.startswith("---"):
            first_meaningful = stripped
            break
    return first_meaningful[:150] if first_meaningful else content[:150]


def simulate_l1_overview(content: str) -> str:
    """模拟 L1 概览 — 提取标题和首段，限制约 2000 字符"""
    lines = content.strip().splitlines()
    overview_parts = []
    total_chars = 0
    in_code_block = False

    for line in lines:
        if line.strip().startswith("```"):
            in_code_block = not in_code_block
            continue
        if in_code_block:
            continue

        if line.startswith("#"):
            overview_parts.append(line)
            total_chars += len(line)
        elif line.strip() and total_chars < 2000:
            overview_parts.append(line)
            total_chars += len(line)

        if total_chars >= 2000:
            break

    return "\n".join(overview_parts)


def main():
    if len(sys.argv) < 2:
        print("用法: python3 demo-token-compare.py <docs_directory>")
        print()
        print("示例: python3 demo-token-compare.py ./my-project-docs/")
        print()
        print("如果没有文档目录，将使用内置演示数据。")
        use_demo = True
    else:
        use_demo = False

    print()
    print("═" * 60)
    print("  OpenViking 分层加载 Token 消耗对比演示")
    print("═" * 60)
    print()
    print("场景: Agent 收到自然语言需求 '帮我写一个用户认证模块'")
    print("对比: 传统全量加载 vs OpenViking L0/L1/L2 分层按需加载")
    print()

    if use_demo:
        docs = _generate_demo_docs()
        print(f"使用内置演示数据 ({len(docs)} 个模拟文档)")
    else:
        target_dir = sys.argv[1]
        if not os.path.isdir(target_dir):
            print(f"ERROR: 目录不存在: {target_dir}", file=sys.stderr)
            sys.exit(1)
        docs = read_all_docs(target_dir)
        print(f"扫描目录: {target_dir} ({len(docs)} 个文档)")

    if not docs:
        print("未找到文档文件")
        sys.exit(1)

    print()
    print("─" * 60)

    # === 方案 1: 传统全量加载 ===
    total_full_tokens = sum(d["tokens"] for d in docs)
    print(f"\n▸ 方案 1: 传统全量加载 (把所有文档塞进 prompt)")
    print(f"  文件数:    {len(docs)}")
    print(f"  总 Token:  {total_full_tokens:,}")
    print(f"  特点:      无差别加载，token 浪费严重")

    # === 方案 2: L0 扫描 ===
    l0_results = []
    for d in docs:
        abstract = simulate_l0_abstract(d["content"])
        l0_results.append({
            "path": d["path"],
            "abstract": abstract,
            "tokens": estimate_tokens(abstract),
            "full_tokens": d["tokens"],
        })

    total_l0_tokens = sum(r["tokens"] for r in l0_results)
    print(f"\n▸ 方案 2: OpenViking L0 扫描 (仅加载摘要)")
    print(f"  文件数:    {len(docs)}")
    print(f"  L0 Token:  {total_l0_tokens:,}")
    print(f"  节省:      {(1 - total_l0_tokens / max(total_full_tokens, 1)) * 100:.1f}%")
    print(f"  特点:      Agent 用 L0 摘要判断哪些资源相关")

    # === 方案 3: L0 + L1 按需 ===
    relevant_count = max(len(docs) // 3, 2)
    relevant_docs = sorted(l0_results, key=lambda x: x["full_tokens"], reverse=True)[:relevant_count]

    l1_tokens = 0
    for d in relevant_docs:
        full_doc = next((doc for doc in docs if doc["path"] == d["path"]), None)
        if full_doc:
            overview = simulate_l1_overview(full_doc["content"])
            l1_tokens += estimate_tokens(overview)

    total_l0_l1_tokens = total_l0_tokens + l1_tokens
    print(f"\n▸ 方案 3: OpenViking L0 + L1 (摘要 + 概览)")
    print(f"  L0 全部:   {total_l0_tokens:,} tokens ({len(docs)} 个文件)")
    print(f"  L1 选读:   {l1_tokens:,} tokens ({relevant_count} 个相关文件)")
    print(f"  合计:      {total_l0_l1_tokens:,}")
    print(f"  节省:      {(1 - total_l0_l1_tokens / max(total_full_tokens, 1)) * 100:.1f}%")
    print(f"  特点:      Agent 用 L1 概览理解架构，制定计划")

    # === 方案 4: L0 + L1 + L2 按需深读 ===
    deep_read_count = max(relevant_count // 2, 1)
    deep_docs = relevant_docs[:deep_read_count]
    l2_tokens = sum(d["full_tokens"] for d in deep_docs)

    total_layered = total_l0_tokens + l1_tokens + l2_tokens
    print(f"\n▸ 方案 4: OpenViking L0 + L1 + L2 (完整分层)")
    print(f"  L0 全部:   {total_l0_tokens:,} tokens ({len(docs)} 个文件)")
    print(f"  L1 选读:   {l1_tokens:,} tokens ({relevant_count} 个相关文件)")
    print(f"  L2 深读:   {l2_tokens:,} tokens ({deep_read_count} 个必要文件)")
    print(f"  合计:      {total_layered:,}")
    print(f"  节省:      {(1 - total_layered / max(total_full_tokens, 1)) * 100:.1f}%")
    print(f"  特点:      仅深读写代码必需的具体文件")

    # === 汇总 ===
    print()
    print("═" * 60)
    print("  对比汇总")
    print("═" * 60)
    print()
    print(f"  {'方案':<35} {'Token':>10}  {'节省':>8}")
    print(f"  {'─' * 35} {'─' * 10}  {'─' * 8}")
    print(f"  {'传统全量加载':<33} {total_full_tokens:>10,}  {'基准':>8}")
    print(f"  {'OpenViking L0 扫描':<33} {total_l0_tokens:>10,}  {(1 - total_l0_tokens / max(total_full_tokens, 1)) * 100:>7.1f}%")
    print(f"  {'OpenViking L0 + L1':<33} {total_l0_l1_tokens:>10,}  {(1 - total_l0_l1_tokens / max(total_full_tokens, 1)) * 100:>7.1f}%")
    print(f"  {'OpenViking L0 + L1 + L2 按需':<33} {total_layered:>10,}  {(1 - total_layered / max(total_full_tokens, 1)) * 100:>7.1f}%")
    print()

    if total_full_tokens > 0:
        ratio = total_layered / total_full_tokens
        print(f"  结论: OpenViking 分层加载消耗仅为全量加载的 {ratio * 100:.1f}%，")
        print(f"        节省 {(1 - ratio) * 100:.1f}% 的 token 开销。")
    print()

    # === 输出 JSON 报告 ===
    report = {
        "scenario": "自然语言开发 — 用户认证模块",
        "total_files": len(docs),
        "comparison": {
            "full_load": {"tokens": total_full_tokens, "description": "传统全量加载"},
            "l0_scan": {"tokens": total_l0_tokens, "saving_pct": round((1 - total_l0_tokens / max(total_full_tokens, 1)) * 100, 1)},
            "l0_l1": {"tokens": total_l0_l1_tokens, "relevant_files": relevant_count, "saving_pct": round((1 - total_l0_l1_tokens / max(total_full_tokens, 1)) * 100, 1)},
            "l0_l1_l2": {"tokens": total_layered, "deep_read_files": deep_read_count, "saving_pct": round((1 - total_layered / max(total_full_tokens, 1)) * 100, 1)},
        },
    }
    report_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "token-report.json")
    with open(report_path, "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    print(f"  报告已保存: {report_path}")


def _generate_demo_docs() -> list[dict]:
    """生成内置演示数据 — 模拟一个中型项目的文档结构"""
    demo_files = {
        "docs/architecture.md": """# 项目架构

## 概述
本项目采用微服务架构，包含以下核心模块：
- 用户服务 (User Service)
- 认证服务 (Auth Service)
- 订单服务 (Order Service)
- 支付服务 (Payment Service)

## 技术栈
- 后端: Python FastAPI
- 数据库: PostgreSQL
- 缓存: Redis
- 消息队列: RabbitMQ

## 服务通信
服务间通过 gRPC 通信，外部 API 使用 RESTful 接口。
每个服务独立部署，通过 Kubernetes 编排管理。

## 数据流
1. 客户端请求 → API Gateway → 对应微服务
2. 微服务处理 → 数据库读写 → 响应返回
3. 异步任务 → 消息队列 → Worker 处理
""",
        "docs/auth/jwt-config.md": """# JWT 配置说明

## Token 结构
```json
{
  "sub": "user_id",
  "exp": 1234567890,
  "iat": 1234567890,
  "roles": ["admin", "user"],
  "permissions": ["read", "write"]
}
```

## 密钥管理
- 使用 RS256 算法
- 私钥存储在 Vault 中
- 公钥通过 JWKS 端点分发
- 密钥轮换周期: 90 天

## Token 生命周期
- Access Token: 15 分钟
- Refresh Token: 7 天
- 支持 Token 黑名单机制

## 配置项
```yaml
auth:
  jwt:
    algorithm: RS256
    access_token_expire: 900
    refresh_token_expire: 604800
    issuer: "auth-service"
    audience: "api-gateway"
```
""",
        "docs/auth/oauth2-flow.md": """# OAuth2 认证流程

## 支持的 Grant Types
1. Authorization Code (推荐)
2. Client Credentials (服务间调用)
3. Refresh Token

## Authorization Code 流程
1. 客户端重定向到授权端点
2. 用户登录并授权
3. 回调携带 authorization code
4. 客户端用 code 换取 token
5. 返回 access_token + refresh_token

## 第三方登录集成
- Google OAuth2
- GitHub OAuth
- WeChat OAuth (微信登录)

## 安全措施
- PKCE 扩展 (S256)
- State 参数防 CSRF
- Redirect URI 白名单
- Token 加密存储
""",
        "docs/auth/rbac.md": """# RBAC 权限模型

## 角色定义
| 角色 | 权限 | 说明 |
|------|------|------|
| super_admin | * | 超级管理员 |
| admin | user.*, order.read | 普通管理员 |
| user | self.*, order.create | 普通用户 |
| guest | public.read | 访客 |

## 权限检查中间件
```python
@require_permission("user.write")
async def update_user(user_id: str, data: UserUpdate):
    ...
```

## 数据库模型
- users: 用户表
- roles: 角色表
- permissions: 权限表
- user_roles: 用户-角色关联
- role_permissions: 角色-权限关联
""",
        "docs/api/endpoints.md": """# API 端点文档

## 用户管理
- POST /api/v1/users - 创建用户
- GET /api/v1/users/:id - 获取用户详情
- PUT /api/v1/users/:id - 更新用户
- DELETE /api/v1/users/:id - 删除用户
- GET /api/v1/users - 用户列表 (分页)

## 认证相关
- POST /api/v1/auth/login - 登录
- POST /api/v1/auth/logout - 登出
- POST /api/v1/auth/refresh - 刷新 Token
- POST /api/v1/auth/register - 注册
- POST /api/v1/auth/forgot-password - 忘记密码
- POST /api/v1/auth/reset-password - 重置密码

## 订单管理
- POST /api/v1/orders - 创建订单
- GET /api/v1/orders/:id - 订单详情
- GET /api/v1/orders - 订单列表
- PUT /api/v1/orders/:id/cancel - 取消订单

## 支付
- POST /api/v1/payments - 发起支付
- GET /api/v1/payments/:id - 支付状态
- POST /api/v1/payments/:id/callback - 支付回调

## 通用参数
- page: 页码 (默认 1)
- page_size: 每页数量 (默认 20, 最大 100)
- sort: 排序字段
- order: asc / desc
""",
        "docs/database/schema.md": """# 数据库 Schema

## users 表
```sql
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email VARCHAR(255) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    nickname VARCHAR(100),
    avatar_url TEXT,
    status VARCHAR(20) DEFAULT 'active',
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);
```

## orders 表
```sql
CREATE TABLE orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID REFERENCES users(id),
    total_amount DECIMAL(10,2) NOT NULL,
    status VARCHAR(20) DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT NOW()
);
```

## payments 表
```sql
CREATE TABLE payments (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    order_id UUID REFERENCES orders(id),
    amount DECIMAL(10,2) NOT NULL,
    method VARCHAR(50),
    status VARCHAR(20) DEFAULT 'pending',
    paid_at TIMESTAMP
);
```
""",
        "docs/deployment/docker.md": """# Docker 部署指南

## 构建镜像
```bash
docker build -t myapp:latest .
```

## Docker Compose
```yaml
version: '3.8'
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://...
      - REDIS_URL=redis://redis:6379
    depends_on:
      - postgres
      - redis

  postgres:
    image: postgres:15
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine
```

## 环境变量
| 变量 | 必填 | 说明 |
|------|------|------|
| DATABASE_URL | Y | PostgreSQL 连接串 |
| REDIS_URL | Y | Redis 连接串 |
| JWT_SECRET | Y | JWT 签名密钥 |
| LOG_LEVEL | N | 日志级别，默认 INFO |
""",
        "docs/testing/guide.md": """# 测试指南

## 单元测试
```bash
pytest tests/unit/ -v
```

## 集成测试
```bash
pytest tests/integration/ -v --cov=app
```

## E2E 测试
```bash
pytest tests/e2e/ -v
```

## 覆盖率要求
- 核心模块: >= 80%
- 工具模块: >= 60%

## Mock 规范
- 外部 API 调用必须 mock
- 数据库使用 test fixtures
- 时间相关使用 freezegun
""",
        "src/auth/service.py": """# Auth Service 实现

from datetime import datetime, timedelta
from typing import Optional
import jwt
from passlib.hash import bcrypt

from app.config import settings
from app.models import User
from app.database import get_db

class AuthService:
    def __init__(self):
        self.secret_key = settings.JWT_SECRET
        self.algorithm = "RS256"
        self.access_expire = timedelta(minutes=15)
        self.refresh_expire = timedelta(days=7)

    async def authenticate(self, email: str, password: str) -> Optional[User]:
        db = get_db()
        user = await db.users.find_one({"email": email})
        if not user or not bcrypt.verify(password, user.password_hash):
            return None
        return user

    def create_access_token(self, user_id: str, roles: list) -> str:
        payload = {
            "sub": user_id,
            "roles": roles,
            "exp": datetime.utcnow() + self.access_expire,
            "iat": datetime.utcnow(),
            "type": "access"
        }
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

    def create_refresh_token(self, user_id: str) -> str:
        payload = {
            "sub": user_id,
            "exp": datetime.utcnow() + self.refresh_expire,
            "iat": datetime.utcnow(),
            "type": "refresh"
        }
        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

    def verify_token(self, token: str) -> dict:
        return jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
""",
        "src/auth/middleware.py": """# Auth Middleware

from functools import wraps
from fastapi import Request, HTTPException
from app.auth.service import AuthService

auth_service = AuthService()

def require_auth(func):
    @wraps(func)
    async def wrapper(request: Request, *args, **kwargs):
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not token:
            raise HTTPException(status_code=401, detail="Missing token")
        try:
            payload = auth_service.verify_token(token)
            request.state.user = payload
        except Exception:
            raise HTTPException(status_code=401, detail="Invalid token")
        return await func(request, *args, **kwargs)
    return wrapper

def require_permission(permission: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            user = getattr(request.state, "user", None)
            if not user:
                raise HTTPException(status_code=401)
            roles = user.get("roles", [])
            if "super_admin" not in roles:
                # check specific permission
                pass
            return await func(request, *args, **kwargs)
        return wrapper
    return decorator
""",
    }

    docs = []
    for path, content in demo_files.items():
        docs.append({"path": path, "content": content, "tokens": estimate_tokens(content)})
    return docs


if __name__ == "__main__":
    main()

FILE:scripts/setup-config.sh
#!/usr/bin/env bash
set -euo pipefail

BOLD='\033[1m'
GREEN='\033[0;32m'
YELLOW='\033[0;33m'
CYAN='\033[0;36m'
NC='\033[0m'

CONFIG_DIR="$HOME/.openviking"
WORKSPACE_DIR="-$HOME/openviking_workspace"

log_info()  { echo -e "GREEN[✓]NC $1"; }
log_step()  { echo -e "\nBOLD▸ $1NC"; }
prompt()    { echo -en "CYAN  $1NC"; }

mkdir -p "$CONFIG_DIR"

echo -e "BOLD═══════════════════════════════════════════NC"
echo -e "BOLD  OpenViking 交互式配置向导NC"
echo -e "BOLD═══════════════════════════════════════════NC"

# --- 选择提供商 ---
log_step "选择模型提供商"
echo "  1) openai       — OpenAI / 兼容 API"
echo "  2) volcengine   — 火山引擎豆包"
echo "  3) nvidia       — NVIDIA NIM (免费)"
echo "  4) litellm      — LiteLLM (Anthropic/DeepSeek/Gemini/Ollama...)"
echo ""
prompt "请选择 [1-4]: "
read -r PROVIDER_CHOICE

case "$PROVIDER_CHOICE" in
    1)
        VLM_PROVIDER="openai"
        EMB_PROVIDER="openai"
        DEFAULT_VLM_MODEL="gpt-4o"
        DEFAULT_EMB_MODEL="text-embedding-3-large"
        DEFAULT_EMB_DIM=3072
        DEFAULT_API_BASE="https://api.openai.com/v1"
        ;;
    2)
        VLM_PROVIDER="volcengine"
        EMB_PROVIDER="volcengine"
        DEFAULT_VLM_MODEL="doubao-seed-2-0-pro-260215"
        DEFAULT_EMB_MODEL="doubao-embedding-vision-250615"
        DEFAULT_EMB_DIM=1024
        DEFAULT_API_BASE="https://ark.cn-beijing.volces.com/api/v3"
        ;;
    3)
        VLM_PROVIDER="openai"
        EMB_PROVIDER="openai"
        DEFAULT_VLM_MODEL="meta/llama-3.3-70b-instruct"
        DEFAULT_EMB_MODEL="nvidia/nv-embed-v1"
        DEFAULT_EMB_DIM=4096
        DEFAULT_API_BASE="https://integrate.api.nvidia.com/v1"
        ;;
    4)
        VLM_PROVIDER="litellm"
        EMB_PROVIDER="openai"
        DEFAULT_VLM_MODEL="claude-3-5-sonnet-20240620"
        DEFAULT_EMB_MODEL="text-embedding-3-large"
        DEFAULT_EMB_DIM=3072
        DEFAULT_API_BASE=""
        ;;
    *)
        echo "无效选择，使用 openai 默认值"
        VLM_PROVIDER="openai"
        EMB_PROVIDER="openai"
        DEFAULT_VLM_MODEL="gpt-4o"
        DEFAULT_EMB_MODEL="text-embedding-3-large"
        DEFAULT_EMB_DIM=3072
        DEFAULT_API_BASE="https://api.openai.com/v1"
        ;;
esac

# --- API Key ---
log_step "配置 API Key"
prompt "API Key: "
read -rs API_KEY
echo ""

if [ -z "$API_KEY" ]; then
    echo -e "YELLOW[!] 未提供 API Key，你可以稍后在配置文件中设置NC"
    API_KEY="YOUR_API_KEY_HERE"
fi

# --- API Base ---
log_step "配置 API 端点"
prompt "API Base [$DEFAULT_API_BASE]: "
read -r API_BASE
API_BASE="-$DEFAULT_API_BASE"

# --- VLM Model ---
log_step "配置 VLM 模型（用于语义理解）"
prompt "VLM Model [$DEFAULT_VLM_MODEL]: "
read -r VLM_MODEL
VLM_MODEL="-$DEFAULT_VLM_MODEL"

# --- Embedding Model ---
log_step "配置 Embedding 模型（用于向量检索）"
prompt "Embedding Model [$DEFAULT_EMB_MODEL]: "
read -r EMB_MODEL
EMB_MODEL="-$DEFAULT_EMB_MODEL"

prompt "Embedding Dimension [$DEFAULT_EMB_DIM]: "
read -r EMB_DIM
EMB_DIM="-$DEFAULT_EMB_DIM"

# --- Workspace ---
log_step "配置工作目录"
prompt "Workspace [$WORKSPACE_DIR]: "
read -r WS
WS="-$WORKSPACE_DIR"
mkdir -p "$WS"

# --- 生成 ov.conf ---
log_step "生成服务器配置"

cat > "$CONFIG_DIR/ov.conf" << OVEOF
{
  "storage": {
    "workspace": "$WS"
  },
  "log": {
    "level": "INFO",
    "output": "stdout"
  },
  "embedding": {
    "dense": {
      "api_base": "$API_BASE",
      "api_key": "$API_KEY",
      "provider": "$EMB_PROVIDER",
      "dimension": $EMB_DIM,
      "model": "$EMB_MODEL"
    },
    "max_concurrent": 10
  },
  "vlm": {
    "api_base": "$API_BASE",
    "api_key": "$API_KEY",
    "provider": "$VLM_PROVIDER",
    "model": "$VLM_MODEL",
    "max_concurrent": 100
  }
}
OVEOF

log_info "服务器配置已写入 $CONFIG_DIR/ov.conf"

# --- 生成 ovcli.conf ---
cat > "$CONFIG_DIR/ovcli.conf" << CLIEOF
{
  "url": "http://localhost:1933",
  "timeout": 60.0,
  "output": "table"
}
CLIEOF

log_info "CLI 配置已写入 $CONFIG_DIR/ovcli.conf"

# --- 设置环境变量 ---
export OPENVIKING_CONFIG_FILE="$CONFIG_DIR/ov.conf"
export OPENVIKING_CLI_CONFIG_FILE="$CONFIG_DIR/ovcli.conf"

echo ""
echo -e "GREEN════════════════════════════════════════NC"
echo -e "GREEN  配置完成！NC"
echo -e "GREEN════════════════════════════════════════NC"
echo ""
echo "  配置文件:  $CONFIG_DIR/ov.conf"
echo "  CLI 配置:  $CONFIG_DIR/ovcli.conf"
echo "  工作目录:  $WS"
echo "  VLM:       $VLM_PROVIDER / $VLM_MODEL"
echo "  Embedding: $EMB_PROVIDER / $EMB_MODEL (dim=$EMB_DIM)"
echo ""
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"

echo "下一步："
echo "  1. 启动服务器:  openviking-server"
echo "  2. 测试连接:    python3 $SCRIPT_DIR/viking.py info"
echo "  3. 添加资源:    python3 $SCRIPT_DIR/viking.py add ./your-docs/"
echo ""

FILE:scripts/viking.py
#!/usr/bin/env python3
"""
OpenViking CLI Wrapper — Agent 可直接调用的上下文操作工具。
支持 add / search / ls / tree / abstract / overview / read / info / commit / stats 命令。
每次操作自动追踪 token 消耗，展示分层加载相比全量加载的节省效果。
"""

import argparse
import glob
import json
import os
import sys
import textwrap
import time

STATS_FILE = os.path.expanduser("~/.openviking/session_stats.json")


# ─── Session Token Tracker ───────────────────────────────────────────

class SessionTracker:
    """追踪会话级 token 消耗：实际消耗 vs 全量加载假设值"""

    def __init__(self):
        self._data = self._load()

    def _load(self):
        if os.path.exists(STATS_FILE):
            try:
                with open(STATS_FILE, encoding="utf-8") as f:
                    return json.load(f)
            except Exception:
                pass
        return self._empty()

    @staticmethod
    def _empty():
        return {
            "session_start": time.strftime("%Y-%m-%d %H:%M:%S"),
            "ops": [],
            "totals": {"actual": 0, "full_load": 0},
        }

    def _save(self):
        os.makedirs(os.path.dirname(STATS_FILE), exist_ok=True)
        with open(STATS_FILE, "w", encoding="utf-8") as f:
            json.dump(self._data, f, ensure_ascii=False, indent=2)

    def record(self, op: str, uri: str, layer: str, actual: int, full_load: int):
        entry = {
            "time": time.strftime("%H:%M:%S"),
            "op": op,
            "uri": uri,
            "layer": layer,
            "actual_tokens": actual,
            "full_load_tokens": full_load,
            "saved_tokens": full_load - actual,
        }
        self._data["ops"].append(entry)
        self._data["totals"]["actual"] += actual
        self._data["totals"]["full_load"] += full_load
        self._save()
        return entry

    def reset(self):
        self._data = self._empty()
        self._save()

    @property
    def totals(self):
        return self._data["totals"]

    @property
    def ops(self):
        return self._data["ops"]

    @property
    def session_start(self):
        return self._data.get("session_start", "?")

    def print_summary_line(self):
        t = self.totals
        actual = t["actual"]
        full = t["full_load"]
        saved = full - actual
        if full > 0:
            pct = saved / full * 100
            print(f"\n📊 会话累计 | 实际: {actual:,} tokens | 全量: {full:,} tokens | 节省: {saved:,} ({pct:.1f}%)")
        else:
            print(f"\n📊 会话累计 | 实际: {actual:,} tokens")


tracker = SessionTracker()


# ─── Helpers ─────────────────────────────────────────────────────────

def _estimate_tokens(text):
    if not text:
        return 0
    cn_chars = sum(1 for c in text if "\u4e00" <= c <= "\u9fff")
    en_chars = len(text) - cn_chars
    return int(cn_chars * 1.0 + en_chars / 4.0)


def _get_full_load_tokens(client, uri):
    """获取资源 L2 全文 token 数，作为"全量加载"基准"""
    try:
        full_text = str(client.read(uri))
        return _estimate_tokens(full_text)
    except Exception:
        return 0


def get_client(path=None):
    try:
        import openviking as ov
    except ImportError:
        print("ERROR: openviking 未安装。运行: pip install openviking", file=sys.stderr)
        sys.exit(1)

    workspace = path or os.environ.get(
        "OPENVIKING_WORKSPACE",
        os.path.expanduser("~/openviking_workspace"),
    )
    client = ov.SyncOpenViking(path=workspace)
    client.initialize()
    return client


# ─── Commands ────────────────────────────────────────────────────────

def cmd_add(args):
    client = get_client()
    target = args.target

    if target.startswith("http://") or target.startswith("https://"):
        result = client.add_resource(path=target)
        print(f"已添加 URL: {target}")
        print(f"结果: {result}")
    elif os.path.isdir(target):
        pattern = os.path.join(target, "**", "*.*")
        files = glob.glob(pattern, recursive=True)
        added = 0
        for f in files:
            if os.path.isfile(f) and not f.startswith("."):
                try:
                    client.add_resource(path=f)
                    added += 1
                    print(f"  + {f}")
                except Exception as e:
                    print(f"  ! {f}: {e}", file=sys.stderr)
        print(f"\n共添加 {added} 个文件")
    elif os.path.isfile(target):
        result = client.add_resource(path=target)
        print(f"已添加文件: {target}")
        print(f"结果: {result}")
    else:
        print(f"ERROR: 路径不存在: {target}", file=sys.stderr)
        sys.exit(1)

    if args.wait:
        print("等待语义处理完成...")
        client.wait_processed()
        print("处理完成 ✓")

    client.close()


def cmd_search(args):
    client = get_client()
    results = client.find(args.query, limit=args.limit)

    if not results.resources:
        print("未找到匹配结果")
        client.close()
        return

    total_actual = 0
    total_full = 0

    print(f"找到 {len(results.resources)} 个结果:\n")
    for i, r in enumerate(results.resources, 1):
        score = f"{r.score:.4f}" if hasattr(r, "score") else "N/A"
        abstract_text = ""
        if hasattr(r, "abstract") and r.abstract:
            abstract_text = str(r.abstract)

        print(f"  {i}. [{score}] {r.uri}")
        if abstract_text:
            print(f"     {textwrap.shorten(abstract_text, width=80)}")

        actual = _estimate_tokens(abstract_text) if abstract_text else 0
        full = _get_full_load_tokens(client, r.uri)
        total_actual += actual
        total_full += full

        print(f"     [L0: {actual} tokens | 全量: {full} tokens]")
        print()

    tracker.record("search", args.query, "L0", total_actual, total_full)
    tracker.print_summary_line()
    client.close()


def cmd_ls(args):
    client = get_client()
    uri = args.uri or "viking://resources"
    result = client.ls(uri)
    print(f"目录: {uri}\n")
    if hasattr(result, "__iter__"):
        for item in result:
            print(f"  {item}")
    else:
        print(result)
    client.close()


def cmd_tree(args):
    client = get_client()
    uri = args.uri or "viking://resources"
    try:
        result = client.tree(uri, depth=args.level)
        print(result)
    except AttributeError:
        print(f"tree 命令需要 ov CLI 支持，尝试: ov tree {uri} -L {args.level}")
    client.close()


def cmd_abstract(args):
    client = get_client()
    result = client.abstract(args.uri)
    content = str(result)

    print(f"═══ L0 摘要 ({args.uri}) ═══\n")
    print(content)

    actual = _estimate_tokens(content)
    full = _get_full_load_tokens(client, args.uri)

    print(f"\n[L0: {actual} tokens | 全量: {full} tokens | 节省: {full - actual} tokens]")
    tracker.record("abstract", args.uri, "L0", actual, full)
    tracker.print_summary_line()
    client.close()


def cmd_overview(args):
    client = get_client()
    result = client.overview(args.uri)
    content = str(result)

    print(f"═══ L1 概览 ({args.uri}) ═══\n")
    print(content)

    actual = _estimate_tokens(content)
    full = _get_full_load_tokens(client, args.uri)

    print(f"\n[L1: {actual} tokens | 全量: {full} tokens | 节省: {full - actual} tokens]")
    tracker.record("overview", args.uri, "L1", actual, full)
    tracker.print_summary_line()
    client.close()


def cmd_read(args):
    client = get_client()
    result = client.read(args.uri)
    content = str(result)

    print(f"═══ L2 全文 ({args.uri}) ═══\n")
    if args.head:
        lines = content.splitlines()
        display = "\n".join(lines[: args.head])
        print(display)
        if len(lines) > args.head:
            print(f"\n... (截断，共 {len(lines)} 行)")
    else:
        print(content)

    actual = _estimate_tokens(content)

    print(f"\n[L2 全文: {actual} tokens — 此次为全量读取，无节省]")
    tracker.record("read", args.uri, "L2", actual, actual)
    tracker.print_summary_line()
    client.close()


def cmd_info(args):
    config_file = os.environ.get(
        "OPENVIKING_CONFIG_FILE",
        os.path.expanduser("~/.openviking/ov.conf"),
    )

    print("═══ OpenViking 状态 ═══\n")

    if os.path.exists(config_file):
        print(f"配置文件: {config_file} ✓")
        try:
            with open(config_file) as f:
                conf = json.load(f)
            vlm = conf.get("vlm", {})
            emb = conf.get("embedding", {}).get("dense", {})
            print(f"VLM:       {vlm.get('provider', '?')} / {vlm.get('model', '?')}")
            print(f"Embedding: {emb.get('provider', '?')} / {emb.get('model', '?')} (dim={emb.get('dimension', '?')})")
            print(f"Workspace: {conf.get('storage', {}).get('workspace', '?')}")
        except Exception as e:
            print(f"配置读取错误: {e}")
    else:
        print(f"配置文件: {config_file} ✗ (未找到)")

    print()
    try:
        client = get_client()
        print("OpenViking 连接: ✓")
        client.close()
    except Exception as e:
        print(f"OpenViking 连接: ✗ ({e})")

    try:
        import openviking
        print(f"版本: {openviking.__version__}")
    except Exception:
        pass


def cmd_commit(args):
    client = get_client()
    try:
        result = client.commit()
        print("记忆提取已触发")
        print(f"结果: {result}")
    except Exception as e:
        print(f"记忆提取失败: {e}", file=sys.stderr)
    client.close()


def cmd_stats(args):
    """展示当前会话的 token 消耗汇总"""
    ops = tracker.ops
    t = tracker.totals
    actual = t["actual"]
    full = t["full_load"]
    saved = full - actual

    print(f"═══ Token 消耗统计 ═══")
    print(f"  会话开始: {tracker.session_start}")
    print(f"  操作次数: {len(ops)}")
    print()

    if not ops:
        print("  暂无操作记录。使用 search/abstract/overview/read 后自动记录。")
        return

    print(f"  {'#':<4} {'时间':<10} {'操作':<10} {'层级':<5} {'实际':>8} {'全量':>8} {'节省':>8} {'URI'}")
    print(f"  {'─'*4} {'─'*10} {'─'*10} {'─'*5} {'─'*8} {'─'*8} {'─'*8} {'─'*30}")

    for i, op in enumerate(ops, 1):
        uri_short = op["uri"][:30] + "..." if len(op["uri"]) > 30 else op["uri"]
        print(
            f"  {i:<4} {op['time']:<10} {op['op']:<10} {op['layer']:<5} "
            f"{op['actual_tokens']:>8,} {op['full_load_tokens']:>8,} {op['saved_tokens']:>8,} {uri_short}"
        )

    print()
    print(f"  ┌─────────────────────────────────────┐")
    print(f"  │  全量加载 (传统方式): {full:>10,} tokens │")
    print(f"  │  实际消耗 (分层加载): {actual:>10,} tokens │")
    print(f"  │  节省 token 数量:     {saved:>10,} tokens │")
    if full > 0:
        pct = saved / full * 100
        print(f"  │  节省比例:            {pct:>9.1f}%        │")
    print(f"  └─────────────────────────────────────┘")

    if args.reset:
        tracker.reset()
        print("\n  ✓ 统计数据已重置")


# ─── Main ────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        prog="viking",
        description="OpenViking CLI Wrapper — Agent 上下文操作工具（含 token 追踪）",
    )
    sub = parser.add_subparsers(dest="command", help="可用命令")

    p_add = sub.add_parser("add", help="添加资源（文件/目录/URL）")
    p_add.add_argument("target", help="文件路径、目录路径或 URL")
    p_add.add_argument("--wait", action="store_true", help="等待语义处理完成")

    p_search = sub.add_parser("search", help="语义搜索")
    p_search.add_argument("query", help="搜索查询")
    p_search.add_argument("--limit", type=int, default=5, help="最大结果数")

    p_ls = sub.add_parser("ls", help="浏览资源目录")
    p_ls.add_argument("uri", nargs="?", help="viking:// URI")

    p_tree = sub.add_parser("tree", help="树形展示资源")
    p_tree.add_argument("uri", nargs="?", help="viking:// URI")
    p_tree.add_argument("-L", "--level", type=int, default=3, help="展示层级")

    p_abstract = sub.add_parser("abstract", help="读取 L0 摘要")
    p_abstract.add_argument("uri", help="viking:// URI")

    p_overview = sub.add_parser("overview", help="读取 L1 概览")
    p_overview.add_argument("uri", help="viking:// URI")

    p_read = sub.add_parser("read", help="读取 L2 全文")
    p_read.add_argument("uri", help="viking:// URI")
    p_read.add_argument("--head", type=int, help="只显示前 N 行")

    sub.add_parser("info", help="检查服务状态和配置")
    sub.add_parser("commit", help="提取当前会话记忆")

    p_stats = sub.add_parser("stats", help="查看 token 消耗统计")
    p_stats.add_argument("--reset", action="store_true", help="重置统计数据")

    args = parser.parse_args()

    if not args.command:
        parser.print_help()
        sys.exit(0)

    handlers = {
        "add": cmd_add,
        "search": cmd_search,
        "ls": cmd_ls,
        "tree": cmd_tree,
        "abstract": cmd_abstract,
        "overview": cmd_overview,
        "read": cmd_read,
        "info": cmd_info,
        "commit": cmd_commit,
        "stats": cmd_stats,
    }

    handlers[args.command](args)


if __name__ == "__main__":
    main()