WikiSkill:把Agent执行经验编译为可复用技能库的完整方案
之前开发 Agent 类应用时我一直被一个问题困扰Agent 在每次任务里都会产生大量轨迹、决策记录和试错过程但任务一结束这些经验就基本丢掉了。下一次遇到类似问题模型还是从零开始推理同样的坑还会再踩一遍。后来我尝试把 Agent 的完整执行过程“编译”成结构化知识再沉淀为可复用的技能效果比想象中好不少。本文就把这套方案完整拆开包括概念模型、系统分层、Python 代码实现、运行验证和常见坑点适合正在做 Agent 应用、工具链或知识管理系统的开发者参考。1. 背景Agent 的经验为什么总是“用过即忘”1.1 从一次 Agent 任务说起假设你有一个客服工单分类 Agent它的职责是根据用户描述判断工单类型、紧急程度并推荐处理部门。第一次运行时Agent 可能走了很多弯路先误以为是“账号问题”后来才发现是“支付失败”于是修改推理方向最终完成分类。整个过程中最有价值的信息其实是它从误判到修正的这条路径。但如果你没有做任何经验沉淀下一次它遇到“支付失败但用户先说账号登不上”的场景时又会重复同样的误判。这里面的核心问题不是模型能力不够而是“经验没有变成知识”。1.2 经验、知识与技能的区别很多同学把这三个词混着用但在 WikiSkill 这套体系里它们定义完全不同概念含义生命周期示例经验ExperienceAgent 执行具体任务时产生的原始轨迹包括输入、动作、观察、中间推理、结果短时、碎片化某一次工单分类中 Agent 的完整日志知识Knowledge对经验进行整理、去噪、归纳后的结构化信息脱离具体任务也能独立存在持久、可检索“支付失败类工单常被误判为账号问题”技能Skill将知识进一步提炼为可指导后续任务执行的流程、规则、提示模板或工具调用模式复用、可进化“遇到支付失败描述时优先检查支付网关回调日志再决定派单部门”可以这样理解经验是一次性的原始日志知识是整理后的信息技能是能直接指导下一步行动的能力。1.3 WikiSkill 的闭环思路WikiSkill 要做的就是把“经验 → 知识 → 技能”这条链路自动化并形成闭环执行任务产生经验 → 编译经验生成知识 → 沉淀入库 → 提炼技能 → 下一次任务复用 → 再次产生新经验重点在于“编译”二字。它不是简单把日志存起来而是像编译器把高级语言翻译成可执行代码一样把非结构化的 Agent 轨迹转换成结构化、可被检索和推理的知识条目最终驱动技能进化。2. 总体架构设计2.1 四大核心模块WikiSkill 的架构可以拆成四个部分职责边界非常清晰模块职责输入输出Experience Harvester经验采集器收集 Agent 执行过程中的原始日志、轨迹、决策记录Agent 运行时日志标准化经验记录Knowledge Compiler知识编译器对经验进行摘要、去噪、结构化提取可复用的结论标准化经验记录知识条目Persistent Store持久化存储保存知识条目提供检索和版本管理知识条目可查询的知识库Skill Evolver技能进化器从知识库中归纳、生成、评估技能模板知识库条目可复用技能另外还有一个跨模块组件Evaluator评估器。它负责监控技能在下一轮任务中的表现把结果反馈给编译器形成闭环。2.2 数据流转模型整个系统的数据流转可以用下面这条线表示Agent Log → Raw Experience → Normalized Experience → Knowledge Entry → Skill Template → Task Guidance每一步都是一次“压缩”和“提纯”Agent Log可能是 JSON Lines、控制台输出或其他格式。Raw Experience保留完整轨迹但不适合直接入库。Normalized Experience统一字段结构如 task_type、objective、steps、result。Knowledge Entry去掉冗余提炼结论附加上下文标签。Skill Template把知识组织成可执行的指导流程。Task Guidance实际注入到下一轮 Agent 输入中的内容。2.3 为什么需要持久知识库很多人会问直接把经验丢回 LLM 上下文里不行吗可行但有几个问题上下文长度有限完整轨迹可能超出 Token 限制。原始轨迹噪声太多直接回放会干扰模型判断。经验之间可能存在矛盾不经过编译模型无法判断哪条更可靠。经验是一次性的无法跨任务积累形成长期能力。持久知识库的价值在于它让经验可以被重复检索、比较、归纳最终形成稳定可用的技能而不是一次性的临时记忆。3. 环境准备与项目结构3.1 运行环境与版本说明本文示例使用 Python 编写版本需要根据你的项目实际情况调整下面以常见环境为例Python 3.9 及以上操作系统Windows / macOS / Linux 均可依赖库PyYAML、Jieba用于中文文本分词、Numpy如果只是跑通演示逻辑不需要接入真实大模型。本文会把“调用大模型”抽象成一个函数接口方便替换成你自己的模型网关。3.2 项目目录结构wikiskill-demo/ ├── data/ │ ├── raw_logs/ # 原始 Agent 日志 │ ├── knowledge/ # 编译后的知识条目 │ └── skills/ # 生成的技能模板 ├── wikiskill/ │ ├── __init__.py │ ├── models.py # 数据模型定义 │ ├── harvester.py # 经验采集模块 │ ├── compiler.py # 知识编译模块 │ ├── storage.py # 持久化存储模块 │ └── evolver.py # 技能进化模块 ├── main.py # 主流程入口 └── requirements.txt下面我们先实现核心模块再通过一个模拟案例串联整个流程。3.3 数据格式约定我们约定 Agent 原始日志使用 JSON Lines 格式每行包含以下字段{ trace_id: 任务唯一标识, task_type: 任务类型, objective: 任务目标描述, timestamp: 执行时间, steps: [ { action: Agent 执行的动作, observation: 执行后的观察结果, thinking: 中间推理如果有, status: success / error / retry } ], result: 最终结果, success: true }这样设计的好处是Agent 日志来源可以多种多样只要在采集层转换成统一格式编译器就能以相同逻辑处理。4. 核心代码实现4.1 数据模型定义先定义系统内部的数据模型。文件路径wikiskill/models.pyfrom dataclasses import dataclass, field from typing import List, Optional dataclass class StepRecord: action: str observation: str thinking: str status: str success # success / error / retry dataclass class ExperienceRecord: trace_id: str task_type: str objective: str timestamp: str steps: List[StepRecord] result: str success: bool dataclass class KnowledgeEntry: entry_id: str task_type: str title: str content: str tags: List[str] field(default_factorylist) source_trace_ids: List[str] field(default_factorylist) confidence: float 0.5 dataclass class SkillTemplate: skill_id: str name: str description: str trigger_conditions: List[str] field(default_factorylist) procedure: List[str] field(default_factorylist) examples: List[str] field(default_factorylist) source_knowledge_ids: List[str] field(default_factorylist)这些模型是整个系统的基础。后面所有模块都围绕这几个数据结构展开。4.2 经验采集把原始日志转成标准化记录文件路径wikiskill/harvester.pyimport json from typing import List, Dict, Any from .models import ExperienceRecord, StepRecord class ExperienceHarvester: 从原始日志中采集并标准化经验数据。 def load_jsonl(self, file_path: str) - List[Dict[str, Any]]: 读取 JSON Lines 文件。 records [] with open(file_path, r, encodingutf-8) as f: for line in f: line line.strip() if not line: continue records.append(json.loads(line)) return records def normalize(self, raw: Dict[str, Any]) - ExperienceRecord: 将原始日志字典转换为标准 ExperienceRecord。 steps [] for step in raw.get(steps, []): steps.append( StepRecord( actionstep.get(action, ), observationstep.get(observation, ), thinkingstep.get(thinking, ), statusstep.get(status, success), ) ) return ExperienceRecord( trace_idraw.get(trace_id, ), task_typeraw.get(task_type, unknown), objectiveraw.get(objective, ), timestampraw.get(timestamp, ), stepssteps, resultraw.get(result, ), successraw.get(success, False), ) def harvest(self, file_path: str) - List[ExperienceRecord]: 采集并标准化一个日志文件中的全部经验。 raw_records self.load_jsonl(file_path) return [self.normalize(r) for r in raw_records]这个模块的逻辑很简单但很重要它解决了日志来源格式不统一的问题。如果你的 Agent 输出不是 JSON可以在 normalize 之前做一层字段映射。4.3 知识编译从轨迹中提炼结论这是 WikiSkill 的核心模块。它的工作有两部分从失败的步骤中提取“错误模式”。从成功的步骤中提取“可用方法”。文件路径wikiskill/compiler.pyimport uuid from typing import List, Dict, Any from .models import ExperienceRecord, KnowledgeEntry class KnowledgeCompiler: 将经验记录编译为结构化知识条目。 def __init__(self, llm_funcNone): # llm_func 是一个可选的函数接口用于调用大模型做摘要 # 实际项目中可以传入自己封装好的模型调用函数 self.llm_func llm_func def _extract_error_patterns(self, exp: ExperienceRecord) - List[str]: 提取错误模式从 status 为 error 的步骤中总结。 patterns [] for step in exp.steps: if step.status error: patterns.append(f{step.action} 时出现异常观察结果{step.observation}) return patterns def _extract_success_methods(self, exp: ExperienceRecord) - List[str]: 提取成功方法从 status 为 success 的步骤中总结。 methods [] for step in exp.steps: if step.status success and step.observation: methods.append(f{step.action} 后观察{step.observation}) return methods def compile_entry(self, exp: ExperienceRecord) - KnowledgeEntry: 从单条经验编译出一条知识条目。 task_type exp.task_type title f{task_type} 执行经验 # 如果配置了 LLM可以通过模型生成更精炼的摘要 if self.llm_func is not None: content self._generate_content_with_llm(exp) else: # 兜底逻辑拼接错误模式和成功方法 error_part .join(self._extract_error_patterns(exp)) or 无错误路径 success_part .join(self._extract_success_methods(exp)) or 无成功路径 content f经验来源{exp.objective}。错误模式{error_part}。可用方法{success_part}。 entry KnowledgeEntry( entry_idstr(uuid.uuid4()), task_typetask_type, titletitle, contentcontent, tags[task_type, 经验沉淀], source_trace_ids[exp.trace_id], confidence0.8 if exp.success else 0.5, ) return entry def _generate_content_with_llm(self, exp: ExperienceRecord) - str: 调用大模型生成摘要。这只是一个框架示例。 steps_text \n.join( [f动作{s.action}观察{s.observation}状态{s.status} for s in exp.steps] ) prompt f请根据以下 Agent 执行轨迹提炼一条结构化经验。 任务类型{exp.task_type} 任务目标{exp.objective} 执行过程 {steps_text} 最终结果{exp.result} 请输出 1. 这条经验解决什么问题。 2. 有哪些值得注意的错误点。 3. 下次执行可以采用的策略。 return self.llm_func(prompt)这段代码体现了“编译”的基本思想不直接把原始轨迹塞进知识库而是先抽取模式再组织成可读、可复用的表述。4.4 持久化存储与检索文件路径wikiskill/storage.pyimport json import os from typing import List, Optional from .models import KnowledgeEntry, SkillTemplate class PersistentStore: 基于本地 JSON 文件的持久化存储便于演示和调试。 def __init__(self, knowledge_dir: str, skills_dir: str): self.knowledge_dir knowledge_dir self.skills_dir skills_dir os.makedirs(knowledge_dir, exist_okTrue) os.makedirs(skills_dir, exist_okTrue) def save_knowledge(self, entry: KnowledgeEntry) - str: 保存知识条目。 file_path os.path.join(self.knowledge_dir, f{entry.entry_id}.json) with open(file_path, w, encodingutf-8) as f: json.dump(entry.__dict__, f, ensure_asciiFalse, indent2) return file_path def save_skill(self, skill: SkillTemplate) - str: 保存技能模板。 file_path os.path.join(self.skills_dir, f{skill.skill_id}.json) with open(file_path, w, encodingutf-8) as f: json.dump(skill.__dict__, f, ensure_asciiFalse, indent2) return file_path def load_all_knowledge(self) - List[KnowledgeEntry]: 加载全部知识条目。 entries [] if not os.path.isdir(self.knowledge_dir): return entries for file_name in os.listdir(self.knowledge_dir): if not file_name.endswith(.json): continue with open(os.path.join(self.knowledge_dir, file_name), r, encodingutf-8) as f: data json.load(f) entries.append(KnowledgeEntry(**data)) return entries def search_knowledge(self, query: str, top_k: int 5) - List[KnowledgeEntry]: 基于简单关键词匹配的知识检索。 生产环境中建议替换为向量检索方案比如使用向量数据库。 entries self.load_all_knowledge() scored [] for entry in entries: score 0 query_lower query.lower() if query_lower in entry.content.lower(): score len(query_lower) for tag in entry.tags: if tag.lower() in query_lower: score 1 if entry.title and query_lower in entry.title.lower(): score 2 scored.append((score, entry)) scored.sort(keylambda x: x[0], reverseTrue) return [entry for _, entry in scored[:top_k] if _ 0]这里的检索是“关键词匹配”版本演示系统运行的完整链路足够了。真实项目建议换成语义向量检索后面会展开讲。4.5 技能进化从知识条目到可复用技能文件路径wikiskill/evolver.pyfrom typing import List from .models import KnowledgeEntry, SkillTemplate class SkillEvolver: 从知识库中归纳技能模板实现技能进化。 def __init__(self, min_entries: int 2): # 同一任务类型至少需要多少条知识条目才能生成技能 self.min_entries min_entries def evolve(self, entries: List[KnowledgeEntry]) - List[SkillTemplate]: 根据知识条目生成或更新技能模板。 # 按任务类型分组 grouped: dict[str, List[KnowledgeEntry]] {} for entry in entries: grouped.setdefault(entry.task_type, []).append(entry) skills [] for task_type, group in grouped.items(): if len(group) self.min_entries: continue # 经验不足暂不生成技能 # 综合多条知识条目归纳技能流程 procedure [] for entry in group: content entry.content # 简化提取逻辑实际项目中可以结合 LLM 做更精细的归纳 if 错误模式 in content: procedure.append(f先确认是否属于常见误区{content[:80]}...) else: procedure.append(f参考经验{content[:80]}...) skill SkillTemplate( skill_idfskill_{task_type}_{len(group)}, namef{task_type} 标准处理流程, descriptionf该技能由 {len(group)} 条知识条目归纳而来, trigger_conditions[task_type], procedureprocedure, source_knowledge_ids[e.entry_id for e in group], ) skills.append(skill) return skills技能进化不是一次性的。随着知识条目增加同一类任务的技能模板会不断更新这就是“进化”的含义。下一轮任务结束后系统会重新评估技能的命中率和效果决定保留还是覆盖旧版本。4.6 主流程串联文件路径main.pyimport sys from wikiskill.harvester import ExperienceHarvester from wikiskill.compiler import KnowledgeCompiler from wikiskill.storage import PersistentStore from wikiskill.evolver import SkillEvolver def llm_demo(prompt: str) - str: 演示用的大模型调用函数实际项目请替换为真实模型网关。 # 这里不做真实调用只返回固定的框架结果 return [LLM 摘要] 该任务的关键经验是优先检查异常步骤确认错误模式。 def main(): if len(sys.argv) 2: print(用法python main.py agent_log_file) return log_file sys.argv[1] # 1. 采集经验 harvester ExperienceHarvester() experiences harvester.harvest(log_file) print(f[1/4] 采集到 {len(experiences)} 条经验记录) # 2. 编译知识 compiler KnowledgeCompiler(llm_funcllm_demo) store PersistentStore(data/knowledge, data/skills) entries [] for exp in experiences: entry compiler.compile_entry(exp) store.save_knowledge(entry) entries.append(entry) print(f[2/4] 编译并保存 {len(entries)} 条知识条目) # 3. 加载知识库 knowledge_list store.load_all_knowledge() print(f[3/4] 当前知识库共 {len(knowledge_list)} 条知识) # 4. 技能进化 evolver SkillEvolver(min_entries2) skills evolver.evolve(knowledge_list) for skill in skills: store.save_skill(skill) print(f[4/4] 技能进化完成生成 {len(skills)} 个技能模板) # 展示技能内容 for skill in skills: print(f\n技能名称{skill.name}) print(f触发条件{skill.trigger_conditions}) for step in skill.procedure: print(f - {step}) if __name__ __main__: main()5. 运行与验证5.1 准备模拟 Agent 经验数据在data/raw_logs/task_log.jsonl中放两条模拟记录{trace_id: 001, task_type: 客服工单分类, objective: 判断用户反馈的支付失败属于哪类工单, timestamp: 2025-01-10 10:00:00, steps: [{action: 分析用户描述, observation: 用户说账号登录不上, thinking: 可能是账号问题, status: error}, {action: 查询支付日志, observation: 发现支付网关回调超时, thinking: 问题出在支付环节, status: success}], result: 支付失败工单, success: true} {trace_id: 002, task_type: 客服工单分类, objective: 判断用户反馈的余额不对属于哪类工单, timestamp: 2025-01-10 10:30:00, steps: [{action: 查询账户流水, observation: 存在未到账交易, thinking: 可能涉及支付渠道, status: success}, {action: 核对订单状态, observation: 订单显示支付成功但余额未更新, thinking: 需要触发对账流程, status: success}], result: 支付对账工单, success: true}5.2 执行主流程在项目根目录运行python main.py data/raw_logs/task_log.jsonl预期输出如下[1/4] 采集到 2 条经验记录 [2/4] 编译并保存 2 条知识条目 [3/4] 当前知识库共 2 条知识 [4/4] 技能进化完成生成 1 个技能模板 技能名称客服工单分类 标准处理流程 触发条件[客服工单分类] - 参考经验经验来源判断用户反馈的支付失败属于哪类工单。错误模式分析用户描述 时出现异常... - 参考经验经验来源判断用户反馈的余额不对属于哪类工单。错误模式无错误路径...5.3 验证闭环效果技能生成后下一轮 Agent 在执行类似任务时可以这样使用技能根据技能模板“客服工单分类 标准处理流程” 1. 如果用户描述中存在“支付”、“余额”、“订单”相关关键词优先检查支付侧日志。 2. 不要仅凭“登录不上”判断为账号问题先确认支付网关状态。这样一来Agent 的初始推理方向就从“从零猜测”变成了“依据沉淀经验”命中率会明显提升这也是技能进化的直接收益。6. 常见问题与排查思路问题现象常见原因解决思路采集阶段读不到日志文件路径配置错误或日志编码不是 UTF-8检查文件路径统一使用 UTF-8 编码编译出的知识内容过于碎片化日志步骤太细没有经过模式归纳增加 LLM 摘要步骤或按任务阶段合并步骤知识库条目越来越多但检索不准关键词匹配无法表达语义升级为向量检索基于 Embedding 做相似度搜索技能模板重复且互相矛盾相同任务类型的知识条目未去重合并按任务类型分组对相似条目做聚类合并技能进化后效果反而下降低质量经验污染了知识库增加置信度过滤引入人工审核节点存储层在大规模数据下变慢本地 JSON 文件不适合高并发检索迁移到 SQLite/PostgreSQL 或向量数据库Agent 没有按照生成的技能执行技能没有注入到 Prompt 或 Agent 流程中在任务启动前检索技能拼接进系统提示词或工具说明这里的排查思路适用于大多数知识管理类 Agent 系统。最重要的一点是知识库的质量决定了技能进化的上限数据源必须先做清洗和审核而不是盲目累积。7. 最佳实践与工程建议7.1 数据治理与质量控制经验采集是所有环节的基础如果原始数据质量差后面编译出的知识和技能也会不可靠。建议做到日志字段统一至少包含动作、观察、状态、时间。对成功和失败的轨迹分别标记失败轨迹中包含的教训通常更有价值。设置置信度阈值低置信度知识不参与技能生成。重要领域引入人工审核知识入库之前由经验丰富的工程师确认。7.2 安全与权限边界Agent 在执行业务任务时会接触敏感数据经验日志可能包含用户信息、内部系统状态和 API 密钥。因此日志采集阶段要做脱敏处理例如用占位符替换手机号、身份证号、Token。知识库存储涉及业务敏感数据时必须做好访问控制遵循最小权限原则。如果使用大模型做知识摘要先确认模型服务允许传输这些数据并且建议通过内部模型网关统一接入。对知识条目的修改操作保留审计记录避免错误知识被写入后影响后续任务。7.3 存储层设计本地 JSON 文件适合原型演示生产环境建议元数据使用 PostgreSQL 或 SQLite便于事务和版本管理。文本向量使用向量数据库例如 Milvus、Weaviate 或云服务提供的向量检索能力。知识条目设计版本号字段每次技能进化生成新版本保留历史版本以便回滚。检索时结合关键词过滤和语义相似度兼顾效率和准确性。7.4 评估反馈闭环技能进化需要评估闭环否则容易变成“盲目更新”。建议在系统里加入以下指标技能命中率目标任务中有多少比例触发并使用了技能。任务成功率使用技能后任务完成率是否提升。推理步数相比未使用技能平均执行步数是减少还是增加。人工纠偏次数知识库中是否需要频繁人工修正。这些指标既能反映技能质量也能反向指导知识编译策略。7.5 成本与性能控制调用大模型做经验摘要会带来额外成本。建议批量离线编译经验而不是每条日志实时调用模型。对相似经验做聚类每条聚成一个代表条目减少存储和计算开销。检索技能时限制候选数量控制注入 Prompt 的内容长度。对生成技能设置冷却期避免频繁改动导致 Agent 行为不稳定。8. 总结与学习路线本文围绕 WikiSkill 的核心思想完整演示了如何把 Agent 经验编译成持久知识并进一步驱动技能进化。总结下来关键点有这么几个经验、知识、技能是三件事分别对应原始记录、结构化信息、可复用能力。经验采集要统一格式并做脱敏清洗。知识编译是核心决定了知识库质量。技能进化需要批量知识和最小样本量保证不是一条经验就能形成技能。评估闭环和安全控制是生产落地的必要条件。如果你准备在自己项目里实践建议从一个小范围开始先选定一种任务类型采集 50 到 100 条真实轨迹手动分析哪些经验值得沉淀再写代码自动化编译最后接入 Agent 流程观察效果。等验证了价值之后再扩展到更多任务类型。下一步可以继续学习向量检索、反思机制、长期记忆架构、技能冲突消解这些方向。这些技术和 WikiSkill 的思路可以互相结合进一步把 Agent 从“每次从零开始”推进到“越用越聪明”的状态。如果本文对你有帮助建议收藏备用动手跑一遍完整流程会比只看文章收获大得多。