大家好，今天给大家分享一套可直接落地的 ERP Agent 记忆系统架构。在开发 ERP 智能助手时，相信很多开发者都会遇到一个核心问题：长对话场景下，上下文超限、历史信息丢失、跨会话无法复用知识——这套三层记忆体系，正是为解决这些痛点而生，结合了 CLAW 压缩策略和 RAGFlow 记忆设计，兼顾性能与实用性。

        先给大家划重点：这套架构无需复杂依赖，基于 Python 实现，支持无限长对话、服务重启自动恢复，还能保证关键信息不丢失，适合各类 ERP 场景（如财务 AR/AP 表查询、业务流程咨询等）的智能助手开发。

一、整体架构：三层记忆体系（核心设计）

不同于传统的单层记忆，这套架构分为「实时上下文压缩→短期记忆→长期记忆」三层，层层递进、分工明确，既防止单次请求超限，又能实现跨会话知识积累。

架构流程图（可直接复制到本地查看）：

三层核心分工（表格清晰对比，一目了然）：

层级	触发时机	压缩方式	存储位置	核心用途
实时压缩	每轮对话前	结构化摘要（非截断）	内存	防止单次请求超限
短期记忆	每次添加消息	滑动窗口 + Token 预算	JSON 文件	保存当前会话上下文
长期记忆	消息数 > 100 条	LLM 摘要 + 原始归档	Markdown 文件	跨会话知识积累

二、分层详解（附核心代码+实操）

下面逐一层拆解，每个层级都附上可直接复用的核心代码，新手也能快速上手。

第一层：实时上下文压缩（防止单次请求超限）

核心作用：每轮对话前检查 Token 数量，当超过阈值时，对历史上下文进行结构化压缩，避免模型请求超限（比如 GPT、Claude 等模型的 Token 上限）。

1. 核心配置（可直接修改）

# backend/core/graph.py
COMPRESS_THRESHOLD = 60000  # 6万 Token 触发压缩（可调整）
PRESERVE_COUNT = 4          # 保留最近4条消息（保证对话连贯性）
FORCE_DELETE_THRESHOLD = 100000  # 压缩后仍超10万，强制删除旧消息
ERROR_THRESHOLD = 120000    # 超12万 Token 报错，提示新建会话

2. 工作原理（核心代码）

每轮对话前自动检查 Token 数，超限则触发压缩：

# 每轮对话前检查（backend/core/graph.py 353-404行）
estimated_tokens = estimate_tokens(current_messages)  # Token 估算
if estimated_tokens > 60000:  # 达到触发阈值
    current_messages = self._compress_tool_messages(current_messages)  # 执行压缩

3. 压缩效果（直观对比）

不同于简单截断，基于 CLAW 策略的压缩会生成结构化摘要，保留关键信息：

原始消息（冗余，含大量工具返回内容）：

[SystemMessage]
[HumanMessage] "查询 AR 表"
[AIMessage] "让我检索..."
[ToolMessage] [21万字符的文档内容]
[AIMessage] "根据文档..."
[HumanMessage] "再查询 AP 表"
[AIMessage] "让我检索..."
[ToolMessage] [15万字符的文档内容]
[AIMessage] "根据文档..."
[HumanMessage] "总结一下"
[AIMessage] "..."

压缩后（保留关键信息，大幅缩减 Token）：

[SystemMessage]
[SystemMessage - 压缩摘要]
  """
  [上下文压缩摘要]
  压缩范围：
  - 共 6 条消息被压缩（用户=2, 助手=2, 工具=2）
  - 使用的工具：search_knowledge_base, get_document_by_name
  - 最近的用户请求：
    • 查询 AR 表
    • 再查询 AP 表
  - 关键时间线：
    • 用户: 查询 AR 表
    • 助手: 让我检索...
    • 工具: 检索到5个文档...
    ... 省略 2 条消息 ...
  """
[HumanMessage] "总结一下"  ← 保留最近4条
[AIMessage] "..."           ← 保留最近4条

4. 关键特性（避坑重点）

• 结构化摘要：统计消息数量、工具使用记录、用户请求，不丢失核心信息；

• 三级防护：触发压缩→强制删除→报错，层层兜底，避免系统崩溃；

• 去重机制：检测工具返回结果前100字符，跳过重复内容，减少冗余。

第二层：短期记忆（当前会话上下文管理）

核心作用：管理当前会话的对话历史，控制上下文大小，提高响应速度，同时支持持久化，服务重启后可自动恢复。

1. 存储结构（一目了然）

rag_data/memory/
  ├── {session_id}.json          ← 消息历史（核心文件）
  └── {session_id}.meta.json     ← 会话元数据（创建时间、最后活跃时间等）

2. 核心逻辑（可复用代码）

滑动窗口裁剪 + 自动持久化：

# backend/core/memory.py - MemoryManager 类
window_size = 10      # 保底保留最近10轮对话
max_tokens = 8000     # 短期记忆 Token 预算（可调整）

# 添加消息并自动裁剪、持久化
def add_message(session_id, message):
    history.append(message)
    self._trim(session_id)   # 滑动窗口裁剪（超出Token则删除最早消息）
    self._save(session_id)   # 持久化到 JSON 文件

# Token 估算算法（精准估算，避免超限）
def _estimate_tokens(content):
    chinese_chars = sum(1 for c in content if '\u4e00' <= c <= '\u9fff')
    english_words = len(content.split()) - chinese_chars
    return int(chinese_chars * 1.5 + english_words)

3. 数据流（清晰易懂）

用户输入 → add_message() → 内存缓存 → 滑动窗口裁剪 → 持久化到 JSON → 服务重启自动恢复

第三层：长期记忆（跨会话知识积累）

核心作用：当会话消息数过多时，自动将历史对话压缩为 LLM 摘要，存储到 Markdown 文件，实现跨会话知识复用（比如用户下次问相同的 AR 表问题，助手可直接调用长期记忆，无需重复检索）。

1. 存储结构

rag_data/memory/{session_id}/
  ├── MEMORY.md      ← LLM 生成的结构化摘要（核心）
  └── HISTORY.md     ← 原始对话归档（备份，可追溯）

2. 核心代码（自动触发压缩）

# backend/core/memory.py 300-456行
# 当消息数 > 100 条时，自动触发长期记忆压缩
if len(history) > 100:
    await consolidate_memory(session_id, llm)  # 调用LLM生成摘要

# LLM 摘要生成逻辑（关键代码）
def consolidate_memory(session_id, llm):
    current_memory = read_long_term_memory(session_id)  # 读取当前长期记忆
    formatted_messages = format_messages(to_consolidate)  # 格式化待压缩消息
    
    # LLM 提示词（可直接复用）
    prompt = f"""请分析以下对话，提取关键信息并更新长期记忆。
## 当前长期记忆
{current_memory}
## 需要处理的对话
{formatted_messages}
请以 Markdown 格式输出更新后的长期记忆，包括：
1. **用户偏好和习惯**
2. **已知的数据库信息**（表名、字段名、连接信息）
3. **已解决的问题和答案**
4. **待解决的问题**
"""
    # 调用 LLM 生成摘要并写入 MEMORY.md
    response = await llm.ainvoke([HumanMessage(content=prompt)])
    write_long_term_memory(session_id, response.content)
    
    # 原始对话归档到 HISTORY.md（备份，防止摘要丢失）
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
    history_entry = f"[{timestamp}] 归档 {len(to_consolidate)} 条消息\n\n{formatted_messages[:500]}..."
    append_history_log(session_id, history_entry)

3. 容错机制（避坑关键）

如果 LLM 摘要失败（比如网络问题、模型超时），系统会自动降级为原始归档，确保数据不丢失：

# LLM 摘要失败降级逻辑
if self._consolidation_failures[session_id] >= 3:
    return self._raw_archive(session_id, to_consolidate)  # 直接归档原始消息

三、完整工作流程（实战场景）

以用户查询 ERP 中 AR/AP 表的长对话为例，完整流程如下（可直接对应实际开发场景）：

第1轮：用户问 "AR 表结构"
  ↓
[短期记忆] 添加 HumanMessage + AIMessage
  ↓
[持久化] 保存到 {session_id}.json

第2轮：用户问 "AP 表结构"
  ↓
[短期记忆] 添加消息
  ↓
[滑动窗口] 检查 Token 数（未超限）

...（继续对话，累计消息）...

第50轮：上下文达到 6万 Token
  ↓
[实时压缩] 触发压缩 → 前46条消息压缩为1条摘要，保留最近4条
  ↓
继续对话

第100轮：消息数 > 100 条
  ↓
[长期记忆] 触发自动压缩 → 取出最早50条消息
  ↓
调用 LLM 生成摘要 → 写入 MEMORY.md；原始对话归档 → 写入 HISTORY.md
  ↓
从短期记忆中删除这50条消息，继续对话

...（无限循环，支持长对话）...

四、Token 消耗对比（核心优势）

很多开发者担心压缩会影响体验，这里直接上对比，一目了然：

1. 无压缩机制（痛点明显）

第1轮：2k Token
第2轮：4k Token
...
第50轮：100k Token → 超限报错 ❌（对话中断）

2. 有压缩机制（完美解决）

第1轮：2k Token
第2轮：4k Token
...
第30轮：60k Token → 触发压缩 → 降到10k Token
第50轮：40k Token → 再次压缩 → 降到8k Token
...
无限对话 ✅（不超限、不丢失关键信息）

五、常见问题（FAQ）

整理了开发中最常遇到的4个问题，直接对号入座解决：

Q1：为什么需要三层记忆？不能用单层吗？

A：单层记忆无法兼顾「长对话支持」和「知识复用」：比如单层内存记忆，服务重启后丢失；单层文件记忆，响应速度慢；三层分工，既解决超限问题，又实现跨会话复用。

Q2：压缩会丢失关键信息吗？

A：不会！实时压缩生成结构化摘要，保留用户请求、工具使用、关键时间线；长期记忆有原始对话归档（HISTORY.md），可随时追溯。

Q3：如何调整压缩阈值，适配自己的模型？

A：直接修改配置参数即可（比如模型Token上限是8万，可将COMPRESS_THRESHOLD改为7万，预留冗余）：

# 更激进的压缩（更早触发，适合Token上限低的模型） COMPRESS_THRESHOLD = 40000 # 更保守的压缩（更晚触发，适合Token上限高的模型） COMPRESS_THRESHOLD = 80000

Q4：LLM 摘要失败怎么办？

A：系统有三重兜底：1. 重试3次；2. 重试失败则自动归档原始消息；3. 压缩后仍超限，强制删除旧消息，避免报错。

六、总结

这套 ERP Agent 记忆系统，核心优势在于「三层防护、分工明确、可落地、易调试」：

• ✅ 支持无限长对话，解决 Token 超限痛点；

• ✅ 关键信息不丢失，结构化压缩+原始归档双重保障；

• ✅ 服务重启自动恢复，无需担心数据丢失；

• ✅ 基于 CLAW、RAGFlow 业界最佳实践，稳定性有保障；

• ✅ 代码可直接复用，适配各类 ERP 智能助手场景。

目前这套架构已经在我们的 ERP 智能助手项目中落地，运行稳定，解决了长对话和跨会话知识复用的核心痛点。如果大家在使用过程中有疑问，或者需要调整适配自己的项目，可以在评论区留言交流～

最后，附上核心参考来源（方便大家深入学习）：

• CLAW (Claude Code)：reference-project/claw-code-main/rust/crates/runtime/src/compact.rs

• RAGFlow：reference-project/ragflow-main/rag/prompts/prompts.py（59-88行）

• LangChain：消息序列化/反序列化（langchain_core.messages）

觉得有用的话，记得点赞+收藏，关注我，后续分享更多 ERP 智能助手开发实战技巧！