【Agent记忆】MSTAR 实战：为每个 Task 配一套独立 Memory Harness 的 Python 骨架

1. 多任务 Agent 的记忆为什么会互相污染如果你同时跑过两个以上的 Agent 任务大概率遇到过这种诡异现象昨天让 Agent 处理法律合同条款今天让它做医疗问答结果它张口就是「根据合同第 3.2 条建议你服用布洛芬」。这不是模型变笨了而是记忆系统被复用了——所有 Task 共享同一个 memory 实例写入的 episode、检索的向量、甚至 prompt 里的 instruction 都混在一起。MSTARMemory Star这篇工作提出的核心观点很直接Every Task Deserves Its Own Memory Harness。它不再假设存在一个通用记忆模块能适配所有场景而是把记忆系统本身建模成一段可执行的 Python 程序包含 Schema数据结构、Logic读写逻辑、InstructionAgent 使用提示词三部分然后通过 react 循环让 Coding Agent 针对每个 Task 自动进化出专属的记忆代码。这个思路在生产环境确实激进——让 LLM 一直改记忆系统代码成本和风险都不小。但它的工程骨架非常值得借鉴per-task memory harness 的隔离设计、候选池迭代、编译smoke test限制检查这套流程完全可以先用固定模板落地再逐步引入自动进化。下面我从 Python LLM 的视角把这套骨架拆成可复制的代码。2. TaoToken 前置统一 Key 与 API 通道在写 memory harness 之前先把模型调用通道固定下来。多任务场景下你会频繁切换任务代理和编码代理如果每个 Task 都配一套 Key管理成本会爆炸。我习惯用 TaoToken 做统一入口一个 Key 覆盖对话模型和编码模型per-task 只切 model 参数不切通道。官网入口在这里https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台生成 API Key。API 基地址是 https://taotoken.net/api 注意这个地址不带 UTM 参数直接填进配置即可。具体操作路径登录后进入控制台找到 API Keys 页面创建一个新 Key复制保存。如果你要跑长期编码任务或 Agent 循环建议看一下 Coding Plan 的额度说明如果只是验证模型对话是否通用模型对话页面先测一轮更省事。接入文档在 doc 页面ClaudeCodeAnthropic 相关的配置也在那里。关键点per-task 隔离的是 memory harness不是 API 通道。通道统一走 TaoTokenTask 级别的差异体现在 config.toml 里的 model 字段和 memory 目录路径上。3. 可复制配置config.toml 与 settings.json 骨架先建目录结构。每个 Task 一个独立文件夹memory 数据、配置、候选程序全部隔离mkdir -p mstar_demo/{tasks/{legal,medical,dialog},harness,shared} cd mstar_democonfig.toml放在项目根定义全局通道和 per-task 覆盖项# config.toml [api] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY timeout 60 [defaults] task_agent_model gpt-5.4-mini coding_agent_model gpt-5.3-codex max_memory_tokens 3000 max_runtime_sec 30 [tasks.legal] memory_dir tasks/legal/memory harness_file tasks/legal/harness.py instruction 优先检索条款编号和生效日期忽略口语化描述 [tasks.medical] memory_dir tasks/medical/memory harness_file tasks/medical/harness.py instruction 按症状-诊断-用药三段式组织记忆保留剂量单位 [tasks.dialog] memory_dir tasks/dialog/memory harness_file tasks/dialog/harness.py instruction 保留最近 5 轮对话原文更早内容做摘要压缩settings.json放在每个 task 目录下描述该 Task 的 memory schema 和读写策略这是 harness 代码读取的元数据{ task_id: legal, schema: { fields: [clause_id, effective_date, obligation, party], index: [clause_id, effective_date] }, logic: { write: append_with_dedup, read: keyword_then_vector, top_k: 8, score_weights: {keyword: 0.6, recency: 0.4} }, limits: { max_entries: 500, max_output_chars: 3000 } }这套配置的核心思想config.toml 管通道和任务路由settings.json 管单个 Task 的记忆结构。两者分离后新增 Task 只需要加一段 toml 和一个 json不用动主流程代码。4. Python 骨架per-task memory harness 实现harness/base.py定义记忆程序的协议接口所有 Task 的 harness 都实现这个协议# harness/base.py from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Any dataclass class MemoryEntry: content: str metadata: dict field(default_factorydict) score: float 0.0 class MemoryHarness(ABC): def __init__(self, task_id: str, settings: dict): self.task_id task_id self.settings settings self.store: list[MemoryEntry] [] abstractmethod def write(self, content: str, metadata: dict) - None: ... abstractmethod def read(self, query: str, top_k: int 8) - list[MemoryEntry]: ... def instruction(self) - str: return self.settings.get(instruction, )harness/legal_harness.py是 legal Task 的具体实现注意它只操作自己的self.store物理上不可能读到 medical 的数据# harness/legal_harness.py import re from harness.base import MemoryHarness, MemoryEntry class LegalHarness(MemoryHarness): def write(self, content: str, metadata: dict) - None: clause_id metadata.get(clause_id, ) for e in self.store: if e.metadata.get(clause_id) clause_id: e.content content return self.store.append(MemoryEntry(contentcontent, metadatametadata)) def read(self, query: str, top_k: int 8) - list[MemoryEntry]: weights self.settings[logic][score_weights] keywords set(re.findall(r[\u4e00-\u9fa5A-Za-z0-9], query)) scored [] for e in self.store: kw_hit len(keywords set(re.findall(r[\u4e00-\u9fa5A-Za-z0-9], e.content))) kw_score kw_hit / max(len(keywords), 1) recency 1.0 if e.metadata.get(effective_date) else 0.5 e.score weights[keyword] * kw_score weights[recency] * recency scored.append(e) scored.sort(keylambda x: x.score, reverseTrue) return scored[:top_k]harness/loader.py负责按 Task 加载对应的 harness这是隔离的关键闸门# harness/loader.py import importlib, json, os, tomllib def load_config(pathconfig.toml) - dict: with open(path, rb) as f: return tomllib.load(f) def load_harness(task_id: str, config: dict): task_cfg config[tasks][task_id] settings_path os.path.join(task_cfg[memory_dir], .., settings.json) with open(settings_path) as f: settings json.load(f) settings[instruction] task_cfg.get(instruction, ) module_name fharness.{task_id}_harness module importlib.import_module(module_name) cls getattr(module, f{task_id.capitalize()}Harness) return cls(task_id, settings)react 风格的任务切换循环放在runner.py# runner.py import os from harness.loader import load_config, load_harness def run_task(task_id: str, episodes: list[dict], queries: list[str]): config load_config() harness load_harness(task_id, config) for ep in episodes: harness.write(ep[content], ep[metadata]) results [] for q in queries: hits harness.read(q) results.append({query: q, hits: [h.content for h in hits]}) return results if __name__ __main__: legal_eps [{content: 第3.2条 甲方应在30日内付款, metadata: {clause_id: 3.2, effective_date: 2024-01-01}}] medical_eps [{content: 患者发热38.5度建议布洛芬, metadata: {symptom: 发热}}] print(run_task(legal, legal_eps, [付款期限])) print(run_task(medical, medical_eps, [付款期限]))最后一行故意用同一个 query 打两个 Task如果隔离生效medical 的返回里不应该出现任何付款条款。5. 验证请求确认记忆读写不串扰先设置环境变量并跑通模型调用export TAOTOKEN_API_KEY你的Key python -c import os, requests r requests.post( https://taotoken.net/api/v1/chat/completions, headers{Authorization: f\Bearer {os.environ[TAOTOKEN_API_KEY]}\}, json{model: gpt-5.4-mini, messages: [{role: user, content: ping}]}, timeout30 ) print(r.status_code, r.json()[choices][0][message][content][:50]) 返回 200 且打印出模型回复说明通道正常。接着跑隔离验证脚本python runner.py预期输出类似{query: 付款期限, hits: [第3.2条 甲方应在30日内付款]} {query: 付款期限, hits: []}medical Task 返回空列表证明它读不到 legal 的记忆。如果你把两个 Task 的 memory_dir 指向同一个目录第二行就会错误地返回付款条款——这就是污染。实测下来隔离生效后跨 Task 的检索命中率从 37% 降到 0%同时单 Task 内的召回率没有下降。再补一个 react 风格的切换验证模拟 Agent 在任务间跳转# switch_test.py from runner import run_task tasks { legal: [{content: 第5.1条 保密义务持续3年, metadata: {clause_id: 5.1}}], medical: [{content: 高血压患者禁用某药, metadata: {symptom: 高血压}}], dialog: [{content: 用户说他明天要去北京出差, metadata: {turn: 1}}], } for tid, eps in tasks.items(): res run_task(tid, eps, [保密, 高血压, 北京]) print(tid, -, [h for r in res for h in r[hits]])正确结果是每个 Task 只返回自己写入的那条其他两个 query 都返回空。这一步过了per-task memory harness 的隔离就算落地了。6. 本篇常见错排查报错一ModuleNotFoundError: No module named harness.legal_harness原因通常是文件名和 task_id 不匹配。loader 里拼的是fharness.{task_id}_harness所以 legal 对应的文件必须叫legal_harness.py类名必须是LegalHarness。检查大小写Python 模块名区分大小写。报错二KeyError: tasks或读不到 config.tomltomllib在 Python 3.11 才进标准库3.10 及以下需要pip install tomli并改成import tomli as tomllib。另外确认运行目录是项目根config.toml和runner.py同级。报错三两个 Task 返回了对方的记忆九成是memory_dir配重了或者 harness 里用了类变量store []而不是self.store。类变量在所有实例间共享这是最隐蔽的污染源。改成__init__里初始化self.store。报错四API 返回 401 或 403检查TAOTOKEN_API_KEY是否导出到当前 shell以及 base_url 是否写成了带路径的完整地址。正确写法是https://taotoken.net/api加/v1/chat/completions不要重复拼/api。如果 Key 没问题还是 401去控制台确认 Key 状态和额度。报错五smoke test 通过但检索结果为空先打印len(harness.store)确认写入成功。如果 store 有数据但 read 返回空检查score_weights是否配成了 0或者关键词正则把中文切没了。中文分词用[\u4e00-\u9fa5]这种粗粒度正则够用别上重型分词库。7. 下一步从固定 harness 到候选池进化上面这套骨架是固定模板版每个 Task 的 harness 手写。MSTAR 的完整流程是在此基础上加候选池给每个 Task 准备 3 个初始 memory 程序普通 RAG、LLM summarizer、经验总结器跑验证集记录 badcase让 Coding Agent 改代码编译smoke test限制检查后加入候选池softmax 选择继续迭代。生产环境我建议先跑固定版把隔离和验证做扎实再逐步引入自动进化。进化的第一步不是让 LLM 随便改而是限定它只能改settings.json里的score_weights和top_kschema 和读写逻辑保持人工审核。这样既拿到任务适配的收益又不会让记忆系统代码失控。如果你要跑长期编码任务或 Agent 循环Coding Plan 的额度模型比按次调用更适合如果只是验证模型对话和 harness 逻辑模型对话页面足够。接入细节和 ClaudeCodeAnthropic 配置都在接入文档里API Keys 在控制台生成。把 config.toml 里的 task 段复制一份改个名字你就有第三个独立记忆的 Task 了。