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 五大管控维度

二、安全管控

2.1 工具权限分级

fromenumimportEnumclassPermissionLevel(Enum):"""工具权限级别"""SAFE="safe"# 只读，无副作用（查天气、搜索）LOW_RISK="low_risk"# 低风险，可逆（写文件到指定目录、发草稿）HIGH_RISK="high_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:"""安全的工具调用"""level=TOOL_PERMISSIONS.get(tool_name,PermissionLevel.HIGH_RISK)iflevel==PermissionLevel.SAFE:returnawaittool_map[tool_name](**args)eliflevel==PermissionLevel.LOW_RISK:# 低风险：记录日志，直接执行log_tool_call(tool_name,args,level)result=awaittool_map[tool_name](**args)log_tool_result(tool_name,result,level)returnresulteliflevel==PermissionLevel.HIGH_RISK:# 高风险：需要人工审批log_tool_call(tool_name,args,level)# 发送审批请求approval=awaitrequest_human_approval(tool_name=tool_name,args=args,reason=f"高风险操作需要人工确认",)ifapproval.approved:result=awaittool_map[tool_name](**args)log_tool_result(tool_name,result,level)returnresultelse:returnf"操作被用户拒绝：{approval.reason}"

2.2 输入/输出过滤

deffilter_output(text:str)->str:"""过滤输出中的敏感信息"""importre# 过滤可能包含的API Keytext=re.sub(r'(api[_-]?key|token|secret)["\s]*[:=]["\s]*["\'][\w-]{20,}["\']','[已脱敏]',text,flags=re.IGNORECASE)# 过滤手机号text=re.sub(r'1[3-9]\d{9}','[手机号已脱敏]',text)# 过滤邮箱text=re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+','[邮箱已脱敏]',text)returntext

三、成本管控

3.1 Token预算与熔断

classTokenBudget:"""Token预算管理"""def__init__(self,max_tokens:int,warn_at:float=0.8):self.max_tokens=max_tokens self.warn_at=warn_at self.used_tokens=0defconsume(self,tokens:int)->bool:"""消耗Token，返回是否还有预算"""self.used_tokens+=tokensifself.used_tokens>=self.max_tokens:print(f"🚫 Token预算用完：{self.used_tokens}/{self.max_tokens}")returnFalseifself.used_tokens>=self.max_tokens*self.warn_at:print(f"⚠️ Token预算警告：已使用{self.used_tokens/self.max_tokens*100:.0f}%")returnTrue@propertydefremaining(self)->int:returnself.max_tokens-self.used_tokens# 在Agent Loop中使用asyncdefagent_loop_with_budget(messages,tools,budget:TokenBudget,max_rounds=10):forround_numinrange(max_rounds):response=awaitllm_call(messages,tools)usage=response.usageifnotbudget.consume(usage.total_tokens):return"抱歉，本次对话的Token预算已用完。"# ... 正常Agent Loop逻辑

3.2 循环检测

Agent陷入死循环是成本爆炸的常见原因：

defdetect_loop(messages:list,window:int=4)->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循环，强制终止")break

3.3 超时控制

importasyncioasyncdefcall_with_timeout(coro,timeout_seconds:int=30):"""带超时的LLM调用"""try:returnawaitasyncio.wait_for(coro,timeout=timeout_seconds)exceptasyncio.TimeoutError:print(f"⏰ LLM调用超时（{timeout_seconds}秒）")returnNone

四、可观测性

4.1 调用链路追踪

importuuidimporttimefromdataclassesimportdataclass,field@dataclassclassAgentTrace:"""Agent调用链路追踪"""trace_id:str=field(default_factory=lambda:str(uuid.uuid4())[:8])spans:list=field(default_factory=list)defstart_span(self,name:str,meta:dict=None):"""开始一个追踪片段"""span={"id":str(uuid.uuid4())[:8],"name":name,"start":time.time(),"meta":metaor{},}self.spans.append(span)returnspandefend_span(self,span_id:str,result:str=None,error:str=None):"""结束一个追踪片段"""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:duration=span.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)# 使用trace=AgentTrace()span=trace.start_span("LLM调用",{"model":"deepseek-chat","prompt_tokens":1500})response=awaitllm_call(messages,tools)trace.end_span(span["id"],result=response.choices[0].message.content[:100])span=trace.start_span("工具调用",{"tool":"get_weather"})result=awaitget_weather(city="北京")trace.end_span(span["id"],result=result)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"],}filepath=f"traces/{trace_id}_round_{round_num}.json"withopen(filepath,"w")asf:json.dump(snapshot,f,ensure_ascii=False,indent=2)

五、可恢复性

5.1 状态持久化

importjsonimportsqlite3fromdatetimeimportdatetimeclassAgentCheckpoint:"""Agent检查点：用于断点续跑"""def__init__(self,db_path:str="agent_checkpoints.db"):self.conn=sqlite3.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:int=None)->tuple:"""加载检查点"""ifround_num:row=self.conn.execute("SELECT messages, state FROM checkpoints WHERE session_id=? AND round_num=?",(session_id,round_num)).fetchone()else:row=self.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每次完成一轮后保存检查点checkpoint=AgentCheckpoint()# 断点续跑messages,state=checkpoint.load(session_id="abc123")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 每篇的核心收获

七、下一步学什么？

学完这9篇，你已经掌握了AI应用开发的核心知识。接下来可以深入的方向：

八、本章总结

你学到了什么：

1. Harness Engineering：安全管控（权限分级）、成本管控（Token预算+循环检测+超时）、可观测性（调用链路追踪）、可恢复性（状态持久化+断点续跑）
2. 知识地图：9篇内容从底层API到系统管控的完整知识体系
3. 学习路径：入门（API+Loop）→ 进阶（推理+Skills+Memory）→ 高级（Multi-Agent+Context+Harness）
4. 核心公式：Agent = LLM + Loop + Tools + Memory + Reasoning + Skills

最后一句话：

AI应用开发的核心不是"调API"，而是理解Agent如何思考、如何行动、如何记忆、如何协作、如何被管控。掌握了这些，你就能构建真正有用的Agent系统。

参考资料

1. Anthropic Building Effective Agents：https://www.anthropic.com/research/building-effective-agents
2. LangGraph文档：https://langchain-ai.github.io/langgraph/
3. OpenAI Safety Best Practices：https://platform.openai.com/docs/guides/safety
4. Prompt Security：https://owasp.org/www-project-top-10-for-large-language-model-applications/

上一篇：第8篇 Prompt Engineering 与 Context Engineering
系列完结🎉