拓冰建站拓冰建站
首页 / 资讯中心 / 正文

Agent Harness:智能体安全带的运行框架与工程实践

之前有朋友问我你天天说 Agent到底写的是一套while True循环调大模型还是一个能上生产环境的东西这个问题其实问到了关键点上。很多团队在 Demo 阶段都能让模型“自己用工具”但一旦要处理超时、断点、权限、日志、回放、评估这些工程问题时发现代码里乱成一团。这时候缺的往往不是更强的模型而是一个承载 Agent 运行的基础设施。在 Agent 开发领域这个基础设施有一个越来越常被提起的名字智能体安全带Agent Harness。这篇文章会围绕“什么是 Agent Harness”展开梳理它和 Agent、Prompt、Tool、Skill 这些概念之间的关系然后用一个可运行的最小代码示例带大家理解 Agent Harness 到底在系统中扮演什么角色。无论你是刚接触大模型应用开发还是已经写过不少 Agent 代码但觉得工程化很吃力这篇文章都会对你有帮助。1. 什么是智能体安全带Agent Harness1.1 从一个常见误区说起先来看一个非常常见的“伪 Agent”实现把用户问题拼进 Prompt调用一次大模型把返回结果打印出来。很多初学者以为这就是 Agent。严格来说这只是“模型对话”它不具备自主决策、工具调用、多步推理这些 Agent 能力。当项目往前走你会发现 Agent 需要解决几类问题模型在什么时候调用工具调用哪个工具工具执行出错之后是重试、换工具还是直接终止多轮对话的上下文由谁保存存到哪里超长之后怎么截断Agent 卡在死循环里怎么拦截线上出问题时能不能通过日志完整还原 Agent 每一步的决策过程这些问题都不是“换一个更强的模型”能解决的它们属于运行时管理问题。而 Agent Harness就是用来承载和处理这些问题的系统层组件。1.2 Agent Harness 的通俗定义不绕弯子先给一个通俗理解Agent Harness 是 Agent 的“生命支持系统”和“运行框架”。它负责拉起 Agent 的执行环境管理 Agent 与模型、工具、记忆、安全策略之间的交互让 Agent 的行为可控制、可观测、可复现。如果你开过车可以把 Agent 理解成司机把 Harness 理解成安全带。司机决定往哪开但如果没系安全带急刹车时人就会被甩出去。Agent Harness 做的事情就是在 Agent 自由行动的同时给它加上约束、保护、缓冲和记录。从工程角度Agent Harness 通常是一套代码库或服务框架它包含但不限于以下能力模型接入与调用管理超时、重试、切换模型工具注册与调度工具的发现、白名单、参数校验上下文和记忆状态管理短期会话、长期记忆、截断策略执行循环控制最大步数、终止条件、异常恢复安全策略权限控制、敏感操作审批、隔离可观测性结构化日志、链路追踪、运行回放1.3 为什么 Agent 越来越需要 Harness大语言模型本质上是一个概率系统同一个 Prompt 在不同时间可能产生不同输出。单独一次模型调用输出不稳定是可以接受的但当你把模型放到一个可以调用工具、操作外部系统的 Agent 循环里这种不确定性就会被放大甚至产生不可控的风险。举个例子模型在一次对话中决定调用“删除数据库记录”的工具这个决策本身可能只是概率性的但后果却是确定的。如果没有 Harness 这一层来做权限校验、二次确认、操作审计一旦模型决策错误损失将无法挽回。另外从工程协作角度来说Prompt 工程属于提示词层Agent 的决策逻辑属于应用层而 Harness 属于框架层。把这三层分开团队里不同角色才能并行工作提示词工程师专注优化 Prompt算法工程师专注模型和工具效果后端工程师专注 Harness 的稳定性与安全。2. Agent、Harness、Prompt、Skill 到底有什么区别很多人在学习 Agent 开发时会把 Prompt、Agent、Tool、Skill、Harness 混在一起。下面把这几组概念梳理清楚。2.1 通过一个类比理解概念分层可以这样理解一个完整的智能体系统就像一家餐厅。Prompt 是菜谱它告诉厨房模型怎么做菜。Model 是厨师它根据菜谱和经验做出决策。Tool / Skill 是厨房里的锅碗瓢盆和厨师技艺负责执行具体操作。Agent 是主厨它根据当前客人的需求决定先做什么菜、用什么工具。Harness 是餐厅的管理制度与后厨安全规范它规定什么时候能开火、什么东西必须戴手套、出问题怎么报备。单有厨师模型和菜谱Prompt做一两个菜没问题。但要稳定地经营一家餐厅就必须有后厨管理制度Harness。2.2 核心概念对比表概念一句话解释典型问题与 Harness 的关系Prompt给模型的指令文本输出格式不稳定、语气不对Harness 负责拼装和管理 Prompt 模板Model大语言模型本身幻觉、超时、上下文有限Harness 负责模型接入、重试、降级Tool可被模型调用的具体函数/API参数格式错误、权限缺失Harness 负责注册、校验、调度工具Skill围绕某个领域的能力封装能力边界不清、复用困难Harness 负责技能的加载与生命周期管理Agent有自主决策能力的应用实体决策错误、死循环Harness 为 Agent 提供运行与保护机制HarnessAgent 的运行框架与基础设施不稳定、不可观测、难回放Harness 是整个系统的骨架2.3 “模型调用”和“Agent 运行”是两个层次单独调用模型时输入是消息列表输出是文本。你不需要处理工具、状态、权限。而 Agent 运行是一个循环过程接收用户输入。构建上下文Prompt 历史 工具描述。调用模型获得决策结果。如果模型要求调用工具执行工具。把工具结果写回上下文。回到第 3 步直到模型不再调用工具或达到终止条件。这个循环就是经典的 Agent 执行循环Agent loop。Harness 的核心职责之一就是把这个循环用一个稳定、可扩展、可观测的框架实现出来。没有 Harness你也能在脚本里写一个循环但可维护性、安全性和可观测性都会大打折扣。在开源社区和各大厂商的 AI 应用框架里Agent Harness 思想已经非常普遍。不同的框架可能叫法不同有的叫 Runtime有的叫 Executor有的叫 Runner但底层做的事情基本一致。理解 Agent Harness 的原理后再去看各种框架的源码或配置会轻松很多。3. Agent Harness 的核心组成与工作原理下面从六个方面拆解 Agent Harness 的典型构成。注意不是每个 Harness 都必须包含全部模块生产环境通常会按需裁剪。3.1 LLM 接入层LLM 接入层负责屏蔽不同模型提供方的差异。无论是 OpenAI、Anthropic、阿里云百炼、智谱、DeepSeek还是本地部署的模型服务Harness 都应该通过统一接口调用。LLM 接入层要考虑的问题包括API 地址和密钥管理。请求超时和重试策略。模型版本管理。Token 消耗统计。流式输出支持。多模型切换与降级。一个设计良好的 LLM 接入层通常对外暴露一个类似chat(messages, tools, config) - ChatResult的方法。内部再把不同厂商的请求和响应转换成统一数据格式。3.2 工具注册与调度机制工具是 Agent 能够影响外部世界的通道。Harness 提供一套注册机制让开发者可以方便地把函数、API、SDK 方法暴露给模型。工具注册机制需要支持声明式元信息工具名、描述、参数 Schema。生命周期管理初始化、启用、禁用、销毁。权限控制哪些 Agent 可以用哪些工具。并发控制工具被多个任务同时调用时如何隔离。工具调度则解决“模型选择调用了某个工具Harness 如何安全执行它”的问题。执行时通常要做参数校验、超时控制、异常捕获和结果规范化。3.3 上下文与记忆管理模型上下文窗口是有限的。Harness 需要负责维护 Agent 的记忆并决定什么信息能被写入模型输入。上下文管理通常分三个层级短期记忆当前任务内的多轮对话和工具执行结果。长期记忆跨会话的用户偏好、历史结论、业务知识。工作记忆Agent 当前正在处理的数据和中间状态。当上下文超过模型窗口限制时Harness 需要执行截断策略丢弃早期的对话、压缩历史摘要、摘除不重要的工具输出。这些策略直接影响到 Agent 的长对话稳定性和记忆准确性。3.4 安全与权限边界这是 Agent Harness 最重要的价值之一。模型可以“建议”调用工具但 Harness 通过权限校验来最终决定“能不能调用”。安全边界包括工具白名单即使模型在 Prompt 里看到了某个工具如果不在当前会话的白名单内也不能执行。操作审批对删除、修改、转账、发布等高风险操作Harness 可以拦截并要求人工二次确认。参数校验模型给出的参数可能不合法或超范围Harness 需要做严格的 Schema 校验。隔离与限额限制 Agent 调用工具的频次、次数、资源消耗。审计日志记录谁在什么时间、通过什么 Prompt、调用了什么工具、传了什么参数、结果如何。3.5 可观测性与日志追踪Agent 的决策链路过长如果没有完整的日志排错会非常痛苦。Harness 会为每一次任务生成一个唯一的 Trace ID并把整个运行过程中的关键事件记录下来每一步的输入消息。模型返回的原始输出。工具调用的参数与结果。上下文截断和摘要操作。错误和重试记录。耗时和 Token 消耗。有了这些记录开发者可以回放 Agent 的完整决策过程分析它为什么在某个步骤做出了错误决定。3.6 生命周期与运行策略Agent 不是无限运行的。Harness 负责控制 Agent 的生命周期从初始化、启动、运行、暂停到最终停止。运行策略则包括最大执行步数限制例如max_steps20防止死循环。最大 Token 消耗限制。任务超时时间。遇到特定错误时的降级策略。用户取消任务时的清理逻辑。一个没有运行策略的 Agent在生产环境就像一个没有刹车的车非常危险。之前有团队在测试时让 Agent 自动调用某个数据同步工具结果模型在循环里反复触发同步接口直到被监控系统发现才停机这就是典型的“没有 max_steps”导致的线上故障。4. 从零手写一个极简 Agent Harness下面用一个最小可运行示例来演示 Agent Harness 的核心思想。这个示例不使用任何第三方 SDK只用 Python 标准库实现工具注册、执行循环、日志追踪和步数限制。真实的 Harness 框架会复杂很多但核心思想完全一致。4.1 设计目标我们想实现一个极简 Harness具备以下能力支持工具注册。支持调用 LLM先用 Mock 模式演示再替换为 OpenAI 兼容接口。支持多轮 Agent 循环模型决定调用工具Harness 执行再把结果写回上下文。支持最大步数限制。输出结构化运行日志。4.2 项目结构agent_harness_demo/ ├── config.yaml # 运行配置 ├── harness.py # Harness 核心实现 ├── tools.py # 演示用工具 └── main.py # 启动入口4.3 配置文件先看配置文件它定义了 Harness 的常用参数。# config.yaml harness: max_steps: 10 timeout_seconds: 30 trace_dir: ./traces logging: level: INFO llm: provider: mock # 可切换为 openai_compatible base_url: https://your-api.example.com/v1 model: your-model-name api_key_env: LLM_API_KEY security: enable_tool_whitelist: true tool_whitelist: - get_current_time - add_numbers4.4 Harness 核心代码核心类是ToolRegistry和AgentHarness。# harness.py import json import logging import os import time from typing import Any, Callable, Dict, Optional logger logging.getLogger(AgentHarness) class ToolSpec: 描述一个可被 Agent 调用的工具。 def __init__(self, name: str, description: str, func: Callable): self.name name self.description description self.func func class ToolRegistry: 工具注册中心负责保存、校验、调用工具。 def __init__(self, whitelist: Optional[list] None): self._tools: Dict[str, ToolSpec] {} self._whitelist set(whitelist or []) def register(self, name: str, description: str): 装饰器把普通函数变成可被 Harness 调用的工具。 def wrapper(func: Callable): self._tools[name] ToolSpec( namename, descriptiondescription, funcfunc, ) return func return wrapper def list_tools(self) - list: 返回给模型看的工具描述列表。 tools [] for name, spec in self._tools.items(): if self._whitelist and name not in self._whitelist: continue tools.append({ name: name, description: spec.description, }) return tools def call(self, name: str, **kwargs: Any) - Any: 调用工具执行前做权限校验。 if name not in self._tools: raise ValueError(f工具不存在: {name}) if self._whitelist and name not in self._whitelist: raise PermissionError(f工具不在白名单中禁止调用: {name}) spec self._tools[name] logger.info(f[ToolCall] name{name}, args{kwargs}) result spec.func(**kwargs) logger.info(f[ToolResult] name{name}, result{result}) return result class AgentHarness: Agent Harness 核心管理 Agent 的执行循环。 def __init__(self, config: dict): self.config config self.registry ToolRegistry( whitelistconfig.get(security, {}).get(tool_whitelist) ) self._running False self._trace_id None property def running(self) - bool: return self._running def start(self): 启动 Harness完成初始化。 self._running True self._trace_id ftrace-{int(time.time() * 1000)} logger.info(f[Harness] started, trace_id{self._trace_id}) def stop(self): 停止 Harness。 self._running False logger.info([Harness] stopped) def build_system_prompt(self) - str: 构造系统提示词。 tools self.registry.list_tools() tool_lines \n.join( [f- {t[name]}: {t[description]} for t in tools] ) return ( 你是运行在 Agent Harness 中的助手。\n 你可以使用以下工具\n f{tool_lines}\n 当需要调用工具时请严格输出 JSON\n {action: tool_name, args: {key: value}}\n ) def run(self, user_input: str) - str: 执行一次完整的 Agent 任务。 if not self._running: raise RuntimeError(Harness 尚未启动请先调用 start()) max_steps self.config.get(harness, {}).get(max_steps, 10) messages [ {role: system, content: self.build_system_prompt()}, {role: user, content: user_input}, ] for step in range(1, max_steps 1): logger.info(f[Step {step}] 调用 LLM...) # 这里调用模型。当前使用 mock_llm 模拟模型输出。 llm_output self._call_llm(messages) if llm_output is None: # 模型没有要调用工具返回最终回复。 return 当前模型未返回有效结果 try: parsed json.loads(llm_output) action parsed[action] args parsed.get(args, {}) except Exception: # 模型没有输出标准 JSON视为普通回复。 return llm_output if action finish: return args.get(answer, 任务结束) # Harness 核心执行工具调用 result self.registry.call(action, **args) messages.append({role: assistant, content: llm_output}) messages.append({ role: tool, name: action, content: json.dumps(result, ensure_asciiFalse), }) raise RuntimeError(f超过最大步数限制 {max_steps}强制终止) def _call_llm(self, messages: list) - Optional[str]: 调用 LLM 的抽象方法。 provider self.config.get(llm, {}).get(provider, mock) if provider mock: return self._mock_llm(messages) if provider openai_compatible: return self._call_openai_compatible_llm(messages) raise ValueError(f未知 provider: {provider}) def _mock_llm(self, messages: list) - Optional[str]: 模拟模型输出用于本地演示。 真实项目中这里会替换为对 OpenAI / 其他模型的 HTTP 调用。 user_content messages[-1][content] if 时间 in user_content: return {action: get_current_time, args: {}} if 加法 in user_content or 加上 in user_content: return {action: add_numbers, args: {a: 1, b: 2}} return None def _call_openai_compatible_llm(self, messages: list) - Optional[str]: 以 OpenAI 兼容接口为例展示真实接入逻辑。 需要安装 requestspip install requests import requests llm_config self.config.get(llm, {}) api_key_env llm_config.get(api_key_env, LLM_API_KEY) api_key os.environ.get(api_key_env) if not api_key: raise RuntimeError(f缺少环境变量: {api_key_env}) url llm_config[base_url].rstrip(/) /chat/completions headers { Authorization: fBearer {api_key}, Content-Type: application/json, } payload { model: llm_config[model], messages: messages, temperature: 0.2, } response requests.post(url, headersheaders, jsonpayload, timeout30) response.raise_for_status() content response.json()[choices][0][message][content] return content4.5 工具定义与启动入口接下来定义两个演示工具并编写启动入口。# tools.py from datetime import datetime from harness import AgentHarness def get_current_time() - str: 获取当前服务器时间。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def add_numbers(a: int, b: int) - int: 计算两个整数的和。 return int(a) int(b) def register_tools(harness: AgentHarness): 把工具注册到 Harness 的工具注册中心。 harness.registry.register(get_current_time, 获取当前服务器时间)(get_current_time) harness.registry.register(add_numbers, 计算两个整数的和)(add_numbers)# main.py import logging import sys from pathlib import Path import yaml from harness import AgentHarness from tools import register_tools def load_config(): config_path Path(__file__).parent / config.yaml with open(config_path, r, encodingutf-8) as f: return yaml.safe_load(f) def setup_logging(level: str INFO): logging.basicConfig( levelgetattr(logging, level.upper()), format%(asctime)s | %(name)s | %(levelname)s | %(message)s, streamsys.stdout, ) def main(): config load_config() setup_logging(config.get(logging, {}).get(level, INFO)) harness AgentHarness(config) register_tools(harness) harness.start() try: user_input 现在几点了 reply harness.run(user_input) print(f[Agent 回复] {reply}) user_input 帮我算一下 1 加 2 等于多少 reply harness.run(user_input) print(f[Agent 回复] {reply}) finally: harness.stop() if __name__ __main__: main()4.6 运行与验证先安装依赖pip install pyyaml requests运行程序cd agent_harness_demo python main.py预期输出类似2025-01-01 12:00:01 | AgentHarness | INFO | [Harness] started, trace_idtrace-1735700400123 2025-01-01 12:00:01 | AgentHarness | INFO | [Step 1] 调用 LLM... 2025-01-01 12:00:01 | AgentHarness | INFO | [ToolCall] nameget_current_time, args{} 2025-01-01 12:00:01 | AgentHarness | INFO | [ToolResult] nameget_current_time, result2025-01-01 12:00:01 2025-01-01 12:00:01 | AgentHarness | INFO | [Step 2] 调用 LLM... 2025-01-01 12:00:02 | AgentHarness | INFO | [Harness] stopped注意在 Mock 模式下模型只返回一次工具调用就结束了所以第二个问题“1 加 2”可能直接返回None也就是返回“当前模型未返回有效结果”。这正好说明了一个真实问题模型输出不稳定Harness 需要设计好容错逻辑。如果你把config.yaml中的 provider 改为openai_compatible并设置好环境变量LLM_API_KEY程序就会走真实模型调用流程。这个例子虽然简单但已经包含了一个 Agent Harness 的核心闭环工具注册、权限白名单、执行循环、日志追踪、步数限制、LLM 接入抽象。你在生产环境使用的各种 Harness 框架本质上都是在这些基础能力上不断扩展和强化。5. 常见问题与排查思路在落地 Agent Harness 时团队通常会遇到下面几类问题。这里整理成排查清单的形式。5.1 Agent 总是调用同一个工具陷入局部循环问题现象常见原因排查思路Agent 反复调用同一个工具不往下推进工具返回结果没有改变上下文状态模型没有足够信息判断“任务已完成”检查工具结果是否被完整写回上下文调整系统 Prompt明确工具调用后必须继续判断引入 max_steps 强制截断5.2 上下文增长过快很快超过模型窗口问题现象常见原因排查思路多轮工具调用后请求报上下文超长每轮都把完整工具结果拼进消息列表没有做压缩实现上下文压缩策略历史窗口截断、工具结果摘要化、关键信息提取后移除原始结果5.3 Harness 中工具权限过大问题现象常见原因排查思路模型在 Prompt 中看到了工具但部分工具不应被当前场景调用工具注册时没有做细分权限控制所有 Agent 共用全部工具引入工具白名单机制按 Agent 用途拆分工具集合高风险工具加人工审批回调5.4 日志里看不出 Agent 为什么做错决策问题现象常见原因排查思路线上 Agent 输出错误结果但找不到完整调用链日志只记录最终回复没有记录中间步骤和工具调用参数每次任务生成 trace_id在 Harness 里记录每一步 messages、LLM 输出、工具调用结果提供回放调试工具5.5 Mock 模式正常切换真实模型后表现不稳定问题现象常见原因排查思路真实模型经常不按 Prompt 要求输出 JSON模型对工具调用格式理解不稳定Prompt 中工具描述不清晰使用官方工具调用function calling协议而不是让模型输出 JSON 字符串增加格式校验和修复重试机制6. 最佳实践与工程建议把 Agent Harness 落到工程里有几个原则值得长期坚持。6.1 把“模型决策”和“实际执行”解耦模型负责“建议”Harness 负责“决定”。即使模型在输出里要求调用某个工具Harness 也必须做权限校验、参数校验和操作审批。不要盲目信任模型输出。严格区分模型决策层和工具执行层能让你在出问题时快速定位责任边界。6.2 最小权限原则给 Agent 的工具权限应该尽量小。能只读就不要给写权限能只操作单条数据就不要给批量操作权限能限制在测试环境就不要放开生产环境。Harness 的配置中心可以集中管理工具白名单、环境标签和审批策略。对于高风险操作例如删除资源、修改线上配置、发送消息、转账建议实现“人工二次确认”机制。6.3 日志记录要做到可以完整回放Agent 排错最怕的是“不知道它为什么这么做”。Harness 应该为每次任务生成独立 Trace ID并完整记录每个步骤的完整消息列表。模型返回的原始输出。工具调用的参数和返回结果。上下文截断行为。错误和重试记录。这些日志是 Agent 调优最重要的依据。没有回放能力的 Agent 系统上线后基本等于黑盒。6.4 评估先于上线在把 Agent 应用到生产之前准备一组评估用例集。用例要覆盖正常流程、边界输入、恶意输入、工具异常等场景。每次修改 Prompt、升级模型、调整 Harness 策略后都跑一遍评估集对比通过率、工具调用准确率、任务完成率和 Token 消耗。评估集不需要一开始就很大50 到 100 条高质量用例就足以发现大部分问题。关键是长期维护和持续回归。6.5 生产环境的额外注意点密钥不要写进配置文件和代码库从环境变量或密钥管理服务读取。配置变更要经过测试环境验证并保留回滚能力。为 Harness 设置资源配额例如单任务最大 Token 消耗、QPS 上限、超时时间。对模型的输出做校验不合法就重试或降级不要直接进入工具调用环节。定期检查工具和依赖的版本避免供应链风险。7. 总结与学习方向Agent Harness 是 Agent 工程化过程中绕不开的一层基础设施。它不负责“更聪明”但负责“更可控”。模型负责决策工具负责执行Harness 负责把决策和执行安全地衔接起来。本文的核心要点可以概括为Agent Harness 是 Agent 的运行框架解决控制、安全、观测、生命周期管理问题。Agent 是决策实体Harness 是承载 Agent 的运行环境两者职责不同。Prompt、Tool、Skill、Model 都是 Harness 管理下的资源。一个最小 Harness 至少包含工具注册、执行循环、步数限制、日志追踪。生产环境必须重视权限管控和日志回放能力。如果你接下来想深入学习可以从这几个方向入手阅读成熟 Agent 框架的源码观察它们是如何实现工具注册和调度。研究函数调用function calling协议理解模型与工具之间的交互格式。实践上下文压缩和记忆管理解决长任务场景下的稳定性问题。设计一套自己的评估用例集把 Agent 的效果量化出来。最后说一点个人经验我在落地这类系统时最大的感受是“边界感很重要”。Agent 的自由度不是越大越好而是要在可控范围内逐步放开。先把 Harness 的机制、日志、权限、步数限制做扎实再尝试更复杂的任务编排。建议你从今天这个极简版本开始加上更多的工具接入真实模型跑通你自己的第一个带安全带的 Agent。
分享:

看完干货,该让你的企业上线了

免费需求沟通 · 48 小时内出具建站方案 · 河南本地可上门