- AI Agent
- 数据分析
【免费下载链接】TaskWeaver
The first "code-first" agent framework for seamlessly planning and executing data analytics tasks.
TaskWeaver 的 experience(经验)模块允许 Agent 把过往对话中沉淀下来的"经验教训"保存到经验池,并在后续相似请求到来时自动检索、注入到 Planner 与 CodeInterpreter 的系统提示词中,从而避免重蹈覆辙。本文以 TaskWeaver 官方博客 Experience Selection in TaskWeaver 为主线,完整讲解两种经验选择方式:通过配置各 role 独立经验目录实现的静态经验选择,以及借助子目录与共享内存(Shared Memory)实现、可根据任务类型动态切换经验子目录的动态经验选择。读完本文,你将掌握从配置项、目录结构到角色代码编写的完整实战方案,并能直接照搬到自己的 TaskWeaver 项目中。
背景:Experience 模块解决了什么问题
在深入"选择"之前,先简要回顾经验模块的定位(详见 Experience 模块文档)。实践中,用户常常要求 TaskWeaver 解决难题:Agent 可能先走弯路、出错,经过多轮尝试或用户补充说明后最终成功。但如果下一次用户提出相似甚至完全相同的问题,Agent 依旧可能一上来就给不出正确答案——因为它"没有记住"上次的经验。为此 TaskWeaver 提出了 experience memory 机制:
- 用户在控制台输入
/save,把当前会话的聊天历史保存为raw_exp_{session_id}.yaml; - 重启后,初始化阶段会调用 LLM 将原始对话总结为经验文件
exp_{session_id}.yaml(总结提示词模板见 default_exp_prompt.yaml,要求聚焦错误与用户偏好); - 之后收到相似查询时,TaskWeaver 检索经验池,把命中的经验拼接到 Planner / CodeInterpreter 的提示词中,指导后续规划与代码生成。
除了/save自动沉淀的经验,开发者也可以手工编写经验文件(文件名前缀handcrafted_exp_{exp_id}.yaml,内含exp_id与experience_text两个字段),详见 Handcrafted Experience。无论哪种来源,最终都要经过"选择"环节决定哪些经验被当前角色使用——这正是本文的核心。
静态经验选择:为每个角色配置独立经验目录
配置方式
TaskWeaver 中的每个 role(角色)都可以拥有自己的经验目录。在项目配置文件(默认 taskweaver_config.json)中,通过role_name.experience_dir字段设置即可:
{ "planner.use_experience": true, "planner.experience_dir": "planner_exp_dir" }其中:
planner.use_experience:是否启用 Planner 角色的经验检索,默认false;planner.experience_dir:Planner 角色的经验目录路径,未配置时默认为项目目录下的experience文件夹。
对于Planner和CodeInterpreter两个角色,分别对应planner.experience_dir与code_generator.experience_dir两个配置字段。与planner同理,启用 CodeInterpreter 的经验检索需要设置code_generator.use_experience: true。一旦开启,两个角色都会在每次reply时检索并加载经验到自身提示词中(调用链见下文)。
角色名的确定规则
配置前缀role_name指的是哪个名字?官方文档明确指出:角色名默认取该角色实现文件的文件名(去掉扩展名),除非你在实现文件中通过_set_name显式指定了角色名。因此,若某角色实现文件名为planner.py,则对应的配置前缀就是planner;若你在代码里调用_set_name("my_planner")改名,则配置前缀相应变为my_planner。这一规则在源码 RoleConfig 中也能得到印证:_set_role_name通过inspect.getfile(child_class)取实现文件路径,再os.path.basename(file_name).split(".")[0]得到角色名。
源码视角:静态配置如何生效
静态经验选择的底层由 RoleConfig 与 role_load_experience 实现:
RoleConfig解析use_experience、experience_dir(默认os.path.join(app_base_path, "experience"))以及dynamic_experience_sub_path三个配置项;- 每次角色回复时,
role_load_experience调用prepare_loading检查启用状态,然后把ExperienceGenerator的经验目录设置为experience_dir,执行refresh()(若exp_{exp_id}.yaml不存在或 embedding 缺失则重建)、load_experience()(载入所有exp_*.yaml及其 embedding),最后用retrieve_experience(query)按余弦相似度筛出命中的经验(相似度阈值retrieve_threshold,默认0.2,见 ExperienceConfig)。
以 Planner 为例,该调用发生在 planner.py 的 reply 方法 中:
user_query = rounds[-1].user_query self.role_load_experience(query=user_query, memory=memory)CodeInterpreter 侧同理,见 code_generator.py。
通过为不同角色配置不同经验目录,即可"静态地"让每个角色拥有各自专属的经验集。这种方式的优点是简单直接、隔离清晰,缺点是无法针对单次请求动态切换经验来源——当经验需要随任务类型变化时,就需要动态方案了。
动态经验选择:为什么仅靠相似度检索不够
TaskWeaver 的经验检索本质上是基于查询内容与经验内容的向量相似度(余弦相似度)完成的。但作者在博客中明确指出:仅凭查询内容与经验内容之间的相似度,有时很难拿到正确的经验。
作者给出了一个真实案例:项目中存在大量任务类型(如task_type_1、task_type_2、task_type_3),每种任务类型都遵循一组指令完成工作。不同任务类型的指令内容不同,但结构高度相似——它们都有类似的步骤step_1、step_2、step_3,甚至多数步骤的标题都一样,只是步骤内容略有差异。每种任务类型都有自己专属的经验,目标就是按任务类型选择经验。
问题在于:即使用户输入(往往只是一个任务 ID)与经验内容做相似度匹配,由于各任务经验的文本结构高度雷同,很难可靠地区分到底该用哪一份经验。因此项目需要先根据任务 ID 判断任务类型,再依据任务类型选择经验——这是典型的"按业务维度路由经验"的场景,无法靠文本相似度自动完成。
动态经验选择的实现:经验子目录 + 共享内存
方案核心:允许经验目录下存在子目录
为了支持按任务类型路由,TaskWeaver 在经验选择流程中增加了一层"子路径"(sub path)机制——经验目录下允许存在子目录,每个子目录对应一种任务类型的经验集。例如:
planner_experience └── task_type_1 ├── exp_1.yaml ├── exp_2.yaml └── ...当能够根据任务 ID 识别出任务类型后,就把经验子目录设置为task_type_1,ExperienceGenerator便只在该子目录内刷新、加载与检索经验。这一机制在源码中有直接实现——get_experience_dir:
def get_experience_dir(self): assert self.experience_dir is not None, "Experience directory is not set. Call set_experience_dir() first." return os.path.join(self.experience_dir, self.sub_path) if self.sub_path else self.experience_dir配合 set_sub_path,refresh()、load_experience()、retrieve_experience()都会基于"经验目录 + 子路径"拼接出的实际目录工作。
如何在运行期动态设置子目录:共享内存
关键问题随之而来:经验子目录需要在运行期动态设置,而配置是静态的。博客给出的答案是:通过 Role 概念 在对话流程中动态写入子路径,具体载体则是 TaskWeaver 的共享内存(Shared Memory)机制。
共享内存允许一个角色向其他角色分享信息。它的载体是特殊的 Attachment,挂在某个 post 上,extra字段携带一个 SharedMemoryEntry 实例。SharedMemoryEntry 包含四个字段:
| 字段 | 类型/取值 | 含义 |
|---|---|---|
type | plan/experience_sub_path/example_sub_path(见 type_vars.py) | 共享信息的类型 |
content | str | 共享信息内容 |
scope | round/conversation | 生效范围:仅当前轮,或整个对话 |
id | str | 条目唯一 ID(create()时自动生成,前缀sme-) |
对应地,角色配置中还新增了开关dynamic_experience_sub_path(默认false),只有开启后,角色才会在加载经验时去共享内存里读取experience_sub_path类型的条目作为子目录。
写入侧:新增 TaskTypeIdentifier 角色
首先新增一个名为TaskTypeIdentifier的角色,专门根据任务 ID 识别任务类型,并把识别结果写入共享内存。其reply方法的关键代码如下(原博客完整代码):
def reply(self, memory: Memory, **kwargs: ...) -> Post: # ... # get the task type from the last post message task_type = get_task_type(last_post.message) # create an attachment post_proxy.update_attachment( type=AttachmentType.shared_memory_entry, message="Add experience sub path", extra=SharedMemoryEntry.create( type="experience_sub_path", scope="conversation", # define the effective scope of the shared memory entry to be the whole conversation content="task_type_1", ), ) return post_proxy.end()要点解读:
AttachmentType.shared_memory_entry是 Attachment 的一个合法类型枚举值(见 attachment.py);scope="conversation"表示该经验子路径在整个对话期间持续有效,而非仅当前轮次有效;content="task_type_1"就是要注入的经验子目录名,与经验目录planner_experience下的子目录一一对应。
读取侧:在需要经验的角色中读取子目录
在需要使用经验、需要动态切换子目录的角色(例如 Planner)的reply中,先从共享内存取出子路径,再加载经验(原博客完整代码):
def reply( self, memory: Memory, post_proxy: Optional[PostEventProxy] = None, prompt_log_path: Optional[str] = None, **kwargs: ..., ) -> Post: ... rounds = memory.get_role_rounds( role=self.alias, include_failure_rounds=False, ) # obtain the query from the last round query = rounds[-1].post_list[-1].message # retrieve the experience based on the query self.role_load_experience(query=query, memory=memory) ...这段代码中self.role_load_experience(query=query, memory=memory)正是 role.py 中定义的加载入口。它在内部通过 prepare_loading 完成动态子路径解析:
sub_path = "" if dynamic_sub_path: assert memory is not None, f"Memory should be provided when dynamic_{item_type}_sub_path is True" sub_paths = memory.get_shared_memory_entries(entry_type=f"{item_type}_sub_path") if sub_paths: # todo: handle multiple sub paths sub_path = sub_paths[0].content也就是说,当dynamic_experience_sub_path为true时,角色会调用memory.get_shared_memory_entries(entry_type="experience_sub_path"),取第一个条目的content作为子路径,然后experience_generator.set_sub_path(sub_path),再依次refresh()、load_experience()、retrieve_experience(query)——检索出的经验即限定在task_type_1子目录内。
共享内存条目的检索语义
Memory.get_shared_memory_entries 的实现细节值得注意:
- 遍历所有轮次与所有 post 的 attachment,只筛选
type == AttachmentType.shared_memory_entry且entry.type == 目标类型的条目; - 条目仅在
scope == "conversation"或所属轮次为最后一轮时有效——这保证了"当前轮产生的共享信息立即生效,历史轮次中的 round 级信息失效"; - 若同一角色写入了多个同类型条目,只保留最新一条(后面的覆盖前面的),避免多角色/多轮写入造成歧义。
完整配置示例
综合静态与动态两种机制,一个支持"按任务类型动态路由经验"的 Planner 配置如下:
{ "planner.use_experience": true, "planner.experience_dir": "planner_experience", "planner.dynamic_experience_sub_path": true, "code_generator.use_experience": true, "code_generator.experience_dir": "code_generator_experience" }其中planner.dynamic_experience_sub_path是动态方案的关键开关(默认false),它与experience_dir配合决定了经验最终从planner_experience/task_type_1这样的子目录中加载。经验目录结构则形如:
planner_experience ├── task_type_1 │ ├── exp_1.yaml │ ├── exp_2.yaml │ └── ... └── task_type_2 ├── exp_1.yaml └── ...静态与动态方案对比
| 维度 | 静态经验选择 | 动态经验选择 |
|---|---|---|
| 配置入口 | role_name.experience_dir | 在上述基础上增加role_name.dynamic_experience_sub_path: true |
| 经验划分方式 | 不同角色使用不同目录 | 同一目录下按子目录细分,运行期按需切换 |
| 路由依据 | 角色身份(静态绑定) | 共享内存中的experience_sub_path条目(可依据任务类型等动态写入) |
| 适用场景 | 角色间经验天然隔离,无需按请求变化 | 任务类型多样、经验需随请求动态路由(如按任务 ID 识别类型) |
| 实现要点 | 配置即可生效 | 需自定义TaskTypeIdentifier类角色写入共享内存,目标角色在reply中通过role_load_experience读取 |
注意事项
博客原文特别提示:动态经验子路径是 TaskWeaver 当前的实验性功能,未来可能发生变更(dynamic_experience_sub_path与experience_sub_path共享内存条目均属于此类)。此外,从源码prepare_loading可以看出,同样的"动态子路径"机制也适用于 example(示例)模块(对应配置dynamic_example_sub_path、共享内存类型example_sub_path),理解本文的经验路由原理后可以举一反三。若需要了解经验文件的完整结构与手工编写方式,可继续阅读 Experience 模块文档 与 Handcrafted Experience。
结论
本文围绕 TaskWeaver 官方博客,系统梳理了经验选择的两种方案:静态经验选择通过role_name.experience_dir为每个角色配置独立经验目录,简单直接、开箱即用;动态经验选择则在经验目录之上引入子目录机制,让TaskTypeIdentifier这类角色通过共享内存写入experience_sub_path条目,目标角色借助dynamic_experience_sub_path开关与role_load_experience在运行期动态切换经验子目录,从而突破"纯文本相似度"的局限,实现按任务类型等业务维度精准路由经验。这套机制让 TaskWeaver 既保留了"经验即提示词注入"的简洁理念,又具备了面向复杂业务场景的灵活路由能力。
- AI Agent
- 数据分析
【免费下载链接】TaskWeaver
The first "code-first" agent framework for seamlessly planning and executing data analytics tasks.
相关推荐
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考