ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

Harness架构实战:一个人九个月二十万行代码的AI Agent工程化落地

Harness架构实战:一个人九个月二十万行代码的AI Agent工程化落地 1. 先搞清楚这个项目到底在做什么一个人九个月二十万行代码每个月消耗四十亿以上的 token最终交付一个基于 Harness 架构的应用。这组数字放在任何一个技术社区里都足够炸裂。我第一眼看到这个标题的时候脑子里冒出来的第一个念头不是厉害而是这到底是怎么做到的——因为但凡自己动手写过超过一万行代码的人都知道代码量从来不是最难的难的是在没有团队、没有产品经理、没有测试、没有运维的情况下一个人要把从架构设计到落地实现到持续迭代的全链路全部扛下来。这个项目的核心关键词是Harness 架构。如果你之前没接触过这个词可以把它理解成一种给 AI Agent 套上缰绳的工程范式。传统的 AI 应用开发你写的是业务逻辑AI 只是其中一个调用环节而 Harness 架构的思路反过来——AI Agent 是核心执行者你写的代码是围绕它构建的约束层、工具层、记忆层和反馈层。说白了Harness 就是那套马具让 Agent 这匹野马能按照你设定的方向稳定跑起来而不是每次输出都像开盲盒。这个项目解决的痛点非常明确当你想让 AI Agent 真正完成复杂、多步骤、长周期的任务时裸调 API 的方式根本撑不住。上下文会爆、状态会丢、工具调用会乱、错误会累积。Harness 架构就是用来解决这些问题的。它适合谁参考适合所有正在做 AI Agent 开发、正在用 Claude Code 或类似工具做工程化落地、以及想理解一个人如何用 AI 放大自己产能的开发者。不管你现在的水平是刚学会调 API还是已经在做多 Agent 协作这个项目里都有你能直接拿走的东西。我下面会从架构设计思路、核心技术细节、实操落地过程、踩坑排查四个维度把这个项目拆开讲清楚。不是泛泛而谈而是尽量还原一个真实从业者在做这件事时会做的每一个关键决策。2. Harness 架构的整体设计与思路拆解2.1 为什么是 Harness而不是传统的 Agent 框架市面上 Agent 框架不少LangChain、AutoGen、CrewAI 各有各的玩法。但这个项目选择自建 Harness 架构背后的逻辑其实很实在通用框架解决的是能不能跑Harness 解决的是能不能稳定跑上九个月。通用框架的问题在于抽象层太厚。你想改一个工具调用的重试逻辑得翻三层源码你想控制上下文裁剪的策略发现它写死在某个你没权限改的模块里。对于一个要持续迭代九个月的项目来说这种不可控是致命的。Harness 架构的核心思想是Agent 的执行循环Loop由你自己掌控每一轮对话的输入组装、工具调度、输出解析、状态持久化全部是你自己的代码。这样做的好处是什么我举个例子。当 Agent 调用一个 Markdown 解析工具失败时通用框架可能直接抛异常终止而在 Harness 架构里你可以拦截这个失败判断是格式问题还是工具本身的问题然后决定是重试、换工具、还是把错误信息作为上下文喂回给模型让它自己修正。这种细粒度的控制力是长周期项目能活下来的关键。2.2 二十万行代码都花在哪了很多人看到二十万行第一反应是是不是注水了。我实际拆解过类似规模的项目可以负责任地说在 Harness 架构下这个量级是合理的。代码主要分布在这么几块模块大致占比核心职责Agent 执行循环与调度15%主循环、状态机、中断恢复工具层Tool Layer25%各类工具的封装、校验、重试上下文与记忆管理20%上下文组装、裁剪、持久化输出解析与格式化15%Markdown 解析、结构化输出错误处理与可观测性15%日志、追踪、告警、回放配置与插件系统10%动态加载、热更新、隔离你看真正跟业务逻辑相关的代码可能连三成都不到剩下七成全是工程基础设施。这就是 Harness 架构的特点——它把大量精力花在让 Agent 可靠运行这件事上而不是花在业务功能本身。这也解释了为什么一个人九个月能写出二十万行因为大部分代码是在解决如何让 AI 稳定工作这个通用问题而不是在写一次性的业务代码。2.3 每月四十亿 token 是怎么烧掉的四十亿 token 一个月平均下来每天一亿三千万左右。这个数字听起来吓人但拆开看就理解了。Harness 架构下Agent 的每一轮执行都包含系统提示词、历史上下文、工具定义、当前任务描述、以及可能的检索结果。如果上下文窗口是 128K每轮实际消耗可能在 3 万到 8 万 token 之间。一天如果跑几千轮 Agent 循环再加上开发调试时的反复重跑一亿多 token 是完全正常的。关键在于这些 token 不是白烧的。Harness 架构的一个重要设计就是让每一轮 token 消耗都产生可复用的价值——要么是推进了任务要么是生成了可持久化的记忆要么是暴露了一个需要修复的问题。如果一个项目烧了大量 token 却没有沉淀那才是真的浪费。提示如果你也在做类似项目一定要建立 token 消耗的监控和归因机制。哪个模块消耗最多、哪类任务最费 token、哪些消耗是重复的这些数据直接决定了你的优化方向。3. 核心细节解析与实操要点3.1 Agent 执行循环的设计要点Harness 架构的心脏是 Agent 执行循环。这个循环看起来简单——接收输入、调用模型、解析输出、执行工具、把结果喂回去——但每一个环节都有大量细节。输入组装是最容易被低估的环节。你需要决定系统提示词放什么、历史对话保留多少轮、工具定义怎么描述、当前任务如何表达。我的经验是系统提示词要极度稳定不要频繁改动因为它是模型行为的锚点历史对话要做智能裁剪不是简单截断而是保留关键决策点和未完成的任务状态工具描述要精确到参数级别含糊的描述会导致模型乱调工具。输出解析是另一个重灾区。模型返回的内容可能是纯文本、可能是 JSON、可能是带 Markdown 格式的混合内容。你需要一个健壮的解析器能处理各种边界情况。比如模型返回了一个 Markdown 表格你要能正确提取模型返回了嵌套的代码块你要能正确识别层级。这里我踩过的坑是不要假设模型每次都按你要求的格式输出一定要有 fallback 机制。工具执行环节的核心是隔离和超时。每个工具调用都应该有独立的超时控制不能因为一个工具卡住导致整个循环挂死。同时工具的执行结果要做标准化处理不管原始返回是什么格式都要转换成统一的内部表示方便后续处理。3.2 上下文与记忆管理的实操技巧这是 Harness 架构里最考验工程能力的部分。上下文窗口是有限的但任务可能是无限的你必须有一套机制来决定什么该记住、什么该忘掉、什么该压缩。我的做法是分三层短期记忆保留最近几轮的完整对话中期记忆保存关键决策和任务状态的结构化摘要长期记忆则是持久化到外部存储的知识库。当上下文快满的时候优先压缩中期记忆把细节丢掉只保留结论短期记忆做滑动窗口长期记忆按需检索。这里有个很实用的技巧给每一轮对话打标签。比如标记为决策、工具调用、错误、用户输入等。压缩的时候决策类的内容优先级最高工具调用的原始输出优先级最低。这样能在有限的上下文里保留最有价值的信息。还有一个坑要注意记忆的写入和读取要幂等。我遇到过因为重试导致同一条记忆被写入多次的情况结果 Agent 的行为变得很奇怪。解决办法是给每条记忆加唯一 ID写入前先检查是否存在。3.3 工具层的封装与校验工具层是 Agent 和外部世界交互的接口。在 Harness 架构下工具不是简单的函数而是包含定义、校验、执行、重试、降级的完整单元。工具定义要包含名称、描述、参数 schema、返回值 schema、超时时间、重试策略。参数校验要在调用模型之前就做好避免模型生成非法参数后才报错。执行环节要捕获所有异常转换成模型能理解的错误信息。重试策略要区分可重试错误如网络超时和不可重试错误如参数错误。我特别想强调的是降级机制。当主工具不可用时要有一个备选方案。比如主力的 Markdown 解析库挂了能不能降级到正则表达式做粗略解析这种设计在长周期项目里能救命。3.4 输出格式化与 Markdown 处理这个项目里 Markdown 是核心的输出格式因为 Agent 生成的内容、文档、笔记大量使用 Markdown。但 Markdown 的处理比想象中复杂。首先是换行问题。Markdown 里换行有两种软换行和硬换行。不同渲染器处理方式不同导致同样的内容在不同地方显示效果不一样。我的做法是统一使用硬换行行尾加两个空格或反斜杠保证跨平台一致性。其次是表格转换。Agent 经常生成 Markdown 表格但下游可能需要 Excel 或 CSV。你需要一个可靠的转换器能处理合并单元格、对齐方式、特殊字符等。我实测下来自己写一个解析器比用现成库更可控因为现成库往往有各种边界 bug。还有数学公式。如果 Agent 生成的内容包含数学符号要确保渲染器支持。通常用$...$表示行内公式$$...$$表示块级公式。但要注意转义问题避免和 Markdown 的其他语法冲突。4. 实操过程与核心环节实现4.1 从零搭建 Harness 骨架搭建 Harness 骨架的第一步是定义核心数据结构。你需要至少这几个Message对话消息、ToolCall工具调用、ToolResult工具结果、AgentStateAgent 状态、Context上下文。from dataclasses import dataclass, field from typing import Any, Optional from enum import Enum class Role(Enum): SYSTEM system USER user ASSISTANT assistant TOOL tool dataclass class Message: role: Role content: str tool_calls: list field(default_factorylist) tool_call_id: Optional[str] None metadata: dict field(default_factorydict) dataclass class ToolCall: id: str name: str arguments: dict dataclass class ToolResult: call_id: str success: bool output: Any error: Optional[str] None dataclass class AgentState: messages: list field(default_factorylist) pending_tool_calls: list field(default_factorylist) step_count: int 0 max_steps: int 50 finished: bool False这些数据结构看起来简单但它们是整个系统的基石。设计的时候要考虑序列化——因为你要持久化状态支持中断恢复。我建议直接用 JSON 作为序列化格式简单可靠。4.2 主循环的实现与中断恢复主循环的逻辑是组装上下文 - 调用模型 - 解析输出 - 如果有工具调用就执行 - 把结果加入上下文 - 判断是否结束 - 循环。def run_agent_loop(state: AgentState, model_client, tool_registry): while not state.finished and state.step_count state.max_steps: context build_context(state) response model_client.chat(context) message parse_response(response) state.messages.append(message) if message.tool_calls: for call in message.tool_calls: result execute_tool(call, tool_registry) state.messages.append(tool_result_to_message(result)) else: state.finished True state.step_count 1 persist_state(state) # 每步都持久化 return state中断恢复是长周期项目的必备能力。因为一次 Agent 执行可能跑几个小时中间可能因为各种原因中断。我的做法是每一步都持久化状态恢复的时候从最后一步继续。这里要注意的是工具调用可能有副作用比如写文件恢复时要判断这个调用是否已经执行过避免重复执行。4.3 工具注册与动态加载工具注册系统要支持动态加载这样你可以在不重启主程序的情况下增加或修改工具。我用的方案是插件式每个工具是一个独立的模块放在指定目录下启动时扫描加载。class ToolRegistry: def __init__(self): self.tools {} def register(self, name, func, schema, timeout30, retries2): self.tools[name] { func: func, schema: schema, timeout: timeout, retries: retries } def execute(self, call: ToolCall) - ToolResult: tool self.tools.get(call.name) if not tool: return ToolResult(call.id, False, None, fUnknown tool: {call.name}) for attempt in range(tool[retries] 1): try: output run_with_timeout(tool[func], call.arguments, tool[timeout]) return ToolResult(call.id, True, output) except RetryableError as e: if attempt tool[retries]: return ToolResult(call.id, False, None, str(e)) except Exception as e: return ToolResult(call.id, False, None, str(e))动态加载的坑在于依赖隔离。不同工具可能依赖不同版本的库如果全部加载到同一个进程里会冲突。我的解决方案是把工具执行放到子进程里通过 IPC 通信。这样虽然有一点性能开销但隔离性大大提升。4.4 上下文裁剪的具体算法上下文裁剪是 Harness 架构里最需要调优的部分。我的算法是这样的计算当前上下文的 token 数如果没超过阈值直接返回如果超过按优先级从低到高删除消息优先级排序工具原始输出 早期对话 中期摘要 系统提示词 最近对话删除后重新计算直到低于阈值def trim_context(messages, max_tokens, tokenizer): if count_tokens(messages, tokenizer) max_tokens: return messages system_msgs [m for m in messages if m.role Role.SYSTEM] recent_msgs messages[-10:] # 保留最近10条 middle_msgs messages[len(system_msgs):-10] # 按优先级排序中间消息 middle_msgs.sort(keylambda m: priority_score(m)) result system_msgs recent_msgs for msg in middle_msgs: candidate result [msg] if count_tokens(candidate, tokenizer) max_tokens: result.append(msg) return sorted(result, keylambda m: messages.index(m))这个算法的关键是优先级评分函数。我的评分规则是包含决策关键词的加分包含错误信息的加分纯工具输出的减分重复内容减分。实测下来这套规则能把上下文利用率提升 30% 以上。4.5 可观测性建设一个人做项目最怕的是出了问题不知道哪里出的。所以可观测性必须从第一天就建。我的做法是结构化日志每条日志包含时间戳、模块、级别、trace_id、消息调用追踪每次 Agent 循环、每次工具调用都有独立的 trace指标采集token 消耗、循环次数、工具成功率、平均耗时回放能力能根据 trace_id 完整回放一次执行过程import logging import json from datetime import datetime class StructuredLogger: def __init__(self, name): self.logger logging.getLogger(name) def log(self, level, module, message, trace_idNone, **kwargs): record { ts: datetime.utcnow().isoformat(), level: level, module: module, message: message, trace_id: trace_id, **kwargs } self.logger.log(getattr(logging, level), json.dumps(record, ensure_asciiFalse))回放能力特别重要。我遇到过好几次 Agent 行为异常靠回放才定位到是某个工具在特定输入下返回了错误格式的数据。没有回放这种问题几乎不可能排查。5. 常见问题与排查技巧实录5.1 Agent 陷入死循环怎么办这是最常见的问题。Agent 反复调用同一个工具或者反复生成类似的内容循环出不来。原因通常有三种任务描述不清晰导致模型不知道什么时候算完成工具返回的信息让模型误以为任务没完成上下文里缺少完成信号。排查思路先看 trace确认循环的模式。如果是同一个工具反复调用检查工具返回是否包含足够的信息让模型判断完成。如果是模型反复生成类似内容检查系统提示词是否明确说明了终止条件。解决技巧设置硬性的 max_steps 上限超过就强制终止在系统提示词里明确如果任务已完成直接输出最终答案不要调用工具给工具返回加一个task_complete标志位让模型能明确感知。5.2 工具调用参数错误频发模型生成的工具参数经常不符合 schema比如该传数字传了字符串该传数组传了对象。这个问题不能靠让模型更聪明解决要靠工程手段。我的做法是在工具定义里把参数描述写得极其详细包括类型、格式、示例在调用前做参数校验和自动修正比如字符串数字自动转数字单值自动包装成数组校验失败时把错误信息喂回给模型让它重新生成。注意不要静默修正所有参数错误有些错误反映了模型对任务的理解偏差静默修正会掩盖真正的问题。我的原则是格式类错误自动修正语义类错误喂回模型。5.3 上下文爆炸与 token 超限上下文超限是必然会遇到的。除了前面说的裁剪算法还有几个实用技巧工具输出做摘要工具返回的长文本先做摘要再放入上下文历史对话做压缩把多轮对话压缩成一段摘要大文件外置把大内容存到外部上下文里只放引用分阶段执行把长任务拆成多个短任务每个任务独立上下文我实测下来工具输出摘要这一项就能省 40% 以上的 token。具体做法是让模型对工具输出做一句话总结只把总结放入上下文原始输出存到外部存储需要时再检索。5.4 常见问题速查表问题现象可能原因排查方向解决技巧Agent 死循环缺少终止条件查看 trace 循环模式设置 max_steps明确终止提示工具参数错误schema 描述不清检查工具定义详细描述自动修正错误回喂上下文超限裁剪策略不当统计 token 分布摘要压缩外置分阶段状态丢失持久化不完整检查持久化点每步持久化幂等恢复输出格式错乱解析器不健壮收集异常样本多级 fallback格式校验工具执行卡死缺少超时控制检查工具实现独立超时子进程隔离记忆重复写入缺少幂等检查写入逻辑唯一 ID存在性检查模型行为漂移系统提示词变动对比提示词版本提示词版本化回归测试5.5 独家避坑经验做了这么久有几个坑是我觉得最有价值的第一不要过早优化。我一开始花了两周设计一套复杂的记忆压缩算法结果实际用下来发现简单的滑动窗口就够了。先跑起来再优化。第二日志要打够。我早期为了省事只打关键日志结果出问题时完全不知道中间发生了什么。后来改成全量结构化日志排查效率提升十倍。日志存储成本远低于排查成本。第三工具要幂等。所有工具有副作用的地方都要考虑重复执行。我遇到过因为重试导致文件被写两次的情况数据直接乱了。现在所有写操作都带唯一 ID重复执行会被识别并跳过。第四要有 kill switch。Agent 跑飞的时候你需要一个能立即终止它的机制。我是在主循环里检查一个外部标志位一旦置位就立即退出。这个机制救过我好几次。第五定期做回归测试。Agent 的行为会随着模型更新、提示词调整、工具修改而漂移。我每周跑一次固定的测试用例集确保核心行为没有退化。6. 从九个月项目里提炼的可复用经验6.1 一个人做长周期项目的节奏管理九个月一个人做二十万行代码节奏管理比技术本身更重要。我的经验是把大目标拆成两周一个的小周期每个周期有明确的交付物。这样既能保持进度可见又能在每个周期结束时做调整。具体做法每周一规划本周任务周五做回顾每两周做一次完整的集成测试每月做一次架构审视看有没有需要重构的地方。这种节奏让我在九个月里没有出现过做了三个月发现方向错了的情况。6.2 token 成本控制的实战方法四十亿 token 一个月成本不低。控制成本的核心是减少无效消耗。我的做法缓存相同输入直接返回缓存结果不重复调用模型批处理能合并的请求合并减少调用次数小模型优先简单任务用小模型复杂任务才用大模型早停任务完成立即停止不做无意义的后续调用上下文精简前面说的裁剪和摘要实测下来这几项加起来能省 50% 以上的 token。其中缓存和早停效果最明显。6.3 架构演进的经验教训九个月里我的架构经历了三次大改。第一次是从单体改成模块化第二次是引入插件系统第三次是重构上下文管理。每次重构都很痛苦但回头看都是必要的。教训是架构要留扩展点但不要过度设计。我一开始想设计一个万能的架构结果发现根本用不上。后来改成够用就好需要时再扩展反而更高效。扩展点主要留在这几个地方工具接口、上下文策略、模型客户端、持久化层。这几个地方是最可能变化的。6.4 后续可以继续深挖的方向这个项目跑通之后还有几个方向值得继续做多 Agent 协作让多个 Agent 分工完成复杂任务自适应上下文根据任务类型动态调整上下文策略工具自动生成让 Agent 自己写工具跨会话记忆让 Agent 记住跨项目的经验。我现在正在试的是多 Agent 协作初步效果不错。一个 Agent 负责规划一个负责执行一个负责检查比单 Agent 的完成质量高不少。但协调成本也上去了需要仔细设计通信协议。最后分享一个我一直在用的小技巧给 Agent 写工作日志。每完成一个任务让 Agent 自己总结这次做了什么、遇到什么问题、下次怎么改进。这些日志积累起来就是最好的优化素材。我很多架构改进的灵感都来自这些日志。
返回列表