最近很多读者在后台留言视频里天天讲 Harness、核心组件、最佳实践但看完还是不清楚 Harness 到底解决什么问题面试时被问到也只能背几个名词。过去几个月我反复折腾了 Harness 相关的工程框架、源码和项目案例踩了不少坑也整理出一套从概念到落地的闭环笔记。这篇文章会从 Harness 的底层定位讲起拆解核心组件手写一个轻量级 Harness Demo再结合真实项目给出工程化建议和面试答题思路希望能帮你把 Harness 真正“吃透”。文章内容较多建议先收藏再阅读。如果你是零基础可以按顺序阅读如果已经写过 Agent 或折腾过自动化工具链可以直接跳到第 4 节看实战代码再回来补齐概念。1. Harness 架构到底是什么1.1 从“马具”到“控制层”一个容易混淆的概念Harness 英文原意是“马具”“挽具”在工程领域的引申含义是“把某个能力约束、固定在一个可控轨道里运行的装置”。在软件领域Harness 这个词至少有三种常见理解语境含义典型对象测试领域测试夹具、测试执行框架负责准备数据、调用被测对象、收集结果JUnit、pytest fixtureDevOps 领域持续交付/发布编排平台负责把构建产物安全发布到环境Harness.io 等 CD 平台AI Agent 领域大模型与外部工具之间的编排与控制层负责上下文管理、工具调度、安全策略Codex Harness、DeepSeek Harness 等本文讨论的 Harness 架构重点聚焦在“AI 应用开发中模型外围的编排与控制层”。这是当前社区讨论热度最高、面试也最容易问到的方向。但要理解它你不能只盯着 AI 场景因为 Harness 的设计思想来自更早的测试工程和 DevOps 实践。1.2 底层驱动为什么人人都开始聊 Harness过去两年大模型应用开发经历了一次明显的范式转变。早期大家写 Agent代码通常是这样的def handle_user_query(query): if 天气 in query: return get_weather(city) elif 日历 in query: return create_event(event_info) else: return llm.chat(query)这种硬编码路由方式在小规模场景下勉强能用但一旦工具数量变多、对话上下文变长、权限控制变得复杂代码会迅速腐化。你很难回答下面这些问题模型到底调用了哪些工具调用顺序是什么一次请求消耗了多少 token失败重试了几次某个工具只允许特定角色调用这个约束写在哪里当模型幻觉导致工具参数错误时谁负责兜底Harness 架构的核心目标就是把“模型能力”和“业务流程”之间的所有非业务逻辑抽出来统一管理。它解决的不是单次调用而是 Agent 生命周期中的可观测、可控制、可编排问题。这个定位和测试工程中的 Test Harness“准备环境、执行用例、汇总报告”非常相似也和 DevOps 中的 CD Harness“发布前检查、灰度、回滚”一脉相承。理解了这一层你就能看懂为什么开源社区和面试官如此重视 Harness。1.3 Harness、Agent、框架三者的关系面试时经常有人把 Harness 和 Agent、框架混为一谈这里先做一个边界区分。Agent 是“智能体”强调自主决策能力它内部会依赖模型推理。Framework 是“开发框架”提供构建 Agent 的 API 和基础设施例如 LangChain、LlamaIndex。Harness 是“控制/编排层”它不负责模型推理本身而是负责模型之外的上下文、工具、策略、观测和错误处理。一个 Harness 可以驱动多个 Agent 运行也可以被嵌入到任何框架之上。架构分层大致如下业务应用层 ↓ Harness 编排控制层 ← 本文重点 ↓ Agent / 模型调用层 ↓ 工具 / 外部系统把 Harness 理解成“模型与现实世界之间的隔离舱”会更容易。它既保证模型能触达外部工具又保证外部工具不会因为模型失控而被误用。2. Harness 架构的核心组件拆解一个完整的 Harness 架构通常包含 6 个核心组件。下面逐一拆解。2.1 上下文管理模块上下文管理负责维护一次会话中模型能看到的所有信息包括用户输入和系统提示词工具调用结果历史消息摘要当前会话的业务状态权限与角色信息。设计上要注意两点一是上下文不能无限增长需要做截断或摘要策略二是必须保持“可见性”即模型每次决策前能依据的上下文应当是确定、可审计的。常见实现class HarnessContext: def __init__(self, system_prompt: str, max_turns: int 20): self.messages [{role: system, content: system_prompt}] self.max_turns max_turns def add_user_message(self, content: str): self.messages.append({role: user, content: content}) def add_assistant_message(self, content: str): self.messages.append({role: assistant, content: content}) def add_tool_result(self, tool_name: str, result: str): self.messages.append({ role: tool, name: tool_name, content: result }) def compact_if_needed(self): if len(self.messages) self.max_turns * 2: # 这里可以调用摘要模型把早期消息压缩成 summary self.messages [self.messages[0]] self.messages[-20:]2.2 工具注册与路由中心工具注册中心是 Harness 的“能力清单”它维护了所有可被模型调用的工具及其元信息例如工具名称、描述、参数 JSON Schema、权限要求、是否幂等等。为什么不能把工具直接硬编码在业务逻辑里因为工具注册中心把“能力描述”和“能力实现”分离模型只需要读取描述就能决定是否调用、怎么调用而 Harness 负责把模型给的参数安全地传给真实实现。class ToolRegistry: def __init__(self): self._tools {} def register(self, tool): self._tools[tool.name] tool return self def get(self, name): return self._tools.get(name) def list_tools(self): return [ {name: t.name, description: t.description, schema: t.parameters_schema} for t in self._tools.values() ]2.3 执行引擎与调度器执行引擎是 Harness 的核心循环。它负责接收模型返回的“意图”和“工具调用请求”做参数校验真正执行工具把结果写回上下文决定是否继续让模型推理还是返回最终答案。一个简化版的执行引擎调度逻辑def execute_tool_call(tool_call, registry, context, policy): tool_name tool_call[tool] params tool_call[params] # 1. 校验参数 schema validate_params(tool_name, params) # 2. 权限检查 check_permission(policy, tool_name, params) # 3. 执行工具 tool registry.get(tool_name) result tool.run(**params) # 4. 写回上下文 context.add_tool_result(tool_name, json.dumps(result, ensure_asciiFalse)) return result2.4 策略、限流与错误恢复Harness 之所以适合生产环境是因为它把“策略”从业务代码中抽出来集中管理。常见策略包括工具调用次数上限防止模型陷入死循环单次工具执行超时时间限流与配额控制敏感操作二次确认失败重试与降级方案。错误恢复是 Harness 最容易忽略的部分。模型经常出现“工具参数格式错误”或“工具不存在”这类问题Harness 应该把这些错误原样写回上下文让模型自行修正而不是直接抛异常终止整个会话。2.5 观测、日志与追踪生产中排查 Agent 问题时最痛苦的是“不知道模型内部发生了什么”。Harness 需要在每个关键节点埋点模型请求/响应耗时token 消耗工具调用记录错误堆栈上下文截断情况。日志建议采用结构化 JSON 格式写入独立的检索系统。这样面试时你可以直接说“我们的 Harness 支持从 traceId 查询一次完整会话中每一次模型调用、工具执行和上下文变化”。2.6 安全与权限边界大模型应用的安全问题不能等到上线后再补。Harness 必须在设计阶段就包含权限边界典型手段包括工具级权限不同角色只能调用某些工具参数级校验限制参数取值范围例如金额、地址等敏感字段人工审批节点高风险操作必须暂停等待人工确认网络隔离内部系统与公网工具使用不同网络策略。class PermissionPolicy: def __init__(self, role_permissions: dict): self.role_permissions role_permissions def can_call(self, role, tool_name): return tool_name in self.role_permissions.get(role, [])3. Harness 分层设计从原理到架构图很多新手拿到 Harness 源码后不知道从哪看起原因是缺少分层视角。Harness 并不是一个单独的大类而是由多个层次组合而成。3.1 编排层编排层是 Harness 对外的门面提供 Agent 运行入口、会话管理、任务分发能力。对外暴露的是run()、stream()、invoke()这类 API。在这个层面你需要考虑同步调用和流式调用的区别。流式输出在用户体验上明显更好但会显著增加 Harness 的复杂度因为每个 token 都要经过管道转发。3.2 语义层语义层负责“理解模型在想什么”。具体来说它需要从模型返回结果中解析出工具调用意图判断当前是否应该继续执行工具还是终止对模型输出做结构化处理。在很多开源实现中语义层通过 function calling 机制来完成也有部分实现通过提示词约束模型输出 JSON再由语义层解析。两种方案各有优劣function calling 更稳定prompt 解析则兼容性更强。3.3 执行层执行层是真正接触外部系统的地方。它负责工具调用、数据访问、API 请求。执行层必须实现“失败隔离”不能让单个工具异常拖垮整个 Harness 进程。def safe_call_tool(tool, **params): try: return {status: success, data: tool.run(**params)} except TimeoutError: return {status: error, error: tool timeout} except Exception as e: return {status: error, error: str(e)}每层职责独立后Harness 的扩展性会大幅提升你可以不改执行层只新增一个策略插件也可以替换底层模型但不影响工具注册逻辑。3.4 观测层观测层负责把 Harness 运行过程中产生的指标、日志、链路追踪数据暴露出去。包括 Prometheus 指标、结构化日志、OpenTelemetry 链路追踪等。生产环境建议至少覆盖以下指标Harness 整体延迟分位数p50/p95/p99模型调用失败率工具调用成功率上下文 token 使用量单次会话工具调用次数分布。3.5 配置层配置层负责隔离不同环境的策略、模型参数、工具列表。例如开发环境允许调试工具生产环境禁用不同部门可能使用不同的模型 API Key。配置推荐使用 YAML 或 JSON 文件维护并在 Harness 启动时加载为配置对象。不建议把配置散落在代码各处。harness: max_iterations: 10 timeout_seconds: 60 roles: admin: [get_order, create_order, cancel_order] visitor: [get_order]4. 实战从零实现一个轻量级 Harness纸上谈兵容易动手实现才是理解 Harness 的最佳路径。下面我们基于 Python 手写一个“轻量级 Harness Demo”你可以直接复制运行再逐步扩展成自己的工程框架。4.1 项目结构与依赖harness_demo/ ├── core/ │ ├── __init__.py │ ├── context.py │ ├── registry.py │ ├── harness.py │ └── llm_client.py ├── tools/ │ ├── __init__.py │ └── weather_tool.py ├── main.py └── README.md本项目依赖仅使用 Python 标准库运行环境建议 Python 3.10无需额外安装第三方包。4.2 实现上下文模块# core/context.py from typing import List, Dict class HarnessContext: def __init__(self, system_prompt: str): self.messages: List[Dict[str, str]] [ {role: system, content: system_prompt} ] def add_user_message(self, content: str): self.messages.append({role: user, content: content}) def add_assistant_message(self, content: str): self.messages.append({role: assistant, content: content}) def add_tool_result(self, tool_name: str, content: str): self.messages.append({ role: tool, name: tool_name, content: content }) def to_messages(self) - List[Dict[str, str]]: return self.messages def compact(self, keep_last: int 6): if len(self.messages) keep_last 1: # 保留 system prompt 和最后 keep_last 条消息 self.messages [self.messages[0]] self.messages[-keep_last:]4.3 实现工具注册中心# core/registry.py from typing import Any, Callable, Dict, List class Tool: def __init__( self, name: str, description: str, func: Callable[..., Any], parameters_schema: Dict[str, Any], require_permission: bool False ): self.name name self.description description self.func func self.parameters_schema parameters_schema self.require_permission require_permission def run(self, **kwargs): return self.func(**kwargs) class ToolRegistry: def __init__(self): self._tools: Dict[str, Tool] {} def register(self, tool: Tool) - ToolRegistry: self._tools[tool.name] tool return self def get(self, name: str) - Tool | None: return self._tools.get(name) def list_tools(self) - List[Dict[str, Any]]: return [ { name: tool.name, description: tool.description, parameters_schema: tool.parameters_schema, } for tool in self._tools.values() ]4.4 实现 Harness 主流程下面是整个 Harness 的核心循环。为了让 Demo 不依赖外部 API Key我们先实现一个FakeLLMClient它通过规则代替真实的模型推理后面会给出真实大模型的接入方式。# core/harness.py import json import logging import time from .context import HarnessContext from .registry import ToolRegistry logger logging.getLogger(__name__) class Harness: def __init__( self, llm_client, registry: ToolRegistry, max_iterations: int 5, ): self.llm_client llm_client self.registry registry self.max_iterations max_iterations def run(self, context: HarnessContext, user_input: str) - str: context.add_user_message(user_input) tool_schema self.registry.list_tools() for i in range(self.max_iterations): # 1. 调用模型得到意图 response self.llm_client.invoke( messagescontext.to_messages(), toolstool_schema, ) # 2. 如果没有工具调用说明可以直接返回 if not response.get(tool_calls): final_answer response.get(content, ) context.add_assistant_message(final_answer) return final_answer # 3. 执行工具调用 for tool_call in response[tool_calls]: tool_name tool_call[tool] params tool_call.get(params, {}) tool self.registry.get(tool_name) if not tool: context.add_tool_result( tool_name, json.dumps({error: tool not found}) ) continue logger.info(iteration%s tool%s params%s, i, tool_name, params) start time.time() try: result tool.run(**params) result_str json.dumps(result, ensure_asciiFalse) except Exception as exc: logger.exception(tool execution failed) result_str json.dumps({error: str(exc)}) duration_ms (time.time() - start) * 1000 logger.info(tool%s duration_ms%.1f, tool_name, duration_ms) context.add_tool_result(tool_name, result_str) context.add_assistant_message(达到最大迭代次数终止任务。) return 执行超时已达到最大迭代次数4.5 实现一个简单的 LLM 客户端为了让你直接跑通这里用规则模拟模型行为。真实项目中你只需要把invoke方法替换为对大模型 API 的调用。# core/llm_client.py from typing import List, Dict, Any class FakeLLMClient: 规则型伪模型用于本地演示。 仅用于理解 Harness 流程不代表真实模型能力。 def __init__(self, rules: Dict[str, Any]): self.rules rules def invoke(self, messages, tools, **kwargs) - Dict[str, Any]: last_user_message messages[-1][content] for keyword, action in self.rules.items(): if keyword in last_user_message: return { tool_calls: [action], content: None, } return { tool_calls: [], content: 我理解你的问题但当前没有可用工具处理。 }4.6 注册业务工具我们模拟一个查询天气的工具并假设天气数据来自内部接口。# tools/weather_tool.py import json import random def get_weather(city: str, date: str today): # 真实项目中应调用天气服务 API这里用随机数模拟 temperature random.randint(-5, 35) return { city: city, date: date, temperature: temperature, condition: 晴 if temperature 10 else 多云 }4.7 组装主程序# main.py import logging from core.context import HarnessContext from core.registry import ToolRegistry, Tool from core.harness import Harness from core.llm_client import FakeLLMClient from tools.weather_tool import get_weather logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, ) SYSTEM_PROMPT 你是一个智能助手可以通过工具获取天气信息并回答用户问题。 def main(): # 1. 创建工具注册中心 registry ToolRegistry() registry.register( Tool( nameget_weather, description查询指定城市的天气信息, funcget_weather, parameters_schema{ type: object, properties: { city: {type: string, description: 城市名}, date: {type: string, description: 日期默认今天}, }, required: [city], }, ) ) # 2. 创建伪模型客户端 llm_client FakeLLMClient( rules{ 天气: { tool: get_weather, params: {city: 上海, date: today}, }, } ) # 3. 组装 Harness harness Harness( llm_clientllm_client, registryregistry, max_iterations3, ) # 4. 运行 context HarnessContext(system_promptSYSTEM_PROMPT) answer harness.run(context, 帮我查一下上海的天气) print(最终回答:, answer) print(\n 当前上下文 ) for msg in context.to_messages(): print(msg) if __name__ __main__: main()运行命令python main.py预期输出类似随机数据温度数值不定2024-xx-xx xx:xx:xx - ... - iteration0 toolget_weather params{city: 上海, date: today} 最终回答: 执行超时已达到最大迭代次数 当前上下文 {role: system, content: 你是一个智能助手可以通过工具获取天气信息并回答用户问题。} ...这里出现“最终回答是执行超时”是因为我们的 FakeLLMClient 在工具执行结束后不再继续生成最终回答。真实场景中模型会在拿到工具结果后继续推理并生成回复。为了让 Demo 更完整可以优化 FakeLLMClient当上下文存在 tool 角色消息时自动生成一句总结。这里留作练习你可以自行完善。4.8 接入真实大模型 API接入真实模型时核心工作是把llm_client.invoke替换为 HTTP 调用。需要注意不同模型的 function calling 协议不同需要把 Harness 的工具描述转换为模型要求的格式模型返回的 tool_call 需要解析后再交给 Harness。示例思路如下以 OpenAI 兼容接口为例import requests import json class OpenAIClient: def __init__(self, api_key: str, base_url: str, model: str): self.api_key api_key self.base_url base_url self.model model def invoke(self, messages, tools, **kwargs): # 转换为 OpenAI 格式 openai_tools [] for tool in tools: openai_tools.append({ type: function, function: { name: tool[name], description: tool[description], parameters: tool[parameters_schema], }, }) resp requests.post( f{self.base_url}/chat/completions, headers{Authorization: fBearer {self.api_key}}, json{ model: self.model, messages: messages, tools: openai_tools, }, timeout30, ) resp.raise_for_status() data resp.json() # 把 OpenAI 返回值转换为 Harness 内部格式 tool_calls [] content None if data[choices]: choice data[choices][0][message] content choice.get(content) if choice.get(tool_calls): for tc in choice[tool_calls]: tool_calls.append({ tool: tc[function][name], params: json.loads(tc[function][arguments]), }) return {tool_calls: tool_calls, content: content}这里强调一下不同版本 SDK 的返回字段可能存在差异实际接入时以官方文档为准。生产环境不建议直接使用 requests 裸调应优先使用官方 SDK 或封装良好的 HTTP 客户端并补充重试、超时、熔断机制。5. 从实战到工程化Harness 在真实项目中的应用模式Demo 能跑通只是第一步。下面结合真实业务场景聊聊 Harness 在不同项目中的应用模式。5.1 AI 应用模型能力与业务规则之间的隔离层在 AI 应用中Harness 最大的价值是“隔离”。它把模型能力和业务规则隔离开使系统不会因为模型输出不稳定而失控。举例来说一个客服机器人可能需要调用订单查询、退款申请、物流跟踪三个工具。其中“退款申请”是高风险操作Harness 可以配置为只有用户角色为 VIP 且金额小于 500 元时才允许直接执行否则转入人工审批。这类逻辑如果写在业务代码里散落各处很难维护写在 Harness 策略层则一目了然。5.2 自动化测试测试 harness 的设计思路传统测试领域同样大量使用 Harness 思想。测试 Harness 需要处理测试环境准备数据库、Redis、外部依赖 mock用例执行与断言测试报告生成失败用例自动重跑。一个设计良好的测试 Harness允许测试人员只关注“测试数据”和“断言逻辑”而不用关心环境怎么启动、数据怎么回滚、报告怎么生成。5.3 持续交付CI/CD 平台中的 harness 概念DevOps 领域的 Harness 更偏向“发布编排”。它的核心能力包括环境检查、灰度发布、自动回滚、审批流。虽然和 AI Harness 技术细节不同但设计思想高度一致——都是在复杂系统中引入可控的编排层。对面试者来说能在不同语境中灵活切换 Harness 的定义本身就说明你对架构抽象有深入理解。这是一个很加分的点。6. Harness 架构的最佳实践在真实项目中落地 Harness下面几条经验值得参考。6.1 最小权限原则不要让 Harness 拥有超出必要的权限。模型调用的工具应当按角色、按环境、按操作类型严格控制。生产环境的工具调用必须支持审计回溯。# 生产环境策略示例 production: allowed_roles: customer_service: [query_order, query_logistics] manager: [query_order, refund_order, create_coupon] sensitive_tools: refund_order: require_approval: true max_amount: 5006.2 上下文可见性与透明审计任何写进上下文的信息都应当在日志中留痕。尤其是工具返回结果可能包含敏感字段进入上下文前需要做脱敏处理避免被模型输出到对话中。6.3 工具协议标准化工具注册时不要只写“函数名”和“函数体”还需要定义清晰的参数 schema、错误码、幂等性。推荐内部统一使用 JSON Schema 作为参数描述标准。6.4 配置与版本管理Harness 的配置工具列表、权限策略、模型参数应当纳入版本管理使用 Git 管理变更。任何配置修改都走审批流程保证生产环境可追溯。6.5 可观测性建设上线前先部署观测体系而不是上线后再补。至少包含日志检索、指标监控、链路追踪。6.6 灰度发布与回滚Harness 本身也可能是业务逻辑的一部分当 Harness 代码变更时会直接影响生产行为。建议对 Harness 版本做灰度发布并准备一键回滚方案。7. 高频面试题与面试准备思路7.1 面试官真正想考什么面试官问 Harness通常不是考你背诵的概念而是想确认三件事你是否理解复杂系统中“控制层”存在的必要你是否能在实际项目中识别出需要 Harness 的场景你是否有能力设计一个可控、可观测、可扩展的编排层。7.2 高频面试题与参考回答问题 1什么是 Harness它和 Agent 的区别是什么参考回答思路Harness 是模型外围的编排与控制层负责上下文管理、工具调度、权限控制、错误恢复和观测。Agent 是智能体本身负责决策和推理。Harness 不参与模型推理而是为 Agent 提供安全可控的运行环境。可以理解成“Agent 是发动机Harness 是整车控制系统”。问题 2你如何设计一个 Harness 架构核心组件有哪些回答时建议画分层图面试中口述即可然后按顺序说六层上下文管理工具注册中心执行引擎策略层观测层配置层。问题 3模型调用工具后如何把结果安全地传给模型回答重点工具结果先经过脱敏和长度限制失败信息也要结构化返回返回结果会写回上下文模型下一轮推理才能看到对于超大结果应先做摘要再传给模型。问题 4如果你的 Agent 在循环调用某个工具停不下来怎么处理回答重点设置最大迭代次数检测相同工具连续调用次数并触发熔断增加 token 消耗限制引入人工干预接口。问题 5Harness 如何保证权限安全回答要点工具级权限参数级校验敏感操作人工审批完整的审计日志。7.3 面试答题框架STAR 分层思考如果你有实际项目经验推荐用 STAR 法则组织语言Situation项目背景为什么需要 HarnessTask要解决的架构问题Action你做了什么设计核心组件怎么落地Result最终效果量化指标如工具调用成功率提升到 99%错误恢复时间下降 50%。如果没有生产项目经验把第 4 节 Demo 扩写成一个小型项目本质上是完全可行的。面试官更看重你的思考深度而不是一定要你有千万级流量的项目。8. 总结与后续学习路线本文从 Harness 的概念讲起拆解了上下文、工具注册、执行引擎、策略、观测、安全 6 大核心组件并结合 Python 实现了一个可运行的轻量级 Harness Demo。同时给出了真实项目中的工程化建议和面试答题框架。如果你希望继续深入建议按下面顺序学习精读一个开源 Harness 项目的源码重点关注它的执行循环、工具协议、错误处理扩展你的 Demo加入真实模型、权限策略、日志追踪研究 function calling 协议理解不同模型在工具调用上的差异测试你自己的 Harness 在长时间运行、高并发、输入异常情况下的表现。Harness 架构的价值是在复杂系统中体现出来的。当你面对的不再是“模型能不能回答”的问题而是“模型会不会乱调用工具、会不会占用过多资源、失败了怎么恢复”这些问题时你就会真正理解 Harness 存在的意义。希望这篇文章能帮你少踩一些坑顺利把 Harness 落地到自己的项目中。