WorkBuddy_AI Harness架构解析7_上下文管理器

上下文管理器摘要在多轮任务型 AI 系统如自动化编码助手、Agent Harness、工作流引擎中上下文管理器Context Manager是决定系统能否稳定完成复杂任务的核心组件。它负责在多轮交互之间维护三类关键状态工作记忆Working Memory当前任务目标、子任务进度、推理中间结果。文件状态File State项目目录结构、文件内容、版本差异。历史操作Operation History已经执行过的命令、工具调用、失败与回滚记录。本文将从工程实现角度出发系统讲解上下文管理器的设计目标、数据结构、生命周期、持久化策略、压缩与检索机制并给出一个完整可运行的 Python 示例实现。文章适合正在构建 Agent、Copilot、自动化流水线或任务编排系统的工程师阅读。引言为什么需要上下文管理器假设你正在构建一个名为Harness的自动化编码代理它可以接收用户的多轮指令“帮我重构这个模块”、“现在加上单元测试”、“刚才那个测试失败了修一下”调用工具读写文件、执行 shell、运行测试在数十轮甚至上百轮交互后仍然记得“我们最初的目标是什么”、“哪些文件被改过”、“上一步为什么失败”如果没有上下文管理器会出现以下典型问题问题表现目标漂移多轮之后 AI 忘记原始需求开始做无关的事文件幻读AI 以为文件内容是 A实际磁盘上已经是 B重复操作重复执行已经成功过的命令浪费 token 和时间无法回滚出错后不知道之前改了什么无法恢复到稳定状态上下文爆炸把全部历史塞进 prompt超出模型窗口成本飙升上下文管理器就是为了解决这些问题而存在的。它不是简单的“消息列表”而是一个带状态、可查询、可压缩、可持久化的运行时子系统。本文要解决的问题上下文管理器应该保存哪些数据如何设计工作记忆的数据结构如何跟踪文件状态并检测“陈旧读取”如何记录历史操作并支持回滚如何在有限的 token 预算下做上下文压缩与检索如何把这些整合成一个可运行的实现正文1. 上下文管理器的整体架构一个成熟的上下文管理器通常由四个子系统组成┌─────────────────────────────────────────────────────┐ │ Context Manager │ │ │ │ ┌──────────────┐ ┌──────────────┐ │ │ │ Working │ │ File State │ │ │ │ Memory │ │ Tracker │ │ │ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ ┌──────▼───────┐ ┌──────▼───────┐ │ │ │ Operation │ │ Compressor / │ │ │ │ History │ │ Retriever │ │ │ └──────────────┘ └──────────────┘ │ │ │ │ ┌──────────────┐ │ │ │ Persistence │ │ │ │ (SQLite/JSON)│ │ │ └──────────────┘ │ └─────────────────────────────────────────────────────┘Working Memory任务级状态如目标、计划、已完成步骤。File State Tracker文件快照与哈希用于检测变更。Operation History所有工具调用与结果支持审计与回滚。Compressor / Retriever把长历史压缩成短摘要并按需检索。Persistence把状态落盘支持跨进程恢复。下面我们逐一展开。2. 工作记忆Working Memory的设计工作记忆回答三个问题我们要做什么Goal现在做到哪了Progress下一步做什么Next Action一个实用的工作记忆结构如下fromdataclassesimportdataclass,fieldfromtypingimportList,Dict,OptionalfromdatetimeimportdatetimedataclassclassSubTask:id:strdescription:strstatus:strpending# pending / in_progress / done / failedresult:Optional[str]Noneupdated_at:strfield(default_factorylambda:datetime.utcnow().isoformat())dataclassclassWorkingMemory:goal:str# 原始目标constraints:List[str]field(default_factorylist)# 约束条件subtasks:List[SubTask]field(default_factorylist)scratchpad:Dict[str,str]field(default_factorydict)# 临时推理结果current_focus:Optional[str]None# 当前聚焦的子任务 iddefadd_subtask(self,sid:str,desc:str):self.subtasks.append(SubTask(idsid,descriptiondesc))defupdate_subtask(self,sid:str,status:str,result:Optional[str]None):fortinself.subtasks:ift.idsid:t.statusstatus t.resultresult t.updated_atdatetime.utcnow().isoformat()returnraiseKeyError(fsubtask{sid}not found)defsummary(self)-str:生成人类可读的摘要用于注入 promptlines[fGoal:{self.goal}]ifself.constraints:lines.append(Constraints: ; .join(self.constraints))lines.append(Subtasks:)fortinself.subtasks:mark{done:✔,failed:✘,in_progress:▶,pending:○}[t.status]lines.append(f{mark}[{t.id}]{t.description})ifself.current_focus:lines.append(fCurrent focus:{self.current_focus})return\n.join(lines)关键设计点Goal 与 Constraints 一旦设定就应尽量不变它们是防止“目标漂移”的锚点。SubTask 是结构化进度比自由文本的“对话历史”更容易压缩和检索。scratchpad 用于存放中间推理例如“我怀疑 bug 在第 42 行”它不需要进入长期历史但当前轮次需要。3. 文件状态跟踪File State Tracker在多轮任务中文件状态是最容易出错的部分。常见陷阱AI 在第 3 轮读了main.py第 10 轮基于旧内容做修改但中间用户手动改了文件。AI 认为某个文件不存在其实已经被上一步的工具创建了。解决方案是内容哈希 快照importhashlibimportosfromdataclassesimportdataclass,fieldfromtypingimportDict,OptionalfromdatetimeimportdatetimedataclassclassFileSnapshot:path:strcontent:strhash:strread_at:strmtime:floatclassFileStateTracker:def__init__(self,root:str):self.rootos.path.abspath(root)self._snapshots:Dict[str,FileSnapshot]{}def_abs(self,path:str)-str:returnos.path.abspath(os.path.join(self.root,path))staticmethoddef_hash(content:str)-str:returnhashlib.sha256(content.encode(utf-8)).hexdigest()[:16]defread(self,path:str)-str:读取文件并记录快照abs_pathself._abs(path)withopen(abs_path,r,encodingutf-8)asf:contentf.read()snapFileSnapshot(pathpath,contentcontent,hashself._hash(content),read_atdatetime.utcnow().isoformat(),mtimeos.path.getmtime(abs_path),)self._snapshots[path]snapreturncontentdefwrite(self,path:str,content:str)-None:写入文件并更新快照abs_pathself._abs(path)os.makedirs(os.path.dirname(abs_path),exist_okTrue)withopen(abs_path,w,encodingutf-8)asf:f.write(content)self._snapshots[path]FileSnapshot(pathpath,contentcontent,hashself._hash(content),read_atdatetime.utcnow().isoformat(),mtimeos.path.getmtime(abs_path),)defis_stale(self,path:str)-bool:判断快照是否已过期磁盘内容与快照不一致snapself._snapshots.get(path)ifsnapisNone:returnTrueabs_pathself._abs(path)ifnotos.path.exists(abs_path):returnTrueifos.path.getmtime(abs_path)!snap.mtime:withopen(abs_path,r,encodingutf-8)asf:currentf.read()returnself._hash(current)!snap.hashreturnFalsedefdiff_against_disk(self,path:str)-Optional[str]:返回快照与磁盘的差异提示ifnotself.is_stale(path):returnNonereturnfFile {path} has changed on disk since last read. Please re-read before editing.使用方式trackerFileStateTracker(root./project)tracker.read(src/main.py)# ... 经过若干轮后用户手动修改了文件 ...iftracker.is_stale(src/main.py):print(tracker.diff_against_disk(src/main.py))# 强制重新读取tracker.read(src/main.py)这个机制能显著减少“基于陈旧内容做修改”的低级错误。4. 历史操作记录Operation History每一次工具调用都应当被记录包含操作类型read / write / shell / test输入参数输出结果成功/失败/异常时间戳关联的子任务 idfromdataclassesimportdataclass,field,asdictfromtypingimportAny,Dict,List,OptionalfromdatetimeimportdatetimeimportjsondataclassclassOperation:op_id:intkind:str# read / write / shell / testargs:Dict[str,Any]result:Optional[str]success:boolsubtask_id:Optional[str]Nonetimestamp:strfield(default_factorylambda:datetime.utcnow().isoformat())classOperationHistory:def__init__(self,max_len:int1000):self._ops:List[Operation][]self._next_id1self.max_lenmax_lendefrecord(self,kind:str,args:Dict[str,Any],result:Optional[str],success:bool,subtask_id:Optional[str]None)-Operation:opOperation(op_idself._next_id,kindkind,argsargs,resultresult,successsuccess,subtask_idsubtask_id,)self._next_id1self._ops.append(op)iflen(self._ops)self.max_len:self._opsself._ops[-self.max_len:]returnopdefrecent(self,n:int10)-List[Operation]:returnself._ops[-n:]defby_subtask(self,sid:str)-List[Operation]:return[opforopinself._opsifop.subtask_idsid]deffailed(self)-List[Operation]:return[opforopinself._opsifnotop.success]defto_prompt(self,n:int5)-str:把最近 n 条操作转成简短文本注入 promptlines[]foropinself.recent(n):statusOKifop.successelseFAILlines.append(f[{op.op_id}]{op.kind}{op.args}-{status})return\n.join(lines)defdump(self,path:str)-None:withopen(path,w,encodingutf-8)asf:json.dump([asdict(op)foropinself._ops],f,ensure_asciiFalse,indent2)defload(self,path:str)-None:withopen(path,r,encodingutf-8)asf:datajson.load(f)self._ops[Operation(**d)fordindata]self._next_id(max(op.op_idforopinself._ops)1)ifself._opselse1为什么要有subtask_id因为当某个子任务失败时我们需要快速定位“这个子任务相关的所有操作”从而决定回滚范围或重试策略。5. 上下文压缩与检索多轮任务动辄几十上百轮把全部历史塞进 prompt 不现实。上下文管理器需要做两件事压缩把长历史折叠成短摘要。检索根据当前问题动态拉取相关历史。一个简单但有效的策略是分层摘要classContextCompressor:def__init__(self,history:OperationHistory,memory:WorkingMemory):self.historyhistory self.memorymemorydefbuild_context(self,token_budget:int2000)-str:按优先级拼装上下文超预算时逐层裁剪sections[]# 1. 目标与约束最高优先级永不裁剪sections.append((GOAL,self.memory.summary()))# 2. 当前聚焦子任务的相关操作ifself.memory.current_focus:opsself.history.by_subtask(self.memory.current_focus)[-5:]ifops:text\n.join(f[{op.op_id}]{op.kind}{OKifop.successelseFAIL}foropinops)sections.append((FOCUS_OPS,text))# 3. 最近失败的操作帮助避免重蹈覆辙failsself.history.failed()[-3:]iffails:text\n.join(f[{op.op_id}]{op.kind}{op.args}-{op.result[:80] if op.result else }foropinfails)sections.append((RECENT_FAILURES,text))# 4. 最近操作低优先级可裁剪sections.append((RECENT_OPS,self.history.to_prompt(5)))# 5. 按预算裁剪returnself._fit(sections,token_budget)staticmethoddef_fit(sections,budget:int)-str:# 粗略估算1 token ≈ 4 字符out[]used0forname,textinsections:blockf###{name}\n{text}\ncostlen(block)//4ifusedcostbudgetandnamenotin(GOAL,):continueout.append(block)usedcostreturn\n.join(out)检索增强如果历史很长可以引入向量检索。把每条 Operation 的args result编码成向量按当前 query 检索 top-k。这属于进阶话题但接口设计上应预留classHistoryRetriever:def__init__(self,history:OperationHistory):self.historyhistorydefsearch(self,query:str,k:int5)-List[Operation]:简化版关键词匹配。生产环境可替换为向量检索。keywordsset(query.lower().split())scored[]foropinself.history._ops:text(str(op.args) (op.resultor)).lower()scoresum(1forkwinkeywordsifkwintext)ifscore0:scored.append((score,op))scored.sort(keylambdax:-x[0])return[opfor_,opinscored[:k]]6. 整合一个完整的 Harness 上下文管理器把上面的组件拼起来classContextManager:def__init__(self,root:str,persist_path:str./context.json):self.memoryWorkingMemory(goal)self.filesFileStateTracker(rootroot)self.historyOperationHistory()self.persist_pathpersist_path# ---------- 生命周期 ----------defstart_task(self,goal:str,constraints:Optional[List[str]]None):self.memory.goalgoal self.memory.constraintsconstraintsor[]# ---------- 工具封装 ----------defread_file(self,path:str,subtask_id:Optional[str]None)-str:try:contentself.files.read(path)self.history.record(read,{path:path},content[:200],True,subtask_id)returncontentexceptExceptionase:self.history.record(read,{path:path},str(e),False,subtask_id)raisedefwrite_file(self,path:str,content:str,subtask_id:Optional[str]None)-None:try:self.files.write(path,content)self.history.record(write,{path:path,size:len(content)},ok,True,subtask_id)exceptExceptionase:self.history.record(write,{path:path},