AI脚本怎么变成可用工具？任务边界、输入输出与最小产品架构-开发者社区

一段脚本在开发者电脑上跑通，往往只说明“这一次输入能得到一个结果”。它还没有回答普通使用者真正会遇到的问题：上传的是哪一份文件、同一次点击会不会重复执行、关闭页面后任务是否还在、失败的原因能不能看懂、最终下载的结果能不能证明来自这次输入。

本文用一个脱敏的媒体转换工具说明最小产品架构。固定任务PT-20261007-001上传source.mp4，选择受控预设WEB_1080P，后台异步生成result.mp4和一份媒体摘要。例子中的文件名、任务号和参数均为教学数据；不涉及真实素材、内部服务地址、密钥或产品编排。

环境边界：Python 3.11 Worker、Java 17、Spring Boot 风格服务层、MySQL 8.x。媒体处理命令仅作为可替换的 Worker 能力，本文讨论的是任务产品化，不公开生产命令和内部实现细节。

脚本跑通为什么还不算工具
固定案例：先约定输入、输出和不做什么
最小任务架构：把一次点击变成可追溯任务
数据模型：任务、幂等键和产物不能混在一起
Worker 实现：受控执行与临时产物
服务层实现：提交事务不等待长任务
预期输出与自动测试
SQL 验证：上线后怎样发现状态和产物不一致
异常边界与上线验收
小结和延伸阅读

一、脚本跑通为什么还不算工具

假设原始脚本接受一个本地路径和一个预设：

python transform.py --input source.mp4 --preset WEB_1080P

开发者看到输出文件出现，就会认为它成功了。但页面上的一次“开始处理”至少跨越四个独立事实：请求是否被接收、任务是否排队、Worker 是否执行、结果是否通过验收并可下载。把这四件事都塞进一次 HTTP 请求，会出现三个常见故障：浏览器超时后用户再次点击；应用重启时正在运行的工作消失；Worker 写出半个文件，页面却已经显示成功。

产品化的第一步不是加一个更漂亮的页面，而是把“调用脚本”改成“创建任务并交付产物”。请求完成只代表任务已受理；只有验收后的正式产物出现，才代表用户拿到了结果。

图1：命令行脚本只覆盖执行；可用工具还要记录输入、状态、产物和验收证据。

二、固定案例：先约定输入、输出和不做什么

任务PT-20261007-001的输入不是任意命令文本，而是一份经校验的文件记录和一套枚举配置：

项目	固定值或规则	为什么要这样约束
输入文件	`source.mp4`，上传后保存文件大小、摘要值和媒体信息	文件名可重复，文件身份不能只靠路径判断
可选预设	`WEB_1080P`、`ARCHIVE_SOURCE`	预设表达允许的能力，避免页面透传危险命令参数
幂等键	用户本次提交生成的`request_key`	网络重试或重复点击不能创建两份同类任务
成功产物	`result.mp4`与`result-summary.json`	下载文件和它的验收依据需要同时存在
完成条件	输出非空、媒体摘要可读取、状态与产物登记一致	“进程退出 0”本身不能代表可交付

该工具不负责识别画面内容，也不替用户修改文件；它只把受控的输入转换为一个可验证的交付结果。把边界写在输入契约里，后续才能拒绝不支持的格式、预设和状态，而不是让 Worker 在运行一半后猜测该怎么做。

图2：产品入口不是“传一个路径并执行”，而是创建一份可以复查的任务契约。

三、最小任务架构：把一次点击变成可追溯任务

最小架构不需要一开始就引入复杂的工作流平台，但必须拆开三个责任：服务层负责受理和状态转换；队列或调度器负责把待执行任务交给 Worker；Worker 负责运行受控步骤、写临时结果并提交产物。对象存储、磁盘目录或数据库都可以保存文件，关键是它们都通过任务号关联，而不由前端直接猜路径。

POST /tool-jobs -> 校验上传文件和预设 -> 事务内创建 QUEUED 任务、写入 request_key -> 提交后投递 taskId -> Worker 领取任务并置为 RUNNING -> 写入 temporary/result.mp4 -> 探测验收，通过后登记 RESULT 并置为 READY

建议的状态机只有六个状态：QUEUED、RUNNING、READY、FAILED、CANCEL_REQUESTED和CANCELLED。READY只能由验收通过后进入；FAILED必须留下可展示的错误分类，而不是把完整命令、路径或敏感输入回显给用户。

图3：请求线程只负责受理任务；Worker 完成后才允许将临时结果提交为正式产物。

四、数据模型：任务、幂等键和产物不能混在一起

若把输入路径、输出路径和最终状态都塞进一张“任务表”，后面很难表达一个任务对应多个产物、一个产物有不同角色、或同一输入的多次受控运行。这里用tool_job保存生命周期，用tool_job_artifact保存每一份可追溯文件。

CREATETABLEtool_job(idBIGINTPRIMARYKEYAUTO_INCREMENT,job_noVARCHAR(40)NOTNULL,request_keyVARCHAR(64)NOTNULL,preset_codeVARCHAR(32)NOTNULL,statusVARCHAR(24)NOTNULL,error_codeVARCHAR(48)NULL,config_versionVARCHAR(32)NOTNULL,created_atDATETIMENOTNULL,started_atDATETIMENULL,finished_atDATETIMENULL,UNIQUEKEYuk_tool_job_no(job_no),UNIQUEKEYuk_tool_job_request_key(request_key),CONSTRAINTck_tool_job_statusCHECK(statusIN('QUEUED','RUNNING','READY','FAILED','CANCEL_REQUESTED','CANCELLED')));CREATETABLEtool_job_artifact(idBIGINTPRIMARYKEYAUTO_INCREMENT,job_idBIGINTNOTNULL,artifact_roleVARCHAR(20)NOTNULL,object_keyVARCHAR(255)NOTNULL,sha256CHAR(64)NOTNULL,byte_sizeBIGINTNOTNULL,media_summary_json JSONNULL,created_atDATETIMENOTNULL,UNIQUEKEYuk_job_artifact_role(job_id,artifact_role),CONSTRAINTck_artifact_roleCHECK(artifact_roleIN('SOURCE','RESULT','SUMMARY')),CONSTRAINTfk_artifact_jobFOREIGNKEY(job_id)REFERENCEStool_job(id));

request_key解决“同一次用户意图”只能受理一次；job_no解决对外查询和日志关联；artifact_role则明确哪一份是输入、结果或摘要。MySQL 的CHECK能挡住无效枚举，但“READY 必须有 RESULT”这种跨表规则仍要由服务层和上线 SQL 同时验证，不能只依赖字段类型。

图4：状态、幂等键和产物角色分别建模，才可以对重复提交和半成品交付做出明确判断。

五、Worker 实现：受控执行与临时产物

Worker 不应该接收浏览器传来的任意 shell 字符串。它先根据preset_code构造固定参数列表，再在专属临时目录中执行；完成后校验临时产物，最后由服务层登记正式对象。Python 的subprocess.run可以用参数列表、超时和受控工作目录执行子进程，避免以shell=True拼接用户输入。

fromdataclassesimportdataclassfrompathlibimportPathimportsubprocess@dataclass(frozen=True)classRunResult:exit_code:intoutput_path:Path PRESET_ARGS={"WEB_1080P":("-vf","scale=-2:1080","-c:v","libx264"),"ARCHIVE_SOURCE":("-c","copy"),}defrun_job(source:Path,temporary_output:Path,preset:str)->RunResult:ifpresetnotinPRESET_ARGS:raiseValueError("UNSUPPORTED_PRESET")command=["media-tool","-i",str(source),*PRESET_ARGS[preset],str(temporary_output)]completed=subprocess.run(command,check=False,timeout=30*60,cwd=temporary_output.parent,capture_output=True,text=True,)ifcompleted.returncode!=0:raiseRuntimeError("WORKER_COMMAND_FAILED")ifnottemporary_output.exists()ortemporary_output.stat().st_size==0:raiseRuntimeError("RESULT_MISSING")returnRunResult(completed.returncode,temporary_output)

这里故意没有把completed.stderr原样存进用户可见字段。它可能含环境路径、文件名或命令细节。Worker 应记录脱敏后的诊断摘要，并把错误归类为INPUT_INVALID、WORKER_TIMEOUT、WORKER_COMMAND_FAILED或RESULT_INVALID，让页面、告警和重试策略都能基于同一套语义工作。

六、服务层实现：提交事务不等待长任务

数据库事务应该覆盖“校验后创建任务、写入幂等键、登记输入文件”这些短操作，而不应包住几十分钟的媒体或模型处理。长事务会占用连接、增加锁冲突，也不能让数据库回滚一个已经启动的外部进程。

@ServicepublicclassToolJobService{@TransactionalpublicCreateJobResponsecreate(CreateJobCommandcommand){ToolJobexisting=jobRepository.findByRequestKey(command.requestKey());if(existing!=null){returnCreateJobResponse.reused(existing.getJobNo(),existing.getStatus());}UploadFilesource=uploadValidator.requireSupportedVideo(command.fileId());ToolJobjob=ToolJob.queued(JobNo.next(),command.requestKey(),command.presetCode(),"tool-config-v1");jobRepository.insert(job);artifactRepository.insert(Artifact.source(job.getId(),source.objectKey(),source.sha256()));outboxRepository.insert(OutboxEvent.forJob(job.getId()));returnCreateJobResponse.accepted(job.getJobNo());}}

这里用 outbox 记录“任务已创建，需要投递”的事实：事务成功后，投递器才读取该事件并通知 Worker。即使应用恰好在提交后重启，未投递事件仍可再次扫描；Worker 领取任务时还要以条件更新保证只有一个执行者能将QUEUED改成RUNNING。这是下一篇任务队列和进度设计的基础。

七、预期输出与自动测试

固定案例的成功结果不只是一句“处理完成”，而应当是：

{"jobNo":"PT-20261007-001","status":"READY","preset":"WEB_1080P","artifacts":["SOURCE","RESULT","SUMMARY"],"downloadable":true}

正常测试验证幂等受理和成功提交；异常测试验证未通过验收的临时文件绝不登记为READY。下面以 Java 服务层为例：

@TestvoidsameRequestKeyReturnsTheExistingJob(){CreateJobCommandcommand=command("request-20261007-001","WEB_1080P");CreateJobResponsefirst=service.create(command);CreateJobResponsesecond=service.create(command);assertThat(second.jobNo()).isEqualTo(first.jobNo());assertThat(jobRepository.count()).isEqualTo(1);}@TestvoidresultCannotBecomeReadyBeforeValidationPasses(){ToolJobjob=jobRepository.save(queuedJob());assertThatThrownBy(()->completionService.commitResult(job.getId(),emptyTemporaryFile())).hasMessage("RESULT_MISSING");assertThat(jobRepository.find(job.getId()).getStatus()).isEqualTo("FAILED");assertThat(artifactRepository.findByJobAndRole(job.getId(),"RESULT")).isNull();}

Worker 也应有独立测试：未知预设必须在启动外部进程前失败；命令超时必须转换为确定的错误码；输出文件不存在时不能继续到正式提交。测试不需要调用真实模型或真实视频，临时目录和受控替身即可覆盖这些产品边界。

八、SQL 验证：上线后怎样发现状态和产物不一致

上线验收不能只看“最近请求返回 200”。以下查询针对的是最危险的状态断裂：已完成却没有结果、存在结果却未完成、同一提交重复建单。

-- 预期结果：0 行。READY 任务必须有 RESULT 产物。SELECTj.job_noFROMtool_job jLEFTJOINtool_job_artifact aONa.job_id=j.idANDa.artifact_role='RESULT'WHEREj.status='READY'ANDa.idISNULL;-- 预期结果：0 行。未完成任务不得暴露正式结果。SELECTj.job_no,j.statusFROMtool_job jJOINtool_job_artifact aONa.job_id=j.idANDa.artifact_role='RESULT'WHEREj.statusNOTIN('READY');-- 预期结果：0 组。request_key 的唯一性不允许重复受理。SELECTrequest_key,COUNT(*)AStotalFROMtool_jobGROUPBYrequest_keyHAVINGCOUNT(*)>1;

这些查询不是替代业务代码，而是上线后的独立核对。它们可以放进发布检查、定时巡检或告警看板；一旦出现结果，先阻止下载入口继续扩大影响，再根据任务日志和临时产物判断是状态转换还是存储提交出了问题。

九、异常边界与上线验收

最小产品也要明确“不自动替用户猜”的边界：

场景	系统应做什么	不能做什么
重复点击或网络重试	返回同一个`request_key`对应任务	再创建一个相同任务
Worker 超时或进程异常	标记`FAILED`，保留脱敏诊断和可重试依据	把半成品标记为`READY`
用户申请取消	标记取消请求，Worker 在安全点停止并清理临时产物	强行删除已正式交付且可能被下载的结果
存储提交失败	保持非`READY`，可从临时结果恢复或重新运行	只因命令退出成功就宣告完成
输入格式不支持	在受理阶段拒绝并解释支持范围	让 Worker 运行到中途才报模糊错误