s08

长期记忆系统

智能层

Fact 置信度管理

LLM 抽取 → 置信度评分 → 自动清洗

上传事件自动剔除、per-agent 记忆空间隔离、最大 Fact 数量约束

🧠

为什么需要基于置信度的记忆系统？

Agent 每次对话都从零开始——不记得用户偏好、项目背景、之前的决策。简单地把所有对话存下来既浪费 token 又引入噪声。DeerFlow 的记忆系统以 Fact 为最小单位，通过 LLM 提取→置信度评分→max_facts 容量控制→上传事件自动清洗，实现精准的跨会话记忆。

提取

→

评分

→

清洗

→

持久化

→

召回

Fact 示例

用户偏好 Python

0.92

项目使用 FastAPI

0.88

上传了 test.csv

0.15

用户是后端工程师

0.76

长期记忆系统总览

DeerFlow 的记忆系统以 Fact 为最小单位，通过 MemoryUpdater 实现 LLM 提取→置信度过滤→去重→容量控制→上传清洗的完整管道。每个 Agent 拥有独立的记忆空间（per-agent isolation），记忆通过 MemoryMiddleware 在对话前自动注入、对话后异步更新。

1/6

Key Source Code

MemoryUpdater._apply_updates — LLM 提取 Fact 后的入库逻辑：置信度过滤 + 去重 + 容量淘汰

agents/memory/updater.py

def _apply_updates(self, current_memory, update_data, thread_id=None):
    config = get_memory_config()
    # 添加新 Fact：置信度必须 >= 阈值
    for fact in update_data.get("newFacts", []):
        confidence = fact.get("confidence", 0.5)
        if confidence >= config.fact_confidence_threshold:
            # 内容去重：相同内容不重复入库
            fact_key = _fact_content_key(normalized_content)
            if fact_key in existing_fact_keys:
                continue
            current_memory["facts"].append(fact_entry)
    # 容量控制：超出 max_facts 时按置信度排序淘汰
    if len(current_memory["facts"]) > config.max_facts:
        current_memory["facts"] = sorted(
            current_memory["facts"],
            key=lambda f: f.get("confidence", 0),
            reverse=True
        )[:config.max_facts]
    return current_memory

python

_strip_upload_mentions_from_memory — 自动清洗上传事件，防止 Agent 在未来会话中搜索已不存在的文件

agents/memory/updater.py

_UPLOAD_SENTENCE_RE = re.compile(
    r"[^.!?]*\b(?:"
    r"upload(?:ed|ing)?(?:\s+\w+){0,3}\s+(?:file|document|attachment)"
    r"|file\s+upload"
    r"|/mnt/user-data/uploads/"
    r"|<uploaded_files>"
    r")[^.!?]*[.!?]?\s*", re.IGNORECASE)

def _strip_upload_mentions_from_memory(memory_data):
    """上传文件是会话级的，持久化上传事件会导致
    Agent 在未来会话中尝试访问不存在的文件"""
    for section in ("user", "history"):
        for _key, val in memory_data.get(section, {}).items():
            if isinstance(val, dict) and "summary" in val:
                val["summary"] = _UPLOAD_SENTENCE_RE.sub("", val["summary"]).strip()
    memory_data["facts"] = [f for f in memory_data.get("facts", [])
                            if not _UPLOAD_SENTENCE_RE.search(f.get("content", ""))]
    return memory_data

python