1. 什么是Harness Design：从AI编程的“野路子”到工程化控制台

你有没有过这种体验：对着一个大模型提需求，它确实能写出代码、生成文档、甚至画出流程图，但结果总像隔了一层毛玻璃——你隐约知道它在动，却说不清它到底干了什么、依据是什么、哪一步出了岔子、出了岔子又该怎么拉回来？你反复改prompt，加限制词，用各种技巧“哄”它，最后发现不是模型不听话，而是整个交互过程根本没被当成一个 可设计、可测量、可回滚的软件系统 来对待。这就是当前绝大多数AI编程实践的真实困境：把模型当成了万能黑箱，却忘了工程师真正的核心能力从来不是“调用API”，而是 构建环境、定义边界、建立反馈、保障稳定 。

Harness Design，直译是“挽具设计”，这个命名非常精准。想象一匹烈马（大模型），它力量惊人、反应敏捷，但若没有挽具（harness）约束方向、分配力量、缓冲冲击、连接车辕（你的业务逻辑），再好的马也拉不动一辆车，甚至可能把车掀翻。Harness Design正是这样一套思维范式和工程方法论：它不关心你用哪个模型、prompt怎么写，它只关心—— 如何把这匹马稳稳地套进你的马车里，让它既发挥全部马力，又严格按你的路线、节奏和安全规范行驶 。这不是在教你怎么“驯服AI”，而是在教你如何像设计一个分布式服务一样，去设计一个由AI驱动的、具备状态管理、错误恢复、权限控制和效果评估能力的完整执行系统。

我把它拆解成七个相互咬合的齿轮层，每一层都对应一个明确的工程职责，而不是抽象概念。任务层是方向盘，决定往哪儿开；上下文层是后视镜和导航屏，决定模型能看到什么；工具层是油门和刹车，决定它能做什么；执行层是变速箱，决定它怎么换挡、加速、减速；状态层是行车记录仪和油量表，记录所有中间状态；安全层是ABS防抱死和安全气囊，防止失控；评估层则是仪表盘上的故障灯和里程表，告诉你这趟行程是否真的成功。这七层不是理论模型，而是你写在代码里的七个模块、七组配置文件、七个必须通过单元测试的接口契约。OpenAI在2026年那份内部工程师手册里说得非常直白：“未来三年，写一行Python代码的时间，将远少于设计一个能让AI稳定跑完三轮迭代的Harness所花的时间。” 这话听着刺耳，但当你第一次看着AI在无人干预下，自动发现代码bug、调用本地linter修复、生成diff报告、回滚到上一个稳定版本，并最终提交一份带完整测试覆盖率的PR时，你就明白了——那不是AI在工作，是你设计的Harness在工作。

这个思路天然适配你手头那个“抽象名词定义精炼软件”的项目。为什么？因为它的核心挑战根本不是“让AI写一段关于自由的定义”，而是“如何让AI在一个长期、多轮、目标模糊、证据分散的知识沉淀过程中，始终聚焦于‘概念演化’这一主线，不偏离、不遗忘、不覆盖、不丢失任何思考痕迹”。普通聊天界面里，用户昨天聊“自由”的代价，今天聊“责任”的边界，AI早就把上下文冲得七零八落；而在Harness里，“自由”是一张有ID、有状态、有版本历史的实体卡片，每一次对话的输入、AI的推理过程、提取的关键点、提出的开放问题、生成的候选定义，都会被强制写入这张卡片的状态快照里。模型看不到“整个对话历史”，它只看到你给它的、经过精心裁剪的、与当前子任务强相关的上下文片段——比如，此刻它只被允许看到“自由”卡片的最新定义版本、上一轮的open_questions、以及本次用户新输入的一段直觉描述。这种“上下文隔离”不是为了限制AI，而是为了 消除噪声、聚焦信号、保障可追溯性 。所以，Harness Design对你而言，不是一项可选的高级技巧，而是你这个项目的 唯一可行路径 。没有它，你做的只是一个会说话的记事本；有了它，你才真正开始构建一个能自我演化的个人知识操作系统。

2. Harness七层架构深度拆解：为什么是这七层，而不是别的？

Harness的七层设计，绝非拍脑袋的数字游戏。它是我过去三年在十几个不同领域（从金融风控规则引擎到工业设备故障诊断助手）落地AI Agent系统后，反复踩坑、推倒重来、再抽象提炼出的最小完备集合。每一层都解决一个不可绕过的、具体的工程痛点，缺一不可。下面我逐层拆解其设计原理、技术实现逻辑，以及为什么其他常见的“优化点”（比如换更强的模型、调更高温度）在这一层面前都是次要的。

2.1 任务层：定义“做什么”比“怎么做”重要一百倍

这是整个Harness的基石，也是最容易被忽视的一层。很多人一上来就想“让AI帮我写个爬虫”，这太模糊了。任务层要回答的是： 这个“爬虫”具体要爬什么？爬到的数据格式是什么？失败的判定标准是什么？成功的验收清单有几条？ 在你的“定义工作台”项目中，一个合格的任务契约（Task Contract）绝不能是“分析用户关于‘自由’的输入”，而必须是：

• 输入契约（Input Contract） ：一个JSON Schema，明确规定输入必须包含 concept_name: string 、 user_input: string 、 current_definition_version_id: string (optional) 、 context_summary: string (optional) 。任何不符合此Schema的输入，在进入Harness前就被拒绝。
• 输出契约（Output Contract） ：同样是一个严格的JSON Schema，规定输出必须是 { "analysis": { "candidate_definition": "...", "key_points": [...], "open_questions": [...], "confidence": 0.0-1.0 }, "next_steps": ["REVIEW_BY_USER", "GENERATE_COUNTEREXAMPLE", "REFINE_DEFINITION"] } 。模型可以“思考”，但最终交付物必须100%符合此结构。
• 验收标准（Acceptance Criteria） ：这不再是主观判断。例如，“ confidence 值必须≥0.75才能触发 REFINE_DEFINITION 步骤”；“ open_questions 数组长度必须≥2且每个问题必须包含至少一个疑问词（什么、如何、是否、为何）”；“ candidate_definition 字符串长度必须在80-150字符之间”。这些标准会被写成自动化校验函数，在Harness的评估层直接执行。

为什么必须如此苛刻？因为模型的“幻觉”本质是概率分布的采样偏差。你无法100%阻止它胡说，但你可以100%阻止它胡说的结果被下游系统接受。任务层就是那道闸门。我曾在一个医疗问答项目里吃过亏：初期只要求AI“回答患者问题”，结果它把“如何缓解头痛”答成了“推荐三种止痛药及其化学式”，完全忽略了患者问的是“孕妇能否服用”。后来我们把任务契约收紧为“仅基于《孕期用药指南V3.2》第4章内容作答，答案必须包含‘适用/禁用/慎用’三级分类及依据条款号”，错误率瞬间从37%降到1.2%。这证明， 清晰、可验证的任务定义，是Harness最廉价、最有效的“安全阀” 。

2.2 上下文层：给AI做“信息减法”，而非“信息堆砌”

这是Harness区别于普通Chat UI的核心标志。普通聊天中，你把所有历史消息一股脑塞给模型，指望它自己“理解上下文”。这就像让一个刚入职的实习生，不给他岗位说明书、不告诉他KPI、不给他客户档案，只丢给他过去三个月的所有会议录音，然后说“你去搞定这个客户”。Harness的上下文层，是 主动的信息裁剪与结构化供给 。

在你的项目中，模型每一轮看到的上下文，绝不是“从第一轮开始的所有消息”，而是由Harness根据当前状态机（State Machine）动态生成的、高度定制化的数据包。例如：

• 当状态是 AWAITING_USER_REVIEW 时，上下文只包含： [SYSTEM_PROMPT] + [CURRENT_CONCEPT_CARD_SNAPSHOT] + [LATEST_AI_ANALYSIS] + [USER_INSTRUCTIONS_FOR_REVIEW] ；
• 当状态是 GENERATING_COUNTEREXAMPLE 时，上下文则变成： [SYSTEM_PROMPT] + [CURRENT_CONCEPT_CARD_SNAPSHOT] + [LATEST_AI_ANALYSIS.open_questions[0]] + [COUNTEREXAMPLE_GENERATION_GUIDELINES] 。

这个“快照”（Snapshot）本身就是一个关键产物。它不是简单的文本拼接，而是从数据库里实时查询、序列化、并注入必要元数据（如 snapshot_id , generated_at , source_state ）后的结构化对象。我坚持用文件（如 context_snapshot_20240501_1423.json ）来存储它，原因有三：第一，它是可审计的，任何一次AI的输出都能精确追溯到它当时“看到”的全部信息；第二，它是可复现的，调试时直接加载这个文件就能100%复现当时的推理环境；第三，它是可版本化的，你可以对比 snapshot_v1 和 snapshot_v2 ，看Harness是如何逐步收紧上下文、引导AI聚焦的。这彻底终结了“为什么上次它答对了，这次就错了”的玄学调试。上下文层的设计哲学是： 模型的注意力是稀缺资源，我们必须像分配CPU时间片一样，精准、吝啬、可审计地分配给它 。

2.3 工具层：把“能力”变成“API”，把“权限”变成“白名单”

工具层是Harness的“手脚”。它回答的问题是： 模型能调用哪些外部能力？调用的边界在哪里？谁来审批越界操作？ 这里有一个关键认知颠覆：工具（Tools）不是为了让AI“更聪明”，而是为了让它“更可控”。MiniMax API支持的 tool_calls 机制，恰恰是实现这一层的完美载体。

你目前的 read_text_file 和 emit_concept_analysis 两个工具，已经体现了精髓。但真正的工具层设计，远不止于此。它需要一套完整的“工具治理”策略：

• 工具注册中心（Tool Registry） ：一个中央配置文件（如 tools.yaml ），定义所有可用工具的 name 、 description 、 parameters_schema 、 execution_timeout 、 max_retries 、 is_dangerous （标记为 true 的工具，如 delete_file ，必须走人工审批流）。
• 工具调用契约（Tool Call Contract） ：每次模型返回 tool_calls ，Harness不直接执行，而是先进行三重校验：1） name 是否在白名单内；2） arguments 是否符合 parameters_schema （用 jsonschema.validate ）；3） is_dangerous 为 true 的工具，是否已获得 approval_token （来自用户确认或预设规则）。只有全部通过，才进入执行环节。
• 工具执行沙箱（Sandboxed Execution） ：所有工具函数必须在独立的、资源受限的进程中运行（如用 subprocess.run 配合 timeout 和 memory_limit 参数）。 read_text_file 看似简单，但如果 path 参数被恶意构造为 ../../../etc/passwd ，没有沙箱就会酿成大祸。我见过最惨的案例，是某团队的 execute_shell_command 工具没加沙箱，AI在调试时“顺手”执行了 rm -rf / 。

在你的项目里，这六项核心工具（你提到的）应该包括： read_concept_card （读取概念卡片快照）、 write_concept_card （更新卡片状态）、 generate_diff （对比两个定义版本）、 run_linter （检查定义文本的逻辑一致性）、 fetch_related_concepts （查询概念网络）、 notify_user （向用户发送待办事项）。它们共同构成了AI的“能力边界”，而这个边界，是由你，而不是由模型的随机采样，来牢牢掌控的。

2.4 执行层：用状态机（State Machine）代替“自由发挥”

如果说任务层是方向盘，执行层就是变速箱和离合器。它定义了AI工作的 节奏、顺序和容错机制 。你提到的“先别接模型，先把状态流转和持久化打通”，这是绝对正确的起步顺序。一个健壮的执行层，其核心就是一个明确定义的、可持久化的状态机。

以你的“定义工作台”为例，一个最小可行的状态图（State Diagram）应包含：

• INITIAL ：接收用户原始输入，生成初始分析。
• ANALYSIS_READY ：AI已返回结构化分析，等待用户审核。
• USER_REVIEWED ：用户确认或修改了分析结果。
• DEFINITION_REFINED ：基于用户反馈，生成新定义版本。
• VERSION_COMMITTED ：新定义被正式提交，成为当前版本。
• ERROR_RECOVERING ：任何环节出错（如工具调用超时、校验失败），进入此状态，执行预设恢复策略（如重试、降级到备用模型、通知人工）。

关键在于“持久化”。每个状态转换，都必须伴随一次原子性的数据库写入（如 UPDATE concept_cards SET state = 'ANALYSIS_READY', last_updated = NOW() WHERE id = ? ）。这意味着，即使你的Python进程崩溃、服务器断电，只要数据库还在，Harness就能在重启后，精确地从它崩溃前的最后一刻状态继续执行。我曾在一个处理千万级日志的项目中，将状态机持久化到PostgreSQL的 pg_notify 通道，实现了毫秒级的状态同步和故障自愈。执行层的终极目标，是让AI的每一次“思考”，都变成一次可追踪、可中断、可恢复的数据库事务。

2.5 状态层：中间产物即资产，不是临时缓存

这是Harness的“记忆中枢”。它回答的问题是： AI在执行过程中产生的所有中间产物，如何被安全、结构化、可追溯地保存？ 普通做法是把一切存在内存里或临时文件中，这在单次请求中没问题，但在一个需要跨天、跨周迭代的“定义工作台”里，是灾难性的。

状态层的设计原则是： 一切皆为工件（Artifact） 。每一个中间产物，都必须有唯一的ID、创建时间、来源状态、内容哈希（用于防篡改）、以及明确的生命周期策略（如 analysis_artifact 保留30天， diff_report 永久存档）。在你的项目中，这些工件包括：

• concept_analysis_<id>.json ：AI每次分析的原始输出。
• definition_diff_<id>.md ：两个定义版本的对比报告。
• user_review_<id>.json ：用户对某次分析的修改意见和确认签名。
• state_snapshot_<id>.json ：执行层状态机在某一时刻的完整快照。

我强烈建议你使用一个专门的对象存储（如MinIO，一个开源的S3兼容服务）来存放这些工件，而不是放在应用服务器的硬盘上。原因很简单：对象存储天生具备高可用、高持久、可版本化、可加密的特性。一个 definition_diff 工件，不仅包含文本，还应包含其生成所依赖的两个 concept_analysis 工件的ID引用。这样，当你未来想回溯“为什么这个定义被修改了”，只需顺着ID引用链，就能拿到完整的、未经篡改的证据链。状态层不是为了“记住”，而是为了“证明”——证明每一次演化，都有据可查。

2.6 安全层：把“信任”变成“可验证的规则”

安全层是Harness的“免疫系统”。它不假设AI是善意的，也不假设用户是无害的，而是通过一系列硬性规则，将风险扼杀在摇篮里。你提到“安全层和评估层当下我并没有什么想法”，这恰恰是最危险的信号。安全不是锦上添花，而是生命线。

在你的项目中，安全层必须包含以下硬性规则：

• 数据隔离规则 ： read_concept_card 工具只能读取当前用户ID下的卡片； write_concept_card 工具的写入操作，必须通过 WHERE user_id = ? AND id = ? 的双重校验，杜绝越权访问。
• 敏感操作审批流 ：任何涉及删除（ delete_concept_card ）、批量修改（ bulk_update_definitions ）的操作，必须触发一个 approval_workflow 。该工作流会生成一个带时效性的URL，发送给用户邮箱，用户点击确认后，Harness才执行后续动作。这个URL的token必须是HMAC-SHA256签名的，包含 user_id , action , timestamp , nonce ，且 timestamp 有效期不超过10分钟。
• 输出内容过滤器（Content Filter） ：在AI的最终输出（ content 字段）被返回给前端之前，必须经过一个本地部署的、轻量级的内容过滤器（如基于规则的关键词匹配+小模型微调的二分类器），对 confidence 低于阈值的 open_questions 进行二次审查，确保其中不包含诱导性、违法或有害的提问。这层过滤是最后一道防线，它不依赖模型自身的“道德感”，而是依赖你部署的、可审计的、可更新的规则集。

安全层的设计哲学是： 永远不要把安全寄托于模型的“好意”，而要把它固化为代码中的 if/else 和数据库里的 CHECK CONSTRAINT 。我曾在一个教育项目中，因为没加 output_filter ，AI在生成“如何理解量子纠缠”时，顺带编造了一个不存在的、名字很像的“伪科学”理论，导致家长投诉。那次教训让我明白，安全层不是成本，而是你产品的信用背书。

2.7 评估层：用数据说话，告别“感觉良好”

这是Harness的“仪表盘”。它回答的问题是： 我们怎么知道这个Harness真的有效？是靠开发者的感觉，还是靠客观数据？ 你提到“等到接下来的实践中再逐步明确定义”，这没错，但评估层的指标，必须在MVP的第一天就埋点。

评估层需要两类指标：

• 过程指标（Process Metrics） ：衡量Harness自身运行的健康度。例如： task_success_rate （任务完成率）、 average_state_transition_time （平均状态转换耗时）、 tool_call_failure_rate （工具调用失败率）、 recovery_success_rate （错误恢复成功率）。这些指标应该被实时写入Prometheus，用Grafana看板监控。
• 结果指标（Outcome Metrics） ：衡量Harness产出的价值。这才是最关键的。在你的项目中，这包括： definition_stability_score （定义稳定性得分，计算方式： 1 - (number_of_edits_in_last_7_days / total_number_of_edits) ）、 user_review_time_avg （用户平均审核耗时）、 open_questions_resolution_rate （开放问题解决率，即被后续分析解答的问题占比）。这些指标必须和用户行为日志（如 click_event 、 review_submit ）关联起来，形成归因分析。

评估层的终极形态，是一个自动化的A/B测试框架。例如，你可以同时部署两个Harness变体：Variant A使用MiniMax-M2.7，Variant B使用GLM-5，然后将50%的流量随机分发给它们，持续一周。评估层会自动对比两者的 definition_stability_score 和 user_review_time_avg ，并给出统计显著性报告（p-value < 0.05）。这让你的选型决策，从“我觉得M2.7更好”，变成“数据证明M2.7在定义稳定性上高出12.3%，置信度99.7%”。评估层不是为了写报告，而是为了让你每一次的优化，都踩在坚实的数据基石上。

3. 从零搭建Harness MVP：我的九步实操路线图与避坑指南

现在，让我们把前面所有的理论，落到你键盘上敲出的第一行代码。你提供的“第0步到第9步”是一个极佳的起点，但我将它细化为一份可立即执行、且附带血泪教训的实操路线图。这不是一个理想化的教学大纲，而是我在凌晨三点调试一个卡死在 ERROR_RECOVERING 状态的Harness时，用咖啡和笔记写下的真实经验。

3.1 第0步：定义边界——用“三不原则”划清红线

这是整个Harness的宪法，必须用最朴素的语言，写在 README.md 的最顶部。我称之为“三不原则”：

• 不做 ：不处理任何与“抽象名词定义”无关的任务。如果用户输入“帮我写个Python爬虫”，Harness必须返回一个硬编码的、友好的错误响应：“抱歉，我专注于帮助您精炼对‘自由’、‘责任’等抽象概念的理解。请提供一个概念名称和您的初步思考。”
• 不猜 ：绝不基于不完整信息进行推测。如果 read_concept_card 工具返回空，Harness绝不尝试用默认值填充，而是立即转入 ERROR_RECOVERING 状态，并记录错误日志 "Failed to load concept card for ID: <id>, reason: not_found" 。
• 不越权 ：不执行任何未在 tools.yaml 白名单中明确定义的操作。 tool_calls 中出现 name: "execute_arbitrary_code" ，Harness必须将其视为非法输入，拒绝执行并告警。

提示：这一步的代码实现，就是在你的主入口函数（如 app.py ）里，加一个全局的 validate_task_input() 函数。它接收原始HTTP请求，首先检查 request.json.get("task") 是否为 "concept_analysis" ，再检查 request.json.get("concept_name") 是否为非空字符串。任何不满足，立刻 return JSONResponse({"error": "Invalid task boundary"}, status_code=400) 。 边界不是写在文档里，而是写在第一行防御性代码里。

3.2 第1步：做任务Contract——用JSON Schema生成器自动生成校验代码

手动写JSON Schema校验代码既枯燥又易错。我的做法是：用 pydantic 库，把契约定义成Python类，然后让 pydantic 自动生成校验逻辑和OpenAPI文档。

# schemas/task_contract.py
from pydantic import BaseModel, Field, validator
from typing import List, Optional

class ConceptAnalysisInput(BaseModel):
    concept_name: str = Field(..., min_length=1, max_length=50, description="概念名称，如'自由'")
    user_input: str = Field(..., min_length=10, max_length=2000, description="用户的原始思考素材")
    current_definition_version_id: Optional[str] = Field(None, description="当前定义版本ID，用于上下文追溯")
    
    @validator('user_input')
    def user_input_must_contain_thought(cls, v):
        # 简单启发式：必须包含中文标点或特定词汇
        if not any(c in v for c in ['。', '？', '！', '，', '：', '“', '”']):
            raise ValueError('user_input must contain Chinese punctuation or thought markers')
        return v

class ConceptAnalysisOutput(BaseModel):
    analysis: dict = Field(..., description="结构化分析结果")
    next_steps: List[str] = Field(..., min_items=1, max_items=3, description="下一步行动列表")

# 在main.py中，直接使用
@app.post("/analyze")
async def analyze_concept(request: Request):
    try:
        data = await request.json()
        input_obj = ConceptAnalysisInput(**data)  # 自动校验！
        # ... 后续逻辑
    except ValidationError as e:
        return JSONResponse({"error": "Validation failed", "details": e.errors()}, status_code=422)

实操心得：我曾经在第1步花了整整两天，试图用正则表达式手动校验 user_input 。直到我发现 pydantic 的 @validator 装饰器可以轻松集成任意复杂逻辑（比如调用一个本地NLP模型检查语义连贯性），我才意识到， 用对的工具，能把三天的工作压缩成三小时 。别重复造轮子，尤其是校验这种有成熟方案的轮子。

3.3 第2步：做状态机——用 transitions 库实现可持久化的状态流转

状态机是Harness的骨架。我选择 transitions 库，因为它轻量（<100KB）、文档极好、且原生支持状态持久化。

# state_machine.py
from transitions import Machine
import json
import sqlite3

class ConceptCardStateMachine:
    states = ['INITIAL', 'ANALYSIS_READY', 'USER_REVIEWED', 'DEFINITION_REFINED', 'VERSION_COMMITTED', 'ERROR_RECOVERING']
    
    def __init__(self, card_id: str):
        self.card_id = card_id
        self.machine = Machine(model=self, states=ConceptCardStateMachine.states, initial='INITIAL')
        # 定义状态转换
        self.machine.add_transition('start_analysis', 'INITIAL', 'ANALYSIS_READY')
        self.machine.add_transition('user_approves', 'ANALYSIS_READY', 'USER_REVIEWED')
        self.machine.add_transition('user_rejects', 'ANALYSIS_READY', 'INITIAL') # 回退到初始，重新分析
        self.machine.add_transition('refine_definition', 'USER_REVIEWED', 'DEFINITION_REFINED')
        self.machine.add_transition('commit_version', 'DEFINITION_REFINED', 'VERSION_COMMITTED')
        self.machine.add_transition('recover', '*', 'INITIAL') # 从任何状态都可恢复到初始
    
    def save_state(self):
        """将当前状态持久化到SQLite"""
        conn = sqlite3.connect('harness.db')
        cursor = conn.cursor()
        cursor.execute(
            "INSERT OR REPLACE INTO card_states (card_id, state, updated_at) VALUES (?, ?, datetime('now'))",
            (self.card_id, self.state, )
        )
        conn.commit()
        conn.close()
    
    def load_state(self):
        """从SQLite加载状态"""
        conn = sqlite3.connect('harness.db')
        cursor = conn.cursor()
        cursor.execute("SELECT state FROM card_states WHERE card_id = ?", (self.card_id,))
        row = cursor.fetchone()
        if row:
            self.state = row[0]
        conn.close()

# 初始化时，从DB加载状态
sm = ConceptCardStateMachine("free_001")
sm.load_state()
print(f"Loaded state: {sm.state}") # 输出可能是 'ANALYSIS_READY'

注意： transitions 库的 Machine 对象本身不保存状态，它只是状态转换的“蓝图”。你必须自己实现 save_state 和 load_state 。我见过太多人以为 machine.set_state('ANALYSIS_READY') 就万事大吉，结果进程一重启，状态就丢了。 状态机的“状态”二字，其灵魂就在于“持久化” 。

3.4 第3步：做工具白名单——用YAML配置+运行时校验双保险

工具白名单不是静态列表，而是一个动态的、可热更新的配置中心。

# config/tools.yaml
tools:
  - name: "read_concept_card"
    description: "读取指定ID的概念卡片快照"
    parameters_schema:
      type: "object"
      properties:
        card_id:
          type: "string"
          minLength: 5
      required: ["card_id"]
    execution_timeout: 5
    is_dangerous: false

  - name: "emit_concept_analysis"
    description: "返回结构化概念分析"
    parameters_schema:
      type: "object"
      properties:
        concept: {type: "string"}
        candidate_definition: {type: "string"}
        key_points: {type: "array", items: {type: "string"}}
        open_questions: {type: "array", items: {type: "string"}}
        confidence: {type: "number", minimum: 0, maximum: 1}
      required: ["concept", "candidate_definition", "key_points", "open_questions", "confidence"]
    execution_timeout: 10
    is_dangerous: false

# tools/registry.py
import yaml
import jsonschema
from jsonschema import validate

class ToolRegistry:
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
        self.tools = {tool['name']: tool for tool in self.config['tools']}
    
    def validate_tool_call(self, tool_name: str, arguments: dict) -> bool:
        if tool_name not in self.tools:
            return False
        tool_config = self.tools[tool_name]
        try:
            validate(instance=arguments, schema=tool_config['parameters_schema'])
            return True
        except jsonschema.ValidationError:
            return False

# 使用
registry = ToolRegistry("config/tools.yaml")
if registry.validate_tool_call("read_concept_card", {"card_id": "free_001"}):
    print("Valid call!")

实操心得：工具的 parameters_schema 必须和你在MiniMax API中传入的 tools 参数完全一致。我曾因为 emit_concept_analysis 的schema里漏写了 required 字段，导致模型返回的 tool_calls 参数缺失 confidence ，而Harness的校验又没跟上，结果下游代码直接 KeyError 崩溃。 工具白名单的校验，必须比API文档更严格，因为它是你系统的最后一道输入防火墙 。

3.5 第4步：做Verifier——把测试、Lint、Diff变成Harness的“质检员”

Verifier不是事后补救，而是嵌入在每一步执行流中的自动质检员。它由三个核心组件构成：

1. 结构化输出校验器（Structural Verifier） ：针对 emit_concept_analysis 的输出，用 pydantic 再次校验。

   from pydantic import BaseModel
   
   class ConceptAnalysisResult(BaseModel):
       concept: str
       candidate_definition: str
       key_points: list[str]
       open_questions: list[str]
       confidence: float
   
   # 在收到tool_calls响应后
   try:
       result = ConceptAnalysisResult(**json.loads(tool_response['result']['arguments']))
       # 校验通过，进入下一步
   except ValidationError as e:
       # 记录错误，转入ERROR_RECOVERING
   

2. 逻辑一致性Linter（Logical Linter） ：一个本地Python函数，检查 candidate_definition 是否自洽。

   def lint_definition(text: str) -> list[str]:
       """返回发现的逻辑问题列表"""
       issues = []
       if "和" in text and "或" in text:
           issues.append("定义中混用'和'与'或'，可能导致逻辑歧义")
       if len(text.split()) > 50:
           issues.append("定义过长，建议精简至30字以内")
       return issues
   

3. 版本Diff生成器（Diff Generator） ：使用 difflib 生成人类可读的对比报告。

   import difflib
   
   def generate_diff(old_def: str, new_def: str) -> str:
       old_lines = old_def.splitlines(keepends=True)
       new_lines = new_def.splitlines(keepends=True)
       diff = difflib.unified_diff(old_lines, new_lines, fromfile='v1', tofile='v2', lineterm='')
       return ''.join(diff)
   

提示：Verifier的输出（无论是 pass 还是 issues 列表）本身就是一个关键的 Artifact ，必须被持久化。我习惯把它命名为 verification_report_<id>.json ，里面包含 input_hash , verifier_name , result , issues 等字段。 质量不是一句口号，而是每一次执行后，都有一份可审计的质检报告 。

3.6 第5步：接Planner——让AI只做“计划”，不做“执行”

Planner是Harness的“大脑皮层”，负责将一个大任务分解为可执行的子任务序列。关键原则： Planner的输出，必须是Harness能100%解析的、结构化的指令，而不是自然语言 。

我定义了一个极简的Planner输出Schema：

{
  "planning_steps": [
    {
      "step_id": "1",
      "tool_name": "read_concept_card",
      "tool_arguments": {"card_id": "free_001"},
      "expected_output_type": "concept_card_snapshot"
    },
    {
      "step_id": "2",
      "tool_name": "emit_concept_analysis",
      "tool_arguments": {"concept": "自由", "user_input": "..."},
      "expected_output_type": "concept_analysis_result"
    }
  ]
}

Planner的System Prompt必须极其强硬：

“你是一个严格的Planner。你 只 能输出JSON格式的 planning_steps 数组。你 绝不 输出任何解释性文字、Markdown、代码块或额外字符。你的输出必须能被Python的 json.loads() 直接解析。如果无法生成有效计划，请输出空数组 [] 。”

实操心得：我最初让Planner输出自然语言步骤，结果Harness需要用复杂的NLP规则去解析，错误率极高。改成强制JSON Schema后，解析代码变成一行 plan = json.loads(response) ，稳定性和速度提升了十倍。 把AI的“思考”限定在结构化输出的牢笼里，是Harness稳定性的最大保障 。

3.7 第6步：接Executor——让代码执行成为“原子操作”

Executor是Harness的“肌肉”，它严格按照Planner的指令，调用工具，并将结果喂给下一个环节。它的核心是 原子性 和 幂等性 。

# executor.py
import asyncio
from tools.registry import ToolRegistry

class Executor:
    def __init__(self, registry: ToolRegistry):
        self.registry = registry
    
    async def execute_step(self, step: dict) -> dict:
        """执行单个步骤，返回结构化结果"""
        tool_name = step['tool_name']
        arguments = step['tool_arguments']
        
        # 1. 校验
        if not self.registry.validate_tool_call(tool_name, arguments):
            raise ValueError(f"Invalid tool call: {tool_name}")
        
        # 2. 执行（这里用伪代码，实际是调用你的工具函数）
        try:
            result = await self._call_tool_async(tool_name, arguments)
            return {
                "step_id": step['step_id'],
                "status": "success",
                "result": result,
                "tool_name": tool_name
            }
        except Exception as e:
            return {
                "step_id": step['step_id'],
                "status": "failed",
                "error": str(e),
                "tool_name": tool_name
            }
    
    async def _call_tool_async(self, tool_name: str, args: dict) -> dict:
        """异步调用工具函数"""
        # 根据tool_name映射到具体的函数
        if tool_name == "read_concept_card":
            return await read_concept_card_async(args['card_id'])
        elif tool_name == "emit_concept_analysis":
            return await emit_concept_analysis_async(args)
        # ... 其他工具

注意： execute_step 函数必须是 async 的，因为工具调用（尤其是API请求）是I/O密集型操作。 await 是