学会这套架构，你也能写Claude Code！

Claude Code底层架构完全拆解：手撸一套可运行的Agent核心（精简版）

从源码读懂2026年最硬核的AI Agent设计

引言

2026年，AI已经从“聊天玩具”进化为真正的“生产力工具”。在这场转型中，Claude Code无疑是佼佼者——它不仅仅是一个大模型，更是一套完整的Agent系统。

很多开发者看源码时，往往只看到了TypeScript语法，却忽略了背后的工程化思想。今天，我将结合官方源码架构文档，从Harness与Framework的哲学定义出发，剥开它的内核，并用精简的Python伪代码复刻这套工业级架构。

本文核心亮点：

• 🧠 架构本质：Harness与Framework的分离哲学
• 🔄 核心循环：agent_loop如何实现思考→行动的闭环
• 🛠️ 工具调度：并发读与串行写的精妙平衡
• 🛡️ 安全防御：Bash命令的纵深防御体系
• 💾 状态管理：QueryEngine如何解决上下文爆炸

---

一、核心思想：Harness vs. Framework

在开始写代码之前，必须先搞懂两个概念：

概念	定义	特点	示例
Framework（框架）	系统的基础设施	相对稳定，长期存在	状态管理、工具注册表
Harness（外壳）	套住模型运行的逻辑	随模型能力进化，易被替换	测试逻辑、调用适配层

Claude Code的本质公式：

Claude Code = 大模型（Brain）+ 工具系统（Hands）+ 状态管理（Memory）+ 安全压缩（Guardrail）

工作流：
用户输入 → 组装上下文 → 调用模型 → 决策 → 执行工具 → 结果回写 → 循环

---

二、主循环与状态管理：系统的“中控台”

QueryEngine负责准备资料、清理垃圾、维持记忆。核心任务是：把过去发生了什么，组织成模型能理解的当前输入。

Python实战代码：QueryEngine（精简版）

class QueryEngine:
    def __init__(self, token_limit: int = 200000):
        self.history = []      # 会话历史
        self.memory = []       # 长期记忆
        self.system_prompt = ""
        self.token_limit = token_limit

    def load_memory(self, memory_files: list[str]):
        for file_name in memory_files:
            with open(file_name, "r") as f:
                self.memory.append({"role": "user", "content": f"【记忆】\n{f.read()}"})

    def build_messages(self) -> list[dict]:
        messages = []
        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})
        messages.extend(self.memory)
        messages.extend(self.history)
        return messages

    def add_user_message(self, text: str):
        self.history.append({"role": "user", "content": text})

    def add_tool_results(self, tool_name: str, content: str):
        self.history.append({"role": "tool", "content": f"{tool_name}结果:\n{content}"})

    def maybe_compact(self):
        """上下文压缩：Token超限时总结历史"""
        if len(self.history) > 5:
            summary = self._call_llm_for_summary(self.history[:-5])
            self.history = [{"role": "system", "content": f"【摘要】{summary}"}] + self.history[-5:]

    def _call_llm_for_summary(self, history_part) -> str:
        # 可接入KoalaAPI等统一接口生成摘要
        return "用户与助手讨论了项目需求，执行了若干工具操作。"

💡 设计亮点：maybe_compact用“压缩摘要”替代“暴力截断”，让Agent在长对话中保持完整“记忆”。实际接入大模型时，可借助koalaapi等统一接口降低开发成本。

---

三、工具系统与调度：Agent的“手”

模型只负责“决策”，工具负责“执行”。

1. 工具基类（精简版）

import abc, subprocess, re
from concurrent.futures import ThreadPoolExecutor

class Tool(abc.ABC):
    name: str = ""
    description: str = ""  # 会作为Prompt告诉模型怎么用

    @abc.abstractmethod
    def validate(self, params) -> tuple[bool, str]:
        pass

    @abc.abstractmethod
    def run(self, params, context) -> str:
        pass

class ReadFileTool(Tool):
    name = "read_file"
    description = "读取文件内容。参数: {\"path\": \"文件路径\"}"

    def validate(self, params):
        return (True, "") if params.get("path") else (False, "缺少path")

    def run(self, params, context):
        with open(params["path"], "r") as f:
            return f"文件内容:\n{f.read()}"

2. 工具调度器：并发读 + 串行写

class ToolOrchestration:
    def __init__(self):
        self.tools = {"read_file": ReadFileTool()}

    def run_tools(self, tool_calls: list[dict]):
        readonly, write = [], []
        for call in tool_calls:
            (readonly if call["name"] == "read_file" else write).append(call)
        
        # 并发执行只读任务
        with ThreadPoolExecutor(max_workers=5) as executor:
            r_results = list(executor.map(self._run_one, readonly))
        
        # 串行执行写任务
        w_results = [self._run_one(call) for call in write]
        return r_results + w_results

    def _run_one(self, call):
        tool = self.tools.get(call["name"])
        if tool and tool.validate(call["params"])[0]:
            return tool.run(call["params"], None)
        return f"工具{call['name']}执行失败"

---

四、安全与防御：Bash的“紧箍咒”

纵深防御：正则拦截危险命令。

class BashTool(Tool):
    name = "bash"
    description = "执行Shell命令"
    
    DANGEROUS = [(r"\$\(.+\)", "命令替换禁止"), (r"rm\s+-rf", "强制删除禁止")]

    def validate(self, params):
        cmd = params.get("command", "")
        for pattern, msg in self.DANGEROUS:
            if re.search(pattern, cmd):
                return False, msg
        return True, ""

    def run(self, params, context):
        result = subprocess.run(params["command"], shell=True, capture_output=True, timeout=30)
        return result.stdout.decode()[:2000]

---

五、主循环：让一切转起来

def agent_loop(user_input: str, engine: QueryEngine, orchestrator: ToolOrchestration):
    engine.add_user_message(user_input)
    
    for _ in range(10):  # 最多10轮
        response = mock_model_call(engine.build_messages())  # 调用模型API
        
        if response.get("tool_calls"):
            results = orchestrator.run_tools(response["tool_calls"])
            for r in results:
                engine.add_tool_results("system", r)
            engine.maybe_compact()  # 自动压缩上下文
        else:
            return response.get("content", "完成")
    
    return "达到轮数限制"

def mock_model_call(messages):
    """模拟模型返回（实际应调用Claude/GPT）"""
    return {"tool_calls": [{"name": "read_file", "params": {"path": "test.py"}}]}

---

六、总结与工程启示

核心设计模式回顾：

组件	职责	关键技巧	代码行数
QueryEngine	状态管理+上下文压缩	摘要压缩替代截断	~25行
ToolOrchestration	工具调度	并发读+串行写	~15行
BashTool	安全防御	正则拦截危险命令	~12行
Agent Loop	主循环	决策-执行-回写闭环	~10行

总计约60行核心代码，完整呈现Agent三大机制。

这套架构适用于代码生成、办公自动化、运维排障等任何需要“逻辑+执行”的场景。理解它，你就掌握了2026年Agent系统的核心设计范式