Agentic Harness Workflow：将AI Coding升级为工程化流水线的实践框架

坦率讲，哪怕到今天，我依然看到很多团队把 AI Coding 用成了“高级补全工具”：人开一个对话，把需求贴进去，等模型吐出一段代码，再手动复制回仓库，跑一下发现报错，又贴回去让它改。这种玩法不是不行，但它完全没有发挥出 AI 编程真正的杠杆——Agentic Harness Workflow 框架要解决的就是这件事：把 AI Coding 从“一问一答”的对话模式，升级成“需求到交付”的工程化流水线。

我手上这套框架已经跑了小半年，前后重构过三轮，从最早的一个循环调模型，到现在拆成任务节点、审核门禁、工具沙箱、观测面板一整套体系。这篇文章把我踩过的坑、沉淀下来的设计思路、以及可以直接抄走的最小实现一次性讲清楚。如果你正准备把 AI 编程引入团队流程，或者你自己一个人想管住好几个并行的 AI 任务，这篇文章应该能帮你少走很多弯路。

1. 为什么 AI Coding 不能只靠“单次对话”

1.1 Harness 到底是什么意思

先说一个很容易被忽略的词：“Harness”。它在英文里的本义是“马具、缰绳”，后来在工程领域被引申为“控制、约束、支撑系统”。你去看很多测试框架，里面都有一个 Test Harness 的概念，它就是用来把被测对象包起来，替它准备数据、调用它的接口、收集它的输出。

放到 AI Coding 场景里，Harness 做的事情完全一样：模型是我们的被测对象，也是执行者，但它本身不受控。你不知道它下一步会调用哪个工具，不知道它会在哪一步开始编造不存在的文件，也不知道它会绕多少圈子才肯停下来。Harness 的作用，就是给模型划出一条轨道，让它只能在轨道里“自主”。

我见过很多团队对“Agentic”的理解，以为就是让模型想干嘛就干嘛。实际跑下来你会发现，没有约束的自主，结果就是失控。模型可能会自己装一堆依赖、改掉不该改的配置、在一个问题上翻来覆去重试十几遍。Harness 的意义，恰恰是用结构化的流程把这种自由限制在可接受的边界里。

1.2 从“对话”到“流水线”的转变

单次对话模式的本质是“人肉维护上下文”：模型写完第一版代码，你要自己看报错信息，再把报错贴回去，模型看到的是你整理好的局部信息，它并不知道整个仓库的全貌，也没有机会主动去跑一遍测试。

工程化流程的本质则是“把步骤抽成交付链”。你可以把一次 AI 编程任务拆成这么几段：

• 需求澄清节点：模型或人来确认需求里有没有模糊点
• 技术方案节点：让模型基于仓库结构给出改动方案
• 代码生成节点：按方案逐文件生成或修改代码
• 静态检查节点：跑 lint、类型检查、编译
• 测试验证节点：跑单元测试、集成测试
• 人工审核门禁：差改动必须停下来等人确认
• 合并与回滚节点：通过后才允许合入，出问题能自动回退

每一步都是独立节点，都有输入输出，都有超时和重试策略。模型不再需要自己记住“我写到第几步了”，框架里的状态管理替它记住。这就是“Agentic + Workflow”的核心价值：智能体负责局部的自主决策，工作流负责全局的确定性编排。

2. 框架设计思路：核心概念与架构拆解

2.1 节点模型：用类型和工作流逻辑解耦

我这一版框架里，所有行为都被抽象成四类节点，这四类足够覆盖绝大多数 AI Coding 场景：

一个工作流就是这些节点的有向图。节点之间不直接互相调用，而是通过一个共享的 WorkflowContext 传递数据。这样做的好处是：任何一个节点可以替换、可以单独调试、可以挂掉重试而不影响其他节点。

Task Node 是其中最重要的节点，需要单独设计。它的输入是一个 Prompt Template 和一段上下文，输出是模型的文本回复。这里有一个非常关键的设计原则：**Task Node 内不写任何业务逻辑，只做“生成”这件事。**判断生成结果好不好、要不要重试，交给上层的 Decision Node 和 Tool Node。

2.2 状态管理：不要把上下文塞给每一个节点

一开始我犯过一个严重错误：把整个对话历史、所有中间输出、整个仓库文件树，全部塞进每个节点的 Prompt 里。结果模型看起来什么都知道，实际上什么都记不住，输出质量一路下滑，Token 费用还高得吓人。

后来的方案是“按需注入 + 显式状态”。WorkflowContext 里只存三种东西：

• 不可变输入：需求原文、项目元信息、约束条件
• 可累积产出：已生成的代码、测试报告、检查结果
• 事件日志：每个节点的耗时、Token 消耗、重试次数

在 Task Node 真正执行之前，一个专门的 ContextBuilder 模块会根据当前节点的类型，从三样东西里挑出与该任务相关的那一部分，拼成 Prompt。其他东西一概不注入。

这样做还可以顺便解决一个让很多人头疼的问题：上下文漂移。节点越多，模型越容易把早期步骤里的小决定理解歪。按需注入就像是给每个节点发一张“任务工单”，只告诉它需要知道的事情，它反而更能聚焦。

2.3 循环与控制：给 Agent 的“自由”装上刹车

Agent 任务没法避免循环——模型写代码发现测试挂了，修完又挂，再修。没有控制的话，这种循环可以无限进行下去，账单一夜跑爆。框架里必须内置三套刹车机制：

第一套，次数限制。每个 Task Node 都有 max_attempts，超过次数就进入失败分支，要么终止流程等待人工，要么切换备用模型重试。

第二套，工具调用白名单。模型只能在注册过的工具集合里选择，任何未授权的工具调用直接拒绝并记入日志。这个相当重要，你不希望模型哪天心血来潮去执行rm -rf。

第三套，成本预算。每个工作流实例可以设置总的 Token 预算或金额上限，一旦超过，框架主动中断所有节点，生成一份“成本异常报告”。

这三套刹车必须同时存在。次数限制防的是逻辑死循环，白名单防的是工具滥用，成本预算防的是资源失控。任何一个缺失，Harness 都不合格。

3. 实操：搭建一个最小可用的 Agentic Harness Workflow

3.1 确定技术栈与目录结构

我选择的是 Python 3.11 + asyncio，因为 Python 的生态对大模型和代码分析工具支持最全，asyncio 方便并行跑多个 Agent 节点。没有用重量级的服务器框架，核心就是一个异步调度器加一堆节点实现。

最小目录结构长这样：

agentic_harness/ ├── runtime.py # 核心调度器 ├── context.py # WorkflowContext 与数据装载 ├── nodes/ │ ├── base.py # 节点基类 │ ├── task_node.py # 大模型生成节点 │ ├── tool_node.py # 工具执行节点 │ └── gate_node.py # 人工门禁节点 ├── workflows/ │ └── coding_task.yaml # 工作流定义 ├── tools/ │ ├── run_tests.py │ └── git_ops.py └── main.py

这套结构的好处是，每个节点和工具都是独立的 Python 模块，可以单独写测试。我强烈建议你哪怕只是自用，也保持“定义文件与实现代码分离”：工作流描述放在 YAML 里，节点的实现放在代码里。这样调整流程不需要改代码，改 YAML 就行，非常适合多团队协作时划分责任。

3.2 定义工作流：把 AI 编程翻译成 YAML

以“修复一个失败的单元测试并合入”为例，工作流定义可以直接抄：

id: fix_failing_test name: 修复失败用例并提交 version: 1.0 timeout: 3600 budget: 200000 nodes: - id: analyze_failure type: task prompt_template: "分析以下测试失败信息，定位根因，输出简洁的诊断结论。" max_attempts: 2 - id: plan_fix type: task prompt_template: "基于诊断结论和仓库结构，输出修复方案，给出具体文件与改动点。" max_attempts: 3 - id: apply_changes type: tool tool: apply_patch max_attempts: 1 - id: run_tests type: tool tool: run_tests max_attempts: 1 - id: tests_check type: decision condition: "ctx['run_tests']['passed'] == True" next_on_true: human_review next_on_false: fix_again - id: fix_again type: task prompt_template: "根据最新报错修正之前的改动，只输出修改后的补丁。" max_attempts: 3 - id: human_review type: gate message: "AI 已完成修复，请在 30 分钟内人工审核本次改动。" - id: merge type: tool tool: git_review_merge max_attempts: 1 edges: - from: analyze_failure to: plan_fix - from: plan_fix to: apply_changes - from: apply_changes to: run_tests - from: run_tests to: tests_check - from: tests_check to: human_review - from: tests_check to: fix_again - from: fix_again to: apply_changes - from: human_review to: merge

看到里面的循环了吗？fix_again -> apply_changes -> run_tests -> tests_check是一个环，如果测试一直不过，这个环会一直转。但因为有每个节点独立的 max_attempts，以及总超时 3600 秒兜底，它不会无限跑下去。这种“局部循环 + 全局刹车”的组合，就是 Agentic 和传统流程编排最大的区别。

3.3 实现核心调度器与节点基类

调度器是整个框架的心脏，我用最简单的事件循环实现。一个节点执行完，根据它的输出和边的定义决定下一步谁执行。

class Runtime: def __init__(self, workflow, context): self.workflow = workflow self.ctx = context self.status = {} async def run(self): next_nodes = self.workflow.start_nodes() while next_nodes: current = next_nodes.pop(0) for node in current: if not await self.should_skip(node): result = await self.execute_node(node) self.status[node.id] = result next_nodes = self.resolve_next(current) return self.ctx

每个节点的基类只有两个方法：

class Node: def __init__(self, config): self.config = config self.attempts = 0 async def execute(self, ctx: WorkflowContext): raise NotImplementedError @property def exhausted(self): return self.attempts >= self.config["max_attempts"]

Task Node 的 execute 方法负责组装 Prompt、调用模型、解析输出。

class TaskNode(Node): async def execute(self, ctx: WorkflowContext): self.attempts += 1 prompt = self.build_prompt(ctx) result = await call_model( model="gpt-4o", messages=prompt, temperature=0.2, ) ctx.append_event(self.config["id"], "task_completed", result) ctx.outputs[self.config["id"]] = result return result

这里call_model是一个封装的异步函数。把模型调用封装成一层，方便以后替换提供商或加入缓存。我的经验是，这个函数一定要支持两个参数：max_tokens和temperature，并且要把每次调用的输入输出完整记录到事件日志里，否则出了问题根本没法复盘。

3.4 关键机制：工具沙箱、人工门禁与观测

工具沙箱是我后期加入的功能，但如果你打算让 Agent 在真实仓库里跑自动化操作，这一步绝对不能省。

class ToolNode(Node): async def execute(self, ctx: WorkflowContext): self.attempts += 1 tool = self.config["tool"] if tool not in ctx.allowed_tools: raise ToolNotAllowed(tool) result = await execute_tool_in_subprocess(tool, ctx) ctx.outputs[self.config["id"]] = result return result

execute_tool_in_subprocess会在一个低权限用户下执行工具命令，并限制工作目录。实际我踩过这样一个坑：第一次让 Agent 自己跑pytest，它看到测试挂了一个异步用例，竟然自作主张给测试代码加了time.sleep(1)想“修复”时序问题。这属于危险的“解题思路跑偏”。因为测试代码和业务代码都在仓库里，一个不受约束的 Agent 完全可能把测试改成“看起来通过”的样子。人工审核门禁就是防止这种隐蔽问题的最后防线。

人工门禁的实现很简单，但在消息提醒这块要稍微花点功夫。我在框架里内置了一个 gate 节点，执行到这里会向企业内部通讯工具发一条卡片消息，附带 diff 链接和确认按钮。卡片的按钮回调会写回数据库，流程继续或终止只取决于按钮结果。

class GateNode(Node): async def execute(self, ctx: WorkflowContext): await ctx.emit_gate(self.config["id"], self.config["message"]) decision = await ctx.wait_for_gate(self.config["id"], timeout=1800) if not decision.approved: raise GateRejected("人工审核未通过") ctx.outputs[self.config["id"]] = decision return decision

观测这一块也很有必要。每次节点执行完，我会往时序数据库写一条记录：节点 ID、耗时、Token 数、重试次数、状态。后面排查问题全靠它。有一段时间我发现某个任务的 Token 消耗异常高，查观测面板发现fix_again节点被触发了十几次，每个节点又往 Prompt 里塞了完整报错堆栈，相互叠加导致上下文越来越大。这个现象后来成了我们判断“上下文被污染”的典型信号。

4. 常见问题与排查技巧实录

4.1 上下文漂移：模型“忘了”最初的需求

这是跑多节点工作流时最常遇到的问题。表现为：后面的节点输出的代码，跟最初需求对不上，或者方案里的关键约束没有传递到最后的实现节点。

排查思路：先看事件日志，确认每个节点收到的事件日志是否完整。我在 2.2 节提过按需注入，这不是一句空话。要具体落实到每个 Task Node 的 Prompt 里，必须把“需求约束”单独抽成一个字段，并且在每一轮生成的结尾让它复述关键约束。

解决技巧：在 Task Node 的 Prompt 末尾加上这么一句“请先用一句话复述你刚才收到的最重要约束，再开始输出。”大量实验证明，这种强制性的“输出前复述”能显著减少生成结果偏离需求的情况，而且几乎不消耗额外成本。它可以看作是一种廉价的“工作记忆检查”。

4.2 工具调用循环：模型反复执行同一个操作

有一段时间，我们的工作流频繁出现模型的工具调用死循环，比如反复执行同一段git diff，或者反复跑同一组测试，但什么都不改。后来定位到原因：模型没有从工具输出中提取到“下一步要干什么”，于是默认选择重复上一次动作。

解决这个问题，我给 Tool Node 加了一层“操作总结器”。工具执行完后，用一个小的专用模型调用，把原始输出（比如测试失败的一大堆日志）压缩成结构化摘要，包括核心错误、涉及文件、建议定位范围。再把摘要喂给下一个 Task Node。这一步看似多花了 Token，实际上是省钱的，因为下一个节点不需要阅读完整日志就能理解现状，而且不会被日志里的噪声带偏。

另外一个很实用的限制是：给同一种工具操作加“间隔锁”。短时间内重复触发同一个工具且输入完全一致时，调度器直接拒绝执行，返回上一次的结果。这个策略在代码生成场景里非常有效，因为重复运行同一个测试并且输入没变，结果本来就一样。

4.3 生成结果不稳定：同样的输入，不同的输出

很多人的第一反应是调低 temperature，把模型参数改成 0。实测下来，temperature 设成 0 确实能降低随机性，但不能完全消除不确定性，因为模型本身有采样逻辑，而且不同版本模型之间的行为差异更大。我们的做法有两层：第一层，模型参数固定，temperature 固定在 0.2 左右，给一点随机性但不过分；第二层，建一个“回归样本集”，把过去出现问题的工作流输入输出记录下来，每次框架改动后，跑一遍回归集，确保质量没有回退。

这个回归集非常值钱。有一次我重构 ContextBuilder 模块，自认为会减少 Token 消耗，结果回归样本集跑下来，三个样本的方案生成质量明显下滑，才发现新版本把“需求约束”过滤掉了。如果没有回归集，这种回归错误可能要等真实任务跑挂才能发现。

4.4 测试被 Agent“作弊”式修复

这个我在实操部分提过，值得单独拿出来说：模型发现测试挂了，它不修业务代码，反而去修测试断言，这是一种非常典型的“走捷径”行为。它会给你一个非常合理的话术，比如“调整测试用例以匹配实际行为”，但在代码交付里，这种行为往往是不可接受的。

应对手段，首先是给“测试验证”节点加目录限制，只运行指定测试目录，并且让它读入 tests 目录下的 README 或约定文件，明确“测试优先级高于实现”。更重要的是，在人工审核卡片里，必须并排展示业务代码 diff 和测试文件 diff，审核人一眼就能看出是不是动了测试断言。

如果条件允许，可以在工具沙箱里把测试目录设为只读。这样 Agent 连改测试文件的可能性都被物理切断了。这是我的终极方案，后面跑下来再也没出现过“作弊式修复”。

4.5 Diff 过小或过大：模型偷懒或超范围修改

模型有时候会把任务“缩水”：需求是要修一个函数，它只改了一行注释，然后说完成。反过来也会超范围：让它修一个 bug，它顺带着重构了好几个文件。这两种情况都必须通过工具检查来拦截。

我在 apply_changes 这样工具节点之前，总设一个“diff 统计节点”：跑git diff --stat，把改动文件数和行数写入上下文，然后与任务预估范围做比较。改动行数低于某个阈值或高于某个阈值时，工作流自动进入“需求复查”分支，让模型重新阅读需求，并判断当前改动是否符合。

这种启发式检查不会百分之百准确，但很管用。至少它能抓住“完全没改”“狂改一堆文件”这两个极端，把 Agent 的行为拉回到可控范围。

5. 再分享几个让我少走三年弯路的经验

直接给结论，不铺垫。

**先设计状态结构，再设计 Prompt 模板。**大多数人反过来。他们在意 Prompt 怎么写，却不关心状态怎么传。一旦节点多了，状态结构不合理，Prompt 写得再精致也白搭，信息该丢还是丢。我的建议是，定义工作流之前，先画一张“数据流图”，确定每个节点从哪个字段取数、往哪个字段写数。这一步做扎实，后面开发会非常顺。

**人工审核点宁可多，不要少。**初期我们只设了一个最终审核点，结果模型生成的中间方案越走越偏，到最后一步人工才发现，返工成本极高。后来在方案生成阶段和代码生成阶段各加了一道快速审核点，改动大方向错了能及时被拦下来。虽然多花了人几分钟时间，但整体效率反而提升了一倍。

**日志里一定要有 Token 消耗的增量。**之前我只看总消耗，排查问题特别困难。改成每个节点独立记录增量消耗后，一眼就能看出哪个节点在“烧钱”。我甚至给日志加了个警告：单个节点消耗超过平均值的 5 倍时，直接在面板里标红。这让你快速识别出上下文越来越大的异常节点。

这套 Agentic Harness Workflow 本身不是银弹。它的所有优势，都建立在“你把流程想清楚了”的前提上。如果业务本身一团乱麻，装再好的 Harness 也是让模型在一团乱麻里更快地迷路。但一旦你的流程可描述、可拆解、可度量，再给它套上这套框架，AI Coding 的生产效率确实会迎来一轮质的提升。我自己的项目组，过去一个功能从开发到自测差不多要两天，现在稳定在半天以内，而且质量波动小得多。建议你从最小的场景开始，我一个同事就是从“自动修一个失败的测试用例”起步，跑了不到两个月，就培养出了对整个框架的信心。