什么是驾驭工程？
定义：驾驭工程是围绕 AI Agent 设计约束、反馈循环、工具系统和验证机制的工程学科。
这个定义听起来很学术，让我们用一个生动的比喻来理解：
驾驭一匹千里马：千里马（AI Agent）拥有强大的奔跑能力，但如果没有人驾驭，它可能会随意奔跑、撞伤路人、甚至冲悬崖。驾驭工程就是为这匹千里马配备缰绳（约束）、刹车（安全控制）、马鞭（激励机制）和骑手（监控），让它在正确的道路上安全前行。
核心理念：Human Steer, Agent Execute
- Human Steer：人类设定目标、监控过程、最终决策
- Agent Execute：AI 在约束框架内自主执行
这个理念解决了 Context Engineering 无法解决的根本问题：AI 有了知识，但没有行为的约束和验证。
起源与发展
Mitchell Hashimoto 的开创性贡献
2026 年 2 月 5 日，HashiCorp 联合创始人 Mitchell Hashimoto 正式提出了"驾驭工程"这个概念。需要特别注意的是：是 Mitchell Hashimoto（HashiCorp 联合创始人），而不是 CTO。
Mitchell 在其博客中写道：
“驾驭工程不是限制 AI 的能力，而是确保 AI 的能力得到负责任的发挥。就像给赛车手配备安全装备，不是限制他跑得快，而是确保他能安全地快跑。”
OpenAI 的技术推动
2026 年 2 月 11 日，OpenAI 发表了《Harness engineering: leveraging Codex》，从技术角度阐述了驾驭工程的重要性：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| # OpenAI 的驾驭工程示例
class CodexHarness:
def __init__(self, codex_model, safety_constraints):
self.model = codex_model
self.constraints = safety_constraints
self.execution_monitor = ExecutionMonitor()
def safe_execute(self, task):
# 1. 任务安全性验证
if not self.validate_task(task):
return "Task violates safety constraints"
# 2. 执行过程监控
with self.execution_monitor:
result = self.model.execute(task)
# 3. 结果验证
if not self.validate_result(result):
return "Execution result validation failed"
return result
|
王欣的理论贡献
技术专家王欣在其个人网站 wangxin.io 上发表了多篇关于驾驭工程的文章，从理论层面完善了这个概念。他提出：
“驾驭工程的核心是’可控性’。AI Agent 再强大，如果不能被人类控制和验证，就不能被用于生产环境。”
驾驭工程的公式
Mitchell Hashimoto 提出了一个简洁的公式：
Agent = LLM + Harness
这个公式揭示了现代 AI Agent 的本质：
- LLM：提供认知能力和执行能力
- Harness：提供安全控制和约束框架
四大核心子系统
驾驭工程包含四个相互关联的核心子系统，它们共同构成了 AI Agent 的安全框架。
负责管理 AI Agent 可以调用的工具，确保工具调用的安全性和可控性。
核心组件：
- 函数调用协议：标准化的工具调用接口
- 工具注册表：可调用工具的元数据管理
- 权限控制：基于角色的工具访问控制
- 沙箱隔离：工具执行的隔离环境
技术实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| class ToolInjectionSystem:
def __init__(self):
self.tool_registry = ToolRegistry()
self.permission_manager = PermissionManager()
self.sandbox = ExecutionSandbox()
def inject_tools(self, agent, required_tools):
"""安全地注入工具到 Agent"""
# 1. 工具权限验证
for tool in required_tools:
if not self.permission_manager.can_use(agent, tool):
raise PermissionError(f"Agent {agent} cannot use tool {tool}")
# 2. 工具注册和验证
validated_tools = []
for tool in required_tools:
tool_info = self.tool_registry.get_tool(tool)
if self.validate_tool(tool_info):
validated_tools.append(tool_info)
# 3. 沙箱环境配置
sandbox_config = self.sandbox.create_config(validated_tools)
return {
'tools': validated_tools,
'sandbox': sandbox_config,
'permissions': self.permission_manager.get_permissions(agent)
}
|
Claude Code 实践案例：
Claude Code（2025年9月发布）是工具注入系统的优秀实践：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| # Claude Code 的工具注入机制
claude_code_config = {
'allowed_tools': [
'file_read',
'file_write',
'bash_execute',
'web_search',
'git_operations'
],
'restricted_directories': [
'/system',
'/private',
'/config'
],
'execution_timeout': 30, # 秒
'max_file_size': '10MB'
}
|
2. 状态管理系统（State Management）
负责跟踪和管理 AI Agent 的执行状态，确保任务的可追溯性和可恢复性。
核心组件：
- 任务进度跟踪：实时监控任务执行状态
- 状态持久化：将中间状态保存到数据库
- 中断恢复：从中断点继续执行
- 并发隔离：防止多个任务相互干扰
技术实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
| class StateManager:
def __init__(self, storage_backend):
self.storage = storage_backend
self.state_cache = {}
def create_task_state(self, task_id, initial_state):
"""创建任务状态"""
state = {
'task_id': task_id,
'status': 'initialized',
'progress': 0.0,
'subtasks': [],
'created_at': datetime.now(),
'last_updated': datetime.now(),
'data': initial_state
}
self.storage.save_state(task_id, state)
return state
def update_progress(self, task_id, progress, subtask=None):
"""更新任务进度"""
state = self.storage.get_state(task_id)
state['progress'] = progress
state['last_updated'] = datetime.now()
if subtask:
state['subtasks'].append({
'name': subtask,
'completed_at': datetime.now()
})
self.storage.save_state(task_id, state)
return state
def save_checkpoint(self, task_id, checkpoint_data):
"""保存检查点"""
state = self.storage.get_state(task_id)
state['checkpoint'] = {
'data': checkpoint_data,
'saved_at': datetime.now()
}
self.storage.save_state(task_id, state)
def restore_from_checkpoint(self, task_id):
"""从检查点恢复"""
state = self.storage.get_state(task_id)
if 'checkpoint' in state:
return state['checkpoint']['data']
return None
|
并发控制策略：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| class ConcurrencyManager:
def __init__(self, max_concurrent_tasks=5):
self.max_concurrent = max_concurrent_tasks
self.active_tasks = {}
self.task_locks = {}
def execute_task(self, task):
"""执行任务，处理并发"""
# 1. 检查并发限制
if len(self.active_tasks) >= self.max_concurrent:
raise ConcurrencyError(f"Maximum {self.max_concurrent} tasks allowed")
# 2. 创建任务锁
task_lock = asyncio.Lock()
self.task_locks[task.id] = task_lock
try:
# 3. 执行任务
async with task_lock:
result = await task.execute()
self.active_tasks[task.id] = result
return result
finally:
# 4. 清理资源
del self.task_locks[task.id]
if task.id in self.active_tasks:
del self.active_tasks[task.id]
|
3. 验证循环系统（Verification Loop）
负责验证 AI Agent 的执行结果，确保输出的质量和安全性。
核心组件：
- 输出质量检查：验证生成内容的质量
- 错误检测：识别执行中的异常
- 自动修正：对错误进行自动修正
- 人工审核触发：复杂情况下的人工介入
技术实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
| class VerificationLoop:
def __init__(self):
self.checkers = [
QualityChecker(),
SafetyChecker(),
FormatChecker(),
CompletenessChecker()
]
self.error_handlers = [
AutoFixHandler(),
HumanReviewHandler(),
LogHandler()
]
async def verify_execution(self, task, result):
"""验证执行结果"""
verification_results = {}
# 1. 执行所有检查
for checker in self.checkers:
try:
check_result = await checker.check(result)
verification_results[checker.name] = check_result
except Exception as e:
verification_results[checker.name] = {
'passed': False,
'error': str(e)
}
# 2. 处理检查结果
overall_passed = all(r['passed'] for r in verification_results.values())
if not overall_passed:
# 3. 错误处理
error_summary = self.summarize_errors(verification_results)
corrected_result = await self.handle_errors(task, result, error_summary)
return corrected_result
return result
def summarize_errors(self, verification_results):
"""总结错误信息"""
errors = []
for checker_name, result in verification_results.items():
if not result['passed']:
errors.append({
'checker': checker_name,
'issue': result.get('issue', 'Unknown issue'),
'severity': result.get('severity', 'medium')
})
return errors
async def handle_errors(self, task, result, errors):
"""处理错误"""
for error in errors:
# 按严重程度处理
if error['severity'] == 'critical':
# 触发人工审核
await self.trigger_human_review(task, result, error)
elif error['severity'] == 'high':
# 自动修正
await self.auto_fix(task, result, error)
else:
# 记录日志
await self.log_error(task, result, error)
return result
|
LangGraph 工作流编排：
LangGraph（2024年1月发布）为验证循环提供了强大的工作流编排能力：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| # LangGraph 验证循环示例
verification_workflow = {
'nodes': [
{'id': 'input', 'type': 'input'},
{'id': 'quality_check', 'type': 'quality_check'},
{'id': 'safety_check', 'type': 'safety_check'},
{'id': 'format_check', 'type': 'format_check'},
{'id': 'auto_fix', 'type': 'auto_fix'},
{'id': 'human_review', 'type': 'human_review'},
{'id': 'output', 'type': 'output'}
],
'edges': [
{'from': 'input', 'to': 'quality_check'},
{'from': 'quality_check', 'to': 'safety_check'},
{'from': 'safety_check', 'to': 'format_check'},
{'from': 'format_check', 'to': 'auto_fix', 'condition': 'needs_fix'},
{'from': 'format_check', 'to': 'human_review', 'condition': 'needs_review'},
{'from': 'auto_fix', 'to': 'output'},
{'from': 'human_review', 'to': 'output'}
]
}
|
4. 约束分层系统（Guardrails）
负责为 AI Agent 设置多层次的约束，确保行为的合规性和安全性。
核心组件：
- 规则引擎：可配置的业务规则
- 格式强制：输出格式的要求
- 安全策略：安全行为的约束
- 合规审计：合规性的检查和记录
技术实现：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
| class GuardrailSystem:
def __init__(self):
self.rule_engine = RuleEngine()
self.format_enforcer = FormatEnforcer()
self.security_policy = SecurityPolicy()
self.compliance_auditor = ComplianceAuditor()
def apply_constraints(self, task, input_data):
"""应用约束系统"""
# 1. 规则检查
rule_violations = self.rule_engine.check_rules(task, input_data)
if rule_violations:
raise RuleViolationError(rule_violations)
# 2. 格式强制
formatted_input = self.format_enforcer.enforce_format(input_data)
# 3. 安全策略检查
security_check = self.security_policy.check_security(task, formatted_input)
if not security_check['passed']:
raise SecurityError(security_check['issues'])
# 4. 合规审计
self.compliance_auditor.log_audit_event(task, formatted_input)
return formatted_input
def create_rule_set(self, domain):
"""创建特定领域的规则集"""
rule_sets = {
'finance': [
{'rule': 'no_financial_data_leakage', 'severity': 'critical'},
{'rule': 'require_approval_for_large_transactions', 'severity': 'high'}
],
'healthcare': [
{'rule': 'hipaa_compliance', 'severity': 'critical'},
{'rule': 'patient_data_anonymization', 'severity': 'high'}
],
'general': [
{'rule': 'no_harmful_content', 'severity': 'high'},
{'rule': 'respect_privacy', 'severity': 'medium'}
]
}
return rule_sets.get(domain, rule_sets['general'])
|
设计原则
前馈控制（Feedforward Control）
在执行前设置约束和预期，防患于未然：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
| class FeedforwardController:
def __init__(self):
self.constraints = []
self.expectations = []
def setup_constraints(self, task):
"""设置前馈约束"""
# 1. 权限预检
self.validate_permissions(task)
# 2. 资源预分配
self.allocate_resources(task)
# 3. 风险评估
self.assess_risks(task)
# 4. 预期设定
self.set_expectations(task)
def validate_permissions(self, task):
"""权限验证"""
required_perms = task.get_required_permissions()
for perm in required_perms:
if not self.check_permission(perm):
raise PermissionError(f"Missing permission: {perm}")
def allocate_resources(self, task):
"""资源分配"""
resources = task.get_required_resources()
for resource, amount in resources.items():
if not self.allocate(resource, amount):
raise ResourceError(f"Cannot allocate {resource}: {amount}")
|
反馈修正（Feedback Correction）
在执行后进行验证和修正，持续改进：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| class FeedbackController:
def __init__(self):
self.validators = []
self.correctors = []
def correct_execution(self, task, result):
"""执行修正"""
# 1. 结果验证
validation_result = self.validate_result(result)
if not validation_result['passed']:
# 2. 自动修正
corrected_result = self.auto_correct(result, validation_result['issues'])
# 3. 重新验证
if not self.validate_result(corrected_result)['passed']:
# 4. 人工介入
return self.human_intervention(task, corrected_result)
return corrected_result
return result
|
渐进式自动化
从人工监督到完全自动化的渐进过程：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| class ProgressiveAutomation:
def __init__(self):
self.automation_levels = {
'level_1': 'human_supervised',
'level_2': 'human_in_the_loop',
'level_3': 'autonomous_with_monitoring',
'level_4': 'fully_autonomous'
}
def get_automation_level(self, task_complexity, risk_level):
"""根据任务复杂度和风险等级确定自动化级别"""
if risk_level == 'critical':
return 'level_1'
elif task_complexity == 'simple' and risk_level == 'low':
return 'level_4'
else:
return 'level_2' # 默认有人类在环路中
|
可观测性（Observability）
确保系统的透明度和可监控性：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
| class ObservabilitySystem:
def __init__(self):
self.logger = Logger()
self.metrics = Metrics()
self.tracer = Tracer()
def log_execution(self, task, agent, result):
"""记录执行日志"""
log_entry = {
'timestamp': datetime.now(),
'task': task,
'agent': agent,
'result': result,
'duration': result.get('duration', 0),
'status': result.get('status', 'unknown')
}
self.logger.info(log_entry)
self.metrics.record_task_completion(task, result['status'])
def trace_execution(self, task_id, steps):
"""追踪执行步骤"""
trace = {
'task_id': task_id,
'steps': steps,
'start_time': datetime.now(),
'end_time': None,
'errors': []
}
return trace
|
局限与挑战
当前局限
尽管驾驭工程提供了强大的安全控制，但仍存在一些局限：
- 任务初始化的人类依赖：仍需要人类来启动任务和设定目标
- 任务间的自主性缺失：无法在多个任务之间自主切换和协调
- 固定循环结构：预设的循环结构无法动态适应新情况
- 适应性不足：对意外情况的处理能力有限
动力与未来
这些局限正是推动下一阶段工程演进的动力：
- 任务间的自主协调：AI 可以自主在多个任务之间切换
- 动态循环结构：循环结构可以根据需要动态调整
- 自适应决策：基于执行结果动态调整策略
- 完全自主：从"需要人类启动"到"自主任务管理"
这些发展将推动 AI 工程进入下一个阶段：Loop Engineering（循环工程）。
小结：驾驭的价值
驾驭工程的价值在于：让 AI 从 demo 变为可部署的生产系统。
| 特性 | 简单 demo | 驾驭工程系统 |
|---|
| 安全性 | 无安全保障 | 多层约束保护 |
| 可靠性 | 结果不可控 | 执行可验证 |
| 可维护性 | 难以维护 | 状态可追踪 |
| 可扩展性 | 难以扩展 | 模块化设计 |
驾驭工程是 AI 工程化的关键一步。它让我们不仅拥有了能回答问题的 AI，更拥有了能安全可靠地执行任务的 AI。这标志着 AI 技术从实验室走向实际应用的重要里程碑。