ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

Harness工程实战:Multi-Agent、SandBox与Skill构建可控AI智能体

Harness工程实战:Multi-Agent、SandBox与Skill构建可控AI智能体 这次我们来看 Harness 工程。它不是某个单一开源项目的名字而是 2025 年 AI 智能体开发里最值得补的一课把大模型从“能聊天”推进到“能干活、能稳定地干活”的工程方法。最近 DeepSeek Harness、Codex Harness 这些词频繁出现在各种讨论里网上也出现了一大批以 Harness Multi-Agent SandBox Skill 为骨架的实战教程其中就包括这套 60 集的 Harness 工程全套教程。先给结论如果你已经在本地部署过 AI 大模型或者跑过简单的 Agent Demo但发现“演示能跑、一上生产就崩”那问题一般不在模型本身而在 harness——也就是模型外面的那层工程壳。模型负责“理解”和“生成”harness 负责“让它按规则干活、出错了能回滚、执行环境不失控、能力可以沉淀复用”。这篇文章把 Harness 工程拆成四个部分来讲Multi-Agent 多智能体协作、SandBox 沙箱隔离、Skill 能力沉淀、以及 API 与批量任务对接。读完你可以照着搭一个最小的 Harness知道每一步应该验证什么、踩坑了往哪里排查。文章会从环境准备开始给出一套可复制的代码框架再逐步加上 SandBox 和 Skill最后讨论多 Agent 编排、接口暴露、批量任务和性能观察。涉及本地部署和大模型接入时我也把合规边界单独列了一节各位在动手前先过一眼。1. Harness 工程核心能力速览能力项说明工程定位大模型 Agent 的工程化控制层不是单一模型而是一套“模型 工具 沙箱 技能 编排”的完整运行框架核心模块Multi-Agent 多智能体协作、SandBox 沙箱隔离、Skill 技能封装、API 服务典型参考OpenAI Codex Harness、DeepSeek Harness 等均采用“模型外挂工具 沙箱执行 技能库”的架构思路硬件要求取决于接入的模型。纯 API 接入如 DeepSeek API无显卡门槛本地部署 7B~70B 模型建议 8GB 以上显存并按量化版本测试具体以实际模型为准启动方式命令行 / Python 脚本 / FastAPI 服务 / Docker 沙箱是否支持 API支持通常以 FastAPI 或 Flask 暴露任务执行接口是否支持批量任务支持通过任务队列 并发 Worker 实现是否支持 Skill 扩展支持Skill 是核心设计之一可注册、复用、版本化管理适合人群已了解 Prompt 工程想进入 Agent 工程化和生产落地的开发者说明上表是对 Harness 工程通用能力的内容框架总结。具体到某一套教程或开源项目参数和功能以它的实际文档和章节安排为准。2. 为什么 2025 年大家都在学 Harness 工程先看两个现象。第一个现象是 OpenAI Codex 带火了 “Harness” 这个词。Codex 本质上不是一个单纯的 CLI它是一整套围绕模型打造的工程系统模型通过工具调用操作终端、读写文件、运行测试所有执行都被限制在沙箱里技能以 SKILL.md 的形式沉淀。这套系统被称为 Codex Harness意味着重点已经从“模型多聪明”转移到了“外面这层壳多可靠”。第二个现象是 DeepSeek Harness 的出现。中文技术社区里“deepseek harness 安装”“deepseek harness 怎么使用”很快成了高频搜索词。大家关心的是能不能用类似 Codex 的方式把 DeepSeek 这类国产大模型也接进一套可控的 Agent 框架里。这也让越来越多开发者意识到真正值钱的不是把模型 API 调通而是把“模型 环境 工具 流程”做成一个能上线的东西。那 Harness 和 Agent 到底有什么区别这是新手最容易混淆的问题。Agent 是指一个能够自主规划、调用工具、完成目标的运行实体它的核心是“感知—决策—行动”的循环。而 Harness 是承载并约束 Agent 的工程框架它负责的是 Agent 之外的一切上下文怎么管理、工具怎么注册、代码在哪里执行、超时怎么处理、日志怎么留、权限怎么控制、技能怎么积累。可以说Agent 是演员Harness 是舞台、灯光、剧本和安保系统。很多失败项目都是只有 Agent 没有 Harness模型一通乱调工具代码在宿主机上直接执行技能散落在各种脚本里任务一多就乱。这也是这套 60 集教程把 Multi-Agent、SandBox、Skill 并列放在标题里的原因——它们正好对应 Agent 生产化的三个最痛的点多智能体怎么写、执行环境怎么隔离、能力怎么复用。3. Harness 工程的三大核心模块3.1 Multi-Agent多智能体协作架构单 Agent 能做的事情有上限。当任务涉及多个专业领域或者流程需要分阶段推进时把不同职责拆给不同 Agent 往往更可控。典型的分工方式是规划 Agent拆解任务生成执行计划。代码 Agent负责写代码、改代码、跑测试。审查 Agent检查输出质量和安全问题。执行 Agent负责调用外部系统比如发请求、查数据库。Multi-Agent 的重点不是“多开几个 Agent”而是定义清楚它们之间的通信协议和任务边界。常见的协作模式有串行管道、并行分发、主从编排三种后面第 6 节会给出具体代码示例。初学者最容易犯的错是让两个 Agent 反复来回对话token 消耗巨大且结果不可控所以工程上要优先用流程来约束交互而不是靠模型自由发挥。3.2 SandBox沙箱隔离SandBox 是 Harness 工程里最容易忽略、也最不能省的一层。模型生成的代码、模型执行的命令都应该在一个受控环境里运行而不是直接在宿主机上执行。沙箱要解决的安全问题很明确代码可能读取敏感文件、可能调用网络请求、可能写入任意路径、可能陷入死循环。所以一个最小可用的沙箱至少要有文件系统隔离在临时目录或容器内运行。网络控制默认禁用外网访问按需开放白名单。超时控制进程超过设定时间直接杀掉。资源限制限制内存和 CPU防止模型生成的代码把机器打满。Hot search 里那句“disabled no sandbox”对应的就是关闭沙箱的配置项。开发调试时可以打开但生产环境强烈建议保持沙箱开启。后面第 5 节会写一个基于 subprocess 的最小沙箱实现生产环境可以在此基础上换成 Docker 或 gVisor。3.3 Skill可复用的技能封装Skill 是 Harness 工程里让能力“越用越多”的关键设计。它把一段可复用的能力封装成“描述 参数 实现代码”Agent 在执行任务时按需加载。一个 Skill 通常包含名称和用途描述告诉模型什么场景下该用这个 Skill。参数定义声明调用时需要传入哪些字段。实现逻辑具体执行代码。使用示例给模型参考的调用样例。Skill 的价值在于沉淀。写一次搜索 Skill之后所有 Agent 都能用写一个 Excel 处理 Skill就不需要每次重新生成同样功能的代码。教程里常提到的 Agent Skill、Skill Creator、Skill Recorder 都是围绕这个体系展开的前者是技能的消费方后两个是技能的生成和录制工具。4. Harness 工程本地部署环境准备Harness 本身不是一个重组件它对机器没有特别苛刻的要求真正的资源大头是底层模型。在动手之前先按下面的清单检查环境。4.1 操作系统与运行环境Harness 工程的主体代码一般用 Python 编写建议使用 Python 3.10 及以上版本。如果教程里有 Node.js 版本的实现比如模仿 Codex CLI 的工具则需要 Node.js 18。# 检查 Python 版本 python --version # 检查 Node 版本如果用到 CLI 工具 node --version4.2 依赖管理建议使用虚拟环境避免依赖冲突污染系统环境。python -m venv .venv source .venv/bin/activate # Windows 下是 .venv\Scripts\activate pip install --upgrade pip pip install requests openai fastapi uvicorn # 按实际项目补充4.3 模型接入方式按照教程内容Harness 工程通常会支持两种模型接入方式接入方式说明硬件门槛API 接入通过 OpenAI 兼容接口调用 DeepSeek、通义、Kimi 等大模型无显卡要求需要 API Key本地模型通过 llama.cpp、Ollama、vLLM 等部署开源模型建议 8GB 以上显存具体以模型量化版本为准本地部署 AI 大模型时常用的启动方式是 Ollama 或 vLLM。Ollama 更轻量适合学习和测试vLLM 吞吐更高适合生产环境。无论哪种方式Harness 只需要对接一个兼容 OpenAI 的 HTTP 接口即可。4.4 磁盘与端口磁盘纯 API 接入 10GB 足够本地模型按模型文件大小预留 20GB 以上。端口Harness API 服务默认建议用 7860 或 8000启动前用下面的命令检查占用。# Linux / macOS lsof -i :8000 # Windows netstat -ano | findstr :8000如果端口被占用启动参数里改一个端口即可不需要强制杀掉其他进程。5. 从零实现最小 Harness 实战这一节不依赖具体教程环境给出一个最小可运行的结构一个 Agent 循环 一个 Skill 注册表 一个沙箱执行器。整个框架不到 150 行代码可以跑通后再逐步扩充。5.1 搭建 Agent 基础循环Agent 循环是 Harness 的心脏。模型每轮输出可能是普通文本回复也可能是工具调用请求Harness 负责解析、执行、把结果回填给模型。import json from typing import Callable, Optional class Message: def __init__(self, role: str, content: str, tool_call: Optional[dict] None): self.role role self.content content self.tool_call tool_call # {name: str, arguments: dict} class Agent: def __init__( self, model_fn: Callable[[list[dict]], Message], tools: Optional[dict[str, Callable]] None, max_steps: int 10, ): self.model_fn model_fn self.tools tools or {} self.messages [] self.max_steps max_steps def run(self, task: str) - str: self.messages.append({role: user, content: task}) for step in range(self.max_steps): response self.model_fn(self.messages) self.messages.append({ role: assistant, content: response.content, }) if response.tool_call is None: return response.content tool_name response.tool_call[name] tool_args response.tool_call[arguments] if tool_name not in self.tools: result f错误工具 {tool_name} 不存在 else: try: result self.tools[tool_name](**tool_args) if not isinstance(result, str): result json.dumps(result, ensure_asciiFalse) except Exception as e: result f工具执行异常{str(e)} self.messages.append({ role: tool, name: tool_name, content: result, }) raise TimeoutError(Agent 执行步数超过上限)model_fn是模型接入层的统一入口。你可以直接在里面调用任何兼容 OpenAI 的 APIDeepSeek API、本地 Ollama 都能在这一层做适配。5.2 注册 SkillSkill 的注册表本质是一个字典key 是技能名value 是技能实现。重点是让模型知道“这个技能什么时候用、参数怎么传”所以描述必须写清楚。class SkillRegistry: def __init__(self): self._skills {} def register(self, name: str, description: str, func: Callable): self._skills[name] { description: description, func: func, } def get_tool_schemas(self) - list[dict]: schemas [] for name, skill in self._skills.items(): schemas.append({ type: function, function: { name: name, description: skill[description], parameters: { type: object, properties: {}, }, }, }) return schemas def execute(self, name: str, **kwargs): if name not in self._skills: raise KeyError(fSkill {name} 未注册) return self._skills[name][func](**kwargs) registry SkillRegistry() def skill(name: str, description: str): def decorator(func: Callable): registry.register(name, description, func) return func return decorator skill(namesum_numbers, description对两个整数求和) def sum_numbers(a: int, b: int) - str: return str(a b)Skill 的注册本身不复杂复杂的是维护一套高质量技能库。教程里提到 Skill Creator、Skill Recorder核心就是在解决“如何把一次成功的任务过程自动沉淀成一个可复用 Skill”的问题。你可以在每次 Agent 成功完成任务后把有用的工具调用序列存下来作为新 Skill 的雏形。5.3 接入沙箱执行沙箱的目的是让 Agent 生成的代码在受控环境里执行。下面这个实现基于subprocess限制超时和临时工作目录适合学习和原型验证。生产环境建议换成 Docker 容器。import os import subprocess import tempfile class CodeSandbox: def __init__(self, timeout: int 10, memory_limit_mb: int 512): self.timeout timeout self.memory_limit_mb memory_limit_mb def execute_python(self, code: str) - dict: with tempfile.TemporaryDirectory() as tmp_dir: script_path os.path.join(tmp_dir, script.py) with open(script_path, w, encodingutf-8) as f: f.write(code) try: result subprocess.run( [python, script_path], capture_outputTrue, textTrue, timeoutself.timeout, cwdtmp_dir, ) return { stdout: result.stdout, stderr: result.stderr, returncode: result.returncode, } except subprocess.TimeoutExpired: return { stdout: , stderr: f执行超时{self.timeout}s, returncode: -1, }在真实项目里这个沙箱还要加上内存限制、网络白名单、文件系统只读等策略。常用的方案是 Docker# 用 Docker 执行模型生成的代码 echo print(hello sandbox) | docker run --rm -i --network none --memory 512m python:3.11-slim python ---network none表示禁用网络--memory 512m限制内存--rm保证执行完自动清理。模型生成代码越自由沙箱策略越要严格。6. Multi-Agent 编排与协作模式6.1 串行管道模式任务按阶段拆分上一个 Agent 的输出作为下一个 Agent 的输入适合“规划—编码—审查”这类流程。class SerialPipeline: def __init__(self, agents: list[Agent]): self.agents agents def run(self, task: str) - str: current task for i, agent in enumerate(self.agents): print(f[Pipeline] 第 {i 1} 个 Agent 开始处理) current agent.run(current) return current6.2 并行分发模式多个 Agent 同时处理独立子任务最后汇总结果。适合数据分析、多源信息搜集等场景。from concurrent.futures import ThreadPoolExecutor, as_completed class ParallelDispatcher: def __init__(self, agents: dict[str, Agent], max_workers: int 3): self.agents agents self.max_workers max_workers def run(self, tasks: dict[str, str]) - dict[str, str]: results {} with ThreadPoolExecutor(max_workersself.max_workers) as executor: future_map { executor.submit(self.agents[name].run, task): name for name, task in tasks.items() } for future in as_completed(future_map): name future_map[future] try: results[name] future.result() except Exception as e: results[name] fAgent {name} 执行失败{str(e)} return results6.3 编排模式怎么选模式适用场景风险串行管道有明确前后依赖的任务流单点失败会中断整条链路并行分发子任务互相独立并发过高会打满 API 配额主从编排需要规划者统一分配任务规划 Agent 的决策质量是瓶颈教程里的 Multi-Agent 部分主要就是在教大家根据任务类型选择合适的编排模式而不是把所有任务都丢给一个 Agent 自由发挥。实际开发时建议先画一张任务 DAG有向无环图明确每个节点的输入输出再决定用串行还是并行。7. 接口 API 与批量任务设计Harness 工程在完成基础能力后下一步就是对外提供服务。常见的做法是用 FastAPI 暴露接口让其他系统可以提交任务、查询结果。7.1 API 服务启动from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleHarness API) class RunTaskRequest(BaseModel): task: str # 任务描述 agent: str default # 使用哪个 Agent timeout: int 60 # 超时时间秒 class RunTaskResponse(BaseModel): status: str # ok / error result: str agent: str app.post(/api/run, response_modelRunTaskResponse) def run_task(req: RunTaskRequest): try: agent agent_pool[req.agent] result agent.run(req.task) return RunTaskResponse(statusok, resultresult, agentreq.agent) except Exception as e: return RunTaskResponse(statuserror, resultstr(e), agentreq.agent) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动服务python api_server.py启动后用 curl 验证接口curl -X POST http://127.0.0.1:8000/api/run \ -H Content-Type: application/json \ -d {task: 计算 12 和 27 的和, agent: default}7.2 批量任务处理批量任务的核心是引入任务队列。提交的 HTTP 请求只负责把任务写入队列后台 Worker 负责慢慢消费避免一个长任务把接口请求卡死。import threading import uuid from queue import Queue task_queue Queue() task_store {} def worker_loop(): while True: task_id, agent, task task_queue.get() try: task_store[task_id] {status: running, result: None} task_store[task_id][result] agent.run(task) task_store[task_id][status] done except Exception as e: task_store[task_id][status] error task_store[task_id][result] str(e) finally: task_queue.task_done() def submit_task(task: str, agent: Agent) - str: task_id str(uuid.uuid4()) task_queue.put((task_id, agent, task)) task_store[task_id] {status: queued, result: None} return task_id def get_result(task_id: str) - dict: return task_store.get(task_id, {status: not_found, result: None})批量任务设计时要考虑三个点失败重试单个任务失败不应该影响整个批次建议每个任务独立捕获异常并记录。日志每个任务要有 task_id 贯穿日志方便定位。限流调用外部模型 API 时要控制并发数避免触发配额限制。7.3 模型接入示例用 DeepSeek API 兼容 OpenAI 接口时model_fn可以这样写from openai import OpenAI client OpenAI( base_urlhttps://api.deepseek.com, api_keyyour-api-key, ) def deepseek_model_fn(messages: list[dict]) - Message: response client.chat.completions.create( modeldeepseek-chat, messagesmessages, ) content response.choices[0].message.content return Message(roleassistant, contentcontent, tool_callNone)如果是本地部署的 Ollama把base_url换成http://127.0.0.1:11434即可。接口设计上尽量保持 OpenAI 兼容这样 Harness 的模型接入层可以随时切换底层模型。8. 资源占用与性能观察Harness 工程的资源占用主要来自三块模型推理、沙箱执行、日志存储。8.1 模型推理开销单次 Agent 任务消耗的 token 初始 Prompt 多轮工具调用上下文 工具返回结果 最终回答。工具返回结果越大下一轮请求的 prompt 就越长消耗成倍增加。建议对每次任务记录 token 用量用于成本控制。8.2 沙箱执行开销沙箱执行会额外产生进程创建和临时文件读写。在容器化方案里每次执行还需要拉镜像、启动容器延迟会比直接执行高几秒。如果是高频调用建议容器预热常驻一个运行中的容器通过管道传入代码而不是每次重建。8.3 性能观察方法观察 Agent 循环轮数轮数越多延迟和成本越高。观察工具调用失败率失败后模型会用额外轮次尝试恢复消耗更多 token。观察沙箱超时频率如果模型频繁生成超时代码应该在 Prompt 层加上约束。本地部署模型时可以用nvidia-smi观察显存占用# 每隔 2 秒刷新一次显存占用 watch -n 2 nvidia-smi显存占用需要以实际模型版本和推理参数为准不同量化级别、上下文长度的差异很大不要拿别人的数字直接套用到自己的环境。9. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 不调用工具只返回文本模型没有收到工具 Schema或 Prompt 未说明工具可用检查请求 payload 里是否带上了 tools 参数在模型请求中显式附加工具定义工具参数经常传错工具的参数 Schema 描述不够清晰查看模型传出的 arguments 和报错信息补充参数描述和示例增加类型校验沙箱执行超时模型生成了死循环代码或任务本身耗时过长查看沙箱日志中的超时记录降低单次执行超时阈值在 Prompt 中约束代码风格本地模型推理很慢显存不足导致模型部分落在 CPU 上用 nvidia-smi 观察显存占用换小尺寸模型或降低量化精度批量任务大量失败并发过高触发 API 限流查看 API 返回的限流状态码和错误信息增加重试和退避机制降低并发数端口被占用无法启动之前服务未正常退出检查端口监听进程换端口或结束残留进程Skill 注册后找不到Skill 名称拼写不一致或注册时机晚于 Agent 初始化检查注册表字典内容统一名称规范启动时做技能加载自检多 Agent 交互跑飞缺少流程约束两个 Agent 无限对话查看日志中双方交互轮数改为串行管道或主从编排限制单 Agent 最大轮数10. 最佳实践与合规边界10.1 工程实践建议第一次跑通时用小模型、小任务先验证链路再上规模。每个 Agent 任务都要有“最大步数”兜底防止死循环。沙箱默认开启网络默认禁用按需开放白名单。Skill 要写清楚使用条件和参数说明模糊的描述会导致模型滥用技能。任务队列要记录完整日志至少包含 task_id、开始时间、结束时间、状态、错误信息。API 服务默认监听127.0.0.1不要直接暴露公网如果需要远程访问加鉴权。模型输出结果在发布或商用前要做人工复核特别是在代码、金融、医疗等高风险场景。10.2 合规与安全边界Harness 工程涉及大模型调用、代码执行、数据处理使用时必须注意调用模型 API 时不要提交未经授权的隐私数据、源代码和敏感文件。沙箱内执行的代码必须视为不可信内容严格限制权限。涉及人脸、声音、版权素材等内容生成时必须确认已获得合法授权不得用于侵权用途。本地部署模型时要注意模型的开源协议明确商业使用边界。批量抓取和处理外部数据时要遵守目标站点的使用条款和相关法律法规。教程里演示的多数是技术可行性验证落地时一定要根据实际场景补齐安全策略。11. 总结与学习路线建议Harness 工程最值得投入精力验证的三个功能点一是 SandBox 隔离是否可靠二是 Skill 注册与复用的完整链路三是 API 暴露之后批量任务能否稳定跑完。最容易踩的坑是跳过沙箱直接让模型生成的代码在宿主机执行以及多 Agent 之间缺乏流程约束导致 token 无止境消耗。如果你准备跟着这套 60 集教程系统学习我建议按下面的节奏推进阶段学习内容验证目标第 1~2 天Harness 基础概念、Agent 与 Harness 的区别跑通最小 Agent 循环第 3~4 天Skill 设计与 Skill Creator完成一个 Skill 的注册和调用第 5 天SandBox 隔离与安全策略恶意代码被沙箱正确拦截或超时第 6 天Multi-Agent 编排串行管道 并行分发各跑通一个案例第 7 天API 服务与批量任务提交 50 个批量任务并成功收集结果学习过程中建议自己维护一套“最小可运行配置”一个会调用工具的 Agent、一个沙箱、两个 Skill这个最小骨架在你换模型、换任务时都能快速复用。后续可以继续扩展的方向包括Skill 自动录制、Agent 记忆持久化、沙箱升级为 Docker 隔离、以及把 Harness 接入到现有的业务系统中。把这套流程跑通你就完成了从“会调 API”到“能交付 Agent 系统”的关键一步。
返回列表