1. 项目概述当智能体需要“记性”与“判断力”在智能体Agent技术快速发展的今天我们正面临一个核心瓶颈如何让智能体像人类一样在复杂、长期的任务中既能记住关键信息又能灵活地判断何时该“深思熟虑”何时该“快速行动”传统的智能体工作流要么将每一步都交给大语言模型LLM处理导致成本高昂、速度缓慢且容易在长程任务中迷失方向要么过度依赖预设的规则和工具丧失了应对未知场景的灵活性。“SKILL.nb”这个项目正是为了解决这一痛点而提出的一个精巧框架。它的全称“Selective Formalization and Gated Execution for Durable Agent Workflows”直译过来是“面向持久化智能体工作流的选择性形式化与门控执行”。听起来很学术但拆解开来其核心理念非常务实为智能体工作流构建一个“双系统”大脑。想象一下一个负责市场调研的智能体。它需要浏览几十份报告从中提取关键数据、识别趋势并最终生成一份分析摘要。如果每一步都调用LLM去阅读理解全文不仅耗时费钱还可能因为信息过载而抓不住重点。SKILL.nb的思路是让智能体学会“选择性记笔记”Selective Formalization只把那些真正关键、需要后续反复使用或推理的信息用一种结构化的方式比如JSON、知识图谱节点记录下来存入一个持久化的“工作记忆”中。同时它具备“门控判断力”Gated Execution在面对一个新步骤时能自动判断这个问题很简单可以直接查“笔记”回答吗还是需要动用“重型武器”LLM进行深度分析抑或是调用某个专用工具如计算器、API这个框架的目标是打造高效、可靠且成本可控的持久化智能体。它特别适合那些需要多步骤协作、信息有累积性、且对响应时间和计算资源有要求的场景比如自动化数据分析、复杂文档处理、长期对话陪伴型助手、以及需要与外部系统频繁交互的业务流程自动化。2. 核心设计理念双系统协同与状态持久化2.1 为何需要“选择性形式化”在长周期任务中智能体与环境的交互会产生海量的中间信息包括原始观察如网页文本、API响应、内部推理过程LLM的思考链、以及执行结果。如果全部存储不仅效率低下而且会给后续的信息检索带来巨大噪音。“选择性形式化”的精髓在于“提炼”与“结构化”。它不是一个简单的缓存而是一个主动的知识管理过程。提炼智能体需要根据当前任务目标和历史上下文判断哪些信息是“高价值”的。例如在阅读一篇财报时“公司Q3营收同比增长15%”这个事实远比大段的行业背景描述更有价值。提炼的规则可以由人工预设如“总是提取数字和实体”也可以由另一个轻量级模型或规则引擎来动态判断。结构化将提炼出的信息转换为机器更易理解和推理的格式。常见的形式包括键值对Key-Value{“metric”: “revenue_growth”, “value”: “15%”, “period”: “Q3”}知识三元组Subject-Predicate-Object(Company_A, hasRevenueGrowth, 15%)自定义的Pydantic模型定义严格的字段和类型确保数据质量。这样做的直接好处是极大压缩了工作记忆的规模提升了后续检索的速度和准确性。当智能体需要回答“Q3业绩如何”时它可以直接查询结构化的“营收增长”记录而无需重新解析整篇财报。2.2 “门控执行”如何运作“门控执行”是SKILL.nb的决策中枢它决定了工作流在每一个岔路口该走哪条路。其核心是一个路由决策函数。这个函数接收当前状态包括任务目标、工作记忆、可用工具列表作为输入输出一个执行指令。典型的执行分支包括记忆查询与推理如果问题可以直接从已形式化的记忆中得到答案或通过简单的逻辑推理如比较、计算得出则直接返回结果。这是最快、最廉价的路径。工具调用如果问题涉及特定操作如数学计算、数据库查询、调用某个API则路由到相应的工具。工具执行结果是确定性的可靠且高效。LLM调用当问题需要理解、创意、总结或复杂推理时才动用LLM。这是成本最高、最慢的路径但能力也最强。门控的关键在于决策依据。一个简单的实现可以是基于规则如果查询包含“计算”则路由到计算器工具如果查询是“总结上文”则路由到LLM。更高级的实现则会使用一个轻量级的分类模型甚至是一个提示词精巧的小型LLM来动态分析查询意图和上下文做出更精准的路由。实操心得门控逻辑的设计是平衡性能与智能的跷跷板。初期建议从简单的规则和关键词匹配开始快速验证流程。随着任务复杂化再引入基于嵌入向量的语义匹配或微调的小型分类器。切忌一开始就追求完美的智能路由那会引入不必要的复杂性。2.3 持久化工作流的状态管理“持久化”Durable是SKILL.nb应对长任务的核心。这意味着智能体的工作状态任务进度、工作记忆、中间结果必须能够被保存、加载和恢复。这通常通过以下方式实现状态快照在每一个关键步骤如一个子任务完成、或门控决策后后将整个智能体的状态包括记忆、任务栈、环境变量序列化如转为JSON并存储到数据库或文件系统中。任务标识符每个工作流实例都有一个唯一ID。当需要恢复时根据ID加载最新的状态快照智能体就能从上次中断的地方继续执行仿佛从未停止过。事件溯源更高级的模式是记录工作流执行过程中的所有事件如“记忆了A”、“调用了工具B”、“LLM生成了C”。通过重放事件流可以精确重建任何历史状态便于调试和审计。这种设计使得SKILL.nb智能体能够处理耗时数小时甚至数天的任务例如监控一个持续更新的数据源或者与用户进行跨越多天的间歇性对话。3. 架构拆解与核心组件实现一个典型的SKILL.nb系统包含以下几个核心模块我们可以用Python伪代码和概念图来理解它们是如何协作的。3.1 工作记忆模块这是选择性形式化发生的地方。我们可以将其设计为一个版本化的键值存储并附带一些智能的“提炼”逻辑。from typing import Dict, Any, List from pydantic import BaseModel import json class FormalizedMemory: 结构化工作记忆 def __init__(self): self.memory_store: Dict[str, Any] {} # 核心存储 self.memory_index: Dict[str, List[str]] {} # 反向索引便于检索 def formalize_and_store(self, raw_data: str, context: Dict[str, Any]) - str: 选择性形式化并存储。 raw_data: 原始观察如LLM输出、工具返回 context: 当前任务上下文用于指导提炼 # 1. 提炼根据上下文决定提取什么 key_points self._extract_key_points(raw_data, context) # 2. 结构化转换为标准格式 for point in key_points: # 假设我们使用一个简单的实体-关系提取器 structured_entry self._to_structured_entry(point) entry_id fentry_{len(self.memory_store)} self.memory_store[entry_id] structured_entry # 3. 索引为关键字段建立索引 for key in [subject, metric, date]: if key in structured_entry: index_key f{key}:{structured_entry[key]} self.memory_index.setdefault(index_key, []).append(entry_id) return fFormalized {len(key_points)} points into memory. def query(self, query: str) - List[Any]: 查询工作记忆 # 简单实现关键词匹配索引 results [] for index_key, entry_ids in self.memory_index.items(): if query.lower() in index_key.lower(): for eid in entry_ids: results.append(self.memory_store.get(eid)) return results if results else [] def _extract_key_points(self, text: str, context: Dict) - List[str]: # 这里可以实现基于规则或轻量级模型的提取逻辑 # 例如如果上下文是“财务分析”则提取所有百分比和货币数字 # 简化版用简单正则匹配 import re # 匹配百分比和金额 percentages re.findall(r\d\.?\d*%, text) amounts re.findall(r\$\d[\d,]*\.?\d*, text) return percentages amounts def _to_structured_entry(self, point: str) - Dict: # 将提取的点转换为结构化对象 return {value: point, type: self._infer_type(point), timestamp: ...} def _infer_type(self, point: str) - str: if % in point: return percentage elif $ in point: return currency else: return fact3.2 门控执行器门控执行器是工作流的大脑它评估当前状态并做出路由决策。class GatedExecutor: def __init__(self, llm_client, tools: Dict[str, callable], memory: FormalizedMemory): self.llm llm_client self.tools tools self.memory memory def execute_step(self, task_state: Dict) - Dict: 执行一个步骤。 task_state: 包含当前查询、历史等 current_query task_state.get(current_query) # 第一步尝试从记忆回答 memory_answer self._try_answer_from_memory(current_query) if memory_answer[can_answer]: return {type: memory, result: memory_answer[answer]} # 第二步判断是否需要调用工具 tool_to_use self._should_use_tool(current_query) if tool_to_use: tool_result self.tools[tool_to_use](current_query) # 工具结果可能也需要被形式化存储 self.memory.formalize_and_store(str(tool_result), task_state) return {type: tool, tool: tool_to_use, result: tool_result} # 第三步最终手段调用LLM llm_response self.llm.generate(promptself._build_llm_prompt(task_state)) # LLM的精彩回答也值得被记下来 self.memory.formalize_and_store(llm_response, task_state) return {type: llm, result: llm_response} def _try_answer_from_memory(self, query: str) - Dict: results self.memory.query(query) if results: # 简单合并记忆中的答案 answer ; .join([str(r) for r in results[:3]]) # 取最相关的前三条 return {can_answer: True, answer: f根据记录{answer}} return {can_answer: False, answer: None} def _should_use_tool(self, query: str) - str: # 基于规则的简单路由 tool_keywords { calculator: [计算, 加起来, 等于多少, , -, *, /], search_web: [搜索, 查找, 什么是, 谁], get_weather: [天气, 气温, 下雨], } for tool_name, keywords in tool_keywords.items(): if any(kw in query for kw in keywords): return tool_name return None3.3 持久化状态管理器负责保存和加载工作流的状态确保其持久性。import pickle import redis # 或使用其他数据库 class StateManager: def __init__(self, storage_backendredis): # 这里以Redis为例也可以使用SQLite、文件系统等 self.client redis.Redis(hostlocalhost, port6379, db0) def save_state(self, workflow_id: str, executor: GatedExecutor, memory: FormalizedMemory, other_state: Dict): 保存整个工作流状态 state { executor_memory: memory.memory_store, executor_index: memory.memory_index, task_stack: other_state.get(task_stack, []), current_step: other_state.get(current_step, 0), } # 序列化并存储 serialized pickle.dumps(state) self.client.set(fworkflow:{workflow_id}, serialized) print(fState saved for {workflow_id}) def load_state(self, workflow_id: str) - Dict: 加载工作流状态 serialized self.client.get(fworkflow:{workflow_id}) if serialized: return pickle.loads(serialized) return None # 新工作流3.4 工作流编排引擎这是将以上所有组件串联起来的主循环。它管理任务的生命周期处理中断与恢复。class DurableWorkflowEngine: def __init__(self, executor: GatedExecutor, memory: FormalizedMemory, state_manager: StateManager): self.executor executor self.memory memory self.state_manager state_manager self.workflow_id None def run(self, initial_query: str, workflow_id: str None): 运行或恢复一个工作流 self.workflow_id workflow_id or fwf_{uuid.uuid4().hex[:8]} # 尝试加载历史状态 saved_state self.state_manager.load_state(self.workflow_id) if saved_state: print(fResuming workflow {self.workflow_id}) self.memory.memory_store saved_state[executor_memory] self.memory.memory_index saved_state[executor_index] task_state saved_state else: print(fStarting new workflow {self.workflow_id}) task_state {current_query: initial_query, task_stack: [], current_step: 0} # 主执行循环简化版实际可能更复杂处理多步任务 try: while not self._is_task_complete(task_state): result self.executor.execute_step(task_state) print(fStep {task_state[current_step]}: [{result[type]}] - {result[result][:50]}...) # 更新任务状态 task_state[current_step] 1 # 这里可以根据result决定下一步的查询例如分解任务 # task_state[current_query] self._plan_next_step(result, task_state) # 定期保存状态例如每5步 if task_state[current_step] % 5 0: self.state_manager.save_state(self.workflow_id, self.executor, self.memory, task_state) print(fWorkflow {self.workflow_id} completed.) except KeyboardInterrupt: # 用户中断立即保存状态 print(\nWorkflow interrupted. Saving state...) self.state_manager.save_state(self.workflow_id, self.executor, self.memory, task_state) print(fState saved. Resume with ID: {self.workflow_id})4. 实战演练构建一个智能研报分析助手让我们用一个具体场景来串联所有概念构建一个能自动阅读多份行业研报并回答特定问题的智能助手。4.1 场景定义与初始化目标用户上传三份关于“新能源汽车电池”的PDF研报然后可以连续提问例如“三元锂电池和磷酸铁锂电池的能量密度对比如何”、“报告中提到的头部公司有哪些”。初始化步骤文档预处理使用PyPDF2或pdfplumber提取三份PDF的文本内容。初始化SKILL.nb组件FormalizedMemory: 创建一个空的工作记忆。GatedExecutor: 配置LLM如OpenAI GPT-4、工具计算器、网络搜索、并注入记忆实例。StateManager: 连接到Redis。DurableWorkflowEngine: 将以上组件组装起来。首次信息摄入选择性形式化将三份研报的文本依次输入给一个“提炼LLM”可以是比主LLM更小、更快的模型并给出提示词“请从以下行业研报文本中提取关于电池技术如类型、能量密度、成本、优缺点、公司名称、市场份额数据、以及未来预测的关键事实并以JSON列表形式输出。”将LLM输出的JSON列表逐一调用memory.formalize_and_store()方法存入工作记忆。此时记忆里不是原始文本而是类似[{subject: 三元锂电池, attribute: 能量密度, value: 250-300Wh/kg, source: report1.pdf}, ...]的结构化条目。4.2 交互式问答流程当用户提问“三元锂电池和磷酸铁锂电池的能量密度对比如何”时工作流引擎启动门控判断GatedExecutor首先调用_try_answer_from_memory。记忆查询记忆模块在索引中查找“三元锂电池 能量密度”和“磷酸铁锂 能量密度”。由于我们在初始化时已经结构化存储了这些信息查询会立刻返回相关条目。直接回答执行器发现记忆可以回答can_answer为True于是直接组合记忆中的信息生成对比回答“根据记录三元锂电池能量密度为250-300Wh/kg磷酸铁锂电池能量密度为150-210Wh/kg。” 这个过程完全无需调用LLM和外部工具速度极快成本为零。状态保存引擎在完成此步骤后可以选择保存当前状态虽然记忆未变但对话历史增加了。接下来用户问“宁德时代在这几份报告里被提及了多少次”门控判断再次尝试记忆。但我们的结构化存储可能只提取了“公司名称”作为实体没有存储“提及次数”这种需要全文统计的信息。路由决策规则判断“提及了多少次”可能涉及统计不属于简单查询。但我们的工具列表里没有“统计工具”。因此门控器判断需要调用LLM。LLM调用执行器构建提示词“请基于以下结构化记忆附上相关记忆片段和原始文本附上原始报告文本中关于宁德时代的段落统计‘宁德时代’在三份报告中出现的总次数。” 然后调用LLM。结果处理与再形式化LLM返回结果“共计提及15次”。执行器将这个新的高价值信息——“宁德时代提及次数15”——再次调用formalize_and_store存入记忆。这样如果下次再问同样或类似问题就可以直接从记忆回答了。4.3 处理复杂与未知问题用户问“根据这些信息预测一下明年磷酸铁锂电池的市场份额趋势。”门控判断记忆查询可能返回历史市场份额数据但“预测趋势”需要推理和综合。路由决策规则或分类器判断这是一个需要“分析、预测”的问题属于LLM的强项。LLM深度分析执行器构建一个复杂的提示词包含任务指令预测趋势、相关记忆磷酸铁锂的成本、性能、当前市场份额数据、以及原始报告中的相关观点段落。然后调用LLM进行深度分析。生成与存储LLM生成一段预测文本。执行器可以从中再次提炼关键结论例如“预测明年份额将提升至65%”并将其形式化存储。注意事项在这个场景中选择性形式化的策略是动态演进的。一开始我们可能只提取基础事实。随着问答的深入我们发现“提及次数”、“预测数据”也是高频需求那么就可以调整提炼LLM的提示词或者在门控执行后主动将LLM产出中的关键结论进行形式化。这是一个“记忆”与“智能”共同成长的过程。5. 性能优化与高级技巧5.1 提升记忆检索效率当记忆条目成千上万时线性搜索或简单关键词匹配就不够了。需要引入更高效的检索机制向量索引将每条结构化记忆的文本描述如“三元锂电池能量密度250-300Wh/kg”通过嵌入模型如text-embedding-3-small转换为向量存入向量数据库如Chroma、Weaviate。查询时将用户问题也转换为向量进行相似度搜索。这能实现语义层面的模糊匹配。混合检索结合关键词索引用于精确匹配实体名、数字和向量索引用于语义匹配。例如先通过关键词筛选出与“电池”、“能量密度”相关的条目再用向量检索在这些条目中找最匹配“对比”这个意图的。记忆分级根据信息的“活性”分级存储。高频访问的热数据放在内存缓存如Redis中全量数据放在向量数据库或关系型数据库中。5.2 精细化门控策略简单的规则路由容易误判。更精细的策略包括基于置信度的路由让记忆查询和工具调用都返回一个“置信度”。例如记忆检索返回的结果如果相似度分数高于0.9则直接使用如果在0.7-0.9之间则可以将记忆结果作为上下文再让LLM做最终确认或润色低于0.7则直接交给LLM。LLM作为路由器使用一个快速、廉价的小模型如GPT-3.5-turbo或Claude Haiku专门负责路由决策。提示词如“请判断以下问题最适合哪种方式解决A) 直接回答已知信息需附上信息B) 调用计算工具C) 调用搜索工具D) 需要深度推理分析。问题{用户问题} 已知结构化信息{相关记忆摘要}”。让小模型输出A/B/C/D主流程再根据结果分支。成本预算控制为工作流设置LLM调用成本预算。门控器在决定调用LLM前检查已消耗的成本和当前问题的优先级如果成本超支或问题价值不高可以降级处理如返回记忆中的近似答案或拒绝回答。5.3 确保工作流的持久性与可靠性状态快照的粒度保存状态的频率需要权衡。每步都保存最安全但开销大。通常是在完成一个逻辑上完整的子任务后或是在调用外部不可靠服务如网络API之前进行保存。这样即使后续步骤失败也能回滚到最近的安全点。错误处理与重试在门控执行器的每个分支记忆、工具、LLM都要有完善的错误处理。网络超时、API限流、LLM输出格式错误等都需要被捕获。对于可重试的错误如网络超时应实现指数退避重试机制。对于不可恢复的错误应记录详细日志并保存状态方便人工介入。异步与并行化对于可以并行执行的任务如同时处理多份文档的初始形式化使用异步编程asyncio或任务队列Celery来提升吞吐量。注意并行任务的状态管理更复杂需要确保对共享记忆的访问是线程安全的。6. 常见问题与排查实录在实际开发和运行SKILL.nb类系统时你会遇到一些典型问题。以下是一些实录与解决方案。6.1 记忆检索不准或遗漏问题表现用户问的问题明明记忆里有但系统却触发了LLM调用增加了成本和延迟。排查思路检查形式化内容首先确认你认为应该被记住的信息是否真的被成功提取并结构化了。查看memory_store里的原始数据。检查索引策略你的查询是如何匹配索引的如果是关键词匹配是否因为同义词如“营收” vs “收入”而匹配失败考虑引入同义词词典或使用词干提取。检查查询预处理用户的问题是否在查询前进行了适当的处理例如是否应该去除停用词、进行词干化解决方案引入向量检索这是解决语义匹配问题的根本方法。在形式化阶段为同一条信息生成多个索引键。例如对于“营收增长15%”除了索引“营收”还可以索引“收入”、“增长”、“15%”。实现一个查询重写步骤在查询记忆前先用一个非常轻量的规则或模型将用户问题重写成更可能匹配记忆索引的形式。6.2 门控决策错误该用LLM时用了工具或记忆问题表现系统用记忆里的一个过时或片面的答案回答了复杂问题或者调用了一个不合适的工具导致错误。排查思路分析决策日志记录下每次门控决策的输入问题、上下文和输出选择的分支及其置信度。这是最重要的调试信息。分类错误类型收集一批决策错误的案例人工标注它们本应走哪条路径。看看是规则覆盖不全还是小模型路由器判断失误。解决方案丰富规则集针对新发现的错误案例补充新的路由规则。优化路由模型如果使用小模型路由用标注好的错误案例作为训练数据对其进行微调few-shot learning或fine-tuning。增加安全阀对于从记忆或工具返回的答案计算一个“答案质量评分”。如果评分过低例如记忆检索的相似度太低或工具返回了错误码则自动降级到LLM分支进行复核或重做。6.3 工作流状态恢复后行为异常问题表现从保存的状态恢复工作流后智能体似乎“失忆”了或者执行步骤错乱。排查思路检查序列化/反序列化确保所有需要持久化的对象尤其是自定义类都是可序列化的picklable。复杂的对象如打开的数据库连接、网络会话不能直接序列化需要在保存状态前关闭在恢复后重新初始化。检查状态完整性对比中断前刚保存的状态文件和恢复后加载的状态确保所有关键变量任务栈指针、循环索引、临时变量都正确还原。检查外部依赖恢复后的环境API密钥、数据库连接、外部服务状态是否与中断前一致解决方案使用__getstate__和__setstate__魔法方法来自定义复杂对象的序列化行为。将工作流状态设计为纯数据只有字典、列表、基本类型而将所有外部依赖客户端、连接作为“运行时环境”在引擎初始化时注入。恢复时只加载纯数据状态然后将其应用到新的运行时环境中。实现状态版本管理。当代码升级后旧版本保存的状态可能无法兼容。可以在状态中保存一个版本号恢复时根据版本号进行数据迁移或报错提示。6.4 LLM调用成本失控问题表现运行一段时间后发现LLM的token消耗远超预期。排查思路分析调用日志统计哪些类型的问题最频繁地触发了LLM调用是不是门控太“懒”总把问题抛给LLM检查提示词长度每次调用LLM时是否携带了过多不必要的上下文例如把整个对话历史都塞进去记忆检索返回的结果是否过于冗长解决方案优化门控阈值提高直接使用记忆或工具的信心阈值。压缩上下文对需要送入LLM的长上下文如多段记忆进行摘要。可以用一个更便宜的模型先对上下文进行总结再将总结送入主LLM。设置硬性限制在门控器中实现一个令牌预算计数器并设置硬性上限。当预算耗尽时所有新问题强制降级到非LLM路径或直接返回“预算已用尽”的提示。缓存LLM响应对于完全相同的输入提示词其输出结果可以缓存起来。下次遇到相同问题时直接返回缓存结果。注意这需要谨慎处理确保问题场景没有发生本质变化。构建一个成熟的SKILL.nb系统是一个迭代过程。从最简单的规则和内存字典开始逐步引入向量检索、小模型路由、异步处理等高级特性同时通过详尽的日志和监控不断观察、分析和优化其决策质量与执行效率。这个框架的核心优势在于其清晰的模块化设计让你可以针对具体场景对每一个环节形式化、门控、记忆、持久化进行深度定制最终打造出一个既聪明又经济的专属持久化智能体。