项目实训 | Vue 3 + FastAPI | A-Mem 工程化 | 分层检索 | HNSW 风格导航 | 可解释记忆路径

一、第一版的问题

上一篇我写的是把 A-Mem 接入法律文书分析系统。那一版解决了“AI 助手没有长期记忆”的问题:系统能从对话里抽取知识点,保存成 note,并在后续问答中召回。

但接着我发现,仅仅“能记住”还不够。

法律场景里的记忆不是平铺的。用户画像、当前案件事实、历史案件经验、通用法律规则、全局办案流程,它们的作用完全不同。如果只用一个向量检索池去召回,语义相似但层级不合适的内容很容易混进回答。

例如用户问:

这个电脑花屏纠纷下一步怎么走?

系统可能同时召回:

用户偏好低成本维权。
消费者权益保护法支持质量问题退换修。
另一个租房押金案件中交接照片很关键。
当前电脑案件缺少检测报告。

这些内容都可能有用,但它们不是同一种东西。第一条是用户画像,第二条是通用法律,第三条是跨案经验,第四条才是当前案件事实。回答时应该有主次。

所以第二阶段我做的核心优化是:把记忆分层,并且让检索路径可解释。

二、记忆分层:不是所有 note 都应该同权

我给每条记忆增加了 memory_scope 字段,目前主要分为五类:

层级 含义 例子
用户画像 用户长期偏好和习惯 用户希望先给结论,再列证据清单
全局记忆 通用办案流程或系统级经验 先固定事实,再补证据,再匹配法条
跨案经验 从多个案件中沉淀的模式 押金扣减案件中交接照片权重高
通用法律 法条、规则、一般法律依据 消费者可就质量问题主张退换修
历史案件 具体案件事实和证据 某次电脑花屏纠纷缺少检测报告

这一步不是 A-Mem 原论文直接提供的法律层级,而是我根据业务场景加的工程设计。

分层以后,检索不再只是“谁和 query 最像”。系统会先生成一个检索计划,再按层级过滤、加权、合并:

用户问题
  ↓
判断需要哪些记忆层
  ↓
按层检索候选 note
  ↓
结合语义相似度、关键词重叠、importance、confidence 加权
  ↓
合并去重后注入聊天上下文

这样做的目的不是追求算法复杂,而是让法律问答更稳:当前案件事实优先,用户偏好和通用法律作为补充,跨案经验用来提供策略参考。

三、为什么要做 HNSW 风格可视化

普通图谱能告诉我“哪些 note 有边”,但不能回答另一个更关键的问题:

系统为什么会从一堆记忆里找到这条?

用户点击一个具体案件事实时,我希望前端不是只弹出详情,而是展示:

入口点 → 高层索引 → 中间跳转 → L0 具体记忆

这就是我引入 HNSW 风格导航图的原因。

这里必须实事求是地说明:我目前实现的是“面向可视化和解释的 HNSW 风格导航图”,不是完整替代 ChromaDB 的工业级 HNSW 索引,也不是 Faiss 的完整实现。

它做了这些事:

  1. 为每个 note 确定一个 deterministic 的 hnsw_level
  2. 高层节点少,底层节点多。
  3. 每层只连接该层可见节点的近邻。
  4. 每个节点的可视化度数做限制,避免图谱变成一团线。
  5. 搜索时从最高入口点开始,高层 greedy,下到 L0 后做候选扩展。
  6. 点击节点时,后端返回真实 trace,前端按 trace 高亮路径。

它没有做这些事:

  1. 没有完全复刻 HNSW 插入阶段的邻居启发式选择。
  2. 没有替代 ChromaDB 的底层向量索引。
  3. 没有宣称达到 ANN benchmark 的性能指标。
  4. 当前主要服务于小规模记忆图的解释和调试。

我认为这样写更诚实,也更符合这个实训项目的定位。

四、后端图构建

后端入口在 memory_navigation.py。构图过程大致如下:

从 SQLite 读取 amem_notes
  ↓
按 note_id 稳定分配 hnsw_level
  ↓
按层筛选 eligible nodes
  ↓
用向量相似度和词面重叠寻找近邻候选
  ↓
每层加边并限制度数
  ↓
补 L0 连通性,避免底层出现不可达孤点
  ↓
生成前端需要的 nodes / edges / layers / params

几个参数目前是保守设置:

M = 4        # 每层最大度数
M0 = M       # L0 最大度数(这里和高层一致)
EF_SEARCH = 16
HNSW_LEVEL_MULT = 1 / math.log(M)

我故意没有把 M 设得很大。因为前端需要看图,不是只追求召回。边太多会让图变成一张网,用户看不懂。

4.1 层级分配

每个 note 的 hnsw_level 通过 SHA1 哈希确定性分配,确保同一 note 每次构图都在同一层:

def _assign_hnsw_levels(rows, max_level):
    levels = {}
    for row in rows:
        note_id = row["note_id"]
        level = 0
        while level < max_level and _stable_unit(f"{note_id}:{level}") < (1 / M):
            level += 1
        levels[note_id] = level

    # 保证每层至少有一个节点,否则图会断层
    if rows and max(levels.values(), default=0) < max_level:
        promoted = max(rows, key=lambda r: (float(r.get("importance") or 1.0), ...))
        levels[promoted["note_id"]] = max_level
    for level in range(max_level - 1, 0, -1):
        if any(nl == level for nl in levels.values()):
            continue
        candidates = [r for r in rows if levels.get(r["note_id"], 0) < level]
        if candidates:
            promoted = max(candidates, key=lambda r: (float(r.get("importance") or 1.0), ...))
            levels[promoted["note_id"]] = level
    return levels

这里 _stable_unit 把 note_id 的 SHA1 哈希映射到 [0, 1) 区间。每层的晋升概率是 1/M = 0.25,所以大约 25% 的节点能进入 L1,6.25% 进入 L2,以此类推。这个分布近似几何分布,和标准 HNSW 的随机层级分配在统计上一致。

4.2 分层建边

每层只连接该层可见的节点。候选来源有两个:ChromaDB 向量相似度和词面重叠(Jaccard)。

def _add_hnsw_edges(rows, levels, max_level, edges, keys):
    for level in range(max_level, -1, -1):
        eligible = {nid for nid, nl in levels.items() if nl >= level}
        if len(eligible) < 2:
            continue
        degree = {nid: 0 for nid in eligible}
        candidates = _level_candidates(row_by_id, eligible)
        for source, target, score in candidates:
            if degree[source] >= M or degree[target] >= M:
                continue
            _upsert_hnsw_edge(edges, ..., source, target, score, level)
            degree[source] += 1
            degree[target] += 1

从最高层向下建边,高层节点少、边少,底层节点多、边多。同一对节点如果在多层都相邻,会合并成一条带 levels 数组的边。

4.3 L0 连通性修正

后来测试时我发现一个问题:如果只做度数限制,L0 可能出现多个连通块。这样点击某些底层球时,真实检索路径无法从入口点走到它。

def _ensure_l0_connected(rows, edges):
    while True:
        components = _l0_components(row_by_id.keys(), edges)
        if len(components) <= 1:
            return
        main = max(components, key=len)
        rest = [c for c in components if c is not main]
        # 在主连通块和每个孤立块之间找词面最近的一对节点
        best = None
        for component in rest:
            for source in main:
                for target in component:
                    score = _lexical_score(source_row, target_row)
                    if best is None or score > best[2]:
                        best = (source, target, score)
        if not best:
            return
        _upsert_hnsw_edge(edges, ..., best[0], best[1], score, 0)

这是纯工程妥协:通过补桥边把所有连通块连起来,保证 L0 可导航。

五、真实检索路径,而不是前端编线

这一点是我后来特别改的。

一开始前端会根据当前图自己推断一条路径,看起来像“入口点到当前节点”。但这其实不够诚实,因为它不是后端真实检索过程。

现在改成:

点击一个节点
  ↓
前端请求 /api/mmem/nav-path/{note_id}
  ↓
后端使用该 note 的 content_text 作为 query
  ↓
从入口点开始执行 HNSW 风格 trace
  ↓
返回经过的 nodes、edges、是否命中、候选结果
  ↓
前端高亮真实路径

后端 trace 的过程分两段:

高层 greedy 下降:

for level in range(max_level, 0, -1):
    improved = True
    while improved:
        improved = False
        neighbors = _neighbors_at_level(current["id"], adj, node_by_id, level)
        best, best_edge, best_score = current, None, scores.get(current["id"], 0.0)
        for node, edge in neighbors:
            if scores.get(node["id"], 0.0) > best_score:
                best, best_edge, best_score = node, edge, scores.get(node["id"], 0.0)
        if best_edge and best["id"] != current["id"]:
            path_edges.append({**best_edge, "type": "hnsw_search_path"})
            current = best
            improved = True

L0 beam search(优先队列多跳扩展):

ef = max(k, EF_SEARCH)
heap = [(-scores.get(current["id"], 0.0), current["id"])]
result_set = {current["id"]: scores.get(current["id"], 0.0)}
while heap and len(result_set) < ef:
    _neg, cur_id = heapq.heappop(heap)
    for node, edge in _neighbors_at_level(cur_id, adj, node_by_id, 0):
        nid = node["id"]
        if nid not in l0_visited:
            l0_visited.add(nid)
            result_set[nid] = scores.get(nid, 0.0)
            heapq.heappush(heap, (-scores.get(nid, 0.0), nid))

L0 用优先队列做多跳扩展(与标准 HNSW ef_search 一致),不是只看落点的 1-hop 邻居。搜索和 trace 都通过预建的 adjacency[level][node_id] 索引做 O(度数) 查找,避免线性扫描全部边。

前端展示的是类似这样的路径:

真实 HNSW trace:入口点 -> L2 -> L0 -> L0 -> L0 -> 命中当前节点

这比单纯显示“相关 note 列表”更直观。用户能看到系统是从哪个入口点开始,经过哪些中间记忆,最后找到当前记忆。

六、API 接口

后端在 mmem.py 注册了三个 HNSW 相关路由:

@router.get("/nav-graph")
def nav_graph(session_id=None, memory_scope=None, limit=200):
    return memory_navigation.build_nav_graph(limit=limit, session_id=session_id, memory_scope=memory_scope)

@router.get("/nav-path/{note_id}")
def nav_path(note_id: str, session_id=None, memory_scope=None, limit=300):
    return memory_navigation.nav_path_to_note(note_id, session_id=session_id, memory_scope=memory_scope, limit=limit)

@router.post("/search")
def search(req: SearchRequest):
    if req.strategy == "layered_nav":
        return memory_navigation.layered_nav_search(req.query, k=req.k, session_id=req.session_id)
  • /nav-graph 返回完整的 HNSW 图(nodes, edges, layers, params),前端 3D 渲染用。
  • /nav-path/{note_id} 对指定 note 执行真实 HNSW trace,返回路径高亮数据。
  • /search 支持 strategy="layered_nav",用 HNSW 路由 + ChromaDB 向量检索融合。

七、真实数据:不再靠前端手写 demo

这次我也修正了一个展示层面的虚假感:如果前端图谱里的 note 都是手写 demo,那它再漂亮也不能说明系统真的能抽取记忆。

所以我做了一个真实生成和抽取链路:

  1. 让大模型生成一个同一用户、一个月、多事件的法律咨询剧本。
  2. 事件包括电脑花屏、租房押金、课程退款、加班费、维修费等。
  3. 每一轮对话都走真实的 extract_and_store_knowledge
  4. 每条 note 写入 SQLite,并带上 source_type=chat_story 和具体 source_id
  5. 过程文件保存到 backend/data/mmem_showcase_runs/

过程文件包括:

00_api_research.md
01_story_generation_request.json
02_story_generation_response.json
03_story_exchanges.json
turns/*.json
90_stored_notes.json
91_graph_summary.json

这样做的好处是:前端看到的月度记忆图,不是我手写出来的假数据,而是系统实际抽取、入库、构图之后的结果。

当然,这里的“真实”也要说明边界:剧情本身是模型生成的,不是真实用户隐私数据;但抽取、存储、构图、检索路径都走的是系统真实链路。

八、前端交互优化

前端页面主要做了几件事:

  1. 左侧从普通统计面板改成“月度剧情索引”,按事件筛选。
  2. 3D 图按 HNSW 层展示节点。
  3. 点击节点后右侧展示内容、层级、来源、source_id。
  4. 同时请求真实检索路径,并高亮 trace。
  5. 打开右侧详情栏时,3D 图主体会向左移动,避免被详情栏盖住。

这里有一个容易误解的点:有些球看起来没有连线,并不代表后端没有边。选中节点后,非路径边会被降透明度,所以视觉上可能像断开。后端当前图里没有孤立节点,每个节点至少有连接。

另外,同级节点之间有边是正常的。HNSW 每层本来就是该层节点之间的近邻图;跨层下降是搜索过程,不一定表现为每个球都有一条竖线连接上下层。

九、现在的局限

这次优化后,系统更像一个可解释的长程记忆系统,但它仍然不是完美版本。

当前局限包括:

  1. HNSW 图是可视化和解释层,不是工业级 ANN 索引替代。
  2. hnsw_level 是确定性分配,不等同于完整 HNSW 插入算法里的随机层级过程。
  3. L0 为了可达性做了连通性修正,这更偏工程实用。
  4. 记忆抽取仍依赖 LLM,可能漏抽或误抽。
  5. 分层规则还比较粗,需要更多真实法律任务验证。
  6. 图谱现在适合几十到几百条 note,规模更大时需要分页、聚类或按需加载。

我觉得这些限制必须写出来。否则博客会变成“包装算法”,而不是工程复盘。

十、这次优化带来的变化

第一阶段的系统可以回答:

我记住了什么?

第二阶段希望进一步回答:

这条记忆属于哪一层?
它为什么被召回?
系统是沿着什么路径找到它的?
这条路径里哪些是用户画像,哪些是法律规则,哪些是案件事实?

这对法律助手很重要。法律问答不是闲聊,用户需要知道结论从哪里来。记忆系统如果只是黑盒召回,很难让用户信任;但如果能展示路径,哪怕只是工程近似,也能帮助调试和解释。

十一、总结

这次分层优化让我重新理解了长程记忆系统:

  1. A-Mem 解决的是“记忆如何被组织成 note 网络”。
  2. 法律业务分层解决的是“不同类型记忆如何承担不同角色”。
  3. HNSW 风格导航解决的是“系统如何解释自己找到某条记忆”。
  4. 真实抽取链路解决的是“展示数据是否可信”。

如果说上一篇是让系统从 0 到 1 拥有记忆,那么这一篇就是让记忆从“能存能搜”走向“有层级、有路径、可解释”。

这还不是最终答案,但已经比第一版更接近我想要的法律 AI 助手:它不只是回答问题,也能说明自己依据了哪些长期记忆,以及这些记忆之间是如何连接的。

参考

  • A-Mem: Agentic Memory for LLM Agents, NeurIPS 2025 Main Conference Track
  • HNSW: Efficient and robust approximate nearest neighbor search using Hierarchical Navigable Small World graphs
Logo

有“AI”的1024 = 2048,欢迎大家加入2048 AI社区

更多推荐