)
1. 引言Harness Engineering 是什么Harness Engineering工程化編排是近年來在 AI Agent、自動化流水線和復(fù)雜系統(tǒng)集成領(lǐng)域快速興起的一類工程實踐。它的核心目標(biāo)是把多個松散的組件——模型、工具、數(shù)據(jù)源、人工審批、外部服務(wù)——通過一套可編排、可觀測、可回滾的工程框架組織成穩(wěn)定、可控、可復(fù)用的自動化流程。簡單來說Harness Engineering 解決的是「如何把能力變成可靠的工程系統(tǒng)」的問題。它關(guān)注的不只是單個模型或單個工具的效果而是整條鏈路的穩(wěn)定性、可維護(hù)性和可治理性。2. 核心概念拆解要理解 Harness Engineering需要先厘清幾個關(guān)鍵概念Harness編排框架承載流程定義、狀態(tài)管理、錯誤處理和資源調(diào)度的運行容器。Step步驟流程中的最小執(zhí)行單元可以是調(diào)用模型、執(zhí)行代碼、查詢數(shù)據(jù)庫或觸發(fā)外部 API。Workflow工作流由多個 Step 按順序或條件組合而成的完整執(zhí)行鏈路。Guardrail護(hù)欄對輸入輸出進(jìn)行校驗、限流、審計和人工確認(rèn)的機制是 Harness 區(qū)別于普通腳本的關(guān)鍵。Observability可觀測性對每一步的輸入、輸出、耗時、成本和失敗原因進(jìn)行記錄與追蹤。3. Harness Engineering 與普通腳本的區(qū)別很多人會問這不就是寫腳本把幾個 API 串起來嗎區(qū)別在于工程化程度維度普通腳本Harness Engineering錯誤處理try-catch 散落各處統(tǒng)一的重試、降級、熔斷策略狀態(tài)管理全局變量顯式的工作流狀態(tài)機可觀測性print 日志結(jié)構(gòu)化追蹤、指標(biāo)采集、鏈路回溯人工介入難以實現(xiàn)內(nèi)置審批節(jié)點、暫停恢復(fù)復(fù)用性復(fù)制粘貼Step 組件化、版本化4. 代碼實戰(zhàn)構(gòu)建一個最小 Harness 框架下面我們用 Python 從零實現(xiàn)一個輕量級 Harness 框架包含 Step 抽象、工作流編排、重試機制和結(jié)構(gòu)化日志。先定義基礎(chǔ)組件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: 所有步驟的基類子類實現(xiàn) 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. 工作流引擎實現(xiàn) 有了 Step 基類接下來實現(xiàn) Workflow 引擎負(fù)責(zé)按順序執(zhí)行步驟、傳遞上下文、收集結(jié)果 dataclass class WorkflowResult: workflow_id: str status: StepStatus step_results: List[StepResult] field(default_factorylist) context: Dict[str, Any] field(default_factorydict) class Workflow: 按順序執(zhí)行一組 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: # 把輸出寫入共享上下文供后續(xù)步驟使用 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; 實戰(zhàn)示例構(gòu)建一個帶護(hù)欄的 AI 內(nèi)容審核工作流 下面用一個真實場景串聯(lián)整個框架對用戶提交的文本先做敏感詞過濾再調(diào)用大模型生成摘要最后經(jīng)過人工審批節(jié)點。先實現(xiàn)具體的 Step class SensitiveWordFilter(Step): 護(hù)欄步驟檢查輸入是否包含敏感詞。 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): 調(diào)用大模型生成摘要此處用模擬實現(xiàn)。 def execute(self, context: Dict[str, Any]) - Any: text context[input_text] 真實場景這里調(diào)用 OpenAI / Claude / 本地模型 API summary call_llm(f請總結(jié){text}) summary f[模擬摘要] 原文共 {len(text)} 字主題為示例內(nèi)容。 return {summary: summary} class HumanApproval(Step): 人工審批節(jié)點模擬等待人工確認(rèn)。 def execute(self, context: Dict[str, Any]) - Any: summary context[LLMSummarizer][summary] 真實場景這里會推送審批任務(wù)到 IM/郵件等待回調(diào) approved True # 模擬審批通過 if not approved: raise ValueError(人工審批未通過) return {approved: True, summary: summary}/code/pre 7. 組裝并運行工作流 def main(): 1. 定義護(hù)欄詞表 sensitive_words [違規(guī)詞A, 違規(guī)詞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: 這是一段需要審核的正常內(nèi)容用于演示 Harness 工作流。}) 4. 輸出結(jié)果 print(f工作流狀態(tài): {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 工作流狀態(tài): success SensitiveWordFilter: success (0.2ms) LLMSummarizer: success (1.1ms) HumanApproval: success (0.3ms) 最終摘要: [模擬摘要] 原文共 28 字主題為示例內(nèi)容。 進(jìn)階條件分支與并行執(zhí)行 真實場景往往不是簡單的線性鏈路。下面擴(kuò)展 Workflow 支持條件分支 class ConditionalStep(Step): 根據(jù)條件決定執(zhí)行哪個子步驟。 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} 使用示例內(nèi)容長度超過閾值才走詳細(xì)審核 def is_long_text(ctx): return len(ctx.get(input_text, )) 50 wf Workflow(conditional_pipeline) wf.add_step(SensitiveWordFilter(SensitiveWordFilter, [違規(guī)詞A])) wf.add_step(ConditionalStep( RouteByLength, conditionis_long_text, if_stepLLMSummarizer(LLMSummarizer), else_stepHumanApproval(HumanApproval), )) 9. 可觀測性結(jié)構(gòu)化追蹤 生產(chǎn)環(huán)境必須能回溯每一步的執(zhí)行情況。在 Step.run 中已經(jīng)記錄了耗時和重試次數(shù)進(jìn)一步可以接入追蹤系統(tǒng) import json import datetime def export_trace(result: WorkflowResult) - str: 把工作流執(zhí)行結(jié)果導(dǎo)出為 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. 生產(chǎn)落地的關(guān)鍵考量 從 Demo 到生產(chǎn)Harness Engineering 還需要關(guān)注以下幾點 持久化工作流狀態(tài)要寫入數(shù)據(jù)庫支持中斷恢復(fù)和重新執(zhí)行。 冪等性每個 Step 要設(shè)計成可重復(fù)執(zhí)行且結(jié)果一致避免重試造成副作用。 超時控制外部 API 調(diào)用必須設(shè)置超時和熔斷防止鏈路阻塞。 審計日志涉及人工審批和敏感數(shù)據(jù)的步驟要記錄完整的操作軌跡。 版本管理工作流定義要納入版本控制支持灰度發(fā)布和快速回滾。 成本控制對模型調(diào)用等昂貴步驟做預(yù)算限制和用量統(tǒng)計。 11. 總結(jié) Harness Engineering 的本質(zhì)是把「能跑通的腳本」升級為「可治理的工程系統(tǒng)」。它通過 Step 抽象、工作流編排、護(hù)欄機制和可觀測性讓復(fù)雜的自動化鏈路變得穩(wěn)定、可控、可審計。本文從零實現(xiàn)了一個輕量級框架并演示了帶敏感詞過濾、模型調(diào)用和人工審批的完整工作流。生產(chǎn)環(huán)境中可以基于同樣的思想借助成熟的編排平臺或自研框架把 Harness Engineering 落地到實際業(yè)務(wù)中。 Harness Engineering 到底在做什么：從概念到代碼實戰(zhàn)
1. 引言Harness Engineering 是什么Harness Engineering工程化編排是近年來在 AI Agent、自動化流水線和復(fù)雜系統(tǒng)集成領(lǐng)域快速興起的一類工程實踐。它的核心目標(biāo)是把多個松散的組件——模型、工具、數(shù)據(jù)源、人工審批、外部服務(wù)——通過一套可編排、可觀測、可回滾的工程框架組織成穩(wěn)定、可控、可復(fù)用的自動化流程。簡單來說Harness Engineering 解決的是「如何把能力變成可靠的工程系統(tǒng)」的問題。它關(guān)注的不只是單個模型或單個工具的效果而是整條鏈路的穩(wěn)定性、可維護(hù)性和可治理性。2. 核心概念拆解要理解 Harness Engineering需要先厘清幾個關(guān)鍵概念Harness編排框架承載流程定義、狀態(tài)管理、錯誤處理和資源調(diào)度的運行容器。Step步驟流程中的最小執(zhí)行單元可以是調(diào)用模型、執(zhí)行代碼、查詢數(shù)據(jù)庫或觸發(fā)外部 API。Workflow工作流由多個 Step 按順序或條件組合而成的完整執(zhí)行鏈路。Guardrail護(hù)欄對輸入輸出進(jìn)行校驗、限流、審計和人工確認(rèn)的機制是 Harness 區(qū)別于普通腳本的關(guān)鍵。Observability可觀測性對每一步的輸入、輸出、耗時、成本和失敗原因進(jìn)行記錄與追蹤。3. Harness Engineering 與普通腳本的區(qū)別很多人會問這不就是寫腳本把幾個 API 串起來嗎區(qū)別在于工程化程度維度普通腳本Harness Engineering錯誤處理try-catch 散落各處統(tǒng)一的重試、降級、熔斷策略狀態(tài)管理全局變量顯式的工作流狀態(tài)機可觀測性print 日志結(jié)構(gòu)化追蹤、指標(biāo)采集、鏈路回溯人工介入難以實現(xiàn)內(nèi)置審批節(jié)點、暫停恢復(fù)復(fù)用性復(fù)制粘貼Step 組件化、版本化4. 代碼實戰(zhàn)構(gòu)建一個最小 Harness 框架下面我們用 Python 從零實現(xiàn)一個輕量級 Harness 框架包含 Step 抽象、工作流編排、重試機制和結(jié)構(gòu)化日志。先定義基礎(chǔ)組件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: 所有步驟的基類子類實現(xiàn) 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. 工作流引擎實現(xiàn) 有了 Step 基類接下來實現(xiàn) Workflow 引擎負(fù)責(zé)按順序執(zhí)行步驟、傳遞上下文、收集結(jié)果 dataclass class WorkflowResult: workflow_id: str status: StepStatus step_results: List[StepResult] field(default_factorylist) context: Dict[str, Any] field(default_factorydict) class Workflow: 按順序執(zhí)行一組 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: # 把輸出寫入共享上下文供后續(xù)步驟使用 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; 實戰(zhàn)示例構(gòu)建一個帶護(hù)欄的 AI 內(nèi)容審核工作流 下面用一個真實場景串聯(lián)整個框架對用戶提交的文本先做敏感詞過濾再調(diào)用大模型生成摘要最后經(jīng)過人工審批節(jié)點。先實現(xiàn)具體的 Step class SensitiveWordFilter(Step): 護(hù)欄步驟檢查輸入是否包含敏感詞。 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): 調(diào)用大模型生成摘要此處用模擬實現(xiàn)。 def execute(self, context: Dict[str, Any]) - Any: text context[input_text] 真實場景這里調(diào)用 OpenAI / Claude / 本地模型 API summary call_llm(f請總結(jié){text}) summary f[模擬摘要] 原文共 {len(text)} 字主題為示例內(nèi)容。 return {summary: summary} class HumanApproval(Step): 人工審批節(jié)點模擬等待人工確認(rèn)。 def execute(self, context: Dict[str, Any]) - Any: summary context[LLMSummarizer][summary] 真實場景這里會推送審批任務(wù)到 IM/郵件等待回調(diào) approved True # 模擬審批通過 if not approved: raise ValueError(人工審批未通過) return {approved: True, summary: summary}/code/pre 7. 組裝并運行工作流 def main(): 1. 定義護(hù)欄詞表 sensitive_words [違規(guī)詞A, 違規(guī)詞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: 這是一段需要審核的正常內(nèi)容用于演示 Harness 工作流。}) 4. 輸出結(jié)果 print(f工作流狀態(tài): {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 工作流狀態(tài): success SensitiveWordFilter: success (0.2ms) LLMSummarizer: success (1.1ms) HumanApproval: success (0.3ms) 最終摘要: [模擬摘要] 原文共 28 字主題為示例內(nèi)容。 進(jìn)階條件分支與并行執(zhí)行 真實場景往往不是簡單的線性鏈路。下面擴(kuò)展 Workflow 支持條件分支 class ConditionalStep(Step): 根據(jù)條件決定執(zhí)行哪個子步驟。 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} 使用示例內(nèi)容長度超過閾值才走詳細(xì)審核 def is_long_text(ctx): return len(ctx.get(input_text, )) 50 wf Workflow(conditional_pipeline) wf.add_step(SensitiveWordFilter(SensitiveWordFilter, [違規(guī)詞A])) wf.add_step(ConditionalStep( RouteByLength, conditionis_long_text, if_stepLLMSummarizer(LLMSummarizer), else_stepHumanApproval(HumanApproval), )) 9. 可觀測性結(jié)構(gòu)化追蹤 生產(chǎn)環(huán)境必須能回溯每一步的執(zhí)行情況。在 Step.run 中已經(jīng)記錄了耗時和重試次數(shù)進(jìn)一步可以接入追蹤系統(tǒng) import json import datetime def export_trace(result: WorkflowResult) - str: 把工作流執(zhí)行結(jié)果導(dǎo)出為 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. 生產(chǎn)落地的關(guān)鍵考量 從 Demo 到生產(chǎn)Harness Engineering 還需要關(guān)注以下幾點 持久化工作流狀態(tài)要寫入數(shù)據(jù)庫支持中斷恢復(fù)和重新執(zhí)行。 冪等性每個 Step 要設(shè)計成可重復(fù)執(zhí)行且結(jié)果一致避免重試造成副作用。 超時控制外部 API 調(diào)用必須設(shè)置超時和熔斷防止鏈路阻塞。 審計日志涉及人工審批和敏感數(shù)據(jù)的步驟要記錄完整的操作軌跡。 版本管理工作流定義要納入版本控制支持灰度發(fā)布和快速回滾。 成本控制對模型調(diào)用等昂貴步驟做預(yù)算限制和用量統(tǒng)計。 11. 總結(jié) Harness Engineering 的本質(zhì)是把「能跑通的腳本」升級為「可治理的工程系統(tǒng)」。它通過 Step 抽象、工作流編排、護(hù)欄機制和可觀測性讓復(fù)雜的自動化鏈路變得穩(wěn)定、可控、可審計。本文從零實現(xiàn)了一個輕量級框架并演示了帶敏感詞過濾、模型調(diào)用和人工審批的完整工作流。生產(chǎn)環(huán)境中可以基于同樣的思想借助成熟的編排平臺或自研框架把 Harness Engineering 落地到實際業(yè)務(wù)中。