Merge remote-tracking branch 'origin/main'

# Conflicts:
#	README.md
#	README_ZH.md
This commit is contained in:
方应 2026-03-30 19:24:53 +08:00
commit 949b3bf695
39 changed files with 1793 additions and 628 deletions

176
README.md
View file

@ -17,6 +17,10 @@
<a href="https://deepwiki.com/agentscope-ai/ReMe"><img src="https://img.shields.io/badge/DeepWiki-Ask_Devin-navy.svg" alt="DeepWiki"></a>
</p>
<p align="center">
<a href="https://trendshift.io/repositories/20528" target="_blank"><img src="https://trendshift.io/api/badge/repositories/20528" alt="agentscope-ai%2FReMe | Trendshift" style="width: 250px; height: 55px;" width="250" height="55"/></a>
</p>
<p align="center">
<strong>A memory management toolkit for AI agents — Remember Me, Refine Me.</strong><br>
</p>
@ -25,7 +29,8 @@
---
🧠 ReMe is a memory management framework designed for **AI agents**, providing both [file-based](#-file-based-memory-system-remelight) and [vector-based](#-vector-based-memory-system) memory systems.
🧠 ReMe is a memory management framework designed for **AI agents**, providing
both [file-based](#-file-based-memory-system-remelight) and [vector-based](#-vector-based-memory-system) memory systems.
It tackles two core problems of agent memory: **limited context window** (early information is truncated or lost in long
conversations) and **stateless sessions** (new sessions cannot inherit history and always start from scratch).
@ -33,6 +38,8 @@ conversations) and **stateless sessions** (new sessions cannot inherit history a
ReMe gives agents **real memory** — old conversations are automatically compacted, important information is persistently
stored, and relevant context is automatically recalled in future interactions.
ReMe achieves state-of-the-art results on the LoCoMo and HaluMem benchmarks; see the [Experimental results](#experimental-results).
<details>
<summary><b>What you can do with ReMe</b></summary>
@ -73,6 +80,8 @@ working_dir/
├── MEMORY.md # Long-term memory: persistent info such as user preferences
├── memory/
│ └── YYYY-MM-DD.md # Daily journal: automatically written after each conversation
├── dialog/ # Raw conversation records: full dialog before compression
│ └── YYYY-MM-DD.jsonl # Daily conversation messages in JSONL format
└── tool_result/ # Cache for long tool outputs (auto-managed, expired entries auto-cleaned)
└── <uuid>.txt
```
@ -90,6 +99,8 @@ capabilities for AI agents:
<tr><td><code>pre_reasoning_hook</code></td><td>🔄 Pre-reasoning hook</td><td><code>compact_tool_result</code> + <code>check_context</code> + <code>compact_memory</code> + <code>summary_memory</code> (async)</td></tr>
<tr><td rowspan="2">Long-term Memory</td><td><code>summary_memory</code></td><td>📝 Persist important memory to files</td><td><a href="reme/memory/file_based/components/summarizer.py">Summarizer</a> — ReActAgent + file tools (<code>read</code> / <code>write</code> / <code>edit</code>)</td></tr>
<tr><td><code>memory_search</code></td><td>🔍 Semantic memory search</td><td><a href="reme/memory/file_based/tools/memory_search.py">MemorySearch</a> — hybrid retrieval with vectors + BM25</td></tr>
<tr><td rowspan="2">Session Memory</td><td><code>get_in_memory_memory</code></td><td>💾 Create in-session memory instance</td><td>Returns ReMeInMemoryMemory with dialog_path configured for persistence</td></tr>
<tr><td><code>await_summary_tasks</code></td><td>⏳ Wait for async summary tasks</td><td>Block until all background summary tasks complete</td></tr>
<tr><td>-</td><td><code>start</code></td><td>🚀 Start memory system</td><td>Initialize file storage, file watcher, and embedding cache; clean up expired tool result files</td></tr>
<tr><td>-</td><td><code>close</code></td><td>📕 Shutdown and cleanup</td><td>Clean up tool result files, stop file watcher, and persist embedding cache</td></tr>
</table>
@ -146,8 +157,12 @@ async def main():
messages = [...] # List of conversation messages
# 1. Compact long tool outputs (prevent tool results from blowing up context)
messages = await reme.compact_tool_result(messages)
# 1. Check context size (token counting, determine if compaction is needed)
messages_to_compact, messages_to_keep, is_valid = await reme.check_context(
messages=messages,
memory_compact_threshold=90000, # Threshold to trigger compaction (tokens)
memory_compact_reserve=10000, # Token count to reserve for recent messages
)
# 2. Compact conversation history into a structured summary
summary = await reme.compact_memory(
@ -158,10 +173,10 @@ async def main():
language="zh", # Summary language (e.g., "zh" / "")
)
# 3. Submit summary task asynchronously (non-blocking, writes to memory/YYYY-MM-DD.md)
reme.add_async_summary_task(messages=messages)
# 3. Compact long tool outputs (prevent tool results from blowing up context)
messages = await reme.compact_tool_result(messages)
# 4. Pre-reasoning hook (auto compact tool results + generate summaries)
# 4. Pre-reasoning hook (auto compact tool results + check context + generate summaries)
processed_messages, compressed_summary = await reme.pre_reasoning_hook(
messages=messages,
system_prompt="You are a helpful AI assistant.",
@ -173,12 +188,17 @@ async def main():
tool_result_compact_keep_n=3,
)
# 5. Semantic memory search (vector + BM25 hybrid retrieval)
# 5. Persist important memory to files (writes to memory/YYYY-MM-DD.md)
summary_result = await reme.summary_memory(
messages=messages,
language="zh",
)
# 6. Semantic memory search (vector + BM25 hybrid retrieval)
result = await reme.memory_search(query="Python version preference", max_results=5)
# 6. Create in-session memory instance (manages context for one conversation)
from reme.memory.file_based.reme_in_memory_memory import ReMeInMemoryMemory
memory = ReMeInMemoryMemory()
# 7. Create in-session memory instance (manages context for one conversation)
memory = reme.get_in_memory_memory() # Auto-configures dialog_path
for msg in messages:
await memory.add(msg)
token_stats = await memory.estimate_tokens(max_input_length=128000)
@ -186,8 +206,8 @@ async def main():
print(f"Message token count: {token_stats['messages_tokens']}")
print(f"Estimated total tokens: {token_stats['estimated_tokens']}")
# 7. Wait for background summary tasks to complete before shutdown
summary_result = await reme.await_summary_tasks()
# 8. Mark messages as compressed (auto-persists to dialog/YYYY-MM-DD.jsonl)
# await memory.mark_messages_compressed(messages_to_compact)
# Shutdown ReMeLight
await reme.close()
@ -203,9 +223,22 @@ if __name__ == "__main__":
### Architecture of the file-based ReMeLight memory system
[CoPaw MemoryManager](https://github.com/agentscope-ai/CoPaw/blob/main/src/copaw/agents/memory/memory_manager.py)
inherits
`ReMeLight` and integrates its memory capabilities into the agent reasoning loop:
#### Context data structure
```mermaid
flowchart TD
A[Context] --> B[compact_summary]
B --> C[dialog path guide + Goal/Constraints/Progress/KeyDecisions/NextSteps]
A --> E[messages: full dialogue history]
A --> F[File System Cache]
F --> G[dialog/YYYY-MM-DD.jsonl]
F --> H[tool_result/uuid.txt N-day TTL]
```
---
[CoPaw MemoryManager](https://github.com/agentscope-ai/CoPaw/blob/main/src/copaw/agents/memory/reme_light_memory_manager.py)
inherits `ReMeLight` and integrates its memory capabilities into the agent reasoning loop:
```mermaid
graph LR
@ -215,8 +248,11 @@ graph LR
CC -->|Exceeds limit| CM[compact_memory<br>Generate summary]
CC -->|Exceeds limit| SM[summary_memory<br>Async persistence]
SM -->|ReAct + FileIO| Files[memory/*.md]
CC -->|Exceeds limit| MMC[mark_messages_compressed<br>Persist raw dialog]
MMC --> Dialog[dialog/*.jsonl]
Agent -->|Explicit call| Search[memory_search<br>Vector+BM25]
Agent -->|In - session| InMem[ReMeInMemoryMemory<br>Token-aware memory]
InMem -->|Compress/Clear| Dialog
Files -.->|FileWatcher| Store[(FileStore<br>Vector+FTS index)]
Search --> Store
```
@ -260,16 +296,18 @@ graph LR
**Summary structure** (context checkpoints):
| Field | Description |
|-----------------------|------------------------------------------------------------------------|
| `## Goal` | User goals |
| `## Constraints` | Constraints and preferences |
| `## Progress` | Task progress |
| `## Key Decisions` | Key decisions |
| `## Next Steps` | Next step plans |
| `## Critical Context` | Critical data such as file paths, function names, error messages, etc. |
| Field | Description |
|-----------------------|-----------------------------------------------------------------------------------------|
| `## Goal` | User goals |
| `## Constraints` | Constraints and preferences |
| `## Progress` | Task progress |
| `## Key Decisions` | Key decisions |
| `## Next Steps` | Next step plans |
| `## Critical Context` | Critical data such as file paths, function names, error messages, etc. |
- **Incremental updates**: when `previous_summary` is provided, new conversations are merged into the existing summary.
- **Thinking enhancement**: with `add_thinking_block=True` (default), a reasoning step is added before generating the
summary to improve quality.
---
@ -302,18 +340,25 @@ graph LR
#### 4. `compact_tool_result` — tool result compaction
[ToolResultCompactor](reme/memory/file_based/components/tool_result_compactor.py) addresses the problem of long tool
outputs bloating the context.
outputs bloating the context. It applies two different truncation strategies depending on whether a message falls within
the `recent_n` window:
```mermaid
graph LR
M[messages] --> L{Iterate tool_result<br>len > threshold?}
L -->|No| K[Keep as-is]
L -->|Yes| T[truncate_text<br>Truncate to threshold]
T --> S[Write full content<br>tool_result/uuid.txt]
S --> R[Append file path reference<br>to message]
R --> C[cleanup_expired_files<br>Delete expired files]
M[messages] --> B{Within recent_n?}
B -->|Yes - recent| C[Low truncation recent_max_bytes=100KB<br>Save full content to tool_result/uuid.txt<br>Hint: 'Read from line N']
B -->|No - old| D[High truncation old_max_bytes=3KB<br>Reference existing file<br>More aggressive truncation]
C --> E[cleanup_expired_files<br>Delete expired files]
D --> E
```
| Parameter | Default | Description |
|--------------------|-----------------------|-------------------------------------------------------------------------------------------------------------------------------|
| `recent_n` | `1` | Minimum number of trailing consecutive tool-result messages treated as "recent" (use low truncation) |
| `recent_max_bytes` | `100 * 1024` (100 KB) | Truncation threshold for recent messages; content beyond this is saved to `tool_result/` with a file path and start-line hint |
| `old_max_bytes` | `3000` (3 KB) | Truncation threshold for older messages; truncation is more aggressive |
| `retention_days` | `3` | Number of days to retain tool result files; expired files are auto-cleaned |
- **Auto cleanup**: expired files (older than `retention_days`) are deleted automatically during `start` / `close` /
`compact_tool_result`.
@ -341,7 +386,7 @@ graph LR
#### 6. `ReMeInMemoryMemory` — in-session memory
[ReMeInMemoryMemory](reme/memory/file_based/reme_in_memory_memory.py) extends AgentScope's `InMemoryMemory` to provide
token-aware memory management.
token-aware memory management and raw conversation persistence.
```mermaid
graph LR
@ -351,13 +396,20 @@ graph LR
P -->|Yes| S[Prepend previous summary]
S --> O[Output messages]
P -->|No| O
M[mark_messages_compressed] --> D[Persist to dialog/YYYY-MM-DD.jsonl]
D --> R[Remove from memory]
```
| Function | Description |
|----------------------------------|---------------------------------------------------|
| `get_memory` | Filter messages by mark and auto-append summary |
| `estimate_tokens` | Estimate token usage of the context |
| `state_dict` / `load_state_dict` | Serialize/deserialize state (session persistence) |
| Function | Description |
|----------------------------------|----------------------------------------------------------|
| `get_memory` | Filter messages by mark and auto-append summary |
| `estimate_tokens` | Estimate token usage of the context |
| `state_dict` / `load_state_dict` | Serialize/deserialize state (session persistence) |
| `mark_messages_compressed` | Mark messages compressed and persist to dialog directory |
| `clear_content` | Persist all messages before clearing memory |
**Raw conversation persistence**: When messages are compressed or cleared, they are automatically saved to
`{dialog_path}/{date}.jsonl` with one JSON-formatted message per line.
---
@ -381,10 +433,18 @@ graph LR
**Execution flow**:
1. `compact_tool_result` — compact long tool outputs.
2. `check_context` — check whether the context exceeds limits.
3. `compact_memory` — generate compact summary (sync).
4. `summary_memory` — persist memory (async in the background).
1. `compact_tool_result` — compact long tool outputs for all messages except the most recent
`tool_result_compact_keep_n`.
2. `check_context` — check whether the context exceeds limits (remaining space = threshold minus tokens used by system
prompt and compressed summary).
3. `compact_memory` — generate compact summary (sync), appended into `compact_summary`.
4. `summary_memory` — persist memory to `memory/*.md` (async in the background, non-blocking).
| Key parameter | Default | Description |
|------------------------------|---------|-------------------------------------------------------------------------------------|
| `tool_result_compact_keep_n` | `3` | Skip tool result compaction for the most recent N messages (preserve full content) |
| `memory_compact_reserve` | `10000` | Token count to reserve for recent messages; messages beyond this trigger compaction |
| `compact_ratio` | `0.7` | Compaction threshold ratio: `max_input_length × compact_ratio × 0.95` |
---
@ -416,7 +476,6 @@ memories:
Installation and environment configuration are the same as [ReMeLight](#installation).
API keys are configured via environment variables and can be stored in a `.env` file at the project root.
### Python usage
```python
@ -532,7 +591,7 @@ graph LR
### Experimental results
Evaluations are conducted on Two benchmarks: **LoCoMo** and **HaluMem**. Experimental settings:
Evaluations are conducted on two benchmarks: **LoCoMo** and **HaluMem**. Experimental settings:
1. **ReMe backbone**: as specified in each table.
2. **Evaluation protocol**: LLM-as-a-Judge following MemOS — each answer is scored by GPT-4o-mini.
@ -541,29 +600,28 @@ Baseline results are reproduced from their respective papers under aligned setti
### LoCoMo
| Method | Single Hop | Multi Hop | Temporal | Open Domain | Overall |
|--------|------------|-----------|-----------|-------------|-----------|
| Method | Single Hop | Multi Hop | Temporal | Open Domain | Overall |
|----------|------------|-----------|-----------|-------------|-----------|
| MemoryOS | 62.43 | 56.50 | 37.18 | 40.28 | 54.70 |
| Mem0 | 66.71 | 58.16 | 55.45 | 40.62 | 61.00 |
| MemU | 72.77 | 62.41 | 33.96 | 46.88 | 61.15 |
| MemOS | 81.45 | 69.15 | 72.27 | 60.42 | 75.87 |
| HiMem | 89.22 | 70.92 | 74.77 | 54.86 | 80.71 |
| Zep | 88.11 | 71.99 | 74.45 | 66.67 | 81.06 |
| TiMem | 81.43 | 62.20 | 77.63 | 52.08 | 75.30 |
| TSM | 84.30 | 66.67 | 71.03 | 58.33 | 76.69 |
| MemR3 | 89.44 | 71.39 | 76.22 | 61.11 | 81.55 |
| Mem0 | 66.71 | 58.16 | 55.45 | 40.62 | 61.00 |
| MemU | 72.77 | 62.41 | 33.96 | 46.88 | 61.15 |
| MemOS | 81.45 | 69.15 | 72.27 | 60.42 | 75.87 |
| HiMem | 89.22 | 70.92 | 74.77 | 54.86 | 80.71 |
| Zep | 88.11 | 71.99 | 74.45 | 66.67 | 81.06 |
| TiMem | 81.43 | 62.20 | 77.63 | 52.08 | 75.30 |
| TSM | 84.30 | 66.67 | 71.03 | 58.33 | 76.69 |
| MemR3 | 89.44 | 71.39 | 76.22 | 61.11 | 81.55 |
| **ReMe** | **89.89** | **82.98** | **83.80** | **71.88** | **86.23** |
### HaluMem
| Method | Memory Integrity | Memory Accuracy | QA Accuracy |
|-------------|------------------|---------------|-------------|
| MemoBase | 14.55 | 92.24 | 35.53 |
| Supermemory | 41.53 | 90.32 | 54.07 |
| Mem0 | 42.91 | 86.26 | 53.02 |
| ProMem | **73.80** | 89.47 | 62.26 |
| **ReMe** | 67.72 | **94.06** | **88.78** |
|-------------|------------------|-----------------|-------------|
| MemoBase | 14.55 | 92.24 | 35.53 |
| Supermemory | 41.53 | 90.32 | 54.07 |
| Mem0 | 42.91 | 86.26 | 53.02 |
| ProMem | **73.80** | 89.47 | 62.26 |
| **ReMe** | 67.72 | **94.06** | **88.78** |
---

View file

@ -17,6 +17,10 @@
<a href="https://deepwiki.com/agentscope-ai/ReMe"><img src="https://img.shields.io/badge/DeepWiki-Ask_Devin-navy.svg" alt="DeepWiki"></a>
</p>
<p align="center">
<a href="https://trendshift.io/repositories/20528" target="_blank"><img src="https://trendshift.io/api/badge/repositories/20528" alt="agentscope-ai%2FReMe | Trendshift" style="width: 250px; height: 55px;" width="250" height="55"/></a>
</p>
<p align="center">
<strong>面向智能体的记忆管理工具包Remember Me, Refine Me.</strong><br>
</p>
@ -25,13 +29,15 @@
---
🧠 ReMe 是一个专为 **AI 智能体** 打造的记忆管理框架,同时提供基于[文件系统](#-基于文件的记忆系统-remelight)和基于[向量库](#-基于向量库的记忆系统)的记忆系统。
🧠 ReMe 是一个专为 **AI 智能体** 打造的记忆管理框架,同时提供基于[文件系统](#-基于文件的记忆系统-remelight)
和基于[向量库](#-基于向量库的记忆系统)的记忆系统。
它解决智能体记忆的两类核心问题:**上下文窗口有限**(长对话时早期信息被截断或丢失)、**会话无状态**(新对话无法继承历史,每次从零开始)。
ReMe 让智能体拥有**真正的记忆力**——旧对话自动浓缩,重要信息持久保存,下次对话自动想起来。
在 LoCoMo 与 HaluMem 基准测试中ReMe 取得了领先结果,详见[实验效果](#实验效果)。
<details>
<summary><b>你可以用 ReMe 做什么</b></summary>
@ -67,6 +73,8 @@ working_dir/
├── MEMORY.md # 长期记忆:用户偏好等持久信息
├── memory/
│ └── YYYY-MM-DD.md # 每日日记:对话结束后自动写入
├── dialog/ # 原始对话记录:压缩前的完整对话
│ └── YYYY-MM-DD.jsonl # 按日期存储的对话消息JSONL 格式)
└── tool_result/ # 超长工具输出缓存(自动管理,超期自动清理)
└── <uuid>.txt
```
@ -83,6 +91,8 @@ working_dir/
<tr><td><code>pre_reasoning_hook</code></td><td>🔄 推理前预处理钩子</td><td>compact_tool_result + check_context + compact_memory + summary_memory(async)</td></tr>
<tr><td rowspan="2">长期记忆</td><td><code>summary_memory</code></td><td>📝 将重要记忆写入文件</td><td><a href="reme/memory/file_based/components/summarizer.py">Summarizer</a> — ReActAgent + 文件工具read / write / edit</td></tr>
<tr><td><code>memory_search</code></td><td>🔍 语义搜索记忆</td><td><a href="reme/memory/file_based/tools/memory_search.py">MemorySearch</a> — 向量 + BM25 混合检索</td></tr>
<tr><td rowspan="2">会话内存</td><td><code>get_in_memory_memory</code></td><td>💾 创建会话内存实例</td><td>返回 ReMeInMemoryMemory自动配置 dialog_path 实现对话持久化</td></tr>
<tr><td><code>await_summary_tasks</code></td><td>⏳ 等待异步摘要任务</td><td>阻塞等待所有后台摘要任务完成</td></tr>
<tr><td>-</td><td><code>start</code></td><td>🚀 启动记忆系统</td><td>初始化文件存储、文件监控、Embedding 缓存;清理过期工具结果文件</td></tr>
<tr><td>-</td><td><code>close</code></td><td>📕 关闭并清理</td><td>清理工具结果文件、停止文件监控、保存 Embedding 缓存</td></tr>
</table>
@ -139,8 +149,12 @@ async def main():
messages = [...] # 对话消息列表
# 1. 压缩超长工具输出(防止工具结果撑爆上下文)
messages = await reme.compact_tool_result(messages)
# 1. 检查上下文大小Token 计数,判断是否需要压缩)
messages_to_compact, messages_to_keep, is_valid = await reme.check_context(
messages=messages,
memory_compact_threshold=90000, # 触发压缩的阈值tokens
memory_compact_reserve=10000, # 保留的近期消息 token 数
)
# 2. 将历史对话压缩为结构化摘要(可传入上轮摘要,实现增量更新)
summary = await reme.compact_memory(
@ -151,10 +165,10 @@ async def main():
language="zh", # 摘要语言zh / ""
)
# 3. 后台异步提交摘要任务(不阻塞对话,摘要写入 memory/YYYY-MM-DD.md
reme.add_async_summary_task(messages=messages)
# 3. 压缩超长工具输出(防止工具结果撑爆上下文
messages = await reme.compact_tool_result(messages)
# 4. 推理前预处理钩子(自动压缩工具结果 + 生成摘要)
# 4. 推理前预处理钩子(自动压缩工具结果 + 检查上下文 + 生成摘要)
processed_messages, compressed_summary = await reme.pre_reasoning_hook(
messages=messages,
system_prompt="你是一个有帮助的 AI 助手。",
@ -166,12 +180,18 @@ async def main():
tool_result_compact_keep_n=3,
)
# 5. 语义搜索记忆(向量 + BM25 混合检索)
# 5. 将重要记忆写入文件(摘要写入 memory/YYYY-MM-DD.md
summary_result = await reme.summary_memory(
messages=messages,
language="zh",
)
# 6. 语义搜索记忆(向量 + BM25 混合检索)
result = await reme.memory_search(query="Python 版本偏好", max_results=5)
# 6. 创建会话内存实例(管理单次对话的上下文)
# 7. 创建会话内存实例(管理单次对话的上下文)
from reme.memory.file_based.reme_in_memory_memory import ReMeInMemoryMemory
memory = ReMeInMemoryMemory()
memory = reme.get_in_memory_memory() # 自动配置 dialog_path
for msg in messages:
await memory.add(msg)
token_stats = await memory.estimate_tokens(max_input_length=128000)
@ -179,8 +199,8 @@ async def main():
print(f"消息 Token 数: {token_stats['messages_tokens']}")
print(f"预估总 Token 数: {token_stats['estimated_tokens']}")
# 7. 关闭前等待后台任务完成
summary_result = await reme.await_summary_tasks()
# 8. 标记消息为压缩状态(自动持久化到 dialog/YYYY-MM-DD.jsonl
# await memory.mark_messages_compressed(messages_to_compact)
# 关闭 ReMeLight
await reme.close()
@ -195,7 +215,21 @@ if __name__ == "__main__":
### 基于文件的 ReMeLight 记忆系统架构
[CoPaw MemoryManager](https://github.com/agentscope-ai/CoPaw/blob/main/src/copaw/agents/memory/memory_manager.py) 继承
#### 上下文数据结构
```mermaid
flowchart TD
A[Context] --> B[compact_summary]
B --> C[dialog 路径引导 + Goal/Constraints/Progress/KeyDecisions/NextSteps]
A --> E[messages: 完整对话历史]
A --> F[文件系统缓存]
F --> G[dialog/YYYY-MM-DD.jsonl]
F --> H[tool_result/uuid.txt N天TTL]
```
---
[CoPaw MemoryManager](https://github.com/agentscope-ai/CoPaw/blob/main/src/copaw/agents/memory/reme_light_memory_manager.py) 继承
`ReMeLight`,将记忆能力集成到 Agent 推理流程中:
```mermaid
@ -206,8 +240,11 @@ graph LR
CC -->|超限| CM[compact_memory<br>生成摘要]
CC -->|超限| SM[summary_memory<br>异步持久化]
SM -->|ReAct + FileIO| Files[memory/*.md]
CC -->|超限| MMC[mark_messages_compressed<br>持久化原始对话]
MMC --> Dialog[dialog/*.jsonl]
Agent -->|主动调用| Search[memory_search<br>向量+BM25]
Agent -->|会话内存| InMem[ReMeInMemoryMemory<br>Token感知内存]
InMem -->|压缩/清空| Dialog
Files -.->|FileWatcher| Store[(FileStore<br>向量+FTS索引)]
Search --> Store
```
@ -258,6 +295,7 @@ graph LR
| `## Critical Context` | 文件路径、函数名、错误信息等关键数据 |
- **增量更新**:传入 `previous_summary` 时,自动将新对话与旧摘要合并
- **思考增强**`add_thinking_block=True`(默认)时,在生成摘要前加入思考步骤,提升摘要质量
---
@ -288,18 +326,24 @@ graph LR
#### 4. compact_tool_result — 工具结果压缩
[ToolResultCompactor](reme/memory/file_based/components/tool_result_compactor.py) 解决工具输出过长导致上下文膨胀的问题。
[ToolResultCompactor](reme/memory/file_based/components/tool_result_compactor.py) 解决工具输出过长导致上下文膨胀的问题。根据消息是否在 `recent_n` 范围内,采用不同的截断策略:
```mermaid
graph LR
M[messages] --> L{遍历 tool_result<br>len > threshold?}
L -->|否| K[保留原样]
L -->|是| T[truncate_text<br>截断到 threshold]
T --> S[完整内容写入<br>tool_result/uuid.txt]
S --> R[消息追加文件路径引用]
R --> C[cleanup_expired_files<br>清理过期文件]
M[messages] --> B{属于 recent_n 范围?}
B -->|是 近期消息| C[低截断 recent_max_bytes=100KB<br>完整内容写入 tool_result/uuid.txt<br>消息追加: 从第N行开始读]
B -->|否 历史消息| D[高截断 old_max_bytes=3KB<br>引用已有文件路径<br>更激进截断]
C --> E[cleanup_expired_files<br>清理过期文件]
D --> E
```
| 参数 | 默认值 | 说明 |
|--------------------|---------------------|----------------------------------------------|
| `recent_n` | `1` | 末尾连续工具结果消息的最小数量,视为"近期",使用低截断阈值 |
| `recent_max_bytes` | `100 * 1024`100KB | 近期消息的截断阈值;超出部分转存到 `tool_result/` 并附注文件路径和起始行 |
| `old_max_bytes` | `3000`3KB | 历史消息的截断阈值,截断更激进 |
| `retention_days` | `3` | 工具结果文件的保留天数,过期自动清理 |
- **自动清理**:过期文件(超过 `retention_days`)在 `start`/`close`/`compact_tool_result` 时自动删除
---
@ -326,7 +370,7 @@ graph LR
#### 6. ReMeInMemoryMemory — 会话内存
[ReMeInMemoryMemory](reme/memory/file_based/reme_in_memory_memory.py) 扩展 AgentScope 的 `InMemoryMemory`,提供 Token
感知的内存管理。
感知的内存管理和原始对话持久化能力
```mermaid
graph LR
@ -336,13 +380,19 @@ graph LR
P -->|是| S[头部插入 previous-summary]
S --> O[输出 messages]
P -->|否| O
M[mark_messages_compressed] --> D[持久化到 dialog/YYYY-MM-DD.jsonl]
D --> R[从内存移除]
```
| 功能 | 说明 |
|----------------------------------|-------------------|
| `get_memory` | 按标记过滤,自动追加压缩摘要 |
| `estimate_tokens` | 估算上下文 Token 用量 |
| `state_dict` / `load_state_dict` | 状态序列化/反序列化(会话持久化) |
| 功能 | 说明 |
|----------------------------------|-----------------------|
| `get_memory` | 按标记过滤,自动追加压缩摘要 |
| `estimate_tokens` | 估算上下文 Token 用量 |
| `state_dict` / `load_state_dict` | 状态序列化/反序列化(会话持久化) |
| `mark_messages_compressed` | 标记消息压缩并持久化到 dialog 目录 |
| `clear_content` | 持久化所有消息后清空内存 |
**原始对话持久化**:当消息被压缩或清空时,自动保存到 `{dialog_path}/{date}.jsonl`,每行一条 JSON 格式的消息记录。
---
@ -365,10 +415,16 @@ graph LR
**执行流程**
1. `compact_tool_result` — 压缩超长工具输出
2. `check_context` — 检查上下文是否超限
3. `compact_memory` — 生成压缩摘要(同步)
4. `summary_memory` — 持久化记忆(异步后台)
1. `compact_tool_result` — 对除最近 `tool_result_compact_keep_n` 条消息之外的历史消息压缩超长工具输出
2. `check_context` — 检查上下文是否超限(扣除 system_prompt 和 compressed_summary 的 token 后计算剩余空间)
3. `compact_memory` — 生成压缩摘要(同步),结果追加到 `compact_summary`
4. `summary_memory` — 持久化记忆到 `memory/*.md`(异步后台,不阻塞推理)
| 关键参数 | 默认值 | 说明 |
|------------------------------|---------|--------------------------------------------------|
| `tool_result_compact_keep_n` | `3` | 最近 N 条消息跳过工具结果压缩(保留完整内容) |
| `memory_compact_reserve` | `10000` | 保留近期消息的 token 数,超出部分触发压缩 |
| `compact_ratio` | `0.7` | 压缩阈值比例:`max_input_length × compact_ratio × 0.95` |
---
@ -514,39 +570,38 @@ graph LR
### 实验效果
本实验部分在 LoCoMoHaluMem 两个数据集上进行评测,实验设置如下:
本实验部分在 LoCoMoHaluMem 两个数据集上进行评测,实验设置如下:
1. **ReMe 使用模型**:如各表 backbone 列所示。
2. **评估使用模型**:采用 LLM-as-a-Judge 协议(参照 MemOS——每条回答由 GPT-4o-mini 裁判模型打分。
实验设置尽量与各基线论文保持一致,以复用其公开结果。
### LoCoMo
#### LoCoMo
| Method | Single Hop | Multi Hop | Temporal | Open Domain | Overall |
|--------|------------|-----------|-----------|-------------|-----------|
| Method | Single Hop | Multi Hop | Temporal | Open Domain | Overall |
|----------|------------|-----------|-----------|-------------|-----------|
| MemoryOS | 62.43 | 56.50 | 37.18 | 40.28 | 54.70 |
| Mem0 | 66.71 | 58.16 | 55.45 | 40.62 | 61.00 |
| MemU | 72.77 | 62.41 | 33.96 | 46.88 | 61.15 |
| MemOS | 81.45 | 69.15 | 72.27 | 60.42 | 75.87 |
| HiMem | 89.22 | 70.92 | 74.77 | 54.86 | 80.71 |
| Zep | 88.11 | 71.99 | 74.45 | 66.67 | 81.06 |
| TiMem | 81.43 | 62.20 | 77.63 | 52.08 | 75.30 |
| TSM | 84.30 | 66.67 | 71.03 | 58.33 | 76.69 |
| MemR3 | 89.44 | 71.39 | 76.22 | 61.11 | 81.55 |
| Mem0 | 66.71 | 58.16 | 55.45 | 40.62 | 61.00 |
| MemU | 72.77 | 62.41 | 33.96 | 46.88 | 61.15 |
| MemOS | 81.45 | 69.15 | 72.27 | 60.42 | 75.87 |
| HiMem | 89.22 | 70.92 | 74.77 | 54.86 | 80.71 |
| Zep | 88.11 | 71.99 | 74.45 | 66.67 | 81.06 |
| TiMem | 81.43 | 62.20 | 77.63 | 52.08 | 75.30 |
| TSM | 84.30 | 66.67 | 71.03 | 58.33 | 76.69 |
| MemR3 | 89.44 | 71.39 | 76.22 | 61.11 | 81.55 |
| **ReMe** | **89.89** | **82.98** | **83.80** | **71.88** | **86.23** |
#### HaluMem
### HaluMem
| Method | Memory Integrity | Memory Accuracy | QA Accuracy |
|-------------|------------------|---------------|-------------|
| MemoBase | 14.55 | 92.24 | 35.53 |
| Supermemory | 41.53 | 90.32 | 54.07 |
| Mem0 | 42.91 | 86.26 | 53.02 |
| ProMem | **73.80** | 89.47 | 62.26 |
| **ReMe** | 67.72 | **94.06** | **88.78** |
|-------------|------------------|-----------------|-------------|
| MemoBase | 14.55 | 92.24 | 35.53 |
| Supermemory | 41.53 | 90.32 | 54.07 |
| Mem0 | 42.91 | 86.26 | 53.02 |
| ProMem | **73.80** | 89.47 | 62.26 |
| **ReMe** | 67.72 | **94.06** | **88.78** |
---
## 🧪 程序化记忆论文

View file

@ -176,7 +176,6 @@ Controls how context space is allocated and how memory is searched:
| `watch_paths` | Directories/files to monitor |
| `suffix_filters` | Which file suffixes to watch (`.md`) |
| `recursive` | Whether to recurse into subdirectories |
| `scan_on_start` | Whether to do a full scan on startup |
**token_counters — Token Counter**

View file

@ -172,7 +172,6 @@ pip install -e .
| `watch_paths` | 要监控的目录/文件 |
| `suffix_filters` | 只关心哪些后缀(`.md` |
| `recursive` | 是否递归子目录 |
| `scan_on_start` | 启动时先全量扫一遍 |
**token_counters — Token 计数器**

View file

@ -0,0 +1,122 @@
## Copaw Context Management V2
> 注:不涉及长期记忆
### 上下文数据结构
#### 1. 上下文-内存
- **compact_summary**(可选):
- **历史对话原始数据引导**:存储于 `dialog/YYYY-MM-DD.jsonl`,共 N 行,按时间顺序排列;回顾时建议从后往前读。
- **历史对话摘要**:包含 `Goal + Constraints + Progress + KeyDecisions + NextSteps`
- **messages**:当前对话上下文(完整消息列表)。
#### 2. 上下文-缓存到文件系统
- **历史对话原始数据**`dialog/YYYY-MM-DD.jsonl`
- **工具调用结果原始数据**`tool_result/{uuid}.txt`(保留 N 天)
```mermaid
flowchart TD
A[Context] --> B[compact_summary]
B --> C[dialog path guide + Goal/Constraints/Progress/KeyDecisions/NextSteps]
A --> E[messages: full dialogue history]
A --> F[File System Cache] --> G[dialog/YYYY-MM-DD.jsonl]
F --> H[tool_result/uuid.txt N-day TTL]
```
---
### 上下文机制Pre-Reasoning Hook
1. **工具结果 Offload** (`ToolCallResultCompact`)
2. **上下文检查** (`ContextChecker`)
3. **若 Token 超阈值**
- 保留最近 **X%** 的 Token保障连贯性
- 其余历史对话生成摘要 (`Compactor`)
4. **被摘要的上下文 Offload 到文件系统** (`SaveDialog`)
```mermaid
flowchart LR
A[Pre-Reasoning Hook] --> B[ToolCallResultCompact]
B --> C[ContextChecker]
C --> D{Token > Threshold?}
D -->|Yes| E[Keep recent X% tokens]
E --> F[Compact & Summary old context]
F --> G[SaveDialog: offload to file]
D -->|No| H[Proceed normally]
```
---
### 工具结果 Offload 机制
1. 所有工具调用结果先放入上下文,等待 Pre-Reasoning Hook 处理。
2. 根据是否属于 **recent_n** 范围,决定截断策略:
- **recent_n 内**:近期内容 → 低截断比例
- **recent_n 外**:远期内容 → 高截断比例
#### 示例Browser Use 类工具
| 阶段 | 行为 |
|----|------------------------------------------------------------------------------|
| 1 | 原始工具调用结果 |
| 2 | 保存原始内容到文件:– 若在 recent_n 内:截断较少– 附注“FullText saved to xxxx” 提示:“请从第 N 行开始读” |
| 3 | 若再次引用且超出 recent_n 二次截断(更激进)– 仍指向原文件路径 |
```mermaid
flowchart LR
A[Tool Call Result] --> B{Within recent_n?}
B -->|Yes| C[Low truncation<br>Save full text to tool_result/uuid.txt<br>Hint: 'Read from line N']
B -->|No| D[High truncation<br>Reference existing file<br>More aggressive truncation]
C --> E[Context includes snippet + file ref]
D --> E
```
---
### ReadFile 工具调用结果变化示例
| 阶段 | 行为 |
|----|---------------------------------------------|
| 1 | 原始工具调用结果 |
| 2 | 若在 recent_n 内:– 不截断– 不保存文件(因内容已由用户指定) |
| 3 | 若超出 recent_n 二次截断(更小)– 保存 FullText 到文件并引用 |
> 注ReadFile 本身读取的是外部文件,因此首次调用通常无需重复保存。
```mermaid
flowchart LR
A[ReadFile Result] --> B{Within recent_n?}
B -->|Yes| C[No truncation<br>No file save needed]
B -->|No| D[Apply secondary truncation<br>Save FullText to tool_result/uuid.txt]
C --> E[Include full content in context]
D --> F[Include snippet + file ref]
```
---
## Copaw Memory
### 触发逻辑
1. **主 Agent 主动写入**
- `Memory.md`(长期记忆主干)
- `YYYY-MM-DD.md`(当日日志)
2. **触发阈值时**,由 **SummarizerReact Agent** 写日志:
- 个性化信息(如偏好、习惯)
- Try-error 信息(失败尝试与修正)
3. **定时任务**(每日 00:00
- 汇总最近的 `YYYY-MM-DD.md` 文件
- 更新 `Memory.md`
```mermaid
flowchart TD
A[Main Agent] --> B[Write Memory.md]
A --> C[Write YYYY-MM-DD.md]
D[Context Threshold Reached?] -->|Yes| E[Summarizer Agent]
E --> F[Log: Personalization]
E --> G[Log: Try-Error Info]
H[Cron @ 00:00 daily] --> I[Aggregate recent YYYY-MM-DD.md]
I --> J[Update Memory.md]
```

View file

@ -33,7 +33,6 @@ classifiers = [
keywords = ["llm", "memory", "experience", "memoryscope", "ai", "mcp", "http", "reme", "personal"]
dependencies = [
"flowllm[reme]>=0.2.0.10",
"sqlite-vec>=0.1.6",
"prompt_toolkit>=3.0.52",
"rich>=14.2.0",
@ -44,7 +43,6 @@ dependencies = [
"fastapi>=0.121.3",
"fastmcp>=2.14.1",
"httpx>=0.28.1",
"litellm>=1.80.0",
"loguru>=0.7.3",
"mcp>=1.25.0",
"numpy>=2.2.6",
@ -80,8 +78,13 @@ full = [
"reme_ai[dev,ray,light]",
]
litellm = [
"litellm==1.80.0",
]
light = [
"agentscope==1.0.17",
"flowllm[reme]>=0.2.0.10",
]
[tool.setuptools.packages.find]

View file

@ -6,7 +6,7 @@ from . import extension
from . import memory
from .reme import ReMe
__version__ = "0.3.1.1"
__version__ = "0.3.1.6"
__all__ = [
"config",

View file

@ -47,5 +47,4 @@ file_watchers:
watch_paths: [ ".reme", ".reme/memory" ]
suffix_filters: [ ".md" ]
recursive: false
scan_on_start: true

View file

@ -3,6 +3,8 @@ as_llms:
backend: openai
model_name: qwen3.5-plus
thread_pool_max_workers: -1
as_llm_formatters:
default:
backend: openai
@ -34,4 +36,3 @@ file_watchers:
file_store: default
suffix_filters: [ ".md" ]
recursive: false
scan_on_start: true

View file

@ -156,13 +156,15 @@ class Application:
if not ray.is_initialized():
ray.init(num_cpus=self.service_config.ray_max_workers)
if (
if self.service_config.thread_pool_max_workers > 0 and (
self.service_context.thread_pool is None
or self.service_context.thread_pool._shutdown # pylint: disable=protected-access
):
self.service_context.thread_pool = ThreadPoolExecutor(
max_workers=self.service_config.thread_pool_max_workers,
)
elif self.service_config.thread_pool_max_workers <= 0:
logger.info("Thread pool is disabled (thread_pool_max_workers <= 0)")
if self.service_context.service_config.enable_logo:
print_logo(service_config=self.service_config)
@ -518,7 +520,7 @@ class Application:
def shutdown_thread_pool(self, wait: bool = True):
"""Shutdown the thread pool executor."""
if self.service_context.thread_pool:
if self.service_context.thread_pool is not None:
self.service_context.thread_pool.shutdown(wait=wait)
def shutdown_ray(self, wait: bool = True):

View file

@ -159,6 +159,7 @@ class BaseEmbeddingModel(ABC):
return
try:
load_start = time.time()
# Read all lines first (to load in reverse order)
with open(cache_file, "r", encoding="utf-8") as f:
lines = f.readlines()
@ -201,7 +202,9 @@ class BaseEmbeddingModel(ABC):
logger.warning(f"Failed to parse line in cache file: {e}")
continue
logger.info(f"Loaded {loaded_count} embeddings from cache file: {cache_file}")
logger.info(
f"Loaded {loaded_count} embeddings from cache file: {cache_file} in {time.time() - load_start:.2f}s",
)
except Exception as e:
logger.error(f"Failed to load cache from {cache_file}: {e}, deleting cache file")
try:

View file

@ -7,6 +7,9 @@ from pathlib import Path
from ..embedding import BaseEmbeddingModel
from ..enumeration import MemorySource
from ..schema import FileMetadata, MemoryChunk, MemorySearchResult
from ..utils import get_logger
logger = get_logger()
class BaseFileStore(ABC):
@ -54,24 +57,46 @@ class BaseFileStore(ABC):
"""Generate a zero vector based on embedding model dimensions."""
return [0.0] * self.embedding_dim
def _disable_vector_search(self, reason: str = "embedding API error") -> None:
"""Disable vector search and log a warning."""
if self.vector_enabled:
logger.warning(
f"[{self.store_name}] Disabling vector search due to {reason}. "
"Falling back to full-text search only.",
)
self.vector_enabled = False
async def get_embedding(self, query: str, **kwargs) -> list[float]:
"""Get embedding for a single query string."""
if not self.vector_enabled:
return self._get_mock_embedding()
return await self.embedding_model.get_embedding(query, **kwargs)
try:
return await self.embedding_model.get_embedding(query, **kwargs)
except Exception as e:
self._disable_vector_search(str(e))
return self._get_mock_embedding()
async def get_embeddings(self, queries: list[str], **kwargs) -> list[list[float]]:
"""Get embeddings for a batch of query strings."""
if not self.vector_enabled:
return [self._get_mock_embedding() for _ in queries]
return await self.embedding_model.get_embeddings(queries, **kwargs)
try:
return await self.embedding_model.get_embeddings(queries, **kwargs)
except Exception as e:
self._disable_vector_search(str(e))
return [self._get_mock_embedding() for _ in queries]
async def get_chunk_embedding(self, chunk: MemoryChunk, **kwargs) -> MemoryChunk:
"""Generate and populate embedding field for a single MemoryChunk object."""
if not self.vector_enabled:
chunk.embedding = self._get_mock_embedding()
return chunk
return await self.embedding_model.get_chunk_embedding(chunk, **kwargs)
try:
return await self.embedding_model.get_chunk_embedding(chunk, **kwargs)
except Exception as e:
self._disable_vector_search(str(e))
chunk.embedding = self._get_mock_embedding()
return chunk
async def get_chunk_embeddings(self, chunks: list[MemoryChunk], **kwargs) -> list[MemoryChunk]:
"""Generate and populate embedding fields for a batch of MemoryChunk objects."""
@ -80,7 +105,14 @@ class BaseFileStore(ABC):
for chunk in chunks:
chunk.embedding = mock_embedding.copy()
return chunks
return await self.embedding_model.get_chunk_embeddings(chunks, **kwargs)
try:
return await self.embedding_model.get_chunk_embeddings(chunks, **kwargs)
except Exception as e:
self._disable_vector_search(str(e))
mock_embedding = self._get_mock_embedding()
for chunk in chunks:
chunk.embedding = mock_embedding.copy()
return chunks
@abstractmethod
async def start(self):

View file

@ -29,13 +29,13 @@ class BaseFileWatcher:
watch_paths: list[str] | str,
suffix_filters: list[str] | None = None,
recursive: bool = False,
debounce: int = 500, # Millisecond debounce
debounce: int = 2000,
chunk_tokens: int = 400,
chunk_overlap: int = 80,
file_store: BaseFileStore | None = None,
callback: Callable[[set[tuple[Change, str]]], None | Coroutine[Any, Any, None]] | None = None,
scan_on_start: bool = True,
clear_on_start: bool = True,
rebuild_index_on_start: bool = True,
poll_delay_ms: int = 2000,
**kwargs,
):
"""
@ -50,9 +50,9 @@ class BaseFileWatcher:
chunk_overlap: Overlap size for chunks
file_store: File store instance
callback: Callback function for changes
scan_on_start: If True, scan existing files on start and trigger on_changes with Change.added
clear_on_start: If True, clear all indexed data on start before scanning.
Useful for full rebuild of the index.
rebuild_index_on_start: If True, clear all indexed data on start and rescan existing files.
If False, only monitor new changes without initialization.
poll_delay_ms: Polling delay in milliseconds. If > 300ms, force_polling will be enabled automatically.
**kwargs: Additional keyword arguments
"""
self.watch_paths: list[str] = [watch_paths] if isinstance(watch_paths, str) else watch_paths
@ -63,8 +63,8 @@ class BaseFileWatcher:
self.chunk_overlap: int = chunk_overlap
self.file_store: BaseFileStore = file_store
self.callback = callback
self.scan_on_start: bool = scan_on_start
self.clear_on_start: bool = clear_on_start
self.rebuild_index_on_start: bool = rebuild_index_on_start
self.poll_delay_ms: int = poll_delay_ms
self.kwargs: dict = kwargs
self._stop_event = asyncio.Event()
@ -78,16 +78,14 @@ class BaseFileWatcher:
self._running = True
# Clear all indexed data if requested
if self.clear_on_start and self.file_store is not None:
await self.file_store.clear_all()
logger.info("Cleared all indexed data on start")
async def _initialize_and_watch():
if self.rebuild_index_on_start:
await self.file_store.clear_all()
logger.info("Cleared all indexed data on start")
await self._scan_existing_files()
await self._watch_loop()
# Scan existing files if requested
if self.scan_on_start:
await self._scan_existing_files()
self._watch_task = asyncio.create_task(self._watch_loop())
self._watch_task = asyncio.create_task(_initialize_and_watch())
logger.info(f"Started watching: {self.watch_paths}")
async def close(self):
@ -188,6 +186,7 @@ class BaseFileWatcher:
watch_filter=self.watch_filter,
recursive=self.recursive,
debounce=self.debounce,
poll_delay_ms=self.poll_delay_ms,
stop_event=self._stop_event,
):
if self._stop_event.is_set():

View file

@ -301,7 +301,7 @@ class BaseOp(metaclass=ABCMeta):
def submit_sync_task(self, fn: Callable, *args, **kwargs) -> "BaseOp":
"""Submit a task to the thread pool or local queue."""
if self.enable_parallel:
if self.enable_parallel and self.service_context.thread_pool is not None:
task = self.service_context.thread_pool.submit(fn, *args, **kwargs)
else:
task = (fn, args, kwargs)

View file

@ -2,8 +2,10 @@
from pydantic import BaseModel, Field
_TRUNCATION_NOTICE_MARKER = "<<<TRUNCATED>>>"
_DEFAULT_MAX_BLOCK_TEXT_PREVIEW_LENGTH = 100
_DEFAULT_MAX_FORMATTER_TEXT_LENGTH = 2000
_DEFAULT_MAX_FORMATTER_TEXT_LENGTH = 1000
class AsBlockStat(BaseModel):
@ -27,7 +29,8 @@ class AsBlockStat(BaseModel):
return self.format(_DEFAULT_MAX_BLOCK_TEXT_PREVIEW_LENGTH)
def _truncate(self, text: str, max_length: int) -> str:
"""Simple truncation with ellipsis."""
"""Truncate text with ellipsis, replacing newlines with spaces."""
text = text.replace("\n", " ")
if len(text) <= max_length:
return text
return text[:max_length] + "..."
@ -46,22 +49,23 @@ class AsBlockStat(BaseModel):
if self.block_type == "text":
if not self.text:
return ""
return f"<text>{self._truncate(self.text, max_length)}</text>"
return f"[text]: {self._truncate(self.text, max_length)}"
if self.block_type == "thinking":
if not include_thinking or not self.text:
return ""
return f"<thinking>{self._truncate(self.text, max_length)}</thinking>"
return f"[think]: {self._truncate(self.text, max_length)}"
if self.block_type in ("image", "audio", "video"):
content = self.media_url if self.media_url else ""
return f"<{self.block_type}>{content}</{self.block_type}>"
return f"[{self.block_type}]: {content}"
if self.block_type == "tool_use":
content = f"{self.tool_name} params={self._truncate(self.tool_input, max_length)}"
return f"<tool_use>{content}</tool_use>"
return f"[tool_use]: {content}"
if self.block_type == "tool_result":
if not self.tool_output:
return ""
content = f"{self.tool_name} output={self._truncate(self.tool_output, max_length)}"
return f"<tool_result>{content}</tool_result>"
display_output = self.tool_output.split(_TRUNCATION_NOTICE_MARKER)[0]
content = f"{self.tool_name} output={self._truncate(display_output, max_length)}"
return f"[tool_result]: {content}"
return ""

View file

@ -116,7 +116,10 @@ class ServiceConfig(BasicConfig):
working_dir: str = Field(default=".reme")
enable_logo: bool = Field(default=True)
language: str = Field(default="")
thread_pool_max_workers: int = Field(default=16)
thread_pool_max_workers: int = Field(
default=16,
description="Number of thread pool workers. Set to -1 to disable thread pool.",
)
ray_max_workers: int = Field(default=-1)
log_to_console: bool = Field(default=True)
disabled_flows: list[str] = Field(default_factory=list)

View file

@ -19,7 +19,6 @@ from .pydantic_utils import create_pydantic_model
from .singleton import singleton
from .time import timer, get_now_time
from .hf_token_counter_utils import get_hf_token_counter
from .truncate_text_utils import truncate_text, truncate_text_head, is_truncated, TRUNCATION_MARKER_START
__all__ = [
"convert_dashscope_to_agentscope",
@ -51,8 +50,4 @@ __all__ = [
"timer",
"get_now_time",
"get_hf_token_counter",
"truncate_text",
"truncate_text_head",
"is_truncated",
"TRUNCATION_MARKER_START",
]

View file

@ -1,82 +0,0 @@
"""Utility functions for truncating long text strings."""
from .std_logger import get_logger
logger = get_logger()
TRUNCATION_MARKER_START = "<<<TRUNCATED>>>"
TRUNCATION_MARKER_END = "<<<END_TRUNCATED>>>"
def truncate_text(text: str, max_length: int) -> str:
"""Truncate text to max length, keeping head and tail portions.
Args:
text: The text to truncate
max_length: Maximum allowed length
Returns:
Truncated text with unique markers indicating truncation
"""
text = str(text) if text else ""
if not text:
return text
if len(text) <= max_length:
return text
half_length = max_length // 2
truncated_chars = len(text) - max_length
logger.debug(
"Text truncated: original %d chars, kept head %d + tail %d, removed %d chars.",
len(text),
half_length,
half_length,
truncated_chars,
)
return (
f"{text[:half_length]}\n\n{TRUNCATION_MARKER_START} "
f"({truncated_chars} characters omitted) "
f"{TRUNCATION_MARKER_END}\n\n{text[-half_length:]}"
)
def truncate_text_head(text: str, max_length: int) -> str:
"""Truncate text from the beginning, keeping only the head portion.
Args:
text: The text to truncate
max_length: Maximum allowed length
Returns:
Truncated text with marker indicating truncation at the end
"""
text = str(text) if text else ""
if not text:
return text
if len(text) <= max_length:
return text
truncated_chars = len(text) - max_length
logger.debug(
"Text truncated from head: original %d chars, kept %d, removed %d chars from tail.",
len(text),
max_length,
truncated_chars,
)
return f"{text[:max_length]}{TRUNCATION_MARKER_START}"
def is_truncated(text: str) -> bool:
"""Check if the text has been truncated (contains truncation marker).
Args:
text: The text to check
Returns:
bool: True if text contains truncation marker, False otherwise
"""
if not text:
return False
return TRUNCATION_MARKER_START in text

View file

@ -113,9 +113,9 @@ class CliAgent(BaseOp):
toolkit = Toolkit()
file_io = FileIO(working_dir=self.working_dir)
toolkit.register_tool_function(file_io.read)
toolkit.register_tool_function(file_io.write)
toolkit.register_tool_function(file_io.edit)
toolkit.register_tool_function(file_io.read_file)
toolkit.register_tool_function(file_io.write_file)
toolkit.register_tool_function(file_io.edit_file)
return toolkit

View file

@ -10,6 +10,22 @@ from ....core.utils import get_logger
logger = get_logger()
def _is_valid_summary(content: str) -> bool:
"""Check if the summary content is valid.
Args:
content: The summary content to validate.
Returns:
True if valid, False otherwise.
"""
if not content or not content.strip():
return False
if "##" not in content:
return False
return True
class Compactor(BaseOp):
"""Compactor class for compacting memory messages."""
@ -17,17 +33,24 @@ class Compactor(BaseOp):
self,
memory_compact_threshold: int,
console_enabled: bool = False,
return_dict: bool = False,
add_thinking_block: bool = True,
**kwargs,
):
super().__init__(**kwargs)
self.memory_compact_threshold: int = memory_compact_threshold
self.console_enabled: bool = console_enabled
self.return_dict: bool = return_dict
self.add_thinking_block: bool = add_thinking_block
# pylint: disable=too-many-return-statements
async def execute(self):
messages: list[Msg] = self.context.get("messages", [])
previous_summary: str = self.context.get("previous_summary", "")
if not messages:
if self.return_dict:
return {"user_message": "", "history_compact": "", "is_valid": False}
return ""
msg_handler = AsMsgHandler(self.as_token_counter)
@ -35,12 +58,15 @@ class Compactor(BaseOp):
history_formatted_str: str = await msg_handler.format_msgs_to_str(
messages=messages,
memory_compact_threshold=self.memory_compact_threshold,
include_thinking=self.add_thinking_block,
)
after_token_count = await msg_handler.count_str_token(history_formatted_str)
logger.info(f"Compactor before_token_count={before_token_count} after_token_count={after_token_count}")
if not history_formatted_str:
logger.warning(f"No history to compact. messages={messages}")
if self.return_dict:
return {"user_message": "", "history_compact": "", "is_valid": False}
return ""
agent = ReActAgent(
@ -52,18 +78,12 @@ class Compactor(BaseOp):
agent.set_console_output_enabled(self.console_enabled)
if previous_summary:
prefix: str = self.get_prompt("update_user_message_prefix")
suffix: str = self.get_prompt("update_user_message_suffix")
user_message: str = (
f"<conversation>\n{history_formatted_str}\n</conversation>\n\n"
f"{prefix}\n\n"
f"<previous-summary>\n{previous_summary}\n</previous-summary>\n\n"
f"{suffix}"
f"# conversation\n{history_formatted_str}\n\n"
f"# previous-summary\n{previous_summary}\n\n" + self.get_prompt("update_user_message")
)
else:
user_message: str = f"<conversation>\n{history_formatted_str}\n</conversation>\n\n" + self.get_prompt(
"initial_user_message",
)
user_message: str = f"# conversation\n{history_formatted_str}\n\n" + self.get_prompt("initial_user_message")
logger.info(f"Compactor sys_prompt={agent.sys_prompt} user_message={user_message}")
compact_msg: Msg = await agent.reply(
@ -75,5 +95,16 @@ class Compactor(BaseOp):
)
history_compact: str = compact_msg.get_text_content()
is_valid: bool = _is_valid_summary(history_compact)
if not is_valid:
logger.warning(f"Invalid summary result: {history_compact[:200]}...")
if self.return_dict:
return {"user_message": user_message, "history_compact": history_compact, "is_valid": False}
return ""
logger.info(f"Compactor Result:\n{history_compact}")
if self.return_dict:
return {"user_message": user_message, "history_compact": history_compact, "is_valid": True}
return history_compact

View file

@ -7,10 +7,14 @@ system_prompt_zh: |
这些摘要可以在未来会话中用于恢复上下文。专注于保留关键信息同时减少token数量。
initial_user_message: |
The messages above are a conversation to summarize. Create a structured context checkpoint summary
that another LLM will use to continue the work.
# Task
Create a structured summary from the conversation above.
Use this EXACT format:
# Rules:
- Keep each section concise
- Preserve exact file paths, function names, and error messages
# Output Format:
## Goal
[What is the user trying to accomplish? Can be multiple items if the session covers different tasks.]
@ -39,13 +43,17 @@ initial_user_message: |
- [Any data, examples, or references needed to continue]
- [Or "(none)" if not applicable]
Keep each section concise. Preserve exact file paths, function names, and error messages.
Output the structured summary following the format above.
initial_user_message_zh: |
上述消息是一场需要总结的对话。创建一个结构化的上下文检查点摘要,
以便另一个LLM可以用来继续工作
# 任务
根据上面的对话创建一个结构化摘要
使用此确切格式:
# 规则:
- 保持每个部分简洁
- 保留确切的文件路径、函数名称和错误消息
# 输出示例:
## 目标
[用户试图完成什么?如果会话涵盖不同任务,可以有多个项目。]
@ -74,14 +82,13 @@ initial_user_message_zh: |
- [任何继续工作所需的数据、示例或参考资料]
- [或者如果不适用则为"(none)"]
保持每个部分简洁。保留确切的文件路径、函数名称和错误消息
请按照上面示例的格式,输出结构化摘要
update_user_message_prefix: |
The messages above are NEW conversation messages to incorporate into the existing summary provided in
<previous-summary> tags.
update_user_message: |
# Task
Update the structured summary with new conversation messages.
update_user_message_suffix: |
Update the existing structured summary with new information. RULES:
# Rules:
- PRESERVE all existing information from the previous summary
- ADD new progress, decisions, and context from the new messages
- UPDATE the Progress section: move items from "In Progress" to "Done" when completed
@ -89,7 +96,7 @@ update_user_message_suffix: |
- PRESERVE exact file paths, function names, and error messages
- If something is no longer relevant, you may remove it
Use this EXACT format:
# Output Format:
## Goal
[Preserve existing goals, add new ones if the task expanded]
@ -116,13 +123,13 @@ update_user_message_suffix: |
## Critical Context
- [Preserve important context, add new if needed]
Keep each section concise. Preserve exact file paths, function names, and error messages.
Output the structured summary following the format above.
update_user_message_prefix_zh: |
以上消息是需要整合到现有摘要中的新对话内容,现有摘要位于<previous-summary>标签中。
update_user_message_zh: |
# 任务
使用新的对话内容来更新结构化摘要。
update_user_message_suffix_zh: |
用新信息更新现有的结构化摘要。规则:
# 规则:
- 保留来自先前摘要的所有现有信息
- 从新消息中添加新的进展、决策和上下文
- 更新进度部分:当完成时将项目从"进行中"移到"已完成"
@ -130,7 +137,7 @@ update_user_message_suffix_zh: |
- 保留确切的文件路径、函数名称和错误消息
- 如果某些内容不再相关,您可以删除它
使用此确切格式
# 输出示例
## 目标
[保留现有目标,如果任务扩展则添加新目标]
@ -157,4 +164,4 @@ update_user_message_suffix_zh: |
## 关键上下文
- [保留重要上下文,如需要则添加新的]
保持每个部分简洁。保留确切的文件路径、函数名称和错误消息
请按照上面示例的格式,输出结构化摘要

View file

@ -25,6 +25,7 @@ class Summarizer(BaseOp):
toolkit: Toolkit | None = None,
console_enabled: bool = False,
timezone: str | None = None,
add_thinking_block: bool = True,
**kwargs,
):
super().__init__(**kwargs)
@ -34,6 +35,16 @@ class Summarizer(BaseOp):
self.toolkit: Toolkit | None = toolkit
self.console_enabled: bool = console_enabled
self.timezone: str | None = timezone
self.add_thinking_block: bool = add_thinking_block
def _get_current_datetime(self) -> datetime.datetime:
"""Get current datetime with timezone, fallback to local time if timezone is invalid."""
if self.timezone:
try:
return datetime.datetime.now(zoneinfo.ZoneInfo(self.timezone))
except Exception as e:
logger.error(f"Invalid timezone: {self.timezone}, falling back to local time error={e}")
return datetime.datetime.now()
async def execute(self):
messages: list[Msg] = self.context.get("messages", [])
@ -46,6 +57,7 @@ class Summarizer(BaseOp):
history_formatted_str: str = await msg_handler.format_msgs_to_str(
messages=messages,
memory_compact_threshold=self.memory_compact_threshold,
include_thinking=self.add_thinking_block,
)
after_token_count = await msg_handler.count_str_token(history_formatted_str)
logger.info(f"Summarizer before_token_count={before_token_count} after_token_count={after_token_count}")
@ -63,15 +75,9 @@ class Summarizer(BaseOp):
)
agent.set_console_output_enabled(self.console_enabled)
user_message: str = f"<conversation>\n{history_formatted_str}\n</conversation>\n" + self.prompt_format(
user_message: str = f"# conversation\n{history_formatted_str}\n\n" + self.prompt_format(
"user_message",
date=(
datetime.datetime.now(
zoneinfo.ZoneInfo(self.timezone),
)
if self.timezone
else datetime.datetime.now()
).strftime("%Y-%m-%d"),
date=self._get_current_datetime().strftime("%Y-%m-%d"),
working_dir=self.working_dir,
memory_dir=self.memory_dir,
)

View file

@ -1,28 +1,29 @@
user_message: |
Memory Pre-compression Flush Cycle Initiated
Memory Pre-compression Flush Cycle.
The current session is about to enter the automatic compression phase. Please capture persistent memory AND session reflections, then write them to disk.
Current date: {date}
Working directory: {working_dir}
# Task
Immediately store persistent memory and reflections to: {memory_dir}/YYYY-MM-DD.md
Workflow:
1. First, `read` {memory_dir}/YYYY-MM-DD.md (if the file doesnt exist, an error message will be returned).
2. Extract and synthesize content from the current session:
# Workflow
1. Extract and synthesize content from the current session:
- Persistent Memory: Facts, user profile updates, project states, and important events.
- Experience Reflection: Reusable thinking logic derived from user feedback, successful problem-solving strategies, mistakes made/pitfalls to avoid, and actionable insights for future interactions.
3. Intelligently merge new information with existing content (skip merging if the file doesnt exist):
2. `read` {memory_dir}/YYYY-MM-DD.md (if the file doesnt exist, an error message will be returned)
- If the file doesnt exist, use `write` tool directly.
- If the file exists, intelligently merge new information with existing content, prefer using `edit` to update specific sections.
- Use `write` to overwrite the entire file only if substantial restructuring is required.
# Principles
- Intelligently merge new information with existing content:
- Categorize clearly (e.g., separate "Factual Memory" from "Reflections & Logic").
- Avoid duplicating already recorded information.
- Enrich existing entries with new details where relevant.
- Maintain chronological order wherever applicable.
4. Write the updated content:
- Prefer using `edit` to update specific sections when possible.
- Use `write` to overwrite the entire file only if substantial restructuring is required.
Principles:
- Always preserve timestamps and any date/time-related context.
- Add only genuinely new or meaningfully enriching information.
- Reflections MUST focus on forming reusable cognitive frameworks based on user feedback, aiming to improve future task execution.
@ -37,23 +38,23 @@ user_message_zh: |
当前日期:{date}
工作目录:{working_dir}
# 任务
立即存储持久化记忆与反思(使用路径 {memory_dir}/YYYY-MM-DD.md
工作流程:
1. 先 `read` {memory_dir}/YYYY-MM-DD.md如文件不存在会返回错误提示
2. 从当前会话中提取并综合两类内容:
# 工作流程
1. 从当前会话中提取并综合两类内容:
- 持久化记忆:客观事实、用户信息更新、项目状态及重要事件。
- 经验反思:基于用户反馈形成的可复用思考逻辑、成功的问题解决策略、犯下的错误/应避免的陷阱,以及对未来交互有帮助的行动指南。
3. 智能合并新信息与现有内容(若文件不存在则跳过合并):
2. `read` {memory_dir}/YYYY-MM-DD.md如文件不存在会返回错误提示
- 若文件不存在,直接使用 `write` 工具写入。
- 若文件已存在,智能合并新信息与现有内容,尽可能使用 `edit` 更新特定部分,仅在需要大幅重构时使用 `write` 覆盖整个文件。
# 原则
- 智能合并新信息与现有内容:
- 将内容进行清晰的分类(例如明确区分“事实记忆”与“反思与逻辑”)。
- 避免重复已记录的信息。
- 在相关时丰富现有条目的新细节。
- 在适用时保持时间顺序。
4. 写入更新后的内容:
- 尽可能使用 `edit` 更新特定部分。
- 如需大幅重构则使用 `write` 覆盖整个文件。
原则:
- 始终保留时间戳、日期和时间相关上下文。
- 仅添加真正新的或有丰富价值的信息。
- 反思内容必须侧重于根据用户反馈构建可复用的思维逻辑,以改善未来的任务执行。

View file

@ -1,33 +1,19 @@
"""Tool Result Compactor: truncate large tool results and save full content to files."""
import os
import sys
import uuid
from datetime import datetime, timedelta
from pathlib import Path
from agentscope.message import Msg
from ..utils import truncate_text_output, DEFAULT_MAX_BYTES, TRUNCATION_NOTICE_MARKER
from ....core.op import BaseOp
from ....core.utils import get_logger
from ....core.utils import truncate_text_head, TRUNCATION_MARKER_START
logger = get_logger()
MAX_LINE_LENGTH = 10000
def _split_long_lines(text: str, max_len: int = MAX_LINE_LENGTH) -> str:
"""Split lines that exceed max_len by inserting newlines."""
lines = text.split("\n")
result = []
for line in lines:
if len(line) <= max_len:
result.append(line)
else:
# Split line into chunks of max_len
for i in range(0, len(line), max_len):
result.append(line[i : i + max_len])
return "\n".join(result)
class ToolResultCompactor(BaseOp):
"""Truncate large tool_result outputs and save full content to files."""
@ -35,64 +21,59 @@ class ToolResultCompactor(BaseOp):
def __init__(
self,
tool_result_dir: str | Path,
retention_days: int = 7,
retention_days: int = 3,
old_max_bytes: int = 3000,
recent_max_bytes: int = DEFAULT_MAX_BYTES,
recent_n: int = 1,
old_threshold: int = 500,
recent_threshold: int = 30000,
encoding: str = "utf-8",
**kwargs,
):
super().__init__(**kwargs)
self.tool_result_dir = Path(tool_result_dir)
self.retention_days = retention_days
self.old_max_bytes = old_max_bytes
self.recent_max_bytes = recent_max_bytes
self.recent_n = recent_n
self.old_threshold = old_threshold
self.recent_threshold = recent_threshold
self.encoding = encoding
self.tool_result_dir.mkdir(parents=True, exist_ok=True)
def _save_and_truncate(self, content: str, tool_name: str, threshold: int) -> str:
"""Save full content to file and return truncated version with file reference."""
def _truncate(self, content: str, max_bytes: int) -> str:
if not content:
return content
# Check if content was previously truncated
if TRUNCATION_MARKER_START in content:
parts = content.split(TRUNCATION_MARKER_START, 1)
if len(parts[0]) <= threshold:
return content
return f"{truncate_text_head(parts[0], threshold)}{parts[1]}"
try:
if TRUNCATION_NOTICE_MARKER in content:
return truncate_text_output(content, max_bytes=max_bytes, encoding=self.encoding)
# Not truncated before
if len(content) <= threshold:
if len(content.encode(self.encoding)) <= max_bytes + 100:
return content
saved_path: str | None = None
fp = self.tool_result_dir / f"{uuid.uuid4().hex}.txt"
fp.write_text(content, encoding=self.encoding)
saved_path = str(fp)
return truncate_text_output(
content,
1,
content.count("\n") + 1,
max_bytes,
file_path=saved_path,
encoding=self.encoding,
)
except Exception as e:
logger.warning("Failed to truncate content, returning original: %s", e)
return content
# Save full content with long lines split
self.tool_result_dir.mkdir(parents=True, exist_ok=True)
file_path = self.tool_result_dir / f"{uuid.uuid4().hex}.txt"
created_at = datetime.now().isoformat()
def _compact(self, output: str | list[dict], max_bytes: int) -> str | list[dict]:
"""Truncate output to max_bytes, saving full content to file if needed."""
processed_content = _split_long_lines(content)
file_path.write_text(
f"# tool_name: {tool_name}\n# created_at: {created_at}\n# ---\n{processed_content}",
encoding="utf-8",
)
logger.debug("Saved tool result to %s (len=%d)", file_path, len(content))
# Return truncated with file reference
return f"{truncate_text_head(content, threshold)}\n\n[Full content saved to: {file_path}]"
def _process_output(self, output: str | list[dict], tool_name: str, threshold: int) -> str | list[dict]:
"""Process tool result output, truncating if necessary."""
if isinstance(output, str):
return self._save_and_truncate(output, tool_name, threshold)
return self._truncate(output, max_bytes)
if isinstance(output, list):
return [
(
{**b, "text": self._save_and_truncate(b.get("text", ""), tool_name, threshold)}
if isinstance(b, dict) and b.get("type") == "text"
else b
)
for b in output
]
for b in output:
if isinstance(b, dict) and b.get("type") == "text":
b["text"] = self._truncate(b.get("text", ""), max_bytes)
return output
async def execute(self) -> list[Msg]:
@ -101,43 +82,80 @@ class ToolResultCompactor(BaseOp):
if not messages:
return messages
# Split messages into old and recent parts
split_index = max(0, len(messages) - self.recent_n)
recent_n = 0
for msg in reversed(messages):
if not isinstance(msg.content, list) or not any(
isinstance(b, dict) and b.get("type") == "tool_result" for b in msg.content
):
break
recent_n += 1
split_index = max(0, len(messages) - max(recent_n, self.recent_n))
md_file_tool_ids = set()
try:
for msg in messages:
if not isinstance(msg.content, list):
continue
for block in msg.content:
if isinstance(block, dict) and block.get("type") == "tool_use":
tool_id = block.get("id", "")
if not tool_id:
continue
if (
block.get("name", "").lower() == "read_file"
and ".md" in (block.get("raw_input") or "").lower()
):
md_file_tool_ids.add(tool_id)
except Exception as e:
logger.warning("Failed to detect md file tool ids: %s", e)
logger.info(f"md_file_tool_ids: {md_file_tool_ids}")
for idx, msg in enumerate(messages):
if not isinstance(msg.content, list):
continue
# Determine threshold based on message position
threshold = self.recent_threshold if idx >= split_index else self.old_threshold
is_recent = idx >= split_index
max_bytes = self.recent_max_bytes if is_recent else self.old_max_bytes
for block in msg.content:
if isinstance(block, dict) and block.get("type") == "tool_result":
output = block.get("output")
if output:
block["output"] = self._process_output(output, block.get("name", "unknown"), threshold)
if isinstance(block, dict) and block.get("type") == "tool_result" and block.get("output"):
tool_use_id = block.get("id", "")
if tool_use_id in md_file_tool_ids:
effective_max_bytes = self.recent_max_bytes
else:
effective_max_bytes = max_bytes
block["output"] = self._compact(block["output"], effective_max_bytes)
return messages
def cleanup_expired_files(self) -> int:
"""Clean up files older than retention_days."""
"""Clean up files older than retention_days.
Returns:
Number of files successfully deleted.
"""
if not self.tool_result_dir.exists():
return 0
cutoff = datetime.now() - timedelta(days=self.retention_days)
deleted = 0
deleted = failed = 0
for fp in self.tool_result_dir.glob("*.txt"):
try:
for line in fp.read_text(encoding="utf-8").splitlines()[:3]:
if line.startswith("# created_at:"):
if datetime.fromisoformat(line.split(":", 1)[1].strip()) < cutoff:
fp.unlink()
deleted += 1
break
stat = os.stat(fp)
if sys.platform == "win32":
ts = stat.st_ctime # creation time on Windows
else:
ts = getattr(stat, "st_birthtime", stat.st_mtime) # macOS/BSD; Linux fallback to mtime
if datetime.fromtimestamp(ts) < cutoff:
fp.unlink()
deleted += 1
except FileNotFoundError:
pass # deleted by another process between glob and stat/unlink
except Exception as e:
logger.warning("Failed to process %s: %s", fp, e)
failed += 1
logger.warning("Failed to delete %s: %s", fp, e)
if deleted:
logger.info("Cleaned up %d expired files", deleted)
if deleted or failed:
logger.info("Cleaned up %d expired files (%d failed)", deleted, failed)
return deleted

View file

@ -34,6 +34,7 @@ class ReMeInMemoryMemory(InMemoryMemory):
self._token_counter: HuggingFaceTokenCounter = token_counter
self._msg_handler: AsMsgHandler = AsMsgHandler(token_counter)
self._dialog_path: Path | None = Path(dialog_path) if dialog_path else None
self._long_term_memory: str = ""
def _append_messages_to_dialog(self, messages: list[Msg]) -> int:
"""Append messages to dialog storage file.
@ -124,23 +125,20 @@ class ReMeInMemoryMemory(InMemoryMemory):
"""
filtered_content = [(msg, marks) for msg, marks in self.content if _MemoryMark.COMPRESSED not in marks]
parts = []
if self._long_term_memory:
parts.append(f"# Memories\n\n{self._long_term_memory}")
if prepend_summary and self._compressed_summary:
previous_summary = f"""
<previous-summary>
{self._compressed_summary}
</previous-summary>
The above is a summary of our previous conversation.
Use it as context to maintain continuity.
""".strip()
parts.append(
f"# Summary of previous conversation\n\n"
f"Previous conversation logs are offloaded to dialog/YYYY-MM-DD.jsonl (or nearby date files). "
"Here is the summary:\n\n"
f"{self._compressed_summary}\n"
f"The above is a summary of previous conversation, use it as context to maintain continuity.",
)
return [
Msg(
"user",
previous_summary,
"user",
),
*[msg for msg, _ in filtered_content],
]
if parts:
return [Msg("user", "\n\n".join(parts), "user"), *[msg for msg, _ in filtered_content]]
return [msg for msg, _ in filtered_content]

View file

@ -7,7 +7,7 @@ from typing import Optional
from agentscope.message import TextBlock
from agentscope.tool import ToolResponse
from ..utils import DEFAULT_MAX_BYTES, read_file_safe, truncate_output
from ..utils import read_file_safe, truncate_text_output, TRUNCATION_NOTICE_MARKER
class FileIO:
@ -32,19 +32,19 @@ class FileIO:
Returns:
The resolved absolute file path as string.
"""
path = Path(file_path)
path = Path(file_path).expanduser()
if path.is_absolute():
return str(path)
else:
return str(self.working_dir / file_path)
async def read( # pylint: disable=too-many-return-statements
async def read_file( # pylint: disable=too-many-return-statements
self,
file_path: str,
start_line: Optional[int] = None,
end_line: Optional[int] = None,
) -> ToolResponse:
"""Read a file. Relative paths resolve from working_dir.
"""Read a file. Relative paths resolve from WORKING_DIR.
Use start_line/end_line to read a specific line range (output includes
line numbers). Omit both to read the full file.
@ -57,6 +57,34 @@ class FileIO:
end_line (`int`, optional):
Last line to read (1-based, inclusive).
"""
# Convert start_line/end_line to int if they are strings
if start_line is not None:
try:
start_line = int(start_line)
except (ValueError, TypeError):
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: start_line must be an integer, got {start_line!r}.",
),
],
)
if end_line is not None:
try:
end_line = int(end_line)
except (ValueError, TypeError):
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: end_line must be an integer, got {end_line!r}.",
),
],
)
file_path = self._resolve_file_path(file_path)
if not os.path.exists(file_path):
@ -111,29 +139,29 @@ class FileIO:
# Extract selected lines
selected_content = "\n".join(all_lines[s - 1 : e])
# Apply smart truncation (keep head for file reading)
truncated, was_truncated, output_lines, reason = truncate_output(selected_content, keep="head")
# Apply smart truncation (consistent with shell output format)
text = truncate_text_output(
selected_content,
start_line=s,
total_lines=total,
file_path=file_path,
)
# Build response with truncation hints
if was_truncated:
end_display = s + output_lines - 1
next_line = end_display + 1
if reason == "lines":
hint = f"\n\n[Lines {s}-{end_display} of {total}. Use start_line={next_line} to continue.]"
else:
hint = (
f"\n\n[Lines {s}-{end_display} of {total} ({DEFAULT_MAX_BYTES // 1024}KB limit). "
f"Use start_line={next_line} to continue.]"
)
text = truncated + hint
elif e < total:
remaining = total - e
text = (
f"{file_path} (lines {s}-{e} of {total})\n{truncated}\n\n[{remaining} more lines. "
f"Use start_line={e + 1} to continue.]"
# Add continuation hint if partial read without truncation.
# Use TRUNCATION_NOTICE_MARKER format so ToolResultCompactor can
# re-truncate with the correct start_line when compacting old messages.
if text == selected_content and e < total:
content_bytes = len(text.encode("utf-8"))
notice = (
TRUNCATION_NOTICE_MARKER + f"\nThe output above was truncated."
f"\nThe full content is saved to the file "
f"and contains {total} lines in total."
f"\nThis excerpt starts at line {s} and "
f"covers the next {content_bytes} bytes."
"\nIf the current content is not enough, "
f"call `read_file` with file_path={file_path} start_line={e + 1} to read more."
)
else:
text = truncated
text = text + notice
return ToolResponse(
content=[TextBlock(type="text", text=text)],
@ -149,7 +177,7 @@ class FileIO:
],
)
async def write(
async def write_file(
self,
file_path: str,
content: str,
@ -167,7 +195,7 @@ class FileIO:
content=[
TextBlock(
type="text",
text="Error: No `file_path` provide.",
text="Error: No `file_path` provided.",
),
],
)
@ -195,7 +223,8 @@ class FileIO:
],
)
async def edit(
# pylint: disable=too-many-return-statements
async def edit_file(
self,
file_path: str,
old_text: str,
@ -212,22 +241,50 @@ class FileIO:
new_text (`str`):
Replacement text.
"""
response = await self.read(file_path=file_path)
if response.content and len(response.content) > 0:
error_text = response.content[0].get("text", "")
if error_text.startswith("Error:"):
return response
if not response.content or len(response.content) == 0:
if not file_path:
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: Failed to read file {file_path}.",
text="Error: No `file_path` provided.",
),
],
)
resolved_path = self._resolve_file_path(file_path)
if not os.path.exists(resolved_path):
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: The file {resolved_path} does not exist.",
),
],
)
if not os.path.isfile(resolved_path):
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: The path {resolved_path} is not a file.",
),
],
)
try:
content = read_file_safe(resolved_path)
except Exception as e:
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: Read file failed due to \n{e}",
),
],
)
content = response.content[0].get("text", "")
if old_text not in content:
return ToolResponse(
content=[
@ -239,7 +296,7 @@ class FileIO:
)
new_content = content.replace(old_text, new_text)
write_response = await self.write(file_path=file_path, content=new_content)
write_response = await self.write_file(file_path=resolved_path, content=new_content)
if write_response.content and len(write_response.content) > 0:
write_text = write_response.content[0].get("text", "")
@ -254,3 +311,50 @@ class FileIO:
),
],
)
async def append_file(
self,
file_path: str,
content: str,
) -> ToolResponse:
"""Append content to the end of a file. Relative paths resolve from
working_dir.
Args:
file_path (`str`):
Path to the file.
content (`str`):
Content to append.
"""
if not file_path:
return ToolResponse(
content=[
TextBlock(
type="text",
text="Error: No `file_path` provided.",
),
],
)
file_path = self._resolve_file_path(file_path)
try:
with open(file_path, "a", encoding="utf-8") as file:
file.write(content)
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Appended {len(content)} bytes to {file_path}.",
),
],
)
except Exception as e:
return ToolResponse(
content=[
TextBlock(
type="text",
text=f"Error: Append file failed due to \n{e}",
),
],
)

View file

@ -12,8 +12,6 @@ from pathlib import Path
from agentscope.message import TextBlock
from agentscope.tool import ToolResponse
from ..utils import truncate_shell_output
def _execute_subprocess_sync(
cmd: str,
@ -189,10 +187,6 @@ class Shell:
stdout_str = ""
stderr_str = stderr_suffix
# Apply output truncation
stdout_str = truncate_shell_output(stdout_str)
stderr_str = truncate_shell_output(stderr_str)
# Format the response in a human-friendly way
if returncode == 0:
# Success case: just show the output

View file

@ -1,13 +1,12 @@
"""utils"""
from .as_msg_handler import AsMsgHandler
from .file_utils import truncate_output, truncate_shell_output, read_file_safe, DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES
from .file_utils import truncate_text_output, read_file_safe, DEFAULT_MAX_BYTES, TRUNCATION_NOTICE_MARKER
__all__ = [
"AsMsgHandler",
"truncate_output",
"truncate_shell_output",
"truncate_text_output",
"read_file_safe",
"DEFAULT_MAX_BYTES",
"DEFAULT_MAX_LINES",
"TRUNCATION_NOTICE_MARKER",
]

View file

@ -204,7 +204,7 @@ class AsMsgHandler:
self,
messages: list[Msg],
memory_compact_threshold: int,
include_thinking: bool = False,
include_thinking: bool = True,
) -> str:
"""Format list of messages to a single formatted string.

View file

@ -1,112 +1,217 @@
# -*- coding: utf-8 -*-
"""Shared utilities for file and shell tools."""
# Default truncation limits
DEFAULT_MAX_LINES = 1000
DEFAULT_MAX_BYTES = 30 * 1024 # 30KB
import re
from ....core.utils import get_logger
logger = get_logger()
# Default truncation limit
DEFAULT_MAX_BYTES = 50 * 1024
# Maximum file size to read into memory (1GB)
MAX_FILE_READ_BYTES = 1024 * 1024 * 1024
# Marker prepended to every truncation notice.
# Format:
# <<<TRUNCATED>>>
# The output above was truncated.
# The full content is saved to the file and contains Z lines in total.
# This excerpt starts at line X and covers the next N bytes.
# If the current content is not enough, call `read_file` with file_path=<path> start_line=Y to read more.
#
# Split output on this marker to recover the original (untruncated) portion:
# original = output.split(TRUNCATION_NOTICE_MARKER)[0]
TRUNCATION_NOTICE_MARKER = "<<<TRUNCATED>>>"
def truncate_output(
def _truncate_fresh(
text: str,
max_lines: int = DEFAULT_MAX_LINES,
max_bytes: int = DEFAULT_MAX_BYTES,
keep: str = "head",
) -> tuple[str, bool, int, str]:
"""Smart truncation for large content.
start_line: int,
total_lines: int,
max_bytes: int,
file_path: str | None,
encoding: str,
) -> str:
"""Truncate fresh text (no prior truncation marker) by bytes with line integrity.
Args:
text: Text content to truncate.
max_lines: Maximum number of lines.
max_bytes: Maximum size in bytes.
keep: Which part to keep - "head" (first lines) or "tail" (last lines).
Slices at the byte boundary and appends a truncation notice with a continuation
hint so callers know which line to read next.
Returns:
(truncated_content, was_truncated, output_line_count, truncate_reason)
Returns the original text unchanged when it fits within max_bytes, or when the
last line itself exceeds max_bytes (unhandled edge case).
"""
if not text:
return text, False, 0, ""
text_bytes = text.encode(encoding)
lines = text.split("\n")
total_lines = len(lines)
# Under the byte limit — return as-is without any modification.
if len(text_bytes) <= max_bytes:
return text
# No truncation needed
if total_lines <= max_lines and len(text.encode("utf-8")) <= max_bytes:
return text, False, total_lines, ""
# Slice at the byte boundary.
# Assuming every single line is shorter than DEFAULT_MAX_BYTES, this cut always
# lands mid-line, guaranteeing at least one complete line before the boundary.
# Lines that exceed DEFAULT_MAX_BYTES are not handled and may be skipped entirely.
truncated = text_bytes[:max_bytes]
# Decode back to str; errors="ignore" drops any split multi-byte character
# at the cut boundary without raising an exception.
result = truncated.decode(encoding, errors="ignore")
# Apply line limit
if total_lines > max_lines:
if keep == "tail":
lines = lines[-max_lines:]
else:
lines = lines[:max_lines]
reason = "lines"
# Count '\n' characters to determine how many complete lines are included.
# The tail after the final '\n' is a partial line that will be covered by
# the next read starting at next_line.
newline_count = result.count("\n")
# Compute the first line number not yet fully included in this chunk.
# max(1, ...) prevents next_line from equaling start_line when a single line
# exceeds max_bytes (newline_count == 0), which would make the caller retry
# the same range indefinitely.
next_line = start_line + max(1, newline_count)
if next_line <= total_lines:
# Truncation fell before the last line — continue reading from next_line.
read_from = next_line
elif start_line < total_lines:
# next_line overshot total_lines, meaning the cut landed inside the last line.
# Re-read from the start of the last line so the caller gets it in full.
read_from = total_lines
else:
reason = ""
# start_line == total_lines: the last line itself exceeds DEFAULT_MAX_BYTES.
# This case is outside our handled range — return without a truncation notice.
return result
# Apply byte limit
if len("\n".join(lines).encode("utf-8")) > max_bytes:
if keep == "tail":
while lines and len("\n".join(lines).encode("utf-8")) > max_bytes:
lines.pop(0)
else:
truncated = []
current_bytes = 0
for line in lines:
line_bytes = len(line.encode("utf-8")) + 1
if current_bytes + line_bytes > max_bytes:
break
truncated.append(line)
current_bytes += line_bytes
lines = truncated
reason = "bytes"
notice = (
TRUNCATION_NOTICE_MARKER + f"\nThe output above was truncated."
f"\nThe full content is saved to the file and contains {total_lines} lines in total."
f"\nThis excerpt starts at line {start_line} and covers the next {max_bytes} bytes."
f"\nIf the current content is not enough, call `read_file` with file_path={file_path or ''} "
f"start_line={read_from} to read more."
)
return "\n".join(lines), True, len(lines), reason
return result + notice
def truncate_shell_output(text: str) -> str:
"""Truncate shell output to last N lines or M bytes, with truncation notice.
def _retruncate(
text: str,
max_bytes: int,
encoding: str,
) -> str:
"""Re-truncate text that was previously truncated (contains TRUNCATION_NOTICE_MARKER).
Extracts the original content before the marker, applies the new byte limit, and
updates the embedded notice (byte count and continuation line number) via regex.
Returns the original text unchanged when:
- the content already fits within max_bytes (with a small slack);
- required metadata fields cannot be parsed from the existing notice.
"""
parts = text.split(TRUNCATION_NOTICE_MARKER, 1)
original_content = parts[0]
old_notice = parts[1]
text_bytes = original_content.encode(encoding)
# Allow a small slack to avoid unnecessary re-truncation when content is just
# barely over the limit (e.g. due to minor encoding differences).
if len(text_bytes) <= max_bytes + 100:
return text
# Parse start_line from notice; return text unchanged if not found
start_match = re.search(r"starts at line (\d+)", old_notice)
if not start_match:
return text
start_line_parsed = int(start_match.group(1))
# Re-slice to the new byte limit.
# Because every line is assumed to be shorter than DEFAULT_MAX_BYTES, the cut
# always falls somewhere mid-line, so at least one complete line is preserved.
truncated_bytes = text_bytes[:max_bytes]
# errors="ignore" silently drops any incomplete multi-byte character at the cut boundary.
result = truncated_bytes.decode(encoding, errors="ignore")
# Each '\n' in result corresponds to one fully-included line;
# anything after the last '\n' is a partial line that was cut off.
newline_count = result.count("\n")
# The next read should start at the line immediately after all complete lines.
# max(1, ...) guards against the theoretical zero-newline case
# (impossible when every line is shorter than DEFAULT_MAX_BYTES).
next_line = start_line_parsed + max(1, newline_count)
if not re.search(r"covers the next \d+ bytes", old_notice):
return text
# _truncate_fresh always includes a continuation hint, so both fields are always present.
new_notice = re.sub(r"covers the next \d+ bytes", f"covers the next {max_bytes} bytes", old_notice)
new_notice = re.sub(r"start_line=\d+ to read more", f"start_line={next_line} to read more", new_notice)
return result + TRUNCATION_NOTICE_MARKER + new_notice
def truncate_text_output(
text: str,
start_line: int = 1,
total_lines: int = 0,
max_bytes: int = DEFAULT_MAX_BYTES,
file_path: str | None = None,
encoding: str = "utf-8",
) -> str:
"""Truncate file output by bytes with line integrity.
If text is under byte limit, return as-is.
If over limit, truncate at the last complete line that fits,
allowing the next read to start from a fresh line.
Dispatches to :func:`_truncate_fresh` for text seen for the first time, or to
:func:`_retruncate` when the text already contains a TRUNCATION_NOTICE_MARKER
from a previous pass.
Args:
text: The output text to truncate.
start_line: The starting line number (1-based). Ignored when text already
contains a truncation notice (values are parsed from the notice instead).
total_lines: Total lines in the original file. Ignored when text already
contains a truncation notice (values are parsed from the notice instead).
max_bytes: Maximum size in bytes.
file_path: Optional file path to include in the truncation notice.
encoding: Character encoding used for byte-length calculation and decoding.
Returns:
Truncated text with notice if truncated.
"""
if not text:
return text
if max_bytes <= 0:
return text
try:
total_lines = len(text.split("\n"))
truncated, was_truncated, output_lines, reason = truncate_output(text, keep="tail")
if not was_truncated:
return text
start_line = total_lines - output_lines + 1
if reason == "lines":
notice = f"\n\n[Output truncated: showing lines {start_line}-{total_lines} of {total_lines} total]"
if TRUNCATION_NOTICE_MARKER in text:
return _retruncate(text, max_bytes=max_bytes, encoding=encoding)
else:
notice = (
f"\n\n[Output truncated: showing lines {start_line}-{total_lines} of {total_lines} "
f"({DEFAULT_MAX_BYTES // 1024}KB limit)]"
return _truncate_fresh(
text,
start_line=start_line,
total_lines=total_lines,
max_bytes=max_bytes,
file_path=file_path,
encoding=encoding,
)
return truncated + notice
except Exception:
logger.warning("truncate_text_output failed, returning original text", exc_info=True)
return text
def read_file_safe(file_path: str) -> str:
"""Read file with Unicode error handling.
def read_file_safe(file_path: str, max_bytes: int = MAX_FILE_READ_BYTES) -> str:
"""Read file with Unicode error handling and memory protection.
Args:
file_path: Path to the file.
max_bytes: Maximum bytes to read into memory (default 1GB).
Returns:
File content as string.
File content as string (up to max_bytes).
"""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
return f.read(max_bytes)
except UnicodeDecodeError:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
return f.read()
return f.read(max_bytes)

View file

@ -162,6 +162,28 @@ class ReMe(Application):
return memory_type, memory_target
def _ensure_started(self) -> None:
"""Ensure memory operations run only after services are initialized."""
if not self._started:
raise RuntimeError("ReMe is not started. Call `await reme.start()` before using memory APIs.")
@staticmethod
def _unwrap_memory_result(
result: str | dict,
operation_name: str,
return_dict: bool,
) -> str | dict:
"""Normalize memory API results and fail loudly on swallowed inner errors."""
if not isinstance(result, dict):
raise RuntimeError(f"{operation_name} failed before producing a structured result: {result}")
if "answer" not in result:
raise RuntimeError(f"{operation_name} returned an invalid result payload: missing 'answer'")
if return_dict:
return result
return result["answer"]
async def summarize_memory(
self,
messages: list[Message | dict],
@ -173,10 +195,12 @@ class ReMe(Application):
version: str = "default",
retrieve_top_k: int = 20,
return_dict: bool = False,
raise_exception: bool = False,
llm_config_name: str = "default",
**kwargs,
) -> str | dict:
"""Summarize personal, procedural and tool memories for the given context."""
self._ensure_started()
format_messages: list[Message] = []
for message in messages:
if isinstance(message, dict):
@ -192,12 +216,14 @@ class ReMe(Application):
enable_when_to_use=False,
enable_multiple=True,
top_k=retrieve_top_k,
raise_exception=raise_exception,
),
AddMemory(
enable_thinking_params=enable_thinking_params,
enable_memory_target=False,
enable_when_to_use=False,
enable_multiple=True,
raise_exception=raise_exception,
),
]
if self.enable_profile:
@ -207,18 +233,21 @@ class ReMe(Application):
enable_thinking_params=False,
enable_memory_target=False,
profile_dir=self.profile_dir,
raise_exception=raise_exception,
),
UpdateProfilesV1(
enable_thinking_params=enable_thinking_params,
enable_memory_target=False,
enable_multiple=True,
profile_dir=self.profile_dir,
raise_exception=raise_exception,
),
],
)
personal_summarizer: BaseMemoryAgent = PersonalSummarizer(
llm=llm_config_name,
tools=personal_summarizer_tools,
raise_exception=raise_exception,
)
else:
@ -233,14 +262,17 @@ class ReMe(Application):
enable_when_to_use=False,
enable_multiple=True,
top_k=retrieve_top_k,
raise_exception=raise_exception,
),
AddMemory(
enable_thinking_params=enable_thinking_params,
enable_memory_target=False,
enable_when_to_use=False,
enable_multiple=True,
raise_exception=raise_exception,
),
],
raise_exception=raise_exception,
)
tool_summarizer: BaseMemoryAgent = ToolSummarizer(
llm=llm_config_name,
@ -251,14 +283,17 @@ class ReMe(Application):
enable_when_to_use=False,
enable_multiple=True,
top_k=retrieve_top_k,
raise_exception=raise_exception,
),
AddMemory(
enable_thinking_params=enable_thinking_params,
enable_memory_target=False,
enable_when_to_use=False,
enable_multiple=True,
raise_exception=raise_exception,
),
],
raise_exception=raise_exception,
)
memory_agents = []
@ -306,7 +341,11 @@ class ReMe(Application):
memory_agents = [personal_summarizer, procedural_summarizer, tool_summarizer]
reme_summarizer: BaseMemoryAgent = ReMeSummarizer(
tools=[AddHistory(), DelegateTask(memory_agents=memory_agents)],
tools=[
AddHistory(raise_exception=raise_exception),
DelegateTask(memory_agents=memory_agents, raise_exception=raise_exception),
],
raise_exception=raise_exception,
)
result = await reme_summarizer.call(
@ -317,10 +356,7 @@ class ReMe(Application):
**kwargs,
)
if return_dict:
return result
else:
return result["answer"]
return self._unwrap_memory_result(result, "summarize_memory", return_dict)
async def retrieve_memory(
self,
@ -335,10 +371,12 @@ class ReMe(Application):
retrieve_top_k: int = 20,
enable_time_filter: bool = True,
return_dict: bool = False,
raise_exception: bool = False,
llm_config_name: str = "default",
**kwargs,
) -> str | dict:
"""Retrieve relevant personal, procedural and tool memories for a query."""
self._ensure_started()
if version == "default":
personal_retriever_tools = []
@ -348,6 +386,7 @@ class ReMe(Application):
enable_thinking_params=False,
enable_memory_target=False,
profile_dir=self.profile_dir,
raise_exception=raise_exception,
),
)
personal_retriever_tools.extend(
@ -357,16 +396,19 @@ class ReMe(Application):
enable_thinking_params=enable_thinking_params,
enable_time_filter=enable_time_filter,
enable_multiple=True,
raise_exception=raise_exception,
),
ReadHistory(
enable_thinking_params=enable_thinking_params,
enable_multiple=True,
raise_exception=raise_exception,
),
],
)
personal_retriever: BaseMemoryAgent = PersonalRetriever(
llm=llm_config_name,
tools=personal_retriever_tools,
raise_exception=raise_exception,
)
else:
raise NotImplementedError(f"version={version} is not supported")
@ -379,12 +421,15 @@ class ReMe(Application):
enable_thinking_params=enable_thinking_params,
enable_time_filter=False,
enable_multiple=True,
raise_exception=raise_exception,
),
ReadHistory(
enable_thinking_params=enable_thinking_params,
enable_multiple=True,
raise_exception=raise_exception,
),
],
raise_exception=raise_exception,
)
tool_retriever: BaseMemoryAgent = ToolRetriever(
llm=llm_config_name,
@ -394,12 +439,15 @@ class ReMe(Application):
enable_thinking_params=enable_thinking_params,
enable_time_filter=False,
enable_multiple=True,
raise_exception=raise_exception,
),
ReadHistory(
enable_thinking_params=enable_thinking_params,
enable_multiple=True,
raise_exception=raise_exception,
),
],
raise_exception=raise_exception,
)
memory_agents = []
@ -444,7 +492,8 @@ class ReMe(Application):
memory_agents = [personal_retriever, procedural_retriever, tool_retriever]
reme_retriever: BaseMemoryAgent = ReMeRetriever(
tools=[DelegateTask(memory_agents=memory_agents)],
tools=[DelegateTask(memory_agents=memory_agents, raise_exception=raise_exception)],
raise_exception=raise_exception,
)
result = await reme_retriever.call(
@ -456,10 +505,7 @@ class ReMe(Application):
**kwargs,
)
if return_dict:
return result
else:
return result["answer"]
return self._unwrap_memory_result(result, "retrieve_memory", return_dict)
async def add_memory(
self,

View file

@ -74,6 +74,7 @@ class ReMeLight(Application):
default_as_llm_config: dict | None = None,
default_embedding_model_config: dict | None = None,
default_file_store_config: dict | None = None,
default_file_watcher_config: dict | None = None,
vector_weight: float = 0.7,
candidate_multiplier: float = 3.0,
enable_load_env: bool = False,
@ -101,6 +102,10 @@ class ReMeLight(Application):
dictionary for the embedding model.
default_file_store_config (dict | None): Default configuration
dictionary for the file storage backend.
default_file_watcher_config (dict | None): Default configuration
dictionary for the file watcher. If ``watch_paths`` is included,
it is used as-is. Otherwise the built-in watch paths (MEMORY.md,
memory.md, and the memory directory) are used.
vector_weight (float): Weight assigned to vector similarity search
in hybrid search operations. Range [0.0, 1.0], default 0.7.
Higher values prioritize semantic similarity over keyword matching.
@ -130,6 +135,20 @@ class ReMeLight(Application):
self.vector_weight: float = vector_weight
self.candidate_multiplier: float = candidate_multiplier
# Build the file watcher config: use provided watch_paths if given, otherwise use defaults
_default_watch_paths = [
str(self.working_path / "MEMORY.md"),
str(self.working_path / "memory.md"),
str(self.memory_path),
]
if default_file_watcher_config and default_file_watcher_config.get("watch_paths"):
_merged_file_watcher_config = default_file_watcher_config
else:
_merged_file_watcher_config = {
**(default_file_watcher_config or {}),
"watch_paths": _default_watch_paths,
}
# Initialize the parent Application class with comprehensive configuration
super().__init__(
llm_api_key=llm_api_key,
@ -145,13 +164,7 @@ class ReMeLight(Application):
default_as_llm_config=default_as_llm_config,
default_embedding_model_config=default_embedding_model_config,
default_file_store_config=default_file_store_config,
default_file_watcher_config={
"watch_paths": [
str(self.working_path / "MEMORY.md"),
str(self.working_path / "memory.md"),
str(self.memory_path),
],
},
default_file_watcher_config=_merged_file_watcher_config,
)
# Initialize list to track background summarization tasks
@ -168,7 +181,7 @@ class ReMeLight(Application):
Returns:
Computed compaction threshold as an integer.
"""
return int(max_input_length * compact_ratio * 0.9)
return int(max_input_length * compact_ratio * 0.95)
def _cleanup_tool_results(self) -> int:
"""
@ -231,10 +244,10 @@ class ReMeLight(Application):
async def compact_tool_result(
self,
messages: list[Msg],
old_max_bytes: int = 3000,
recent_max_bytes: int = 100 * 1024,
retention_days: int = 3,
recent_n: int = 1,
old_threshold: int = 500,
recent_threshold: int = 30000,
retention_days: int = 7,
) -> list[Msg]:
"""
Compact tool results by truncating large outputs and saving full content to files.
@ -247,30 +260,37 @@ class ReMeLight(Application):
Args:
messages (list[Msg]): List of messages potentially containing tool results
that may need compaction.
recent_n (int): Number of recent messages to use recent_threshold for.
Default 1.
old_threshold (int): Character threshold for old messages. Default 500.
recent_threshold (int): Character threshold for recent messages. Default 30000.
old_max_bytes (int): Byte threshold for old (non-recent) messages. Default 3000.
recent_max_bytes (int): Byte threshold for recent messages (trailing consecutive
tool-result messages). Default 100KB (102400 bytes). Content exceeding this
limit is saved to disk; the message retains the first 100KB with a
read_file-style truncation notice and the saved file path.
retention_days (int): Number of days to retain tool result files.
Default 7.
Default 3.
recent_n (int): Minimum number of most-recent tool-result messages to treat
as "recent" (using recent_max_bytes). The actual recent window is the
larger of this value and the trailing consecutive tool-result run.
Default 1.
Returns:
list[Msg]: The processed list of messages with large tool results compacted.
If an error occurs, returns the original unmodified messages.
Note:
- Tool results are truncated based on old_threshold/recent_threshold
- Full content of truncated results is saved to tool_result_path
- Expired files are automatically cleaned up during this operation
- Recent tool results (trailing consecutive tool-result messages) are truncated
to recent_max_bytes using read_file-style output with a file path hint.
- Old tool results are truncated to old_max_bytes bytes.
- Full content of truncated results is saved to tool_result_path.
- Expired files are automatically cleaned up during this operation.
"""
try:
# Create compactor with instance configuration
compactor = ToolResultCompactor(
tool_result_dir=self.tool_result_path,
retention_days=retention_days,
old_max_bytes=old_max_bytes,
recent_max_bytes=recent_max_bytes,
recent_n=recent_n,
old_threshold=old_threshold,
recent_threshold=recent_threshold,
)
# Execute compaction and get processed messages
@ -347,7 +367,9 @@ class ReMeLight(Application):
max_input_length: float = 128 * 1024,
compact_ratio: float = 0.7,
previous_summary: str = "",
) -> str:
return_dict: bool = False,
add_thinking_block: bool = True,
) -> str | dict:
"""
Compact a list of messages into a condensed summary.
@ -371,10 +393,13 @@ class ReMeLight(Application):
Defaults to 0.7.
previous_summary (str): Previous summary to incorporate into the new
summary for continuity. Defaults to empty string.
return_dict (bool): If True, returns a dict with user_message,
history_compact, and is_valid. Defaults to False.
Returns:
str: The condensed summary of the messages, or an empty string if
an error occurred during compaction.
str | dict: The condensed summary string, or a dict containing
user_message, history_compact, and is_valid if return_dict=True.
Returns empty string or dict with empty values if an error occurred.
"""
try:
compactor = Compactor(
@ -383,6 +408,8 @@ class ReMeLight(Application):
as_llm_formatter=as_llm_formatter,
as_token_counter=as_token_counter,
language=language if language == "zh" else "",
return_dict=return_dict,
add_thinking_block=add_thinking_block,
)
return await compactor.call(
@ -392,8 +419,10 @@ class ReMeLight(Application):
)
except Exception as e:
# Log error and return empty string to indicate failure
# Log error and return appropriate empty result
logger.exception(f"Error compacting memory: {e}")
if return_dict:
return {"user_message": str(e), "history_compact": str(e), "is_valid": False}
return ""
async def summary_memory(
@ -407,6 +436,7 @@ class ReMeLight(Application):
max_input_length: float = 128 * 1024,
compact_ratio: float = 0.7,
timezone: str | None = None,
add_thinking_block: bool = True,
) -> str:
"""
Generate a comprehensive summary of the given messages.
@ -445,9 +475,9 @@ class ReMeLight(Application):
if toolkit is None:
toolkit = Toolkit()
file_io = FileIO(working_dir=str(self.working_path))
toolkit.register_tool_function(file_io.read)
toolkit.register_tool_function(file_io.write)
toolkit.register_tool_function(file_io.edit)
toolkit.register_tool_function(file_io.read_file)
toolkit.register_tool_function(file_io.write_file)
toolkit.register_tool_function(file_io.edit_file)
summarizer = Summarizer(
working_dir=str(self.working_path),
@ -459,6 +489,7 @@ class ReMeLight(Application):
as_token_counter=as_token_counter,
language=language if language == "zh" else "",
timezone=timezone,
add_thinking_block=add_thinking_block,
)
return await summarizer.call(messages=messages, service_context=self.service_context)

View file

@ -295,7 +295,6 @@ async def test_file_watch_integration():
"watch_paths": [TestConfig.WORKING_DIR, f"{TestConfig.WORKING_DIR}/memory"],
"suffix_filters": [".md"],
"recursive": False,
"scan_on_start": True,
},
)

View file

@ -101,9 +101,9 @@ def create_toolkit(working_dir: str) -> Toolkit:
"""Create a default Toolkit with FileIO tools for testing."""
toolkit = Toolkit()
file_io = FileIO(working_dir=working_dir)
toolkit.register_tool_function(file_io.read)
toolkit.register_tool_function(file_io.write)
toolkit.register_tool_function(file_io.edit)
toolkit.register_tool_function(file_io.read_file)
toolkit.register_tool_function(file_io.write_file)
toolkit.register_tool_function(file_io.edit_file)
return toolkit

View file

@ -7,8 +7,8 @@ from pathlib import Path
from agentscope.message import Msg
from reme.core.utils import is_truncated
from reme.memory.file_based.components import ToolResultCompactor
from reme.memory.file_based.utils import TRUNCATION_NOTICE_MARKER
def create_tool_result_msg(output: str | list, tool_name: str = "test_tool") -> Msg:
@ -33,7 +33,7 @@ class TestToolResultCompactor:
def test_no_truncation_when_under_threshold(self):
"""Test that short content is not truncated."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=1000)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=1000)
messages = [create_tool_result_msg("short content")]
result = asyncio.run(op.call(messages=messages))
@ -45,14 +45,14 @@ class TestToolResultCompactor:
def test_truncation_when_over_threshold(self):
"""Test that long content is truncated and saved to file."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
long_content = "x" * 500
messages = [create_tool_result_msg(long_content)]
_ = asyncio.run(op.call(messages=messages))
output = messages[0].content[0]["output"]
assert is_truncated(output)
assert TRUNCATION_NOTICE_MARKER in output
assert "[Full content saved to:" in output
# Verify file was created
@ -68,7 +68,7 @@ class TestToolResultCompactor:
def test_skip_already_truncated(self):
"""Test that already truncated content is not re-truncated."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
truncated_content = "head<<<TRUNCATED>>>(100 chars omitted)<<<END_TRUNCATED>>>tail"
messages = [create_tool_result_msg(truncated_content)]
@ -80,20 +80,20 @@ class TestToolResultCompactor:
def test_truncation_list_output(self):
"""Test truncation of list output with text blocks."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
list_output = [{"type": "text", "text": "y" * 500}]
messages = [create_tool_result_msg(list_output)]
asyncio.run(op.call(messages=messages))
text_block = messages[0].content[0]["output"][0]
assert is_truncated(text_block["text"])
assert TRUNCATION_NOTICE_MARKER in text_block["text"]
assert len(list(Path(tmpdir).glob("*.txt"))) == 1
def test_list_output_no_truncation_when_short(self):
"""Test that short list output is not truncated."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=1000)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=1000)
list_output = [{"type": "text", "text": "short"}]
messages = [create_tool_result_msg(list_output)]
@ -105,7 +105,7 @@ class TestToolResultCompactor:
def test_list_output_multiple_text_blocks(self):
"""Test truncation of multiple text blocks in list output."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
list_output = [
{"type": "text", "text": "a" * 500},
{"type": "text", "text": "short"},
@ -116,15 +116,15 @@ class TestToolResultCompactor:
asyncio.run(op.call(messages=messages))
output = messages[0].content[0]["output"]
assert is_truncated(output[0]["text"])
assert TRUNCATION_NOTICE_MARKER in output[0]["text"]
assert output[1]["text"] == "short" # unchanged
assert is_truncated(output[2]["text"])
assert TRUNCATION_NOTICE_MARKER in output[2]["text"]
assert len(list(Path(tmpdir).glob("*.txt"))) == 2
def test_list_output_mixed_block_types(self):
"""Test that non-text blocks in list output are unchanged."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
list_output = [
{"type": "text", "text": "c" * 500},
{"type": "image", "source": {"type": "url", "url": "http://example.com/img.png"}},
@ -134,14 +134,14 @@ class TestToolResultCompactor:
asyncio.run(op.call(messages=messages))
output = messages[0].content[0]["output"]
assert is_truncated(output[0]["text"])
assert TRUNCATION_NOTICE_MARKER in output[0]["text"]
assert output[1] == {"type": "image", "source": {"type": "url", "url": "http://example.com/img.png"}}
assert len(list(Path(tmpdir).glob("*.txt"))) == 1
def test_cleanup_expired_files(self):
"""Test cleanup of expired files."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100, retention_days=1)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100, retention_days=1)
# Create an old file
old_time = (datetime.now() - timedelta(days=2)).isoformat()
@ -162,7 +162,7 @@ class TestToolResultCompactor:
def test_string_content_msg_unchanged(self):
"""Test that messages with string content are unchanged."""
with tempfile.TemporaryDirectory() as tmpdir:
op = ToolResultCompactor(tool_result_dir=tmpdir, tool_result_threshold=100)
op = ToolResultCompactor(tool_result_dir=tmpdir, recent_max_bytes=100)
messages = [Msg(name="user", role="user", content="hello world")]
asyncio.run(op.call(messages=messages))

View file

@ -4,6 +4,7 @@
import asyncio
import os
import re
import shutil
import tempfile
@ -11,7 +12,7 @@ import pytest
from reme.memory.file_based.tools.file_io import FileIO
from reme.memory.file_based.tools.shell import Shell
from reme.memory.file_based.utils import DEFAULT_MAX_LINES, DEFAULT_MAX_BYTES
from reme.memory.file_based.utils import DEFAULT_MAX_BYTES
# ============ Shell Tests ============
@ -76,23 +77,6 @@ def test_shell_multiline_output(shell_env):
assert "line3" in text
def test_shell_truncated_output(shell_env):
"""Test output truncation for large output."""
lines_to_generate = DEFAULT_MAX_LINES + 500
cmd = f"seq 1 {lines_to_generate}"
result = asyncio.run(shell_env["shell"].execute_shell_command(cmd))
text = result.content[0].get("text", "")
# Should contain truncation notice
assert "truncated" in text.lower()
# Should contain the last line (tail is kept)
assert str(lines_to_generate) in text
# Verify first numeric line is > 1 (truncated from head)
numeric_lines = [tl for tl in text.strip().split("\n") if tl.isdigit()]
if numeric_lines:
assert int(numeric_lines[0]) > 1
def test_shell_timeout(shell_env):
"""Test command timeout handling."""
result = asyncio.run(
@ -116,13 +100,17 @@ def fileio_env():
with open(simple_file, "w", encoding="utf-8") as f:
f.write("line1\nline2\nline3\nline4\nline5")
# Create large file (exceeds DEFAULT_MAX_LINES)
# Create large file (exceeds DEFAULT_MAX_BYTES)
large_file = os.path.join(test_dir, "large.txt")
with open(large_file, "w", encoding="utf-8") as f:
for i in range(1, DEFAULT_MAX_LINES + 500):
# Each line is ~7-10 bytes ("line N\n"); generate enough to exceed limit.
# Line 1 is literally "line 1" so the head-kept assertion can match it.
line_count = (DEFAULT_MAX_BYTES // 7) + 1000
for i in range(1, line_count + 1):
f.write(f"line {i}\n")
# Create large bytes file (exceeds DEFAULT_MAX_BYTES)
# Lines are 101 bytes each; at DEFAULT_MAX_BYTES the cut lands mid-line → else branch
large_bytes_file = os.path.join(test_dir, "large_bytes.txt")
with open(large_bytes_file, "w", encoding="utf-8") as f:
content = "x" * 100 + "\n"
@ -130,19 +118,31 @@ def fileio_env():
for _ in range(lines_needed):
f.write(content)
# Single line larger than DEFAULT_MAX_BYTES → newline_count==0 branch in truncate
huge_line_file = os.path.join(test_dir, "huge_line.txt")
with open(huge_line_file, "w", encoding="utf-8") as f:
f.write("A" * (DEFAULT_MAX_BYTES + 1000) + "\nline2\n")
# Empty file
empty_file = os.path.join(test_dir, "empty.txt")
with open(empty_file, "w", encoding="utf-8") as f:
f.write("")
yield {
"dir": test_dir,
"file_io": file_io,
"simple_file": simple_file,
"large_file": large_file,
"large_bytes_file": large_bytes_file,
"huge_line_file": huge_line_file,
"empty_file": empty_file,
}
shutil.rmtree(test_dir, ignore_errors=True)
def test_read_file_success(fileio_env):
"""Test successful file reading."""
result = asyncio.run(fileio_env["file_io"].read(fileio_env["simple_file"]))
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["simple_file"]))
text = result.content[0].get("text", "")
assert "line1" in text
assert "line5" in text
@ -150,14 +150,14 @@ def test_read_file_success(fileio_env):
def test_read_file_relative_path(fileio_env):
"""Test reading file with relative path."""
result = asyncio.run(fileio_env["file_io"].read("simple.txt"))
result = asyncio.run(fileio_env["file_io"].read_file("simple.txt"))
text = result.content[0].get("text", "")
assert "line1" in text
def test_read_file_not_exists(fileio_env):
"""Test reading non-existent file."""
result = asyncio.run(fileio_env["file_io"].read("nonexistent.txt"))
result = asyncio.run(fileio_env["file_io"].read_file("nonexistent.txt"))
text = result.content[0].get("text", "")
assert "Error" in text
assert "does not exist" in text
@ -166,7 +166,7 @@ def test_read_file_not_exists(fileio_env):
def test_read_file_with_line_range(fileio_env):
"""Test reading specific line range."""
result = asyncio.run(
fileio_env["file_io"].read(fileio_env["simple_file"], start_line=2, end_line=4),
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=2, end_line=4),
)
text = result.content[0].get("text", "")
assert "line2" in text
@ -177,7 +177,7 @@ def test_read_file_with_line_range(fileio_env):
def test_read_file_start_line_exceeds(fileio_env):
"""Test start_line exceeding file length."""
result = asyncio.run(
fileio_env["file_io"].read(fileio_env["simple_file"], start_line=100),
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=100),
)
text = result.content[0].get("text", "")
assert "Error" in text
@ -187,15 +187,15 @@ def test_read_file_start_line_exceeds(fileio_env):
def test_read_file_invalid_range(fileio_env):
"""Test invalid line range (start > end)."""
result = asyncio.run(
fileio_env["file_io"].read(fileio_env["simple_file"], start_line=4, end_line=2),
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=4, end_line=2),
)
text = result.content[0].get("text", "")
assert "Error" in text
def test_read_file_truncated_by_lines(fileio_env):
"""Test file truncation by line limit."""
result = asyncio.run(fileio_env["file_io"].read(fileio_env["large_file"]))
def test_read_file_truncated(fileio_env):
"""Test file truncation by byte limit."""
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_file"]))
text = result.content[0].get("text", "")
assert "line 1" in text # Head is kept
assert "continue" in text.lower()
@ -203,19 +203,140 @@ def test_read_file_truncated_by_lines(fileio_env):
def test_read_file_truncated_by_bytes(fileio_env):
"""Test file truncation by byte limit."""
result = asyncio.run(fileio_env["file_io"].read(fileio_env["large_bytes_file"]))
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_bytes_file"]))
text = result.content[0].get("text", "")
assert "continue" in text.lower() or "KB" in text
assert "continue" in text.lower()
assert "KB limit" in text
def test_read_directory_error(fileio_env):
"""Test reading a directory returns error."""
result = asyncio.run(fileio_env["file_io"].read(fileio_env["dir"]))
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["dir"]))
text = result.content[0].get("text", "")
assert "Error" in text
assert "not a file" in text
def test_read_file_single_line_range(fileio_env):
"""Test reading exactly one line (start_line == end_line)."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=3, end_line=3),
)
text = result.content[0].get("text", "")
assert "line3" in text
assert "line2" not in text
assert "line4" not in text
def test_read_file_only_start_line(fileio_env):
"""Test reading from start_line to end of file (no end_line)."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=4),
)
text = result.content[0].get("text", "")
assert "line4" in text
assert "line5" in text
assert "line1" not in text
assert "line3" not in text
def test_read_file_only_end_line(fileio_env):
"""Test reading from beginning to end_line (no start_line)."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], end_line=2),
)
text = result.content[0].get("text", "")
assert "line1" in text
assert "line2" in text
assert "line4" not in text
assert "line5" not in text
def test_read_file_end_line_clamped(fileio_env):
"""Test end_line beyond total lines is silently clamped to file end."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=1, end_line=999),
)
text = result.content[0].get("text", "")
assert "Error" not in text
assert "line1" in text
assert "line5" in text
def test_read_file_continuation_hint(fileio_env):
"""Partial range read without truncation shows remaining-lines continuation hint."""
# simple.txt has 5 lines; reading 1-3 leaves 2 more
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line=1, end_line=3),
)
text = result.content[0].get("text", "")
assert "more lines" in text
assert "start_line=4" in text
def test_read_file_truncated_next_line_hint(fileio_env):
"""Truncated large file provides a valid start_line > 1 to continue."""
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_file"]))
text = result.content[0].get("text", "")
match = re.search(r"start_line=(\d+)", text)
assert match is not None, "Expected start_line hint in truncated output"
assert int(match.group(1)) > 1
def test_read_file_truncated_mid_line_message(fileio_env):
"""Truncation mid-line reports which line is truncated (else branch)."""
# large_bytes_file lines are 101 bytes; truncation lands mid-line
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["large_bytes_file"]))
text = result.content[0].get("text", "")
assert "is truncated" in text.lower()
def test_read_file_huge_single_line(fileio_env):
"""Single line exceeding byte limit triggers 'partially shown' notice (newline_count==0 branch)."""
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["huge_line_file"]))
text = result.content[0].get("text", "")
assert "partially shown" in text.lower()
assert "start_line=2" in text
def test_read_file_invalid_start_line_type(fileio_env):
"""Non-integer start_line returns a descriptive error."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line="abc"),
)
text = result.content[0].get("text", "")
assert "Error" in text
assert "start_line" in text
def test_read_file_invalid_end_line_type(fileio_env):
"""Non-integer end_line returns a descriptive error."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], end_line="xyz"),
)
text = result.content[0].get("text", "")
assert "Error" in text
assert "end_line" in text
def test_read_file_start_line_as_string(fileio_env):
"""Numeric-string start_line/end_line are coerced to int successfully."""
result = asyncio.run(
fileio_env["file_io"].read_file(fileio_env["simple_file"], start_line="2", end_line="4"),
)
text = result.content[0].get("text", "")
assert "Error" not in text
assert "line2" in text
assert "line4" in text
def test_read_file_empty(fileio_env):
"""Reading an empty file returns without error."""
result = asyncio.run(fileio_env["file_io"].read_file(fileio_env["empty_file"]))
text = result.content[0].get("text", "")
assert "Error" not in text
# ============ FileIO Write Tests ============
@ -231,7 +352,7 @@ def write_env():
def test_write_new_file(write_env):
"""Test writing a new file."""
file_path = os.path.join(write_env["dir"], "new_file.txt")
result = asyncio.run(write_env["file_io"].write(file_path, "test content"))
result = asyncio.run(write_env["file_io"].write_file(file_path, "test content"))
text = result.content[0].get("text", "")
assert "Wrote" in text
@ -245,7 +366,7 @@ def test_write_overwrite_file(write_env):
with open(file_path, "w", encoding="utf-8") as f:
f.write("old content")
result = asyncio.run(write_env["file_io"].write(file_path, "new content"))
result = asyncio.run(write_env["file_io"].write_file(file_path, "new content"))
text = result.content[0].get("text", "")
assert "Wrote" in text
@ -255,14 +376,14 @@ def test_write_overwrite_file(write_env):
def test_write_empty_path(write_env):
"""Test writing with empty path."""
result = asyncio.run(write_env["file_io"].write("", "content"))
result = asyncio.run(write_env["file_io"].write_file("", "content"))
text = result.content[0].get("text", "")
assert "Error" in text
def test_write_relative_path(write_env):
"""Test writing file with relative path."""
result = asyncio.run(write_env["file_io"].write("relative.txt", "relative content"))
result = asyncio.run(write_env["file_io"].write_file("relative.txt", "relative content"))
text = result.content[0].get("text", "")
assert "Wrote" in text
@ -290,7 +411,7 @@ def edit_env():
def test_edit_replace_text(edit_env):
"""Test replacing text in file."""
result = asyncio.run(
edit_env["file_io"].edit(edit_env["edit_file"], "Hello", "Hi"),
edit_env["file_io"].edit_file(edit_env["edit_file"], "Hello", "Hi"),
)
text = result.content[0].get("text", "")
assert "Successfully" in text
@ -305,7 +426,7 @@ def test_edit_replace_text(edit_env):
def test_edit_text_not_found(edit_env):
"""Test editing when text not found."""
result = asyncio.run(
edit_env["file_io"].edit(edit_env["edit_file"], "NotExists", "Replacement"),
edit_env["file_io"].edit_file(edit_env["edit_file"], "NotExists", "Replacement"),
)
text = result.content[0].get("text", "")
assert "Error" in text
@ -315,7 +436,7 @@ def test_edit_text_not_found(edit_env):
def test_edit_nonexistent_file(edit_env):
"""Test editing non-existent file."""
result = asyncio.run(
edit_env["file_io"].edit("nonexistent.txt", "old", "new"),
edit_env["file_io"].edit_file("nonexistent.txt", "old", "new"),
)
text = result.content[0].get("text", "")
assert "Error" in text

View file

@ -0,0 +1,381 @@
# -*- coding: utf-8 -*-
# pylint: disable=missing-function-docstring
"""Unit tests for _truncate_fresh, _retruncate, and truncate_text_output.
Assumptions that mirror the production code:
- Every single line is shorter than DEFAULT_MAX_BYTES.
- Lines that exceed DEFAULT_MAX_BYTES are explicitly ignored / not fully read.
- _truncate_fresh always includes a continuation hint in the notice, so
_retruncate can assume the hint is always present.
Run:
cd tests/light && python test_truncate_text_output.py
"""
import re
from reme.memory.file_based.utils.file_utils import (
TRUNCATION_NOTICE_MARKER,
_truncate_fresh,
_retruncate,
truncate_text_output,
)
# ── helpers ──────────────────────────────────────────────────────────────────
LINE_BYTES = 20 # every test line is exactly this many bytes
ENC = "utf-8"
def make_line(i: int) -> str:
"""Return a line that is exactly LINE_BYTES bytes in UTF-8.
Format: "L{i}" padded with underscores, terminated with "\\n".
Works for i up to 999.
"""
prefix = f"L{i}"
return prefix + "_" * (LINE_BYTES - 1 - len(prefix)) + "\n"
def make_text(n: int, start: int = 1) -> str:
"""Build n lines starting from line number `start`."""
return "".join(make_line(i) for i in range(start, start + n))
def content_of(result: str) -> str:
"""Return the portion before the truncation marker."""
return result.split(TRUNCATION_NOTICE_MARKER)[0]
def notice_of(result: str) -> str:
"""Return the portion after the truncation marker, or '' if absent."""
parts = result.split(TRUNCATION_NOTICE_MARKER, 1)
return parts[1] if len(parts) > 1 else ""
def parse_next_line(result: str) -> int:
"""Extract 'start_line=X' from the notice, or -1 if absent."""
m = re.search(r"start_line=(\d+) to read more", notice_of(result))
return int(m.group(1)) if m else -1
def parse_covers_bytes(result: str) -> int:
"""Extract 'covers the next X bytes' from the notice."""
m = re.search(r"covers the next (\d+) bytes", notice_of(result))
return int(m.group(1)) if m else -1
def assert_eq(a, b, msg=""):
assert a == b, f"{msg}: expected {b!r}, got {a!r}"
def assert_in(needle, haystack, msg=""):
assert needle in haystack, f"{msg}: {needle!r} not found in {haystack!r}"
def assert_not_in(needle, haystack, msg=""):
assert needle not in haystack, f"{msg}: {needle!r} unexpectedly found in {haystack!r}"
# ── _truncate_fresh tests ─────────────────────────────────────────────────────
def test_fresh_no_truncation_when_under_limit():
text = make_text(3) # 60 bytes
result = _truncate_fresh(text, start_line=1, total_lines=3, max_bytes=100, file_path=None, encoding=ENC)
assert_eq(result, text, "under limit: no change")
assert_not_in(TRUNCATION_NOTICE_MARKER, result)
def test_fresh_no_truncation_at_exact_limit():
text = make_text(3) # 60 bytes
result = _truncate_fresh(text, start_line=1, total_lines=3, max_bytes=60, file_path=None, encoding=ENC)
assert_eq(result, text, "at exact limit: no change")
assert_not_in(TRUNCATION_NOTICE_MARKER, result)
def test_fresh_mid_file_correct_next_line():
# 10 lines × 20 bytes = 200 bytes; max_bytes=95
# 95 bytes → 4 complete lines (80 bytes) + 15 bytes into line 5
# newline_count=4, next_line=1+4=5, 5<=10 → read_from=5
text = make_text(10)
result = _truncate_fresh(text, start_line=1, total_lines=10, max_bytes=95, file_path=None, encoding=ENC)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_next_line(result), 5, "next_line after mid-file cut")
assert_eq(parse_covers_bytes(result), 95)
assert_in("L4", content_of(result), "line 4 fully included")
assert_not_in("L5\n", content_of(result), "line 5 not fully included")
def test_fresh_notice_contains_total_lines_and_start_line():
text = make_text(10, start=3)
result = _truncate_fresh(text, start_line=3, total_lines=12, max_bytes=95, file_path="/tmp/foo.txt", encoding=ENC)
notice = notice_of(result)
assert_in("contains 12 lines in total", notice)
assert_in("starts at line 3", notice)
assert_in("file_path=/tmp/foo.txt", notice)
def test_fresh_next_line_equals_total_lines_reads_from_last():
# 5 lines × 20 = 100 bytes; max_bytes=85
# 85 bytes → 4 complete lines (80 bytes) + 5 bytes into line 5
# newline_count=4, next_line=1+4=5 = total_lines → read_from=5
text = make_text(5)
result = _truncate_fresh(text, start_line=1, total_lines=5, max_bytes=85, file_path=None, encoding=ENC)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_next_line(result), 5, "should re-read the last line")
def test_fresh_next_line_overshoots_total_lines():
# max_bytes=115 → 5 complete lines (100 bytes) + 15 bytes into line 6
# newline_count=5, next_line=1+5=6 > total_lines(5), start_line(1) < 5
# → read_from = total_lines = 5
text = make_text(10)
result = _truncate_fresh(text, start_line=1, total_lines=5, max_bytes=115, file_path=None, encoding=ENC)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_next_line(result), 5, "overshoot: fall back to total_lines")
def test_fresh_long_first_line_skips_to_next_line():
# Line 1 is 200 bytes (> max_bytes=100), no '\n' in truncated result.
# newline_count=0, next_line=1+max(1,0)=2, 2<=5 → read_from=2
long_line = "A" * 199 + "\n" # 200 bytes
text = long_line + make_text(4, start=2)
result = _truncate_fresh(text, start_line=1, total_lines=5, max_bytes=100, file_path=None, encoding=ENC)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_next_line(result), 2, "skip to line 2 after long line 1")
assert_not_in("A" * 101, content_of(result))
def test_fresh_long_middle_line_skips_to_next_line():
# Lines 1-2 normal (40 bytes); line 3 is 200 bytes; lines 4-5 normal.
# max_bytes=100: fits lines 1-2 (40 bytes) + 60 bytes of line 3 (no '\n')
# newline_count=2, next_line=1+2=3, 3<=5 → read_from=3
text = make_text(2, start=1) + "B" * 199 + "\n" + make_text(2, start=4)
result = _truncate_fresh(text, start_line=1, total_lines=5, max_bytes=100, file_path=None, encoding=ENC)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_next_line(result), 3, "restart from long line 3")
def test_fresh_long_last_line_start_equals_total_no_notice():
# start_line == total_lines, single line too long → unhandled case, no notice.
# newline_count=0, next_line=5+1=6 > 5, start_line(5)==total_lines(5)
long_line = "C" * 199 + "\n" # 200 bytes
result = _truncate_fresh(long_line, start_line=5, total_lines=5, max_bytes=100, file_path=None, encoding=ENC)
assert_not_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(len(result), 100, "only max_bytes bytes returned")
def test_fresh_long_last_line_arrived_from_previous_chunk_no_notice():
long_line = "D" * 199 + "\n"
result = _truncate_fresh(long_line, start_line=5, total_lines=5, max_bytes=80, file_path=None, encoding=ENC)
assert_not_in(TRUNCATION_NOTICE_MARKER, result)
# ── _retruncate tests ─────────────────────────────────────────────────────────
def _first_pass(n_lines: int = 50, max_bytes: int = 490) -> str:
"""Produce a first-truncated text via _truncate_fresh."""
return _truncate_fresh(
make_text(n_lines),
start_line=1,
total_lines=n_lines,
max_bytes=max_bytes,
file_path="/file.txt",
encoding=ENC,
)
def test_retruncate_within_slack_returns_unchanged():
# content ≈ 490 bytes; re-truncate with max_bytes=400.
# 490 <= 400+100=500 → slack hit, return unchanged.
pass1 = _first_pass(n_lines=50, max_bytes=490)
result = _retruncate(pass1, max_bytes=400, encoding=ENC)
assert_eq(result, pass1, "within slack: unchanged")
def test_retruncate_updates_byte_count_in_notice():
pass1 = _first_pass(n_lines=50, max_bytes=490)
result = _retruncate(pass1, max_bytes=195, encoding=ENC)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_covers_bytes(result), 195, "notice byte count updated")
def test_retruncate_updates_next_line():
# pass1 content ≈ 490 bytes (24 complete lines), start_line=1.
# re-truncate at 195 bytes → 9 complete lines, next_line=1+9=10.
pass1 = _first_pass(n_lines=50, max_bytes=490)
result = _retruncate(pass1, max_bytes=195, encoding=ENC)
assert_eq(parse_next_line(result), 10, "next_line updated to 10")
def test_retruncate_content_is_smaller():
pass1 = _first_pass(n_lines=50, max_bytes=490)
result = _retruncate(pass1, max_bytes=195, encoding=ENC)
assert len(content_of(result)) < len(content_of(pass1)), "content should shrink"
def test_retruncate_missing_starts_at_line_returns_unchanged():
# Notice missing "starts at line X" → return unchanged.
# Use large content (600 bytes) to bypass the 100-byte slack.
large_content = make_text(30) # 600 bytes
broken = (
large_content + TRUNCATION_NOTICE_MARKER + "\nThe output above was truncated."
"\nThe full content is saved to the file and contains 30 lines in total."
"\nThis excerpt covers the next 600 bytes." # no "starts at line X"
"\nIf the current content is not enough, call `read_file` with file_path=/f.txt start_line=5 to read more."
)
result = _retruncate(broken, max_bytes=10, encoding=ENC)
assert_eq(result, broken, "missing 'starts at line': unchanged")
def test_retruncate_missing_covers_next_bytes_returns_unchanged():
# Notice has malformed "covers ??? bytes" → return unchanged.
large_content = "X" * 500 + "\n"
broken = (
large_content + TRUNCATION_NOTICE_MARKER + "\nThe output above was truncated."
"\nThe full content is saved to the file and contains 10 lines in total."
"\nThis excerpt starts at line 1 and covers ??? bytes."
"\nIf the current content is not enough, call `read_file` with file_path=/f.txt start_line=5 to read more."
)
result = _retruncate(broken, max_bytes=10, encoding=ENC)
assert_eq(result, broken, "missing 'covers the next N bytes': unchanged")
# ── truncate_text_output dispatch / guard tests ───────────────────────────────
def test_dispatch_empty_string():
result = truncate_text_output("", start_line=1, total_lines=0, max_bytes=10)
assert_eq(result, "", "empty string bypassed")
def test_dispatch_max_bytes_zero():
text = make_text(5)
result = truncate_text_output(text, max_bytes=0)
assert_eq(result, text, "max_bytes=0 bypassed")
def test_dispatch_routes_to_fresh_when_no_marker():
text = make_text(10)
result = truncate_text_output(text, start_line=1, total_lines=10, max_bytes=95)
assert_in(TRUNCATION_NOTICE_MARKER, result)
assert_eq(parse_next_line(result), 5)
def test_dispatch_routes_to_retruncate_when_marker_present():
pass1 = _first_pass(n_lines=50, max_bytes=490)
pass2 = truncate_text_output(pass1, max_bytes=195)
assert_in(TRUNCATION_NOTICE_MARKER, pass2)
assert_eq(parse_covers_bytes(pass2), 195)
# ── multi-pass integration tests ──────────────────────────────────────────────
def test_three_pass_decreasing_truncation():
"""Three successive truncations with shrinking max_bytes."""
n_lines = 50
text = make_text(n_lines) # 1000 bytes
pass1 = truncate_text_output(text, start_line=1, total_lines=n_lines, max_bytes=490, file_path="/f.txt")
assert TRUNCATION_NOTICE_MARKER in pass1
assert parse_covers_bytes(pass1) == 490
assert parse_next_line(pass1) > 1
# 490 > 195+100=295 → re-truncation proceeds
pass2 = truncate_text_output(pass1, max_bytes=195)
assert TRUNCATION_NOTICE_MARKER in pass2
assert parse_covers_bytes(pass2) == 195
assert parse_next_line(pass2) < parse_next_line(pass1), "next_line regresses"
assert len(content_of(pass2)) < len(content_of(pass1))
# content_of(pass2) ≈ 195 bytes; 195 > 90+100=190 → re-truncation proceeds
pass3 = truncate_text_output(pass2, max_bytes=90)
assert TRUNCATION_NOTICE_MARKER in pass3
assert parse_covers_bytes(pass3) == 90
assert parse_next_line(pass3) < parse_next_line(pass2), "next_line regresses further"
assert len(content_of(pass3)) < len(content_of(pass2))
def test_three_pass_next_lines_are_consistent():
"""next_line values should monotonically decrease with each re-truncation."""
n_lines = 50
text = make_text(n_lines)
pass1 = truncate_text_output(text, start_line=1, total_lines=n_lines, max_bytes=490, file_path="/f.txt")
pass2 = truncate_text_output(pass1, max_bytes=195)
pass3 = truncate_text_output(pass2, max_bytes=90)
n1 = parse_next_line(pass1)
n2 = parse_next_line(pass2)
n3 = parse_next_line(pass3)
assert n1 > n2 > n3 > 1, f"Expected n1 > n2 > n3 > 1, got {n1} > {n2} > {n3}"
# ── runner ────────────────────────────────────────────────────────────────────
def run_all():
tests = [
# _truncate_fresh
test_fresh_no_truncation_when_under_limit,
test_fresh_no_truncation_at_exact_limit,
test_fresh_mid_file_correct_next_line,
test_fresh_notice_contains_total_lines_and_start_line,
test_fresh_next_line_equals_total_lines_reads_from_last,
test_fresh_next_line_overshoots_total_lines,
test_fresh_long_first_line_skips_to_next_line,
test_fresh_long_middle_line_skips_to_next_line,
test_fresh_long_last_line_start_equals_total_no_notice,
test_fresh_long_last_line_arrived_from_previous_chunk_no_notice,
# _retruncate
test_retruncate_within_slack_returns_unchanged,
test_retruncate_updates_byte_count_in_notice,
test_retruncate_updates_next_line,
test_retruncate_content_is_smaller,
test_retruncate_missing_starts_at_line_returns_unchanged,
test_retruncate_missing_covers_next_bytes_returns_unchanged,
# truncate_text_output dispatch / guard
test_dispatch_empty_string,
test_dispatch_max_bytes_zero,
test_dispatch_routes_to_fresh_when_no_marker,
test_dispatch_routes_to_retruncate_when_marker_present,
# multi-pass integration
test_three_pass_decreasing_truncation,
test_three_pass_next_lines_are_consistent,
]
passed = 0
failed = 0
for t in tests:
try:
t()
print(f" PASS {t.__name__}")
passed += 1
except Exception as e:
print(f" FAIL {t.__name__}: {e}")
failed += 1
print(f"\n{passed} passed, {failed} failed out of {len(tests)} tests.")
return failed == 0
if __name__ == "__main__":
import sys
ok = run_all()
sys.exit(0 if ok else 1)

View file

@ -5,7 +5,7 @@ Async unit tests for BaseFileWatcher covering:
- File suffix filtering
- Start/stop lifecycle
- Callback functionality
- scan_on_start feature
- rebuild_index_on_start feature
Usage:
pytest tests/test_base_file_watcher.py -v
@ -369,12 +369,12 @@ class TestCallbackFunctionality:
# ==================== Test Scan on Start ====================
class TestScanOnStart:
"""Tests for scan_on_start feature."""
class TestRebuildIndexOnStart:
"""Tests for rebuild_index_on_start feature."""
@pytest.mark.asyncio
async def test_scan_on_start_false(self, temp_files, temp_dir: Path):
"""Test that scan_on_start=False doesn't scan existing files."""
async def test_rebuild_index_on_start_false(self, temp_files, temp_dir: Path):
"""Test that rebuild_index_on_start=False doesn't scan existing files."""
callback_called = []
async def callback(changes):
@ -387,7 +387,7 @@ class TestScanOnStart:
watcher = BaseFileWatcher(
watch_paths=str(temp_dir),
scan_on_start=False,
rebuild_index_on_start=False,
callback=callback,
file_store=mock_file_store,
)
@ -400,8 +400,8 @@ class TestScanOnStart:
assert len(callback_called) == 0
@pytest.mark.asyncio
async def test_scan_on_start_true_with_files(self, temp_files, temp_dir: Path):
"""Test that scan_on_start=True scans existing files."""
async def test_rebuild_index_on_start_true_with_files(self, temp_files, temp_dir: Path):
"""Test that rebuild_index_on_start=True scans existing files."""
callback_called = []
async def callback(changes):
@ -414,7 +414,7 @@ class TestScanOnStart:
watcher = BaseFileWatcher(
watch_paths=str(temp_dir),
scan_on_start=True,
rebuild_index_on_start=True,
callback=callback,
file_store=mock_file_store,
)
@ -434,8 +434,8 @@ class TestScanOnStart:
assert all(change == Change.added for change, _ in all_changes)
@pytest.mark.asyncio
async def test_scan_on_start_with_suffix_filter(self, temp_files, temp_dir: Path):
"""Test scan_on_start respects suffix filters."""
async def test_rebuild_index_on_start_with_suffix_filter(self, temp_files, temp_dir: Path):
"""Test rebuild_index_on_start respects suffix filters."""
callback_called = []
async def callback(changes):
@ -447,7 +447,7 @@ class TestScanOnStart:
watcher = BaseFileWatcher(
watch_paths=str(temp_dir),
scan_on_start=True,
rebuild_index_on_start=True,
suffix_filters=[".txt"],
callback=callback,
file_store=mock_file_store,
@ -467,8 +467,8 @@ class TestScanOnStart:
assert path.endswith(".txt"), f"Expected .txt file, got {path}"
@pytest.mark.asyncio
async def test_scan_on_start_recursive(self, temp_nested_dir: Path):
"""Test scan_on_start with recursive=True."""
async def test_rebuild_index_on_start_recursive(self, temp_nested_dir: Path):
"""Test rebuild_index_on_start with recursive=True."""
callback_called = []
async def callback(changes):
@ -480,7 +480,7 @@ class TestScanOnStart:
watcher = BaseFileWatcher(
watch_paths=str(temp_nested_dir),
scan_on_start=True,
rebuild_index_on_start=True,
recursive=True,
suffix_filters=[".txt"],
callback=callback,
@ -503,8 +503,8 @@ class TestScanOnStart:
assert nested_found, "Should find files in nested directories"
@pytest.mark.asyncio
async def test_scan_on_start_non_recursive(self, temp_nested_dir: Path):
"""Test scan_on_start with recursive=False."""
async def test_rebuild_index_on_start_non_recursive(self, temp_nested_dir: Path):
"""Test rebuild_index_on_start with recursive=False."""
callback_called = []
async def callback(changes):
@ -516,7 +516,7 @@ class TestScanOnStart:
watcher = BaseFileWatcher(
watch_paths=str(temp_nested_dir),
scan_on_start=True,
rebuild_index_on_start=True,
recursive=False,
suffix_filters=[".txt"],
callback=callback,
@ -538,8 +538,8 @@ class TestScanOnStart:
assert not nested_found, "Should not find files in nested directories"
@pytest.mark.asyncio
async def test_scan_on_start_nonexistent_path(self):
"""Test scan_on_start with non-existent path."""
async def test_rebuild_index_on_start_nonexistent_path(self):
"""Test rebuild_index_on_start with non-existent path."""
callback_called = []
async def callback(changes):
@ -551,7 +551,7 @@ class TestScanOnStart:
watcher = BaseFileWatcher(
watch_paths="/nonexistent/path",
scan_on_start=True,
rebuild_index_on_start=True,
callback=callback,
file_store=mock_file_store,
)
@ -709,7 +709,7 @@ class TestEdgeCases:
watcher = BaseFileWatcher(
watch_paths=str(file_path),
scan_on_start=True,
rebuild_index_on_start=True,
callback=callback,
file_store=mock_file_store,
)
@ -742,7 +742,7 @@ class TestEdgeCases:
watcher = BaseFileWatcher(
watch_paths=str(empty_dir),
scan_on_start=True,
rebuild_index_on_start=True,
callback=callback,
file_store=mock_file_store,
)

View file

@ -0,0 +1,132 @@
"""Tests for ReMe memory error handling and raise_exception propagation."""
import pytest
import reme.reme as reme_module
from reme import ReMe
class Recorder:
"""Stub that records constructor args for later inspection."""
instances = []
def __init__(self, *args, **kwargs):
self.args = args
self.kwargs = kwargs
self.call_kwargs = None
Recorder.instances.append(self)
class TopLevelAgent(Recorder):
"""Stub agent that returns a successful structured result."""
async def call(self, **kwargs):
"""Simulate a successful agent call."""
self.call_kwargs = kwargs
return {"answer": "ok", "success": True}
def _make_reme() -> ReMe:
"""Create a ReMe instance with startup bypassed for unit testing."""
reme = ReMe(enable_logo=False, log_to_console=False, enable_profile=False)
reme._started = True # pylint: disable=protected-access
return reme
def _patch_summarize_dependencies(monkeypatch: pytest.MonkeyPatch) -> None:
"""Patch all summarize-path dependencies with stubs."""
Recorder.instances = []
monkeypatch.setattr(reme_module, "AddDraftAndRetrieveSimilarMemory", Recorder)
monkeypatch.setattr(reme_module, "AddMemory", Recorder)
monkeypatch.setattr(reme_module, "AddHistory", Recorder)
monkeypatch.setattr(reme_module, "DelegateTask", Recorder)
monkeypatch.setattr(reme_module, "PersonalSummarizer", Recorder)
monkeypatch.setattr(reme_module, "ProceduralSummarizer", Recorder)
monkeypatch.setattr(reme_module, "ToolSummarizer", Recorder)
monkeypatch.setattr(reme_module, "ReMeSummarizer", TopLevelAgent)
def _patch_retrieve_dependencies(monkeypatch: pytest.MonkeyPatch) -> None:
"""Patch all retrieve-path dependencies with stubs."""
Recorder.instances = []
monkeypatch.setattr(reme_module, "RetrieveMemory", Recorder)
monkeypatch.setattr(reme_module, "ReadHistory", Recorder)
monkeypatch.setattr(reme_module, "DelegateTask", Recorder)
monkeypatch.setattr(reme_module, "PersonalRetriever", Recorder)
monkeypatch.setattr(reme_module, "ProceduralRetriever", Recorder)
monkeypatch.setattr(reme_module, "ToolRetriever", Recorder)
monkeypatch.setattr(reme_module, "ReMeRetriever", TopLevelAgent)
@pytest.mark.asyncio
@pytest.mark.parametrize("raise_exception", [False, True])
async def test_summarize_memory_propagates_raise_exception(
monkeypatch: pytest.MonkeyPatch,
raise_exception: bool,
):
"""Verify raise_exception is forwarded to every sub-agent in summarize."""
_patch_summarize_dependencies(monkeypatch)
reme = _make_reme()
result = await reme.summarize_memory(
messages=[{"role": "user", "content": "hi", "time_created": "2026-03-20 10:00:00"}],
task_name="demo-task",
raise_exception=raise_exception,
)
assert result == "ok"
assert Recorder.instances
assert all(instance.kwargs.get("raise_exception") is raise_exception for instance in Recorder.instances)
@pytest.mark.asyncio
@pytest.mark.parametrize("raise_exception", [False, True])
async def test_retrieve_memory_propagates_raise_exception(
monkeypatch: pytest.MonkeyPatch,
raise_exception: bool,
):
"""Verify raise_exception is forwarded to every sub-agent in retrieve."""
_patch_retrieve_dependencies(monkeypatch)
reme = _make_reme()
result = await reme.retrieve_memory(
query="hello",
task_name="demo-task",
raise_exception=raise_exception,
)
assert result == "ok"
assert Recorder.instances
assert all(instance.kwargs.get("raise_exception") is raise_exception for instance in Recorder.instances)
@pytest.mark.asyncio
async def test_summarize_memory_raises_runtime_error_for_unstructured_result(monkeypatch: pytest.MonkeyPatch):
"""Verify RuntimeError is raised when the top-level summarizer returns a plain string."""
Recorder.instances = []
reme = _make_reme()
monkeypatch.setattr(reme_module, "PersonalSummarizer", Recorder)
monkeypatch.setattr(reme_module, "ProceduralSummarizer", Recorder)
monkeypatch.setattr(reme_module, "ToolSummarizer", Recorder)
monkeypatch.setattr(reme_module, "AddDraftAndRetrieveSimilarMemory", Recorder)
monkeypatch.setattr(reme_module, "AddMemory", Recorder)
monkeypatch.setattr(reme_module, "AddHistory", Recorder)
monkeypatch.setattr(reme_module, "DelegateTask", Recorder)
class FailingTopLevelAgent(Recorder):
"""Stub agent that returns a failure string instead of a dict."""
async def call(self, **kwargs):
"""Simulate a failed agent call returning a plain error string."""
self.call_kwargs = kwargs
return "[ReMeSummarizer] failed: boom"
monkeypatch.setattr(reme_module, "ReMeSummarizer", FailingTopLevelAgent)
with pytest.raises(RuntimeError, match="summarize_memory failed before producing a structured result"):
await reme.summarize_memory(
messages=[{"role": "user", "content": "hi", "time_created": "2026-03-20 10:00:00"}],
task_name="demo-task",
)