企业官网建设流程全解析

1. Harness 不是新名词，而是 AI 工程里被长期忽视的“承重墙”

你打开 GitHub 搜索harness，大概率会看到一堆 CI/CD 流水线配置、Kubernetes 的 Helm Harness 插件，或者某个硬件测试框架——这恰恰是问题所在：Harness 在 AI 领域根本不是新概念，而是被强行“借名”、又被严重误读的工程实践代号。它既不是模型，也不是框架，更不是某种神秘中间件。我第一次在 DeepSeek 内部文档里看到harness这个词时，以为是某个新 SDK 的缩写，结果发现它只是指代“把 LLM 调用封装成可复用、可监控、可灰度的最小执行单元”的整套约定和脚手架。它不解决推理本身，但决定了你写的llm.invoke()是不是能扛住每秒 200 次并发、能不能自动 fallback 到备用模型、有没有记录完整 trace 供 debug——这些事，90% 的 Agent 教程里都跳过不讲，却恰恰是上线后最要命的部分。

关键词里反复出现的deepseek harness、harness engineering、harness anything，其实都在指向同一个事实：当 LLM 调用从 demo 阶段进入生产阶段，你就必须给它加一层“工程化外壳”，而这个外壳，业内就叫 harness。它和 Agent 的关系，不是父子，而是“钢筋”和“房子”；和 Skill 的关系，不是包含，而是“供电接口”和“电器”；和 LLM 的关系，更不是替代，而是“遥控器”和“电视”。很多人一上来就学 LangChain 的 AgentExecutor，却连底层LLM实例连超时重试都没配好，结果就是：本地跑通，一上压测就 timeout；单次调用正常，批量请求就返回空 JSON；日志里全是{"error": "model overloaded"}却找不到源头——这些都不是模型问题，是 harness 缺位导致的工程塌方。

我去年帮一家做金融问答的团队重构 Agent 架构，他们原来的代码里llm = OpenAI(model="gpt-4-turbo")直接 new 出来就用，没有熔断、没有降级、没有输入长度校验。结果客户一上传 50 页 PDF，LLM 就因 context overflow 返回乱码，前端直接崩溃。我们没动模型，只加了一层 harness：输入预检（截断+摘要）、输出 schema 强校验（用 Pydantic V2 做 JSON 结构约束）、失败自动 fallback 到 gpt-3.5-turbo + 重试指数退避。上线后错误率从 17% 降到 0.3%，而开发量只增加了 200 行代码。这就是 harness 的真实价值：它不让你的模型更聪明，但让你的模型更可靠、更可控、更像一个工业级服务。如果你正在查harness 和 agent 区别，答案很简单：Agent 是业务逻辑编排器，Harness 是 LLM 调用的“工业控制器”。

提示：不要被harness这个词迷惑。它不是 DeepSeek 独创，也不是某家公司的私有协议。它本质是 SRE（站点可靠性工程）思想在 LLM 调用层的落地——就像你不会裸写socket.connect()去发 HTTP 请求，而是用 Requests 库；同理，你不该裸写llm.invoke()，而该用 harness 封装后的llm_call()。

2. Harness 的四根支柱：为什么它必须独立于 Agent 和 Skill 存在

Harness 的核心价值，恰恰在于它的“低存在感”——它不该出现在你的业务代码里，而应像空气一样弥漫在所有 LLM 调用周围。要理解它为何不能被 Agent 或 Skill 吞并，必须拆解它的四个不可替代的工程支柱。这四根支柱，共同构成了 LLM 生产化落地的底线保障，缺一不可。

2.1 输入治理：不是简单的 prompt 拼接，而是“防爆安检”

绝大多数教程教你把用户 query 和 system prompt 拼成字符串丢给 LLM，这在 demo 阶段可行，但在生产环境等于埋雷。Harness 的第一道防线，就是输入治理。它包含三个硬性动作：

• 长度预检与动态截断：LLM 的 context 窗口是硬限制。Harness 必须在调用前计算len(system_prompt) + len(user_query) + len(retrieved_docs)，若超限，不能简单粗暴地 truncate，而要按语义重要性分层裁剪。比如金融问答中，监管条款原文优先级 > 用户历史对话 > 通用说明。我们实测过，用textwrap.shorten()粗暴截断，准确率下降 23%；而用基于 sentence-transformers 的相似度排序后保留 top-k 句子，准确率仅降 1.8%。

• 敏感内容过滤与脱敏：不是靠关键词黑名单（漏报率高），而是用轻量级 NER 模型（如 spaCy 的 en_core_web_sm）实时识别 PII（姓名、身份证号、银行卡号）。识别到后，不是直接拒绝，而是替换为占位符<PERSON_1>、<ID_NUMBER_2>，并在后续 prompt 中明确告知模型：“以下<PERSON_X>代表需保护的隐私实体，请勿生成其具体值”。这样既合规，又不影响推理逻辑。

• 格式标准化与 Schema 注入：很多团队卡在修复 llm 返回 json 的 java 库这个需求上，根源在于没在输入端就强制结构。Harness 会在拼接 prompt 时，自动注入严格的 output schema 指令，例如：

  请严格按以下 JSON Schema 输出，字段名、类型、必填项均不可更改： {"type": "object", "properties": {"answer": {"type": "string"}, "confidence": {"type": "number", "minimum": 0, "maximum": 1}}, "required": ["answer", "confidence"]}

  并配套使用json_repair库（Python）或json-schema-validator（Java）做输出后校验。这比在 Java 层写一堆 try-catch 解析异常高效得多。

注意：输入治理必须发生在 Agent 编排之前。如果 Agent 自己做截断，不同 Skill 可能重复处理；如果 Skill 自己做脱敏，同一份用户数据在多个 Skill 里被多次识别，性能浪费且规则不一致。Harness 是唯一能统一管控输入的地方。

2.2 调用控制：超时、重试、熔断，一个都不能少

LLM API 不是数据库，它是网络服务，有抖动、有排队、有不可预测的延迟。裸调用llm.invoke()的默认行为是：无超时（等死）、无重试（一次失败就崩）、无熔断（雪崩）。Harness 必须接管这三层控制：

• 分级超时策略：不是设一个全局 timeout。我们按场景分三级：

  • fast（如单轮问答）：总 timeout ≤ 8s，其中 LLM 推理 ≤ 5s，网络传输 ≤ 3s；
  • medium（如多步推理）：总 timeout ≤ 30s，允许 LLM 推理 ≤ 20s；
  • slow（如长文档摘要）：总 timeout ≤ 120s，但必须开启 streaming，每 5s push 一次 chunk。 这些阈值不是拍脑袋定的，而是基于历史 P95 延迟 + 20% buffer 计算得出。例如某模型 P95 延迟是 4.2s，则fast场景 timeout 设为 8s。

• 智能重试机制：不是简单 retry(3)。Harness 会根据 error code 分类重试：

  • 429 Too Many Requests：指数退避（1s, 2s, 4s）；
  • 503 Service Unavailable：立即重试（服务瞬时过载）；
  • 500 Internal Error：不重试，直接 fallback；
  • JSON decode error：不重试，触发 schema 修复流程。 关键是，重试时会自动修改seed参数（如果模型支持），避免重复返回相同错误结果。

• 熔断器（Circuit Breaker）：当连续 5 次调用失败率 > 60%，Harness 自动打开熔断器，后续请求直接返回 fallback 响应（如缓存答案、兜底话术），持续 30 秒后半开试探。这能防止一个模型故障拖垮整个 Agent 流程。我们曾在线上观察到，某天 Azure OpenAI 的 us-east 区域因网络抖动，失败率飙升至 92%，熔断器生效后，整体服务可用性保持在 99.95%，而未启用熔断的旧服务直接雪崩。

2.3 输出契约：让 LLM 的“胡言乱语”变成可验证的合同

LLM 的不确定性是双刃剑。Harness 不压制这种不确定性，而是把它框进可验证的契约里。这包括三重保障：

• Schema 强校验：如前所述，不是靠正则匹配"answer": ".*?"，而是用 JSON Schema 定义字段类型、范围、嵌套关系。Harness 调用后，用jsonschema.validate()校验，不通过则触发修复或 fallback。我们统计过，对{"answer": "string", "steps": ["string"]}这样的简单 schema，LLM 直接输出合规 JSON 的概率约 68%；加入 harness 的 schema 注入 + 修复后，提升至 99.2%。

• 语义一致性检查：Schema 只管格式，不管逻辑。Harness 会额外运行轻量级检查：

  • 如果 prompt 要求“用中文回答”，输出含英文单词 > 3 个则标记为不一致；
  • 如果要求“列出 3 点”，输出数组长度 ≠ 3 则标记；
  • 如果要求“比较 A 和 B”，输出中未同时出现 A 和 B 的关键词则标记。 这些规则用 spaCy 的 rule-based matcher 实现，耗时 < 5ms，却能拦截 12% 的“格式正确但逻辑错误”输出。

• 置信度锚定（Confidence Anchoring）：LLM 不会告诉你它有多不确定。Harness 会在 prompt 中强制要求输出confidence字段，并用以下方式校准：

  • 对于分类任务，confidence = softmax 输出的最大概率；
  • 对于生成任务，confidence = 1 - (perplexity / max_perplexity)，其中 perplexity 由小模型（如 distilbert-base-uncased）对生成文本打分。 这样，下游 Agent 就能根据 confidence 值决定是否需要 human-in-the-loop，而不是盲目信任 LLM。

2.4 可观测性：没有 trace 的 LLM 调用，等于盲人开车

Agent 的 debug 之所以痛苦，是因为你不知道哪一步 LLM 调用出了问题。Harness 必须提供开箱即用的可观测性，这是它区别于普通 wrapper 的关键：

• 全链路 Trace ID 注入：每个 harness 调用生成唯一 trace_id，并透传给 LLM（作为 system prompt 的一部分：“本次会话 trace_id: xxx”）。当 LLM 输出中包含 trace_id，就能反向关联到原始请求。我们曾用此定位到一个 bug：某 Skill 在处理长文本时，因截断逻辑错误，将 trace_id 截断，导致日志无法关联，排查耗时 8 小时。

• 结构化日志：不记录 raw prompt（太长且含敏感信息），而是记录：

  { "trace_id": "xxx", "model": "gpt-4-turbo", "input_tokens": 1240, "output_tokens": 321, "latency_ms": 4821, "status": "success", "confidence": 0.87, "fallback_used": false }

  这些字段直接接入 Prometheus + Grafana，可实时监控 token 效率、延迟分布、fallback 率。

• 采样式 Prompt 日志：为平衡安全与 debug 需求，Harness 默认不记 full prompt，但按 1% 概率采样记录，并自动脱敏（替换 PII 为<MASKED>）。线上出问题时，可快速检索采样日志，还原现场。

这四根支柱，共同定义了 harness 的边界：它不关心 Agent 怎么编排 Skill，也不规定 Skill 该实现什么业务逻辑，它只专注一件事——确保每一次 LLM 调用，都是一个符合工程标准的、可度量、可追溯、可恢复的服务调用。把它塞进 Agent 里，Agent 会臃肿失控；把它塞进 Skill 里，Skill 会职责混乱。它必须是独立的、基础设施级的 layer。

3. Harness 与 Agent：不是谁包含谁，而是“供电系统”与“智能家电”的协作关系

网上大量讨论harness 和 agent 区别，甚至有人画出层级图说 “Agent 包含 Harness”，这完全颠倒了因果。正确的理解是：Agent 是消费 harness 的客户端，Harness 是为 Agent 提供稳定电力的电网。这个比喻不是修辞，而是精确的技术映射。下面用一个真实电商客服 Agent 的例子，拆解它们如何协作。

3.1 Agent 的典型工作流：它只负责“决策”，不负责“执行细节”

假设用户问：“我上周买的 iPhone 15，屏幕碎了，能免费换吗？” 一个标准 Agent 的工作流如下：

1. 意图识别：调用 Skill A（intent_classifier），输入用户 query，输出{"intent": "warranty_claim", "confidence": 0.92}；
2. 信息抽取：调用 Skill B（entity_extractor），输入 query + 上步结果，输出{"product": "iPhone 15", "issue": "screen broken", "purchase_date": "2024-05-10"}；
3. 政策查询：调用 Skill C（policy_retriever），输入 product + issue，从知识库召回相关条款；
4. 决策生成：调用 Skill D（decision_generator），输入 policy + purchase_date，生成回复：“根据您的购买日期（2024-05-10），iPhone 15 屏幕碎裂不在免费保修范围内，但可享受 8 折维修服务……”。

注意：在这个流程里，Agent 只做两件事——决定下一步调哪个 Skill，以及把上一步的输出喂给下一步的输入。它不关心 Skill A 的模型是什么、超时设多少、返回的 JSON 是否合法。这些，全部交给 harness 处理。

3.2 Harness 如何在每个环节默默支撑：以 Skill A 为例

当 Agent 决定调用 Skill A（intent_classifier）时，它实际发出的不是llm.invoke(prompt)，而是harness.call(skill_name="intent_classifier", input_data=...)。此时 harness 承担全部执行细节：

• 输入准备：harness 拿到input_data（即用户 query），先做长度检查（query 长度 28 个 token，远低于 gpt-3.5-turbo 的 16k 上限，通过）；再做敏感过滤（未识别到 PII，通过）；最后注入 schema 指令：“请严格按 {intent: string, confidence: number} 输出 JSON”。

• 调用执行：harness 构造最终 prompt，设置 timeout=8s，启动熔断器（当前失败率 0.1%，闭合状态），发起调用。3.2s 后收到响应{"intent": "warranty_claim", "confidence": 0.92}。

• 输出校验：harness 用预设 schema 校验，字段齐全、类型正确；再运行语义检查（prompt 要求输出 intent，响应中有；要求 confidence 是 number，响应中是 0.92，通过）；置信度锚定显示 0.92 与模型 softmax 输出一致。

• 可观测性：harness 记录日志：{"skill": "intent_classifier", "latency_ms": 3200, "status": "success", "confidence": 0.92}，并上报 trace_id。

Agent 收到的，只是一个干净的、带 confidence 的 dict，它无需知道背后发生了什么。如果这一步 harness 检测到输出不合规（比如返回了{"intent": "warranty", "confidence": "high"}），它会自动触发修复（把 "high" 转为 0.8）或 fallback（调用备用小模型），然后仍返回标准 dict 给 Agent。Agent 的代码完全不用改。

3.3 当 harness 缺位时，Agent 的“崩溃现场”

我们曾接手一个故障频发的 Agent，它的代码类似这样：

# 错误示范：Agent 自己管理 LLM 调用 def classify_intent(query): prompt = f"识别用户意图：{query}。只输出 JSON，如 {{'intent': 'order_status', 'confidence': 0.85}}" response = openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) return json.loads(response.choices[0].message.content)

问题立刻暴露：

• 某天 OpenAI API 延迟飙升到 15s，Agent 线程全部卡死，HTTP 超时返回 504；
• 用户 query 含手机号，LLM 在输出中直接回显，触发 GDPR 报警；
• LLM 偶尔返回{"intent": "warranty_claim", "confidence": "very high"}，json.loads()报错，Agent 进程 crash；
• 日志只有ERROR: Expecting value: line 1 column 1 (char 0)，无法定位是哪个 query 导致。

这些问题，100% 是因为 Agent 试图自己扮演 harness 的角色，却缺乏工程化能力。修复方案不是重写 Agent，而是剥离所有openai.ChatCompletion.create()，替换成harness.call("intent_classifier", query)。一行代码替换，问题全消。

提示：判断一个项目是否需要独立 harness，就看它的 Agent 代码里有没有出现openai.、anthropic.、ollama.这样的直接调用。如果有，它就还没准备好上生产。

4. Harness 与 Skill：不是容器与插件，而是“标准化插座”与“可插拔电器”的物理接口

Skill（技能）这个词，在 AI 社区被过度浪漫化了。很多人以为 Skill 是某种高级抽象，能自动学习、自我进化。实际上，在工程视角下，Skill 就是一个封装了特定业务逻辑的函数，而 Harness 就是给这个函数提供标准电源和保险丝的插座。它们的关系，是物理世界里最朴素的接口契约。

4.1 Skill 的本质：一个有明确定义 I/O 的黑盒函数

一个合格的 Skill，必须满足三个硬性条件，否则它就不是 Skill，只是段杂乱代码：

• 清晰的输入契约（Input Contract）：Skill 必须声明它接受什么类型的输入。例如policy_retrieverSkill 的输入是：

  class PolicyInput(BaseModel): product: str # 必填，字符串 issue: str # 必填，字符串 country: str = "CN" # 可选，默认 CN

  Harness 在调用前，会用 Pydantic 验证输入是否符合此契约。如果传入{"product": 123, "issue": "screen broken"}，harness 直接拒绝，返回ValidationError，不浪费一次 LLM 调用。

• 稳定的输出契约（Output Contract）：Skill 必须承诺返回什么。例如decision_generator的输出是：

  class DecisionOutput(BaseModel): answer: str action_required: Literal["none", "contact_support", "schedule_repair"] estimated_cost: Optional[float] = None

  Harness 调用后，强制校验返回值是否符合此模型。不符合？触发修复或 fallback。

• 可声明的元数据（Metadata）：Skill 必须告诉 harness 它的“电气参数”：

  skill_metadata = { "name": "decision_generator", "model": "gpt-4-turbo", "timeout_ms": 30000, "max_input_tokens": 8000, "requires_rag": True # 是否需要检索增强 }

  Harness 根据这些元数据，自动配置调用参数、决定是否启用 RAG pipeline、设置超时。

这才是 Skill 的真相：它不是魔法，而是一个有说明书、有额定功率、有接口标准的工业零件。Harness 就是那个确保所有零件都能插进同一块电路板的标准化插座。

4.2 Harness 如何统一管理异构 Skill：从codex skill到仓颉skill

现实中的 Skill 来源五花八门：有的是调用 OpenAI API（codex skill），有的是调用本地 Llama 3（仓颉skill），有的是调用微调的小模型（impeccable skill），还有的是调用传统规则引擎（workbuddy skill）。Harness 的核心能力，就是抹平这些差异，让 Agent 无感调用。

我们以codex skill（调用 OpenAI）和仓颉skill（调用本地 Qwen2-7B）为例，看 harness 如何统一：

关键点在于：Agent 调用harness.call("codex_skill", input)和harness.call("cangjie_skill", input)，代码完全一样。Harness 根据 skill name 查找 metadata，自动适配底层差异。这解决了agent evals中最大的痛点——评估不同 Skill 时，要写 N 套适配代码。有了 harness，eval 脚本只需遍历 skill list，统一调用即可。

4.3 “skill原版无删减版百度”背后的工程真相

搜索热词里出现的skill原版无删减版百度，折射出一个普遍现象：很多团队把 Skill 当成黑盒下载包，直接集成，却忽略其工程兼容性。一个未经 harness 封装的 “原版 Skill”，往往意味着：

• 无超时控制：可能卡死数分钟；
• 无错误处理：API 失败直接抛异常，Agent 崩溃；
• 无可观测性：不知道它调用了几次、花了多久、成功率多少；
• 无安全防护：输入未过滤，输出未校验，风险敞口大。

所谓“无删减版”，其实是“无工程加固版”。真正的生产级 Skill，必须经过 harness 的“出厂检验”：注入超时、添加熔断、绑定 trace、校验 schema。这个过程不是删减功能，而是增加鲁棒性。我们有个客户，买了某厂商的 “智能合同审查 Skill”，号称“原版无删减”，结果上线三天，因未设超时，一次大文件解析导致整个 Agent 服务不可用。我们只加了一层 harness（150 行代码），问题解决。

Harness 不是 Skill 的敌人，而是 Skill 的“质量认证机构”。它让 Skill 从实验室玩具，变成可部署、可监控、可运维的工业品。

5. Harness 与 LLM：不是替代，而是“驯化者”与“被驯化者”的共生协议

LLM 是强大的，但也是野性的。它不遵循 REST 规范，不保证 SLA，不提供健康检查端点。Harness 的终极使命，就是与 LLM 达成一份“共生协议”，在不改变 LLM 本质的前提下，让它成为可信赖的基础设施组件。这不是技术傲慢，而是工程必然。

5.1 LLM 的三大“野性”，正是 harness 的三大驯化目标

这三大驯化，不是要消灭 LLM 的创造性，而是划定它的“活动边界”。就像给一匹骏马装上缰绳和鞍具，不是限制它奔跑，而是确保它能听从指令、安全抵达目的地。

5.2 harness engineering 的核心：用确定性对抗不确定性

harness engineering这个热词，精准概括了这项工作的本质——它不是 AI 研究，而是软件工程在 LLM 时代的延伸。它的技术栈，90% 是传统后端工程：

• 网络层：HTTP client 配置（连接池、keep-alive）、DNS 缓存、TLS 版本协商；
• 并发控制：线程池/协程池管理、信号量限流、背压（backpressure）处理；
• 存储层：fallback cache（Redis）、trace 存储（Elasticsearch）、采样日志（S3）；
• 监控层：Prometheus metrics（harness_call_duration_seconds,harness_fallback_total）、Grafana dashboard、告警规则（fallback 率 > 5% 时通知）。

唯一新增的，是针对 LLM 特性的适配逻辑：schema 注入、confidence 计算、语义检查。这些逻辑，全部封装在 harness 的preprocess()、postprocess()、validate()方法里，与传统工程代码无缝集成。

我们团队的 harness 工程师，一半时间在写 Pydantic 模型和 JSON Schema，一半时间在调优线程池大小和 Redis 连接数。这很无聊，但极其重要。因为 LLM 的“智能”再耀眼，也掩盖不了一个事实：它跑在 Linux 内核上，走的是 TCP/IP 协议栈，受制于 CPU 和内存——这些，才是 harness engineering 的主战场。

5.3 为什么llm 框架和llm 代理地址不能替代 harness

搜索热词里频繁出现llm框架、llm代理地址，反映出一种误解：以为换个框架（LangChain、LlamaIndex）或配个代理（如http://localhost:8000/v1），就能解决 LLM 生产化问题。这是危险的幻觉。

• LLM 框架（LangChain 等）：它解决的是“怎么编排”，不是“怎么可靠调用”。LangChain 的LLMChain依然裸调llm.invoke()，没有内置超时、没有熔断、没有 schema 校验。它是个乐高积木，harness 才是胶水和螺丝刀。

• LLM 代理地址（如 Ollama、vLLM）：它解决的是“怎么部署模型”，不是“怎么安全调用”。vLLM 提供了高性能推理，但它不关心你的 prompt 是否含敏感信息，不关心你的输出是否符合业务 schema，不关心你调用失败时要不要 fallback。它是个发动机，harness 才是变速箱和 ABS 系统。

真正可靠的 LLM 应用，必须是三层架构：

Agent（业务编排层） → Harness（工程保障层） → LLM Framework / LLM Proxy（模型执行层）

跳过 harness 这一层，就像造车只关注发动机和座椅，却忘了刹车和安全气囊。dify的sql查询内容太多导致llm返回不稳定这个问题，根源不是 Dify 框架不好，也不是 LLM 不行，而是 Dify 的 harness 层（它的llm_client）没有做好输入长度治理和 fallback 机制。

Harness 不是锦上添花，而是雪中送炭。它不让你的 LLM 更强大，但让你的 LLM 更值得托付。

6. 动手实现一个最小可行 harness：从零开始，200 行代码搞定

理论讲完，现在动手。下面是一个生产可用的最小可行 harness（Python），它覆盖了前文所述的四大支柱，代码 197 行，无外部依赖（除requests、pydantic、jsonschema），可直接集成到任何 Agent 项目中。这不是玩具，是我们线上服务的真实简化版。

# harness.py import time import json import logging import random import uuid from typing import Dict, Any, Optional, Callable, List from pydantic import BaseModel, ValidationError from jsonschema import validate, ValidationError as SchemaValidationError import requests # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class HarnessConfig: """harness 全局配置""" DEFAULT_TIMEOUT_MS = 8000 FALLBACK_THRESHOLD = 0.6 # 熔断失败率阈值 CIRCUIT_BREAKER_DURATION = 30 # 熔断持续秒数 class SkillMetadata(BaseModel): """Skill 元数据""" name: str model: str timeout_ms: int = HarnessConfig.DEFAULT_TIMEOUT_MS max_input_tokens: int = 4000 requires_rag: bool = False class HarnessCallResult(BaseModel): """harness 调用结果""" success: bool data: Optional[Dict[str, Any]] = None error: Optional[str] = None latency_ms: float confidence: Optional[float] = None fallback_used: bool = False class Harness: def __init__(self): self.circuit_breakers = {} # {skill_name: {"state": "closed/open/half-open", "failure_count": int, "last_opened": float}} def call(self, skill_name: str, input_data: Dict[str, Any], metadata: SkillMetadata, output_schema: Dict[str, Any]) -> HarnessCallResult: """统一调用入口""" trace_id = str(uuid.uuid4()) start_time = time.time() # 1. 输入治理 try: validated_input = self._validate_input(input_data, metadata) except ValidationError as e: return HarnessCallResult( success=False, error=f"Input validation failed: {e}", latency_ms=(time.time() - start_time) * 1000, fallback_used=False ) # 2. 熔断检查 if self._is_circuit_open(skill_name): logger.warning(f"Circuit breaker open for {skill_name}, using fallback") return self._fallback(skill_name, input_data, start_time, trace_id) # 3. 构造 prompt & 调用 try: prompt = self._build_prompt(validated_input, metadata, output_schema) response = self._llm_request(prompt, metadata, trace_id) # 4. 输出校验 result_data = self._validate_output(response, output_schema) # 5. 置信度锚定（简化版：从 response 中提取 confidence） confidence = self._extract_confidence(result_data) latency_ms = (time.time() - start_time) * 1000 logger.info(f"Success call to {skill_name}: {latency_ms:.0f}ms, confidence={confidence}") return HarnessCallResult( success=True, data=result_data, latency_ms=latency_ms, confidence=confidence, fallback_used=False ) except Exception as e: latency_ms = (time.time() - start_time) * 1000 logger.error(f"Call to {skill_name} failed: {e}, latency={latency_ms:.0f}ms") self._record_failure(skill_name) return self._fallback(skill_name, input_data, start_time, trace_id) def _validate_input(self, input_data: Dict[str, Any], metadata: SkillMetadata) -> Dict[str, Any]: # 这里可集成 PII 过滤、长度检查等 if len(str(input_data)) > metadata.max_input_tokens * 4: # 粗略字节估算 raise ValidationError(f"Input too long for {metadata.name}") return input_data def _build_prompt(self, input_data: Dict[str, Any], metadata: SkillMetadata, output_schema: Dict[str, Any]) -> str: # 注入 schema 指令 schema_str = json.dumps(output_schema, ensure_ascii=False) return f"""你是一个专业助手，请严格按以下 JSON Schema 输出，不要任何额外文字： {schema_str} 用户输入：{json.dumps(input_data, ensure_ascii=False)}""" def _llm_request(self, prompt: str, metadata: SkillMetadata, trace_id: str) -> str: # 模拟 LLM 调用，实际替换为 requests.post # 这里演示超时和重试逻辑 timeout = metadata.timeout_ms / 1000 for attempt in range(3): try: # 实