1. 项目概述为什么自主Agent需要“安全护栏”最近在折腾AI Agent开发的朋友估计都绕不开一个词“失控”。我见过太多这样的场景了一个设计用来处理客户咨询的Agent聊着聊着突然开始生成一些不合规的营销话术一个负责数据整理的Agent在执行复杂任务链时因为一个意外的API响应陷入了死循环疯狂调用外部服务直到把额度刷爆。这些还不是最可怕的最让人后背发凉的是一个拥有工具调用能力的Agent可能会在未经授权的情况下执行一些具有潜在风险的操作比如向外部系统发送未经审核的数据或者执行一条本应被拦截的数据库删除指令。这就是我们今天要深入探讨的核心Harness Engineering。你可以把它理解为给那些越来越“聪明”、也越来越“自主”的AI Agent装上的一套安全护栏和缰绳系统。它不是一个具体的工具而是一整套工程实践和设计范式核心目标是在赋予Agent自主性的同时确保其行为始终处于可控、可预测、符合预期的边界之内。简单说就是让Agent既跑得快又不会“脱缰”。为什么现在这个话题这么热因为Agent的能力边界正在急速扩张。早期的聊天机器人其交互是封闭的、回合制的。而现代基于LangGraph、ReAct等框架构建的Agent具备了状态记忆、工具调用、多步推理和长期规划的能力。它们更像是一个在数字世界里自主行动的“智能体”其行动轨迹不再是线性的而是一个可能包含分支、循环、并行任务的复杂图。这种强大的自主性是一把双刃剑。没有约束的自主等同于混乱和风险。因此Harness Engineering应运而生。它融合了软件工程、人机交互和安全领域的理念旨在通过系统性的设计在Agent的自主决策循环中嵌入检查点、监控器和干预机制。这不仅仅是“如果出错就报错”那么简单它涉及到在Agent行动的事前、事中、事后全周期进行治理。接下来我们就拆解一下要构建这样一套“安全护栏”我们的核心思路应该是什么。1.1 核心需求解析从“黑盒”到“可观测、可干预、可回溯”要给Agent装护栏首先得看清Agent在“想”什么、“做”什么。传统软件的错误是显性的崩溃、异常而Agent的“错误”可能是隐性的逻辑偏差、目标蠕变。因此Harness Engineering的首要需求是实现对Agent内部状态的深度可观测性。这不仅仅是打印日志。我们需要捕获几个关键维度思维过程Agent的推理链Chain-of-Thought。它为什么选择这个工具基于哪些上下文它的“内心独白”是什么这对于诊断逻辑错误至关重要。行动轨迹Agent调用了哪些工具传入的参数是什么工具的返回结果是什么这构成了Agent的行为序列。状态演变在LangGraph这类基于状态机的框架中State对象是如何随着每个节点Node的执行而变化的哪些关键字段被修改了资源消耗调用次数、Token使用量、执行时长。这是预防滥用和成本失控的财务护栏。有了可观测性下一步就是可干预。我们需要在Agent可能“越界”的关键节点预设干预通道。这通常通过“人在回路”Human-in-the-loop模式实现。例如当Agent准备执行一个高风险操作如“发送邮件”、“删除记录”前自动暂停工作流将决策上下文提交给人工审核。或者当Agent的连续失败次数超过阈值时自动中止流程并告警。最后所有上述信息必须能够被可回溯。当出现问题时我们能像查看飞机黑匣子一样完整复现Agent从启动到出错的整个决策历程和状态变化从而精准定位问题根因。所以Harness Engineering的底层逻辑是将Agent从一个“输入-输出”的黑盒转变为一个“透明且带有紧急制动阀”的灰盒系统。其设计必须与Agent框架如LangGraph深度集成而非事后补救。2. 核心架构设计基于LangGraph构建安全可控的Agent工作流LangGraph之所以成为复杂Agent开发的首选正是因为它用“图”的思维清晰地定义了工作流这为实施Harness Engineering提供了绝佳的骨架。我们的安全护栏本质上是在这个骨架上添加一系列监控节点、校验边和状态守卫。2.1 状态State设计定义安全边界的第一道防线在LangGraph中State是所有节点共享的上下文。我们的安全设计从这里开始。一个健壮的、面向安全的State应该包含以下几部分from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 核心任务信息 messages: Annotated[List, add_messages] # 对话历史 user_query: str # 用户原始输入 current_goal: str # 当前分解后的子目标 # 工具执行记录用于可观测性 action_history: List[dict] # 记录每次工具调用{tool: name, input: params, output: result, step: n} error_log: List[str] # 记录运行过程中的错误或警告 # 安全与管控上下文 requires_human_approval: bool # 当前步骤是否需要人工审批 human_feedback: str # 人工审批的输入或反馈 execution_budget: float # 剩余执行预算如Token成本、API调用次数 risk_level: str # 当前任务评估的风险等级low, medium, high # 流程控制标志 is_approved: bool # 人工是否已批准 should_continue: bool # 是否继续执行 max_steps: int # 最大执行步数防死循环 current_step: int # 当前步数计数器设计要点解析action_history和error_log是可观测性的基石。每个工具节点执行后都必须向其追加记录。requires_human_approval和is_approved是可干预的开关。它们将决定工作流是走向人工审核节点还是继续自动执行。execution_budget和max_steps是资源护栏。在每个节点开始前检查预算防止成本失控或无限循环。实操心得State的设计要遵循“最小化但充分”原则。不要把所有东西都塞进去只放对流程控制和事后分析真正有用的字段。TypedDict和Annotated能提供良好的类型提示但更重要的是要在项目初期就团队对齐每个字段的精确含义和更新时机避免后续出现状态污染。2.2 节点Node与边Edge的安全增强在LangGraph中节点是执行单元边是流转逻辑。安全增强需要渗透到这两者中。安全节点示例人工审批节点这个节点不执行具体任务只负责暂停流程、等待输入、并更新状态。def human_approval_node(state: AgentState) - AgentState: 人工审批节点。在实际应用中这里会连接到一个Web界面或消息队列。 # 1. 在实际系统中此处会触发一个通知如短信、邮件、Slack消息将审批上下文发送给负责人。 # 上下文包括state[user_query], state[current_goal], state[action_history][-1]待审批的操作 print(f[审批请求] 待执行操作: {state[current_goal]}) print(f 操作历史: {state[action_history][-1] if state[action_history] else 无}) # 2. 模拟等待人工输入。真实场景中这里会等待一个外部API回调或数据库状态更新。 # 假设我们通过一个模拟函数获取审批结果 feedback, approved simulate_human_review(state) # 3. 更新状态 state[human_feedback] feedback state[is_approved] approved if approved: print([系统] 人工审批通过继续执行。) else: print(f[系统] 人工审批驳回。反馈{feedback}) state[should_continue] False # 终止流程 return state def simulate_human_review(state): # 这是一个模拟函数。真实场景中审批逻辑可能更复杂。 # 例如可以根据 risk_level 自动判断高风险操作默认送审。 if state[risk_level] high: # 模拟人工审核后批准 return 操作已审核风险可控批准执行。, True else: # 模拟自动批准 return 低风险操作自动批准。, True安全边条件动态路由逻辑边Edge决定了工作流的走向。我们可以根据安全状态来动态路由。from langgraph.graph import END def should_require_human_approval(state: AgentState) - str: 判断下一个节点是否需要经过人工审批。 # 规则1如果上一个工具调用被标记为高风险 last_action state[action_history][-1] if state[action_history] else None if last_action and last_action.get(risk_tag) high_risk: return route_to_approval # 规则2如果执行预算低于阈值 if state.get(execution_budget, 100) 10: return route_to_approval # 也可以路由到告警或终止节点 # 规则3如果用户查询中包含敏感关键词需提前定义列表 sensitive_keywords [删除, 重置, 转账, root] if any(keyword in state[user_query] for keyword in sensitive_keywords): return route_to_approval # 默认继续自动执行 return continue_auto def check_after_approval(state: AgentState) - str: 在人工审批节点之后决定下一步去向。 if not state.get(is_approved, False): # 如果被驳回直接结束工作流 return end else: # 如果批准则返回到主工作流继续执行 return proceed通过这种方式我们将安全策略如“高风险操作需审核”、“预算不足需告警”编码到了工作流的路由逻辑中使安全性与业务流程无缝融合。注意事项定义路由条件函数时务必保持其纯净和高效。它们会在每个决策点被调用不应有副作用如修改State或执行耗时操作如网络请求。复杂的风险评估逻辑应该放在专门的“风险评估节点”中该节点运行后将结果如risk_level: ‘high’写入State再由路由条件函数读取。3. 实操构建一个带有安全护栏的查询执行Agent让我们构建一个具体的例子一个可以查询数据库但删除操作必须经过人工审批的Agent。我们将使用LangGraph来编排流程。3.1 定义工具与风险标注首先定义Agent可以使用的工具并为每个工具标注风险等级。from langchain.tools import tool from typing import Optional tool def query_database(sql: str) - str: 执行数据库查询SELECT语句。 # 这里是模拟实现 print(f[安全日志] 执行查询: {sql}) # 模拟执行... return f查询结果: 模拟数据 from {sql} tool def delete_from_database(table: str, condition: Optional[str] None) - str: 从数据库删除数据。高风险操作。 # 这个工具本身不直接执行而是生成待审批的操作指令。 # 真正的执行会在人工审批后由另一个“安全执行”节点完成。 delete_sql fDELETE FROM {table} if condition: delete_sql f WHERE {condition} print(f[安全日志] 生成删除指令待审批: {delete_sql}) return delete_sql # 注意这里返回的是SQL指令字符串而非执行结果 # 工具列表并附带元数据如风险标签 TOOLS [query_database, delete_from_database] TOOL_RISK_MAP { query_database: low, delete_from_database: high, # 标记为高风险 }3.2 构建主工作流图我们将构建一个包含“计划”、“执行”、“审批”、“安全执行”节点的图。from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolExecutor, ToolNode from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.messages import SystemMessage, HumanMessage # 注意以下为简化示意实际构建需结合LangGraph最新API # 初始化模型和工具执行器 llm ChatOpenAI(modelgpt-4o-mini, temperature0) tool_executor ToolExecutor(TOOLS) # 1. 规划节点分析用户意图决定使用哪个工具及参数 def planning_node(state: AgentState): 分析用户查询规划工具调用。 messages [ SystemMessage(content你是一个数据库助手。请分析用户请求决定是进行‘查询’还是‘删除’。对于删除你必须生成准确的SQL DELETE语句。), HumanMessage(contentstate[user_query]) ] response llm.invoke(messages) # 这里简化处理实际应解析LLM响应为结构化指令 # 假设响应中包含了工具调用决定 planned_action { tool: delete_from_database if 删除 in state[user_query] else query_database, input: {sql: SELECT * FROM users WHERE id1} # 简化示例 } state[current_goal] f调用工具 [{planned_action[tool]}] # 将规划结果存入状态供下一个节点使用 state[_planned_action] planned_action return state # 2. 有条件工具调用节点执行低风险工具 def tool_node(state: AgentState): 执行工具调用并记录历史。 planned state.get(_planned_action) if not planned: state[error_log].append(规划节点未提供执行指令。) return state tool_name planned[tool] tool_input planned[input] # 执行前检查如果是高风险工具则设置审批标志并跳过直接执行 if TOOL_RISK_MAP.get(tool_name) high: state[requires_human_approval] True state[_pending_high_risk_action] {tool: tool_name, input: tool_input} print(f[安全拦截] 高风险工具 {tool_name} 被拦截等待人工审批。) # 此时不执行工具直接返回让路由逻辑将其导向审批节点 return state # 执行低风险工具 try: result tool_executor.invoke({tool_name: tool_input}) # 记录到行动历史 state[action_history].append({ step: state[current_step], tool: tool_name, input: tool_input, output: result, risk_tag: TOOL_RISK_MAP.get(tool_name, unknown) }) state[last_tool_result] result except Exception as e: state[error_log].append(f工具执行失败: {e}) finally: state[current_step] 1 return state # 3. 人工审批节点复用之前定义的 human_approval_node # 4. 安全执行节点仅在人工批准后执行高风险操作 def safe_execution_node(state: AgentState): 执行经过审批的高风险操作。 if not state.get(is_approved): state[error_log].append(试图执行未批准的高风险操作。) return state pending_action state.get(_pending_high_risk_action) if not pending_action: state[error_log].append(审批通过但无待执行的高风险操作。) return state tool_name pending_action[tool] tool_input pending_action[input] print(f[安全执行] 正在执行已审批的高风险操作: {tool_name} with {tool_input}) # 这里是真正执行危险操作的地方 # 例如连接到数据库执行DELETE语句。 # 为了演示我们只是模拟执行并记录。 try: # 模拟执行成功 result f[模拟] 高风险操作 {tool_name} 已安全执行。输入: {tool_input} state[action_history].append({ step: state[current_step], tool: tool_name, input: tool_input, output: result, risk_tag: high, approved: True }) state[last_tool_result] result print(result) except Exception as e: state[error_log].append(f安全执行失败: {e}) finally: # 清理待执行动作 state[_pending_high_risk_action] None state[current_step] 1 return state # 5. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planning_node) workflow.add_node(tool_call, tool_node) workflow.add_node(human_approval, human_approval_node) workflow.add_node(safe_executor, safe_execution_node) # 设置边和路由 workflow.add_edge(START, planner) workflow.add_edge(planner, tool_call) # 关键从 tool_call 出来的条件路由 def route_after_tool_call(state: AgentState) - str: if state.get(requires_human_approval, False): return human_approval else: return generate_response # 假设有一个生成最终响应的节点这里简化为END workflow.add_conditional_edges( tool_call, route_after_tool_call, { human_approval: human_approval, generate_response: END # 简化直接结束 } ) # 从人工审批节点出来的路由 workflow.add_conditional_edges( human_approval, check_after_approval, # 使用前面定义的函数 { end: END, proceed: safe_executor } ) workflow.add_edge(safe_executor, END) # 编译图 app workflow.compile()这个工作流实现了核心安全逻辑规划节点分析意图工具调用节点执行低风险操作并拦截高风险操作将其路由至人工审批节点。审批通过后再由一个专用的“安全执行节点”来最终完成高风险操作。整个过程的状态、行动、审批记录都被完整追踪。踩坑实录在早期版本中我曾尝试在同一个tool_node里既做判断又执行所有工具。这导致了状态管理的混乱特别是当工具执行失败时回滚和重试逻辑变得极其复杂。后来才演变为现在的“拦截-审批-安全执行”三段式设计职责分离清晰状态流转也更容易调试。4. 监控、评估与迭代让护栏越来越智能部署了安全护栏并非一劳永逸。我们需要持续监控Agent的表现评估护栏的有效性并迭代安全策略。4.1 构建监控与日志系统我们需要将之前State中的action_history和error_log持久化到外部系统如数据库、Elasticsearch以便进行分析。import json from datetime import datetime def persist_agent_run(session_id: str, final_state: AgentState, conclusion: str): 将一次Agent运行的完整上下文持久化。 log_entry { session_id: session_id, timestamp: datetime.utcnow().isoformat(), user_query: final_state.get(user_query), conclusion: conclusion, # 成功, 人工驳回, 错误中断, 预算耗尽 action_history: final_state.get(action_history, []), error_log: final_state.get(error_log, []), final_state_snapshot: {k: v for k, v in final_state.items() if k not in [messages, action_history, error_log]} # 避免循环引用 } # 这里可以写入数据库、文件或日志服务 print(f[审计日志] Session {session_id}: {conclusion}) # 示例写入JSON文件 with open(fagent_audit_log_{session_id}.json, w) as f: json.dump(log_entry, f, indent2, ensure_asciiFalse)在每次Agent运行结束时无论成功失败都调用此函数。这为我们提供了事后审计和模型行为分析的宝贵数据。4.2 定义安全评估指标有了数据我们就可以定义一些关键指标KPIs来衡量安全护栏的效果人工干预率触发requires_human_approval的会话占比。过高可能意味着规则太严格影响效率过低则可能意味着规则有漏洞。审批通过率人工审批节点中操作被批准的比例。这反映了Agent提出的高风险操作有多少是真正合理且必要的。平均审批耗时从触发审批到收到反馈的平均时间。这直接影响用户体验和流程效率。异常逃逸率在未触发任何安全机制如审批、错误的情况下最终结果仍不符合预期的会话比例。这是衡量护栏完整性的终极指标。成本合规率execution_budget未被耗尽的会话比例。定期如每周分析这些指标可以帮助我们量化安全投入的效果并发现优化点。4.3 迭代安全策略从规则到模型最初的规则可能是硬编码的如“工具名包含‘delete’就送审”。随着数据积累我们可以让安全策略变得更智能规则引擎优化基于历史审批数据发现误报低风险被送审和漏报高风险未送审的模式调整关键词列表或风险判断逻辑。引入预测模型训练一个简单的分类模型基于用户查询、历史动作等特征预测当前操作的风险概率。将基于规则的路由升级为“基于规则模型评分”的混合路由。def advanced_risk_assessment(state: AgentState) - float: 基于简单特征的风险评分示例。 score 0.0 query state[user_query].lower() # 特征1查询中包含破坏性动词 destructive_verbs [删除, 丢弃, 清空, 覆盖, 修改] if any(verb in query for verb in destructive_verbs): score 0.4 # 特征2近期同一会话中失败次数多 recent_errors len([e for e in state[error_log][-3:] if 失败 in e]) score recent_errors * 0.2 # 特征3操作涉及核心数据表需维护核心表列表 if users in query or orders in query: score 0.3 return min(score, 1.0) # 归一化到0-1 def should_require_approval_advanced(state): score advanced_risk_assessment(state) if score 0.7: # 高风险阈值 return route_to_approval elif score 0.3: # 中风险也许可以记录日志并继续或路由到更轻量的确认节点 state[error_log].append(f中风险操作预警评分: {score:.2f}) return continue_auto else: return continue_auto闭环学习将人工审批的反馈human_feedback作为强化学习的奖励信号或者用于微调规划节点Planner的LLM让Agent逐渐学会在提出请求时就规避高风险行为。个人体会Harness Engineering不是一个可以“一次性完成”的模块而是一个需要持续运营的“系统”。它和你的核心业务逻辑一样重要甚至更重要。我建议在团队中设立明确的“安全护栏负责人”像对待生产环境监控告警一样定期审查安全日志和指标。最开始的规则可能很粗糙会误拦很多正常操作但没关系快速迭代是关键。每一次人工审批不仅是一次安全拦截更是一次对Agent行为的宝贵标注利用好这些数据你的护栏会越来越聪明。5. 常见问题与排查技巧实录在实际部署带有安全护栏的Agent时你肯定会遇到各种意想不到的情况。下面是我踩过的一些坑和总结的排查思路。5.1 问题Agent陷入“审批-执行-再审批”的死循环现象一个高风险操作被批准执行后在后续步骤中又触发了同样的审批规则导致流程卡住。根因分析这通常是因为状态State没有在审批后正确清理。例如requires_human_approval标志在审批通过后没有被重置为False或者_pending_high_risk_action没有被清空导致工作流在下一个循环判断时依然认为需要审批。解决方案确保状态机纯净在safe_execution_node执行完高风险操作后必须重置所有与本次审批相关的临时状态标志。def safe_execution_node(state: AgentState): # ... 执行操作 ... # 执行完毕后清理审批相关状态 state[requires_human_approval] False state[is_approved] False # 重置批准状态为下一次独立审批做准备 state[_pending_high_risk_action] None state[human_feedback] return state设计自包含的审批上下文另一种思路是将每次审批及其对应的待执行动作封装成一个独立的对象并赋予一个唯一的approval_id。当该ID对应的动作执行完毕后系统就认为该审批上下文已关闭不会影响后续流程。5.2 问题人工审批响应超时导致工作流僵死现象流程进入human_approval_node后一直在等待外部系统的回调如用户点击审批按钮如果用户一直不处理这个工作流实例就会一直占用资源等待。根因分析LangGraph的普通节点是同步执行的。如果将“等待人工输入”这个可能耗时很长的操作放在一个同步节点里必然会阻塞。解决方案采用异步工作流与外部事件驱动。改造审批节点为“暂停点”human_approval_node不再等待而是执行以下操作 a. 生成一个唯一的pause_id如UUID。 b. 将当前完整的State序列化后持久化到数据库如Redis、PostgreSQL并以pause_id为键。 c. 向审批系统如邮件、钉钉、自定义Webhook发送请求附上pause_id和审批内容。 d. 节点返回一个特殊指令如{should_pause: True, pause_id: pause_id}并让工作流路由到一个“暂停”状态。提供继续执行的端点暴露一个REST API端点例如POST /webhook/continue/{pause_id}。当用户在审批界面点击“批准”或“驳回”时审批系统调用此端点并附上决定。恢复执行该API端点根据pause_id从数据库中恢复保存的State更新is_approved和human_feedback字段然后重新将State注入到LangGraph工作流中并从“暂停”后的下一个节点如safe_executor继续执行。这需要LangGraph支持检查点Checkpoint功能幸运的是这正是LangGraph的核心优势之一它原生支持将工作流状态持久化并在之后恢复。5.3 问题安全规则过于复杂难以维护和调试现象should_require_human_approval函数里塞满了大量的if-else规则逻辑纠缠添加新规则时战战兢兢生怕影响旧逻辑。根因分析将业务逻辑、安全策略、路由逻辑全部耦合在代码中。解决方案采用策略模式Strategy Pattern或规则引擎。策略模式将不同的风险判断逻辑抽象成独立的类或函数。例如定义BudgetRule、KeywordRule、ToolRiskRule等。路由条件函数只需遍历一个规则列表依次评估即可。class SafetyRule: def evaluate(self, state: AgentState) - (bool, str): # (是否触发, 规则名称) raise NotImplementedError class KeywordRule(SafetyRule): def __init__(self, keywords): self.keywords keywords def evaluate(self, state): query state[user_query].lower() if any(kw in query for kw in self.keywords): return True, keyword_rule return False, class BudgetRule(SafetyRule): def evaluate(self, state): if state.get(execution_budget, 100) 5: return True, budget_rule return False, # 配置化规则列表 SAFETY_RULES [ KeywordRule([删除, root, sudo]), BudgetRule(), # ... 可以轻松添加新规则 ] def should_require_approval(state): for rule in SAFETY_RULES: triggered, rule_name rule.evaluate(state) if triggered: print(f[安全规则触发] {rule_name}) return route_to_approval return continue_auto规则引擎对于极其复杂的策略可以考虑使用像Drools、Easy Rules这样的轻量级规则引擎甚至将规则配置存储在数据库中实现动态更新无需重启服务。5.4 问题监控日志数据量巨大难以定位问题现象action_history记录了每个步骤的详细输入输出当处理大量数据或复杂任务时日志体积爆炸搜索特定错误如同大海捞针。解决方案实施结构化日志分级和采样。分级记录不是所有信息都需要全量记录。DEBUG级记录完整的思维链、工具调用的原始请求和响应。此级别日志量最大只在调试特定会话时开启。INFO级记录关键决策点、规则触发、审批请求和结果。这是日常监控的主要级别。WARN/ERROR级记录异常、失败和违反安全策略的事件。必须全量记录并配置告警。会话采样对于DEBUG级日志可以按一定比例如1%随机采样记录或者只记录标记为“重要”的会话如来自内部测试用户、或触发了WARN以上级别事件的会话。聚合分析使用ELKElasticsearch, Logstash, Kibana或类似栈对INFO和ERROR级日志进行索引。可以快速筛选出“所有触发BudgetRule的会话”或“所有最终状态为人工驳回的会话”进行聚合分析。构建一个带安全护栏的Agent系统就像训练一个天赋异禀但涉世未深的实习生。你需要给他清晰的指令规划赋予他工具能力但在他可能犯大错的关键决策点上高风险操作你必须设置检查站审批。同时你还得在他身后安装摄像头监控日志定期复盘他的工作记录评估指标并据此更新你的管理手册迭代策略。这个过程充满挑战但却是将AI Agent安全、可靠地应用于生产环境的必经之路。
1. 项目概述为什么自主Agent需要“安全护栏”最近在折腾AI Agent开发的朋友估计都绕不开一个词“失控”。我见过太多这样的场景了一个设计用来处理客户咨询的Agent聊着聊着突然开始生成一些不合规的营销话术一个负责数据整理的Agent在执行复杂任务链时因为一个意外的API响应陷入了死循环疯狂调用外部服务直到把额度刷爆。这些还不是最可怕的最让人后背发凉的是一个拥有工具调用能力的Agent可能会在未经授权的情况下执行一些具有潜在风险的操作比如向外部系统发送未经审核的数据或者执行一条本应被拦截的数据库删除指令。这就是我们今天要深入探讨的核心Harness Engineering。你可以把它理解为给那些越来越“聪明”、也越来越“自主”的AI Agent装上的一套安全护栏和缰绳系统。它不是一个具体的工具而是一整套工程实践和设计范式核心目标是在赋予Agent自主性的同时确保其行为始终处于可控、可预测、符合预期的边界之内。简单说就是让Agent既跑得快又不会“脱缰”。为什么现在这个话题这么热因为Agent的能力边界正在急速扩张。早期的聊天机器人其交互是封闭的、回合制的。而现代基于LangGraph、ReAct等框架构建的Agent具备了状态记忆、工具调用、多步推理和长期规划的能力。它们更像是一个在数字世界里自主行动的“智能体”其行动轨迹不再是线性的而是一个可能包含分支、循环、并行任务的复杂图。这种强大的自主性是一把双刃剑。没有约束的自主等同于混乱和风险。因此Harness Engineering应运而生。它融合了软件工程、人机交互和安全领域的理念旨在通过系统性的设计在Agent的自主决策循环中嵌入检查点、监控器和干预机制。这不仅仅是“如果出错就报错”那么简单它涉及到在Agent行动的事前、事中、事后全周期进行治理。接下来我们就拆解一下要构建这样一套“安全护栏”我们的核心思路应该是什么。1.1 核心需求解析从“黑盒”到“可观测、可干预、可回溯”要给Agent装护栏首先得看清Agent在“想”什么、“做”什么。传统软件的错误是显性的崩溃、异常而Agent的“错误”可能是隐性的逻辑偏差、目标蠕变。因此Harness Engineering的首要需求是实现对Agent内部状态的深度可观测性。这不仅仅是打印日志。我们需要捕获几个关键维度思维过程Agent的推理链Chain-of-Thought。它为什么选择这个工具基于哪些上下文它的“内心独白”是什么这对于诊断逻辑错误至关重要。行动轨迹Agent调用了哪些工具传入的参数是什么工具的返回结果是什么这构成了Agent的行为序列。状态演变在LangGraph这类基于状态机的框架中State对象是如何随着每个节点Node的执行而变化的哪些关键字段被修改了资源消耗调用次数、Token使用量、执行时长。这是预防滥用和成本失控的财务护栏。有了可观测性下一步就是可干预。我们需要在Agent可能“越界”的关键节点预设干预通道。这通常通过“人在回路”Human-in-the-loop模式实现。例如当Agent准备执行一个高风险操作如“发送邮件”、“删除记录”前自动暂停工作流将决策上下文提交给人工审核。或者当Agent的连续失败次数超过阈值时自动中止流程并告警。最后所有上述信息必须能够被可回溯。当出现问题时我们能像查看飞机黑匣子一样完整复现Agent从启动到出错的整个决策历程和状态变化从而精准定位问题根因。所以Harness Engineering的底层逻辑是将Agent从一个“输入-输出”的黑盒转变为一个“透明且带有紧急制动阀”的灰盒系统。其设计必须与Agent框架如LangGraph深度集成而非事后补救。2. 核心架构设计基于LangGraph构建安全可控的Agent工作流LangGraph之所以成为复杂Agent开发的首选正是因为它用“图”的思维清晰地定义了工作流这为实施Harness Engineering提供了绝佳的骨架。我们的安全护栏本质上是在这个骨架上添加一系列监控节点、校验边和状态守卫。2.1 状态State设计定义安全边界的第一道防线在LangGraph中State是所有节点共享的上下文。我们的安全设计从这里开始。一个健壮的、面向安全的State应该包含以下几部分from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 核心任务信息 messages: Annotated[List, add_messages] # 对话历史 user_query: str # 用户原始输入 current_goal: str # 当前分解后的子目标 # 工具执行记录用于可观测性 action_history: List[dict] # 记录每次工具调用{tool: name, input: params, output: result, step: n} error_log: List[str] # 记录运行过程中的错误或警告 # 安全与管控上下文 requires_human_approval: bool # 当前步骤是否需要人工审批 human_feedback: str # 人工审批的输入或反馈 execution_budget: float # 剩余执行预算如Token成本、API调用次数 risk_level: str # 当前任务评估的风险等级low, medium, high # 流程控制标志 is_approved: bool # 人工是否已批准 should_continue: bool # 是否继续执行 max_steps: int # 最大执行步数防死循环 current_step: int # 当前步数计数器设计要点解析action_history和error_log是可观测性的基石。每个工具节点执行后都必须向其追加记录。requires_human_approval和is_approved是可干预的开关。它们将决定工作流是走向人工审核节点还是继续自动执行。execution_budget和max_steps是资源护栏。在每个节点开始前检查预算防止成本失控或无限循环。实操心得State的设计要遵循“最小化但充分”原则。不要把所有东西都塞进去只放对流程控制和事后分析真正有用的字段。TypedDict和Annotated能提供良好的类型提示但更重要的是要在项目初期就团队对齐每个字段的精确含义和更新时机避免后续出现状态污染。2.2 节点Node与边Edge的安全增强在LangGraph中节点是执行单元边是流转逻辑。安全增强需要渗透到这两者中。安全节点示例人工审批节点这个节点不执行具体任务只负责暂停流程、等待输入、并更新状态。def human_approval_node(state: AgentState) - AgentState: 人工审批节点。在实际应用中这里会连接到一个Web界面或消息队列。 # 1. 在实际系统中此处会触发一个通知如短信、邮件、Slack消息将审批上下文发送给负责人。 # 上下文包括state[user_query], state[current_goal], state[action_history][-1]待审批的操作 print(f[审批请求] 待执行操作: {state[current_goal]}) print(f 操作历史: {state[action_history][-1] if state[action_history] else 无}) # 2. 模拟等待人工输入。真实场景中这里会等待一个外部API回调或数据库状态更新。 # 假设我们通过一个模拟函数获取审批结果 feedback, approved simulate_human_review(state) # 3. 更新状态 state[human_feedback] feedback state[is_approved] approved if approved: print([系统] 人工审批通过继续执行。) else: print(f[系统] 人工审批驳回。反馈{feedback}) state[should_continue] False # 终止流程 return state def simulate_human_review(state): # 这是一个模拟函数。真实场景中审批逻辑可能更复杂。 # 例如可以根据 risk_level 自动判断高风险操作默认送审。 if state[risk_level] high: # 模拟人工审核后批准 return 操作已审核风险可控批准执行。, True else: # 模拟自动批准 return 低风险操作自动批准。, True安全边条件动态路由逻辑边Edge决定了工作流的走向。我们可以根据安全状态来动态路由。from langgraph.graph import END def should_require_human_approval(state: AgentState) - str: 判断下一个节点是否需要经过人工审批。 # 规则1如果上一个工具调用被标记为高风险 last_action state[action_history][-1] if state[action_history] else None if last_action and last_action.get(risk_tag) high_risk: return route_to_approval # 规则2如果执行预算低于阈值 if state.get(execution_budget, 100) 10: return route_to_approval # 也可以路由到告警或终止节点 # 规则3如果用户查询中包含敏感关键词需提前定义列表 sensitive_keywords [删除, 重置, 转账, root] if any(keyword in state[user_query] for keyword in sensitive_keywords): return route_to_approval # 默认继续自动执行 return continue_auto def check_after_approval(state: AgentState) - str: 在人工审批节点之后决定下一步去向。 if not state.get(is_approved, False): # 如果被驳回直接结束工作流 return end else: # 如果批准则返回到主工作流继续执行 return proceed通过这种方式我们将安全策略如“高风险操作需审核”、“预算不足需告警”编码到了工作流的路由逻辑中使安全性与业务流程无缝融合。注意事项定义路由条件函数时务必保持其纯净和高效。它们会在每个决策点被调用不应有副作用如修改State或执行耗时操作如网络请求。复杂的风险评估逻辑应该放在专门的“风险评估节点”中该节点运行后将结果如risk_level: ‘high’写入State再由路由条件函数读取。3. 实操构建一个带有安全护栏的查询执行Agent让我们构建一个具体的例子一个可以查询数据库但删除操作必须经过人工审批的Agent。我们将使用LangGraph来编排流程。3.1 定义工具与风险标注首先定义Agent可以使用的工具并为每个工具标注风险等级。from langchain.tools import tool from typing import Optional tool def query_database(sql: str) - str: 执行数据库查询SELECT语句。 # 这里是模拟实现 print(f[安全日志] 执行查询: {sql}) # 模拟执行... return f查询结果: 模拟数据 from {sql} tool def delete_from_database(table: str, condition: Optional[str] None) - str: 从数据库删除数据。高风险操作。 # 这个工具本身不直接执行而是生成待审批的操作指令。 # 真正的执行会在人工审批后由另一个“安全执行”节点完成。 delete_sql fDELETE FROM {table} if condition: delete_sql f WHERE {condition} print(f[安全日志] 生成删除指令待审批: {delete_sql}) return delete_sql # 注意这里返回的是SQL指令字符串而非执行结果 # 工具列表并附带元数据如风险标签 TOOLS [query_database, delete_from_database] TOOL_RISK_MAP { query_database: low, delete_from_database: high, # 标记为高风险 }3.2 构建主工作流图我们将构建一个包含“计划”、“执行”、“审批”、“安全执行”节点的图。from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolExecutor, ToolNode from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.messages import SystemMessage, HumanMessage # 注意以下为简化示意实际构建需结合LangGraph最新API # 初始化模型和工具执行器 llm ChatOpenAI(modelgpt-4o-mini, temperature0) tool_executor ToolExecutor(TOOLS) # 1. 规划节点分析用户意图决定使用哪个工具及参数 def planning_node(state: AgentState): 分析用户查询规划工具调用。 messages [ SystemMessage(content你是一个数据库助手。请分析用户请求决定是进行‘查询’还是‘删除’。对于删除你必须生成准确的SQL DELETE语句。), HumanMessage(contentstate[user_query]) ] response llm.invoke(messages) # 这里简化处理实际应解析LLM响应为结构化指令 # 假设响应中包含了工具调用决定 planned_action { tool: delete_from_database if 删除 in state[user_query] else query_database, input: {sql: SELECT * FROM users WHERE id1} # 简化示例 } state[current_goal] f调用工具 [{planned_action[tool]}] # 将规划结果存入状态供下一个节点使用 state[_planned_action] planned_action return state # 2. 有条件工具调用节点执行低风险工具 def tool_node(state: AgentState): 执行工具调用并记录历史。 planned state.get(_planned_action) if not planned: state[error_log].append(规划节点未提供执行指令。) return state tool_name planned[tool] tool_input planned[input] # 执行前检查如果是高风险工具则设置审批标志并跳过直接执行 if TOOL_RISK_MAP.get(tool_name) high: state[requires_human_approval] True state[_pending_high_risk_action] {tool: tool_name, input: tool_input} print(f[安全拦截] 高风险工具 {tool_name} 被拦截等待人工审批。) # 此时不执行工具直接返回让路由逻辑将其导向审批节点 return state # 执行低风险工具 try: result tool_executor.invoke({tool_name: tool_input}) # 记录到行动历史 state[action_history].append({ step: state[current_step], tool: tool_name, input: tool_input, output: result, risk_tag: TOOL_RISK_MAP.get(tool_name, unknown) }) state[last_tool_result] result except Exception as e: state[error_log].append(f工具执行失败: {e}) finally: state[current_step] 1 return state # 3. 人工审批节点复用之前定义的 human_approval_node # 4. 安全执行节点仅在人工批准后执行高风险操作 def safe_execution_node(state: AgentState): 执行经过审批的高风险操作。 if not state.get(is_approved): state[error_log].append(试图执行未批准的高风险操作。) return state pending_action state.get(_pending_high_risk_action) if not pending_action: state[error_log].append(审批通过但无待执行的高风险操作。) return state tool_name pending_action[tool] tool_input pending_action[input] print(f[安全执行] 正在执行已审批的高风险操作: {tool_name} with {tool_input}) # 这里是真正执行危险操作的地方 # 例如连接到数据库执行DELETE语句。 # 为了演示我们只是模拟执行并记录。 try: # 模拟执行成功 result f[模拟] 高风险操作 {tool_name} 已安全执行。输入: {tool_input} state[action_history].append({ step: state[current_step], tool: tool_name, input: tool_input, output: result, risk_tag: high, approved: True }) state[last_tool_result] result print(result) except Exception as e: state[error_log].append(f安全执行失败: {e}) finally: # 清理待执行动作 state[_pending_high_risk_action] None state[current_step] 1 return state # 5. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planning_node) workflow.add_node(tool_call, tool_node) workflow.add_node(human_approval, human_approval_node) workflow.add_node(safe_executor, safe_execution_node) # 设置边和路由 workflow.add_edge(START, planner) workflow.add_edge(planner, tool_call) # 关键从 tool_call 出来的条件路由 def route_after_tool_call(state: AgentState) - str: if state.get(requires_human_approval, False): return human_approval else: return generate_response # 假设有一个生成最终响应的节点这里简化为END workflow.add_conditional_edges( tool_call, route_after_tool_call, { human_approval: human_approval, generate_response: END # 简化直接结束 } ) # 从人工审批节点出来的路由 workflow.add_conditional_edges( human_approval, check_after_approval, # 使用前面定义的函数 { end: END, proceed: safe_executor } ) workflow.add_edge(safe_executor, END) # 编译图 app workflow.compile()这个工作流实现了核心安全逻辑规划节点分析意图工具调用节点执行低风险操作并拦截高风险操作将其路由至人工审批节点。审批通过后再由一个专用的“安全执行节点”来最终完成高风险操作。整个过程的状态、行动、审批记录都被完整追踪。踩坑实录在早期版本中我曾尝试在同一个tool_node里既做判断又执行所有工具。这导致了状态管理的混乱特别是当工具执行失败时回滚和重试逻辑变得极其复杂。后来才演变为现在的“拦截-审批-安全执行”三段式设计职责分离清晰状态流转也更容易调试。4. 监控、评估与迭代让护栏越来越智能部署了安全护栏并非一劳永逸。我们需要持续监控Agent的表现评估护栏的有效性并迭代安全策略。4.1 构建监控与日志系统我们需要将之前State中的action_history和error_log持久化到外部系统如数据库、Elasticsearch以便进行分析。import json from datetime import datetime def persist_agent_run(session_id: str, final_state: AgentState, conclusion: str): 将一次Agent运行的完整上下文持久化。 log_entry { session_id: session_id, timestamp: datetime.utcnow().isoformat(), user_query: final_state.get(user_query), conclusion: conclusion, # 成功, 人工驳回, 错误中断, 预算耗尽 action_history: final_state.get(action_history, []), error_log: final_state.get(error_log, []), final_state_snapshot: {k: v for k, v in final_state.items() if k not in [messages, action_history, error_log]} # 避免循环引用 } # 这里可以写入数据库、文件或日志服务 print(f[审计日志] Session {session_id}: {conclusion}) # 示例写入JSON文件 with open(fagent_audit_log_{session_id}.json, w) as f: json.dump(log_entry, f, indent2, ensure_asciiFalse)在每次Agent运行结束时无论成功失败都调用此函数。这为我们提供了事后审计和模型行为分析的宝贵数据。4.2 定义安全评估指标有了数据我们就可以定义一些关键指标KPIs来衡量安全护栏的效果人工干预率触发requires_human_approval的会话占比。过高可能意味着规则太严格影响效率过低则可能意味着规则有漏洞。审批通过率人工审批节点中操作被批准的比例。这反映了Agent提出的高风险操作有多少是真正合理且必要的。平均审批耗时从触发审批到收到反馈的平均时间。这直接影响用户体验和流程效率。异常逃逸率在未触发任何安全机制如审批、错误的情况下最终结果仍不符合预期的会话比例。这是衡量护栏完整性的终极指标。成本合规率execution_budget未被耗尽的会话比例。定期如每周分析这些指标可以帮助我们量化安全投入的效果并发现优化点。4.3 迭代安全策略从规则到模型最初的规则可能是硬编码的如“工具名包含‘delete’就送审”。随着数据积累我们可以让安全策略变得更智能规则引擎优化基于历史审批数据发现误报低风险被送审和漏报高风险未送审的模式调整关键词列表或风险判断逻辑。引入预测模型训练一个简单的分类模型基于用户查询、历史动作等特征预测当前操作的风险概率。将基于规则的路由升级为“基于规则模型评分”的混合路由。def advanced_risk_assessment(state: AgentState) - float: 基于简单特征的风险评分示例。 score 0.0 query state[user_query].lower() # 特征1查询中包含破坏性动词 destructive_verbs [删除, 丢弃, 清空, 覆盖, 修改] if any(verb in query for verb in destructive_verbs): score 0.4 # 特征2近期同一会话中失败次数多 recent_errors len([e for e in state[error_log][-3:] if 失败 in e]) score recent_errors * 0.2 # 特征3操作涉及核心数据表需维护核心表列表 if users in query or orders in query: score 0.3 return min(score, 1.0) # 归一化到0-1 def should_require_approval_advanced(state): score advanced_risk_assessment(state) if score 0.7: # 高风险阈值 return route_to_approval elif score 0.3: # 中风险也许可以记录日志并继续或路由到更轻量的确认节点 state[error_log].append(f中风险操作预警评分: {score:.2f}) return continue_auto else: return continue_auto闭环学习将人工审批的反馈human_feedback作为强化学习的奖励信号或者用于微调规划节点Planner的LLM让Agent逐渐学会在提出请求时就规避高风险行为。个人体会Harness Engineering不是一个可以“一次性完成”的模块而是一个需要持续运营的“系统”。它和你的核心业务逻辑一样重要甚至更重要。我建议在团队中设立明确的“安全护栏负责人”像对待生产环境监控告警一样定期审查安全日志和指标。最开始的规则可能很粗糙会误拦很多正常操作但没关系快速迭代是关键。每一次人工审批不仅是一次安全拦截更是一次对Agent行为的宝贵标注利用好这些数据你的护栏会越来越聪明。5. 常见问题与排查技巧实录在实际部署带有安全护栏的Agent时你肯定会遇到各种意想不到的情况。下面是我踩过的一些坑和总结的排查思路。5.1 问题Agent陷入“审批-执行-再审批”的死循环现象一个高风险操作被批准执行后在后续步骤中又触发了同样的审批规则导致流程卡住。根因分析这通常是因为状态State没有在审批后正确清理。例如requires_human_approval标志在审批通过后没有被重置为False或者_pending_high_risk_action没有被清空导致工作流在下一个循环判断时依然认为需要审批。解决方案确保状态机纯净在safe_execution_node执行完高风险操作后必须重置所有与本次审批相关的临时状态标志。def safe_execution_node(state: AgentState): # ... 执行操作 ... # 执行完毕后清理审批相关状态 state[requires_human_approval] False state[is_approved] False # 重置批准状态为下一次独立审批做准备 state[_pending_high_risk_action] None state[human_feedback] return state设计自包含的审批上下文另一种思路是将每次审批及其对应的待执行动作封装成一个独立的对象并赋予一个唯一的approval_id。当该ID对应的动作执行完毕后系统就认为该审批上下文已关闭不会影响后续流程。5.2 问题人工审批响应超时导致工作流僵死现象流程进入human_approval_node后一直在等待外部系统的回调如用户点击审批按钮如果用户一直不处理这个工作流实例就会一直占用资源等待。根因分析LangGraph的普通节点是同步执行的。如果将“等待人工输入”这个可能耗时很长的操作放在一个同步节点里必然会阻塞。解决方案采用异步工作流与外部事件驱动。改造审批节点为“暂停点”human_approval_node不再等待而是执行以下操作 a. 生成一个唯一的pause_id如UUID。 b. 将当前完整的State序列化后持久化到数据库如Redis、PostgreSQL并以pause_id为键。 c. 向审批系统如邮件、钉钉、自定义Webhook发送请求附上pause_id和审批内容。 d. 节点返回一个特殊指令如{should_pause: True, pause_id: pause_id}并让工作流路由到一个“暂停”状态。提供继续执行的端点暴露一个REST API端点例如POST /webhook/continue/{pause_id}。当用户在审批界面点击“批准”或“驳回”时审批系统调用此端点并附上决定。恢复执行该API端点根据pause_id从数据库中恢复保存的State更新is_approved和human_feedback字段然后重新将State注入到LangGraph工作流中并从“暂停”后的下一个节点如safe_executor继续执行。这需要LangGraph支持检查点Checkpoint功能幸运的是这正是LangGraph的核心优势之一它原生支持将工作流状态持久化并在之后恢复。5.3 问题安全规则过于复杂难以维护和调试现象should_require_human_approval函数里塞满了大量的if-else规则逻辑纠缠添加新规则时战战兢兢生怕影响旧逻辑。根因分析将业务逻辑、安全策略、路由逻辑全部耦合在代码中。解决方案采用策略模式Strategy Pattern或规则引擎。策略模式将不同的风险判断逻辑抽象成独立的类或函数。例如定义BudgetRule、KeywordRule、ToolRiskRule等。路由条件函数只需遍历一个规则列表依次评估即可。class SafetyRule: def evaluate(self, state: AgentState) - (bool, str): # (是否触发, 规则名称) raise NotImplementedError class KeywordRule(SafetyRule): def __init__(self, keywords): self.keywords keywords def evaluate(self, state): query state[user_query].lower() if any(kw in query for kw in self.keywords): return True, keyword_rule return False, class BudgetRule(SafetyRule): def evaluate(self, state): if state.get(execution_budget, 100) 5: return True, budget_rule return False, # 配置化规则列表 SAFETY_RULES [ KeywordRule([删除, root, sudo]), BudgetRule(), # ... 可以轻松添加新规则 ] def should_require_approval(state): for rule in SAFETY_RULES: triggered, rule_name rule.evaluate(state) if triggered: print(f[安全规则触发] {rule_name}) return route_to_approval return continue_auto规则引擎对于极其复杂的策略可以考虑使用像Drools、Easy Rules这样的轻量级规则引擎甚至将规则配置存储在数据库中实现动态更新无需重启服务。5.4 问题监控日志数据量巨大难以定位问题现象action_history记录了每个步骤的详细输入输出当处理大量数据或复杂任务时日志体积爆炸搜索特定错误如同大海捞针。解决方案实施结构化日志分级和采样。分级记录不是所有信息都需要全量记录。DEBUG级记录完整的思维链、工具调用的原始请求和响应。此级别日志量最大只在调试特定会话时开启。INFO级记录关键决策点、规则触发、审批请求和结果。这是日常监控的主要级别。WARN/ERROR级记录异常、失败和违反安全策略的事件。必须全量记录并配置告警。会话采样对于DEBUG级日志可以按一定比例如1%随机采样记录或者只记录标记为“重要”的会话如来自内部测试用户、或触发了WARN以上级别事件的会话。聚合分析使用ELKElasticsearch, Logstash, Kibana或类似栈对INFO和ERROR级日志进行索引。可以快速筛选出“所有触发BudgetRule的会话”或“所有最终状态为人工驳回的会话”进行聚合分析。构建一个带安全护栏的Agent系统就像训练一个天赋异禀但涉世未深的实习生。你需要给他清晰的指令规划赋予他工具能力但在他可能犯大错的关键决策点上高风险操作你必须设置检查站审批。同时你还得在他身后安装摄像头监控日志定期复盘他的工作记录评估指标并据此更新你的管理手册迭代策略。这个过程充满挑战但却是将AI Agent安全、可靠地应用于生产环境的必经之路。
