当前位置: 首页 > news >正文

我把Claude Code泄露的代码改造成python程序了,其中的大模型记忆模块与上下文工程分析

文章目录

    • 项目概述
    • 一、记忆模块
      • 1.1 架构设计
      • 1.2 记忆类型(4 种)
        • 1.2.1 用户记忆
        • 1.2.2 反馈记忆
        • 1.2.3 项目记忆
        • 1.2.4 引用记忆
      • 1.3 记忆保存方式
        • 方式 A:`<memory>` 标签(快速笔记)
        • 方式 B:结构化文件(持久化记忆)
      • 1.4 记忆不保存的内容
      • 1.5 自动梦境整合(Auto Dream)
      • 1.6 梦境整合流程
        • Phase 1 - Orient(定位)
        • Phase 2 - Gather(收集)
        • Phase 3 - Consolidate(整合)
        • Phase 4 - Prune and Index(修剪和索引)
      • 1.7 会话持久化
    • 二、上下文工程
      • 2.1 系统提示架构
      • 2.2 记忆系统注入
      • 2.3 上下文压缩
      • 2.4 压缩触发条件
      • 2.5 消息分割策略
      • 2.6 摘要提示词
      • 2.7 媒体剥离
    • 三、LLM 客户端抽象
      • 3.1 多提供商支持
      • 3.2 流式响应
      • 3.3 错误处理
    • 四、成本追踪
      • 4.1 Token 成本计算
      • 4.2 成本计算公式
      • 4.3 Advisor 成本
    • 五、关键设计模式
      • 5.1 追加式日志(Append-Only Logs)
      • 5.2 索引与内容分离
      • 5.3 锁文件模式
      • 5.4 节流机制
      • 5.5 角色交替保证
    • 六、总结
      • 记忆模块核心特性
      • 上下文工程核心特性
      • 设计亮点

项目概述

Claude Code开源代码是一个交互式 AI 助手系统,下面我来详细解析Claude Code的记忆模块与上下文工程分析,其中采用模块化架构设计,本文档详细分析其记忆模块和上下文工程的实现机制。
我把Claude Code泄露的代码改造成python程序了,并且接入Qwen大模型,来进行问答。下面我将对Claude Code大模型记忆模块与上下文工程进行详细分析。


Claude Code中还有很多隐藏的功能,比如电子宠物:


一、记忆模块

1.1 架构设计

记忆模块位于features/memory.py,采用基于文件的持久化系统,支持跨会话记忆存储和检索。

核心目录结构:

~/.config/cc-mini/ ├── memory/ # 记忆存储目录 │ ├── MEMORY.md # 记忆索引文件(<200 行) │ ├── logs/ # 按日追加的日志 │ │ └── YYYY/MM/YYYY-MM-DD.md │ └── [topic].md # 主题记忆文件 └── sessions/ # 会话记录目录 └── [session_id].jsonl

1.2 记忆类型(4 种)

系统采用与 Claude Code 相同的 4 类型记忆分类:

1.2.1 用户记忆
  • 用途:存储用户的角色、目标、职责和知识
  • 保存时机:了解到用户的角色、偏好、职责或知识时
  • 示例:用户是后端工程师,偏好使用 PostgreSQL
1.2.2 反馈记忆
  • 用途:用户给出的指导或纠正,确保跨会话一致性
  • 保存时机:用户纠正你的行为方式时
  • 结构
    - 规则描述 - **Why:** 原因说明 - **How to apply:** 应用方式
1.2.3 项目记忆
  • 用途:正在进行的工作、目标、bug 或事件
  • 保存时机:了解到谁在做什么、为什么做、何时完成
  • 要求:相对日期必须转换为绝对日期
1.2.4 引用记忆
  • 用途:指向外部系统中信息的位置
  • 示例:“bug 追踪在 Linear 项目 INGEST 中”

1.3 记忆保存方式

方式 A:<memory>标签(快速笔记)
defextract_memory_tags(text:str)->list[str]:"""从文本中提取所有 <memory>...</memory> 标签内容"""return[m.strip()forminre.findall(r"<memory>(.*?)</memory>",text,re.DOTALL)]

在对话中,模型可以输出:

<memory>用户偏好使用 PostgreSQL 而非 MySQL</memory>

系统会自动提取并追加到当日日志。

方式 B:结构化文件(持久化记忆)

写入.md文件到记忆目录,使用 frontmatter 格式:

--- name: 用户数据库偏好 description: 用户对数据库的技术偏好 type: user --- 用户偏好使用 PostgreSQL 而非 MySQL,因为需要复杂查询支持。

然后在MEMORY.md中添加索引指针:

- [用户数据库偏好](user-db-preference.md) — 技术偏好说明

1.4 记忆不保存的内容

系统明确规定了不应保存的内容:

  • 代码模式、架构、文件路径(可从代码库读取)
  • Git 历史、最近变更(git log/git blame是权威来源)
  • 调试解决方案(修复在代码中,上下文在 commit message)
  • CLAUDE.md 文件中已记录的内容
  • 临时任务详情或当前对话上下文

1.5 自动梦境整合(Auto Dream)

系统实现了自动记忆整合机制,定期将日志中的临时记忆整合为持久化主题文件。

触发条件:

defshould_auto_dream(memory_dir,min_hours,min_sessions,current_session_id):""" 检查是否触发自动整合: 1. 时间间隔 ≥ min_hours(默认由配置决定) 2. 新会话数 ≥ min_sessions 3. 包含 10 分钟扫描节流,避免频繁检查 """

锁机制:

  • 锁文件:.consolidate-lock
  • 锁持有者 PID 存储在文件内容中
  • 锁过期时间:1 小时(3600 秒)
  • 支持锁回收(持有者进程死亡后)

1.6 梦境整合流程

整合过程分为 4 个阶段:

Phase 1 - Orient(定位)
  • 使用 Glob 列出记忆目录中的所有文件
  • 读取MEMORY.md了解当前索引
  • 浏览现有主题文件,避免重复创建
Phase 2 - Gather(收集)

按优先级收集新信息:

  1. 日志文件:追加式日志流
  2. 漂移的记忆:与当前代码库矛盾的事实
  3. 转录搜索:使用 grep 搜索 JSONL 转录中的关键术语
Phase 3 - Consolidate(整合)
  • 将新信息合并到现有主题文件
  • 将相对日期转换为绝对日期
  • 删除矛盾的事实
Phase 4 - Prune and Index(修剪和索引)
  • 更新MEMORY.md,保持在 200 行以内
  • 每行 <150 字符:- [标题](文件.md) — 一句话钩子
  • 移除过时、错误或已被取代的记忆指针

1.7 会话持久化

会话记录采用JSONL 格式(每行一个 JSON 对象):

defsave_session(messages:list[dict],session_id:str)->None:"""序列化消息到 JSONL 并更新 last-session 符号链接"""SESSIONS_DIR.mkdir(parents=True,exist_ok=True)path=SESSIONS_DIR/f"{session_id}.jsonl"withpath.open("w",encoding="utf-8")asf:formsginmessages:f.write(json.dumps(serialize_message(msg),default=str)+"\n")

SessionStore 类管理单个会话:

  • append_message():追加消息到 JSONL
  • load_messages():从 JSONL 读取所有消息
  • list_sessions():列出可用会话(按更新时间排序)
  • 自动生成标题(从第一条用户消息提取)

二、上下文工程

2.1 系统提示架构

系统提示采用分块组装方式,位于core/context.py

静态部分(按顺序):

  1. Intro Section:角色介绍和安全准则
  2. System Section:系统行为规则
  3. Doing Tasks Section:任务执行指南
  4. Actions Section:操作风险评估
  5. Using Tools Section:工具使用偏好
  6. Tone and Style Section:沟通风格
  7. Output Efficiency Section:输出效率要求

动态部分:

  • Environment Section:工作目录、Git 状态、平台信息
  • Git Section:当前分支、状态、最近提交
  • CLAUDE.md Section:项目特定指令(如果存在)
  • Memory Section:记忆系统指令和索引
  • Companion Section:AI 伴侣介绍(如果启用)
  • Plan Mode Section:计划模式指令(如果激活)

2.2 记忆系统注入

记忆系统通过build_memory_system_section()函数注入到系统提示:

defbuild_memory_system_section(memory_dir:Path)->str:"""返回记忆指令 + MEMORY.md 内容"""index=load_memory_index(memory_dir)# 读取 MEMORY.md,截断到 10,000 字符section=""" # Auto Memory You have a persistent, file-based memory system at `{memory_dir}/`. ... [完整的记忆类型说明和保存指南] """ifindex:section+=f"\n## Current Memory Index (MEMORY.md)\n{index}\n"returnsection

2.3 上下文压缩

当上下文接近模型限制时,系统会自动压缩历史消息。

核心机制:

classCompactService:defcompact(self,messages,system_prompt,custom_instructions=""):"""压缩消息列表,返回 (new_messages, summary_text)"""history,recent=_split_recent(messages)# 分离历史和最近消息# 对历史部分调用 LLM 生成摘要summary=self._client.create_message(model=self._model,max_tokens=COMPACT_MAX_OUTPUT_TOKENS,# 4096system=COMPACT_SYSTEM,messages=cleaned_history,)# 构建新消息列表:[摘要用户消息] [确认助手消息] [最近消息]new_messages=[{"role":"user","content":summary},{"role":"assistant","content":"Understood..."},]new_messages.extend(recent)returnnew_messages,summary

2.4 压缩触发条件

自动压缩阈值计算:

def_auto_compact_threshold(model:str)->int:"""context_window - max_output_reserve - buffer"""cw=_context_window_for_model(model)# 模型上下文窗口max_out_reserve=min(20_000,cw//5)# 预留输出空间returncw-max_out_reserve-AUTOCOMPACT_BUFFER_TOKENS# buffer=13,000

模型上下文窗口:

模型系列上下文窗口
claude-opus-4-61,000,000
claude-sonnet-4-61,000,000
claude-opus-4200,000
claude-sonnet-4200,000
claude-3-5-sonnet200,000

触发判断:

defshould_compact(messages,model=None,last_input_tokens=None):iflast_input_tokensandmodel:# 使用 API 返回的实际 token 数returnlast_input_tokens>=_auto_compact_threshold(model)# 回退到基于字符的估算returnestimate_tokens(messages)>COMPACT_THRESHOLD_TOKENS# 100,000

2.5 消息分割策略

压缩时会保留最近的对话:

def_split_recent(messages):"""分离 (待摘要历史, 保留的最近消息)"""# 向后遍历,直到满足两个条件:# 1. 至少 MIN_RECENT_MESSAGES = 6 条消息# 2. 至少 MIN_RECENT_TOKENS = 10,000 个 token# 保证不分割 tool_use / tool_result 对ifkeep_start>0:msg=messages[keep_start]ifmsg 是纯 tool_results 的用户消息:keep_start-=1# 包含前一条助手消息

2.6 摘要提示词

压缩使用的提示词结构:

## Primary Request 用户的总体目标 ## Key Technical Concepts 建立的技术概念、模式、框架或约束 ## Files and Code 讨论或修改的关键文件及简要说明 ## Errors and Fixes 遇到的错误及解决方式 ## Current Work 最近工作内容及当前状态 ## Pending Tasks 未完成的任务或下一步

2.7 媒体剥离

压缩前会剥离历史消息中的图片/文档以节省 token:

def_strip_media(messages):"""将 images/documents 替换为文本标记"""forblockincontent:ifblock.type=="image":new_blocks.append({"type":"text","text":"[image]"})elifblock.type=="document":new_blocks.append({"type":"text","text":"[document]"})

三、LLM 客户端抽象

3.1 多提供商支持

core/llm.py实现了对 Anthropic 和 OpenAI 的抽象:

classLLMClient:def__init__(self,provider="anthropic",api_key=None,base_url=None):self.provider=validate_provider(provider)ifprovider=="openai":self._client=OpenAI(api_key,base_url)else:self._client=anthropic.Anthropic(api_key,base_url)

3.2 流式响应

支持流式输出和工具调用:

defstream_messages(self,model,max_tokens,messages,system,tools):ifprovider=="openai":return_OpenAIStream(...)return_AnthropicStream(...)

3.3 错误处理

重试机制:

_MAX_RETRIES=10_BASE_DELAY=0.5_MAX_DELAY=32.0def_compute_retry_delay(attempt,retry_after=None):ifretry_after:returnretry_after# 尊重 API 的 Retry-After 头# 指数退避 + 抖动delay=min(_BASE_DELAY*(2**attempt),_MAX_DELAY)jitter=delay*random.uniform(0,0.25)returndelay+jitter

错误类型分类:

  • 认证错误:立即返回,不重试
  • 上下文溢出:减少 max_tokens 后重试
  • 可重试错误(限流、连接、内部错误):指数退避重试
  • 其他 API 错误:返回错误信息

四、成本追踪

4.1 Token 成本计算

features/cost_tracker.py实现成本追踪:

@dataclass(frozen=True)class_PricingTier:input:float# $/MTokoutput:floatcache_write:floatcache_read:float

定价层级示例:

模型输入输出缓存写缓存读
claude-3-5-haiku$0.80$4.0$1.0$0.08
claude-sonnet$3.0$15.0$3.75$0.30
claude-opus-4-1$15.0$75.0$18.75$1.50

4.2 成本计算公式

defcalculate_cost(model,usage):tier=_tier_for_model(model)cost=(input_tokens*tier.input+output_tokens*tier.output+cache_read*tier.cache_read+cache_write*tier.cache_write)/1_000_000returncost

注意:Anthropic API 的input_tokens已排除缓存 token,缓存部分单独计费。

4.3 Advisor 成本

Advisor 功能(智能建议)的 token 单独计费:

ifadvisor_inoradvisor_out:advisor_tier=_tier_for_model(advisor_model)advisor_cost=(advisor_in*advisor_tier.input+advisor_out*advisor_tier.output)/1_000_000

五、关键设计模式

5.1 追加式日志(Append-Only Logs)

日志采用追加式写入,保证数据完整性:

defappend_to_daily_log(memory_dir,entry):path=daily_log_path(memory_dir)# YYYY/MM/YYYY-MM-DD.mdtimestamp=datetime.now().strftime("%H:%M")withpath.open("a",encoding="utf-8")asf:f.write(f"- [{timestamp}]{entry}\n")

5.2 索引与内容分离

MEMORY.md仅作为索引,不包含实际记忆内容:

  • 索引文件:<200 行,每行<150 字符
  • 主题文件:完整记忆内容
  • 优势:系统提示中只注入索引,节省 token

5.3 锁文件模式

使用文件 mtime 作为时间戳,文件内容为 PID:

deftry_acquire_lock(memory_dir):lp=_lock_path(memory_dir)# 检查锁是否过期(>1 小时)# 检查持有者进程是否存活# 写入自己的 PIDlp.write_text(str(os.getpid()))

5.4 节流机制

防止频繁扫描会话文件:

_SESSION_SCAN_INTERVAL_S=600# 10 分钟ifnow-_last_session_scan_at<SESSION_SCAN_INTERVAL_S:returnFalse# 跳过本次检查_last_session_scan_at=now

5.5 角色交替保证

API 要求消息严格的用户/助手交替:

def_fix_alternation(messages):fixed=[messages[0]]formsginmessages[1:]:ifmsg["role"]==fixed[-1]["role"]:# 合并到前一条消息fixed[-1]["content"]=prev+"\n"+curelse:fixed.append(msg)returnfixed


六、总结

记忆模块核心特性

  1. 4 类型分类法:user/feedback/project/reference
  2. 双模式保存<memory>标签(临时)和结构化文件(持久)
  3. 自动整合:基于时间和会话数的触发条件
  4. 锁机制:防止并发整合冲突
  5. 索引分离:MEMORY.md 仅作索引,节省 token

上下文工程核心特性

  1. 分块组装:静态 + 动态部分模块化组合
  2. 自动压缩:基于模型上下文窗口的智能阈值
  3. 消息分割:保留最近 6 条消息或 10k token
  4. 媒体剥离:压缩前移除图片/文档
  5. 结构化摘要:7 部分摘要模板

设计亮点

  • 追加式日志:简单可靠,易于回溯
  • 节流机制:避免不必要的频繁操作
  • 错误恢复:多层次重试和回退策略
  • 成本感知:实时追踪 token 使用和成本
  • 多提供商:统一的 Anthropic/OpenAI 抽象

这套系统实现了持久化记忆智能上下文管理成本优化的平衡,为长周期 AI 助手交互提供了坚实基础。

http://www.jsqmd.com/news/647086/

相关文章:

  • [特殊字符]Openclaw 梦境(Dream)系统详细研究
  • Adobe-GenP通用补丁:如何安全高效地解锁Adobe全家桶功能
  • opencode 配置本地ollama模型编程
  • 从零到一:基于STM32的L298N电机驱动与PWM调速实战
  • 2026深度分析罗兰艺境市场研究专业服务GEO技术案例,测评北京市场调研公司优化过程与效果验证 - 罗兰艺境GEO
  • 互补PWM死区时间如何根据MOSFET开关参数精确计算?
  • 职场里,越亲近越好?怎样的边界感,才是舒服关系?
  • mysql大表数据清理的利器_使用表分区按天删除数据
  • HTML5 Input 类型详解
  • 新都区急着入住怎么快又好?2026高效靠谱、工期准时的装修公司终极推荐! - 推荐官
  • 【MATLAB实战】手把手教你设计超前校正:从原理到代码实现
  • 渗透测试不够用?红蓝对抗如何精准击穿企业安全体系的深层弱点
  • 大麦抢票脚本终极教程:5分钟学会自动化抢票技巧
  • package.json resolutions:从依赖冲突到版本锁定的实战指南
  • 春茶季,教你一眼认出茶山上的“紫芽”
  • 从AlphaGo到ChatGPT:聊聊强化学习(RL)是如何成为AI进化‘隐藏引擎’的
  • 5分钟搞定openEuler防火墙放行vsftp:主动/被动模式全解析
  • ribbon--重点笔记
  • 盐城哪家好吃
  • 提升你的编码效率,Claude-Mem 插件带来无缝记忆体验!
  • RS485通信故障排查与优化实践指南
  • 【太奶学IT】【超好理解】神经网络是个啥?我这老太婆给你唠明白
  • Python 并发编程:asyncio vs threading vs multiprocessing
  • MATLAB柱状图进阶:5分钟搞定分组数据+数值标注(附完整代码)
  • 安装阿帕奇maven的相关配置
  • 生成式AI应用用户流失率飙升的真正原因:不是模型不准,而是这6个隐性体验缺口未被填补
  • 即插即用系列 | CVPR 2024 FADC:频域自适应采样,从根源消除分割“棋盘格”
  • LeRobot实战指南:3步构建你的机器人学习工作流
  • 人大金仓 KingbaseES V8 数据库 Docker 部署指南
  • 从零到一的无人机DIY手记(一):配件采购与兼容性排雷