Harness工程
Harness(英 [ˈhɑːnɪs]):原意为"挽具",即套在马身上用于控制和引导的装备。在AI语境中,Harness工程指的是围绕AI Agent构建的控制系统,用于引导、约束和增强代理的能力,使其能够在生产环境中稳定、可控地运行。
引言:为什么需要Harness工程?
当我们谈论AI Agent时,脑海中浮现的往往是一个能够自主思考、使用工具、完成复杂任务的智能体。然而,现实中的Agent远比这复杂——它需要处理权限控制、错误恢复、上下文管理、任务调度等一系列工程问题。这就好比一匹骏马,如果没有合适的挽具,它无法有效地拉车或载人。
Harness工程的本质,就是为AI Agent提供一套完整的"操作系统"。它不改变Agent的核心能力(推理、生成、决策),而是通过一系列精心设计的子系统,将这些能力组织成可靠、可控、可扩展的生产系统。
Learn Claude Code项目为我们提供了一个绝佳的案例研究。它通过20个渐进式步骤,从102行代码的最小代理循环,逐步构建出1708行代码的生产级代理框架。这个过程完美展示了Harness工程的核心思想:一个循环,周围围绕着使其具备生产形态的各个子系统。
本文将从Agent Loop、Tool Use、Permission、Hooks、Todo、Skill、Context、Memory、Error Recovery、Subagent、Task、Background Tasks、Cron Scheduler、Agent Teams等16个维度,深入剖析Harness工程的设计哲学、实现原理和未来演进方向。
第一章:核心循环——Agent Loop
1.1 是什么:代理的"心跳"
Agent Loop是整个Harness工程的心脏。它的本质极其简单:
while True:
response = model.generate(messages)
if response.has_tool_call:
result = execute_tool(response.tool_call)
messages.append(result)
else:
break这个循环只有三步:
- 调用模型:将当前对话历史发送给语言模型
- 检查响应:模型是否请求使用工具?
- 执行或结束:如果有工具调用,执行并将结果反馈;否则结束
1.2 为什么:简单即强大
为什么如此简单的循环能够成为整个框架的基础?因为它体现了递归分解的哲学:
- 任何复杂任务都可以分解为一系列工具调用
- 每个工具调用都可以进一步分解为更小的工具调用
- 直到最终任务可以直接完成
这种设计的优雅之处在于:循环本身不需要知道它在解决什么问题。它只是一个执行引擎,将模型的意图转化为实际行动。
1.3 怎么设计:最小可行循环
Learn Claude Code项目的s01阶段展示了最小可行循环的实现:
# 102行代码的最小代理循环
import anthropic
client = anthropic.Anthropic()
messages = []
tools = [read_file_tool] # 只有1个工具
while True:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=tools,
messages=messages
)
if response.stop_reason == "tool_use":
# 执行工具并继续循环
tool_result = execute_tool(response.content)
messages.append(tool_result)
else:
# 模型完成,退出循环
print(response.content)
break这个设计的精髓在于:它是一个无限循环,但通过模型的响应来决定何时退出。模型既是决策者,也是终止条件的判断者。
1.4 本质探索:循环即状态机
从更深层次看,Agent Loop本质上是一个有限状态机:
状态:[思考中] --模型响应--> [使用工具] --执行结果--> [思考中]
[思考中] --无工具调用--> [结束]每个状态转移都由模型的输出决定。这意味着:
- 代理的行为完全由模型的推理驱动
- 框架只负责状态转移的执行
- 任何复杂的行为都是这个简单状态机的组合
1.5 当前问题与改进方向
问题1:循环效率
当前循环是串行的——一次只能执行一个工具调用。当多个工具调用之间没有依赖关系时,这造成了不必要的等待。
改进方向:并行工具执行。框架可以分析工具调用之间的依赖关系,将无依赖的调用并行执行。
问题2:循环控制
模型可能陷入无限循环——反复调用相同的工具却无法取得进展。
改进方向:引入循环检测和干预机制。框架可以监控工具调用模式,当检测到循环时强制注入新的提示或终止循环。
第二章:工具系统——Tool Use
2.1 是什么:代理的"手"
如果Agent Loop是心脏,那么Tool Use就是代理的双手。工具是代理与外部世界交互的唯一途径——无论是读取文件、执行命令、调用API,还是操作数据库。
2.2 为什么:能力扩展的边界
语言模型本身只能生成文本。要让它成为真正的代理,必须赋予它行动的能力。工具系统解决的核心问题是:如何让模型知道有哪些工具可用,以及如何正确使用它们。
2.3 怎么设计:分发器模式
Learn Claude Code项目采用分发器模式来设计工具系统:
# 工具注册表
TOOLS = {
"read_file": {
"description": "读取文件内容",
"parameters": {"path": "文件路径"},
"handler": read_file_handler
},
"write_file": {
"description": "写入文件内容",
"parameters": {"path": "文件路径", "content": "文件内容"},
"handler": write_file_handler
}
}
def execute_tool(tool_call):
tool = TOOLS[tool_call.name]
return tool["handler"](**tool_call.parameters)这种设计的优势:
- 添加工具只需注册一行代码
- 主循环完全不知道具体工具有哪些
- 工具的实现细节被完全封装
2.4 本质探索:工具即接口
从软件工程的角度看,工具系统本质上是一个接口抽象层。它定义了代理与外部世界交互的契约:
- 工具描述:告诉模型这个工具能做什么
- 参数定义:告诉模型需要提供什么信息
- 返回格式:告诉模型会得到什么结果
这种抽象使得模型不需要知道工具的实现细节,只需要知道如何描述它的意图。
2.5 当前问题与改进方向
问题1:工具发现
当工具数量增多时,模型可能不知道哪个工具最适合当前任务。
改进方向:智能工具推荐。根据当前上下文和任务类型,动态推荐最相关的工具子集。
问题2:工具组合
复杂的任务往往需要多个工具的组合使用,但模型可能无法一次性规划出最优的工具调用序列。
改进方向:工具编排引擎。框架可以提供工具组合的模板或规划器,帮助模型更好地组织工具调用。
第三章:权限控制——Permission
3.1 是什么:代理的"刹车"
权限系统是Harness工程中的安全阀。它决定了代理在什么条件下可以执行什么操作。没有权限控制的代理就像没有刹车的汽车——可能很快,但极其危险。
3.2 为什么:信任与安全的平衡
代理需要执行各种操作,但并非所有操作都是安全的。权限系统需要在两个极端之间找到平衡:
- 过度限制:代理无法完成任何有用的工作
- 过度放开:代理可能造成不可逆的损害
3.3 怎么设计:决策点模式
权限控制通常在工具执行前插入一个决策点:
def execute_tool_with_permission(tool_call):
# 决策点:检查权限
if needs_permission(tool_call):
user_decision = ask_user_permission(tool_call)
if user_decision == "deny":
return PermissionDeniedError()
# 执行工具
return execute_tool(tool_call)权限决策可以基于多种因素:
- 操作类型:读取 vs 写入 vs 删除
- 影响范围:单个文件 vs 整个目录
- 可逆性:可撤销 vs 不可撤销
- 用户偏好:自动允许 vs 每次询问
3.4 本质探索:权限即策略
权限系统本质上是一个策略引擎。它将"什么可以做"的决策从执行逻辑中分离出来,使得:
- 策略可以独立于实现进行调整
- 不同的场景可以应用不同的策略
- 策略可以被审计和追踪
3.5 当前问题与改进方向
问题1:权限粒度
当前权限系统通常是二元的——允许或拒绝。但在实际场景中,往往需要更细粒度的控制。
改进方向:条件权限。例如,"允许写入/tmp目录,但不允许写入/etc目录"。
问题2:权限学习
每次操作都需要用户决策,这会造成"权限疲劳"。
改进方向:权限学习。框架可以学习用户的决策模式,自动应用已知的安全策略。
第四章:钩子系统——Hooks
4.1 是什么:代理的"神经系统"
钩子系统是Harness工程中的横切关注点处理器。它允许在代理生命周期的关键点插入自定义逻辑,而不修改核心循环代码。
4.2 为什么:关注点分离
代理需要处理许多与核心逻辑无关但至关重要的任务:
- 日志记录:记录代理的行为以便调试
- 监控指标:收集性能和使用数据
- 审计追踪:记录所有操作以便合规
- 缓存优化:缓存频繁使用的工具结果
如果将这些逻辑直接写入循环,会导致代码臃肿且难以维护。
4.3 怎么设计:事件驱动模式
钩子系统通常采用事件驱动模式:
# 钩子注册
hooks = {
"before_tool_call": [log_tool_call, check_rate_limit],
"after_tool_call": [cache_result, update_metrics],
"on_error": [retry_with_backoff, notify_user]
}
def execute_tool_with_hooks(tool_call):
# 触发前置钩子
for hook in hooks["before_tool_call"]:
hook(tool_call)
# 执行工具
result = execute_tool(tool_call)
# 触发后置钩子
for hook in hooks["after_tool_call"]:
hook(tool_call, result)
return result4.4 本质探索:钩子即中间件
钩子系统本质上是一种中间件模式。它将横切关注点从业务逻辑中解耦,使得:
- 每个钩子只关注单一职责
- 钩子可以独立添加、删除或修改
- 核心循环保持简洁和专注
4.5 当前问题与改进方向
问题1:钩子顺序
当多个钩子注册到同一个事件时,执行顺序可能很重要。
改进方向:优先级系统。为每个钩子分配优先级,确保关键钩子(如安全检查)优先执行。
问题2:钩子性能
过多的钩子可能影响代理的响应速度。
改进方向:异步钩子。非关键钩子可以在后台异步执行,不阻塞主循环。
第五章:计划管理——Todo
5.1 是什么:代理的"任务清单"
Todo系统为代理提供了显式的计划管理能力。它允许代理将复杂任务分解为可追踪的子任务,并跟踪每个子任务的完成状态。
5.2 为什么:长期任务的导航
没有计划的代理就像没有地图的旅行者——可能不断前进,但很容易迷失方向。Todo系统解决的核心问题是:
- 任务分解:将大目标拆分为可执行的小步骤
- 进度跟踪:知道已经完成了什么,还剩什么
- 错误恢复:当某个步骤失败时,知道从哪里重新开始
5.3 怎么设计:状态机模式
Todo系统通常采用状态机模式:
class TodoItem:
def __init__(self, description):
self.description = description
self.status = "pending" # pending, in_progress, completed, failed
self.dependencies = []
self.result = None
class TodoManager:
def __init__(self):
self.items = []
def add_item(self, description, dependencies=[]):
item = TodoItem(description)
item.dependencies = dependencies
self.items.append(item)
def get_next_task(self):
# 返回下一个可以执行的任务(所有依赖已完成)
for item in self.items:
if item.status == "pending":
if all(self.is_completed(dep) for dep in item.dependencies):
return item
return None
def mark_completed(self, item, result):
item.status = "completed"
item.result = result5.4 本质探索:计划即状态
Todo系统本质上是一个状态管理器。它将代理的"记忆"从对话历史中提取出来,以结构化的方式存储,使得:
- 计划可以被持久化:即使对话中断,计划也不会丢失
- 计划可以被修改:根据新信息调整后续步骤
- 计划可以被共享:多个代理可以协作执行同一个计划
5.5 当前问题与改进方向
问题1:计划僵化
代理可能严格按照计划执行,即使发现计划不再最优。
改进方向:动态计划调整。代理可以根据执行结果和新信息,主动修改后续计划。
问题2:依赖管理
复杂的依赖关系可能导致死锁或循环依赖。
改进方向:依赖图分析。框架可以检测并预防依赖问题,确保计划的可行性。
第六章:技能加载——Skill
6.1 是什么:代理的"专业知识"
技能系统允许代理按需加载专业领域的知识和能力。它不是一个独立的功能模块,而是一种知识注入机制。
6.2 为什么:上下文的稀缺性
语言模型的上下文窗口是有限的。如果将所有可能需要的知识都放入系统提示,会:
- 浪费宝贵的上下文空间
- 增加模型的处理负担
- 降低响应的相关性
技能系统通过按需加载解决了这个问题。
6.3 怎么设计:动态注入模式
技能系统通常采用动态注入模式:
class SkillLoader:
def __init__(self):
self.skills = {
"python": "你是一个Python专家...",
"database": "你是一个数据库专家...",
"security": "你是一个安全专家..."
}
def load_skill(self, skill_name):
if skill_name in self.skills:
return self.skills[skill_name]
return None
def inject_skill(self, messages, skill_name):
skill_content = self.load_skill(skill_name)
if skill_content:
# 在系统提示中注入技能
messages.insert(0, {"role": "system", "content": skill_content})6.4 本质探索:技能即模式
技能系统本质上是一种模式匹配和注入机制。它基于当前任务的特征,识别需要的专业知识,并将其注入到代理的上下文中,使得:
- 代理可以动态适应不同领域的任务
- 专业知识可以独立于核心代理进行更新
- 不同任务可以使用不同的专业知识组合
6.5 当前问题与改进方向
问题1:技能识别
代理可能无法准确判断当前任务需要哪些技能。
改进方向:智能技能推荐。基于任务描述和历史数据,自动推荐最相关的技能。
问题2:技能冲突
多个技能可能提供相互矛盾的指导。
改进方向:技能优先级和冲突解决机制。定义技能之间的优先级关系,当冲突发生时进行仲裁。
第七章:系统提示——Prompt
7.1 是什么:代理的"操作系统"
系统提示是代理的"宪法"——它定义了代理的身份、能力、限制和行为准则。与技能不同,系统提示是持久的、基础的,不会随任务变化而改变。
7.2 为什么:行为的可预测性
没有系统提示的代理就像没有规则的组织——行为不可预测,难以控制。系统提示解决的核心问题是:
- 身份定义:告诉模型它是谁,能做什么
- 行为约束:定义什么可以做,什么不可以做
- 输出格式:规定响应的结构和风格
7.3 怎么设计:运行时组装模式
现代Harness工程通常采用运行时组装模式来构建系统提示:
def build_system_prompt(context):
prompt_parts = []
# 1. 基础身份
prompt_parts.append("你是一个专业的编程助手...")
# 2. 能力声明
prompt_parts.append("你可以读取和修改文件、执行命令...")
# 3. 行为约束
prompt_parts.append("你必须始终遵循安全最佳实践...")
# 4. 动态上下文
if context.current_project:
prompt_parts.append(f"当前项目:{context.current_project}")
# 5. 加载的技能
for skill in context.loaded_skills:
prompt_parts.append(skill)
return "\n\n".join(prompt_parts)7.4 本质探索:提示即代码
系统提示本质上是一种声明式编程语言。它通过自然语言描述代理应该具有的行为,而不是通过代码实现这些行为,使得:
- 非技术人员也可以定义代理行为
- 行为可以快速迭代和调整
- 不同场景可以使用不同的行为定义
7.5 当前问题与改进方向
问题1:提示工程
编写有效的系统提示需要大量经验和试错。
改进方向:提示生成和优化工具。通过自动化测试和优化,生成更有效的系统提示。
问题2:提示膨胀
随着功能增加,系统提示可能变得非常长,影响模型性能。
改进方向:提示压缩和分层。将系统提示分为核心层和扩展层,根据需要动态加载。
第八章:上下文管理——Context
8.1 是什么:代理的"短期记忆"
上下文是代理在单次会话中的"工作记忆"。它包含了当前对话的所有历史消息、工具调用结果和中间状态。
8.2 为什么:上下文窗口的限制
语言模型的上下文窗口是有限的(通常为128K-200K tokens)。当对话历史超过这个限制时,代理会"遗忘"早期的信息。上下文管理解决的核心问题是:
- 如何在有限的窗口内保留最重要的信息
- 如何在对话过长时进行压缩
- 如何在需要时恢复丢失的上下文
8.3 怎么设计:压缩与检索模式
上下文管理通常采用压缩与检索模式:
class ContextManager:
def __init__(self, max_tokens=100000):
self.max_tokens = max_tokens
self.full_history = []
self.compressed_history = []
def add_message(self, message):
self.full_history.append(message)
# 检查是否需要压缩
if self.total_tokens() > self.max_tokens:
self.compress()
def compress(self):
# 将旧消息压缩为摘要
old_messages = self.full_history[:len(self.full_history)//2]
summary = self.summarize(old_messages)
# 保留最近的消息和摘要
self.compressed_history.append({"role": "system", "content": summary})
self.full_history = self.full_history[len(old_messages):]
def get_context(self):
return self.compressed_history + self.full_history8.4 本质探索:上下文即缓存
上下文本质上是一种缓存系统。它缓存了代理的"思考过程",使得:
- 代理可以引用之前的工作:不需要重新计算
- 代理可以保持对话的连贯性:用户感觉在和同一个人对话
- 代理可以从错误中学习:记住什么方法有效,什么无效
8.5 当前问题与改进方向
问题1:信息丢失
压缩过程不可避免地会丢失一些信息,可能导致代理遗忘重要细节。
改进方向:智能压缩。使用更先进的摘要算法,保留关键信息的同时减少token数量。
问题2:检索效率
当上下文很长时,检索特定信息可能很慢。
改进方向:向量检索。将上下文转换为向量表示,通过语义搜索快速定位相关信息。
第九章:记忆系统——Memory
9.1 是什么:代理的"长期记忆"
记忆系统是代理的持久化存储,用于保存跨越会话的重要信息。与上下文不同,记忆不会随会话结束而消失。
9.2 为什么:学习与个性化
没有记忆的代理每次会话都从零开始,无法:
- 学习用户偏好:记住用户喜欢的代码风格、工具等
- 积累项目知识:记住项目的结构、约定、常见问题
- 避免重复错误:记住之前尝试过但失败的方法
9.3 怎么设计:分层存储模式
记忆系统通常采用分层存储模式:
class MemorySystem:
def __init__(self):
self.short_term = {} # 会话内记忆
self.long_term = {} # 跨会话记忆
self.episodic = [] # 事件记忆
def store(self, key, value, importance="medium"):
if importance == "high":
# 存储到长期记忆
self.long_term[key] = {
"value": value,
"timestamp": datetime.now(),
"access_count": 0
}
else:
# 存储到短期记忆
self.short_term[key] = value
def recall(self, key):
# 先查短期记忆
if key in self.short_term:
return self.short_term[key]
# 再查长期记忆
if key in self.long_term:
self.long_term[key]["access_count"] += 1
return self.long_term[key]["value"]
return None
def consolidate(self):
# 将频繁访问的短期记忆转移到长期记忆
for key, value in self.short_term.items():
if self.is_important(key):
self.store(key, value, importance="high")9.4 本质探索:记忆即数据库
记忆系统本质上是一个智能数据库。它不仅存储数据,还:
- 自动判断信息的重要性
- 根据访问模式优化存储
- 提供语义检索能力
9.5 当前问题与改进方向
问题1:记忆冲突
新信息可能与旧信息矛盾,导致记忆不一致。
改进方向:记忆版本控制。保留信息的历史版本,当冲突发生时进行仲裁。
问题2:记忆检索
当记忆量很大时,如何快速找到相关信息是一个挑战。
改进方向:语义索引。使用向量数据库存储记忆,通过语义相似性进行检索。
第十章:错误恢复——Error Recovery
10.1 是什么:代理的"免疫系统"
错误恢复系统是代理处理异常情况的能力。它确保代理在遇到错误时不会崩溃,而是能够优雅地恢复并继续工作。
10.2 为什么:现实世界的复杂性
在真实环境中,各种错误都可能发生:
- 工具执行失败:文件不存在、权限不足、网络超时
- 模型输出错误:格式错误、逻辑错误、幻觉
- 外部依赖故障:API不可用、服务宕机
没有错误恢复的代理就像没有免疫系统的身体——遇到第一个病原体就会倒下。
10.3 怎么设计:分类与重试模式
错误恢复通常采用分类与重试模式:
class ErrorRecovery:
def __init__(self):
self.max_retries = 3
self.retry_strategies = {
"transient": self.retry_with_backoff,
"permanent": self.abort_and_report,
"recoverable": self.try_alternative
}
def handle_error(self, error, context):
# 错误分类
error_type = self.classify_error(error)
# 选择恢复策略
strategy = self.retry_strategies[error_type]
# 执行恢复
return strategy(error, context)
def retry_with_backoff(self, error, context):
# 指数退避重试
for attempt in range(self.max_retries):
time.sleep(2 ** attempt)
try:
return self.retry_operation(context)
except Exception as e:
continue
raise MaxRetriesExceeded()
def try_alternative(self, error, context):
# 尝试替代方案
alternatives = self.get_alternatives(context)
for alt in alternatives:
try:
return alt.execute()
except Exception:
continue
raise NoAlternativesAvailable()10.4 本质探索:错误即信息
错误恢复系统本质上是一个学习系统。它将错误视为宝贵的信息来源,用于:
- 识别系统薄弱点:哪些工具最容易失败?
- 优化重试策略:什么类型的错误值得重试?
- 改进未来决策:避免重复相同的错误
10.5 当前问题与改进方向
问题1:错误传播
一个工具的错误可能级联影响整个任务。
改进方向:错误隔离。为每个工具调用设置独立的错误边界,防止错误传播。
问题2:恢复成本
重试和恢复可能消耗大量时间和资源。
改进方向:智能重试决策。根据错误类型、历史成功率和剩余预算,决定是否值得重试。
第十一章:子代理——Subagent
11.1 是什么:代理的"分身"
子代理系统允许主代理将任务委托给专门的子代理。每个子代理都有独立的上下文和工具集,专注于完成特定的子任务。
11.2 为什么:复杂性的分解
当任务变得复杂时,单一代理可能面临:
- 上下文过载:所有信息塞进一个窗口
- 职责混乱:一个代理处理太多不同类型的任务
- 效率低下:串行处理可以并行的子任务
子代理通过分而治之解决了这些问题。
11.3 怎么设计:委托与聚合模式
子代理系统通常采用委托与聚合模式:
class MainAgent:
def __init__(self):
self.subagent_pool = []
def delegate_task(self, task):
# 创建专门的子代理
subagent = self.create_subagent(task)
# 分配任务
result = subagent.execute(task)
# 聚合结果
return self.aggregate_results(result)
def create_subagent(self, task):
# 根据任务类型创建合适的子代理
if task.type == "code_review":
return CodeReviewSubagent()
elif task.type == "testing":
return TestingSubagent()
else:
return GeneralSubagent()
class Subagent:
def __init__(self):
self.context = [] # 独立的上下文
self.tools = [] # 专门的工具集
def execute(self, task):
# 在独立上下文中执行任务
messages = [{"role": "user", "content": task.description}]
while True:
response = self.model.generate(messages)
if response.has_tool_call:
result = self.execute_tool(response.tool_call)
messages.append(result)
else:
return response.content11.4 本质探索:子代理即微服务
子代理系统本质上是一种微服务架构。每个子代理都是一个独立的"服务",具有:
- 单一职责:只处理特定类型的任务
- 独立上下文:不会被其他任务的信息干扰
- 可组合性:多个子代理可以协作完成复杂任务
11.5 当前问题与改进方向
问题1:通信开销
主代理和子代理之间的通信需要传递大量上下文。
改进方向:共享记忆。使用共享的记忆空间,减少通信开销。
问题2:协调复杂性
当多个子代理并行工作时,协调它们的活动变得复杂。
改进方向:协调协议。定义明确的协调协议,确保子代理之间的有效协作。
第十二章:任务系统——Task
12.1 是什么:代理的"项目管理"
任务系统将大目标分解为结构化的任务图。它不仅是待办事项的集合,更是一个工作流引擎,管理任务之间的依赖、优先级和执行顺序。
12.2 为什么:从目标到执行的桥梁
用户给出的通常是高层目标("修复这个bug"、"添加这个功能"),而不是具体的执行步骤。任务系统解决的核心问题是:
- 目标分解:将模糊的目标转化为具体的步骤
- 依赖管理:确定哪些步骤可以并行,哪些必须串行
- 进度追踪:知道整体进展,预测完成时间
12.3 怎么设计:有向无环图模式
任务系统通常采用有向无环图(DAG)模式:
class TaskGraph:
def __init__(self):
self.tasks = {}
self.dependencies = {}
def add_task(self, task_id, description, dependencies=[]):
self.tasks[task_id] = {
"description": description,
"status": "pending",
"result": None
}
self.dependencies[task_id] = dependencies
def get_executable_tasks(self):
# 返回所有依赖已满足的任务
executable = []
for task_id, deps in self.dependencies.items():
if self.tasks[task_id]["status"] == "pending":
if all(self.tasks[dep]["status"] == "completed" for dep in deps):
executable.append(task_id)
return executable
def mark_completed(self, task_id, result):
self.tasks[task_id]["status"] = "completed"
self.tasks[task_id]["result"] = result
def is_complete(self):
return all(t["status"] == "completed" for t in self.tasks.values())12.4 本质探索:任务即工作流
任务系统本质上是一个工作流引擎。它将执行逻辑从任务定义中分离出来,使得:
- 任务可以被可视化:任务图可以直观地展示工作进展
- 任务可以被优化:通过分析依赖关系,找到最优的执行顺序
- 任务可以被恢复:当某个任务失败时,可以从断点继续
12.5 当前问题与改进方向
问题1:任务粒度
任务分解的粒度很难把握——太细会导致管理开销,太粗会失去并行性。
改进方向:自适应粒度。根据任务的复杂度和依赖关系,动态调整任务粒度。
问题2:动态调整
当执行过程中发现新信息时,可能需要调整任务图。
改进方向:动态任务图。允许在执行过程中添加、删除或修改任务。
第十三章:后台任务——Background Tasks
13.1 是什么:代理的"异步能力"
后台任务系统允许代理将耗时操作放到后台执行,同时继续处理其他工作。它是代理实现非阻塞操作的关键机制。
13.2 为什么:效率与响应性
某些操作可能需要很长时间(编译代码、运行测试、下载文件)。如果代理必须等待这些操作完成,会导致:
- 响应延迟:用户等待时间过长
- 资源浪费:代理在等待期间无法做其他工作
- 用户体验差:感觉代理"卡住"了
13.3 怎么设计:异步执行模式
后台任务系统通常采用异步执行模式:
class BackgroundTaskManager:
def __init__(self):
self.tasks = {}
self.task_queue = queue.Queue()
self.workers = []
def submit_task(self, task_id, task_func, callback):
# 提交任务到后台
self.tasks[task_id] = {
"status": "running",
"result": None,
"callback": callback
}
# 异步执行
thread = threading.Thread(
target=self._execute_task,
args=(task_id, task_func)
)
thread.start()
def _execute_task(self, task_id, task_func):
try:
result = task_func()
self.tasks[task_id]["status"] = "completed"
self.tasks[task_id]["result"] = result
# 调用回调函数
callback = self.tasks[task_id]["callback"]
if callback:
callback(result)
except Exception as e:
self.tasks[task_id]["status"] = "failed"
self.tasks[task_id]["error"] = str(e)
def get_task_status(self, task_id):
return self.tasks[task_id]["status"]
def get_task_result(self, task_id):
task = self.tasks[task_id]
if task["status"] == "completed":
return task["result"]
return None13.4 本质探索:后台即并发
后台任务系统本质上是一个并发执行框架。它允许代理:
- 同时处理多个任务:提高资源利用率
- 保持响应性:用户不会感觉代理"卡住"
- 实现流水线:一个任务的输出可以作为另一个任务的输入
13.5 当前问题与改进方向
问题1:任务监控
后台任务可能失败,但代理可能不知道。
改进方向:主动监控。定期检查后台任务状态,当任务失败时及时通知用户。
问题2:资源管理
过多的后台任务可能耗尽系统资源。
改进方向:资源限制。设置最大并发任务数,防止系统过载。
第十四章:定时调度——Cron Scheduler
14.1 是什么:代理的"闹钟"
定时调度系统允许代理按计划执行任务。它是代理实现自动化和周期性工作的关键机制。
14.2 为什么:从被动到主动
没有定时调度的代理只能被动响应用户请求。定时调度使代理能够:
- 主动执行任务:不需要用户每次手动触发
- 维护系统健康:定期检查、清理、备份
- 响应时间事件:在特定时间执行特定操作
14.3 怎么设计:调度器模式
定时调度系统通常采用调度器模式:
class CronScheduler:
def __init__(self):
self.jobs = {}
self.running = False
def add_job(self, job_id, schedule, task_func, args=[]):
self.jobs[job_id] = {
"schedule": schedule,
"task": task_func,
"args": args,
"last_run": None,
"next_run": self.calculate_next_run(schedule)
}
def start(self):
self.running = True
while self.running:
now = datetime.now()
# 检查是否有任务需要执行
for job_id, job in self.jobs.items():
if now >= job["next_run"]:
# 执行任务
self.execute_job(job_id)
# 更新下次执行时间
job["last_run"] = now
job["next_run"] = self.calculate_next_run(job["schedule"])
# 等待一段时间再检查
time.sleep(1)
def execute_job(self, job_id):
job = self.jobs[job_id]
try:
job["task"](*job["args"])
except Exception as e:
self.handle_job_error(job_id, e)14.4 本质探索:调度即时间管理
定时调度系统本质上是一个时间管理系统。它将时间维度引入代理的行为,使得:
- 代理可以预测未来:知道什么时候需要做什么
- 代理可以优化时间:在空闲时间执行低优先级任务
- 代理可以响应时间事件:在特定时间触发特定行为
14.5 当前问题与改进方向
问题1:调度冲突
多个任务可能在同一时间触发,导致资源竞争。
改进方向:优先级调度。为任务设置优先级,高优先级任务优先执行。
问题2:调度失效
如果代理重启,所有定时任务都会丢失。
改进方向:持久化调度。将调度配置持久化存储,代理重启后自动恢复。
第十五章:多代理协作——Agent Teams
15.1 是什么:代理的"团队"
Agent Teams系统允许多个代理协同工作,共同完成复杂任务。它是Harness工程的最高级形态,将单个代理的能力扩展到团队级别。
15.2 为什么:单代理的极限
单个代理面临固有的限制:
- 上下文窗口限制:无法同时处理所有相关信息
- 专业能力限制:一个代理无法精通所有领域
- 并行性限制:单线程执行无法充分利用资源
Agent Teams通过分工协作突破了这些限制。
15.3 怎么设计:团队协议模式
Agent Teams通常采用团队协议模式:
class AgentTeam:
def __init__(self):
self.agents = {}
self.shared_memory = SharedMemory()
self.communication_protocol = CommunicationProtocol()
def add_agent(self, agent_id, agent, role):
self.agents[agent_id] = {
"agent": agent,
"role": role,
"status": "idle"
}
def assign_task(self, task):
# 分析任务,决定分配给哪个代理
suitable_agent = self.find_suitable_agent(task)
# 分配任务
self.agents[suitable_agent]["status"] = "busy"
result = self.agents[suitable_agent]["agent"].execute(task)
# 更新共享记忆
self.shared_memory.store(task.id, result)
# 通知其他代理
self.communication_protocol.broadcast(
"task_completed",
{"task_id": task.id, "result": result}
)
return result
def find_suitable_agent(self, task):
# 根据代理的角色和能力选择最合适的代理
for agent_id, agent_info in self.agents.items():
if agent_info["role"] == task.required_role:
if agent_info["status"] == "idle":
return agent_id
return None15.4 本质探索:团队即分布式系统
Agent Teams本质上是一个分布式系统。它将单个代理的复杂性分散到多个专门的代理中,使得:
- 每个代理只关注自己擅长的领域
- 多个代理可以并行处理不同的子任务
- 系统可以通过增加代理来扩展能力
15.5 当前问题与改进方向
问题1:协调开销
代理之间的通信和协调需要消耗大量资源。
改进方向:高效协议。设计更高效的通信协议,减少协调开销。
问题2:一致性保证
多个代理可能对同一问题有不同的理解,导致不一致。
改进方向:共识机制。引入类似区块链的共识机制,确保团队决策的一致性。
第十六章:综合代理——Comprehensive Agent Turn
16.1 是什么:Harness工程的"最终形态"
综合代理是所有Harness工程组件的集成形态。它将Agent Loop、Tool Use、Permission、Hooks、Todo、Skill、Context、Memory、Error Recovery、Subagent、Task、Background Tasks、Cron Scheduler、Agent Teams等所有机制汇聚在一个循环中,构成一个完整的生产级代理。
16.2 为什么:从组件到系统
单独的组件就像汽车的零件——每个都有其功能,但只有组装在一起才能成为一辆可用的汽车。综合代理解决的核心问题是:
- 组件集成:确保所有组件能够协同工作
- 行为一致性:不同组件的行为不会相互冲突
- 性能优化:整体性能优于各部分之和
16.3 怎么设计:分层架构模式
综合代理通常采用分层架构模式:
class ComprehensiveAgent:
def __init__(self):
# 核心层
self.agent_loop = AgentLoop()
# 工具层
self.tool_registry = ToolRegistry()
self.permission_manager = PermissionManager()
# 规划层
self.todo_manager = TodoManager()
self.task_graph = TaskGraph()
self.skill_loader = SkillLoader()
# 记忆层
self.context_manager = ContextManager()
self.memory_system = MemorySystem()
# 执行层
self.background_task_manager = BackgroundTaskManager()
self.cron_scheduler = CronScheduler()
# 协作层
self.subagent_manager = SubagentManager()
self.agent_team = AgentTeam()
# 横切关注点
self.hook_system = HookSystem()
self.error_recovery = ErrorRecovery()
def process_request(self, user_request):
# 1. 构建系统提示
system_prompt = self.build_system_prompt()
# 2. 加载相关技能
relevant_skills = self.skill_loader.identify_skills(user_request)
# 3. 创建任务计划
task_plan = self.create_task_plan(user_request)
# 4. 执行代理循环
while not task_plan.is_complete():
# 获取下一个任务
task = task_plan.get_next_task()
# 决定是自己处理还是委托
if self.should_delegate(task):
result = self.delegate_to_subagent(task)
else:
result = self.execute_task(task)
# 更新任务状态
task_plan.mark_completed(task, result)
# 更新记忆
self.memory_system.store(task.id, result)
# 5. 聚合结果
final_result = self.aggregate_results(task_plan)
return final_result16.4 本质探索:综合代理即操作系统
综合代理本质上是一个操作系统。它管理代理的所有资源(上下文、记忆、工具、时间),提供所有服务(规划、执行、协作、恢复),使得:
- 代理可以专注于思考和决策:不需要关心底层细节
- 系统可以保证可靠性和安全性:通过权限和错误恢复机制
- 能力可以水平扩展:通过子代理和团队机制
16.5 当前问题与改进方向
问题1:复杂性管理
随着组件增多,系统的复杂性呈指数增长。
改进方向:模块化设计。将系统分解为独立的模块,每个模块只关注单一职责。
问题2:性能瓶颈
多个组件的交互可能产生性能瓶颈。
改进方向:性能分析和优化。通过性能分析工具识别瓶颈,进行针对性优化。
第十七章:当前问题与未来改进
17.1 Coding Agent的特殊挑战
Coding Agent(编程代理)是Harness工程最典型的应用场景,也是挑战最大的领域:
问题1:代码理解的深度
当前代理对代码的理解往往是表面的——能够识别语法结构,但难以理解深层的语义和设计意图。
改进方向:
- 程序分析技术:引入静态分析、数据流分析等传统编译器技术
- 知识图谱:构建代码知识图谱,捕获代码之间的复杂关系
- 多模态理解:结合代码、注释、文档、测试等多种信息源
问题2:长期记忆的局限
编程任务往往需要理解项目的整体架构和历史决策,但当前代理的记忆能力有限。
改进方向:
- 项目记忆:为每个项目维护专门的记忆空间
- 决策追溯:记录每个设计决策的原因和上下文
- 知识演化:跟踪代码库的演化,理解变化的原因
问题3:协作能力的不足
大型软件项目需要多人协作,但当前代理难以与人类开发者有效协作。
改进方向:
- 人机协作协议:定义清晰的人机协作流程
- 意图理解:更好地理解开发者的意图和偏好
- 冲突解决:当代理的建议与开发者的意见冲突时,如何处理
17.2 通用Agent的改进方向
方向1:自主性的提升
当前代理仍然是"被动"的——需要用户触发才能行动。未来的代理应该能够:
- 主动发现问题:监控系统状态,主动识别潜在问题
- 主动提出建议:基于观察和经验,主动提出改进建议
- 主动学习:从每次交互中学习,不断改进自己的能力
方向2:可解释性的增强
当前代理的决策过程往往是"黑盒"的。未来的代理应该能够:
- 解释决策原因:清楚地说明为什么做出某个决策
- 提供替代方案:展示其他可能的选择及其权衡
- 支持审计追踪:记录完整的决策过程,便于事后审查
方向3:安全性的强化
随着代理能力的增强,安全性变得越来越重要。未来的代理应该能够:
- 风险评估:在执行操作前评估潜在风险
- 沙箱执行:在隔离环境中执行高风险操作
- 异常检测:识别并阻止异常或恶意行为
17.3 技术演进趋势
趋势1:从规则到学习
当前的Harness工程大量依赖规则(权限规则、错误处理规则、调度规则等)。未来的趋势是:
- 学习型规则:通过机器学习自动发现和优化规则
- 自适应策略:根据环境和历史数据动态调整策略
- 预测性干预:预测可能的问题,提前采取预防措施
趋势2:从单体到分布式
当前的代理通常是单体的——所有功能集中在一个进程中。未来的趋势是:
- 微服务架构:将代理分解为独立的微服务
- 边缘计算:将部分计算任务下放到边缘设备
- 云原生部署:利用云原生技术实现弹性伸缩
趋势3:从工具到伙伴
当前的代理更多是"工具"——被动执行用户指令。未来的趋势是:
- 伙伴关系:代理成为人类的合作伙伴,共同解决问题
- 情感理解:理解用户的情感状态,提供更人性化的交互
- 价值对齐:确保代理的行为与人类的价值观一致
第十八章:设计哲学与最佳实践
18.1 核心设计原则
原则1:简单性
Harness工程的核心思想是"一个循环,周围围绕着子系统"。这种设计的优雅之处在于:
- 核心循环保持简单:只做最基本的事情
- 复杂性被推到外围:通过子系统处理复杂情况
- 易于理解和维护:开发者可以快速理解系统的工作原理
原则2:渐进性
Learn Claude Code项目展示了渐进式设计的力量:
- 从最小可行产品开始:102行代码的最小循环
- 逐步添加功能:每次只添加一个子系统
- 保持每一步的可用性:每个阶段都是可工作的系统
原则3:可组合性
每个子系统都是独立的,可以自由组合:
- 模块化设计:每个子系统只关注单一职责
- 标准接口:子系统之间通过标准接口通信
- 插件化扩展:新的子系统可以通过插件方式添加
18.2 实施最佳实践
实践1:测试驱动开发
Harness工程应该采用测试驱动开发:
- 单元测试:每个子系统都应该有完整的单元测试
- 集成测试:测试子系统之间的交互
- 端到端测试:测试完整的代理行为
实践2:监控与可观测性
生产环境中的代理需要完善的监控:
- 性能指标:响应时间、吞吐量、错误率
- 业务指标:任务完成率、用户满意度
- 安全指标:权限拒绝率、异常行为检测
实践3:渐进式部署
代理系统应该采用渐进式部署策略:
- 灰度发布:先在小范围内测试
- A/B测试:比较新旧版本的性能
- 回滚机制:当出现问题时能够快速回滚
18.3 常见陷阱与规避方法
陷阱1:过度工程化
不要过早优化,不要添加不需要的功能。
规避方法:遵循YAGNI原则(You Aren't Gonna Need It)
陷阱2:忽视错误处理
错误处理是Harness工程中最容易被忽视的部分。
规避方法:将错误处理作为一等公民,从设计阶段就考虑
陷阱3:性能瓶颈
代理系统可能面临各种性能瓶颈。
规避方法:从设计阶段就考虑性能,定期进行性能测试和优化
结论:Harness工程的未来
Harness工程代表了AI Agent从实验室走向生产环境的关键一步。它不是一个单一的技术,而是一套完整的工程方法论,涵盖了代理的整个生命周期:从设计、开发、测试、部署到运维。
通过Learn Claude Code项目的学习,我们可以看到:
- Harness工程的核心是简单性:一个循环,周围围绕着子系统
- Harness工程的关键是渐进性:从最小可行产品开始,逐步添加功能
- Harness工程的价值是可靠性:将不稳定的AI模型转化为可靠的生产系统
未来的Harness工程将朝着以下方向发展:
- 更智能:通过机器学习自动优化代理的行为
- 更自主:代理能够主动发现问题和机会
- 更协作:代理成为人类的真正合作伙伴
- 更安全:确保代理的行为符合人类的价值观
Harness工程不仅是技术问题,更是哲学问题:我们如何构建一个既能发挥AI强大能力,又能被人类有效控制的系统? 这个问题的答案,将决定AI Agent在未来的角色和影响。
作为开发者,我们正处于一个激动人心的时代。Harness工程为我们提供了工具和方法,让我们能够构建真正有用的AI Agent。让我们拥抱这个挑战,共同塑造AI的未来。
附录:Harness工程检查清单
在构建自己的Harness工程时,可以参考以下检查清单:
核心循环
工具系统
权限控制
钩子系统
计划管理
技能加载
系统提示
上下文管理
记忆系统
错误恢复
子代理
任务系统
后台任务
定时调度
多代理协作
通过遵循这个检查清单,你可以构建一个完整、可靠、可扩展的Harness工程,为AI Agent提供坚实的基础设施。
Prompt
写一篇关于AI Harness工程的文章,参考Learn Claude Code项目,从Agent Loop、Prompt、Tool Use、MCP、Permission、Hooks、Todo、Skill、Context、Memory、Error Recovery、Subagent、Task、Background Tasks、Cron Scheduler、Agent Teams等方面展开,要求深入浅出、切中要害、前后串连,先将明白是什么、为什么、怎么设计,再根据目前最新研究深入探索其本质,并讲明目前Agent尤其是Coding Agent在这方面的问题以及后续会如何改进。全文不限字数,力求一篇文章讲明讲透Harness工程。