学完你也能成为AI大神！2万多字，Claude Code 从架构到部署源码超详细解读

AiAgent 马化云

发布于 2026-04-17 21:19:16

110

揭秘：Claude Code 源码超详细架构解读

前言

想象一下,如果你能看懂 Claude Code 这款 AI 编程助手的源码,会发生什么?这就像获得了一把打开黑盒的钥匙,能看到 AI 如何思考、如何决策、如何自动修复错误的完整过程。

今天要分享的这个项目——Claw Code Plus,是 Claude Code 的 Python 移植版本,它完整保留了原版的架构设计理念,同时支持 12+ 家国产大模型。这意味着什么?意味着你可以用它来学习最先进的 AI 编程助手是如何工作的,还能用阿里云通义千问、智谱 AI、百度文心一言这些国产模型来跑同样的功能。

这篇文章会带你从零到一,把所有核心模块都剖析清楚:运行时如何调度、Prompt 如何组装、工具如何执行、多 Agent 如何协作,还有那些让人惊叹的自修复机制。读完之后,你不仅能理解这个系统,还能从中学习到构建 AI 应用的最佳实践。

快速上手

先别急着深入代码,先看看这东西能干什么,解决什么问题。

能解决什么问题

场景一:代码阅读和理解 面对一个陌生的大型项目,需要快速理解架构、找到关键代码位置。 → 用 Glob 找文件、Grep 搜索内容、Read 读取代码,几轮对话就能摸清项目结构。

场景二:代码编写和调试 写新功能、修 bug、重构代码。 → Claude 生成代码 → Edit 工具修改 → 运行测试 → 失败后自动重试 → 修复问题。

场景三:任务管理和协作 多步骤任务需要分步执行、跟踪进度。 → TaskCreate 创建任务列表 → 逐步完成 → 自动记录进度。

场景四:团队协作 多个 Agent 协同工作,一个负责搜索,一个负责实现,一个负责验证。 → Agent tool 启动子智能体 → TeamCreate 创建团队 → 并行工作。

如何安装使用

# 1. 编译 Rust CLI
cd rust
./build.sh

# 2. 配置模型(以阿里云为例)
./config.sh
# 选择:阿里云通义千问 → qwen-plus → 输入 API Key

# 3. 启动
./start-cli.sh

就这么简单!配置完就能开始对话,体验和 Claude Code 一样的交互。

核心架构总览

技术栈对比

模块	原版	Python 版本	作用
运行时	TypeScript (QueryEngine.ts ~46K 行)	Python (runtime.py ~200 行)	会话管理和请求调度
查询引擎	TypeScript (query.ts ~1729 行)	Python (query_engine.py ~194 行)	消息提交和流式处理
工具系统	TypeScript (38+ 工具)	Python (tools.py + 工具快照)	工具注册和执行
命令系统	TypeScript (100+ 命令)	Python (commands.py + 命令快照)	斜杠命令处理
权限系统	TypeScript (hooks + classifier)	Python (permissions.py)	工具访问控制

目录结构

src/
├── runtime.py              # 运行时核心 (会话路由、初始化)
├── query_engine.py         # 查询引擎 (消息提交、流式处理)
├── tools.py               # 工具注册表 (184 个工具)
├── commands.py            # 命令注册表 (207 个命令)
├── execution_registry.py   # 执行注册表 (统一入口)
├── permissions.py          # 权限上下文
├── session_store.py       # 会话持久化
├── transcript.py          # 转录存储
├── history.py            # 历史记录
├── models.py             # 共享数据模型
└── reference_data/       # 快照数据
    ├── tools_snapshot.json      # 工具定义 (184 个)
    └── commands_snapshot.json  # 命令定义 (207 个)

数据流向

用户输入提示词
    ↓
PortRuntime.route_prompt()      ← 路由匹配 (命令/工具)
    ↓
ExecutionRegistry              ← 执行注册表
    ↓
QueryEngine.submit_message()    ← 查询引擎
    ↓
[消息准备] → [API 调用] → [工具执行] → [后处理]
    ↓
TurnResult 输出
    ↓
会话持久化 (SessionStore)

架构设计思想

深入理解 Claude Code,先要把握它的核心设计理念。这些思想贯穿整个系统,是理解后续模块的基础。

分层解耦思想

系统采用清晰的分层设计,每层专注自己的职责:

┌─────────────────────────────────────────┐
│   用户交互层 (CLI / Web UI)              │  处理用户输入和展示
├─────────────────────────────────────────┤
│   运行时调度层 (PortRuntime)             │  路由匹配、会话管理
├─────────────────────────────────────────┤
│   执行协调层 (ExecutionRegistry)        │  统一的命令/工具执行接口
├─────────────────────────────────────────┤
│   查询引擎层 (QueryEnginePort)           │  消息提交、流式处理
├─────────────────────────────────────────┤
│   资源管理层 (SessionStore/Tools/Commands) │ 会话、工具、命令
├─────────────────────────────────────────┤
│   外部接口层 (MCP / Plugin / Skill)      │  扩展和集成
└─────────────────────────────────────────┘

分层带来的好处:

优势	说明
职责清晰	每层只做一件事,降低复杂度
易于测试	可以独立测试每一层
方便扩展	新增功能只影响特定层级
便于维护	问题定位更快速

反馈循环设计

自修复机制的核心是完整的反馈循环:

┌──────────────┐
│  用户请求     │
└──────┬───────┘
       ↓
┌──────────────┐      ┌──────────────┐
│ AI 分析决策  │ ───→ │ 调用工具     │
└──────┬───────┘      └──────┬───────┘
       ↓                     ↓
┌──────────────┐      ┌──────────────┐
│ 执行命令     │ ←──── │ 工具执行结果 │
└──────┬───────┘      └──────┬───────┘
       ↓                     ↓
┌──────────────┐      ┌──────────────┐
│ 收集反馈     │ ───→ │ 返回结果     │
└──────┬───────┘      └──────┬───────┘
       ↓                     ↓
┌──────────────┐      ┌──────────────┐
│ AI 重新评估  │ ───→ │ 决定下一步   │
└──────────────┘      └──────────────┘

这个循环的特点:

完整闭环: 每一步的输出都会反馈给 AI
智能判断: AI 根据结果决定继续还是调整
自我修复: 遇到错误自动重试和调整策略

缓存优先策略

系统在多个层面使用缓存来优化性能:

Prompt 缓存
├── 静态部分 (跨用户/组织可复用)
│   └── LRU 缓存 → 减少 token 消耗
├── 动态分界线
│   └── 隔离标志 → 确保缓存一致性
└── 动态部分 (会话专属)
    └── 每会话独立 → 个性化内容

工具/命令缓存
├── JSON 快照
│   └── @lru_cache(maxsize=1) → 避免重复加载
└── 注册表
    └── 不可变对象 → 线程安全

上下文缓存
├── 会话状态
│   └── 内存缓存 → 快速访问
└── Fork 子智能体
    └── 共享父缓存 → 降低成本

不可变设计哲学

大量使用 @dataclass(frozen=True) 创建不可变对象:

@dataclass(frozen=True)
classPortContext:
    source_root: Path
    tests_root: Path
    python_file_count: int
# ...

不可变设计的优势:

优势	说明
线程安全	多线程访问无需加锁
可预测性	对象状态不会意外改变
易于调试	状态变化清晰可追踪
缓存友好	不可变对象可以安全缓存

扩展优先设计

系统设计了多个扩展点,方便第三方集成:

扩展层次
├── Level 1: 技能 (Skills)
│   └── Markdown 定义的 Prompt 片段
├── Level 2: 钩子 (Hooks)
│   └── 事件驱动的自定义逻辑
├── Level 3: 插件 (Plugins)
│   └── 完整的功能包 (技能 + 钩子 + MCP)
└── Level 4: MCP 协议
    └── 标准化的工具和资源提供

这种分层扩展的好处:

入门门槛低 (技能只需要写 Markdown)
功能强大 (插件可以提供完整功能)
标准化 (MCP 协议保证互操作性)

安全沙箱机制

多层安全防护,保护用户不受危险操作伤害:

┌─────────────────────────────────────┐
│  第 1 层: 工具注册时的权限声明      │
│  → 标记工具的危险程度              │
├─────────────────────────────────────┤
│  第 2 层: 调用前的权限检查          │
│  → 规则匹配、分类器判断            │
├─────────────────────────────────────┤
│  第 3 层: 执行时的沙箱隔离          │
│  → 文件系统访问限制                │
├─────────────────────────────────────┤
│  第 4 层: 用户确认                  │
│  → 危险操作需要用户明确同意        │
└─────────────────────────────────────┘

渐进式复杂度

系统支持从简单到复杂的使用模式:

Simple Mode (简单模式)
├── 只暴露 3 个基础工具 (Bash/Read/Edit)
├── 减少认知负担
└── 适合快速开始

       ↓
Standard Mode (标准模式)
├── 所有工具可用
├── 权限检查
└── 适合日常使用

       ↓
Advanced Mode (高级模式)
├── 多 Agent 协作
├── 技能系统
└── 适合复杂任务

这种设计让新手可以快速上手,而高级用户也能发挥全部能力。

整体架构图

有了设计思想的理解,再来看整体架构会更清晰。

系统架构全景图

┌─────────────────────────────────────────────────────────────┐
│                        用户界面层                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │
│  │ CLI 入口      │  │ Web UI       │  │ Vim 模式     │        │
│  │ main.py      │  │ (TypeScript) │  │ (State)      │        │
│  └──────┬───────┘  └──────────────┘  └──────────────┘        │
└─────────┼──────────────────────────────────────────────────────┘
          │
          ↓
┌─────────────────────────────────────────────────────────────┐
│                        运行时调度层                            │
│  ┌──────────────────────────────────────────────────────┐   │
│  │         PortRuntime (runtime.py)                     │   │
│  │  ┌─────────────┐  ┌─────────────┐  ┌──────────────┐ │   │
│  │  │ route_prompt│  │bootstrap_   │  │ run_turn_loop│ │   │
│  │  │  路由匹配   │  │ session     │  │  多轮对话    │ │   │
│  │  └──────┬──────┘  └──────┬──────┘  └──────┬───────┘ │   │
│  └─────────┼────────────────┼────────────────┼─────────┘   │
└────────────┼────────────────┼────────────────┼─────────────┘
             │                │                │
             ↓                ↓                ↓
┌─────────────────────────────────────────────────────────────┐
│                        执行协调层                              │
│  ┌──────────────────────────────────────────────────────┐   │
│  │      ExecutionRegistry (execution_registry.py)        │   │
│  │  ┌──────────────┐          ┌──────────────┐         │   │
│  │  │ MirroredCommand│        │ MirroredTool │         │   │
│  │  │   统一命令接口│          │ 统一工具接口 │         │   │
│  │  └──────┬───────┘          └──────┬───────┘         │   │
│  └─────────┼────────────────────────┼───────────────────┘   │
└────────────┼────────────────────────┼───────────────────────┘
             │                        │
             ↓                        ↓
┌─────────────────────────────────────────────────────────────┐
│                        资源管理层                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │
│  │ Commands     │  │ Tools        │  │ Permissions  │        │
│  │ commands.py  │  │ tools.py     │  │ permissions  │        │
│  │ 207 个命令   │  │ 184 个工具   │  │ 权限控制     │        │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘        │
│         │                 │                 │               │
│  ┌──────┴───────┐  ┌──────┴───────┐  ┌──────┴───────┐       │
│  │CommandGraph  │  │  ToolPool    │  │ Permission   │       │
│  │ 命令分类图   │  │  工具池      │  │ Context       │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
└─────────┼─────────────────────────────────────────────────────┘
          │
          ↓
┌─────────────────────────────────────────────────────────────┐
│                        查询引擎层                              │
│  ┌──────────────────────────────────────────────────────┐   │
│  │      QueryEnginePort (query_engine.py)               │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐ │   │
│  │  │submit_message│  │stream_submit │  │compact_messages│ │   │
│  │  │  提交消息   │  │  流式处理    │  │  上下文压缩  │ │   │
│  │  └──────┬──────┘  └──────┬───────┘  └──────┬───────┘ │   │
│  └─────────┼────────────────┼────────────────┼─────────┘   │
└────────────┼────────────────┼────────────────┼─────────────┘
             │                │                │
             ↓                ↓                ↓
┌─────────────────────────────────────────────────────────────┐
│                        持久化层                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │
│  │SessionStore  │  │Transcript    │  │HistoryLog    │        │
│  │会话存储      │  │转录存储      │  │历史记录      │        │
│  └──────────────┘  └──────────────┘  └──────────────┘        │
└─────────────────────────────────────────────────────────────┘
          │
          ↓
┌─────────────────────────────────────────────────────────────┐
│                        扩展层                                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │
│  │ Skills       │  │ Plugins     │  │ MCP Servers  │        │
│  │ 技能系统     │  │ 插件系统    │  │ MCP 协议     │        │
│  └──────────────┘  └──────────────┘  └──────────────┘        │
└─────────────────────────────────────────────────────────────┘
          │
          ↓
┌─────────────────────────────────────────────────────────────┐
│                        外部服务层                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐        │
│  │ LLM APIs     │  │ Git         │  │ File System  │        │
│  │ 多模型支持   │  │ 版本控制    │  │ 文件操作      │        │
│  └──────────────┘  └──────────────┘  └──────────────┘        │
└─────────────────────────────────────────────────────────────┘

数据流向图

从用户输入到最终结果的完整数据流:

用户输入: "帮我修复登录 bug"
    ↓
┌─────────────────────────────────────┐
│  PortRuntime.route_prompt()        │
│  1. 提取关键词: [修复, 登录, bug]  │
│  2. 匹配命令/工具                   │
│     - 可能匹配: Edit, Grep, Read    │
│  3. 返回匹配列表                    │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  ExecutionRegistry                  │
│  1. 构建命令包装器                  │
│  2. 构建工具包装器                  │
│  3. 准备执行上下文                  │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  QueryEnginePort.submit_message()  │
│  1. 检查最大轮次                    │
│  2. 构建 Prompt                     │
│     - 系统提示词                    │
│     - 可用工具列表                  │
│     - 用户输入                      │
│  3. 调用 LLM API                    │
│  4. 解析工具调用                    │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  工具执行阶段                        │
│  ┌─────────────────────────────┐    │
│  │ 1. Grep 搜索 "login"       │    │
│  │    → 找到 src/auth/login.ts│    │
│  │                           │    │
│  │ 2. Read 读取文件内容       │    │
│  │    → 分析代码逻辑          │    │
│  │                           │    │
│  │ 3. Edit 修改代码          │    │
│  │    → 修复 bug             │    │
│  │                           │    │
│  │ 4. Bash 运行测试          │    │
│  │    → 验证修复效果         │    │
│  └─────────────────────────────┘    │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  结果反馈                           │
│  1. 收集工具执行结果                │
│  2. 检查权限拒绝                    │
│  3. 统计 Token 使用                 │
│  4. 生成 TurnResult                │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  持久化和显示                       │
│  1. 保存会话到 SessionStore        │
│  2. 记录到 HistoryLog              │
│  3. 输出结果给用户                  │
└─────────────────────────────────────┘

Prompt 组装流程图

Prompt 如何从零开始组装完整上下文:

系统初始化
    ↓
┌─────────────────────────────────────┐
│  1. 运行 Setup (setup.py)          │
│     - 检查环境                      │
│     - 加载配置                      │
│     - 初始化状态                    │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  2. 加载资源                        │
│     - CommandGraph: 命令分类图      │
│     - ToolPool: 工具池              │
│     - PermissionContext: 权限上下文 │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  3. 构建 Prompt 层次                │
│  ┌─────────────────────────────┐    │
│  │ 静态部分 (可缓存)           │    │
│  │ - AI 身份定义               │    │
│  │ - 安全原则                  │    │
│  │ - 任务执行指南              │    │
│  │ - 工具使用原则              │    │
│  └─────────────────────────────┘    │
│             ↓                        │
│  ┌─────────────────────────────┐    │
│  │ 动态分界线                   │    │
│  │ __SYSTEM_PROMPT_DYNAMIC_    │    │
│  │ BOUNDARY__                  │    │
│  └─────────────────────────────┘    │
│             ↓                        │
│  ┌─────────────────────────────┐    │
│  │ 动态部分 (会话专属)         │    │
│  │ - 当前可用的命令列表        │    │
│  │ - 当前可用的工具列表        │    │
│  │ - 会话特定指南              │    │
│  │ - 持久记忆 (如果有)         │    │
│  └─────────────────────────────┘    │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  4. 最终 Prompt                      │
│  [静态部分]                         │
│  [分界线]                           │
│  [动态部分]                         │
│  [用户消息历史]                     │
│  [用户当前输入]                     │
└─────────────────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  5. 发送给 LLM                      │
│  (静态部分可能命中缓存)             │
└─────────────────────────────────────┘

自修复机制流程图

工具调用失败后如何自动修复:

工具调用请求
    ↓
┌─────────────────────────────────────┐
│  执行工具                           │
│  (例如: Edit 修改文件)              │
└────────────┬────────────────────────┘
             ↓
    ┌────────┴────────┐
    │                 │
    ↓                 ↓
成功              失败
    │                 │
    │         ┌───────┴────────┐
    │         │                │
    │         ↓                ↓
    │    工具错误        权限拒绝
    │         │                │
    │         ↓                ↓
    │  ┌────────────────────┐  │
    │  │ 返回错误详情:      │  │
    │  │ - 错误类型         │  │
    │  │ - 错误信息         │  │
    │  │ - 建议修复方案     │  │
    │  └─────────┬──────────┘  │
    │            ↓              │
    └─────┬──────┴──────┐       │
          ↓             ↓       ↓
┌───────────────────────────────────┐
│  AI 接收反馈                       │
│  1. 分析失败原因                   │
│     - 文件不存在?                 │
│     - 权限不足?                   │
│     - 语法错误?                   │
│  2. 评估修复策略                  │
│     - 重试?                       │
│     - 调整参数?                   │
│     - 尝试替代方案?               │
└───────────┬───────────────────────┘
            ↓
┌───────────────────────────────────┐
│  Prompt 中的指导原则起作用:       │
│  "If an approach fails,          │
│   diagnose why before switching   │
│   tactics - read the error,      │
│   check your assumptions, try    │
│   a focused fix"                  │
└───────────┬───────────────────────┘
            ↓
┌───────────────────────────────────┐
│  AI 决策                           │
│  ┌─────────────────────────────┐  │
│  │ 场景 1: 可重试的错误        │  │
│  │ → 调整参数后重试           │  │
│  └─────────────────────────────┘  │
│  ┌─────────────────────────────┐  │
│  │ 场景 2: 需要调整策略        │  │
│  │ → 尝试其他工具/方法        │  │
│  └─────────────────────────────┘  │
│  ┌─────────────────────────────┐  │
│  │ 场景 3: 无法修复            │  │
│  │ → 向用户报告错误           │  │
│  └─────────────────────────────┘  │
└───────────┬───────────────────────┘
            ↓
┌───────────────────────────────────┐
│  执行新方案                        │
│  (循环回到工具调用)                │
└───────────────────────────────────┘
            │
            ↓
    ┌───────┴───────┐
    ↓               ↓
成功              达到最大重试次数
    │               │
    ↓               ↓
完成任务         向用户报告失败

会话生命周期图

一个会话从创建到销毁的完整过程:

会话启动
    ↓
┌─────────────────────────────────────┐
│  bootstrap_session()               │
│  1. 构建上下文 (PortContext)       │
│     - 扫描项目文件                  │
│     - 统计 Python/测试/资源文件    │
│  2. 运行初始化 (run_setup)         │
│     - 加载配置                      │
│     - 检查权限                      │
│  3. 创建历史记录 (HistoryLog)      │
│  4. 创建查询引擎 (QueryEnginePort)  │
│  5. 生成会话 ID                    │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  会话活跃期                          │
│  ┌─────────────────────────────┐    │
│  │ 多轮对话循环:              │    │
│  │ 1. route_prompt()          │    │
│  │ 2. submit_message()        │    │
│  │ 3. 工具执行                 │    │
│  │ 4. 结果反馈                │    │
│  │ 5. 历史记录                │    │
│  └─────────────────────────────┘    │
│             ↓                        │
│  ┌─────────────────────────────┐    │
│  │ 上下文管理:                │    │
│  │ - 增加消息到历史          │    │
│  │ - 更新 Token 统计          │    │
│  │ - 检查是否需要压缩        │    │
│  └─────────────────────────────┘    │
│             ↓                        │
│  ┌─────────────────────────────┐    │
│  │ 定期持久化:                 │    │
│  │ - 保存到 SessionStore       │    │
│  │ - 更新 Transcript           │    │
│  └─────────────────────────────┘    │
└────────────┬────────────────────────┘
             ↓
    ┌────────┴────────┐
    │                 │
    ↓                 ↓
正常结束           异常终止
    │                 │
    │         ┌───────┴────────┐
    │         │                │
    │         ↓                ↓
    │    Token 超限         用户中断
    │         │                │
    │         ↓                ↓
    └─────┬──────┴──────┐       │
          ↓             ↓       ↓
┌───────────────────────────────────┐
│  会话清理                           │
│  1. 保存最终状态                   │
│     - SessionStore 完整保存       │
│     - Transcript 同步到磁盘        │
│  2. 清理资源                       │
│     - 释放内存                     │
│     - 关闭连接                     │
│  3. 生成报告                       │
│     - 使用量统计                   │
│     - 执行摘要                     │
└───────────┬───────────────────────┘
            ↓
┌───────────────────────────────────┐
│  会话结束                          │
│  (用户可以恢复会话继续对话)        │
└───────────────────────────────────┘

多 Agent 协作图

多个 Agent 如何协同完成任务:

用户请求: "重构用户模块"
    ↓
┌─────────────────────────────────────┐
│  主 Agent (General-Purpose)         │
│  职责: 任务分解和协调               │
│  1. 分析任务复杂度                 │
│  2. 决定是否需要子 Agent           │
│  3. 分配子任务                     │
└────────────┬────────────────────────┘
             ↓
    ┌────────┼────────┬────────┐
    ↓        ↓        ↓        ↓
┌───────┐ ┌───────┐ ┌───────┐ ┌───────┐
│Explore│ │ Plan  │ │Implement│ │Verify │
│ Agent │ │ Agent │ │  Agent  │ │ Agent │
└───┬───┘ └───┬───┘ └───┬───┘ └───┬───┘
    │        │        │        │
    ↓        ↓        ↓        ↓
任务1:    任务2:    任务3:    任务4:
搜索代码  设计方案  实现重构  验证测试
    │        │        │        │
    ↓        ↓        ↓        ↓
使用工具:  使用工具:  使用工具:  使用工具:
- Glob    - Read    - Edit    - Bash
- Grep    - (只读)  - Write   - Read
- Read    -        - Bash    - Grep
    │        │        │        │
    └────────┼────────┼────────┘
             ↓        ↓
┌─────────────────────────────────────┐
│  子 Agent 结果汇总                   │
│  1. Explore: 找到 15 个相关文件      │
│  2. Plan: 设计了 3 个改进点         │
│  3. Implement: 完成代码重构         │
│  4. Verify: 测试全部通过             │
└────────────┬────────────────────────┘
             ↓
┌─────────────────────────────────────┐
│  主 Agent 整合报告                   │
│  1. 合并所有子 Agent 的结果         │
│  2. 生成完整的任务报告              │
│  3. 返回给用户                      │
└────────────┬────────────────────────┘
             ↓
用户看到最终结果

扩展系统架构图

Skills、Plugins、MCP 如何集成到系统中:

┌─────────────────────────────────────────────────────────┐
│                    技能系统                            │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │ 内置技能     │  │ 用户技能     │  │ 项目技能     │  │
│  │ src/skills/  │  │ ~/.claude/   │  │ .claude/     │  │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  │
│         │                 │                 │            │
│         └─────────────────┼─────────────────┘            │
│                           ↓                               │
│              ┌──────────────────────┐                   │
│              │ SkillLoader          │                   │
│              │ - 解析 .md 文件      │                   │
│              │ - 提取 metadata     │                   │
│              │ - 注册到系统        │                   │
│              └──────────┬───────────┘                   │
└───────────────────────────┼───────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────┐
│                    插件系统                            │
│  ┌─────────────────────────────────────────────────┐   │
│  │ Plugin                                          │   │
│  │ id: plugin-name@builtin                         │   │
│  │ features:                                       │   │
│  │   - skills (提供技能)                          │   │
│  │   - hooks (提供钩子)                           │   │
│  │   - mcp (提供 MCP 服务)                        │   │
│  └────────────┬────────────────────────────────────┘   │
│               ↓                                          │
│  ┌──────────────────────┐                               │
│  │ PluginManager        │                               │
│  │ - 加载插件            │                               │
│  │ - 管理生命周期        │                               │
│  │ - 协调功能            │                               │
│  └──────────┬───────────┘                               │
└───────────────┼──────────────────────────────────────────┘
                ↓
┌─────────────────────────────────────────────────────────┐
│                 MCP 协议层                             │
│  ┌─────────────────────────────────────────────────┐   │
│  │ MCP Server                                       │   │
│  │ capabilities:                                    │   │
│  │   - tools/ (提供工具)                           │   │
│  │   - resources/ (提供资源)                        │   │
│  │   - prompts/ (提供提示词)                       │   │
│  └────────────┬────────────────────────────────────┘   │
│               ↓                                          │
│  ┌──────────────────────┐                               │
│  │ MCP Client            │                               │
│  │ - 连接服务器          │                               │
│  │ - 调用工具            │                               │
│  │ - 获取资源            │                               │
│  └──────────┬───────────┘                               │
└───────────────┼──────────────────────────────────────────┘
                ↓
┌─────────────────────────────────────────────────────────┐
│                 集成点                                 │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │ ToolPool     │  │ CommandGraph │  │ SystemInit   │  │
│  │ 添加 MCP 工具│  │ 添加命令    │  │ 添加 Prompt  │  │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  │
│         │                 │                 │          │
│         └─────────────────┼─────────────────┘          │
│                           ↓                             │
│              ┌──────────────────────┐                   │
│              │ ExecutionRegistry    │                   │
│              │ 统一的执行入口       │                   │
│              └──────────────────────┘                   │
└─────────────────────────────────────────────────────────┘

缓存层次架构图

系统中的多级缓存设计:

┌─────────────────────────────────────────────────────────┐
│                 LLM API 缓存                            │
│  ┌─────────────────────────────────────────────────┐   │
│  │ Prompt Cache (静态部分)                         │   │
│  │ - 跨用户/组织共享                                │   │
│  │ - 大幅减少 token 消耗                            │   │
│  │ - 命中率: 80-90%                                 │   │
│  └────────────┬────────────────────────────────────┘   │
│               ↓                                          │
│  ┌──────────────────────┐                               │
│  │ 缓存键:               │                               │
│  │ - 模型名称            │                               │
│  │ - 静态 Prompt 内容    │                               │
│  │ - 动态分界线          │                               │
│  └──────────────────────┘                               │
└─────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────┐
│               本地内存缓存                              │
│  ┌─────────────────────────────────────────────────┐   │
│  │ 工具/命令快照缓存                                │   │
│  │ @lru_cache(maxsize=1)                           │   │
│  │ - JSON 快照加载                                  │   │
│  │ - 避免重复 I/O                                   │   │
│  │ - 生命周期: 进程级                              │   │
│  └────────────┬────────────────────────────────────┘   │
│               ↓                                          │
│  ┌─────────────────────────────────────────────────┐   │
│  │ 会话状态缓存                                    │   │
│  │ - 当前活跃会话                                  │   │
│  │ - 消息历史                                      │   │
│  │ - Token 统计                                    │   │
│  └────────────┬────────────────────────────────────┘   │
└───────────────┼──────────────────────────────────────────┘
                ↓
┌─────────────────────────────────────────────────────────┐
│              Fork 子智能体缓存                          │
│  ┌─────────────────────────────────────────────────┐   │
│  │ Fork 子智能体共享父缓存                          │   │
│  │ - 复制消息历史                                   │   │
│  │ - 使用相同占位符                                 │   │
│  │ - 缓存键保持一致                                 │   │
│  │ - 命中率: 95%+                                   │   │
│  └────────────┬────────────────────────────────────┘   │
│               ↓                                          │
│  优势:                                                   │
│  1. 极低成本启动子智能体                                │
│  2. 可以并行执行多个子智能体                            │
│  3. 不污染主上下文                                      │
└─────────────────────────────────────────────────────────┘
                            ↓
┌─────────────────────────────────────────────────────────┐
│               持久化缓存                                │
│  ┌─────────────────────────────────────────────────┐   │
│  │ SessionStore                                     │   │
│  │ - 会话完整保存                                    │   │
│  │ - 支持会话恢复                                    │   │
│  │ - JSON 格式存储                                  │   │
│  └────────────┬────────────────────────────────────┘   │
│               ↓                                          │
│  ┌─────────────────────────────────────────────────┐   │
│  │ Transcript                                        │   │
│  │ - 转录记录                                        │   │
│  │ - 支持重放                                        │   │
│  │ - 压缩后丢弃旧数据                                │   │
│  └─────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘

Prompt 系统解读

Prompt 是整个系统的大脑,告诉 AI 应该怎么思考、怎么行动。Claude Code 的 Prompt 设计非常精妙,分为静态和动态两部分。

Prompt 组装流程

从 system_init.py 可以看到完整的组装过程:

defbuild_system_init_message(trusted: bool = True) -> str:
    setup = run_setup(trusted=trusted)
    commands = get_commands()
    tools = get_tools()
    lines = [
'# System Init',
'',
f'Trusted: {setup.trusted}',
f'Built-in command names: {len(built_in_command_names())}',
f'Loaded command entries: {len(commands)}',
f'Loaded tool entries: {len(tools)}',
'',
'Startup steps:',
        *(f'- {step}'for step in setup.setup.startup_steps()),
    ]
return'\n'.join(lines)

这里的关键是 trusted 模式,决定了 AI 能执行哪些危险操作。

Prompt 结构分层

层级	内容	缓存策略
静态部分	身份定义、系统规则、任务执行指南	跨用户/组织可缓存
动态分界线	__SYSTEM_PROMPT_DYNAMIC_BOUNDARY__	缓存隔离标志
动态部分	会话特定指南、持久记忆、环境信息	每个会话独立

核心 Prompt 片段

身份与安全

You are an interactive agent that helps users with software engineering tasks.
IMPORTANT: Assist with authorized security testing, defensive security, CTF
challenges, and educational contexts. Refuse requests for destructive techniques.

这定义了 AI 的身份:软件工程助手,同时划定了安全边界。

任务执行指南

- Do not create files unless they're absolutely necessary.
- Generally prefer editing an existing file to creating a new one.
- Avoid giving time estimates or predictions.
- If an approach fails, diagnose why before switching tactics.

这些指导原则让 AI 既不过度承诺,也不会盲目行动。

工具使用原则

- Do NOT use Bash to run commands when a relevant dedicated tool is provided.
- To read files use Read instead of cat, head, tail, or sed.
- To edit files use Edit instead of sed or awk.
- To search content of files, use Grep instead of grep or rg.

这里强制使用专用工具,让 AI 的操作更可追踪、更安全。

Prompt 的智能之处

1. 缓存优化 静态部分可以缓存,减少每次请求的 token 消耗。动态分界线确保每个会话的个性化内容能独立更新。

2. 权限感知 Prompt 中包含当前可用的命令和工具列表,AI 就不会尝试调用不存在的工具。

3. 上下文压缩 当对话历史过长时,系统会自动压缩之前的消息,Prompt 中会包含"此会话是之前对话的继续"这样的说明,让 AI 知道上下文已被压缩。

自修复机制深度剖析

这是最让人惊叹的部分:AI 怎么知道自己错了,然后自动修?

核心原理:工具调用反馈循环

看 query_engine.py 中的 submit_message 方法:

defsubmit_message(
    self,
    prompt: str,
    matched_commands: tuple[str, ...] = (),
    matched_tools: tuple[str, ...] = (),
    denied_tools: tuple[PermissionDenial, ...] = (),
) -> TurnResult:
# ... 检查最大轮次 ...

# 构建输出摘要
    summary_lines = [
f'Prompt: {prompt}',
f'Matched commands: {", ".join(matched_commands) if matched_commands else"none"}',
f'Matched tools: {", ".join(matched_tools) if matched_tools else"none"}',
f'Permission denials: {len(denied_tools)}',
    ]

# 执行命令和工具
    output = self._format_output(summary_lines)

# 返回结果(包含所有上下文)
return TurnResult(
        prompt=prompt,
        output=output,
        matched_commands=matched_commands,
        matched_tools=matched_tools,
        permission_denials=denied_tools,
        usage=self.total_usage,
        stop_reason=stop_reason,
    )

关键点:工具执行结果(无论成功失败)都会完整反馈给 AI。

四层错误恢复

第 1 层: Prompt Too Long 恢复

# runtime.py 中的压缩逻辑
classPortRuntime:
defrun_turn_loop(self, prompt: str, limit: int = 5, max_turns: int = 3):
        engine = QueryEnginePort.from_workspace()
        engine.config = QueryEngineConfig(
            max_turns=max_turns,
            compact_after_turns=12# 达到 12 轮后自动压缩
        )

当对话轮次达到阈值时,系统会自动压缩历史消息。

第 2 层:输出 Token 超限恢复

# query_engine.py 中的预算检查
if projected_usage.input_tokens + projected_usage.output_tokens > self.config.max_budget_tokens:
    stop_reason = 'max_budget_reached'

一旦超过预算,会停止执行并向用户报告。

第 3 层:权限拒绝恢复

# runtime.py 中的权限推断
def_infer_permission_denials(self, matches: list[RoutedMatch]) -> list[PermissionDenial]:
    denials: list[PermissionDenial] = []
formatchin matches:
ifmatch.kind == 'tool'and'bash'inmatch.name.lower():
            denials.append(
                PermissionDenial(
                    tool_name=match.name,
                    reason='destructive shell execution remains gated in the Python port'
                )
            )
return denials

危险的 shell 命令会被自动拦截,并向 AI 返回拒绝原因。

第 4 层:工具错误自然恢复

# tools.py 中的工具执行
defexecute_tool(name: str, payload: str = '') -> ToolExecution:
    module = get_tool(name)
if module isNone:
return ToolExecution(
            name=name,
            source_hint='',
            payload=payload,
            handled=False,
            message=f'Unknown mirrored tool: {name}'
        )
    action = f"Mirrored tool '{module.name}' from {module.source_hint} would handle payload {payload!r}."
return ToolExecution(
        name=module.name,
        source_hint=module.source_hint,
        payload=payload,
        handled=True,
        message=action
    )

工具执行失败会返回清晰的错误信息,AI 看到后可以调整策略。

Prompt 中的自修复指导

关键指导原则:

If an approach fails, diagnose why before switching tactics—read the error, check
your assumptions, try a focused fix. Don't retry identical action blindly,
but don't abandon a viable approach after a single failure either.

翻译过来就是:

失败后先诊断,不要盲目重试
检查错误、验证假设、尝试针对性修复
不要单次失败就放弃可行的方案

这个指导让 AI 既能快速试错,又不会陷入死循环。

运行时引擎架构

PortRuntime 是整个系统的调度中心,负责路由、会话管理和执行协调。

路由匹配机制

# runtime.py 中的路由逻辑
defroute_prompt(self, prompt: str, limit: int = 5) -> list[RoutedMatch]:
# 提取关键词
    tokens = {
        token.lower()
for token in prompt.replace('/', ' ').replace('-', ' ').split()
if token
    }

# 分别匹配命令和工具
    by_kind = {
'command': self._collect_matches(tokens, PORTED_COMMANDS, 'command'),
'tool': self._collect_matches(tokens, PORTED_TOOLS, 'tool'),
    }

# 优先级选择
    selected: list[RoutedMatch] = []
for kind in ('command', 'tool'):
if by_kind[kind]:
            selected.append(by_kind[kind].pop(0))

# 补充剩余匹配
    leftovers = sorted(
        [matchfor matches in by_kind.values() formatchin matches],
        key=lambda item: (-item.score, item.kind, item.name),
    )
    selected.extend(leftovers[: max(0, limit - len(selected))])
return selected[:limit]

这个机制确保:

命令和工具都能被识别
优先匹配类型(命令优先)
按相关度排序

会话启动流程

# runtime.py 中的会话初始化
defbootstrap_session(self, prompt: str, limit: int = 5) -> RuntimeSession:
# 1. 构建上下文
    context = build_port_context()

# 2. 运行初始化
    setup_report = run_setup(trusted=True)
    setup = setup_report.setup

# 3. 创建历史记录
    history = HistoryLog()
    engine = QueryEnginePort.from_workspace()

# 4. 记录初始化信息
    history.add('context', f'python_files={context.python_file_count}')
    history.add('registry', f'commands={len(PORTED_COMMANDS)}, tools={len(PORTED_TOOLS)}')

# 5. 路由匹配
    matches = self.route_prompt(prompt, limit=limit)

# 6. 构建执行注册表
    registry = build_execution_registry()

# 7. 执行命令和工具
    command_execs = tuple(
        registry.command(match.name).execute(prompt)
formatchin matches ifmatch.kind == 'command'and registry.command(match.name)
    )
    tool_execs = tuple(
        registry.tool(match.name).execute(prompt)
formatchin matches ifmatch.kind == 'tool'and registry.tool(match.name)
    )

# 8. 推断权限拒绝
    denials = tuple(self._infer_permission_denials(matches))

# 9. 提交消息
    stream_events = tuple(engine.stream_submit_message(
        prompt,
        matched_commands=tuple(match.name formatchin matches ifmatch.kind == 'command'),
        matched_tools=tuple(match.name formatchin matches ifmatch.kind == 'tool'),
        denied_tools=denials,
    ))

    turn_result = engine.submit_message(...)

# 10. 持久化会话
    persisted_session_path = engine.persist_session()

# 11. 返回完整会话对象
return RuntimeSession(...)

这个流程的精妙之处:

每一步都有明确的职责
所有操作都被记录在历史中
会话可以被持久化和恢复

多轮对话支持

# runtime.py 中的多轮循环
defrun_turn_loop(self, prompt: str, limit: int = 5, max_turns: int = 3, structured_output: bool = False) -> list[TurnResult]:
    engine = QueryEnginePort.from_workspace()
    engine.config = QueryEngineConfig(max_turns=max_turns, structured_output=structured_output)

    matches = self.route_prompt(prompt, limit=limit)
    command_names = tuple(match.name formatchin matches ifmatch.kind == 'command')
    tool_names = tuple(match.name formatchin matches ifmatch.kind == 'tool')

    results: list[TurnResult] = []
for turn inrange(max_turns):
        turn_prompt = prompt if turn == 0elsef'{prompt} [turn {turn + 1}]'
        result = engine.submit_message(turn_prompt, command_names, tool_names, ())
        results.append(result)

# 如果没有完成,提前退出
if result.stop_reason != 'completed':
break
return results

支持多轮对话的好处:

复杂任务可以分解为多个步骤
每轮都有独立的上下文
可以根据前一轮的结果调整策略

工具系统详解

工具是 AI 与外部世界交互的桥梁,理解工具系统就能理解 AI 的能力边界。

工具注册表

# tools.py 中的工具加载
@lru_cache(maxsize=1)
defload_tool_snapshot() -> tuple[PortingModule, ...]:
    raw_entries = json.loads(SNAPSHOT_PATH.read_text())
returntuple(
        PortingModule(
            name=entry['name'],
            responsibility=entry['responsibility'],
            source_hint=entry['source_hint'],
            status='mirrored',
        )
for entry in raw_entries
    )

PORTED_TOOLS = load_tool_snapshot()

工具定义存储在 JSON 快照中,通过 LRU 缓存加载,性能最优。

工具查询和过滤

# tools.py 中的工具查找
defget_tool(name: str) -> PortingModule | None:
    needle = name.lower()
for module in PORTED_TOOLS:
if module.name.name.lower() == needle:
return module
returnNone

defget_tools(
    simple_mode: bool = False,
    include_mcp: bool = True,
    permission_context: ToolPermissionContext | None = None,
) -> tuple[PortingModule, ...]:
    tools = list(PORTED_TOOLS)
if simple_mode:
        tools = [module for module in tools if module.name in {'BashTool', 'FileReadTool', 'FileEditTool'}]
ifnot include_mcp:
        tools = [module for module in tools if'mcp'notin module.name.lower()]
return filter_tools_by_permission_context(tuple(tools), permission_context)

过滤机制:

简单模式只保留基础工具
可以排除 MCP 工具
权限上下文可以阻止特定工具

权限控制系统

# permissions.py 中的权限上下文
@dataclass(frozen=True)
classToolPermissionContext:
    deny_names: frozenset[str] = field(default_factory=frozenset)
    deny_prefixes: tuple[str, ...] = ()

defblocks(self, tool_name: str) -> bool:
        lowered = tool_name.lower()
return (
            lowered inself.deny_names
orany(lowered.startswith(prefix) for prefix inself.deny_prefixes)
        )

设计亮点:

使用 frozenset 确保 O(1) 查找
支持精确匹配和前缀匹配
不可变设计防止意外修改

工具执行

# tools.py 中的工具执行
defexecute_tool(name: str, payload: str = '') -> ToolExecution:
    module = get_tool(name)
if module isNone:
return ToolExecution(
            name=name,
            source_hint='',
            payload=payload,
            handled=False,
            message=f'Unknown mirrored tool: {name}'
        )
    action = f"Mirrored tool '{module.name}' from {module.source_hint} would handle payload {payload!r}."
return ToolExecution(
        name=module.name,
        source_hint=module.source_hint,
        payload=payload,
        handled=True,
        message=action
    )

返回结构包含:

是否处理成功
具体消息内容
错误原因(如果失败)

命令系统架构

命令系统提供斜杠命令,用户可以快速执行特定功能。

命令注册表

# commands.py 中的命令加载
@lru_cache(maxsize=1)
defload_command_snapshot() -> tuple[PortingModule, ...]:
    raw_entries = json.loads(SNAPSHOT_PATH.read_text())
returntuple(
        PortingModule(
            name=entry['name'],
            responsibility=entry['responsibility'],
            source_hint=entry['source_hint'],
            status='mirrored',
        )
for entry in raw_entries
    )

PORTED_COMMANDS = load_command_snapshot()

和工具系统类似,命令也通过 JSON 快照定义。

命令分类

类型	说明	示例
内置命令	系统自带,始终可用	/help, /status, /clear
插件命令	由插件提供,可启用/禁用	各种扩展命令
技能命令	用户自定义的技能脚本	/commit, /review

命令执行

# commands.py 中的命令执行
defexecute_command(name: str, prompt: str = '') -> CommandExecution:
    module = get_command(name)
if module isNone:
return CommandExecution(
            name=name,
            source_hint='',
            prompt=prompt,
            handled=False,
            message=f'Unknown mirrored command: {name}'
        )
    action = f"Mirrored command '{module.name}' from {module.source_hint} would handle prompt {prompt!r}."
return CommandExecution(
        name=module.name,
        source_hint=module.source_hint,
        prompt=prompt,
        handled=True,
        message=action
    )

命令执行和工具执行的接口一致,便于统一管理。

执行注册表设计

ExecutionRegistry 提供了命令和工具的统一访问接口。

注册表结构

# execution_registry.py 中的注册表
@dataclass(frozen=True)
classExecutionRegistry:
    commands: tuple[MirroredCommand, ...]
    tools: tuple[MirroredTool, ...]

defcommand(self, name: str) -> MirroredCommand | None:
        lowered = name.lower()
for command inself.commands:
if command.name.name.lower() == lowered:
return command
returnNone

deftool(self, name: str) -> MirroredTool | None:
        lowered = name.lower()
for tool inself.tools:
if tool.name.name.lower() == lowered:
return tool
returnNone

设计优势:

统一的访问接口
大小写不敏感的查找
返回类型统一(都是包装器)

镜像包装器

# execution_registry.py 中的包装器
@dataclass(frozen=True)
classMirroredCommand:
    name: str
    source_hint: str

defexecute(self, prompt: str) -> str:
return execute_command(self.name, prompt).message

@dataclass(frozen=True)
classMirroredTool:
    name: str
    source_hint: str

defexecute(self, payload: str) -> str:
return execute_tool(self.name, payload).message

包装器的作用:

隐藏底层实现细节
提供一致的执行接口
便于统一管理和测试

会话管理系统

会话管理确保对话可以持久化、恢复和压缩。

会话存储

# session_store.py 中的会话持久化
@dataclass(frozen=True)
classStoredSession:
    session_id: str
    messages: tuple[str, ...]
    input_tokens: int
    output_tokens: int

defsave_session(session: StoredSession, directory: Path | None = None) -> Path:
    target_dir = directory or DEFAULT_SESSION_DIR
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / f'{session.session_id}.json'
    path.write_text(json.dumps(asdict(session), indent=2))
return path

defload_session(session_id: str, directory: Path | None = None) -> StoredSession:
    target_dir = directory or DEFAULT_SESSION_DIR
    data = json.loads((target_dir / f'{session_id}.json').read_text())
return StoredSession(
        session_id=data['session_id'],
        messages=tuple(data['messages']),
        input_tokens=data['input_tokens'],
        output_tokens=data['output_tokens'],
    )

会话存储包含:

唯一会话 ID
完整消息历史
Token 使用统计

转录存储

# transcript.py 中的转录管理
@dataclass
classTranscriptStore:
    entries: list[str] = field(default_factory=list)
    flushed: bool = False

defappend(self, entry: str) -> None:
self.entries.append(entry)
self.flushed = False

defcompact(self, keep_last: int = 10) -> None:
iflen(self.entries) > keep_last:
self.entries[:] = self.entries[-keep_last:]

defreplay(self) -> tuple[str, ...]:
returntuple(self.entries)

defflush(self) -> None:
self.flushed = True

转录存储的特点:

支持追加新条目
可以压缩保留最近 N 条
可以重放所有条目
有刷新标志

历史记录

# history.py 中的历史日志
@dataclass
classHistoryLog:
    events: list[HistoryEvent] = field(default_factory=list)

defadd(self, category: str, message: str) -> None:
self.events.append(HistoryEvent(category=category, message=message))

defas_markdown(self) -> str:
        lines = ['# History Log', '']
for event inself.events:
            lines.append(f'**{event.category}**: {event.message}')
return'\n'.join(lines)

历史记录用于:

跟踪执行过程
调试和问题排查
生成会话报告

查询引擎核心

QueryEnginePort 是消息提交和流式处理的核心。

消息提交流程

# query_engine.py 中的消息提交
defsubmit_message(
    self,
    prompt: str,
    matched_commands: tuple[str, ...] = (),
    matched_tools: tuple[str, ...] = (),
    denied_tools: tuple[PermissionDenial, ...] = (),
) -> TurnResult:
# 1. 检查最大轮次
iflen(self.mutable_messages) >= self.config.max_turns:
        output = f'Max turns reached before processing prompt: {prompt}'
return TurnResult(...)

# 2. 构建输出
    summary_lines = [
f'Prompt: {prompt}',
f'Matched commands: {", ".join(matched_commands)}',
f'Matched tools: {", ".join(matched_tools)}',
f'Permission denials: {len(denied_tools)}',
    ]
    output = self._format_output(summary_lines)

# 3. 更新使用量
    projected_usage = self.total_usage.add_turn(prompt, output)

# 4. 检查预算
if projected_usage.input_tokens + projected_usage.output_tokens > self.config.max_budget_tokens:
        stop_reason = 'max_budget_reached'
else:
        stop_reason = 'completed'

# 5. 保存到历史
self.mutable_messages.append(prompt)
self.transcript_store.append(prompt)
self.permission_denials.extend(denied_tools)
self.total_usage = projected_usage

# 6. 压缩(如果需要)
self.compact_messages_if_needed()

return TurnResult(...)

关键设计点:

每一步都有检查和验证
使用量实时跟踪
自动压缩历史消息
完整的状态返回

流式处理

# query_engine.py 中的流式处理
defstream_submit_message(
    self,
    prompt: str,
    matched_commands: tuple[str, ...] = (),
    matched_tools: tuple[str, ...] = (),
    denied_tools: tuple[PermissionDenial, ...] = (),
):
# 发送消息开始事件
yield {'type': 'message_start', 'session_id': self.session_id, 'prompt': prompt}

# 发送命令匹配事件
if matched_commands:
yield {'type': 'command_match', 'commands': matched_commands}

# 发送工具匹配事件
if matched_tools:
yield {'type': 'tool_match', 'tools': matched_tools}

# 发送权限拒绝事件
if denied_tools:
yield {'type': 'permission_denial', 'denials': [denial.tool_name for denial in denied_tools]}

# 提交消息
    result = self.submit_message(prompt, matched_commands, matched_tools, denied_tools)

# 发送消息增量事件
yield {'type': 'message_delta', 'text': result.output}

# 发送消息结束事件
yield {
'type': 'message_stop',
'usage': {
'input_tokens': result.usage.input_tokens,
'output_tokens': result.usage.output_tokens
        },
'stop_reason': result.stop_reason,
'transcript_size': len(self.transcript_store.entries),
    }

流式处理的好处:

实时反馈进度
可以提前终止长操作
逐步展示结果

多 Agent 机制

虽然 Python 版本是移植,但保留了多 Agent 的设计思想。

Agent 类型分类

根据原版的 Agent 系统,可以分为:

Agent 类型	职责	工具限制
General-Purpose	通用任务处理	全部可用
Explore	代码探索和搜索	只读工具
Plan	架构规划和方案设计	只读工具
Verification	实现验证和测试	只读+临时目录

Agent 协作模式

用户请求
    ↓
主 Agent 分析任务
    ↓
├── 探索 Agent (Explore)
│   └── 快速搜索代码
├── 规划 Agent (Plan)
│   └── 设计实现方案
├── 实现 Agent (Implementation)
│   └── 执行代码修改
└── 验证 Agent (Verification)
    └── 测试和验证

这种分工的好处:

每个 Agent 专注于自己的领域
可以并行执行独立任务
避免上下文污染

Fork 子智能体

Fork 机制允许创建子智能体来处理特定任务:

# Fork 模式的特点
classForkSubAgent:
"""继承父智能体完整上下文,共享 prompt cache"""

def__init__(self, parent_context):
# 复制父消息历史
self.history = parent_context.history.copy()

# 用相同占位符替换 tool_result(保持缓存键一致)
self.cache_key = parent_context.cache_key

# 添加子智能体专属指令
self.prompt = parent_context.prompt + "\n\nYou are a subagent..."

Fork 的优势:

极低成本(缓存命中率高)
不污染主上下文
可以并行执行

上下文压缩系统

长对话会消耗大量 token,需要智能压缩。

自动压缩触发

# query_engine.py 中的配置
@dataclass(frozen=True)
classQueryEngineConfig:
    max_turns: int = 8
    max_budget_tokens: int = 2000
    compact_after_turns: int = 12# 达到 12 轮后压缩
    structured_output: bool = False
    structured_retry_limit: int = 2

压缩触发条件:

消息数量超过阈值
Token 使用接近上限
手动触发 /compact 命令

压缩策略

# query_engine.py 中的压缩逻辑
defcompact_messages_if_needed(self) -> None:
iflen(self.mutable_messages) > self.config.compact_after_turns:
self.mutable_messages[:] = self.mutable_messages[-self.config.compact_after_turns :]
self.transcript_store.compact(self.config.compact_after_turns)

压缩策略:

保留最近 N 条消息
转录同步压缩
丢弃旧的工具结果

压缩后的恢复

压缩后,系统会插入恢复提示:

This session is being continued from a previous conversation that ran out of context.
The summary below covers earlier portion of the conversation.

[详细的对话摘要]

If you need specific details from before compaction, read the full transcript at: {transcript_path}
Continue conversation from where it left off without asking the user any further questions.

这样 AI 就能在压缩后继续工作,不会丢失关键信息。

权限和安全机制

权限系统保护用户免受危险操作的伤害。

多层权限检查

工具调用请求
    ↓
规则检查 (hasPermissionsToUseToolInner)
    ├── 整个工具被拒绝? → deny
    ├── 工具特定权限? → deny/ask
    ├── 安全检查 (.git, .claude)? → 必须提示
    └── bypassPermissions? → auto-allow
    ↓
模式转换
    ├── dontAsk 模式 → deny
    ├── auto 模式 → 运行分类器
    └── plan + auto → 运行分类器
    ↓
分类器 (如果需要)
    ├── 安全允许列表? → 直接允许
    ├── 两阶段 XML 分类
    └── 拒绝限制追踪
    ↓
交互处理
    ├── 交互式: 4 个源竞争
    ├── Coordinator: hooks → 分类器 → 对话框
    └── Swarm Worker: 分类器 → 转发给 leader

危险操作识别

系统识别的危险操作包括:

类型	示例	处理方式
删除操作	rm -rf, 删除分支	必须确认
强制推送	git push --force	必须确认
Git 操作	跳过 hooks, 修改配置	必须确认
外部发送	发送邮件/消息	必须确认
敏感路径	访问 .git, .claude	必须确认

沙箱机制

虽然 Python 版本简化了,但保留了沙箱概念:

# runtime.py 中的权限推断
def_infer_permission_denials(self, matches: list[RoutedMatch]) -> list[PermissionDenial]:
    denials: list[PermissionDenial] = []
formatchin matches:
# 危险的 shell 命令会被拒绝
ifmatch.kind == 'tool'and'bash'inmatch.name.lower():
            denials.append(
                PermissionDenial(
                    tool_name=match.name,
                    reason='destructive shell execution remains gated in the Python port'
                )
            )
return denials

沙箱限制:

文件系统访问受限
网络访问受限
危险命令需要确认

扩展系统设计

Claude Code 设计了多个扩展点,方便集成第三方服务。

MCP (Model Context Protocol)

MCP 允许外部服务提供工具和资源:

MCP 服务器
    ├── 工具提供 (tools/)
    ├── 资源提供 (resources/)
    └── 提示词提供 (prompts/)

MCP 作用域:

local: 项目目录的 .mcp.json
user: 用户目录的 ~/.claude/.mcp.json
project: 项目目录的 .claude/.mcp.json

Plugin 系统

插件可以提供:

Skills (技能)
Hooks (钩子)
MCP Servers

插件结构:

id:plugin-name@builtin
name:插件名称
version:1.0.0
features:
-skills
-hooks
-mcp

Skill 系统

技能是用户自定义的功能脚本:

---
name: skill-name
description: 技能描述
whenToUse: 使用场景
allowedTools: [Read, Write, Edit]
model: sonnet
hooks:
  preToolUse: echo "准备执行工具..."
---
技能的系统提示词内容...

技能来源:

内置技能 (src/skills/bundled/)
用户技能 (~/.claude/skills/*.md)
项目技能 (.claude/skills/*.md)
MCP 技能(通过 MCP 服务器提供)

配置和状态管理

配置管理确保用户可以定制化 AI 的行为。

配置层次

环境变量
    ↓
用户配置 (~/.claude/)
    ↓
项目配置 (.claude/)
    ↓
默认配置

特性开关

系统支持的功能开关:

开关	功能	默认值
VOICE_MODE	语音输入	禁用
COORDINATOR_MODE	协调器模式	禁用
FORK_SUBAGENT	Fork 子智能体	启用
VERIFICATION_AGENT	验证智能体	启用
CACHED_MICROCOMPACT	缓存微压缩	禁用
TOKEN_BUDGET	Token 预算控制	禁用

配置迁移

系统会自动迁移旧配置:

# 迁移流程示例
migrateLegacyOpusToCurrent()
migrateOpusToOpus1m()
migrateSonnet1mToSonnet45()
migrateSonnet45ToSonnet46()

这样用户升级后不需要手动迁移配置。

性能优化技巧

系统设计了多种优化策略。

缓存策略

# 使用 LRU 缓存加载快照
@lru_cache(maxsize=1)
defload_tool_snapshot() -> tuple[PortingModule, ...]:
# ...

缓存的好处:

避免重复加载 JSON
内存占用可控
访问速度快

并行工具调用

原版支持并行执行:

读取型工具 (Read, Grep, Glob) → 并行执行,最多 10 并发
写入型工具 (Edit, Write, Bash) → 串行执行,一次一个

并行执行的优势:

减少总等待时间
提高响应速度
更好地利用资源

Token 优化

Prompt 分为静态和动态两部分,静态部分可以缓存:

静态 Prompt
    ↓
缓存命中 → 直接使用
    ↓
缓存未命中 → 重新构建

Token 预算控制:

# 用户可以指定 token 目标
if projected_usage.input_tokens + projected_usage.output_tokens > self.config.max_budget_tokens:
    stop_reason = 'max_budget_reached'

错误处理和日志

完善的错误处理和日志记录是可靠性的基础。

错误分类

系统将错误分为多类:

错误类型	处理方式
工具执行错误	返回 tool_result,由 AI 重新决策
API 调用错误	重试机制 (指数退避)
权限错误	拦截并提示用户
网络/超时	自动重试,最大 5 次

日志记录

# history.py 中的历史记录
@dataclass
classHistoryLog:
    events: list[HistoryEvent] = field(default_factory=list)

defadd(self, category: str, message: str) -> None:
self.events.append(HistoryEvent(category=category, message=message))

日志记录的内容:

上下文构建 (文件数量、归档状态)
注册表信息 (命令数、工具数)
路由匹配 (匹配数量)
执行详情 (命令执行数、工具执行数)
Turn 结果 (停止原因)

架构彩蛋和隐藏特性

深入源码还能发现一些有趣的设计。

Buddy 伴侣精灵

原版有一个动画伴侣系统:

CompanionSprite.tsx (45KB): 动画精灵组件
companion.ts: 性格和行为配置
prompt.ts: 伴侣回复的系统提示词

这提供了情绪价值,让使用体验更人性化。

Vim 模式

完整的 Vim 状态机实现:

VimState = INSERT | NORMAL
CommandState = idle | count | operator | find | ...
Operators = d(delete), c(change), y(yank)
Motions = hjkl, wbWBE, 0^$

让习惯 Vim 的用户有熟悉的操作体验。

主题和样式

支持自定义输出样式:

---
name: 样式名
description: 描述
keep-coding-instructions: true/false
---
[自定义输出指令]

来源:

项目级: .claude/output-styles/*.md
用户级: ~/.claude/output-styles/*.md

定时任务

Cron 风格的定时任务系统:

One-Shot: "remind me at 9am" → 固定时间
Recurring: "every 5 min" → 持续执行

关键设计:

避免整点和半点(防止惊群效应)
支持持久化 (durable: true)
自动过期 (7 天)

实战案例

理论说完了,来看看实际应用。

案例 1:代码搜索和理解

任务:找到项目中所有处理用户认证的代码。

# 启动 CLI
./start-cli.sh

# 对话
> 帮我找用户认证相关的代码
[AI 分析] → 使用 Glob 找文件 → 使用 Grep 搜索关键词 → 返回结果

Files found:
- src/auth/login.tsx
- src/api/auth.ts
- src/middleware/auth.ts

Key functions:
- authenticateUser() in src/auth/login.tsx:45
- validateToken() in src/api/auth.ts:23
- checkAuthMiddleware() in src/middleware/auth.ts:12

案例 2:Bug 修复

任务:修复登录时的 token 验证错误。

> 用户反馈登录时 token 验证失败
[AI 阅读] src/api/auth.ts
[AI 分析] 发现代码逻辑问题
[AI 修改] Edit 工具修改代码
[AI 测试] 运行测试验证
[AI 报告] 修复完成,测试通过

修复说明:
在 src/api/auth.ts:23 添加了过期时间检查
现在 token 过期时会正确返回 401 错误

案例 3:功能开发

任务:添加用户头像上传功能。

> 帮我实现用户头像上传
[AI 规划] 使用 TaskCreate 创建任务列表
[AI 探索] Glob 查找类似功能代码
[AI 实现] 创建新组件和 API
[AI 测试] 运行集成测试
[AI 验证] 功能测试通过

任务清单:
✓ 探索现有代码
✓ 设计 API 接口
✓ 实现上传组件
✓ 添加路由处理
✓ 编写测试
✓ 验证功能