1. 引言Harness Engineering 是什么Harness Engineering工程化编排是近年来在 AI Agent、自动化流水线和复杂系统集成领域快速兴起的一类工程实践。它的核心目标是把多个松散的组件——模型、工具、数据源、人工审批、外部服务——通过一套可编排、可观测、可回滚的工程框架组织成稳定、可控、可复用的自动化流程。简单来说Harness Engineering 解决的是「如何把能力变成可靠的工程系统」的问题。它关注的不只是单个模型或单个工具的效果而是整条链路的稳定性、可维护性和可治理性。2. 核心概念拆解要理解 Harness Engineering需要先厘清几个关键概念Harness编排框架承载流程定义、状态管理、错误处理和资源调度的运行容器。Step步骤流程中的最小执行单元可以是调用模型、执行代码、查询数据库或触发外部 API。Workflow工作流由多个 Step 按顺序或条件组合而成的完整执行链路。Guardrail护栏对输入输出进行校验、限流、审计和人工确认的机制是 Harness 区别于普通脚本的关键。Observability可观测性对每一步的输入、输出、耗时、成本和失败原因进行记录与追踪。3. Harness Engineering 与普通脚本的区别很多人会问这不就是写脚本把几个 API 串起来吗区别在于工程化程度维度普通脚本Harness Engineering错误处理try-catch 散落各处统一的重试、降级、熔断策略状态管理全局变量显式的工作流状态机可观测性print 日志结构化追踪、指标采集、链路回溯人工介入难以实现内置审批节点、暂停恢复复用性复制粘贴Step 组件化、版本化4. 代码实战构建一个最小 Harness 框架下面我们用 Python 从零实现一个轻量级 Harness 框架包含 Step 抽象、工作流编排、重试机制和结构化日志。先定义基础组件from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional import time import uuid import logging from enum import Enum logging.basicConfig(levellogging.INFO) logger logging.getLogger(harness) class StepStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed SKIPPED skipped dataclass class StepResult: step_name: str status: StepStatus output: Any None error: Optional[str] None duration_ms: float 0.0 retries: int 0 class Step: 所有步骤的基类子类实现 execute 方法即可。 def __init__(self, name: str, max_retries: int 2, timeout_ms: int 5000): self.name name self.max_retries max_retries self.timeout_ms timeout_ms def execute(self, context: Dict[str, Any]) - Any: raise NotImplementedError def run(self, context: Dict[str, Any]) - StepResult: start time.time() attempt 0 while True: try: logger.info(f[{self.name}] attempt{attempt 1} start) output self.execute(context) duration (time.time() - start) * 1000 logger.info(f[{self.name}] success in {duration:.1f}ms) return StepResult( step_nameself.name, statusStepStatus.SUCCESS, outputoutput, duration_msduration, retriesattempt, ) except Exception as e: attempt 1 duration (time.time() - start) * 1000 if attempt gt; self.max_retries: logger.error(f[{self.name}] failed after {attempt} attempts: {e}) return StepResult( step_nameself.name, statusStepStatus.FAILED, errorstr(e), duration_msduration, retriesattempt - 1, ) logger.warning(f[{self.name}] attempt{attempt} error{e}, retrying...) time.sleep(0.2 * attempt)/code/pre 5. 工作流引擎实现 有了 Step 基类接下来实现 Workflow 引擎负责按顺序执行步骤、传递上下文、收集结果 dataclass class WorkflowResult: workflow_id: str status: StepStatus step_results: List[StepResult] field(default_factorylist) context: Dict[str, Any] field(default_factorydict) class Workflow: 按顺序执行一组 Step共享一个 context 字典。 def init(self, name: str): self.name name self.steps: List[Step] [] def add_step(self, step: Step) - Workflow: self.steps.append(step) return self def run(self, initial_context: Optional[Dict[str, Any]] None) - WorkflowResult: workflow_id uuid.uuid4().hex[:8] context dict(initial_context or {}) results: List[StepResult] [] logger.info(f[workflow:{workflow_id}] {self.name} started with {len(self.steps)} steps) for step in self.steps: result step.run(context) results.append(result) if result.status StepStatus.SUCCESS: # 把输出写入共享上下文供后续步骤使用 context[step.name] result.output else: logger.error(f[workflow:{workflow_id}] step {step.name} failed, aborting) return WorkflowResult( workflow_idworkflow_id, statusStepStatus.FAILED, step_resultsresults, contextcontext, ) logger.info(f[workflow:{workflow_id}] completed successfully) return WorkflowResult( workflow_idworkflow_id, statusStepStatus.SUCCESS, step_resultsresults, contextcontext, )lt;/codegt;lt;/pregt; 实战示例构建一个带护栏的 AI 内容审核工作流 下面用一个真实场景串联整个框架对用户提交的文本先做敏感词过滤再调用大模型生成摘要最后经过人工审批节点。先实现具体的 Step class SensitiveWordFilter(Step): 护栏步骤检查输入是否包含敏感词。 def init(self, name: str, sensitive_words: List[str]): super().init(name) self.sensitive_words sensitive_words def execute(self, context: Dict[str, Any]) - Any: text context.get(input_text, ) hit_words [w for w in self.sensitive_words if w in text] if hit_words: raise ValueError(f包含敏感词: {hit_words}) return {filtered: True, text: text} class LLMSummarizer(Step): 调用大模型生成摘要此处用模拟实现。 def execute(self, context: Dict[str, Any]) - Any: text context[input_text] 真实场景这里调用 OpenAI / Claude / 本地模型 API summary call_llm(f请总结{text}) summary f[模拟摘要] 原文共 {len(text)} 字主题为示例内容。 return {summary: summary} class HumanApproval(Step): 人工审批节点模拟等待人工确认。 def execute(self, context: Dict[str, Any]) - Any: summary context[LLMSummarizer][summary] 真实场景这里会推送审批任务到 IM/邮件等待回调 approved True # 模拟审批通过 if not approved: raise ValueError(人工审批未通过) return {approved: True, summary: summary}/code/pre 7. 组装并运行工作流 def main(): 1. 定义护栏词表 sensitive_words [违规词A, 违规词B] 2. 组装工作流 wf Workflow(content_review_pipeline) wf.add_step(SensitiveWordFilter(SensitiveWordFilter, sensitive_words)) wf.add_step(LLMSummarizer(LLMSummarizer)) wf.add_step(HumanApproval(HumanApproval)) 3. 运行 result wf.run({input_text: 这是一段需要审核的正常内容用于演示 Harness 工作流。}) 4. 输出结果 print(f工作流状态: {result.status.value}) for sr in result.step_results: print(f - {sr.step_name}: {sr.status.value} ({sr.duration_ms:.1f}ms)) if result.status StepStatus.SUCCESS: print(f最终摘要: {result.context[HumanApproval][summary]}) if name main: main() 运行输出示例 [workflow:3f2a9c1d] content_review_pipeline started with 3 steps [SensitiveWordFilter] attempt1 start [SensitiveWordFilter] success in 0.2ms [LLMSummarizer] attempt1 start [LLMSummarizer] success in 1.1ms [HumanApproval] attempt1 start [HumanApproval] success in 0.3ms [workflow:3f2a9c1d] completed successfully 工作流状态: success SensitiveWordFilter: success (0.2ms) LLMSummarizer: success (1.1ms) HumanApproval: success (0.3ms) 最终摘要: [模拟摘要] 原文共 28 字主题为示例内容。 进阶条件分支与并行执行 真实场景往往不是简单的线性链路。下面扩展 Workflow 支持条件分支 class ConditionalStep(Step): 根据条件决定执行哪个子步骤。 def init(self, name: str, condition: Callable[[Dict[str, Any]], bool], if_step: Step, else_step: Optional[Step] None): super().init(name) self.condition condition self.if_step if_step self.else_step else_step def execute(self, context: Dict[str, Any]) - Any: if self.condition(context): return self.if_step.run(context) elif self.else_step: return self.else_step.run(context) return {skipped: True} 使用示例内容长度超过阈值才走详细审核 def is_long_text(ctx): return len(ctx.get(input_text, )) 50 wf Workflow(conditional_pipeline) wf.add_step(SensitiveWordFilter(SensitiveWordFilter, [违规词A])) wf.add_step(ConditionalStep( RouteByLength, conditionis_long_text, if_stepLLMSummarizer(LLMSummarizer), else_stepHumanApproval(HumanApproval), )) 9. 可观测性结构化追踪 生产环境必须能回溯每一步的执行情况。在 Step.run 中已经记录了耗时和重试次数进一步可以接入追踪系统 import json import datetime def export_trace(result: WorkflowResult) - str: 把工作流执行结果导出为 JSON 追踪日志。 trace { workflow_id: result.workflow_id, status: result.status.value, timestamp: datetime.datetime.utcnow().isoformat(), steps: [ { name: sr.step_name, status: sr.status.value, duration_ms: round(sr.duration_ms, 2), retries: sr.retries, error: sr.error, } for sr in result.step_results ], } return json.dumps(trace, ensure_asciiFalse, indent2) 使用 trace_json export_trace(result) print(trace_json) 10. 生产落地的关键考量 从 Demo 到生产Harness Engineering 还需要关注以下几点 持久化工作流状态要写入数据库支持中断恢复和重新执行。 幂等性每个 Step 要设计成可重复执行且结果一致避免重试造成副作用。 超时控制外部 API 调用必须设置超时和熔断防止链路阻塞。 审计日志涉及人工审批和敏感数据的步骤要记录完整的操作轨迹。 版本管理工作流定义要纳入版本控制支持灰度发布和快速回滚。 成本控制对模型调用等昂贵步骤做预算限制和用量统计。 11. 总结 Harness Engineering 的本质是把「能跑通的脚本」升级为「可治理的工程系统」。它通过 Step 抽象、工作流编排、护栏机制和可观测性让复杂的自动化链路变得稳定、可控、可审计。本文从零实现了一个轻量级框架并演示了带敏感词过滤、模型调用和人工审批的完整工作流。生产环境中可以基于同样的思想借助成熟的编排平台或自研框架把 Harness Engineering 落地到实际业务中。
1. 引言Harness Engineering 是什么Harness Engineering工程化编排是近年来在 AI Agent、自动化流水线和复杂系统集成领域快速兴起的一类工程实践。它的核心目标是把多个松散的组件——模型、工具、数据源、人工审批、外部服务——通过一套可编排、可观测、可回滚的工程框架组织成稳定、可控、可复用的自动化流程。简单来说Harness Engineering 解决的是「如何把能力变成可靠的工程系统」的问题。它关注的不只是单个模型或单个工具的效果而是整条链路的稳定性、可维护性和可治理性。2. 核心概念拆解要理解 Harness Engineering需要先厘清几个关键概念Harness编排框架承载流程定义、状态管理、错误处理和资源调度的运行容器。Step步骤流程中的最小执行单元可以是调用模型、执行代码、查询数据库或触发外部 API。Workflow工作流由多个 Step 按顺序或条件组合而成的完整执行链路。Guardrail护栏对输入输出进行校验、限流、审计和人工确认的机制是 Harness 区别于普通脚本的关键。Observability可观测性对每一步的输入、输出、耗时、成本和失败原因进行记录与追踪。3. Harness Engineering 与普通脚本的区别很多人会问这不就是写脚本把几个 API 串起来吗区别在于工程化程度维度普通脚本Harness Engineering错误处理try-catch 散落各处统一的重试、降级、熔断策略状态管理全局变量显式的工作流状态机可观测性print 日志结构化追踪、指标采集、链路回溯人工介入难以实现内置审批节点、暂停恢复复用性复制粘贴Step 组件化、版本化4. 代码实战构建一个最小 Harness 框架下面我们用 Python 从零实现一个轻量级 Harness 框架包含 Step 抽象、工作流编排、重试机制和结构化日志。先定义基础组件from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional import time import uuid import logging from enum import Enum logging.basicConfig(levellogging.INFO) logger logging.getLogger(harness) class StepStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed SKIPPED skipped dataclass class StepResult: step_name: str status: StepStatus output: Any None error: Optional[str] None duration_ms: float 0.0 retries: int 0 class Step: 所有步骤的基类子类实现 execute 方法即可。 def __init__(self, name: str, max_retries: int 2, timeout_ms: int 5000): self.name name self.max_retries max_retries self.timeout_ms timeout_ms def execute(self, context: Dict[str, Any]) - Any: raise NotImplementedError def run(self, context: Dict[str, Any]) - StepResult: start time.time() attempt 0 while True: try: logger.info(f[{self.name}] attempt{attempt 1} start) output self.execute(context) duration (time.time() - start) * 1000 logger.info(f[{self.name}] success in {duration:.1f}ms) return StepResult( step_nameself.name, statusStepStatus.SUCCESS, outputoutput, duration_msduration, retriesattempt, ) except Exception as e: attempt 1 duration (time.time() - start) * 1000 if attempt gt; self.max_retries: logger.error(f[{self.name}] failed after {attempt} attempts: {e}) return StepResult( step_nameself.name, statusStepStatus.FAILED, errorstr(e), duration_msduration, retriesattempt - 1, ) logger.warning(f[{self.name}] attempt{attempt} error{e}, retrying...) time.sleep(0.2 * attempt)/code/pre 5. 工作流引擎实现 有了 Step 基类接下来实现 Workflow 引擎负责按顺序执行步骤、传递上下文、收集结果 dataclass class WorkflowResult: workflow_id: str status: StepStatus step_results: List[StepResult] field(default_factorylist) context: Dict[str, Any] field(default_factorydict) class Workflow: 按顺序执行一组 Step共享一个 context 字典。 def init(self, name: str): self.name name self.steps: List[Step] [] def add_step(self, step: Step) - Workflow: self.steps.append(step) return self def run(self, initial_context: Optional[Dict[str, Any]] None) - WorkflowResult: workflow_id uuid.uuid4().hex[:8] context dict(initial_context or {}) results: List[StepResult] [] logger.info(f[workflow:{workflow_id}] {self.name} started with {len(self.steps)} steps) for step in self.steps: result step.run(context) results.append(result) if result.status StepStatus.SUCCESS: # 把输出写入共享上下文供后续步骤使用 context[step.name] result.output else: logger.error(f[workflow:{workflow_id}] step {step.name} failed, aborting) return WorkflowResult( workflow_idworkflow_id, statusStepStatus.FAILED, step_resultsresults, contextcontext, ) logger.info(f[workflow:{workflow_id}] completed successfully) return WorkflowResult( workflow_idworkflow_id, statusStepStatus.SUCCESS, step_resultsresults, contextcontext, )lt;/codegt;lt;/pregt; 实战示例构建一个带护栏的 AI 内容审核工作流 下面用一个真实场景串联整个框架对用户提交的文本先做敏感词过滤再调用大模型生成摘要最后经过人工审批节点。先实现具体的 Step class SensitiveWordFilter(Step): 护栏步骤检查输入是否包含敏感词。 def init(self, name: str, sensitive_words: List[str]): super().init(name) self.sensitive_words sensitive_words def execute(self, context: Dict[str, Any]) - Any: text context.get(input_text, ) hit_words [w for w in self.sensitive_words if w in text] if hit_words: raise ValueError(f包含敏感词: {hit_words}) return {filtered: True, text: text} class LLMSummarizer(Step): 调用大模型生成摘要此处用模拟实现。 def execute(self, context: Dict[str, Any]) - Any: text context[input_text] 真实场景这里调用 OpenAI / Claude / 本地模型 API summary call_llm(f请总结{text}) summary f[模拟摘要] 原文共 {len(text)} 字主题为示例内容。 return {summary: summary} class HumanApproval(Step): 人工审批节点模拟等待人工确认。 def execute(self, context: Dict[str, Any]) - Any: summary context[LLMSummarizer][summary] 真实场景这里会推送审批任务到 IM/邮件等待回调 approved True # 模拟审批通过 if not approved: raise ValueError(人工审批未通过) return {approved: True, summary: summary}/code/pre 7. 组装并运行工作流 def main(): 1. 定义护栏词表 sensitive_words [违规词A, 违规词B] 2. 组装工作流 wf Workflow(content_review_pipeline) wf.add_step(SensitiveWordFilter(SensitiveWordFilter, sensitive_words)) wf.add_step(LLMSummarizer(LLMSummarizer)) wf.add_step(HumanApproval(HumanApproval)) 3. 运行 result wf.run({input_text: 这是一段需要审核的正常内容用于演示 Harness 工作流。}) 4. 输出结果 print(f工作流状态: {result.status.value}) for sr in result.step_results: print(f - {sr.step_name}: {sr.status.value} ({sr.duration_ms:.1f}ms)) if result.status StepStatus.SUCCESS: print(f最终摘要: {result.context[HumanApproval][summary]}) if name main: main() 运行输出示例 [workflow:3f2a9c1d] content_review_pipeline started with 3 steps [SensitiveWordFilter] attempt1 start [SensitiveWordFilter] success in 0.2ms [LLMSummarizer] attempt1 start [LLMSummarizer] success in 1.1ms [HumanApproval] attempt1 start [HumanApproval] success in 0.3ms [workflow:3f2a9c1d] completed successfully 工作流状态: success SensitiveWordFilter: success (0.2ms) LLMSummarizer: success (1.1ms) HumanApproval: success (0.3ms) 最终摘要: [模拟摘要] 原文共 28 字主题为示例内容。 进阶条件分支与并行执行 真实场景往往不是简单的线性链路。下面扩展 Workflow 支持条件分支 class ConditionalStep(Step): 根据条件决定执行哪个子步骤。 def init(self, name: str, condition: Callable[[Dict[str, Any]], bool], if_step: Step, else_step: Optional[Step] None): super().init(name) self.condition condition self.if_step if_step self.else_step else_step def execute(self, context: Dict[str, Any]) - Any: if self.condition(context): return self.if_step.run(context) elif self.else_step: return self.else_step.run(context) return {skipped: True} 使用示例内容长度超过阈值才走详细审核 def is_long_text(ctx): return len(ctx.get(input_text, )) 50 wf Workflow(conditional_pipeline) wf.add_step(SensitiveWordFilter(SensitiveWordFilter, [违规词A])) wf.add_step(ConditionalStep( RouteByLength, conditionis_long_text, if_stepLLMSummarizer(LLMSummarizer), else_stepHumanApproval(HumanApproval), )) 9. 可观测性结构化追踪 生产环境必须能回溯每一步的执行情况。在 Step.run 中已经记录了耗时和重试次数进一步可以接入追踪系统 import json import datetime def export_trace(result: WorkflowResult) - str: 把工作流执行结果导出为 JSON 追踪日志。 trace { workflow_id: result.workflow_id, status: result.status.value, timestamp: datetime.datetime.utcnow().isoformat(), steps: [ { name: sr.step_name, status: sr.status.value, duration_ms: round(sr.duration_ms, 2), retries: sr.retries, error: sr.error, } for sr in result.step_results ], } return json.dumps(trace, ensure_asciiFalse, indent2) 使用 trace_json export_trace(result) print(trace_json) 10. 生产落地的关键考量 从 Demo 到生产Harness Engineering 还需要关注以下几点 持久化工作流状态要写入数据库支持中断恢复和重新执行。 幂等性每个 Step 要设计成可重复执行且结果一致避免重试造成副作用。 超时控制外部 API 调用必须设置超时和熔断防止链路阻塞。 审计日志涉及人工审批和敏感数据的步骤要记录完整的操作轨迹。 版本管理工作流定义要纳入版本控制支持灰度发布和快速回滚。 成本控制对模型调用等昂贵步骤做预算限制和用量统计。 11. 总结 Harness Engineering 的本质是把「能跑通的脚本」升级为「可治理的工程系统」。它通过 Step 抽象、工作流编排、护栏机制和可观测性让复杂的自动化链路变得稳定、可控、可审计。本文从零实现了一个轻量级框架并演示了带敏感词过滤、模型调用和人工审批的完整工作流。生产环境中可以基于同样的思想借助成熟的编排平台或自研框架把 Harness Engineering 落地到实际业务中。
