三亩地 三亩地SAN MU DI · CODE DIARY
交流投稿
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天，欢迎挑你感兴趣的翻一翻。

首页/学习日记/日记详情

Harness Engineering：超越模型调优，构建高可用AI应用系统工程实践

发布时间：2026/8/13 9:57:33 三亩地 编程学习日记
Harness Engineering：超越模型调优，构建高可用AI应用系统工程实践

1. 当模型不再是唯一焦点：重新审视AI编程的效能瓶颈

最近和几位在头部AI公司工作的朋友聊天，发现一个挺有意思的现象。大家聚在一起，话题不再是“哪个模型又刷新了SOTA榜单”，或者“我们内部又训了一个多少参数的怪兽”。相反，他们开始频繁地提到一个词：“Harness Engineering”。起初我以为是什么新的模型架构或者训练框架，深入了解后才发现，这完全不是一回事。它更像是一种工程哲学，或者说，是一套被OpenAI等顶尖团队内部实践，但外界讨论不多的“增效方法论”。

简单来说，Harness Engineering的核心思想是：别只盯着模型本身卷了，把更多精力放在如何高效、可靠、规模化地“驾驭”现有模型上。这听起来有点像“不要重新发明轮子”，但在AI开发领域，尤其是大模型应用开发中，有着更深层的含义。我们过去太习惯于一个线性思维：任务效果不好？那就换更大的模型、找更多的数据、调更久的参数。这当然有效，但成本和边际收益的曲线正在变得极其陡峭。Harness Engineering则提出，在模型能力给定的情况下，通过极致的工程化手段，去挖掘、组合、引导和保障模型能力的稳定输出，往往能以小博大，获得远超预期的效果提升和系统稳定性。

这不仅仅是Prompt Engineering的升级版。Prompt Engineering关注的是与单次模型交互的“输入艺术”，而Harness Engineering关注的是构建一个系统性的、自动化的、可观测的“模型驾驶舱”。它涉及从数据流编排、提示模板的版本化管理、多模型路由与降级、到输出结果的校验、归因与持续迭代的完整闭环。如果你还在为如何让GPT-4在你的业务场景中稳定发挥而头疼，或者觉得每次调用都像开盲盒，那么你可能已经遇到了Harness Engineering旨在解决的问题。

2. Harness Engineering的核心组件：构建模型的控制中枢

那么，一个典型的“Harness”系统由哪些部分构成呢？它不是一个具体的软件，而是一系列设计模式和工具链的组合。根据我在实际项目中的构建经验，可以将其拆解为以下几个关键层次。

2.1 智能编排与路由层：让合适的模型做合适的事

这是Harness系统最外显的一层。其核心思想是避免对单一模型的过度依赖，而是根据任务类型、成本、延迟要求和当前负载，智能地分配请求。

1. 基于语义的路由策略最简单的路由是基于任务描述的。例如，一个客服系统收到用户问题“帮我重置密码”。系统可以先用一个轻量、快速的分类模型（甚至是基于嵌入向量的规则）判断这是一个“账户操作”类问题。对于这类有明确流程、答案固定的任务，可能根本不需要动用GPT-4，而是路由到一个内部的、经过精调的、成本极低的小模型，或者直接调用知识库API。只有对于复杂的、开放性的咨询，才会路由到GPT-4或Claude。

实操配置示例： 我们设计了一个路由决策表，它不是一个硬编码的if-else，而是一个可动态配置的规则引擎。

# routing_rules.yaml rules: - name: "direct_knowledge_base_lookup" condition: "intent_classifier.confidence > 0.9 AND intent IN ['password_reset', 'account_unlock', 'business_hours']" action: type: "api_call" target: "internal_kb_api" params: query: "{{user_query}}" intent: "{{intent}}" fallback: "rule_complex_qa" # 如果内部API失败，降级到下一个规则 - name: "rule_complex_qa" condition: "query_complexity > 0.7 OR contains(user_query, ['opinion', 'compare', 'why'])" action: type: "llm_call" target: "gpt-4-turbo" params: system_prompt: "expert_assistant_v2" temperature: 0.3 cost_limit: 0.05 # 单次调用成本上限（美元） fallback: "rule_general_qa" - name: "rule_general_qa" condition: "default" # 兜底规则 action: type: "llm_call" target: "claude-3-haiku" # 使用成本更低的模型兜底 params: system_prompt: "helpful_assistant" temperature: 0.7

为什么这样设计？成本控制和性能保障。GPT-4处理简单任务是一种巨大的资源浪费。通过前置的分类和路由，我们能将80%的简单查询引流到低成本渠道，整体成本可能下降60%以上，同时因为轻量级路径延迟更低，用户体验的P95延迟反而得到优化。

2. 主动降级与熔断机制模型服务提供商可能会出现高延迟、高错误率或额度耗尽的情况。一个健壮的Harness系统必须包含熔断器模式。例如，连续5次调用某个模型端点超时，或1分钟内错误率超过10%，则自动熔断该路由，将流量切换到备份模型，并在30分钟后尝试恢复。这借鉴了微服务治理的思路，确保了整个AI服务链路的韧性。

2.2 提示词的生命周期管理：从“黑魔法”到可工程化的资产

Prompt是驾驭模型的“缰绳”，但大多数团队的Prompt管理还处于石器时代：散落在各个工程师的笔记本、代码注释或环境变量里。Harness Engineering将提示词视为一等公民的代码资产。

1. 版本化与A/B测试我们使用Git来管理提示词模板，每个模板都是一个独立的文件（如prompts/customer_service/expert_assistant_v2.jinja2）。这带来了几个好处：

  • 版本回溯：当新提示词上线导致效果下降时，可以立即回滚到上一个稳定版本。
  • 差异化测试：可以轻松地为同一任务创建多个变体（A/B/C版），并通过路由层将少量流量导向不同版本，收集效果指标（如任务完成率、用户满意度评分）后，再决定全量推哪个版本。

2. 参数化与上下文注入硬编码的提示词是脆弱的。优秀的提示词模板应该是参数化的，并能动态注入运行时上下文。

{# expert_assistant_v2.jinja2 #} 你是一名资深的{{ domain }}顾问。请根据以下用户背景和公司知识库信息，专业、友好地解答用户问题。 <用户背景> 会员等级: {{ user_tier }} 历史订单: {{ recent_orders|default([]) }} 最近咨询问题: {{ recent_queries|default([]) }} </用户背景> <知识库上下文> {% for doc in retrieved_documents %} [{{ loop.index }}] {{ doc.content }} {% endfor %} </知识库上下文> <用户问题> {{ user_query }} </用户问题> 请确保你的回答： 1. 引用相关上下文（如[1], [2]）。 2. 如果信息不足，请明确询问。 3. 使用与用户等级相匹配的措辞风格。

3. 提示词编译与优化在发送给LLM之前，Harness系统会对提示词进行“编译”。这包括：

  • 变量替换与转义：确保用户输入中的特殊字符不会破坏提示结构。
  • 长度优化：自动截断过长的上下文，优先保留与当前查询语义最相关的部分（通过嵌入向量相似度计算）。
  • 令牌数估算：在调用前预估本次请求的输入输出令牌数，对于可能超长的请求提前进行分块处理或选择支持更长上下文的模型。

2.3 输出规范化与校验层：给模型的输出加上“护栏”

模型的输出是不可控的，但业务应用要求可控。这一层负责在将模型输出交付给下游系统或用户之前，对其进行约束、验证和格式化。

1. 结构化输出强制通过提示词要求模型以JSON、XML或特定标记格式输出，这只是第一步。Harness系统会在收到输出后，运行一个轻量的解析和校验流程。

  • 语法校验：使用JSON Schema或Pydantic模型验证结构是否正确，字段类型是否匹配。
  • 内容校验：定义业务规则。例如，一个生成产品推荐的输出，必须包含“产品ID”、“推荐理由”字段，且“产品ID”必须在当前库存列表中。如果校验失败，则触发重试或降级逻辑。

2. 安全与合规过滤这是一个至关重要的环节。我们需要在输出层部署一套内容安全策略，这通常是一个多阶段的过滤管道：

  • 关键词过滤：维护一个动态更新的拒绝词列表，对输出进行快速扫描。
  • 敏感内容分类模型：使用一个专门训练的小型分类模型（如基于DeBERTa），判断输出是否包含仇恨言论、歧视性内容、不实信息等。这个模型可以本地部署，延迟极低。
  • 自定义规则引擎：针对特定业务，定义规则。例如，在金融场景，任何包含具体股票代码和“买入”、“暴涨”等组合的语句都需要被拦截并人工复核。

注意：这里的安全过滤是内容层面的业务安全，与网络访问无关。所有操作均在合规的业务逻辑层完成，确保生成内容符合平台规范和社会公序良俗。

3. 后处理与增强有时，模型的原始输出需要进一步加工才能使用。例如：

  • 引用溯源：如果输出引用了提供的上下文，系统可以自动提取引用的文档编号，并生成指向原始知识条目的超链接。
  • 语气调整：根据用户反馈或场景，自动为输出添加或调整表情符号、问候语等，使其更自然。
  • 多轮会话摘要：在会话结束时，自动调用一个摘要模型，生成本轮对话的要点记录，并存入CRM系统。

3. 可观测性与持续迭代：从“感觉”到“数据驱动”

没有度量，就没有改进。Harness Engineering强调对每一次模型交互进行全方位的埋点和分析，这远不止是记录成功失败那么简单。

3.1 核心监控指标大盘

我们为每个AI功能模块定义了一套核心指标，并建立实时仪表盘：

1. 性能指标

  • 延迟：P50， P95， P99延迟。区分“总延迟”（用户发起请求到收到响应）和“LLM延迟”（模型API的响应时间）。
  • 吞吐量：每秒处理的请求数（RPS）。
  • 错误率：按错误类型分类（网络超时、模型内部错误、输出解析失败、内容安全拦截等）。

2. 质量与成本指标

  • 任务成功率：由业务逻辑定义的“成功”，例如客服场景中，是否在3轮对话内解决了用户问题（需要定义判断逻辑）。
  • 用户满意度：通过埋点或事后调查收集。
  • 成本消耗：按模型、按API Key、按业务线进行详细的令牌消耗和费用统计。这是优化路由策略的直接依据。
  • 输出分布：监控输出长度的分布、被触发的降级规则频率、各路由路径的流量占比等。

3.2 构建反馈闭环：让系统自我进化

监控是为了行动。Harness系统需要能自动收集反馈并用于迭代。

1. 隐式反馈收集

  • 用户行为信号：用户是否很快关闭了回答？是否紧接着提出了意思相同的问题（暗示没理解）？是否复制了回答内容（暗示有价值）？这些行为可以被埋点捕获，作为模型回答质量的弱监督信号。
  • 人工审核队列：对于低置信度的输出（例如，内容安全模型评分处于临界值，或任务分类模糊），系统自动将其加入人工审核队列。审核员的打标结果（好/坏，及原因）会立即回流到训练数据集中。

2. 数据驱动的提示词与路由优化定期（如每周）分析指标数据。例如：

  • 发现路由到“模型A”的“任务类型B”成功率显著低于“模型C”。那么可以自动生成一个实验，调整路由规则，将“任务类型B”的部分流量切到“模型C”进行A/B测试。
  • 通过分析失败案例，发现某一类问题总是因为上下文信息不足而回答错误。那么可以优化检索环节，或者修改提示词模板，增加“当信息不足时，应如何引导用户”的明确指令。

这个“监控-分析-调整-验证”的闭环，使得整个AI系统从一个静态的“模型调用方”，转变为一个能够持续学习和适应的有机体。

4. 实战架构：从零搭建一个轻量级Harness系统

理论说了这么多，我们来点实际的。如何为一个具体的应用场景（比如一个智能客服助手）搭建一个最小可行（MVP）的Harness系统？我不会推荐某个庞大的商业平台，而是展示如何用开源组件快速拼装。

4.1 技术栈选型与架构图

我们的目标是构建一个高内聚、低耦合的系统。以下是一个参考技术栈：

  • 编排与路由层：使用FastAPI作为总入口网关。它轻量、异步性能好，易于集成。路由逻辑可以用Python代码实现，复杂的话可以集成Celery或Dramatiq作为任务队列进行异步编排。
  • 提示词管理：将提示词模板存储在Git仓库中。应用启动时或通过Webhook监听，从Git拉取最新模板。使用Jinja2作为模板渲染引擎。
  • 模型调用抽象：使用Litellm这个开源库。它完美契合Harness思想，用一个统一的接口调用 OpenAI、Anthropic、Azure、Cohere等几十种模型，内置了重试、轮询API Key、计算成本等功能，极大简化了代码。
  • 输出校验与后处理：使用Pydantic进行结构化数据验证。内容安全过滤可以使用Transformers库加载一个开源的文本分类模型（如unitary/toxic-bert）。
  • 向量检索（用于上下文注入）：使用ChromaDB或Qdrant这类轻量级向量数据库，存储知识库文档的嵌入向量。
  • 可观测性：使用Prometheus收集自定义指标（Python客户端prometheus-client），用Grafana展示。链路追踪可以用OpenTelemetry。日志统一输出到JSON格式，方便用ELK或Loki收集分析。
  • 配置与特性开关：使用Hydra或Dynaconf管理配置，便于不同环境切换。使用Unleash或自己实现一个简单的特性开关服务，用于控制A/B测试和灰度发布。

一个简化的数据流如下：用户请求 -> FastAPI网关 -> 请求分类器 -> 向量检索（如需）-> 从Git获取并渲染提示词模板 -> 通过Litellm调用路由决策后的模型 -> 输出经过Pydantic校验和安全过滤 -> 返回结果。同时，每一个环节的关键指标和日志都被发送到监控系统。

4.2 关键代码片段与避坑指南

1. 使用Litellm实现模型路由与降级

import litellm from litellm import completion import asyncio # 配置多个模型，并设置优先级和降级链 model_fallbacks = [ { "model": "gpt-4-turbo", # 主模型 "api_key": os.getenv("OPENAI_API_KEY"), "litellm_params": {"max_tokens": 1000}, "weight": 10 # 权重，用于负载均衡 }, { "model": "claude-3-haiku-20240307", # 降级模型1 "api_key": os.getenv("ANTHROPIC_API_KEY"), "litellm_params": {"max_tokens": 1000}, "weight": 5 }, { "model": "groq/llama3-70b-8192", # 降级模型2（通过Groq等高速API） "api_key": os.getenv("GROQ_API_KEY"), "litellm_params": {"max_tokens": 1000}, "weight": 3 } ] async def call_llm_with_fallback(messages, fallbacks=model_fallbacks): """ 使用带降级的模型调用 """ for model_config in fallbacks: try: response = await completion( model=model_config["model"], messages=messages, api_key=model_config.get("api_key"), **model_config.get("litellm_params", {}) ) # 记录成功使用的模型和成本 record_metrics(model_config["model"], response.usage) return response except Exception as e: print(f"Model {model_config['model']} failed: {e}") # 记录失败 record_failure(model_config["model"]) continue # 尝试下一个降级模型 raise Exception("All model fallbacks failed") # 在FastAPI路由中使用 @app.post("/chat") async def chat_endpoint(request: ChatRequest): # ... 业务逻辑：分类、检索、构建prompt ... final_messages = build_messages(user_query, context) try: response = await call_llm_with_fallback(final_messages) # 对response进行校验和后处理 validated_output = validate_and_sanitize(response.choices[0].message.content) return {"answer": validated_output} except Exception as e: # 即使所有模型都失败，也有最终兜底策略，如返回预设话术 return {"answer": "抱歉，服务暂时不可用，请稍后再试。"}

避坑点：litellm的重试机制是全局配置的，注意不要和自定义的降级循环冲突，导致无限重试。务必为整个调用链设置总超时时间（例如使用asyncio.wait_for），避免一个请求卡住整个线程。

2. 提示词模板的动态加载与渲染

import jinja2 import os import git class PromptManager: def __init__(self, repo_url, local_path="./prompt_repo"): self.local_path = local_path self.env = jinja2.Environment(loader=jinja2.FileSystemLoader(local_path)) # 克隆或拉取提示词仓库 self._sync_repo(repo_url) def _sync_repo(self, repo_url): if not os.path.exists(self.local_path): git.Repo.clone_from(repo_url, self.local_path) else: repo = git.Repo(self.local_path) origin = repo.remotes.origin origin.pull() def get_prompt(self, template_name, **kwargs): """渲染指定模板""" template = self.env.get_template(f"{template_name}.jinja2") # 可以在这里注入全局变量，如当前日期、公司名称等 context = { "current_date": datetime.now().strftime("%Y-%m-%d"), **kwargs } return template.render(**context) # 使用示例 pm = PromptManager("https://github.com/your-company/prompt-templates.git") prompt_text = pm.get_prompt( "customer_service/expert_assistant", user_query="如何办理退票？", user_tier="黄金会员", retrieved_documents=[...] )

避坑点：Jinja2模板中如果使用了未定义的变量会报错。务必确保传入的上下文字典包含模板所需的所有变量，或者使用|default()过滤器设置默认值。另外，频繁拉取Git仓库可能影响性能，可以考虑使用Webhook通知更新，或者在内存中缓存已编译的模板对象。

3. 输出结构化校验与安全过滤

from pydantic import BaseModel, Field, validator from transformers import pipeline # 1. 定义期望的输出结构 class ProductRecommendation(BaseModel): product_id: str = Field(..., description="推荐产品的唯一ID") product_name: str reason: str = Field(..., max_length=200, description="推荐理由") confidence: float = Field(ge=0.0, le=1.0) @validator('product_id') def validate_product_id(cls, v): # 假设有一个全局的可用产品ID列表 from product_service import valid_product_ids if v not in valid_product_ids: raise ValueError(f'Invalid product ID: {v}') return v # 2. 安全过滤管道 class SafetyFilter: def __init__(self): # 加载一个轻量级文本分类模型（首次运行会下载模型） self.classifier = pipeline( "text-classification", model="unitary/toxic-bert", device=-1 # -1表示CPU， 0表示GPU ) self.deny_words = ["敏感词1", "敏感词2"] # 可从数据库动态加载 def filter(self, text: str) -> dict: """返回过滤结果和是否安全""" # 快速关键词过滤 for word in self.deny_words: if word in text: return {"safe": False, "reason": f"包含拒绝词: {word}", "filtered_text": None} # 模型深度分类 try: result = self.classifier(text[:512]) # 只分类前512个token以控制延迟 # result 示例: [{'label': 'toxic', 'score': 0.98}] if result[0]['label'] in ['toxic', 'hate'] and result[0]['score'] > 0.8: return {"safe": False, "reason": f"内容安全评分过高: {result[0]}", "filtered_text": None} except Exception as e: # 模型过滤失败时，根据策略决定是放行还是拦截。保守策略是拦截并记录。 print(f"Safety model error: {e}") return {"safe": False, "reason": "安全过滤服务异常", "filtered_text": None} return {"safe": True, "reason": "passed", "filtered_text": text} # 在业务流程中整合 def process_llm_output(raw_text: str): # 步骤1: 安全过滤 safety_result = safety_filter.filter(raw_text) if not safety_result['safe']: log_unsafe_content(raw_text, safety_result['reason']) # 触发人工审核或返回默认安全回复 return get_default_safe_response() # 步骤2: 尝试解析为结构化数据（如果期望结构化输出） try: # 假设我们通过提示词让模型输出JSON，这里先提取JSON部分 json_str = extract_json_from_text(safety_result['filtered_text']) recommendation = ProductRecommendation.parse_raw(json_str) return recommendation.dict() except Exception as e: # 解析失败，可能是模型未按格式输出，触发重试或使用备用解析方案 log_parsing_failure(raw_text, e) # 方案A: 用更简单的规则提取信息 # 方案B: 用另一个LLM调用（指定更严格的输出格式）来修复这个输出 return fallback_extraction(safety_result['filtered_text'])

避坑点：安全过滤模型本身也有延迟和计算成本。对于高并发场景，需要将其部署为独立服务，并使用缓存（例如，对完全相同的输入文本缓存过滤结果）。Pydantic校验非常严格，对于模型这种非确定性输出，解析失败率可能不低，必须设计完善的降级和重试逻辑，而不是直接向用户报错。

5. 从项目到平台：Harness Engineering的规模化挑战

当你为单个应用成功搭建了Harness系统后，随着公司内AI应用越来越多，你会面临新的挑战：如何避免每个团队重复造轮子？如何统一技术标准、监控和成本核算？这时，就需要考虑将Harness的能力平台化。

5.1 内部AI能力中台的关键组件

一个内部的AI能力中台，至少需要提供以下服务：

  1. 统一的模型网关：所有应用都通过这个网关调用各种模型。网关负责认证鉴权、限流、负载均衡、全局的熔断降级和路由策略。
  2. 提示词市场/仓库：一个中心化的提示词库，支持版本管理、权限控制、一键部署和效果追踪。团队可以分享和复用优秀的提示词模板。
  3. 工作流编排引擎：提供可视化的拖拽界面，让非工程师也能组合“检索->分类->生成->校验”等AI原子能力，构建复杂的AI工作流。
  4. 实验管理与分析平台：方便地创建和管理A/B测试，对比不同提示词、不同模型、不同参数下的业务指标，用数据驱动决策。
  5. 成本与资源驾驶舱：为公司管理层提供清晰的视图，了解AI花费在哪些业务、哪些模型上，投资回报率如何，为预算和资源分配提供依据。

5.2 文化转变：从模型炼丹师到AI工程师

推行Harness Engineering最大的障碍往往不是技术，而是团队思维和文化。数据科学家和研究员可能更关注模型本身的创新，而软件工程师可能对AI的黑盒特性感到不安。平台化的过程，也是推动两种角色融合，催生“AI工程师”这一新角色的过程。

AI工程师需要既理解机器学习的基本原理和局限性，又具备扎实的软件工程能力，能够设计高可用、可扩展、可观测的系统来“封装”和“放大”模型的能力。他们的KPI不再是单纯的模型准确率，而是端到端的业务指标提升、系统可用性、成本和延迟优化。

在我经历过的项目中，最成功的转变始于一个跨职能的“AI赋能小组”。这个小组由一两名资深后端工程师、一名机器学习工程师和一名产品经理组成。他们首先用Harness Engineering的方法论，将一个核心业务的AI场景做深做透，做出令人信服的效率和效果提升。然后，将这个案例、这套工具链和最佳实践，像“特洛伊木马”一样推广到其他团队。当其他团队看到，不需要等待更强大的模型，仅仅通过更好的工程化就能获得显著收益时，变革的阻力就会小很多。

说到底，Harness Engineering是一种务实的工程思维。在AI能力日益成为基础要素的今天，它的价值不在于创造新的智能，而在于如何将已有的智能，稳定、高效、经济地转化为真实的用户价值和商业成果。这或许不如发表一篇顶会论文那样耀眼，但它正是当下绝大多数企业解锁AI潜能最需要补上的一课。

编程学习 学习日记 实战经验
← 返回列表