上下文工程 —— Hello-Agent Chapter 9 Learning

上下文工程

第九章 上下文工程

代码仓库

上下文工程学

符合工程规范的上下文设计

Transformer 本身有着缺陷，而为了防止上下文腐败和上下文漂移，我们需要人为地将上下文控制在一定长度和结构内，控制长度是为了防止重点事项被稀释，控制结构是为了尽可能地提升信息密度。

长度上能做的不多，只有及时遗忘、控制信息长度，反而是上下文的结构需要我们仔细思考。官方在文档中提到，建议按照以下结构控制上下文：

• 系统提示（System Prompt）：语言清晰、直白，信息层级把握在“刚刚好”的高度。

• 工具（Tools）：工具定义了智能体与信息/行动空间的契约，必须促进效率：既要返回token 友好的信息，又要鼓励高效的智能体行为

• 示例（Few-shot）：始终推荐提供示例，但不建议把“所有边界条件”的罗列一股脑塞进提示。请精挑细选一组多样且典型的示例，直接画像“期望行为”。对 LLM 而言，好的示例胜过千言万语。

总的指导思想是：信息充分但紧致。

上下文检索与 Agent 工作

上下文的结构与长度，是随着 Agent 工作实时变动的，特别体现在：需要的时候主动检索，不需要的时候不加载，也就是说，目前 Agent 的工作思想已经不再是预先加载全部信息才工作，而是工作的时候按需加载信息。

这样将信息索引化，Agent 本身只存储索引，由 LLM 自行判断是否需要调用信息，从而实现动态导航的做法，还实现了渐进式披露：每一步由决策与交互所产生的上下文，可以反过来影响决策与交互。

渐进式披露

文件大小 → 暗示复杂度
文件命名 → 暗示用途
时间戳   → 暗示相关性

举一个渐进式披露的例子，用户让 pi、codex 等 agent 修复一个支付错误的bug，这是一个长链路的工作，通常涉及到工程文件、数据库、工具类、git 历史等，然而此时 agent 对以上环境统统不知道，它只有最小单位的工具 ls、read 等命令行命令。

LLM 拿到这个问题后，看了看自己目前所掌握的上下文（命令行工具），发现压根没有关联，所以 LLM 判断：

嗯，用户让我修复一个支付错误的bug，但是根据我所掌握的上下文来看，我只有一些命令行工具，包含 ls、 read 等，我需要先 ls 当前目录结构再进一步判断。

注意，这里就是第一次披露：本次交互中，产生的 我只有一些命令行工具包含 、 ls 的上下文，本身就是有含义的，LLM 从工具 ls 的存在，推断出下一次 大概率需要执行该命令，并将下一次交互的决策定为 阅读文件目录。

于是 LLM 尝试调用工具主动搜寻信息，它查看了当前文件目录，这是 第一次交互，结果它发现了这是一个 java 后端项目，有着完整的目录结构，于是 LLM 判断：

我看到了目录结构，这是一个 Java Web 项目，用户提到 “支付问题”，有可能是存在于 src/backend/pay 包下的 Pay.java，从命名来看应该包含了所有的支付逻辑，让我仔细阅读一下这个文件，查找存在的问题。

这是第二次披露：LLM 本次交互，将 src/ 这个文件结构存入上下文中，LLM 从中所蕴含的名称信息中，推断出问题可能出现在 pay 下面，进而将下一次交互的决策定为 查看文件。

······

总之，渐进式披露的全过程将是：

任务
 |
 v
目录结构
 |
 v
相关目录
 |
 v
相关文件
 |
 v
相关代码片段
 |
 v
具体配置

上下文处理手段

• 压缩：对上下文进行总结，从而不断减少上下文长度。缺点是但是很容易丢失重点，造成上下文腐败和漂移。

• 结构化笔记：维护本地文件、数据库等组件，将上下文持久化，远古时代的 spec coding 就是这种处理。缺点（个人体验）是 LLM 很容易过度维护文件、过度读取无关文件，抢占上下文。

• 构建子 Agent 上下文：专注构建干净的上下文和充足的上下文窗口，让子 Agent 代理执行任务。

ContextBuilder：构建更精准的提示词

我们将构建以下组件：

HelloAgent/
   │
   ├── context/                         # 上下文
   │   ├── countToken.py                # 计算token函数
   │   ├── contextItem.py               # 上下文实体类
   │   ├── contextConfig.py             # 上下文配置类
   │   └── contextBuilder.py            # 上下文构建器
   ├── agent/                           # 上下文
   │   └── contextAwareAgent.py            # 上下文构建器

不过需要注意的是，Hello Agent 中严格意义上并没有实现完全的上下文管理，ContextBuilder 本质上也只是一个提示词生成器（但是要区分开来，ContextBuilder 是要动态查询 RAG 的），其结果是用于替代各种 agent 中的 system prompt 的。

不过，上下文这个概念本来也就是针对 LLM 的，本质也只是输入给 LLM 的 token 及其数量，ContextBuilder 能对 LLM 的输入 token 进行管理，某种意义上也完成了一部分上下文管理功能。

缺点则有很多，最显而易见的是：ContextBuilder 每次与 LLM 交互都是从零开始构建上下文，我们之前已经输入给 LLM 的上下文真的还有必要重新整体构建吗？实则完全不需要，我们再输入增量片段就行了：新增了什么上下文、之前的上下文有什么需要修改的等。

count_tokens

这个函数的作用是计算字符串的 token 数量：

def count_tokens(text: str) -> int:
    """计算文本token数（使用tiktoken）"""
    try:
        encoding = tiktoken.get_encoding("cl100k_base")
        return len(encoding.encode(text))
    except Exception:
        # 降级方案：粗略估算（1 token ≈ 4 字符）
        return len(text) // 4

ContextItem

@dataclass
class ContextItem:
    """上下文实体类"""
    content: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)
    token_count: int = 0
    relevance_score: float = 0.0  # 0.0-1.0
    
    def __post_init__(self):
        """自动计算token数"""
        if self.token_count == 0:
            self.token_count = count_tokens(self.content)

ContextItem 是上下文基本单位，每个候选信息都会被封装为一个 ContextItem，其核心成员是字符串 content，此外需要存储必要的时间戳、元数据、token 数量、（与当前任务的）关联度。

ContextConfig

@dataclass
class ContextConfig:
    """上下文构建配置"""
    max_tokens: int = 8000  # 总预算
    reserve_ratio: float = 0.15  # 生成余量（10-20%）
    min_relevance: float = 0.3  # 最小相关性阈值
    enable_mmr: bool = True  # 启用最大边际相关性（多样性）
    mmr_lambda: float = 0.7  # MMR平衡参数（0=纯多样性, 1=纯相关性）
    system_prompt_template: str = ""  # 系统提示模板
    enable_compression: bool = True  # 启用压缩
    
    def get_available_tokens(self) -> int:
        """获取可用token预算（扣除余量）"""
        return int(self.max_tokens * (1 - self.reserve_ratio))

ContextBuilder 实现

主要流程

ContextBuilder 主要函数是 build，核心流程是：搜集 Gather -> 筛选 Select -> 组织 Structure -> 压缩 Compress，这就是所谓的 GSSC：

# 1. Gather: 收集候选信息
items = self._gather(
    user_query=user_query,
    conversation_history=conversation_history or [],
    system_instructions=system_instructions,
    additional_items=additional_items or []
)
# 2. Select: 筛选与排序
selected_items = self._select(items, user_query)
# 3. Structure: 组织成结构化模板
structured_context = self._structure(
    selected_items=selected_items,
    user_query=user_query,
    system_instructions=system_instructions
)
# 4. Compress: 压缩与规范化（如果超预算）
final_context = self._compress(structured_context)

搜集

先来看搜集函数 _gather：

def _gather(
    self,
    user_query: str,
    conversation_history: List[Message],
    system_instructions: Optional[str],
    additional_items: List[ContextItem]
) -> List[ContextItem]:
    """Gather: 收集候选信息"""
    items = []
    
    # P0: 系统指令（强约束）
    if system_instructions:
        items.append(ContextItem(
            content=system_instructions,
            relevance_score = 1.0,
            metadata={"type": "instructions"}
        ))
    
    # P1: 从记忆中获取任务状态与关键结论
    if self.memory_tool:
        try:
            ······
    
    # P2: 从RAG中获取事实证据
    if self.rag_tool:
        try:
            ······
    
    # P3: 对话历史（辅助材料）
    if conversation_history:
        # 只保留最近N条
        recent_history = conversation_history[-10:]
        history_text = "\n".join([
            f"[{msg.role}] {msg.content}"
            for msg in recent_history
        ])
        items.append(ContextItem(
            content=history_text,
            metadata={"type": "history", "count": len(recent_history)}
        ))
    
    # 添加额外包
    items.extend(additional_items)
    
    return items

我们有三个主要信源：内存中或者向量数据库中的 Memory 数据、RAG 权威知识库、本次 session 保存的 history，分别搜索后拼接返回给 build 函数。

在此之前如果输入了系统提示词（ Optional 的 str），那么 relevance_score 为 1.0，确保能始终被保留。

另外，历史对话只保留数条，官方说法是：

对话历史只保留最近的几条，避免上下文窗口被历史信息占据。

筛选

到这里为止，我们已经搜索到了相关的知识，为了防止不符合要求的知识被误搜索，我们需要再次进行校验，所以我们有 _select 函数：

def _select(
    self,
    items: List[ContextItem],
    user_query: str
) -> List[ContextItem]:
    """Select: 基于分数与预算的筛选"""
    # 1) 计算相关性（jieba 中文分词后的关键词重叠）
    import jieba
    query_tokens = set(jieba.cut(user_query.lower()))
    # 去掉停用词和标点，只保留有意义的词
    import re
    stopwords = {"了", "的", "是", "吗", "呢", "怎么", "如何", "什么", "在", "我", "你",
                 "一个", "这个", "那个", "与", "和", "及", "或", "？", "?", "，", ",", "。",
                 ".", "！", "!", "：", ":", "；", ";"}
    query_tokens = {t for t in query_tokens if t.strip() and t not in stopwords}
    
    for item in items:
        content_tokens = set(jieba.cut(item.content.lower()))
        content_tokens = {t for t in content_tokens if t.strip() and t not in stopwords}
        if len(query_tokens) > 0:
            overlap = len(query_tokens & content_tokens)
            item.relevance_score = overlap / len(query_tokens)
        else:
            item.relevance_score = 0.0
    
    # 2) 计算新近性（指数衰减）
    def recency_score(ts: datetime) -> float:
        delta = max((datetime.now() - ts).total_seconds(), 0)
        tau = 3600  # 1小时时间尺度，可暴露到配置
        return math.exp(-delta / tau)
    
    # 3) 计算复合分：0.7*相关性 + 0.3*新近性
    scored_items: List[Tuple[float, ContextItem]] = []
    for p in items:
        rec = recency_score(p.timestamp)
        score = 0.7 * p.relevance_score + 0.3 * rec
        scored_items.append((score, p))
    
    # 4) 系统指令单独拿出，固定纳入
    system_items = [p for (_, p) in scored_items if p.metadata.get("type") == "instructions"]
    remaining = [p for (s, p) in sorted(scored_items, key=lambda x: x[0], reverse=True)
                 if p.metadata.get("type") != "instructions"]
    
    # 5) 依据 min_relevance 过滤（对非系统包）
    filtered = [p for p in remaining if p.relevance_score >= self.config.min_relevance]
    
    # 6) 按预算填充
    available_tokens = self.config.get_available_tokens()
    selected: List[ContextItem] = []
    used_tokens = 0
    
    # 先放入系统指令（不排序）
    for p in system_items:
        if used_tokens + p.token_count <= available_tokens:
            selected.append(p)
            used_tokens += p.token_count
    
    # 再按分数加入其余
    for p in filtered:
        if used_tokens + p.token_count > available_tokens:
            continue
        selected.append(p)
        used_tokens += p.token_count
    
    return selected

筛选机制有两个重要指标：相关性 relevance_socre 和新近度 recency_score，根据 0.7 和 0.3 的权重计算最终分数，主要基于最终分数计算做筛选。筛选过程则是从高到低排序，取一直到最大 token 数耗尽前的选项。

另外，如果相关性过低（即分数来源完全是新近度），则进行斩杀，因为几乎无意义，宁可错杀不可放过。

官方的算法有不小的问题：计算相关性上太不严谨，使用的是交集占比算法，这个实现只关心”query 里的词有多少出现在内容里”，不关心内容里有多少词没出现在 query 里；而且这个方法也不适用中文，所以这里我们采用 jieba 包进行分词。另外在真正生产环境中，应当上嵌入模型。

结构化

前面两个步骤分别完成了信息粗略搜集和精细筛选，我们认为存活下来的信息是有价值的，接下来进行结构化输出，我们将提示词设置为以下结构：

1. [Role & Policies]

2. [Task]

3. [Evidence]

4. [Context]

5. [Output]

压缩

结构化的提示词，无法保证不超出 token 限制，如果超限，就必须进行信息压缩。

def _compress(self, context: str) -> str:
    """Compress: 压缩与规范化"""
    if not self.config.enable_compression:
        return context
    
    current_tokens = count_tokens(context)
    available_tokens = self.config.get_available_tokens()
    
    if current_tokens <= available_tokens:
        return context
    
    # 简单截断策略（保留前N个token）
    # 实际应用中可用LLM做高保真摘要
    print(f"⚠️ 上下文超预算 ({current_tokens} > {available_tokens})，执行截断")
    
    # 按段落截断，保留结构
    lines = context.split("\n")
    compressed_lines = []
    used_tokens = 0
    
    for line in lines:
        line_tokens = count_tokens(line)
        if used_tokens + line_tokens > available_tokens:
            break
        compressed_lines.append(line)
        used_tokens += line_tokens
    
    return "\n".join(compressed_lines)

这个算法仍然是十分粗糙原始的简单截断策略，对每个分区内的信息，分别执行截断，割舍掉后续的信息，保证 token 数不超限。

测试

写一个测试程序：

# 1. 初始化工具
memory_tool = MemoryTool(user_id="user123")
rag_tool = RAGTool(knowledge_base_path="./knowledge_base")
# 2. 创建 ContextBuilder
config = ContextConfig(
    max_tokens=3000,
    reserve_ratio=0.2,
    min_relevance=0.2,
    enable_compression=True
)
builder = ContextBuilder(
    memory_tool=memory_tool,
    rag_tool=rag_tool,
    config=config
)
# 3. 准备对话历史
conversation_history = [
    Message(content="我正在开发一个数据分析工具", role="user", timestamp=datetime.now()),
    Message(content="很好!数据分析工具通常需要处理大量数据。您计划使用什么技术栈?", role="assistant", timestamp=datetime.now()),
    Message(content="我打算使用Python和Pandas,已经完成了CSV读取模块", role="user", timestamp=datetime.now()),
    Message(content="不错的选择!Pandas在数据处理方面非常强大。接下来您可能需要考虑数据清洗和转换。", role="assistant", timestamp=datetime.now()),
]
# 4. 添加一些记忆
memory_tool.run({
    "action": "add",
    "content": "用户正在开发数据分析工具,使用Python和Pandas",
    "memory_type": "semantic",
    "importance": 0.8
})
memory_tool.run({
    "action": "add",
    "content": "已完成CSV读取模块的开发",
    "memory_type": "episodic",
    "importance": 0.7
})
# 5. 构建上下文
context = builder.build(
    user_query="如何优化Pandas的内存占用?",
    conversation_history=conversation_history,
    system_instructions="你是一位资深的Python数据工程顾问。你的回答需要:1) 提供具体可行的建议 2) 解释技术原理 3) 给出代码示例"
)
print("=" * 80)
print("构建的上下文:")
print("=" * 80)
print(context)
print("=" * 80)

测试结果：

================================================================================
构建的上下文:
================================================================================
[Role & Policies]
你是一位资深的Python数据工程顾问。你的回答需要:1) 提供具体可行的建议 2) 解释技术原理 3) 给出代码示例

[Task]
用户问题：如何优化Pandas的内存占用?

[Evidence]
事实与引用：

🔍 搜索 '如何优化Pandas的内存占用?' 的结果（共 2 条）:
  [1] [semantic] 用户正在开发数据分析工具,使用Python和Pandas  (importance=0.80, id=5a587bb9...)
  [2] [episodic] 已完成CSV读取模块的开发  (importance=0.70, id=d4262c1b...)


[Context]
对话历史与背景：
[user] 我正在开发一个数据分析工具
[assistant] 很好!数据分析工具通常需要处理大量数据。您计划使用什么技术栈?
[user] 我打算使用Python和Pandas,已经完成了CSV读取模块
[assistant] 不错的选择!Pandas在数据处理方面非常强大。接下来您可能需要考虑数据清洗和转换。

[Output]
                            请按以下格式回答：
                            1. 结论（简洁明确）
                            2. 依据（列出支撑证据及来源）
                            3. 风险与假设（如有）
                            4. 下一步行动建议（如适用）
================================================================================

是正确的。

ContextAwareAgent

我们继承 SimpleAgent，做一个简单的能够运用上下文的 Agent。

class ContextAwareAgent(SimpleAgent):
    """具有上下文感知能力的 Agent"""

    def __init__(self, name: str, llm: LLM, **kwargs):
        super().__init__(name=name, llm=llm, system_prompt=kwargs.get("system_prompt", ""))

        # 初始化上下文构建器
        self.memory_tool = MemoryTool(user_id=kwargs.get("user_id", "default"))
        self.rag_tool = RAGTool(knowledge_base_path=kwargs.get("knowledge_base_path", "./kb"))

        self.context_builder = ContextBuilder(
            memory_tool=self.memory_tool,
            rag_tool=self.rag_tool,
            config=ContextConfig(max_tokens=4000)
        )

        self.conversation_history = []

    def run(self, user_input: str) -> str:
        """运行 Agent,自动构建优化的上下文"""

        # 1. 使用 ContextBuilder 构建优化的上下文
        optimized_context = self.context_builder.build(
            user_query=user_input,
            conversation_history=self.conversation_history,
            system_instructions=self.system_prompt
        )

        # 2. 使用优化后的上下文调用 LLM
        messages = [
            {"role": "system", "content": optimized_context},
            {"role": "user", "content": user_input}
        ]
        response = self.llm.think(messages)

        # 3. 更新对话历史
        from datetime import datetime

        self.conversation_history.append(
            Message(content=user_input, role="user", timestamp=datetime.now())
        )
        self.conversation_history.append(
            Message(content=response, role="assistant", timestamp=datetime.now())
        )

        # 4. 将重要交互记录到记忆系统
        self.memory_tool.run({
            "action": "add",
            "content": f"Q: {user_input}\nA: {response[:200]}...",  # 摘要
            "memory_type": "episodic",
            "importance": 0.6
        })

        return response

我们只是将 run 函数进行了改写，我们集成了 ContextBuilder，生成规范的系统提示词然后再调用 LLM。使用案例：

# 使用示例
agent = ContextAwareAgent(
    name="数据分析顾问",
    llm=LLM(),
    system_prompt="你是一位资深的Python数据工程顾问。",
    user_id="user123",
    knowledge_base_path="./data_science_kb"
)

response = agent.run("如何优化Pandas的内存占用?")
print(response)

结果：

1. **结论（简洁明确）**
优化Pandas内存占用的核心策略包括：**使用更紧凑的数据类型（如`category`、`float32`、`int8`）、分块读取数据、仅加载必要列、及时清理中间变量**。针对你的场景（已完成CSV读取模块），最直接有效的措施是从`read_csv`开始控制数据类型和列选择，并对现有DataFrame进行类型压缩。

2. **依据（列出支撑证据及来源）**
- 根据上下文证据[1]，你正在开发数据分析工具，使用Python和Pandas，因此内存优化需围绕Pandas核心操作展开。
- 根据证据[2]，你已完成CSV读取模块，说明数据加载是主要入口，优化重点应放在`pd.read_csv`参数及加载后的数据清洗上。
- 上述证据未包含具体技术细节，以下建议基于Python数据工程通用最佳实践：
  - **数据类型优化**：`int64`可降为`int8/int16/int32`，`float64`可降为`float32`，低基数字符串列转为`category`类型，可减少50%以上内存。
  - **分块处理**：使用`read_csv(..., chunksize=10000)`迭代处理，避免一次性载入全部数据。
  - **列裁剪**：通过`usecols`仅读取分析所需列，减少无用数据占用。
  - **内存监控与清理**：使用`df.info(memory_usage='deep')`定位高占用列，并定期`del`临时变量配合`gc.collect()`。

3. **风险与假设**
- 假设内存瓶颈来自DataFrame本身的数据量或数据类型，而非系统配置或外部依赖。
- **精度风险**：将`float64`降为`float32`可能丢失微小数值精度，需根据业务场景评估。
- **类别风险**：`category`类型在低基数列（如性别、城市）效果显著，但若列基数过高（如ID），反而可能增加内存。
- **分块副作用**：分块处理会使代码逻辑更复杂，且某些操作（如跨块计算）需要额外聚合，可能增加开发时间。
- 若使用PyArrow后端（`dtype_backend='pyarrow'`），需确保环境已安装`pyarrow`，否则会报错。

4. **下一步行动建议**
- 首先，在现有CSV读取模块中执行`df.info(memory_usage='deep')`，打印各列实际内存，找出高消耗列。
- 其次，针对这些列，在`read_csv`中显式指定`dtype`（如`{'列名': 'int32'}`），并只保留必要列。
- 如果数据量极大（如数十GB），建议将读取方式改为`chunksize`分块，或考虑用`Dask`/`polars`替代。
- 如果已读取的DataFrame无法重新加载，可编写一个类型压缩函数，遍历列并安全转换数据类型（例如自动判断唯一值占比，将适宜列转为`category`）。
- 最后，建立内存基准测试，优化前后对比，确保改进有效且不影响分析结果。

可以看到，LLM 的解答更加具体了，说明我们的 ContextBuilder 确实有用。

NoteTool：结构化笔记

为什么引入 NoteTool？

为什么 Hello Agent 要设计 NoteTool？我们在之前的第八章中已经设计了 MemoryTool，能够将工作记忆、情景记忆、语义记忆存入数据库以供 LLM 随时调用，有什么缺点吗？官方认为是有的：

然而，MemoryTool 主要关注对话式记忆——短期工作记忆、情景记忆和语义记忆。

这意味着只依靠 MemoryTool 无法很好地完成长程任务，为什么呢？deepseek v4 flash 0731 认为主要问题有以下：

1. MemoryTool 在使用中是 append-only 的，几乎不会对已有的信息进行修改，而是存入一条 修改这条信息的信息，本质上是 流水账 或者 还远历史改动。这样一来，信息前后矛盾的状况随着信息增加会越发严重，加上搜索 RAG 时是按照相关性排序，而非正确性，所以LLM 无法知道哪条信息是最新的、可信的。

2. MemoryTool 的存储格式基本固定，即 用户提出什么任务，ai如何如何执行 的对话形式。而在工程日志中，我们必须要记录项目当前状态、阶段主要任务、阶段主要改动与结论、遇到的问题等有时效性，逻辑链路的信息。

MemoryTool 无法解决的最主要问题是：信息是否 100% 可信？显然 MemoryTool 做不到这一点。所以引入了 NoteTool 记录 Note，这个组件比起 MemoryTool 的 还原历史，实际更侧重于修改、修订、维护，目的是维护权威信源。

此外，Note 还兼顾了轻量级、人类可读的特点，使用 .md 格式存储到文件目录，不需要存储到数据库。

Note 的存储格式

单个 Note 文件夹采用 .md 存储，在开头使用 yml 格式存储元数据，例如一个名为 note_20250119_153000_0.md 的文件：

---
id: note_20250119_153000_0
title: 项目进展 - 第一阶段
type: task_state
tags: [refactoring, phase1, backend]
created_at: 2025-01-19T15:30:00
updated_at: 2025-01-19T15:30:00
---

# 项目进展 - 第一阶段

## 完成情况

已完成数据模型层的重构,主要改动包括:

······

此外，我们还需要单独维护一个 json 文件统一建立索引：

{
  "note_20250119_153000_0": {
    "id": "note_20250119_153000_0",
    "title": "项目进展 - 第一阶段",
    "type": "task_state",
    "tags": ["refactoring", "phase1", "backend"],
    "created_at": "2025-01-19T15:30:00",
    "updated_at": "2025-01-19T15:30:00",
    "file_path": "./notes/note_20250119_153000_0.md"
  }
}

NoteTool 实现

我们在 tool.bulitin 下面创建一个 noteTool.NoteTool，这个工具我们不需要连接数据库底层，只是单纯的文件操作，所以全部操作直接在里面实现就行。

另外，我们还是按照统一接口的形式，采用操作符形式进行调用。

初始化

class NoteTool(Tool):
    def __init__(
        self,
        workspace: str = "./notes",
        auto_backup: bool = True,
        max_notes: int = 1000,
        expandable: bool = False
    ):
        super().__init__(
            name="note",
            description="笔记工具 - 创建、读取、更新、删除结构化笔记，支持任务状态、结论、阻塞项等类型",
            expandable=expandable
        )
        
        self.workspace = Path(workspace)
        self.auto_backup = auto_backup
        self.max_notes = max_notes
        
        # 确保工作目录存在
        self.workspace.mkdir(parents=True, exist_ok=True)
        
        # 笔记索引文件
        self.index_file = self.workspace / "notes_index.json"
        self._load_index()

最重要成员是 workspace，作为 Note 存放目录，索引文件 notes_index.json 也会存放到此。

auto_backup 和 expandable 顾名思义，分别是自动备份、可拓展，后者需要搭配 max_notes 进行。

统一入口 run 函数

def run(self, parameters: Dict[str, Any]) -> str:
    """执行工具（非展开模式）"""
    if not self.validate_parameters(parameters):
        return "❌ 参数验证失败"

    action = parameters.get("action")

    # 根据action调用对应的方法，传入提取的参数
    if action == "create":
        return self._create_note(
            title=parameters.get("title"),
            content=parameters.get("content"),
            note_type=parameters.get("note_type", "general"),
            tags=parameters.get("tags")
        )
    elif action == "read":
        # ······
    else:
        return f"❌ 不支持的操作: {action}"

调用时，依照以下方式进行：

NoteTool(workspace="./project_notes").run({
    "action": "create",
    "title": "项目进展",
    "content": "已完成需求分析，下一步：设计方案",
    "note_type": "task_state",
    "tags": ["milestone", "phase1"]
})

主要采用字典输入，必须给出 action 字段，这里我们预设了 create update delete list search summary 六种操作。

代码略。

集成到 Agent

同样继承 SimpleAgent 创建一个 ProjectAssistantAgent：

class ProjectAssistant(SimpleAgent):
    """长期项目助手,集成 NoteTool 和 ContextBuilder"""

    def __init__(self, name: str, project_name: str, **kwargs):
        super().__init__(name=name, llm=LLM(), **kwargs)

        self.project_name = project_name

        # 初始化工具
        self.memory_tool = MemoryTool(user_id=project_name)
        self.rag_tool = RAGTool(knowledge_base_path=f"./{project_name}_kb")
        self.note_tool = NoteTool(workspace=f"./{project_name}_notes")

        # 初始化上下文构建器
        self.context_builder = ContextBuilder(
            memory_tool=self.memory_tool,
            rag_tool=self.rag_tool,
            config=ContextConfig(max_tokens=4000)
        )

        self.conversation_history = []

    def run(self, user_input: str, note_as_action: bool = False) -> str:
        """运行助手,自动集成笔记"""

        # 1. 从 NoteTool 检索相关笔记
        relevant_notes = self._retrieve_relevant_notes(user_input)

        # 2. 将笔记转换为 ContextItem
        note_packets = self._notes_to_packets(relevant_notes)

        # 3. 构建优化的上下文
        context = self.context_builder.build(
            user_query=user_input,
            conversation_history=self.conversation_history,
            system_instructions=self._build_system_instructions(),
            additional_items=note_packets
        )

        # 4. 调用 LLM
        messages = [
            {"role": "system", "content": context},
            {"role": "user", "content": user_input}
        ]
        response = self.llm.think(messages)

        # 5. 如果需要,将交互记录为笔记
        if note_as_action:
            self._save_as_note(user_input, response)

        # 6. 更新对话历史
        self._update_history(user_input, response)

        return response

    def _retrieve_relevant_notes(self, query: str, limit: int = 3) -> List[Dict]:
        """检索相关笔记（结构化）"""
        try:
            # 优先检索 blocker 类型的笔记
            blockers = self.note_tool.search_notes_structured(
                query="",
                note_type="blocker",
                limit=2
            )

            # 通用搜索
            search_results = self.note_tool.search_notes_structured(
                query=query,
                limit=limit
            )

            # 合并并去重
            all_notes = {note['id']: note for note in blockers + search_results}
            return list(all_notes.values())[:limit]

        except Exception as e:
            print(f"[WARNING] 笔记检索失败: {e}")
            return []

    def _notes_to_packets(self, notes: List[Dict]) -> List[ContextItem]:
        """将笔记转换为上下文包"""
        packets = []

        for note in notes:
            content = f"[笔记:{note['title']}]\n{note['content']}"

            packets.append(ContextItem(
                content=content,
                timestamp=datetime.fromisoformat(note['updated_at']),
                token_count=len(content) // 4,  # 简单估算
                relevance_score=0.75,  # 笔记具有较高相关性
                metadata={
                    "type": "note",
                    "note_type": note['type'],
                    "note_id": note['id']
                }
            ))

        return packets

    def _save_as_note(self, user_input: str, response: str):
        """将交互保存为笔记"""
        try:
            # 判断应该保存为什么类型的笔记
            if "问题" in user_input or "阻塞" in user_input:
                note_type = "blocker"
            elif "计划" in user_input or "下一步" in user_input:
                note_type = "action"
            else:
                note_type = "conclusion"

            self.note_tool.run({
                "action": "create",
                "title": f"{user_input[:30]}...",
                "content": f"## 问题\n{user_input}\n\n## 分析\n{response}",
                "note_type": note_type,
                "tags": [self.project_name, "auto_generated"]
            })

        except Exception as e:
            print(f"[WARNING] 保存笔记失败: {e}")

    def _build_system_instructions(self) -> str:
        """构建系统指令"""
        return f"""你是 {self.project_name} 项目的长期助手。

你的职责:
1. 基于历史笔记提供连贯的建议
2. 追踪项目进展和待解决问题
3. 在回答时引用相关的历史笔记
4. 提供具体、可操作的下一步建议

注意:
- 优先关注标记为 blocker 的问题
- 在建议中说明依据来源(笔记、记忆或知识库)
- 保持对项目整体进度的认识"""

    def _update_history(self, user_input: str, response: str):
        """更新对话历史"""
        from core.message import Message

        self.conversation_history.append(
            Message(content=user_input, role="user", timestamp=datetime.now())
        )
        self.conversation_history.append(
            Message(content=response, role="assistant", timestamp=datetime.now())
        )

        # 限制历史长度
        if len(self.conversation_history) > 10:
            self.conversation_history = self.conversation_history[-10:]

使用案例：

# 使用示例
assistant = ProjectAssistant(
    name="项目助手",
    project_name="data_pipeline_refactoring"
)

# 第一次交互:记录项目状态
response = assistant.run(
    "我们已经完成了数据模型层的重构,测试覆盖率达到85%。下一步计划重构业务逻辑层。",
    note_as_action=True
)

# 第二次交互:提出问题
response = assistant.run(
    "在重构业务逻辑层时,我遇到了依赖版本冲突的问题,该如何解决?"
)

# 查看笔记摘要
summary = assistant.note_tool.run({"action": "summary"})
print(summary)

输出结果：

📊 笔记摘要

总笔记数: 6

按类型统计:
  • action: 6

目录会新建一个文件夹 data_pipeline_refactoring_notes，下面会有 md 格式的 note_xxx，还有索引文件 note_index.json：

{
  "notes": [
    {
      "id": "note_20260805_092053_0",
      "title": "我们已经完成了数据模型层的重构,测试覆盖率达到85%。下一步...",
      "type": "action",
      "tags": [
        "data_pipeline_refactoring",
        "auto_generated"
      ],
      "created_at": "2026-08-05T09:20:53.353006"
    },
    {
      "id": "note_20260805_092400_1"
      ······
    }]
}

TerminalTool：即时文件阅读工具

为什么引入 TerminalTool？

前面不管是 ContextBuilder 还是 NoteTool，都是在针对用户与 agent 交互产出的内容：session 对话记录，工程进度等，我们可以人为地强制地控制其长度，不管是从持久化到数据库时，还是通过 RAG 读取时，我们都能通过控制 token 数量，避免占用过大的上下文窗口。

然而我们忽略了一个重要的问题：Agent 工作空间下面的文件长度算不算？假如让 Agent 维护一个超大的生产级别代码库，其长度一万个 agent 同时读都没法放下，这个时候该怎么办？

当然，即便排除这种情况，仅针对于个人而言，目录下的文件也不应该占用过高，甚至不应该占用上下文：因为它和 RAG 一样都是持久化到本地的，自然可以与 RAG 进行一样方式的处理：渐进式披露，需要时才加载，不需要时遗忘。

此外，过大的文件长度还会拖累 agent 的阅读速度，因为 agent 事先是不知道文件系统（NoteTool 尚且还有索引文件）的目录，为了完整地了解文件目录和文件内容，它只能一个一个地读取。假如我们要求 agent 读文件目录是轻量级、实时响应的，就更不应该使 agent 完全扫盘。

所以，我们引入 TerminalTool，将 agent 扫盘等一切读取文件目录的行为，交给这个工具，使得 LLM 能在无预先索引的情况下，更快速地响应文件操作命令。

TerminalTool 的安全策略

多层安全策略：命令白名单、沙箱环境、超时控制和限制输出。

TerminalTool 实现

核心功能

TerminalTool 有两大核心功能：执行命令，和目录导航。

执行命令功能，由 _execute_command 函数执行，而它内部是使用 subprocess 包，才能实现 python 进程内执行 shell 命令。

对于目录导航，本质也就是一条 cd 命令，TerminalTool 之所以不在 _execute_command 内进行而是在 _handle_cd 函数内做特殊处理，是因为其他命令都是 输入命令 - 得到输出 形式，而 cd 没有输出，而是更改 TerminalTool 自身成员 current_dir 的值。

统一入口 run 函数

我们针对 Tool 的规范是支持统一入口，通过字典设定操作符执行不同指令，这里 run 函数：

def run(self, parameters: Dict[str, Any]) -> str:
    """执行工具"""
    if not self.validate_parameters(parameters):
        return "❌ 参数验证失败"
    
    command = parameters.get("command", "").strip()
    
    if not command:
        return "❌ 命令不能为空"
    
    # 解析命令
    try:
        parts = shlex.split(command)
    except ValueError as e:
        return f"❌ 命令解析失败: {e}"
    
    if not parts:
        return "❌ 命令不能为空"
    
    base_command = parts[0]
    
    # 检查命令是否在白名单中
    if base_command not in self.ALLOWED_COMMANDS:
        return f"❌ 不允许的命令: {base_command}\n允许的命令: {', '.join(sorted(self.ALLOWED_COMMANDS))}"
    
    # 特殊处理 cd 命令
    if base_command == 'cd':
        return self._handle_cd(parts)
    
    # 执行命令
    return self._execute_command(command)

白名单安全

我们前面提到了 Terminal 的安全策略，其中命令白名单在 run 函数内部实现，也就是检查命令是否在白名单内：

# 允许的命令白名单（跨平台）
ALLOWED_COMMANDS = {
    # 文件列表与信息
    'ls', 'dir', 'tree',
    # 文件内容查看
    'cat', 'type', 'head', 'tail', 'less', 'more',
    # 文件搜索
    'find', 'where', 'grep', 'egrep', 'fgrep', 'findstr',
    # 文本处理
    'wc', 'sort', 'uniq', 'cut', 'awk', 'sed',
    # 目录操作
    'pwd', 'cd',
    # 文件信息
    'file', 'stat', 'du', 'df',
    # 其他
    'echo', 'which', 'whereis',
    # 代码执行
    'python', 'python3', 'node', 'bash', 'sh', 'powershell', 'cmd',
}

TerminalTool 与 ContextBuilder、NoteTool、MemoryTool 的联合使用

经过前面几个组件的学习，我们已经大致了解了上下文工程的主要架构：

1. ContextBuilder 能够为 LLM 构建较为完备的上下文环境，让 LLM 充分理解任务。

2. NoteTool、MemoryTool 分别将工程信息和互动记忆本地化到 RAG 系统，供 ContextBuilder 查询。

那么 TerminalTool 的角色应该是怎么样的？TerminalTool 和三者均可单独协同工作，不过我们力求将四者融会贯通：

LLM 调用 TerminalTool 的结果按需存入 NoteTool 和 MemoryTool，ContextBuilder 再将完备的上下文输入到 LLM。

总结

本章我们聚焦上下文，结束了之前随性输入 LLM 的做法。

为了将输入 LLM 的上下文完备化，我们构建了 ContextBuilder，它能够从 RAG 系统中查询知识，为 LLM 构建最适合当前任务的上下文环境。在具体实现上，我们遵照 GSSC，即先搜集，再筛选，最后再结构化与压缩。这样一来，我们能够在尽量少的上下文占用中，塞进更高密度的任务相关信息，让 LLM 执行效果更佳。

而接下来，我们提出当前基于 MemoryTool 的 RAG 系统中存在记忆过重、缺乏灵活性、权威性不足等问题，提出引入 NoteTool， 其最大的价值有二：一是能够实时维护权威信息，避免 MemoryTool 的 add-only 带来的信息矛盾而使 LLM 效果下降的问题；二是能够构建结构灵活、轻量且人类可读的 md 文档，增强了可维护性的同时，降低了 token 用量。

最后，我们聚焦于 Agent维护庞大文件系统 的真实问题，引入了 TerminalTool，它能够为 Agent 提供实时的文件系统探索能力，同时避免 Agent 直接扫盘带来的极高上下文占用。

我们应该在真实生产环境中，将以上组件联合使用，各司其职。