记忆系统指南

为 Claude 构建持久化记忆系统

🌐 查看英文原文 | 源码 Notebook

长时间运行 Agent 的上下文编辑与 Memory

AI agents that run across multiple sessions or handle long-running tasks face two key challenges: they lose learned patterns between conversations, and context windows fill up during extended interactions.

This cookbook demonstrates how to address these challenges using Claude’s memory tool and context editing capabilities.

前置条件

所需知识： - Python 基础（函数、类、async/await 基础） - 对 REST API 和 JSON 的基本理解

所需工具： - Python 3.10 或更高版本 - Anthropic API 密钥 (在此获取)

推荐： - 熟悉并发编程概念（线程、async） - 对 LLM 中的上下文窗口有基本理解

环境配置

VSCode 用户

注意：以下内容仅作为示例，具体配置可能因个人环境而异。

# 示例配置代码

注意： 以上代码仅供参考，实际使用时请根据个人需求进行调整。```bash # 1. Create virtual environment python -m venv .venv

2. Activate it

source .venv/bin/activate # macOS/Linux # or: .venv# Windows

3. Install dependencies

pip install -r requirements.txt

4. In VSCode: Select .venv as kernel (top right)


### API 密钥

```bash
cp .env.example .env
# Edit .env and add your ANTHROPIC_API_KEY

从以下链接获取您的API密钥：https://console.anthropic.com/

1. 简介：为什么记忆很重要

本食谱展示了在有效上下文工程：AI代理中描述的上下文工程模式的实际应用。那篇文章涵盖了上下文为何是有限资源，注意力预算如何工作，以及构建有效代理的策略——您在这里将看到这些技术的实际应用。

问题

大型语言模型具有有限的上下文窗口（Claude 4系列模型的200k个token）。虽然这看起来很大，但会带来几个挑战：

上下文限制：长时间对话或复杂任务可能会超出可用上下文
计算成本：处理大量上下文代价高昂——注意力机制呈二次方增长
重复模式：跨对话中的相似任务需要每次都重新解释上下文
信息丢失：当上下文填满时，早期的重要信息会丢失

解决方案

Claude 4模型引入了强大的上下文管理功能：

记忆工具 (memory_20250818): 实现跨对话学习
- Claude可以记录下它学到的内容以供将来参考
- 基于文件的系统位于/memories目录下
- 客户端实现让您拥有完全控制权
上下文编辑：自动使用两种策略管理上下文：
- 工具使用清除 (clear_tool_uses_20250919): 当上下文变大时清除旧的工具结果
- 思维管理 (clear_thinking_20251015): 管理扩展的思维块（需要启用思维功能）
- 可配置的触发器和保留策略

优势

构建AI代理，使其随着时间的推移在特定任务上变得更好：

第1次会话：Claude解决问题，记录下模式
第2次会话：Claude立即应用学到的模式（更快！）
长会话：上下文编辑使对话可管理

想象一下给Claude一本笔记本来记笔记和查阅——就像人类一样。

你将学到什么

在阅读完这本食谱后，您将能够： - 实现跨对话学习的记忆工具 - 配置上下文编辑以管理长时间会话 - 应用记忆安全和组织方面的最佳实践

1. Introduction: Why Memory Matters

This cookbook demonstrates practical implementations of the context engineering patterns described in Effective context engineering for AI agents. That post covers why context is a finite resource, how attention budgets work, and strategies for building effective agents—the techniques you’ll see in action here.

The Problem

大型语言模型具有有限的上下文窗口（Claude 4为200k个token）。虽然这看起来很大，但会带来几个挑战：

Context limits: Long conversations or complex tasks can exceed available context
Computational cost: Processing large contexts is expensive - attention mechanisms scale quadratically
Repeated patterns: Similar tasks across conversations require re-explaining context every time
Information loss: When context fills up, earlier important information gets lost

The Solution

Claude Sonnet 4.6引入了两个强大的功能：

Memory Tool (memory_20250818): Enables cross-conversation learning
- Claude can write down what it learns for future reference
- File-based system under /memories directory
- Client-side implementation gives you full control

支持模型：Claude Opus 4.1 (claude-opus-4-1), Claude Opus 4 (claude-opus-4), Claude Sonnet 4.6 (claude-sonnet-4-6), Claude Sonnet 4 (claude-sonnet-4), 和 Claude Haiku 4.5 (claude-haiku-4-5)

The Benefit

Build AI agents that get better at your specific tasks over time:

Session 1: Claude solves a problem, writes down the pattern
Session 2: Claude applies the learned pattern immediately (faster!)
Long sessions: Context editing keeps conversations manageable

Think of it as giving Claude a notebook to take notes and refer back to - just like humans do.

2. 应用场景

内存和上下文管理使强大的新工作流程成为可能：

🔍 Code Review Assistant

从过去的审查中学习调试模式
在未来的会话中立即识别相似的错误
构建团队特定的代码质量知识
生产就绪：通过claude-code-action与GitHub PR审查集成

📚 Research Assistant

在多个会话中积累关于主题的知识
连接不同研究线索的见解
维护文献和来源跟踪

💬 Customer Support Bot

学习用户偏好和沟通风格
记住常见问题和解决方案
从交互中构建产品知识库

📊 Data Analysis Helper

记住数据集模式和异常
存储有效的分析技术
随时间构建特定领域的见解

支持模型：Claude Opus 4.1 (claude-opus-4-1) 和 Claude Sonnet 4.6 (claude-sonnet-4-6)

本食谱专注于代码审查助手，因为它清楚地展示了内存（学习模式）和上下文编辑（处理长审查）。

3. 快速入门示例

让我们通过简单的示例来观察内存和上下文管理在实际中的应用。

环境配置

首先，安装依赖项并配置您的环境：

%%capture
# Install required packages
# Option 1: From requirements.txt
# %pip install -q -r requirements.txt

# Option 2: Direct install
%pip install -q anthropic python-dotenv ipykernel

⚠️ 重要：在此目录下创建一个 .env 文件：

# Copy .env.example to .env and add your API key
cp .env.example .env

然后编辑 .env 以添加您的 Anthropic API 密钥（从 https://console.anthropic.com/ 获取）

import os
from typing import cast

from anthropic import Anthropic
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Model configuration - use alias for automatic updates
MODEL = "claude-sonnet-4-6"  # Can override via ANTHROPIC_MODEL env var
if os.getenv("ANTHROPIC_MODEL"):
    MODEL = os.getenv("ANTHROPIC_MODEL")

if not API_KEY:
    raise ValueError("ANTHROPIC_API_KEY not found. Copy .env.example to .env and add your API key.")

if not MODEL:
    raise ValueError("ANTHROPIC_MODEL not found. Copy .env.example to .env and set the model.")

MODEL = cast(str, MODEL)

client = Anthropic(api_key=API_KEY)

print("✓ API key loaded")
print(f"✓ Using model: {MODEL}")

✓ API key loaded
✓ Using model: claude-sonnet-4-6

示例 1：基本 Memory 使用

让我们看看 Claude 如何使用内存来存储信息以供将来参考。

Helper Functions

这些示例使用了 demo_helpers.py 中的辅助函数：

run_conversation_loop()：处理 API 对话循环
- 启用内存工具后调用 Claude 的 API
- 执行工具使用（内存操作）
- 直到 Claude 停止使用工具为止
- 返回最终响应
run_conversation_turn()：单轮对话（用于示例 3）
- 与上面相同，但仅在完成一次 API 调用后返回
- 当您需要精细控制时很有用
print_context_management_info()：显示上下文清除统计信息
- 显示保存的标记，清除的工具使用
- 帮助可视化上下文编辑何时触发

⚠️ Note on Memory Clearing

下面的单元格清除所有内存文件，为本次演示提供一个干净的起点。这对于多次运行笔记本以查看一致的结果很有用。

在生产应用中，您应仔细考虑是否清除所有内存，因为这会永久删除学习到的模式。考虑使用选择性删除或将内存组织成项目特定的目录。

# Import helper functions
from memory_demo.demo_helpers import (
    run_conversation_loop,
    run_conversation_turn,
    print_context_management_info,
)
from memory_tool import MemoryToolHandler

# Initialize
client = Anthropic()
memory = MemoryToolHandler(base_path="./demo_memory")

# Clear any existing memories to start fresh
print("🧹 Clearing previous memories...")
memory.clear_all_memory()
print("✓ Memory cleared\n")

# Load example code with a race condition bug
with open("memory_demo/sample_code/web_scraper_v1.py", "r") as f:
    code_to_review = f.read()

messages = [
    {
        "role": "user",
        "content": f"I'm reviewing a multi-threaded web scraper that sometimes returns fewer results than expected. The count is inconsistent across runs. Can you find the issue?\n\n```python\n{code_to_review}\n```",
    }
]

print("=" * 60)
print("📝 SESSION 1: Learning from a bug")
print("=" * 60)

# Run conversation loop
response = run_conversation_loop(
    client=client,
    model=MODEL,
    messages=messages,
    memory_handler=memory,
    系统参数="您是一位代码审查员。",
    max_tokens=2048,
    max_turns=5,
    verbose=True,
)

print("\n" + "=" * 60)
print("✅ Session 1 complete!")
print("=" * 60)

🧹 Clearing previous memories...
✓ Memory cleared

============================================================
📝 SESSION 1: Learning from a bug
============================================================

🔄 Turn 1:
💬 Claude: I'll review this code for the race condition issue. Let me first check my memory, then analyze the problem.

  🔧 内存工具：查看 /memories
  ✓ Result: Directory: /memories
(empty)

🔄 Turn 2:
  🔧 Memory tool: create /memories/review.md
  ✓ Result: File created successfully at /memories/review.md

🔄 Turn 3:
💬 Claude: ## Code Review: Multi-threaded Web Scraper Race Condition

### 🔴 **Critical Issue Found: Race Condition in Shared State**

You've correctly identified the problem! The code has **race conditions** caused by multiple threads modifying shared lists without synchronization.

---

### **The Problem**

**Lines with race conditions:**
```python
self.results.append(result)  # 竞态条件

Problem: - Python’s list.append() is NOT thread-safe for concurrent modifications - Multiple threads simultaneously appending can cause: - Lost updates: One thread’s append overwrites another’s - Inconsistent state: Internal list structure can be corrupted - Unpredictable counts: Results randomly dropped

Why it happens: list.append() involves multiple operations: 1. Read current list size 2. Allocate space 3. Insert item 4. Update size

In scrape_urls():

if “error” in result: self.failed_urls.append(result[“url”]) # ⚠️ NOT THREAD-SAFE else: self.results.append(result) # ⚠️ NOT THREAD-SAFE


**为什么这会导致结果丢失：**
- Python 的 `list.append()` 在所有场景下 **不是原子的**
- 当多个线程同时调用 `append()` 时，内部列表操作可能会交错
- 这可能导致丢失更新，其中一个线程的追加会覆盖另一个线程的

---

### **Solutions**

#### **选项 1：使用线程锁（传统方法）**
```python
import threading

class WebScraper:
    """一个可以同时抓取多个URL的网页抓取器。”

    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.results = []
        self.failed_urls = []
        self.lock = threading.Lock()  # ✅ Add lock

    def scrape_urls(self, urls: List[str]) -> List[Dict[str, Any]]:
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(self.fetch_url, url) for url in urls]

            for future in as_completed(futures):
                result = future.result()

                # ✅ Protect shared state with lock
                with self.lock:
                    if "error" in result:
                        self.failed_urls.append(result["url"])
                    else:
                        self.results.append(result)

        return self.results

选项 2：使用线程安全的集合

from queue import Queue

class WebScraper:
    """Web scraper that fetches multiple URLs concurrently."""

    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.results = Queue()  # ✅ Thread-safe
        self.failed_urls = Queue()  # ✅ Thread-safe

    def scrape_urls(self, urls: List[str]) -> List[Dict[str, Any]]:
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(self.fetch_url, url) for url in urls]

            for future in as_completed(futures):
                result = future.result()

                if "error" in result:
                    self.failed_urls.put(result["url"])
                else:
                    self.results.put(result)

        # Convert Queue to list for return
        return list(self.results.queue)

选项 3：从 Future 收集结果（最佳方法）

def scrape_urls(self, urls: List[str]) -> List[Dict[str, Any]]:
    """
    ✅ FIXED: Collect results in main thread, avoiding shared state.
    """
    results = []  # Local to main thread
    failed_urls = []  # Local to main thread
    
    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        futures = [executor.submit(self.fetch_url, url) for url in urls]

        for future in as_completed(futures):
            result = future.result()

            # ✅ Only main thread accesses these lists
            if "error" in result:
                failed_urls.append(result["url"])
            else:
                results.append(result)

    # Update instance variables once, in main thread
    self.results = results
    self.failed_urls = failed_urls
    
    return self.results

Recommended Fix (Complete)

这里是使用 选项 3（最干净的方法）的完全修正版本：

"""
Concurrent web scraper - FIXED VERSION
Results collected in main thread to avoid race conditions.
"""

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Any

import requests


class WebScraper:
    """Web scraper that fetches multiple URLs concurrently."""

    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.results = []
        self.failed_urls = []

    def fetch_url(self, url: str) -> Dict[str, Any]:
        """抓取单个URL并返回结果。”
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            return {
                "url": url,
                "status": response.status_code,
                "content_length": len(response.content),
            }
        except requests.exceptions.RequestException as e:
            return {"url": url, "error": str(e)}

    def scrape_urls(self, urls: List[str]) -> List[Dict[str, Any]]:
        """
        同时抓取多个URL。
        
        ✅ FIXED: Results collected in main thread only.
        """
        results = []  # ✅ Local variables - no race condition
        failed_urls = []
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(self.fetch_url, url) for url in urls]

            for future in as_completed(futures):
                result = future.result()

                # ✅ Only the main thread modifies these lists
                if "error" in result:
                    failed_urls.append(result["url"])
                else:
                    results.append(result)

        # Update instance variables after all threads complete
        self.results = results
        self.failed_urls = failed_urls
        
        return self.results

    def get_stats(self) -> Dict[str, int]:
        """Get scraping statistics."""
        return {
            "total_results": len(self.results),
            "failed_urls": len(self.failed_urls),
            "success_rate": (
                len(self.results) / (len(self.results) + len(self.failed_urls))
                if (len(self.results) + len(self.failed_urls)) > 0
                else 0
            ),
        }


if __name__ == "__main__":
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/status/200",
        "https://httpbin.org/status/404",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/status/500",
    ] * 10  # 50 URLs total

    scraper = WebScraper(max_workers=10)
    results = scraper.scrape_urls(urls)

    print(f"Expected: 50 results")
    print(f"Got: {len(results)} results")  # ✅ Now always 50
    print(f"Stats: {scraper.get_stats()}")

Summary

根本原因：多个线程在未同步的情况下向共享列表（self.results 和 self.failed_urls）追加
影响：丢失更新导致结果计数不一致
最佳修复：在主线程中收集结果到局部变量，然后一次性更新实例变量
为什么这有效：只有一个线程（主线程）访问结果列表，完全消除竞态条件

关键见解：避免在并发代码中共享可变状态。让工作线程返回值，并在协调线程中收集它们。

🔧 内存工具：str_replace /memories/review.md ✓ 结果：文件 /memories/review.md 已成功编辑

============================================================ ✅ Session 1 complete! ============================================================


**What happened?**

1. Claude checked its memory (empty on first run)
2. Identified the bug: **race condition** - multiple threads modifying shared state (`self.results` and `self.failed_urls`) without synchronization
3. Stored the concurrency pattern in memory for future reference

Now let's see the magic - Claude applying this learned pattern in a **new conversation**:

### Example 2: Cross-Conversation Learning

Start a completely new conversation - memory persists!

```python
# 新对话（空消息）
# 加载具有类似并发问题的 API 客户端代码
with open("memory_demo/sample_code/api_client_v1.py", "r") as f:
    code_to_review = f.read()

messages = [
    {"role": "user", "content": "审查以下API客户端代码：\n\n"}```python\n{code_to_review}\n``````
}
]

print("=" * 60)
打印("🚀 SESSION 2: 应用学习到的模式")
print("=" * 60)

# 运行对话循环
response = run_conversation_loop(
    client=client,
    model=MODEL,
    messages=messages,
    memory_handler=memory,
    system="您是一位代码审查员。",
    max_tokens=2048,
    max_turns=5,
    verbose=True,
)

print("\n" + "=" * 60)
print("✅ Session 2 complete!")
print("=" * 60)

============================================================
🚀 SESSION 2: 应用学习到的模式
============================================================

🔄 Turn 1:
  🔧 Memory tool: view /memories
  ✓ Result: Directory: /memories
- review.md

🔄 Turn 2:
  🔧 内存工具：查看 /memories/review.md
  ✓ 结果：    1: # 代码审查：多线程网页爬虫
   2: 
   3: ## Issue
   4: Revi...

🔄 Turn 3:
  🔧 Memory tool: str_replace /memories/review.md
  ✓ Result: File /memories/review.md has been edited successfully

🔄 Turn 4:
💬 Claude: ## 代码审查：异步API客户端

### 总结
这段代码存在与共享可变状态相关的**并发问题**，多个协程在没有适当同步的情况下访问该状态。虽然作者在注释中正确地将这些问题识别为错误，但让我提供一个全面的审查。

然而，这里的实际竞争条件风险**有些夸大**，因为：
- 在CPython中，列表的 `.append()` 操作是原子的
- 整数的 `+=` 操作也是原子的
- 协程只在 `await` 点切换，操作之间没有 `await`

**但是**，这仍然是个问题，因为：
- 它依赖于CPython的实现细节
- 它没有得到语言规范的保证
- 代码不可移植到其他Python实现
- 未来重构可能会引入导致实际竞态条件的 `await` 点

#### 1. **`self.error_count` 上的竞态条件（第 51 行）**
```python
self.error_count += 1  # Not atomic!

问题：即使在异步代码中，此操作也不是原子的。虽然 asyncio 是单线程的，但 += 运算符涉及： 1. Read current value 2. Add 1 3. Write back

在步骤 1-3 之间，另一个协程中的 await 可能会导致更新丢失。

影响：错误计数将不正确/报告不足。

2. `self.responses` 上的不安全列表追加（第 53 行）

self.responses.append(result)  # Not thread-safe in async context!

问题：虽然由于 GIL，list.append() 在 CPython 中在技术上原子，但这： - 是一个实现细节，语言规范不保证 - 是一种不良实践，使代码脆弱 - 仍然有问题，因为列表在被其他地方读取的同时正在被修改

影响：可能的数据损坏、缺失响应或计数不正确。

3. 逻辑错误：错误未包含在结果中

if "error" in result:
    self.error_count += 1  # Counted but not stored!
else:
    self.responses.append(result)  # Only successes stored

问题：失败的请求增加 error_count 但 从未添加到 self.responses。这意味着： - fetch_all() 的返回值只包含成功的响应 - 错误详情丢失 - 摘要计算不一致

⚠️ Design Issues

4. 有状态类设计对于并发操作是反模式

该类维护可变状态（responses、error_count），在并发操作期间被修改。这是问题，因为： - 使客户端不可重用（调用 fetch_all() 两次将混合结果） - 更难推理状态一致性 - 需要在调用之间手动重置

更好的方法：直接返回数据，而不是累积在实例变量中。

5. Inconsistent Error Handling

错误被捕获并转换为字典，但随后与成功处理不同。这使得API不一致，并丢失了宝贵的错误信息。

🟡 Minor Issues

6. Missing Type Hints

该模块导入了Optional但从未使用它。返回类型可以更具体。

7. 主函数中的硬编码测试数据

* 20乘数和特定端点使得测试不够灵活。

修复：使用asyncio.gather()，这对于此用例来说更简单、更符合习惯用法。

5. 缺少HTTP状态码验证

位置：fetch_endpoint()方法

🔧 Memory tool: str_replace /memories/review.md ✓ Result: File /memories/review.md has been edited successfully

🔄 Turn 5: 💬 Claude：这是修正后的版本：

Fix:

"""
Async API client with proper concurrency handling.
"""

import asyncio
from typing import List, Dict, Any
from dataclasses import dataclass

import aiohttp


@dataclass
class FetchResult:
    """Result from fetching an endpoint."""
    endpoint: str
    status: int = 0
    data: Any = None
    error: str = None
    
    @property
    def is_success(self) -> bool:
        return self.error is None


class AsyncAPIClient:
    """Async API client for fetching data from multiple endpoints."""

    def __init__(self, base_url: str):
        self.base_url = base_url

    async def fetch_endpoint(
        self, session: aiohttp.ClientSession, endpoint: str
    ) -> FetchResult:
        """Fetch a single endpoint."""
        url = f"{self.base_url}/{endpoint}"
        try:
            async with session.get(
                url, timeout=aiohttp.ClientTimeout(total=5)
            ) as response:
                data = await response.json()
                return FetchResult(
                    endpoint=endpoint,
                    status=response.status,
                    data=data,
                )
        except Exception as e:
            return FetchResult(
                endpoint=endpoint,
                error=str(e),
            )

    async def fetch_all(self, endpoints: List[str]) -> List[FetchResult]:
        """
        Fetch multiple endpoints concurrently.
        
        Fixed: Returns all results directly without modifying shared state.
        Uses asyncio.gather() which is cleaner than as_completed() for this use case.
        """
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch_endpoint(session, endpoint) for endpoint in endpoints]
            # gather() returns results in order and handles all concurrency
            results = await asyncio.gather(*tasks)
            return results

    @staticmethod
    def get_summary(results: List[FetchResult]) -> Dict[str, Any]:
        """Get summary statistics from results."""
        successes = sum(1 for r in results if r.is_success)
        errors = len(results) - successes
        
        return {
            "total_responses": len(results),
            "successful": successes,
            "errors": errors,
            "success_rate": successes / len(results) if results else 0,
        }


async def main():
    """Test the async API client."""
    client = AsyncAPIClient("https://jsonplaceholder.typicode.com")

    endpoints = [
        "posts/1",
        "posts/2",
        "posts/3",
        "users/1",
        "users/2",
        "invalid/endpoint",  # Will error
    ] * 20  # 120 requests total

    results = await client.fetch_all(endpoints)

    # Separate successes and errors for clarity
    successes = [r for r in results if r.is_success]
    errors = [r for r in results if not r.is_success]

    print(f"Expected: 120 total responses (100 successful, 20 errors)")
    print(f"Got: {len(results)} responses ({len(successes)} successful, {len(errors)} errors)")
    print(f"Summary: {AsyncAPIClient.get_summary(results)}")
    
    if errors:
        print(f"\nFirst error: {errors[0].endpoint} - {errors[0].error}")


if __name__ == "__main__":
    asyncio.run(main())

修复版本中的关键改进：

✅ 无共享可变状态 - 直接返回结果
✅ 所有结果都被捕获 - 包含成功和错误
✅ 使用dataclass类型安全 - 结果具有清晰的架构
✅ 可重用客户端 - 可以安全地多次调用fetch_all()
✅ 一致的错误处理 - 所有结果具有相同的结构
✅ 更简单的并发 - 使用asyncio.gather()代替as_completed()
✅ 静态摘要方法 - 以参数形式接收结果，无状态依赖

替代方案：如果您需要共享状态

如果您绝对需要在实例中累积结果（例如，用于流式/渐进式更新），请使用 asyncio.Lock：

import asyncio

class AsyncAPIClient:
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.responses = []
        self.error_count = 0
        self._lock = asyncio.Lock()  # Add lock

    async def fetch_all(self, endpoints: List[str]) -> List[Dict[str, Any]]:
        async with aiohttp.ClientSession() as session:
            tasks = [self.fetch_endpoint(session, endpoint) for endpoint in endpoints]
            
            for coro in asyncio.as_completed(tasks):
                result = await coro
                
                # Protect shared state with lock
                async with self._lock:
                    if "error" in result:
                        self.error_count += 1
                    else:
                        self.responses.append(result)
        
        return self.responses

然而，我最初展示的无状态设计对于异步代码来说是强烈推荐的。

import asyncio from typing import List, Dict, Any, TypedDict

Testing Recommendation

添加测试以验证所有120个请求都已计入：

async def test_all_results_captured():
    client = AsyncAPIClient("https://jsonplaceholder.typicode.com")
    endpoints = ["posts/1"] * 100
    results = await client.fetch_all(endpoints)
    assert len(results) == 100, f"Expected 100, got {len(results)}"

============================================================ ✅ Session 2 complete! ============================================================


**Notice the difference:**

- Claude **immediately checked memory** and found the thread-safety/concurrency pattern
- Recognized the similar issue in async code **instantly** without re-learning
- Response was **faster** because it applied stored knowledge about shared mutable state

This is **cross-conversation learning** in action!

### Example 3: Context Clearing While Preserving Memory

What happens during a **long review session** with many code files?

- Context fills up with tool results from previous reviews
- But memory (learned patterns) must persist!

Let's trigger **context editing** to see how Claude manages this automatically.

**Note on configuration:** We use `clear_at_least: 50` tokens because memory tool operations have small results (~50-150 tokens each). In production with larger tool results (like web search or code execution), you'd use higher values like 3000-5000 tokens.

```python
# 使用两种清除策略配置上下文管理
# 对于演示的低阈值 - 在生产中使用30-40k个token
CONTEXT_MANAGEMENT = {
    "edits": [
        # 在结合策略时，思考管理必须排在首位
        {
            "type": "clear_thinking_20251015",
            "keep": {"type": "thinking_turns", "value": 1}  # 仅保留最后一次思考
        },
        {
            "type": "clear_tool_uses_20250919",
            "trigger": {"type": "input_tokens", "value": 5000},  # 演示的低阈值
            "keep": {"type": "tool_uses", "value": 2},  # 保留最后两次工具使用
            "clear_at_least": {"type": "input_tokens", "value": 2000}
        }
    ]
}

# 扩展思考配置（对于清除思考策略是必需的）
THINKING = {
    "type": "enabled",
    "budget_tokens": 1024  # 每次思考的预算token
}

# 从上一个会话继续 - 内存持续存在！
print("=" * 60)
print("📚 SESSION 3: 长时间回顾会话，清除上下文")
print("=" * 60)
print()

# 清理消息 - 从上一个会话中移除任何空内容
# 这确保了我们有一个有效的消息状态以继续
cleaned_messages = []
for msg in messages:
    if isinstance(msg.get("content"), list):
        # 过滤掉空内容块
        content = [c for c in msg["content"] if c]
        if content:
            cleaned_messages.append({"role": msg["role"], "content": content})
    elif msg.get("content"):
        cleaned_messages.append(msg)

messages = cleaned_messages

# 审查1：数据处理器（大文件）
with open("memory_demo/sample_code/data_processor_v1.py", "r") as f:
    data_processor_code = f.read()

messages.append({
    "role": "user",
    "content": f"审查这个数据处理器：\n\n```python\n{data_processor_code}\n```"
})

print("📝 审查1：数据处理器")
response = run_conversation_turn(
    client=client,
    model=MODEL,
    messages=messages,
    memory_handler=memory,
    system="You are a code reviewer.",
    context_management=上下文管理,
    thinking=THINKING,
    max_tokens=4096,
    verbose=True
)

# Add response to messages
messages.append({"role": "assistant", "content": response[1]})
if response[2]:
    messages.append({"role": "user", "content": response[2]})

print(f"  📊 输入令牌数: {response[0].usage.input_tokens:,}")
context_cleared, saved = print_context_management_info(response[0])
print()

# Review 2: SQL code
with open("memory_demo/sample_code/sql_query_builder.py", "r") as f:
    sql_code = f.read()

messages.append({
    "role": "user",
    "content": f"审查这个SQL查询构建器：\n\n```python\n{sql_code}\n```"
})

print("📝 审查2：SQL查询构建器")
response = run_conversation_turn(
    client=client,
    model=MODEL,
    messages=messages,
    memory_handler=memory,
    system="You are a code reviewer.",
    context_management=CONTEXT_MANAGEMENT,
    thinking=THINKING,
    max_tokens=4096,
    verbose=True
)

messages.append({"role": "assistant", "content": response[1]})
if response[2]:
    messages.append({"role": "user", "content": response[2]})

print(f"  📊 输入令牌数: {response[0].usage.input_tokens:,}")
context_cleared, saved = print_context_management_info(response[0])
print()

# 第3次审查：添加一次审查以确保触发清除
with open("memory_demo/sample_code/web_scraper_v1.py", "r") as f:
    scraper_code = f.read()

messages.append({
    "role": "user",
    "content": f"快速检查 - 这里有什么问题吗？\n\n```python\n{scraper_code}\n```"
})

```python
print("📝 第3次审查：网络爬虫（应触发清除）")
response = run_conversation_turn(
    client=client,
    model=MODEL,
    messages=messages,
    memory_handler=memory,
    system="You are a code reviewer.",
    context_management=CONTEXT_MANAGEMENT,
    thinking=THINKING,
    max_tokens=4096,
    verbose=True
)

messages.append({"role": "assistant", "content": response[1]})
if response[2]:
    messages.append({"role": "user", "content": response[2]})

print(f"  📊 输入token数: {response[0].usage.input_tokens:,}")
context_cleared, saved = print_context_management_info(response[0])

print()
print("=" * 60)
print("✅ 第3次会话完成！")
print("=" * 60)

============================================================
📚 第3次会话：长审查会话与上下文清除
============================================================

📝 Review 1: Data processor
🧠 思考：用户希望我审查一个存在多个并发和线程安全问题的数据处理程序。让我...
  🔧 Memory tool: str_replace /memories/review.md
  ✓ Result: File /memories/review.md has been edited successfully
  📊 Input tokens: 6,611
  ℹ️ 上下文低于阈值 - 未触发清除

📝 复查 2：SQL 查询构建器
🧠 思考：用户现在要求我复查一个存在 SQL 注入漏洞的 SQL 查询构建器。这是 ...
  🔧 Memory tool: str_replace /memories/review.md
  ✓ Result: File /memories/review.md has been edited successfully
  📊 Input tokens: 7,923
  ✂️  Context editing triggered!
      • 清除 1 次思考，节省 166 个 token
      • After clearing: 7,923 tokens

📝 复查 3：网页爬虫（应触发清除）
🧠 思考：这是一个针对存在线程问题的网页爬虫的快速检查请求。让我快速识别 ...
  🔧 Memory tool: str_replace /memories/review.md
  ✓ Result: File /memories/review.md has been edited successfully
  📊 Input tokens: 9,052
  ✂️  Context editing triggered!
      • 清除 2 次思考，节省 265 个 token
      • After clearing: 9,052 tokens

============================================================
✅ Session 3 complete!
============================================================

What just happened?

As context grew during multiple reviews with extended thinking enabled, context editing was applied: 1. Thinking blocks cleared - Old thinking from previous turns removed first 2. Tool results cleared - Old memory tool results removed when threshold exceeded 3. Memory files intact - Claude can still query learned patterns 4. Token usage managed - Saved tokens from both thinking and tool results

This demonstrates the key benefit: - Short-term memory (conversation context + thinking) → Cleared to save space - Long-term memory (stored patterns) → Persists across sessions

Let’s verify memory survived the clearing:

# 验证清除上下文后内存是否持续存在
import os

print("📂 demo_memory/ 目录中的内存文件：")
print()

for root, dirs, files in os.walk("./demo_memory"):
    # 计算显示的相对路径
    level = root.replace("./demo_memory", "").count(os.sep)
    indent = "  " * level
    folder_name = os.path.basename(root) or "demo_memory"
    print(f"{indent}{folder_name}/")

    sub_indent = "  " * (level + 1)
    for file in files:
        file_path = os.path.join(root, file)
        size = os.path.getsize(file_path)
        print(f"{sub_indent}├── {file} ({size} bytes)")

print()
print("✅ All learned patterns preserved despite context clearing!")```

```📂 demo_memory/ 中的内存文件：

demo_memory/
  memories/
    ├── review.md (318 字节)

✅ 即使清除上下文，所有学习到的模式都得到了保留！```

## 4. How It Works {#how-it-works}

### Memory Tool Architecture

The memory tool is **client-side** - you control the storage. Claude makes tool calls, your application executes them.

#### Memory Tool Commands

| Command | Description | Example |
|---------|-------------|---------|
| `view` | Show directory or file contents | `{"command": "view", "path": "/memories"}` |
| `create` | Create or overwrite a file | `{"command": "create", "path": "/memories/notes.md", "file_text": "..."}` |
| `str_replace` | Replace text in a file | `{"command": "str_replace", "path": "...", "old_str": "...", "new_str": "..."}` |
| `insert` | Insert text at line number | `{"command": "insert", "path": "...", "insert_line": 2, "insert_text": "..."}` |
| `delete` | Delete a file or directory | `{"command": "delete", "path": "/memories/old.txt"}` |
| `rename` | Rename or move a file | `{"command": "rename", "old_path": "...", "new_path": "..."}` |

See `memory_tool.py` for the complete implementation with path validation and security measures.

### Thinking Management (`clear_thinking_20251015`)

When using extended thinking, thinking blocks accumulate and consume tokens. The `clear_thinking` strategy manages these automatically.

**Important**: This strategy requires `thinking` to be enabled in your API call.

**API Call Pattern** (with extended thinking enabled):

```python
response = client.beta.messages.create(
    betas=["context-management-2025-06-27"],  # 必需的beta标志
    model="claude-sonnet-4-6",
    messages=messages,
    tools=[{"type": "memory_20250818", "name": "memory"}],
    thinking={"type": "enabled", "budget_tokens": 10000},  # 启用思考
    context_management={  # 上下文编辑配置
        "edits": [
            {
                "type": "clear_thinking_20251015",
                "keep": {"type": "thinking_turns", "value": 1}  # 仅保留最后一轮
            },
            {
                "type": "clear_tool_uses_20250919",
                "trigger": {"type": "input_tokens", "value": 35000},
                "keep": {"type": "tool_uses", "value": 5}
            }
        ]
    },
    max_tokens=2048
)```

**Key points:**
- `clear_thinking` must come **first** when combining strategies
- Requires extended thinking to be enabled (`thinking={"type": "enabled", ...}`)
- Use `"keep": "all"` to preserve all thinking blocks for maximum cache hits
- Trigger is optional for thinking (clears based on `keep` value)

### Understanding the Demo Code

Key implementation details from `code_review_demo.py`:

``````markdown
python
class CodeReviewAssistant:
    def __init__(self, memory_storage_path="./memory_storage"):
        self.client = Anthropic()
        self.memory_handler = MemoryToolHandler(base_path=memory_storage_path)
        self.messages = []
    
    def review_code(self, code, filename, description=""):
        # 1. 添加用户消息
        self.messages.append({...})
        
        # 2. 与工具执行的对话循环
        while True:
            response = self.client.beta.messages.create(
                model=MODEL,
                system=self._create_system_prompt(),
                messages=self.messages,
                tools=[{"type": "memory_20250818", "name": "memory"}],
                betas=["context-management-2025-06-27"],
                context_management=CONTEXT_MANAGEMENT
            )
            
            # 3. 执行工具使用
            tool_results = []
            for content in response.content:
                if content.type == "tool_use":
                    result = self._execute_tool_use(content)
                    tool_results.append({...})
            
            # 4. 如果有工具使用，则继续，否则完成
            if tool_results:
                self.messages.append({"role": "user", "content": tool_results})
            else:
                break

The key pattern: Keep calling the API while there are tool uses, executing them and feeding results back.

What Claude Actually Learns

This is what makes memory powerful - semantic pattern recognition, not just syntax:

Session 1: Thread-Based Web Scraper

# Bug: 竞态条件
class WebScraper:
    def __init__(self):
        self.results = []  # 共享状态！
    
    def scrape_urls(self, urls):
        with ThreadPoolExecutor() as executor:
            for future in as_completed(futures):
                self.results.append(future.result())  # 竞赛！```

**What Claude Stores in Memory** (example file: `/memories/concurrency_patterns/thread_safety.md`):

When Claude encounters this pattern, it stores the following insights to its memory files:
- **Symptom**: Inconsistent results in concurrent operations
- **Cause**: Shared mutable state (lists/dicts) modified from multiple threads
- **Solution**: Use locks, thread-safe data structures, or return results instead
- **Red flags**: Instance variables in thread callbacks, unused locks, counter increments

---

**Session 2: Async API Client** (New conversation!)

Claude checks memory FIRST, finds the thread-safety pattern, then:
1. **Recognizes** similar pattern in async code (coroutines can interleave too)
2. **Applies** the solution immediately (no re-learning needed)
3. **Explains** with reference to stored knowledge

```python
# 克劳德立即注意到这一点：
```python
async def fetch_all(self, endpoints):
    for coro in asyncio.as_completed(tasks):
        self.responses.append(await coro)  # 同样的模式！

Why This Matters:

❌ Syntax checkers miss race conditions entirely
✅ Claude learns architectural patterns and applies them across contexts
✅ Cross-language: Pattern applies to Go, Java, Rust concurrency too
✅ Gets better: Each review adds to the knowledge base

Sample Code Files

The demo uses these sample files (all have concurrency/thread-safety bugs):

memory_demo/sample_code/web_scraper_v1.py - Race condition: threads modifying shared state
memory_demo/sample_code/api_client_v1.py - Similar concurrency bug in async context
memory_demo/sample_code/data_processor_v1.py - Multiple concurrency issues for long session demo

Let’s look at one:

memory_demo/sample_code/web_scraper_v1.py

"""
并发网络爬虫存在竞态条件错误。
多个线程在不同步的情况下修改共享状态。
"""

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict

import requests


class WebScraper:
    """Web scraper that fetches multiple URLs concurrently."""

    def __init__(self, max_workers: int = 10):
        self.max_workers = max_workers
        self.results = []  # BUG: 多线程访问共享可变状态！
        self.failed_urls = []  # BUG：另一个竞态条件！

    def fetch_url(self, url: str) -> Dict[str, any]:
        """Fetch a single URL and return the result."""
        try:
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            return {
                "url": url,
                "status": response.status_code,
                "content_length": len(response.content),
            }
        except requests.exceptions.RequestException as e:
            return {"url": url, "error": str(e)}

    def scrape_urls(self, urls: List[str]) -> List[Dict[str, any]]:
        """
        Scrape multiple URLs concurrently.

        BUG：未加锁就从多个线程访问self.results！
        这会导致竞态条件，结果可能会丢失或损坏。
        """
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(self.fetch_url, url) for url in urls]

            for future in as_completed(futures):
                result = future.result()

                # 竞态条件：多个线程同时向 self.results 添加内容
                if "error" in result:
                    self.failed_urls.append(result["url"])  # 竞态条件
                else:
                    self.results.append(result)  # RACE CONDITION

        return self.results```

**Bug**: Multiple threads modify `self.results` and `self.failed_urls` without locking!

Claude will:
1. Identify the race conditions
2. Store the pattern in `/memories/concurrency_patterns/thread_safety.md`
3. Apply this concurrency pattern to async code in Session 2

### Demo Overview

We've built a complete Code Review Assistant. The implementation is in `memory_demo/code_review_demo.py`.

**To run the interactive demo:**
```bash
python memory_demo/code_review_demo.py

The demo demonstrates: 1. 会话 1：审查存在错误的 Python 代码 → Claude 学习模式 2. 会话 2：审查类似代码（新对话）→ Claude 应用模式 3. 会话 3：长时间审查会话 → 上下文编辑保持可管理性

7. 最佳实践与安全

Memory Management

Do: - ✅ 存储与任务相关的模式，而非对话历史 - ✅ 使用清晰的目录结构进行组织 - ✅ 使用描述性的文件名 - ✅ 定期审查和清理内存

Don’t: - ❌ 存储敏感信息（密码、API 密钥、PII） - ❌ 允许内存无限制增长 - ❌ Store everything indiscriminately

安全：路径遍历保护

关键：始终验证路径以防止目录遍历攻击。请参阅 memory_tool.py 以获取实现方法。

Security: Memory Poisoning

⚠️ 关键风险：内存文件被读回到 Claude 的上下文中，使其成为提示注入的潜在向量。

Mitigation strategies: 1. 内容净化：在存储前过滤危险模式 2. 内存作用域隔离：按用户/项目隔离
3. 内存审计：记录和扫描所有内存操作 4. 提示工程：指示Claude忽略内存中的指令

请参阅memory_tool.py以获取完整的实现和测试，位于tests/目录下。

结论

What You Accomplished

在本食谱中，你学习了以下内容： - ✅ 实现内存工具以实现跨对话学习（第1和第2节课展示了模式识别的持续性） - ✅ 配置上下文编辑使用token触发器和保留策略（第3节课展示了自动清除功能） - ✅ 应用安全最佳实践包括路径验证和内存中毒预防

Applying These Patterns

For your projects: 1. 从单个内存文件开始，用于存储模式（例如，/memories/patterns.md） 2. 将上下文编辑触发器设置为30-40k tokens以用于生产 3. 实施每个项目的内存隔离以防止交叉污染

Other applications: - 客户支持：存储用户偏好和常见问题解决方案 - 研究助手：在会话间积累领域知识 - 数据分析：记住数据集特征和成功的技术

后续步骤

生产部署：使用claude-code-action进行GitHub PR审查
安全加固：审查memory_tool.py中的内存中毒缓解措施
扩展思维：探索计算密集型任务的思维管理

Resources

内存和上下文管理目前处于测试版。请分享您的反馈以帮助我们改进！

Tip

中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。

长时间运行 Agent 的上下文编辑与 Memory

目录

前置条件

环境配置

VSCode 用户

2. Activate it

3. Install dependencies

4. In VSCode: Select .venv as kernel (top right)

1. 简介：为什么记忆很重要

问题

解决方案

优势

你将学到什么

1. Introduction: Why Memory Matters

The Problem

The Solution

The Benefit

2. 应用场景

🔍 Code Review Assistant

📚 Research Assistant

💬 Customer Support Bot

📊 Data Analysis Helper

3. 快速入门示例

环境配置

示例 1：基本 Memory 使用

In scrape_urls():

选项 2：使用线程安全的集合

选项 3：从 Future 收集结果（最佳方法）

Recommended Fix (Complete)

Summary

2. self.responses 上的不安全列表追加（第 53 行）

3. 逻辑错误：错误未包含在结果中

⚠️ Design Issues

4. 有状态类设计对于并发操作是反模式

5. Inconsistent Error Handling

🟡 Minor Issues

6. Missing Type Hints

7. 主函数中的硬编码测试数据

5. 缺少HTTP状态码验证

修复版本中的关键改进：

替代方案：如果您需要共享状态

Testing Recommendation

What Claude Actually Learns

Sample Code Files

7. 最佳实践与安全

Memory Management

安全：路径遍历保护

Security: Memory Poisoning

结论

What You Accomplished

Applying These Patterns

后续步骤

Resources

2. `self.responses` 上的不安全列表追加（第 53 行）