当前位置：首页 > news >正文

Harness Engineering：构建企业级多Agent协同系统的工程化实践

news 2026/7/5 11:34:18

🚀 30+款热门AI模型一站整合，DeepSeek/GLM/Qwen 随心用，限时 5 折。 👉 点击领海量免费额度

1. 先搞清楚 Harness Engineering 到底是什么，以及它为什么重要

如果你正在尝试用大模型构建一个能真正干活的 AI Agent，而不是一个只会聊天的玩具，那么“Harness Engineering”（缰绳工程）是你现在最需要关注的概念。它解决的核心问题是：为什么同一个大模型，在不同的工具或环境中，表现出的能力天差地别？

简单来说，Harness Engineering 认为：一个能用的 AI Agent = 大模型本身 + 围绕它构建的“缰绳”系统。这里的“缰绳”，指的是所有为了让模型能稳定、可靠、安全地完成特定任务而添加的代码、配置和执行逻辑。这包括系统提示词、工具调用、沙箱环境、状态管理、钩子函数、子Agent协调、反馈循环和错误恢复路径等一切非模型本身的东西。

一个常见的误区是，当 Agent 表现不佳时，我们总倾向于归咎于模型不够聪明，然后陷入“等下一代模型”的被动等待。Harness Engineering 的思维方式则截然不同：大多数失败不是模型问题，而是配置和工程问题。一个中等水平的模型，配上一套精心设计的“缰绳”，其表现往往能超越一个顶级模型配上糟糕的工程实现。这就像给一位经验丰富的骑手一匹普通的马，配上精良的马鞍和缰绳，其表现可能远超一位新手骑手驾驭一匹千里马。

从实战角度看，Harness Engineering 的价值在于它提供了一套可迭代、可积累的工程化方法。每一次 Agent 的失败，都不是一个需要被遗忘的“坏运气”，而是一个明确的信号，用于加固你的“缰绳”系统，确保同样的错误不再发生。这个过程就像一个“棘轮”，只进不退，让你的 Agent 系统随着使用时间越来越可靠。

2. 拆解一个企业级多Agent协同系统的核心组件

基于 Harness Engineering 的理念，要构建一个“企业级多Agent协同”项目，我们不能只盯着模型选型。我们需要从零开始，设计一套能让多个 Agent 安全、高效、可控地协同工作的基础设施。这套系统通常包含以下几个核心支柱，它们共同构成了 Agent 的“缰绳”。

2.1 沙箱：为 Agent 提供一个安全的“游乐场”

让 AI Agent 直接在你的生产服务器或开发机上运行代码，无异于打开潘多拉魔盒。沙箱是隔离执行环境的核心，它确保了 Agent 的所有操作都被限制在一个可控的范围内。

沙箱的核心职责：

环境隔离：每个 Agent 任务都在一个独立的、临时的容器或虚拟机中运行，任务结束后环境被销毁，避免残留物污染或影响后续任务。
资源限制：严格限制 CPU、内存、磁盘和网络的使用，防止单个 Agent 任务耗尽系统资源。
命令白名单：并非所有 Bash 命令都应对 Agent 开放。沙箱应只允许执行预先审核过的安全命令，并拦截rm -rf /、:(){ :|:& };:（Fork 炸弹）等危险操作。
网络隔离：默认禁止对外网络访问，或仅允许访问特定的内部 API 和资源库（如内部 PyPI、NPM 镜像站）。

实战配置示例（基于 Docker）：一个基础的沙箱配置可能如下。这里我们创建一个临时的、资源受限的 Python 环境。

# Dockerfile.sandbox FROM python:3.11-slim # 设置非 root 用户，增强安全性 RUN useradd -m -s /bin/bash agentuser USER agentuser WORKDIR /home/agentuser/workspace # 限制资源（在 docker run 时通过参数指定） # --memory=512m --cpus=1.0 # 预装一些常用且安全的工具 RUN sudo apt-get update && sudo apt-get install -y \ git \ curl \ jq \ && sudo rm -rf /var/lib/apt/lists/* # 设置默认的命令解释器，并限制其能力（可选，通过 seccomp 或 AppArmor 配置文件）

在运行时，你的 Harness 系统会动态创建并启动这个容器，将任务代码、数据文件挂载进去，执行完毕后再获取结果并销毁容器。

2.2 技能与工具：赋予 Agent “手脚”和“专业知识”

模型本身只会思考和生成文本。要让 Agent 能“做事”，必须为它提供可调用的技能和工具。这里的“技能”可以理解为更高级、更面向特定领域的工具组合或工作流。

工具与技能的设计原则：

描述清晰：每个工具的函数名、描述和参数 schema 都会进入模型的提示词。模糊的描述会导致模型错误调用。例如，“处理文件”不如“读取 JSON 文件并提取user_id字段”明确。
功能聚焦：十个功能明确的工具，远胜五十个功能重叠的工具。模型需要在有限的上下文内记住这些工具是干什么的。
安全第一：工具的实现必须进行严格的输入验证和权限检查。一个执行 SQL 查询的工具，绝不能允许拼接式查询，必须使用参数化查询。

技能示例：代码审查技能一个“代码审查”技能可能不是单个工具，而是一个由多个工具和固定流程组成的“技能包”：

调用“静态代码分析”工具（如pylint,eslint）。
调用“安全检查”工具（如bandit,npm audit）。
调用“代码复杂度分析”工具。
将上述结果汇总，调用大模型生成一份人类可读的审查报告。这个“技能包”可以被封装成一个独立的、可复用的模块，由 Planner Agent 分配给专门的 Reviewer Agent 执行。

2.3 多Agent协同与人工介入：设计工作流与管控点

单个 Agent 能力有限，复杂任务需要分解并由多个各司其职的 Agent 协作完成。同时，必须设计好人工介入的节点，确保最终控制权在人类手中。

典型的协同模式：

规划者-执行者-评审者（Planner-Executor-Reviewer）：
- Planner Agent：接收用户原始需求，进行分析和任务分解，生成一个详细的、可执行的计划（如plan.md）。它负责“做什么”和“先做什么后做什么”。
- Executor Agent(s)：根据计划，调用具体的工具和技能执行子任务。可能是多个并行的 Agent，分别负责前端、后端、数据库等。
- Reviewer Agent：对 Executor 的产出进行验证和评审。例如，检查代码是否通过测试、文档是否符合规范。Anthropic 的研究明确指出，让一个独立的 Agent 进行评审，比让生成者自我评审（Self-Evaluation）更可靠，因为生成者总会倾向于给自己的作品打高分。

人工介入的“钩子”（Hooks）：人工介入不应是随时的打断，而应被设计为工作流中明确的“审批点”或“安全阀”。这通过“钩子”实现。

提交前钩子（Pre-commit Hook）：Agent 完成代码修改后，自动运行测试、代码风格检查。如果失败，自动将错误信息反馈给 Agent 让其修正；如果成功，则生成变更摘要，等待人工确认后再执行git commit。
推送前审批（Pre-push Approval）：任何向主分支（main）推送代码的行为，必须触发一个审批流程，例如发送通知到 Slack 或生成一个待批准的合并请求（MR）。
关键操作拦截（Destructive Action Block）：在沙箱层或工具层拦截高风险命令（如git push --force,DROP TABLE），并强制要求人工输入特定授权码或进行二次确认。

3. 从零搭建一个可运行的多Agent协同Demo

理论说再多不如动手跑一遍。下面我们设计一个最小化的实战场景：让两个Agent协作，一个负责生成配置文件，另一个负责验证配置文件的合法性，最后需要人工确认才能写入磁盘。

环境准备：

Python 3.10+
OpenAI API Key（或其它兼容 OpenAI SDK 的模型 API，如 DeepSeek、通义千问）
Docker（用于沙箱环境）

项目结构：

multi_agent_harness_demo/ ├── harness_core/ # 缰绳系统核心 │ ├── sandbox.py # 沙箱管理模块 │ ├── tool_registry.py # 工具注册与管理 │ ├── agent_base.py # Agent 基类 │ └── hooks.py # 钩子函数定义 ├── skills/ # 技能定义 │ └── config_skill.py # 配置文件处理技能 ├── agents/ # 具体 Agent 定义 │ ├── planner_agent.py │ ├── generator_agent.py │ └── validator_agent.py ├── workspace/ # Agent 共享工作区 ├── main.py # 主流程入口 └── requirements.txt

步骤 1：定义核心工具和沙箱首先，我们在harness_core/tool_registry.py中定义几个最基础的工具。

# harness_core/tool_registry.py import subprocess import json from typing import Dict, Any class ToolRegistry: def __init__(self): self.tools = {} self._register_default_tools() def _register_default_tools(self): self.register_tool("execute_bash", self._execute_bash, description="在安全沙箱内执行bash命令，返回标准输出和错误。") self.register_tool("read_file", self._read_file, description="读取工作区内指定路径的文件内容。") # write_file 工具暂时不注册，因为我们需要用钩子控制写操作 # self.register_tool("write_file", self._write_file, description="向工作区内指定路径写入内容。") def register_tool(self, name: str, func: callable, description: str): self.tools[name] = {"function": func, "description": description} def get_tools_description(self) -> str: """生成供模型使用的工具描述文本""" desc = [] for name, info in self.tools.items(): desc.append(f"- {name}: {info['description']}") return "\n".join(desc) def call_tool(self, name: str, **kwargs): if name not in self.tools: raise ValueError(f"Tool '{name}' not found.") return self.tools[name]["function"](**kwargs) # --- 工具实现 --- def _execute_bash(self, command: str, sandbox_id: str = None) -> Dict[str, Any]: """实际项目中，这里应调用沙箱接口，而不是本地执行。""" # 此处为演示，进行简单模拟。真实环境必须使用沙箱！ if "rm -rf" in command or "format" in command.lower(): return {"stdout": "", "stderr": "ERROR: Destructive command blocked by harness.", "returncode": 1} try: result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30) return {"stdout": result.stdout, "stderr": result.stderr, "returncode": result.returncode} except subprocess.TimeoutExpired: return {"stdout": "", "stderr": "ERROR: Command timed out.", "returncode": -1} def _read_file(self, filepath: str) -> str: """从workspace读取文件""" try: with open(f"workspace/{filepath}", 'r', encoding='utf-8') as f: return f.read() except FileNotFoundError: return "ERROR: File not found."

步骤 2：实现带钩子的 Agent 基类在harness_core/agent_base.py中，我们实现一个基础的 Agent 类，它集成了工具调用和钩子机制。

# harness_core/agent_base.py import openai from harness_core.tool_registry import ToolRegistry class AgentBase: def __init__(self, name: str, model: str, system_prompt: str, tool_registry: ToolRegistry): self.name = name self.model = model self.system_prompt = system_prompt self.tool_registry = tool_registry self.client = openai.OpenAI(api_key="your-api-key") # 请替换 self.conversation_history = [{"role": "system", "content": self.system_prompt}] def run(self, user_query: str) -> str: """运行Agent的主要循环（简化版ReAct）""" self.conversation_history.append({"role": "user", "content": user_query}) max_steps = 10 for step in range(max_steps): # 1. 调用模型，获取思考或工具调用 response = self.client.chat.completions.create( model=self.model, messages=self.conversation_history, tools=self._format_tools_for_openai(), # 将工具格式化为OpenAI格式 tool_choice="auto", ) message = response.choices[0].message self.conversation_history.append(message) # 2. 检查是否是工具调用 if message.tool_calls: for tool_call in message.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) # **关键：在执行工具前，调用“前置钩子”** if not self._run_pre_tool_hook(tool_name, tool_args): result = "Tool execution blocked by pre-hook." else: result = self.tool_registry.call_tool(tool_name, **tool_args) # **关键：在执行工具后，调用“后置钩子”** result = self._run_post_tool_hook(tool_name, tool_args, result) self.conversation_history.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result) }) else: # 没有工具调用，返回最终回答 final_response = message.content return final_response return "Agent reached max steps without final answer." def _format_tools_for_openai(self): """简化版工具格式转换""" # 实际项目需要更完整的schema定义 tools = [] for name, info in self.tool_registry.tools.items(): tools.append({ "type": "function", "function": { "name": name, "description": info["description"], } }) return tools def _run_pre_tool_hook(self, tool_name: str, args: dict) -> bool: """前置钩子：返回False则阻止工具执行""" # 示例：阻止任何名为‘write_file’的工具直接执行 if tool_name == "write_file": print(f"[HARNESS INTERVENTION] Agent '{self.name}' attempted to write file. Blocked. Requiring manual approval.") return False return True def _run_post_tool_hook(self, tool_name: str, args: dict, result: dict) -> dict: """后置钩子：可以修改工具执行结果""" # 示例：为所有bash命令结果添加安全标记 if tool_name == "execute_bash": result["_harness_note"] = "Command executed in simulated sandbox." return result

步骤 3：定义具体的技能和 Agent我们创建一个生成和验证 JSON 配置的技能。

# skills/config_skill.py import json class ConfigSkill: @staticmethod def generate_config(app_name: str, port: int) -> str: """生成一个基础的应用程序配置模板""" config = { "app_name": app_name, "port": port, "debug": False, "database": { "host": "localhost", "name": f"{app_name}_db" } } return json.dumps(config, indent=2) @staticmethod def validate_config(config_str: str) -> tuple: """验证配置是否合法""" try: config = json.loads(config_str) # 简单验证规则 errors = [] if not isinstance(config.get("app_name"), str): errors.append("'app_name' must be a string.") if not isinstance(config.get("port"), int) or not (1 <= config.get("port", 0) <= 65535): errors.append("'port' must be an integer between 1 and 65535.") if errors: return False, errors return True, "Config is valid." except json.JSONDecodeError as e: return False, [f"Invalid JSON: {e}"]

然后，我们创建两个专门的 Agent。

# agents/generator_agent.py from harness_core.agent_base import AgentBase class GeneratorAgent(AgentBase): def __init__(self, tool_registry): system_prompt = """你是一个配置文件生成专家。用户会描述他们需要的应用程序配置，你需要生成一个标准、安全的JSON配置文件。 你可以使用 `read_file` 工具查看现有配置模板，但你不能直接写入文件。生成配置后，请用清晰的语言描述你的配置内容。""" super().__init__(name="ConfigGenerator", model="gpt-4", system_prompt=system_prompt, tool_registry=tool_registry)

# agents/validator_agent.py from harness_core.agent_base import AgentBase from skills.config_skill import ConfigSkill class ValidatorAgent(AgentBase): def __init__(self, tool_registry): system_prompt = """你是一个配置文件验证专家。你将收到一段JSON配置文本，你需要严格检查其语法和逻辑的正确性。 请调用 `read_file` 工具来读取配置内容（如果提供了路径），或者直接分析给出的文本。 你的输出必须是一个清晰的验证报告，指出任何错误或警告。""" super().__init__(name="ConfigValidator", model="gpt-4", system_prompt=system_prompt, tool_registry=tool_registry) def validate(self, config_content: str) -> str: # 这里我们直接使用技能进行验证，也可以让模型分析 is_valid, message = ConfigSkill.validate_config(config_content) if is_valid: return f"Validation PASSED. {message}" else: return f"Validation FAILED. Errors: {message}"

步骤 4：编写主流程并引入人工介入在main.py中，我们编排整个工作流。

# main.py from harness_core.tool_registry import ToolRegistry from agents.generator_agent import GeneratorAgent from agents.validator_agent import ValidatorAgent from skills.config_skill import ConfigSkill import json def manual_approval(action: str, context: str) -> bool: """模拟人工审批环节""" print(f"\n{'='*50}") print(f"[MANUAL APPROVAL REQUIRED]") print(f"Action: {action}") print(f"Context:\n{context}") print(f"{'='*50}") # 在实际系统中，这里可以连接邮件、Slack、审批系统 user_input = input("Approve this action? (yes/no): ").strip().lower() return user_input == 'yes' def main(): print("Starting Multi-Agent Config Generation Demo...") tool_registry = ToolRegistry() # 1. 用户需求 user_request = "请为我的电商后端服务‘ShopBackend’生成一个配置文件，服务端口用 8080。" # 2. Generator Agent 工作 print("\n[Phase 1] Generator Agent is working...") generator = GeneratorAgent(tool_registry) # 在实际中，Generator会通过工具调用和模型思考来生成配置。 # 这里为演示，我们直接使用技能生成。 config_content = ConfigSkill.generate_config("ShopBackend", 8080) print(f"Generated config:\n{config_content}") # 3. Validator Agent 工作 print("\n[Phase 2] Validator Agent is working...") validator = ValidatorAgent(tool_registry) validation_result = validator.validate(config_content) print(f"Validation Result: {validation_result}") # 4. 人工介入：确认是否写入文件 if "PASSED" in validation_result: print("\n[Phase 3] Configuration is valid. Proceeding to write phase.") # **关键人工介入点** if manual_approval("Write configuration to file", f"Content to write:\n{config_content}"): # 只有人工批准后，才执行写操作 file_path = "workspace/config.json" with open(file_path, 'w', encoding='utf-8') as f: f.write(config_content) print(f"[SUCCESS] Configuration has been written to {file_path}") else: print("[CANCELLED] Write operation was rejected by human.") else: print("\n[Phase 3] Validation failed. Not proceeding to write. Please fix the errors.") if __name__ == "__main__": main()

运行与验证：

安装依赖：pip install openai
在项目根目录创建workspace文件夹。
将main.py中的api_key替换为你自己的。
运行python main.py。

你会看到控制台依次输出生成、验证的过程，并在最后**暂停，等待你在终端输入“yes”或“no”**来批准写入操作。这模拟了 Harness 系统中关键的人工管控点。

4. 将Demo升级为生产级系统的关键考量

上面的Demo展示了核心概念，但距离一个企业级系统还有巨大差距。以下是你在实际项目中必须深入处理的环节。

4.1 状态管理与上下文工程

Agent 是有状态的。它需要记住之前的对话、工具调用结果和任务进度。在长周期、多步骤任务中，管理好上下文（Context）是成败关键。

面临的挑战：

上下文窗口限制：所有主流模型都有 token 数量上限。当对话和工具输出很长时，会触及上限。
上下文腐化：无关或过时的信息充斥上下文，会干扰模型当前任务的判断。

Harness 的解决方案：

压缩：当上下文快满时，Harness 系统需要自动将历史对话中不重要的部分进行总结（Summarize），保留核心信息，腾出空间。
卸载：对于大型工具输出（如一个 5000 行的日志文件），不要全部塞进上下文。Harness 应只保留头部和尾部的关键行，将完整内容写入工作区的文件，并告诉 Agent：“完整日志已保存在debug.log中，如需查看第 2500 行，请使用read_file工具”。
技能按需加载：不要在系统提示词里一次性列出所有 100 个工具的说明。采用“渐进式披露”，初始只给通用工具。当模型表现出需要某个专业领域技能时（例如，它说“我需要创建一个图表”），Harness 再将“数据可视化”技能包的详细说明和工具动态注入上下文。
结构化记忆：使用类似AGENTS.md的“记忆文件”。这个文件记录了对本项目、本代码库的长期约定（如“我们使用 pytest 而不是 unittest”）。每次启动 Agent 时，Harness 自动将这个文件的内容注入系统提示词。Agent 在运行过程中也可以修改这个文件，实现跨会话的“持续学习”。

4.2 可观测性与调试

当拥有多个协同的 Agent 时，系统会变得复杂。你必须能看清里面发生了什么。

必须建立的观测点：

追踪：为每个用户请求生成一个唯一的trace_id，贯穿所有子 Agent 和工具调用。这样你可以完整复现一个请求的生命周期。
日志：结构化日志，记录每个 Agent 的输入、输出、调用的工具及其参数/结果、token 消耗、耗时。
成本与延迟度量：监控每个模型调用、工具调用的花费和时间，用于优化和预算控制。
Agent 思维过程：记录模型在调用工具前的“推理”（Reasoning）内容，这是调试 Agent 为什么做出错误决策的最重要依据。

实战建议：使用像 LangSmith、Weights & Biases 或自建的基于 OpenTelemetry 的系统来统一收集这些数据。调试时，优先看“思维过程”和工具调用的输入输出，这比只看最终结果有效得多。

4.3 自我进化与“棘轮”机制

Harness Engineering 的精髓在于“棘轮”机制：每一次失败都转化为一条永不失效的规则。

如何实现：

收集失败案例：当 Agent 犯错（如提交了编译不通过的代码、写了不安全的 SQL、忽略了代码规范），不要仅仅重试。将这个案例（包括输入、Agent 的思维链、错误输出）记录到案例库。
分析根因：是工具描述不清？是系统提示词有歧义？是缺少某个约束？例如，Agent 提交了被注释掉的测试代码，根因是它不知道“不能提交被注释的测试”这条规则。
更新 Harness：
- 更新AGENTS.md：增加一条明确的规则：“Never commit commented-out tests. Either fix them or delete them.”
- 添加钩子：在pre-commit钩子中，加入检查代码 diff 中是否包含it.skip、xit、@unittest.skip等模式的逻辑，如果发现则自动拒绝提交，并将错误信息反馈给 Agent。
- 优化工具：如果是因为工具能力不足，则改进工具。例如，SQL 工具增加更严格的注入检查。
验证与闭环：规则添加后，用之前失败的案例重新测试，确保问题被解决。同时，也要确保新规则不会引入新的问题（如误报）。

这个过程是手动的、需要人工智慧的，但它使得你的 Agent 系统像软件一样，拥有版本号，并随着时间不断变得更健壮。

4.4 安全与权限的纵深防御

多 Agent 系统，尤其是能执行代码的 Agent，是巨大的攻击面扩大器。安全必须是设计时的第一原则。

防御层级：

沙箱层：如前所述，这是最后也是最坚固的防线。确保即使 Agent 被恶意提示词诱导，其破坏也被限制在沙箱内。
工具层：每个工具函数内部都必须进行严格的输入验证、输出编码和权限检查。例如，一个“执行数据库查询”的工具，必须使用参数化查询，并限制其只能访问特定的数据库和表。
提示词层：系统提示词中必须包含明确的安全指令，例如“你绝不能执行任何破坏性操作或尝试绕过限制”。但绝不能只依赖提示词，因为模型可能被“提示词注入”攻击绕过。
钩子层：用于拦截已知的高风险模式（如强制推送、删除命令）。
人工审批层：对于最高风险的操作（如生产环境部署、删除重要数据），流程必须设计为强制中断，等待明确的人工批准。

记住，你赋予 Agent 的每一个工具，都相当于给了它一部分你系统的操作权限。设计 Harness 时，要时刻以“最小权限原则”来思考。

从 Demo 到生产，核心区别不在于 Agent 的数量，而在于这套“缰绳”系统的完备性、可靠性和可观测性。开始的时候可以像 Demo 一样简单，但必须预留出这些核心组件的扩展点。你的迭代速度，将直接取决于你的 Harness 能多快地将一次失败转化为一条固化的规则。这才是 Harness Engineering 在实战中的真正价值。