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