Harness vs 传统Agent框架摘要随着大语言模型LLM能力的快速演进Agent智能体框架成为连接模型能力与真实任务执行的桥梁。从早期的 ReAct、AutoGPT到后来的 LangChain、AutoGen、CrewAI开发者们不断探索如何让模型自主完成复杂任务。然而在实际工程落地中传统 Agent 框架暴露出任务闭环不完整、文件操作能力薄弱、交付验收缺乏标准等痛点。Harness 作为一种新兴的 Agent 运行时范式围绕任务闭环、文件操作、交付验收三大核心重新设计了执行模型。本文将深入对比 Harness 与传统 Agent 框架的设计差异并通过完整代码示例展示 Harness 在真实任务场景下的优势。引言为什么需要重新思考 Agent 框架先说一个真实场景。假设你让一个基于 LangChain 的 Agent 完成这样的任务“帮我读取data/sales.csv按季度聚合销售额生成一张柱状图并把分析报告写到reports/q1_summary.md。”在传统 Agent 框架中你大概率会遇到以下问题任务闭环断裂Agent 可能在读取文件成功后忘记自己下一步该做什么或者在中途因为一次工具调用失败就终止了整个流程。文件操作依赖外部工具Agent 本身没有对文件系统的原生理解必须通过FileReadTool、PythonREPLTool等工具间接操作容易出现路径错误、编码错误、上下文丢失。交付验收模糊任务完成的标准是什么是 Agent 说我完成了还是真的有一个可验证的产物存在传统框架往往只返回一段自然语言总结缺乏对交付物的强约束。Harness 的设计哲学正是针对这三点让 Agent 运行在一个受控的执行框架Harness中任务必须有闭环、文件操作必须是一等公民、交付必须有可验收的产物。下面我们逐层展开对比。一、任务闭环从对话式循环到状态机驱动1.1 传统 Agent 框架的循环模型传统框架以 LangChain 的AgentExecutor为例通常采用思考-行动-观察的循环# 传统 LangChain Agent 的典型循环简化版fromlangchain.agentsimportAgentExecutor,create_react_agentfromlangchain.toolsimportToolfromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o)tools[Tool(nameread_file,funclambdap:open(p).read(),description读取文件),Tool(namewrite_file,funclambdaargs:open(args[0],w).write(args[1]),description写文件),]agentcreate_react_agent(llm,tools,prompt)executorAgentExecutor(agentagent,toolstools,max_iterations10,verboseTrue)# 问题max_iterations 一旦耗尽任务直接失败没有任何恢复机制resultexecutor.invoke({input:读取 data/sales.csv 并生成报告})print(result[output])这种模型的问题在于循环终止条件模糊靠max_iterations或 LLM 输出 “Final Answer” 来终止容易提前退出或死循环。状态不持久每一步的中间结果只存在于对话历史里一旦上下文被截断Agent 就失忆了。失败无恢复某一步工具调用失败整个任务就崩了没有重试、回滚或降级策略。1.2 Harness 的状态机闭环Harness 把任务建模为一个显式状态机每个任务有明确的状态PENDING → RUNNING → VERIFYING → DONE / FAILED并且每一步的产物都持久化。# Harness 风格的任务闭环示意实现fromdataclassesimportdataclass,fieldfromenumimportEnumfromtypingimportAnyimportjsonimportosclassTaskState(Enum):PENDINGpendingRUNNINGrunningVERIFYINGverifyingDONEdoneFAILEDfaileddataclassclassTask:goal:strstate:TaskStateTaskState.PENDING artifacts:dictfield(default_factorydict)# 交付物清单history:listfield(default_factorylist)# 执行轨迹retries:int0classHarness:def__init__(self,workspace:str,max_retries:int3):self.workspaceworkspace self.max_retriesmax_retries os.makedirs(workspace,exist_okTrue)defrun(self,task:Task,steps:list):steps 是一组可重试的原子操作task.stateTaskState.RUNNINGforstepinsteps:successFalsewhiletask.retriesself.max_retriesandnotsuccess:try:resultstep(task)task.history.append({step:step.__name__,result:result})successTrueexceptExceptionase:task.retries1task.history.append({step:step.__name__,error:str(e)})ifnotsuccess:task.stateTaskState.FAILED self._persist(task)returntask# 进入验收阶段task.stateTaskState.VERIFYINGifself.verify(task):task.stateTaskState.DONEelse:task.stateTaskState.FAILED self._persist(task)returntaskdefverify(self,task:Task)-bool:子类实现具体的验收逻辑raiseNotImplementedErrordef_persist(self,task:Task):pathos.path.join(self.workspace,task_state.json)withopen(path,w,encodingutf-8)asf:json.dump({goal:task.goal,state:task.state.value,artifacts:task.artifacts,history:task.history,},f,ensure_asciiFalse,indent2)关键差异维度传统 AgentHarness循环模型对话驱动靠 LLM 判断终止状态机驱动状态显式失败处理整体失败单步重试 状态持久化上下文依赖对话历史依赖持久化状态可恢复性几乎不可恢复可从任意状态续跑二、文件操作从工具调用到一等公民2.1 传统框架的文件操作痛点在 LangChain 中文件操作是通过 Tool 暴露给 LLM 的。这带来几个问题# 传统方式文件操作是字符串参数传给工具tools[Tool(namewrite_file,funclambdax:open(x.split(|)[0],w).write(x.split(|)[1]),description参数格式文件路径|内容,)]# LLM 需要自己拼接字符串容易出错# - 路径中的特殊字符# - 内容中的换行、引号# - 大文件超出 token 限制核心问题参数序列化脆弱LLM 要把路径和内容拼成一个字符串遇到多行文本、特殊字符就崩。无路径校验Agent 可能写到/etc/passwd或工作区外。无版本管理覆盖写就是覆盖没有 diff、没有回滚。大文件无解超过上下文窗口的文件根本读不进来。2.2 Harness 的文件系统抽象Harness 把文件操作提升为运行时的一等公民提供受控的虚拟文件系统# Harness 的受控文件系统importosimportshutilfrompathlibimportPathfromdatetimeimportdatetimeclassWorkspaceFS:受控工作区文件系统所有操作限定在 root 目录内def__init__(self,root:str):self.rootPath(root).resolve()self.root.mkdir(parentsTrue,exist_okTrue)self.snapshotsself.root/.snapshotsself.snapshots.mkdir(exist_okTrue)def_safe_path(self,rel_path:str)-Path:防止路径穿越攻击target(self.root/rel_path).resolve()ifnotstr(target).startswith(str(self.root)):raisePermissionError(f路径越界:{rel_path})returntargetdefwrite(self,rel_path:str,content:str,snapshot:boolTrue):targetself._safe_path(rel_path)target.parent.mkdir(parentsTrue,exist_okTrue)# 写前快照支持回滚ifsnapshotandtarget.exists():tsdatetime.now().strftime(%Y%m%d_%H%M%S)snapself.snapshots/f{rel_path.replace(/,_)}.{ts}.bakshutil.copy2(target,snap)withopen(target,w,encodingutf-8)asf:f.write(content)return{path:rel_path,bytes:len(content.encode(utf-8))}defread(self,rel_path:str,max_bytes:int100_000)-str:targetself._safe_path(rel_path)ifnottarget.exists():raiseFileNotFoundError(rel_path)withopen(target,r,encodingutf-8)asf:dataf.read(max_bytes)returndatadeflist_dir(self,rel_path:str.):targetself._safe_path(rel_path)returnsorted([p.nameforpintarget.iterdir()ifnotp.name.startswith(.)])defrollback(self,rel_path:str):回滚到最近一次快照keyrel_path.replace(/,_)candidatessorted(self.snapshots.glob(f{key}.*.bak))ifnotcandidates:raiseFileNotFoundError(无快照可回滚)latestcandidates[-1]shutil.copy2(latest,self._safe_path(rel_path))return{rolled_back_to:str(latest)}关键差异维度传统 AgentHarness操作方式LLM 拼字符串调工具结构化 API 调用路径安全无校验强制沙箱大文件受 token 限制分块读取 摘要版本管理无自动快照 回滚编码处理靠 LLM 猜显式指定 检测有了这个抽象Agent 在生成代码/报告时不再需要把内容塞进 tool 参数字符串而是直接调用fs.write()内容作为一个独立的 Python 字符串传入多行、特殊字符都不再是问题。三、交付验收从LLM 说完成到产物可验证3.1 传统框架的伪完成传统 Agent 的结束标志通常是 LLM 输出 “Final Answer: …”这是一个自然语言声明不是可验证的事实。# 传统框架任务完成只是 LLM 的一句话resultexecutor.invoke({input:生成 sales_report.md})# 输出可能是# {output: 我已经为您生成了销售报告保存在 sales_report.md 中。}# 但文件可能根本不存在或者内容是空的这就是所谓的“幻觉交付”—— Agent 声称完成了任务但产物并不存在或不符合要求。3.2 Harness 的验收契约Harness 引入Acceptance Contract验收契约概念任务在启动时就声明完成的判定标准运行结束后由独立的验证器检查。# 验收契约 验证器fromabcimportABC,abstractmethodfrompathlibimportPathclassAcceptanceCriterion(ABC):abstractmethoddefcheck(self,workspace:WorkspaceFS)-tuple[bool,str]:返回 (是否通过, 说明)passclassFileExists(AcceptanceCriterion):def__init__(self,path:str,min_bytes:int1):self.pathpath self.min_bytesmin_bytesdefcheck(self,fs:WorkspaceFS)-tuple[bool,str]:try:contentfs.read(self.path)exceptFileNotFoundError:returnFalse,f文件不存在:{self.path}iflen(content.encode(utf-8))self.min_bytes:returnFalse,f文件内容过小:{self.path}returnTrue,f文件存在且大小合规:{self.path}classContainsKeywords(AcceptanceCriterion):def__init__(self,path:str,keywords:list[str]):self.pathpath self.keywordskeywordsdefcheck(self,fs:WorkspaceFS)-tuple[bool,str]:try:contentfs.read(self.path)exceptFileNotFoundError:returnFalse,f文件不存在:{self.path}missing[kforkinself.keywordsifknotincontent]ifmissing:returnFalse,f缺少关键词:{missing}returnTrue,所有关键词均存在classMarkdownHasHeading(AcceptanceCriterion):def__init__(self,path:str,level:int1):self.pathpath self.levelleveldefcheck(self,fs:WorkspaceFS)-tuple[bool,str]:contentfs.read(self.path)prefix#*self.level ifnotany(line.startswith(prefix)forlineincontent.splitlines()):returnFalse,f缺少{self.level}级标题returnTrue,标题结构合规classVerifier:def__init__(self,criteria:list[AcceptanceCriterion]):self.criteriacriteriadefverify(self,fs:WorkspaceFS)-dict:results[]all_passedTrueforcinself.criteria:ok,msgc.check(fs)results.append({criterion:c.__class__.__name__,passed:ok,message:msg})all_passedall_passedandokreturn{passed:all_passed,details:results}3.3 完整示例一个可验收的任务把前面几节的组件组装起来我们跑一个真实任务# 完整示例读取 CSV → 生成 Markdown 报告 → 验收importcsvfrompathlibimportPath# 1. 准备测试数据os.makedirs(workspace/data,exist_okTrue)withopen(workspace/data/sales.csv,w,encodingutf-8,newline)asf:writercsv.writer(f)writer.writerow([quarter,amount])writer.writerow([Q1,12000])writer.writerow([Q2,15000])writer.writerow([Q3,18000])writer.writerow([Q4,21000])# 2. 定义任务步骤fsWorkspaceFS(workspace)defstep_read_csv(task:Task):contentfs.read(data/sales.csv)rowslist(csv.DictReader(content.splitlines()))task.artifacts[rows]rowsreturnf读取{len(rows)}行defstep_aggregate(task:Task):rowstask.artifacts[rows]totalsum(int(r[amount])forrinrows)task.artifacts[total]total task.artifacts[avg]total/len(rows)returnf总额{total}均值{task.artifacts[avg]:.2f}defstep_write_report(task:Task):rowstask.artifacts[rows]lines[# 季度销售报告,,## 数据明细,]lines.append(| 季度 | 销售额 |)lines.append(|------|--------|)forrinrows:lines.append(f|{r[quarter]}|{r[amount]}|)lines[,## 汇总,,f- 全年总额{task.artifacts[total]},f- 季度均值{task.artifacts[avg]:.2f},]fs.write(reports/q1_summary.md,\n.join(lines))task.artifacts[report_path]reports/q1_summary.mdreturn报告已写入# 3. 定义验收契约classSalesReportHarness(Harness):def__init__(self,workspace):super().__init__(workspace)self.fsWorkspaceFS(workspace)self.verifierVerifier([FileExists(reports/q1_summary.md,min_bytes100),MarkdownHasHeading(reports/q1_summary.md,level1),ContainsKeywords(reports/q1_summary.md,[季度销售报告,全年总额,Q1,Q4]),])defverify(self,task:Task)-bool:resultself.verifier.verify(self.fs)task.artifacts[verification]resultreturnresult[passed]# 4. 运行harnessSalesReportHarness(workspace)taskTask(goal生成季度销售报告)finalharness.run(task,[step_read_csv,step_aggregate,step_write_report])print(f最终状态:{final.state.value})print(f验收结果:{final.artifacts[verification][passed]})