NoteTool 是为"长时程任务"提供的结构化外部记忆组件。它以 Markdown 文件作为载体，头部使用 YAML 前置元数据记录关键信息，正文用于记录状态、结论、阻塞与行动项等内容。这种设计结合了人类可读性、版本控制友好性和易于回注上下文的特性，是构建长时程智能体的重要工具。
9.4.1 设计理念与应用场景
在深入实现细节之前，让我们首先理解 NoteTool 的设计理念和典型应用场景。
（1）为什么需要 NoteTool?
在第八章中，我们介绍了 MemoryTool，它提供了强大的记忆管理能力。然而，MemoryTool 主要关注对话式记忆——短期工作记忆、情景记忆和语义记忆。对于需要长期追踪、结构化管理的项目式任务，我们需要一种更轻量、更人类友好的记录方式。
NoteTool 填补了这个gap，它提供了：
- 结构化记录：使用 Markdown + YAML 格式，既适合机器解析，也方便人类阅读和编辑
- 版本友好：纯文本格式，天然支持 Git 等版本控制系统
- 低开销：无需复杂的数据库操作，适合轻量级的状态追踪
- 灵活分类：通过
type和tags灵活组织笔记，支持多维度检索
（2）典型应用场景
NoteTool 特别适合以下场景：
场景1：长期项目追踪
想象一个智能体正在协助完成一个大型代码库的重构任务，这可能需要几天甚至几周。NoteTool 可以记录：
task_state：当前阶段的任务状态和进度conclusion：每个阶段结束后的关键结论blocker：遇到的问题和阻塞点action：下一步的行动计划
# 记录任务状态
notes.run({
"action": "create",
"title": "重构项目 - 第一阶段",
"content": "已完成数据模型层的重构,测试覆盖率达到85%。下一步将重构业务逻辑层。",
"note_type": "task_state",
"tags": ["refactoring", "phase1"]
})
# 记录阻塞点
notes.run({
"action": "create",
"title": "依赖冲突问题",
"content": "发现某些第三方库版本不兼容,需要解决。影响范围:业务逻辑层的3个模块。",
"note_type": "blocker",
"tags": ["dependency", "urgent"]
})
场景2：研究任务管理
一个智能研究助手在进行文献综述时，可以使用 NoteTool 记录：
- 每篇论文的核心观点(
conclusion) - 待深入调研的主题(
action) - 重要的参考文献(
reference)
场景3：与 ContextBuilder 配合
在每轮对话前，Agent 可以通过 search 或 list 操作检索相关笔记，并将其注入到上下文中：
# 在 Agent 的 run 方法中
def run(self, user_input: str) -> str:
# 1. 检索相关笔记
relevant_notes = self.note_tool.run({
"action": "search",
"query": user_input,
"limit": 3
})
# 2. 将笔记内容转换为 ContextPacket
note_packets = []
for note in relevant_notes:
note_packets.append(ContextPacket(
content=note['content'],
timestamp=note['updated_at'],
token_count=self._count_tokens(note['content']),
relevance_score=0.7,
metadata={"type": "note", "note_type": note['type']}
))
# 3. 构建上下文时传入笔记
context = self.context_builder.build(
user_query=user_input,
custom_packets=note_packets,
...
)
9.4.2 存储格式详解
NoteTool 采用了 Markdown + YAML 的混合格式，这种设计兼顾了结构化和可读性。
（1）笔记文件格式
每个笔记都是一个独立的 .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
---
# 项目进展 - 第一阶段
## 完成情况
已完成数据模型层的重构,主要改动包括:
1. 统一了实体类的命名规范
2. 引入了类型提示,提升代码可维护性
3. 优化了数据库查询性能
## 测试覆盖
- 单元测试覆盖率: 85%
- 集成测试覆盖率: 70%
## 下一步计划
1. 重构业务逻辑层
2. 解决依赖冲突问题
3. 提升集成测试覆盖率至85%
这种格式的优势：
- YAML 元数据：机器可解析，支持精确的字段提取和检索
- Markdown 正文：人类可读，支持丰富的格式化(标题、列表、代码块等)
- 文件名即 ID：简化管理，每个笔记的文件名就是其唯一标识
（2）索引文件
NoteTool 维护一个 notes_index.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"
}
}
这个索引文件的作用：
- 快速检索：无需打开每个文件，直接从索引中查找
- 元数据管理：集中管理所有笔记的元数据
- 完整性校验：可以检测文件缺失或损坏
9.4.3 核心操作详解
NoteTool 提供了七个核心操作，覆盖了笔记的完整生命周期管理。
（1）create：创建笔记
def _create_note(
self,
title: str,
content: str,
note_type: str = "general",
tags: Optional[List[str]] = None
) -> str:
"""创建笔记
Args:
title: 笔记标题
content: 笔记内容(Markdown格式)
note_type: 笔记类型(task_state/conclusion/blocker/action/reference/general)
tags: 标签列表
Returns:
str: 笔记ID
"""
from datetime import datetime
# 1. 生成唯一ID
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
note_id = f"note_{timestamp}_{len(self.index)}"
# 2. 构建元数据
metadata = {
"id": note_id,
"title": title,
"type": note_type,
"tags": tags or [],
"created_at": datetime.now().isoformat(),
"updated_at": datetime.now().isoformat()
}
# 3. 构建完整的 Markdown 文件内容
md_content = self._build_markdown(metadata, content)
# 4. 保存到文件
file_path = os.path.join(self.workspace, f"{note_id}.md")
with open(file_path, 'w', encoding='utf-8') as f:
f.write(md_content)
# 5. 更新索引
metadata["file_path"] = file_path
self.index[note_id] = metadata
self._save_index()
return note_id
def _build_markdown(self, metadata: Dict, content: str) -> str:
"""构建 Markdown 文件内容(YAML + 正文)"""
import yaml
# YAML 前置元数据
yaml_header = yaml.dump(metadata, allow_unicode=True, sort_keys=False)
# 组合格式
return f"---\n{yaml_header}---\n\n{content}"
使用示例：
from hello_agents.tools import NoteTool
notes = NoteTool(workspace="./project_notes")
note_id = notes.run({
"action": "create",
"title": "重构项目 - 第一阶段",
"content": """## 完成情况
已完成数据模型层的重构,测试覆盖率达到85%。
## 下一步
重构业务逻辑层""",
"note_type": "task_state",
"tags": ["refactoring", "phase1"]
})
print(f"✅ 笔记创建成功,ID: {note_id}")
（2）read：读取笔记
def _read_note(self, note_id: str) -> Dict:
"""读取笔记内容
Args:
note_id: 笔记ID
Returns:
Dict: 包含元数据和内容的字典
"""
if note_id not in self.index:
raise ValueError(f"笔记不存在: {note_id}")
file_path = self.index[note_id]["file_path"]
# 读取文件
with open(file_path, 'r', encoding='utf-8') as f:
raw_content = f.read()
# 解析 YAML 元数据和 Markdown 正文
metadata, content = self._parse_markdown(raw_content)
return {
"metadata": metadata,
"content": content
}
def _parse_markdown(self, raw_content: str) -> Tuple[Dict, str]:
"""解析 Markdown 文件(分离 YAML 和正文)"""
import yaml
# 查找 YAML 分隔符
parts = raw_content.split('---\n', 2)
if len(parts) >= 3:
# 有 YAML 前置元数据
yaml_str = parts[1]
content = parts[2].strip()
metadata = yaml.safe_load(yaml_str)
else:
# 无元数据,全部作为正文
metadata = {}
content = raw_content.strip()
return metadata, content
（3）update：更新笔记
def _update_note(
self,
note_id: str,
title: Optional[str] = None,
content: Optional[str] = None,
note_type: Optional[str] = None,
tags: Optional[List[str]] = None
) -> str:
"""更新笔记
Args:
note_id: 笔记ID
title: 新标题(可选)
content: 新内容(可选)
note_type: 新类型(可选)
tags: 新标签(可选)
Returns:
str: 操作结果消息
"""
if note_id not in self.index:
raise ValueError(f"笔记不存在: {note_id}")
# 1. 读取现有笔记
note = self._read_note(note_id)
metadata = note["metadata"]
old_content = note["content"]
# 2. 更新字段
if title:
metadata["title"] = title
if note_type:
metadata["type"] = note_type
if tags is not None:
metadata["tags"] = tags
if content is not None:
old_content = content
# 更新时间戳
from datetime import datetime
metadata["updated_at"] = datetime.now().isoformat()
# 3. 重新构建并保存
md_content = self._build_markdown(metadata, old_content)
file_path = metadata["file_path"]
with open(file_path, 'w', encoding='utf-8') as f:
f.write(md_content)
# 4. 更新索引
self.index[note_id] = metadata
self._save_index()
return f"✅ 笔记已更新: {metadata['title']}"
（4）search：搜索笔记
def _search_notes(
self,
query: str,
limit: int = 10,
note_type: Optional[str] = None,
tags: Optional[List[str]] = None
) -> List[Dict]:
"""搜索笔记
Args:
query: 搜索关键词
limit: 返回数量限制
note_type: 按类型过滤(可选)
tags: 按标签过滤(可选)
Returns:
List[Dict]: 匹配的笔记列表
"""
results = []
query_lower = query.lower()
for note_id, metadata in self.index.items():
# 类型过滤
if note_type and metadata.get("type") != note_type:
continue
# 标签过滤
if tags:
note_tags = set(metadata.get("tags", []))
if not note_tags.intersection(tags):
continue
# 读取笔记内容
try:
note = self._read_note(note_id)
content = note["content"]
title = metadata.get("title", "")
# 在标题和内容中搜索
if query_lower in title.lower() or query_lower in content.lower():
results.append({
"note_id": note_id,
"title": title,
"type": metadata.get("type"),
"tags": metadata.get("tags", []),
"content": content,
"updated_at": metadata.get("updated_at")
})
except Exception as e:
print(f"[WARNING] 读取笔记 {note_id} 失败: {e}")
continue
# 按更新时间排序
results.sort(key=lambda x: x["updated_at"], reverse=True)
return results[:limit]
（5）list：列出笔记
def _list_notes(
self,
note_type: Optional[str] = None,
tags: Optional[List[str]] = None,
limit: int = 20
) -> List[Dict]:
"""列出笔记(按更新时间倒序)
Args:
note_type: 按类型过滤(可选)
tags: 按标签过滤(可选)
limit: 返回数量限制
Returns:
List[Dict]: 笔记元数据列表
"""
results = []
for note_id, metadata in self.index.items():
# 类型过滤
if note_type and metadata.get("type") != note_type:
continue
# 标签过滤
if tags:
note_tags = set(metadata.get("tags", []))
if not note_tags.intersection(tags):
continue
results.append(metadata)
# 按更新时间排序
results.sort(key=lambda x: x.get("updated_at", ""), reverse=True)
return results[:limit]
（6）summary：笔记摘要
def _summary(self) -> Dict[str, Any]:
"""生成笔记摘要统计
Returns:
Dict: 统计信息
"""
total_count = len(self.index)
# 按类型统计
type_counts = {}
for metadata in self.index.values():
note_type = metadata.get("type", "general")
type_counts[note_type] = type_counts.get(note_type, 0) + 1
# 最近更新的笔记
recent_notes = sorted(
self.index.values(),
key=lambda x: x.get("updated_at", ""),
reverse=True
)[:5]
return {
"total_notes": total_count,
"type_distribution": type_counts,
"recent_notes": [
{
"id": note["id"],
"title": note.get("title", ""),
"type": note.get("type"),
"updated_at": note.get("updated_at")
}
for note in recent_notes
]
}
（7）delete：删除笔记
def _delete_note(self, note_id: str) -> str:
"""删除笔记
Args:
note_id: 笔记ID
Returns:
str: 操作结果消息
"""
if note_id not in self.index:
raise ValueError(f"笔记不存在: {note_id}")
# 1. 删除文件
file_path = self.index[note_id]["file_path"]
if os.path.exists(file_path):
os.remove(file_path)
# 2. 从索引中移除
title = self.index[note_id].get("title", note_id)
del self.index[note_id]
self._save_index()
return f"✅ 笔记已删除: {title}"
9.4.4 与 ContextBuilder 的深度集成
NoteTool 的真正威力在于与 ContextBuilder 的配合使用。让我们通过一个完整的案例来展示这种集成。
（1）场景设定
假设我们正在构建一个长期项目助手，它需要：
- 记录项目的阶段性进展
- 追踪待解决的问题
- 在每次对话时，自动回顾相关笔记
- 基于历史笔记提供连贯的建议
（2）实现示例
from hello_agents import SimpleAgent, HelloAgentsLLM
from hello_agents.context import ContextBuilder, ContextConfig, ContextPacket
from hello_agents.tools import MemoryTool, RAGTool, NoteTool
from datetime import datetime
class ProjectAssistant(SimpleAgent):
"""长期项目助手,集成 NoteTool 和 ContextBuilder"""
def __init__(self, name: str, project_name: str, **kwargs):
super().__init__(name=name, llm=HelloAgentsLLM(), **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. 将笔记转换为 ContextPacket
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(),
custom_packets=note_packets
)
# 4. 调用 LLM
response = self.llm.invoke(context)
# 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 和 action 类型的笔记
blockers = self.note_tool.run({
"action": "list",
"note_type": "blocker",
"limit": 2
})
# 通用搜索
search_results = self.note_tool.run({
"action": "search",
"query": query,
"limit": limit
})
# 合并并去重
all_notes = {note['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[ContextPacket]:
"""将笔记转换为上下文包"""
packets = []
for note in notes:
content = f"[笔记:{note['title']}]\n{note['content']}"
packets.append(ContextPacket(
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['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 hello_agents.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)
（3）运行效果展示
[ContextBuilder] 汇集了 8 个候选信息包
[ContextBuilder] 选择了 7 个信息包,共 3500 tokens
✅ 助手回答:
我注意到您之前记录的笔记中提到了这个问题。根据笔记[重构项目 - 第一阶段],您当前的测试覆盖率已经达到85%,这是一个很好的基础。
关于依赖版本冲突的问题,我建议:
1. **使用虚拟环境隔离**: 为业务逻辑层创建独立的虚拟环境,避免与其他模块的依赖冲突
2. **锁定版本**: 在 requirements.txt 中明确指定所有依赖的精确版本
3. **使用 pipdeptree**: 分析依赖树,找出冲突的根源
这个问题我会标记为 blocker,建议优先解决。
[依据来源: 笔记 note_20250119_153000_0, 项目知识库]
---
📋 笔记摘要:
{
"total_notes": 2,
"type_distribution": {
"action": 1,
"blocker": 1
},
"recent_notes": [
{
"id": "note_20250119_154500_1",
"title": "在重构业务逻辑层时,我遇到了依赖版本冲突的问题...",
"type": "blocker",
"updated_at": "2025-01-19T15:45:00"
},
{
"id": "note_20250119_153000_0",
"title": "我们已经完成了数据模型层的重构...",
"type": "action",
"updated_at": "2025-01-19T15:30:00"
}
]
}
9.4.5 最佳实践
在实际使用 NoteTool 时，以下最佳实践能帮助您构建更强大的长时程智能体：
-
合理的笔记分类：
task_state：记录阶段性进展和状态conclusion：记录重要的结论和发现blocker：记录阻塞问题，优先级最高action：记录下一步行动计划reference：记录重要的参考资料
-
定期清理和归档：
- 对于已解决的 blocker，更新为 conclusion
- 对于过时的 action，及时删除或更新
- 使用 tags 进行版本管理，如
["v1.0", "completed"]
-
与 ContextBuilder 的配合：
- 在每轮对话前检索相关笔记
- 根据笔记类型设置不同的相关性分数(blocker > action > conclusion)
- 限制笔记数量，避免上下文过载
-
人机协作：
- 笔记是人类可读的 Markdown 格式，支持手动编辑
- 使用 Git 进行版本控制，追踪笔记的演化
- 在关键阶段，人工审核 Agent 生成的笔记
-
自动化工作流：
- 定期生成笔记摘要报告
- 基于笔记自动生成项目进度文档
- 将笔记内容同步到其他系统(如 Notion、Confluence)