在 LLM Agent 的记忆设计里分层图记忆Hierarchical Graph Memory正成为解决长程任务遗忘和错误路径累积的一种重要思路。尤其是当 Agent 需要执行多轮工具调用、不断观察环境并调整计划时把每轮交互都直接塞进上下文既昂贵又容易丢失关键信息。更有效的做法是把历史轨迹组织成一张层次化图在需要时通过路径级定位Path-level Localization找出相关经验再借助重写Rewrite机制修正已过时或错误的记忆。下面会先解释分层图记忆的出发点再给出一个最小可运行实现并整理定位、重写和运维排错的关键点。读完可以直接在自己的 Agent 项目中搭建一版可扩展的记忆模块。1. 先理解 Agent 记忆为什么不能只靠“塞上下文”1.1 上下文窗口的瓶颈每个大语言模型都有固定的上下文窗口长度。表面上看窗口足够容纳几十轮对话但真实 Agent 任务并不只有对话。一次订单查询可能需要调用身份校验、订单服务、物流服务三个工具每个工具返回一大段 JSON 或日志一次数据修复任务可能包含 SQL 执行、结果校验、异常重试多个环节。把这些中间产物全部叠加后很短时间内就会触达窗口上限。更隐蔽的问题是早期关键信息会被淹没在长文本中。模型在处理靠后的 token 时注意力会分散对前面的重要结论记忆减弱。结果就是 Agent 反复查询同一个订单或者遗漏了已经确认过的约束条件。成本方面每次请求都携带全部历史token 费用和端到端延迟都会线性增长。实际项目中简单做法是把最近几轮对话加上一段摘要放入上下文。这个方案在小任务下够用但一旦任务需要跨会话记忆比如“昨天跑批失败的原因是什么”就无法回答因为数据已经被截断。上下文不是数据库它只是当前工作内存。真正稳定的记忆必须落到 Agent 外部的结构化存储中。1.2 从“关键词检索”到“结构化记忆”向量检索和关键词检索解决了“找到相关片段”的问题。但它们返回的是文本块不是可推理的结构。Agent 需要判断“如果执行 A 后出现 B下一步应该做 C”这个顺序关系在文档检索里是分散的。如果只把相似的历史对话片段贴进提示词模型很难理解完整链路。结构化记忆则是把实体、动作、结果和它们之间的关系显式保存。比如“用户登录失败 - 锁定账号 - 需要重置密码”是一条因果路径。保存路径的好处是下次遇到类似问题可以直接把整条路径送入提示词模型能快速复用而不是从一堆对话里重新归纳。对于 Agent 来说路径比片段更有价值因为路径本身就代表了“怎么做”。1.3 分层图记忆的核心思想分层图记忆将知识划分为多个层次。底层是实体和事件节点中间层是任务或会话对应的操作路径顶层是路径与目标之间的语义关系。图结构天然适合表达“节点-边-路径”路径级定位Path-level Localization就是在图中搜索与当前目标最匹配的路径而不是只检索单个节点。重写Rewrite则是在路径被验证为失败或过时后对该路径进行修改、分支或标注避免下次重复犯错。这种设计解决了两个常见问题一是记忆碎片化二是历史路径固化。碎片化让 Agent 看不到全貌固化则让错误经验反复影响后续决策。图的分层结构允许在实体层保留事实在路径层保留流程在语义层做动态匹配三者配合才能支撑长程 Agent 行为。注意不要简单地把分层图记忆理解成“知识图谱 向量数据库”。知识图谱强调实体关系向量数据库强调相似检索而分层图记忆更关心“路径”这个粒度。路径代表了 Agent 行为的时序和因果这是两者难以直接提供的。2. 设计分层图记忆的三层结构2.1 实体层节点存储什么信息实体层是图的基础。节点可以表示一个实体用户、商品、订单、一个动作调用 API、查询数据库或一个事件登录失败、超时。每个节点至少包含类型和内容。示例节点表示{ node_id: order_query_001, type: action, content: 调用订单查询接口, params: {order_id: A1001}, observation: 返回状态已发货, timestamp: 2025-01-10T10:00:00Z }在图中节点是粗粒度的记忆单元。实际设计里不要把每一步底层 API 调用都建模成节点否则图会迅速膨胀。推荐把“有意义的决策点”建模成节点例如“是否重试”“是否切换方案”“是否结束”而把普通日志写在节点的属性里。这样路径才具有可读性。2.2 路径层会话轨迹如何映射成图路径路径层把一次完整任务抽象为节点序列和边。路径不仅保存节点还保存边的方向、触发条件和结果标签。一次 Agent 执行可以产生多条路径比如“成功路径”“失败路径”“重试路径”。路径记录结构{ path_id: task_001_v1, goal: 查询订单状态并通知用户, nodes: [order_query_001, result_check_001, notify_user_001], edges: [ {from: order_query_001, to: result_check_001, condition: api_return_ok}, {from: result_check_001, to: notify_user_001, condition: status_shipped} ], tags: [order, query, notify], result: success, reward: 1.0 }路径层的核心价值是保留执行顺序。模型在提示词中看到路径后能直接理解“先做什么、再判断什么、最后输出什么”这比零散的记忆片段更接近任务执行本身。路径层还可以进一步分层高层子任务路径、低层工具调用路径两者通过边连接形成真正的分层结构。2.3 语义层路径级定位与重写在哪里发生语义层不在图里存另一个大网络而是维护路径索引和匹配规则。它负责回答两个问题当前目标应该复用哪条路径这条路径是否值得重写当 Agent 接到新任务时语义层会把任务目标转成查询特征用路径级定位找到候选路径。如果候选路径结果标签是“failure”或与当前环境冲突就会进入重写流程。重写流程会产生新路径并把旧路径标记为历史版本。因此语义层更像是一个“记忆调度器”它管理路径的定位、验证和演化。三层之间可以这样对应层次数据单元解决的问题典型存储实体层节点保存实体、动作、事件的事实属性图数据库节点路径层路径保存任务执行顺序和因果关系边序列、路径表语义层路径索引与规则定位相似路径、触发重写向量索引、倒排索引这个分层带来的最大好处是可以单独扩展某一层。比如实体层增加新节点不影响路径层结构语义层更换相似度模型也不需要重建整个图。生产环境中这种解耦非常重要。3. 路径级定位如何从图里找到需要重写的轨迹3.1 路径的定义与生成方式路径的生成主要来自 Agent 执行日志。每次执行完成后记录生成一条 PathRecord。生成规则可以简单也可以复杂简单方式把一次任务从开始到结束的“决策节点”按顺序连成路径。复杂方式从日志里用模式识别切割出多个子任务把每个子任务作为子路径再通过边组成层级路径。路径定义需要包含几个关键字段节点序列、边序列、目标、标签、结果。字段越完整后续定位越准确但采集成本也越高。实际项目建议先记录最小字段goal、nodes、tags、result后续再补充 condition 和 reward。3.2 路径相似度计算路径级定位不直接比较节点文本而是比较路径的整体特征。常用三种方式标签重叠用 Jaccard 系数计算查询标签和路径标签的重合度。实现简单适合冷启动。向量余弦把路径描述通过 embedding 模型转成向量计算余弦相似度。能捕获语义相近但关键词不同的路径。结构相似度比较节点顺序、边跳转是否一致。适合对执行顺序敏感的任务。实践中通常组合使用。例如先用标签过滤再用向量排序最后用结构约束验证。下表给出各有适用场景方法计算成本擅长场景缺点Jaccard 标签低标签规范、关键词稳定对语义泛化不敏感Embedding 余弦中问题表达多样、同义改写需要额外模型可能忽略顺序结构匹配高严格流程、依赖顺序实现复杂召回容易偏窄3.3 路径定位算法伪代码下面是一段路径定位的伪代码展示“先粗筛再重排”的思路def locate_paths(memory, query_goal, query_tags, top_k3): # 1. 标签粗筛 candidates [] for path in memory.paths.values(): overlap len(set(query_tags) set(path.tags)) / max(1, len(set(query_tags) | set(path.tags))) if overlap 0.2: candidates.append((path, overlap)) # 2. 语义重排 query_vec embed(query_goal) scored [] for path, overlap in candidates: path_vec path.vector if path.vector else embed( .join(path.nodes)) semantic_score cosine(query_vec, path_vec) scored.append((path, 0.4 * overlap 0.6 * semantic_score)) scored.sort(keylambda x: x[1], reverseTrue) return scored[:top_k]这段代码中的embed函数在真实项目里可以是 OpenAI Embedding 接口、本地 sentence-transformers 模型也可以是简单的 TF-IDF。关键点是定位结果是“路径集合”而不是“文本片段”后续可以整体送入提示词也可以交给重写流程。3.4 定位结果如何验证路径定位的准确性需要通过实验验证。一个简单做法是准备一批历史任务把新任务作为查询看返回的前三条路径是否包含实际成功路径。也可以记录定位后 Agent 执行的成功率。如果定位一直召回失败路径优先检查路径采集是否完整、标签是否准确、向量索引是否更新。在实际项目中可以增加一个“线上抽样评估”流程。每 100 次 Agent 任务中随机抽取 10 次记录下定位返回的路径和 Agent 最终结果。每周做一次复盘将失败样本补充到重写流程中。这个闭环比单纯调相似度阈值更有价值。3.5 增量更新与冷启动策略路径定位并不是一次构建永久使用。每次新路径加入时需要做两件事更新路径索引更新节点图。增量更新可以放在定时任务里也可以在add_path之后同步执行。冷启动是另一个容易被忽略的问题。系统刚上线时图中没有历史路径定位模块无法召回任何经验。此时可以准备一批规则模板路径比如“遇到超时先重试一次再降级兜底”。这些模板路径的 reward 设为 0只在无其他可用路径时才被召回。随着真实任务执行模板路径逐渐被真实路径替代。4. 重写机制定位之后如何更新记忆4.1 重写的输入与输出重写是路径级记忆闭环的关键。当 Agent 执行失败或发现环境变化导致旧路径不再适用时就要触发重写。重写输入包括当前路径 id新任务目标或新观察失败原因 / 风险评估可选的父路径引用输出建议使用结构化 JSON便于解析和校验{ action: create_rewrite, new_nodes: [start, validate_cache, call_api, fallback, answer], new_edges: [ {from: start, to: validate_cache}, {from: validate_cache, to: call_api, condition: cache_miss} ], new_tags: [cache, fallback], reason: 旧路径未考虑缓存命中场景增加缓存校验步骤, keep_old: true }这个 JSON 结构并不需要 LLM 一次性输出完整字段。如果担心模型生成不稳定可以让它逐步输出先生成节点列表再生成边最后生成 reason。每步都做校验降低失败概率。4.2 保留原路径还是生成新路径重写时最需要决策的是“原地修改”还是“创建分支”。原地修改会让图变小但会丢失历史经验创建分支会保留新旧路径但需要处理路径版本和淘汰策略。推荐默认采用“新版本分支”策略原路径标记为旧版本并降低 reward新路径继承旧路径的 goal并关联 parent_id。这样即使新路径失败也能回滚到旧路径。对于反复失败的新路径可以设置最大版本数超过后淘汰整条轨迹。4.3 用 LLM 调用实现 Rewrite 的示例下面的示例展示如何让 LLM 根据失败路径生成新路径。这里用了一个模拟函数代替真实 API 调用落地时替换为实际模型接口即可。def rewrite_path_with_llm(memory, path_id, feedback): path memory.paths[path_id] prompt f 你是一个Agent记忆重写器。基于以下旧路径和失败反馈生成新路径。 goal: {path.goal} tags: {path.tags} nodes: {path.nodes} feedback: {feedback} 输出JSONnew_nodes, new_edges, new_tags, reason, keep_old response call_llm(prompt) # 实际项目中返回真实LLM输出 data parse_json(response) if new_nodes not in data: raise ValueError(LLM 输出缺少 new_nodes 字段) new_path_id memory.add_path_from_llm(path, data) return new_path_id这个流程中的add_path_from_llm会创建一个新 PathRecord同时旧路径的reward减一。这样后续定位时旧路径的排序会自然下降。要注意给 LLM 输出增加 JSON schema 校验和重试机制否则一次格式错误可能导致整个流程中断。4.4 防止重写风暴的守卫逻辑重写机制一旦失控会产生大量重复路径消耗存储和模型调用。建议在重写入口增加守卫逻辑def safe_rewrite(memory, path_id, feedback, max_versions3): path memory.paths[path_id] if path.version max_versions: return None # 不再重写转人工处理 if path.reward 0.5: return None # 成功路径不参与重写 return rewrite_path_with_llm(memory, path_id, feedback)此外要记录每条路径的rewrite_count。如果连续三次重写都没有让 Agent 执行成功就应该停止自动重写把问题转给人工分析。重写风暴的常见信号是路径版本号快速增加、reason 内容高度相似、图节点数量短时暴涨。监控这三个指标能提前发现异常。5. 用不到 100 行 Python 跑通路径定位与重写闭环5.1 环境准备与依赖为了跑通示例使用 Python 3.10 以上版本主要依赖 networkx 和 numpy。安装命令pip install networkx numpy如果希望使用真实 embedding可以额外安装 sentence-transformers但下面示例使用简单的标签特征和节点重合度不依赖外部模型方便验证核心流程。真实项目里再替换为向量模型。5.2 定义路径数据结构首先定义 PathRecord 数据结构from dataclasses import dataclass, field from typing import List, Tuple, Optional dataclass class PathRecord: path_id: str goal: str nodes: List[str] edges: List[Tuple[str, str]] field(default_factorylist) tags: List[str] field(default_factorylist) result: str success reward: float 0.0 version: int 1 parent_id: Optional[str] None然后创建记忆类底层用networkx.MultiDiGraph保存节点和边import networkx as nx class HierarchicalGraphMemory: def __init__(self): self.graph nx.MultiDiGraph() self.paths {} def add_path(self, path: PathRecord): self.paths[path.path_id] path for i, node_id in enumerate(path.nodes): self.graph.add_node(node_id, typeentity) if i 0: self.graph.add_edge(path.nodes[i-1], node_id, path_idpath.path_id)5.3 实现路径定位与重写流程定位函数采用标签 Jaccard 和节点顺序重合度结合def locate_paths(memory, query_tags, top_k2): results [] for path in memory.paths.values(): tag_overlap len(set(query_tags) set(path.tags)) / max(1, len(set(query_tags) | set(path.tags))) node_overlap len(set(query_tags) set(path.nodes)) / max(1, len(set(query_tags) | set(path.nodes))) score 0.6 * tag_overlap 0.4 * node_overlap results.append((path, score)) results.sort(keylambda x: x[1], reverseTrue) return results[:top_k]重写函数模拟 LLM 输出实际项目中替换为真实模型调用import json def rewrite_path(memory, path_id, feedback): old memory.paths[path_id] # 实际项目中替换为 call_llm(prompt) raw_llm json.dumps({ new_nodes: [start, check_cache, call_api, answer], new_tags: [cache, api], reason: add cache check, keep_old: True }) data json.loads(raw_llm) new_path PathRecord( path_idf{path_id}_v{old.version1}, goalold.goal, nodesdata[new_nodes], edges[(data[new_nodes][i-1], data[new_nodes][i]) for i in range(1, len(data[new_nodes]))], tagsdata[new_tags], resultpending, reward0.1, versionold.version 1, parent_idpath_id ) memory.add_path(new_path) old.reward - 0.5 # 降低旧路径权重 return new_path.path_id5.4 运行验证与结果分析下面在主流程中建立两条历史路径并执行定位和重写memory HierarchicalGraphMemory() memory.add_path(PathRecord( path_idtask_order_v1, goal查询订单状态, nodes[query, parse_result, answer], tags[order, query], reward1.0 )) memory.add_path(PathRecord( path_idtask_refund_v1, goal申请退款, nodes[login, check_refund_policy, apply_refund], tags[refund, apply], reward0.5 )) print(定位查询查询订单) candidates locate_paths(memory, [order, query]) for path, score in candidates: print(f{path.path_id} score{score:.2f}) print(重写 task_order_v1) new_id rewrite_path(memory, task_order_v1, 缺少缓存检查导致重复查询) print(新路径:, new_id) print(旧路径 reward:, memory.paths[task_order_v1].reward)预期输出定位查询查询订单 task_order_v1 score0.60 task_refund_v1 score0.00 重写 task_order_v1 新路径: task_order_v1_v2 旧路径 reward: 0.5运行后可以看到路径定位能按标签召回订单路径重写后新路径被加入图旧路径权重下降。这说明记忆模块具备“按路径召回”和“修改路径经验”的基本能力。真实项目只需要把模拟 LLM 替换为真实接口并把PathRecord持久化到图数据库。6. 先排定位不准、重写循环和检索变慢这三个问题6.1 定位结果不准现象查询“用户订单超时怎么办”时召回的是退款路径而不是重试路径。原因可能包括路径标签设置太粗embedding 模型未针对业务微调阈值过低导致无关路径也进入排序路径采集遗漏了关键节点。排查方式打印候选路径的分数构成查看是标签重合度低还是向量分数低。检查实际历史路径中是否记录了“超时重试”这条完整流程。如果没有问题出在路径采集环节而不是定位算法。解决方案先补全路径再调整特征权重。可以在定位后增加一个 LLM 校验步骤让模型从候选路径中选择最相关的一条过滤掉低质量候选。6.2 重写后路径发散或循环现象Agent 每次失败都把路径重写一次不断产生新版本但一直没有改进。原因通常是终止条件缺失或者失败反馈没有真正影响路径的 reward。模型可能每次生成略不同的路径但核心问题未解决导致反复重写。排查方式 LLM Agent分层图记忆：路径级定位与重写机制实战
在 LLM Agent 的记忆设计里分层图记忆Hierarchical Graph Memory正成为解决长程任务遗忘和错误路径累积的一种重要思路。尤其是当 Agent 需要执行多轮工具调用、不断观察环境并调整计划时把每轮交互都直接塞进上下文既昂贵又容易丢失关键信息。更有效的做法是把历史轨迹组织成一张层次化图在需要时通过路径级定位Path-level Localization找出相关经验再借助重写Rewrite机制修正已过时或错误的记忆。下面会先解释分层图记忆的出发点再给出一个最小可运行实现并整理定位、重写和运维排错的关键点。读完可以直接在自己的 Agent 项目中搭建一版可扩展的记忆模块。1. 先理解 Agent 记忆为什么不能只靠“塞上下文”1.1 上下文窗口的瓶颈每个大语言模型都有固定的上下文窗口长度。表面上看窗口足够容纳几十轮对话但真实 Agent 任务并不只有对话。一次订单查询可能需要调用身份校验、订单服务、物流服务三个工具每个工具返回一大段 JSON 或日志一次数据修复任务可能包含 SQL 执行、结果校验、异常重试多个环节。把这些中间产物全部叠加后很短时间内就会触达窗口上限。更隐蔽的问题是早期关键信息会被淹没在长文本中。模型在处理靠后的 token 时注意力会分散对前面的重要结论记忆减弱。结果就是 Agent 反复查询同一个订单或者遗漏了已经确认过的约束条件。成本方面每次请求都携带全部历史token 费用和端到端延迟都会线性增长。实际项目中简单做法是把最近几轮对话加上一段摘要放入上下文。这个方案在小任务下够用但一旦任务需要跨会话记忆比如“昨天跑批失败的原因是什么”就无法回答因为数据已经被截断。上下文不是数据库它只是当前工作内存。真正稳定的记忆必须落到 Agent 外部的结构化存储中。1.2 从“关键词检索”到“结构化记忆”向量检索和关键词检索解决了“找到相关片段”的问题。但它们返回的是文本块不是可推理的结构。Agent 需要判断“如果执行 A 后出现 B下一步应该做 C”这个顺序关系在文档检索里是分散的。如果只把相似的历史对话片段贴进提示词模型很难理解完整链路。结构化记忆则是把实体、动作、结果和它们之间的关系显式保存。比如“用户登录失败 - 锁定账号 - 需要重置密码”是一条因果路径。保存路径的好处是下次遇到类似问题可以直接把整条路径送入提示词模型能快速复用而不是从一堆对话里重新归纳。对于 Agent 来说路径比片段更有价值因为路径本身就代表了“怎么做”。1.3 分层图记忆的核心思想分层图记忆将知识划分为多个层次。底层是实体和事件节点中间层是任务或会话对应的操作路径顶层是路径与目标之间的语义关系。图结构天然适合表达“节点-边-路径”路径级定位Path-level Localization就是在图中搜索与当前目标最匹配的路径而不是只检索单个节点。重写Rewrite则是在路径被验证为失败或过时后对该路径进行修改、分支或标注避免下次重复犯错。这种设计解决了两个常见问题一是记忆碎片化二是历史路径固化。碎片化让 Agent 看不到全貌固化则让错误经验反复影响后续决策。图的分层结构允许在实体层保留事实在路径层保留流程在语义层做动态匹配三者配合才能支撑长程 Agent 行为。注意不要简单地把分层图记忆理解成“知识图谱 向量数据库”。知识图谱强调实体关系向量数据库强调相似检索而分层图记忆更关心“路径”这个粒度。路径代表了 Agent 行为的时序和因果这是两者难以直接提供的。2. 设计分层图记忆的三层结构2.1 实体层节点存储什么信息实体层是图的基础。节点可以表示一个实体用户、商品、订单、一个动作调用 API、查询数据库或一个事件登录失败、超时。每个节点至少包含类型和内容。示例节点表示{ node_id: order_query_001, type: action, content: 调用订单查询接口, params: {order_id: A1001}, observation: 返回状态已发货, timestamp: 2025-01-10T10:00:00Z }在图中节点是粗粒度的记忆单元。实际设计里不要把每一步底层 API 调用都建模成节点否则图会迅速膨胀。推荐把“有意义的决策点”建模成节点例如“是否重试”“是否切换方案”“是否结束”而把普通日志写在节点的属性里。这样路径才具有可读性。2.2 路径层会话轨迹如何映射成图路径路径层把一次完整任务抽象为节点序列和边。路径不仅保存节点还保存边的方向、触发条件和结果标签。一次 Agent 执行可以产生多条路径比如“成功路径”“失败路径”“重试路径”。路径记录结构{ path_id: task_001_v1, goal: 查询订单状态并通知用户, nodes: [order_query_001, result_check_001, notify_user_001], edges: [ {from: order_query_001, to: result_check_001, condition: api_return_ok}, {from: result_check_001, to: notify_user_001, condition: status_shipped} ], tags: [order, query, notify], result: success, reward: 1.0 }路径层的核心价值是保留执行顺序。模型在提示词中看到路径后能直接理解“先做什么、再判断什么、最后输出什么”这比零散的记忆片段更接近任务执行本身。路径层还可以进一步分层高层子任务路径、低层工具调用路径两者通过边连接形成真正的分层结构。2.3 语义层路径级定位与重写在哪里发生语义层不在图里存另一个大网络而是维护路径索引和匹配规则。它负责回答两个问题当前目标应该复用哪条路径这条路径是否值得重写当 Agent 接到新任务时语义层会把任务目标转成查询特征用路径级定位找到候选路径。如果候选路径结果标签是“failure”或与当前环境冲突就会进入重写流程。重写流程会产生新路径并把旧路径标记为历史版本。因此语义层更像是一个“记忆调度器”它管理路径的定位、验证和演化。三层之间可以这样对应层次数据单元解决的问题典型存储实体层节点保存实体、动作、事件的事实属性图数据库节点路径层路径保存任务执行顺序和因果关系边序列、路径表语义层路径索引与规则定位相似路径、触发重写向量索引、倒排索引这个分层带来的最大好处是可以单独扩展某一层。比如实体层增加新节点不影响路径层结构语义层更换相似度模型也不需要重建整个图。生产环境中这种解耦非常重要。3. 路径级定位如何从图里找到需要重写的轨迹3.1 路径的定义与生成方式路径的生成主要来自 Agent 执行日志。每次执行完成后记录生成一条 PathRecord。生成规则可以简单也可以复杂简单方式把一次任务从开始到结束的“决策节点”按顺序连成路径。复杂方式从日志里用模式识别切割出多个子任务把每个子任务作为子路径再通过边组成层级路径。路径定义需要包含几个关键字段节点序列、边序列、目标、标签、结果。字段越完整后续定位越准确但采集成本也越高。实际项目建议先记录最小字段goal、nodes、tags、result后续再补充 condition 和 reward。3.2 路径相似度计算路径级定位不直接比较节点文本而是比较路径的整体特征。常用三种方式标签重叠用 Jaccard 系数计算查询标签和路径标签的重合度。实现简单适合冷启动。向量余弦把路径描述通过 embedding 模型转成向量计算余弦相似度。能捕获语义相近但关键词不同的路径。结构相似度比较节点顺序、边跳转是否一致。适合对执行顺序敏感的任务。实践中通常组合使用。例如先用标签过滤再用向量排序最后用结构约束验证。下表给出各有适用场景方法计算成本擅长场景缺点Jaccard 标签低标签规范、关键词稳定对语义泛化不敏感Embedding 余弦中问题表达多样、同义改写需要额外模型可能忽略顺序结构匹配高严格流程、依赖顺序实现复杂召回容易偏窄3.3 路径定位算法伪代码下面是一段路径定位的伪代码展示“先粗筛再重排”的思路def locate_paths(memory, query_goal, query_tags, top_k3): # 1. 标签粗筛 candidates [] for path in memory.paths.values(): overlap len(set(query_tags) set(path.tags)) / max(1, len(set(query_tags) | set(path.tags))) if overlap 0.2: candidates.append((path, overlap)) # 2. 语义重排 query_vec embed(query_goal) scored [] for path, overlap in candidates: path_vec path.vector if path.vector else embed( .join(path.nodes)) semantic_score cosine(query_vec, path_vec) scored.append((path, 0.4 * overlap 0.6 * semantic_score)) scored.sort(keylambda x: x[1], reverseTrue) return scored[:top_k]这段代码中的embed函数在真实项目里可以是 OpenAI Embedding 接口、本地 sentence-transformers 模型也可以是简单的 TF-IDF。关键点是定位结果是“路径集合”而不是“文本片段”后续可以整体送入提示词也可以交给重写流程。3.4 定位结果如何验证路径定位的准确性需要通过实验验证。一个简单做法是准备一批历史任务把新任务作为查询看返回的前三条路径是否包含实际成功路径。也可以记录定位后 Agent 执行的成功率。如果定位一直召回失败路径优先检查路径采集是否完整、标签是否准确、向量索引是否更新。在实际项目中可以增加一个“线上抽样评估”流程。每 100 次 Agent 任务中随机抽取 10 次记录下定位返回的路径和 Agent 最终结果。每周做一次复盘将失败样本补充到重写流程中。这个闭环比单纯调相似度阈值更有价值。3.5 增量更新与冷启动策略路径定位并不是一次构建永久使用。每次新路径加入时需要做两件事更新路径索引更新节点图。增量更新可以放在定时任务里也可以在add_path之后同步执行。冷启动是另一个容易被忽略的问题。系统刚上线时图中没有历史路径定位模块无法召回任何经验。此时可以准备一批规则模板路径比如“遇到超时先重试一次再降级兜底”。这些模板路径的 reward 设为 0只在无其他可用路径时才被召回。随着真实任务执行模板路径逐渐被真实路径替代。4. 重写机制定位之后如何更新记忆4.1 重写的输入与输出重写是路径级记忆闭环的关键。当 Agent 执行失败或发现环境变化导致旧路径不再适用时就要触发重写。重写输入包括当前路径 id新任务目标或新观察失败原因 / 风险评估可选的父路径引用输出建议使用结构化 JSON便于解析和校验{ action: create_rewrite, new_nodes: [start, validate_cache, call_api, fallback, answer], new_edges: [ {from: start, to: validate_cache}, {from: validate_cache, to: call_api, condition: cache_miss} ], new_tags: [cache, fallback], reason: 旧路径未考虑缓存命中场景增加缓存校验步骤, keep_old: true }这个 JSON 结构并不需要 LLM 一次性输出完整字段。如果担心模型生成不稳定可以让它逐步输出先生成节点列表再生成边最后生成 reason。每步都做校验降低失败概率。4.2 保留原路径还是生成新路径重写时最需要决策的是“原地修改”还是“创建分支”。原地修改会让图变小但会丢失历史经验创建分支会保留新旧路径但需要处理路径版本和淘汰策略。推荐默认采用“新版本分支”策略原路径标记为旧版本并降低 reward新路径继承旧路径的 goal并关联 parent_id。这样即使新路径失败也能回滚到旧路径。对于反复失败的新路径可以设置最大版本数超过后淘汰整条轨迹。4.3 用 LLM 调用实现 Rewrite 的示例下面的示例展示如何让 LLM 根据失败路径生成新路径。这里用了一个模拟函数代替真实 API 调用落地时替换为实际模型接口即可。def rewrite_path_with_llm(memory, path_id, feedback): path memory.paths[path_id] prompt f 你是一个Agent记忆重写器。基于以下旧路径和失败反馈生成新路径。 goal: {path.goal} tags: {path.tags} nodes: {path.nodes} feedback: {feedback} 输出JSONnew_nodes, new_edges, new_tags, reason, keep_old response call_llm(prompt) # 实际项目中返回真实LLM输出 data parse_json(response) if new_nodes not in data: raise ValueError(LLM 输出缺少 new_nodes 字段) new_path_id memory.add_path_from_llm(path, data) return new_path_id这个流程中的add_path_from_llm会创建一个新 PathRecord同时旧路径的reward减一。这样后续定位时旧路径的排序会自然下降。要注意给 LLM 输出增加 JSON schema 校验和重试机制否则一次格式错误可能导致整个流程中断。4.4 防止重写风暴的守卫逻辑重写机制一旦失控会产生大量重复路径消耗存储和模型调用。建议在重写入口增加守卫逻辑def safe_rewrite(memory, path_id, feedback, max_versions3): path memory.paths[path_id] if path.version max_versions: return None # 不再重写转人工处理 if path.reward 0.5: return None # 成功路径不参与重写 return rewrite_path_with_llm(memory, path_id, feedback)此外要记录每条路径的rewrite_count。如果连续三次重写都没有让 Agent 执行成功就应该停止自动重写把问题转给人工分析。重写风暴的常见信号是路径版本号快速增加、reason 内容高度相似、图节点数量短时暴涨。监控这三个指标能提前发现异常。5. 用不到 100 行 Python 跑通路径定位与重写闭环5.1 环境准备与依赖为了跑通示例使用 Python 3.10 以上版本主要依赖 networkx 和 numpy。安装命令pip install networkx numpy如果希望使用真实 embedding可以额外安装 sentence-transformers但下面示例使用简单的标签特征和节点重合度不依赖外部模型方便验证核心流程。真实项目里再替换为向量模型。5.2 定义路径数据结构首先定义 PathRecord 数据结构from dataclasses import dataclass, field from typing import List, Tuple, Optional dataclass class PathRecord: path_id: str goal: str nodes: List[str] edges: List[Tuple[str, str]] field(default_factorylist) tags: List[str] field(default_factorylist) result: str success reward: float 0.0 version: int 1 parent_id: Optional[str] None然后创建记忆类底层用networkx.MultiDiGraph保存节点和边import networkx as nx class HierarchicalGraphMemory: def __init__(self): self.graph nx.MultiDiGraph() self.paths {} def add_path(self, path: PathRecord): self.paths[path.path_id] path for i, node_id in enumerate(path.nodes): self.graph.add_node(node_id, typeentity) if i 0: self.graph.add_edge(path.nodes[i-1], node_id, path_idpath.path_id)5.3 实现路径定位与重写流程定位函数采用标签 Jaccard 和节点顺序重合度结合def locate_paths(memory, query_tags, top_k2): results [] for path in memory.paths.values(): tag_overlap len(set(query_tags) set(path.tags)) / max(1, len(set(query_tags) | set(path.tags))) node_overlap len(set(query_tags) set(path.nodes)) / max(1, len(set(query_tags) | set(path.nodes))) score 0.6 * tag_overlap 0.4 * node_overlap results.append((path, score)) results.sort(keylambda x: x[1], reverseTrue) return results[:top_k]重写函数模拟 LLM 输出实际项目中替换为真实模型调用import json def rewrite_path(memory, path_id, feedback): old memory.paths[path_id] # 实际项目中替换为 call_llm(prompt) raw_llm json.dumps({ new_nodes: [start, check_cache, call_api, answer], new_tags: [cache, api], reason: add cache check, keep_old: True }) data json.loads(raw_llm) new_path PathRecord( path_idf{path_id}_v{old.version1}, goalold.goal, nodesdata[new_nodes], edges[(data[new_nodes][i-1], data[new_nodes][i]) for i in range(1, len(data[new_nodes]))], tagsdata[new_tags], resultpending, reward0.1, versionold.version 1, parent_idpath_id ) memory.add_path(new_path) old.reward - 0.5 # 降低旧路径权重 return new_path.path_id5.4 运行验证与结果分析下面在主流程中建立两条历史路径并执行定位和重写memory HierarchicalGraphMemory() memory.add_path(PathRecord( path_idtask_order_v1, goal查询订单状态, nodes[query, parse_result, answer], tags[order, query], reward1.0 )) memory.add_path(PathRecord( path_idtask_refund_v1, goal申请退款, nodes[login, check_refund_policy, apply_refund], tags[refund, apply], reward0.5 )) print(定位查询查询订单) candidates locate_paths(memory, [order, query]) for path, score in candidates: print(f{path.path_id} score{score:.2f}) print(重写 task_order_v1) new_id rewrite_path(memory, task_order_v1, 缺少缓存检查导致重复查询) print(新路径:, new_id) print(旧路径 reward:, memory.paths[task_order_v1].reward)预期输出定位查询查询订单 task_order_v1 score0.60 task_refund_v1 score0.00 重写 task_order_v1 新路径: task_order_v1_v2 旧路径 reward: 0.5运行后可以看到路径定位能按标签召回订单路径重写后新路径被加入图旧路径权重下降。这说明记忆模块具备“按路径召回”和“修改路径经验”的基本能力。真实项目只需要把模拟 LLM 替换为真实接口并把PathRecord持久化到图数据库。6. 先排定位不准、重写循环和检索变慢这三个问题6.1 定位结果不准现象查询“用户订单超时怎么办”时召回的是退款路径而不是重试路径。原因可能包括路径标签设置太粗embedding 模型未针对业务微调阈值过低导致无关路径也进入排序路径采集遗漏了关键节点。排查方式打印候选路径的分数构成查看是标签重合度低还是向量分数低。检查实际历史路径中是否记录了“超时重试”这条完整流程。如果没有问题出在路径采集环节而不是定位算法。解决方案先补全路径再调整特征权重。可以在定位后增加一个 LLM 校验步骤让模型从候选路径中选择最相关的一条过滤掉低质量候选。6.2 重写后路径发散或循环现象Agent 每次失败都把路径重写一次不断产生新版本但一直没有改进。原因通常是终止条件缺失或者失败反馈没有真正影响路径的 reward。模型可能每次生成略不同的路径但核心问题未解决导致反复重写。排查方式