大家好，我是你们的技术伙伴。👋
在2026年的今天，我们见证了AI从“聊天玩具”向“生产力工具”的剧烈转型。而Claude Code（或称Claude Dev）无疑是这场转型中的佼佼者。它不仅仅是一个大模型，更是一个完整的Agent系统。
很多开发者看源码时，往往只看到了TypeScript的语法，却忽略了其背后的工程化思想。今天，我将结合源码架构文档，带大家从Harness（外壳）与Framework（框架）的哲学定义出发，一层层剥开它的内核，并用Python和TS双语，手把手教你复刻这套工业级架构。
本文核心硬核点：
🧠 第一部分：核心思想——Harness vs. Framework
在开始写代码之前，我们必须先搞懂两个概念：Harness 和 Framework。
- Framework (框架)：这是系统的基础设施，比如状态管理、工具注册表。它相对稳定，长期存在。
- Harness (外壳)：这是把模型“套”起来运行/测试的逻辑。随着模型能力的提升（如上下文窗口变大），Harness很容易被替换或弱化。
Claude Code的本质公式：
Claude Code = 大模型 (Brain) + 工具系统 (Hands) + 状态管理 (Memory) + 安全压缩 (Guardrail)
它的工作流不是简单的“一问一答”，而是一个永不停歇的循环：
用户输入 -> 组装上下文 -> 调用模型 -> 模型决策(回答/调工具) -> 执行工具 -> 结果回写 -> 循环
⚙️ 第二部分：主循环与状态管理——系统的“中控台”
如果说模型是大脑，那么QueryEngine就是大脑的“工作台”，负责准备资料、清理垃圾。这是整个Agent的中控台。
1. QueryEngine：会话状态管理器
在TypeScript源码中，QueryEngine负责维护history（历史消息）、memory（长期记忆）和system_prompt（系统指令）。它的核心任务是：把过去发生了什么，组织成模型能理解的当前输入。
Python实战代码：重构QueryEngine
from typing import List, Dict, Any
import tiktoken # 用于Token计算
class QueryEngine:
def __init__(self, token_limit: int = 200000):
self.history: List[Dict[str, Any]] = []
self.memory: List[Dict[str, Any]] = []
self.system_prompt: str = ""
self.token_limit = token_limit
self.encoding = tiktoken.get_encoding("cl100k_base")
def load_memory(self, memory_files: List[str]):
"""加载长期记忆文件 (模拟源码中的MEMORY.md)"""
for file_name in memory_files:
try:
with open(file_name, "r", encoding="utf-8") as f:
content = f.read()
# 角色设为user，让模型认为这是用户设定的规则或记忆
self.memory.append({"role": "user", "content": f"【记忆文件：{file_name}】\\n{content}"})
except Exception as e:
print(f"加载记忆文件失败 {file_name}: {e}")
def build_messages(self) -> List[Dict[str, Any]]:
"""组装最终发送给模型的消息列表"""
messages = []
# 1. 系统提示词（放在最前面，权重最高）
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
# 2. 长期记忆
messages.extend(self.memory)
# 3. 会话历史
messages.extend(self.history)
return messages
def add_user_message(self, text: str):
self.history.append({"role": "user", "content": text})
def add_assistant_message(self, text: str):
self.history.append({"role": "assistant", "content": text})
def add_tool_results(self, tool_name: str, content: str, status: str = "success"):
"""添加工具执行结果"""
self.history.append({
"role": "tool",
"content": f"工具名称: {tool_name}\\n状态: {status}\\n返回结果:\\n{content}"
})
def count_tokens(self, messages) -> int:
"""粗略计算Token数量"""
return len(self.encoding.encode(str(messages)))
def maybe_compact(self, reserve: int = 30000):
"""
上下文压缩逻辑（核心！）
当Token接近上限时，将历史对话总结为摘要，防止上下文爆炸
"""
current_tokens = self.count_tokens(self.history)
if current_tokens < self.token_limit – reserve:
return
print(f"上下文过长 ({current_tokens} tokens)，正在执行压缩摘要…")
# 这里可以调用LLM API生成摘要，或者简单截断
# 简单示例：保留最近5条，其余总结
if len(self.history) > 5:
to_summarize = self.history[:-5]
summary_text = self._call_llm_for_summary(to_summarize) # 假设的摘要方法
self.history = [ {
"role": "system",
"content": f"【对话摘要】为了节省上下文，之前的对话已被总结：\\n{summary_text}"
} ] + self.history[-5:]
def _call_llm_for_summary(self, history_part) -> str:
# 模拟调用模型进行摘要
return "用户与助手讨论了项目需求，助手执行了若干工具操作。"
🛠️ 第三部分：工具系统与调度——Agent的“手”
这是Claude Code最性感的部分。模型只负责“决策”（我要读文件），工具负责“执行”（把文件内容读出来）。
1. 工具基类与读文件工具 (TS -> Python)
文档中提到，工具不仅仅是函数，它有描述（Description）、参数校验和执行逻辑。描述词（Description）直接决定了模型会不会用它，它是Prompt的一部分。
TypeScript 源码思想 (供参考):
// 这是TS源码中的思想：工具描述即契约
interface Tool {
name: string;
description: string; // 这部分会被塞进Prompt，告诉模型怎么用
parameters: JSONObject;
execute: (params: any) => Promise<any>;
}
Python实战代码：定义Tool基类与ReadFile
import abc
import json
import subprocess
import re
class Tool(abc.ABC):
name: str = "base_tool"
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):
if "path" not in params:
return False, "缺少必填参数 'path'"
return True, ""
def run(self, params, context):
path = params["path"]
try:
with open(path, "r", encoding="utf-8") as f:
content = f.read()
return f"文件内容如下：\\n{content}"
except Exception as e:
return f"读取文件失败：{str(e)}"
class SearchFilesTool(Tool):
name = "search_files"
description = """在指定目录下搜索包含关键词的文件。
参数: { "directory": "目录", "query": "关键词" }
适用场景：当用户需要查找某个功能代码或特定配置在哪里时调用。"""
def validate(self, params):
if not params.get("directory") or not params.get("query"):
return False, "缺少必要参数"
return True, ""
def run(self, params, context):
directory = params["directory"]
query = params["query"]
# 这里简化为打印指令，实际应调用grep或系统搜索
return f"在目录 {directory} 中搜索关键词 '{query}' 的结果：\\n模拟结果：found in file_x.py, line 10."
2. 工具调度器：并发与串行的艺术
文档中提到一个非常高级的工程技巧：只读工具并发，写操作工具串行。
- 为什么？ 读文件可以同时开10个线程，速度飞快；但写文件如果并发，A改了一行，B又覆盖回去，就会导致代码错乱。
Python实战代码：智能工具调度器
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
# 全局锁，用于保护写操作（简化演示）
_write_lock = threading.Lock()
class ToolOrchestration:
def __init__(self):
self.tools_map = {}
# 注册工具
self.register_tool(ReadFileUpTool())
self.register_tool(SearchFilesTool())
# … 其他工具
def register_tool(self, tool: Tool):
self.tools_map[tool.name] = tool
def run_tools(self, tool_calls: List[Dict]):
"""
智能调度：
1. 只读工具 -> 并发执行 (提高速度)
2. 写操作工具 -> 串行执行 (保证安全)
"""
readonly_results = []
write_results = []
readonly_calls = []
write_calls = []
# 分类
for call in tool_calls:
tool_name = call["name"]
tool_params = call["params"]
tool_obj = self.tools_map.get(tool_name)
if not tool_obj:
write_results.append(f"错误：未找到工具 {tool_name}")
continue
# 这里简单通过名称判断读写，实际应由Tool类属性决定
if tool_name in ["read_file", "search_files"]:
readonly_calls.append((tool_obj, tool_params))
else:
write_calls.append((tool_obj, tool_params))
# — 并发执行只读任务 —
print(f"并发执行 {len(readonly_calls)} 个只读任务…")
with ThreadPoolExecutor(max_workers=5) as executor:
futures = [executor.submit(self._safe_run, tool, params) for tool, params in readonly_calls]
for future in as_completed(futures):
readonly_results.append(future.result())
# — 串行执行写任务 —
print(f"串行执行 {len(write_calls)} 个写任务…")
for tool_obj, tool_params in write_calls:
# 模拟加锁，防止并发修改冲突
with _write_lock:
result = self._safe_run(tool_obj, tool_params)
write_results.append(result)
return {
"readonly": readonly_results,
"write": write_results
}
def _safe_run(self, tool: Tool, params):
"""安全执行单个工具"""
is_valid, msg = tool.validate(params)
if not is_valid:
return f"参数校验失败: {msg}"
try:
return tool.run(params, None)
except Exception as e:
return f"工具执行异常: {str(e)}"
🛡️ 第四部分：安全与防御——Bash的“紧箍咒”
文档中特别强调了Bash工具的危险性。在2026年，“模型幻觉+代码执行”是致命的。Claude Code采用了纵深防御（Defense in Depth）。
核心逻辑：
Python实战代码：Bash安全检查器
class BashTool(Tool):
name = "bash"
description = """执行Shell命令。
参数: { "command": "命令字符串" }
警告：这是高危操作，必须经过严格检查。"""
# 危险模式匹配：命令替换、删除、重定向等
DANGEROUS_PATTERNS = [
(r"\\$\\(", "检测到命令替换 $(…)，禁止执行"),
(r"\\|.*\\s(rm|dd|:>\\s)", "检测到管道或危险命令 (rm, dd, 清空文件)"),
(r"rm\\s+-rf", "检测到强制删除命令 rm -rf，绝对禁止"),
(r"chmod\\s+777", "检测到过度开放权限 chmod 777"),
]
def validate(self, params):
command = params.get("command", "")
if not command:
return False, "必须提供 command 参数"
# 检查危险模式
for pattern, message in self.DANGEROUS_PATTERNS:
if re.search(pattern, command, re.IGNORECASE):
return False, message
return True, ""
def run(self, params, context):
command = params["command"]
is_valid, msg = self.validate(params)
if not is_valid:
return f"安全检查未通过：{msg}"
# 这里应该调用Hook或者用户确认
# 模拟确认通过
try:
# shell=True存在风险，实际生产环境应严格限制
result = subprocess.run(command, shell=True, capture_output=True, timeout=30)
output = result.stdout.decode()[:2000] # 限制输出长度
error = result.stderr.decode()[:2000]
return f"执行成功。输出:\\n{output}\\n错误:\\n{error}"
except Exception as e:
return f"执行失败: {str(e)}"
🔄 第五部分：主循环——让一切转起来
最后，我们将上述所有组件拼装起来，形成Agent的心脏。
Python实战代码：Agent主循环
def agent_loop(user_input: str, engine: QueryManag, tool_orchestrator: ToolOrchestration, max_turns: int = 10):
"""
Agent 主循环 (The Engine)
"""
engine.add_user_message(user_input)
for turn in range(max_turns):
print(f"\\n— 第 {turn+1} 轮循环 —")
# 1. 准备上下文
messages = engine.build_messages()
# 2. 调用模型 (这里模拟)
# 实际这里应该是调用Claude或GPT的API
response = mock_model_call(messages) # 你的API调用逻辑
# 3. 决策：是回答还是调工具？
if response.get("tool_calls"):
# 执行工具
tool_results = tool_orchestrator.run_tools(response["tool_calls"])
# 结果回写
for result in tool_results["readonly"] + tool_results["write"]:
engine.add_tool_results("system", result)
# 自动压缩上下文
engine.maybe_compact()
else:
# 直接回答
final_answer = response.get("content", "好的。")
engine.add_assistant_message(final_answer)
print(f"Assistant: {final_answer}")
return final_answer
return "达到最大轮数限制，任务可能未完成。"
# — 启动 —
if __name__ == "__main__":
engine = QueryEngine()
engine.load_memory(["MEMORY.md"]) # 加载记忆
# 初始化工具箱
tools = ToolOrchestration()
# 运行
agent_loop("请帮我分析一下项目中的 requirements.txt 有哪些依赖，并在终端中执行 pip install -r requirements.txt", engine, tools)
🏁 结语
通过上述五个部分的拆解，我们不仅看到了Claude Code的TypeScript源码架构，更亲手用Python复刻了一套可运行的Agent核心框架。
回顾核心亮点：
这套架构不仅仅适用于代码生成，它适用于办公自动化、运维排障、数据分析等任何需要“逻辑+执行”的场景。
如果你觉得这篇博客帮你打通了任督二脉，希望点赞、收藏、关注！你的支持是我持续输出硬核内容的最大动力！
