AI开发基础（第9篇）：Harness Engineering与知识地图

AI 开发基础第9篇Harness Engineering 与知识地图 - 管控整个 Agent 系统适合读者已读完前8篇想了解Agent系统的工程化管控和整体知识体系预计阅读时间35分钟前言从能跑到可管控前8篇我们学会了构建一个完整能力的Agent第1-2篇底层基础API、KV Cache第3篇核心能力Agent Loop Tool Use第4篇推理能力Reasoning Planning第5篇能力扩展Skills MCP第6篇记忆系统Memory RAG第7篇协作架构Subagent Multi-Agent第8篇沟通艺术Prompt Context Engineering但一个能跑的Agent和一个生产可用的Agent系统之间还有一个巨大的鸿沟Harness Engineering 就是填补这个鸿沟的工程。一、Harness Engineering什么是缰绳1.1 概念解释Harness原意是马具中的缰绳。Harness Engineering指的是对Agent系统的管控、约束、监控。Agent就像一匹马。 没有缰绳马随意跑可能跑偏、可能撞到人 有缰绳马在指定范围内跑主人随时可以拉住 Harness就是Agent的缰绳。1.2 五大管控维度维度问题解决方案安全Agent执行了危险操作删数据、发邮件权限控制、审批流程、沙箱环境成本Agent陷入循环Token无限消耗预算限制、超时控制、熔断机制质量Agent输出质量不稳定自动评分、人工审核、A/B测试可观测性Agent出错了不知道哪一步出了问题日志追踪、调用链路、状态快照可恢复性Agent运行到一半崩溃从头再来状态持久化、断点续跑二、安全管控2.1 工具权限分级fromenumimportEnumclassPermissionLevel(Enum):工具权限级别SAFEsafe# 只读无副作用查天气、搜索LOW_RISKlow_risk# 低风险可逆写文件到指定目录、发草稿HIGH_RISKhigh_risk# 高风险需审批发邮件、删数据、发布上线TOOL_PERMISSIONS{get_weather:PermissionLevel.SAFE,search_web:PermissionLevel.SAFE,write_file:PermissionLevel.LOW_RISK,save_draft:PermissionLevel.LOW_RISK,send_email:PermissionLevel.HIGH_RISK,delete_data:PermissionLevel.HIGH_RISK,deploy:PermissionLevel.HIGH_RISK,}asyncdefsafe_tool_call(tool_name:str,args:dict)-str:安全的工具调用levelTOOL_PERMISSIONS.get(tool_name,PermissionLevel.HIGH_RISK)iflevelPermissionLevel.SAFE:returnawaittool_map[tool_name](**args)eliflevelPermissionLevel.LOW_RISK:# 低风险记录日志直接执行log_tool_call(tool_name,args,level)resultawaittool_map[tool_name](**args)log_tool_result(tool_name,result,level)returnresulteliflevelPermissionLevel.HIGH_RISK:# 高风险需要人工审批log_tool_call(tool_name,args,level)# 发送审批请求approvalawaitrequest_human_approval(tool_nametool_name,argsargs,reasonf高风险操作需要人工确认,)ifapproval.approved:resultawaittool_map[tool_name](**args)log_tool_result(tool_name,result,level)returnresultelse:returnf操作被用户拒绝{approval.reason}2.2 输入/输出过滤deffilter_output(text:str)-str:过滤输出中的敏感信息importre# 过滤可能包含的API Keytextre.sub(r(api[_-]?key|token|secret)[\s]*[:][\s]*[\][\w-]{20,}[\],[已脱敏],text,flagsre.IGNORECASE)# 过滤手机号textre.sub(r1[3-9]\d{9},[手机号已脱敏],text)# 过滤邮箱textre.sub(r[\w.-][\w-]\.[\w.],[邮箱已脱敏],text)returntext三、成本管控3.1 Token预算与熔断classTokenBudget:Token预算管理def__init__(self,max_tokens:int,warn_at:float0.8):self.max_tokensmax_tokens self.warn_atwarn_at self.used_tokens0defconsume(self,tokens:int)-bool:消耗Token返回是否还有预算self.used_tokenstokensifself.used_tokensself.max_tokens:print(f Token预算用完{self.used_tokens}/{self.max_tokens})returnFalseifself.used_tokensself.max_tokens*self.warn_at:print(f⚠️ Token预算警告已使用{self.used_tokens/self.max_tokens*100:.0f}%)returnTruepropertydefremaining(self)-int:returnself.max_tokens-self.used_tokens# 在Agent Loop中使用asyncdefagent_loop_with_budget(messages,tools,budget:TokenBudget,max_rounds10):forround_numinrange(max_rounds):responseawaitllm_call(messages,tools)usageresponse.usageifnotbudget.consume(usage.total_tokens):return抱歉本次对话的Token预算已用完。# ... 正常Agent Loop逻辑3.2 循环检测Agent陷入死循环是成本爆炸的常见原因defdetect_loop(messages:list,window:int4)-bool:检测Agent是否在循环重复相同内容iflen(messages)window*2:returnFalse# 比较最近几轮的输出是否高度相似recent_assistant[m[content][:100]forminmessagesifm[role]assistant][-window:]# 如果最近N轮输出的前100字符都一样可能是在循环iflen(set(recent_assistant))1andlen(recent_assistant)2:returnTruereturnFalse# 使用ifdetect_loop(messages):print(⚠️ 检测到Agent循环强制终止)break3.3 超时控制importasyncioasyncdefcall_with_timeout(coro,timeout_seconds:int30):带超时的LLM调用try:returnawaitasyncio.wait_for(coro,timeouttimeout_seconds)exceptasyncio.TimeoutError:print(f⏰ LLM调用超时{timeout_seconds}秒)returnNone四、可观测性4.1 调用链路追踪importuuidimporttimefromdataclassesimportdataclass,fielddataclassclassAgentTrace:Agent调用链路追踪trace_id:strfield(default_factorylambda:str(uuid.uuid4())[:8])spans:listfield(default_factorylist)defstart_span(self,name:str,meta:dictNone):开始一个追踪片段span{id:str(uuid.uuid4())[:8],name:name,start:time.time(),meta:metaor{},}self.spans.append(span)returnspandefend_span(self,span_id:str,result:strNone,error:strNone):结束一个追踪片段forspaninself.spans:ifspan[id]span_id:span[end]time.time()span[duration_ms](span[end]-span[start])*1000span[result]result span[error]errorbreakdefsummary(self)-str:生成追踪摘要lines[f追踪ID:{self.trace_id}]forspaninself.spans:durationspan.get(duration_ms,0)status❌ifspan.get(error)else✅lines.append(f{status}{span[name]}({duration:.0f}ms))ifspan.get(error):lines.append(f 错误:{span[error]})return\n.join(lines)# 使用traceAgentTrace()spantrace.start_span(LLM调用,{model:deepseek-chat,prompt_tokens:1500})responseawaitllm_call(messages,tools)trace.end_span(span[id],resultresponse.choices[0].message.content[:100])spantrace.start_span(工具调用,{tool:get_weather})resultawaitget_weather(city北京)trace.end_span(span[id],resultresult)print(trace.summary())# 输出# 追踪ID: a1b2c3d4# ✅ LLM调用 (1234ms)# ✅ 工具调用 (89ms)4.2 状态快照defsave_state_snapshot(state:dict,trace_id:str,round_num:int):保存Agent状态快照用于调试和恢复importjsonfromdatetimeimportdatetime snapshot{trace_id:trace_id,round:round_num,timestamp:datetime.now().isoformat(),messages:state[messages],tool_calls:state[tool_calls],}filepathftraces/{trace_id}_round_{round_num}.jsonwithopen(filepath,w)asf:json.dump(snapshot,f,ensure_asciiFalse,indent2)五、可恢复性5.1 状态持久化importjsonimportsqlite3fromdatetimeimportdatetimeclassAgentCheckpoint:Agent检查点用于断点续跑def__init__(self,db_path:stragent_checkpoints.db):self.connsqlite3.connect(db_path)self.conn.execute( CREATE TABLE IF NOT EXISTS checkpoints ( session_id TEXT, round_num INTEGER, messages TEXT, state TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (session_id, round_num) ) )self.conn.commit()defsave(self,session_id:str,round_num:int,messages:list,state:dict):保存检查点self.conn.execute(INSERT OR REPLACE INTO checkpoints VALUES (?, ?, ?, ?, ?),(session_id,round_num,json.dumps(messages),json.dumps(state),datetime.now().isoformat()))self.conn.commit()defload(self,session_id:str,round_num:intNone)-tuple:加载检查点ifround_num:rowself.conn.execute(SELECT messages, state FROM checkpoints WHERE session_id? AND round_num?,(session_id,round_num)).fetchone()else:rowself.conn.execute(SELECT messages, state FROM checkpoints WHERE session_id? ORDER BY round_num DESC LIMIT 1,(session_id,)).fetchone()ifrow:returnjson.loads(row[0]),json.loads(row[1])returnNone,None# 使用Agent每次完成一轮后保存检查点checkpointAgentCheckpoint()# 断点续跑messages,statecheckpoint.load(session_idabc123)ifmessages:print(f从第{state.get(round,0)}轮恢复)# 继续Agent Loop六、AI 开发知识地图6.1 9篇内容的关系图┌─────────────┐ │ LLM API │ 第1篇入口 │ (调用基础) │ └──────┬──────┘ │ ┌──────▼──────┐ │ KV Cache │ 第2篇性能原理 │ (推理优化) │ └──────┬──────┘ │ ┌────────────▼────────────┐ │ Agent Loop Tool │ 第3篇核心 │ (让模型能做事) │ └────────────┬────────────┘ │ ┌────────────▼────────────┐ │ Reasoning Planning │ 第4篇思考 │ (让模型想清楚再动手) │ └────────────┬────────────┘ │ ┌────────────▼────────────┐ │ Skills MCP │ 第5篇能力扩展 │ (模块化 标准化) │ └────────────┬────────────┘ │ ┌────────────▼────────────┐ │ Memory RAG │ 第6篇记忆 │ (记住用户 检索知识) │ └────────────┬────────────┘ │ ┌────────────▼────────────┐ │ Subagent Multi-Agent │ 第7篇协作 │ (分工 多智能体) │ └────────────┬────────────┘ │ ┌────────────▼────────────┐ │ Prompt Context Eng. │ 第8篇沟通 │ (管理模型的全部输入) │ └────────────┬────────────┘ │ ┌────────────▼────────────┐ │ Harness 知识地图 │ 第9篇管控 │ (安全/成本/可观测/恢复) │ └─────────────────────────┘6.2 核心概念关联Agent LLM Loop Tools Memory Reasoning Skills LLM ← 第1篇API调用、第2篇KV Cache性能 Loop ← 第3篇Agent Loop、第7篇Multi-Agent协作 Tools ← 第3篇Tool Use、第5篇Skills/MCP标准化 Memory ← 第6篇三层记忆 RAG Reasoning ← 第4篇CoT ReAct Plan-and-Execute Skills ← 第5篇能力模块化、第7篇角色分工 Prompt ← 第8篇提示词 上下文工程 Harness ← 第9篇安全/成本/可观测/恢复6.3 学习路径建议入门路径1-2周第1篇 LLM API → 第3篇 Agent Loop → 动手写一个简单Agent进阶路径3-4周第4篇 Reasoning → 第5篇 Skills/MCP → 第6篇 Memory → 给Agent加上推理能力、工具、记忆高级路径5-6周第7篇 Multi-Agent → 第8篇 Context Engineering → 第9篇 Harness → 构建生产级Agent系统6.4 每篇的核心收获篇章一句话总结第1篇LLM API是所有上层框架的底层理解参数比会调框架更重要第2篇KV Cache让推理快2-3倍但它吃显存PagedAttention是标准解法第3篇Agent Loop LLM判断→工具调用→结果返回→继续循环50行代码就能跑第4篇CoT让模型写出来再回答推理模型自带推理能力但贵且慢第5篇MCP是工具调用的USB接口一次开发处处用第6篇Memory记住你RAG知道更多两者结合效果最好第7篇先单Agent跑通发现瓶颈再拆Subagent复杂场景用Multi-Agent第8篇Context Engineering Prompt Engineering管理全部输入而不是只写提示词第9篇生产环境必须有安全管控、成本限制、可观测性和断点恢复七、下一步学什么学完这9篇你已经掌握了AI应用开发的核心知识。接下来可以深入的方向方向推荐资源Agent框架实战LangGraph、CrewAI、AutoGenRAG深度优化混合检索、重排序、知识图谱模型微调LoRA、QLoRA低成本微调部署运维vLLM、TGI、Triton Inference Server多模态视觉Agent、语音Agent、视频理解评估体系LLM-as-Judge、人工评估、A/B测试八、本章总结你学到了什么Harness Engineering安全管控权限分级、成本管控Token预算循环检测超时、可观测性调用链路追踪、可恢复性状态持久化断点续跑知识地图9篇内容从底层API到系统管控的完整知识体系学习路径入门APILoop→ 进阶推理SkillsMemory→ 高级Multi-AgentContextHarness核心公式Agent LLM Loop Tools Memory Reasoning Skills最后一句话AI应用开发的核心不是调API而是理解Agent如何思考、如何行动、如何记忆、如何协作、如何被管控。掌握了这些你就能构建真正有用的Agent系统。参考资料Anthropic Building Effective Agentshttps://www.anthropic.com/research/building-effective-agentsLangGraph文档https://langchain-ai.github.io/langgraph/OpenAI Safety Best Practiceshttps://platform.openai.com/docs/guides/safetyPrompt Securityhttps://owasp.org/www-project-top-10-for-large-language-model-applications/上一篇第8篇 Prompt Engineering 与 Context Engineering系列完结