简介:manus-manus 是一份围绕 manus 主题的数据处理与三维模型项目资源包,面向对 Python 工程组织、数据预处理及三维姿态数据感兴趣的开发者和研究者。包内共167个文件,体积仅24.19MB,其中以113个 Python 脚本为绝对主体,辅以 YAML 配置、Shell 环境脚本、Blender 三维模型、Pickle 数据文件、JSON 元数据及 Git 版本管理相关文件,能较完整呈现技术项目的常见组成,也适合快速了解基于 Python 的科研项目目录设计。从文件构成推断,项目使用 Git 进行版本控制,并借助子模块管理外部依赖,内部设有数据预处理、资源文件、源代码与数据集描述等目录,主程序、环境搭建脚本和说明文档一应俱全,便于梳理从数据准备到运行环境的整体脉络,并为二次开发提供清晰起点。内容预览中的多组 subject 三维模型文件,提示该资源可能与多对象姿态采集或三维重建相关,可作为该方向入门参考。目前已有293人学习,适合希望借助真实项目案例理解数据处理流程、工程化布局及三维数据组织方式的读者下载研究。
1. manus-manus:先把“手”这件事想清楚
manus-manus 这个代号第一次看很像某个仓库的归属名,拆开后是两个拉丁词“手”。我不打算把它当成某个现成框架的说明书,而是当做一个技术方向:把“手动操作”这件人人都会的平常事,拆成机器能执行、人能复核、出问题能回放的一套原语。这个方向解决的不是“能不能自动化”,而是“自动化的手伸出去之后,你怎么知道它做得对”。很多项目翻车,不是脚本写不出来,而是少盯了下游状态:页面提示成功,数据库里没有记录;接口返回 200,业务线程已报错。manus-manus 的核心产物是一组操作原语、一个执行状态机、一层校验与回放机制。它适合重复且不允许漏的活,比如内部数据录入、跨系统同步、定时巡检。新手在这篇文章里能拿到最小可跑的骨架,熟手可以重点看第四章的参数边界和踩坑记录。
2. 拆开 manus-manus:动作原语、状态机与调度骨架
2.1 为什么把操作拆成“原语”而不是写死流程
手工操作本质上是三层:感知当前状态、决策下一步动作、执行具体操作。把这三层揉进一个写死的流程脚本里,最容易出的问题是“只知道执行,不知道校验”。原语的意义在于把“动作”和“校验”变成同样大小的积木:click、upload、wait、assert 都可以算一个原语,每个原语有自己的参数、超时、重试和状态。这样做最大的好处是可以组合。第一期先做四个原语,第二期再加一个 download 原语,原有流程不用重写,只要在计划列表里插一行。
更隐蔽但更重要的点是状态可查询。自动化任务跑到一半卡住,最怕的是一点中间状态都没有。如果把每个原语包进一个状态机,从 PENDING、RUNNING 到 SUCCESS、FAILED、TIMEOUT,上层调度就能在失败时准确回答“最后一步是哪个原语、参数是什么”,而不是让人去翻日志猜。我一般会把状态机做得很薄,只维护一次执行实例的当前状态,不把业务字段塞进去。业务字段属于计划层,不属于状态层,混在一起之后状态机就会变成一个大泥球,改一处牵连一片。
2.2 最小可运行的原语与执行函数
下面这段是 manus-manus 思路里最核心的骨架:一个 Operation 描述一次“动手”,run_operation 负责执行、超时、重试。真正的动作怎么接,下一章会展开,这里先让状态机跑起来。
from dataclasses import dataclass, field from enum import Enum, auto import time, logging logger = logging.getLogger("manus_manus") class Status(Enum): PENDING = auto() RUNNING = auto() SUCCESS = auto() FAILED = auto() TIMEOUT = auto() @dataclass class Operation: op: str # 原语名:click/upload/wait/assert params: dict = field(default_factory=dict) timeout: float = 10.0 # 单个原语允许的总耗时,秒 retries: int = 2 # 失败后的重试次数,不含首次执行 idempotent: bool = True # 是否可安全重试 status: Status = Status.PENDING def mask_sensitive(params: dict) -> dict: # 日志里过滤 password/token/secret,避免把密钥打出去 keys = {"password", "token", "secret"} return {k: ("***" if k in keys else v) for k, v in params.items()} def execute_primitive(op: str, params: dict): """实际动作的插入点。 接界面用界面自动化驱动,接接口用 API 客户端,不在本骨架展开。""" if op == "wait": time.sleep(params.get("seconds", 1)) # 仅演示,真实场景多用轮询等待 elif op == "assert": if not params.get("condition", False): raise AssertionError(f"assert failed: {params}") else: raise NotImplementedError(op) def run_operation(op: Operation) -> Status: start = time.time() last_err = None for attempt in range(op.retries + 1): if attempt > 0 and not op.idempotent: # 非幂等动作不能重试,直接判失败,避免重复下单、重复扣款 break try: execute_primitive(op.op, op.params) op.status = Status.SUCCESS return op.status except Exception as exc: last_err = exc op.status = Status.FAILED if time.time() - start >= op.timeout: op.status = Status.TIMEOUT break time.sleep(min(2 ** attempt, 5)) # 指数退避,最多等 5 秒 logger.error("op failed op=%s err=%r", op.op, last_err) return op.status这段代码的核心,是把一次操作的最终结果收敛成五个状态之一。run_operation 按 retries 加一次尝试执行,每次失败先判断是否超过总超时:超时就标 TIMEOUT,而不是继续堆重试;没超时才进入下一次尝试。TIMEOUT 和 FAILED 在调度层是两类完全不同的处理:FAILED 可能是环境抖动,原地重试有机会成功;TIMEOUT 往往意味着业务链路卡死,继续重试只会加剧问题,多半要人工介入。
参数设计上有几个值得说透的地方。timeout 建议取正常耗时的 P95 再加 30%,不要拍脑袋填一个大数字。假的宽松超时会让系统等到天荒地老才报错。retries 默认给 2,但前提是 idempotent 为 True。如果动作是“创建订单”“发送短信”“扣减库存”,务必把 idempotent 改成 False,否则一次超时重试,业务侧就可能多出一笔脏数据。这里的退避用2 ** attempt而不是固定 sleep,是为了避免失败任务扎堆时把下游系统压得更死。
2.3 把它串成一个计划:状态与断点续跑
有了单个原语和执行函数,下一步是把它们按顺序排成 plan。run_plan 不直接改 plan 里的元素,而是返回独立的 results 列表,方便上层落库或者打追踪日志。
def run_plan(plan: list[Operation], stop_on_error: bool = True): """按顺序执行一组原语,返回每一步的结果快照。 stop_on_error=False 时,失败原语被记录,但流程继续。""" results = [] for index, op in enumerate(plan, start=1): status = run_operation(op) results.append({"index": index, "op": op.op, "status": status.name}) if status in (Status.FAILED, Status.TIMEOUT) and stop_on_error: break return resultsrun_operation 管“一个动作成不成”,run_plan 管“一个任务停不停”。stop_on_error 这个开关非常重要:给用户批量导入订单,遇到第一笔失败就停,不然你会收获一批半截数据,后续对账想死的心都有;给几十台机器做巡检,遇到第一台告警就停,反而会漏掉更大面积的异常。我一般把 stop_on_error 做成任务级参数,在执行计划时传入,不写死在代码里。
为什么不直接写 if-else 大法?因为一旦流程变成 if-else,你就失去了“执行到第几步、失败在哪一步、能否从那里续跑”的完整视图。把动作拆成原语之后,断点续跑可以做得非常朴素:把每个 Operation 的状态持久化到一张表,任务中断后,重启时过滤掉已经 SUCCESS 的步骤,从第一个非 SUCCESS 原语继续。没有原语层设计,这种恢复只能靠人从中间重新点一遍。
3. 把 manus-manus 接到真实系统:三种接入路线的落地对比
3.1 三种接入方式:界面层、接口层、业务层
同一套原语计划,接入层可以完全不同。我见过的项目里,最常见的落地路线有三条:走界面、走接口、走业务内部入口。选型不是选“最好的”,而是选“你能长期维护的”。
| 接入层 | 常见做法 | 优点 | 主要风险 |
|---|---|---|---|
| 界面层 | 用界面自动化驱动浏览器或桌面程序,模拟点击、输入、等待 | 对现有系统无侵入,最快看到效果 | 响应慢、选择器易失效、断言只能看到表层状态 |
| 接口层 | 直接调目标系统的 HTTP 或 RPC 接口,把操作变成一次请求 | 稳定、快、断言简单 | 拿不到前端交互状态,需要理解业务接口语义 |
| 业务层 | 在业务逻辑代码里增加一个操作入口,复用领域服务 | 最可控,事务、权限、回滚都在同一边 | 需要对方团队配合改造,跨部门成本高 |
我的默认顺序是:先问有没有接口,再问能不能在业务层加入口,最后才碰界面层。界面层放在最后不是因为效率低,而是因为它是三种方式里唯一看不见内部状态的黑匣子:脚本只能看到页面,看不到数据库和消息队列,出问题很难定位是界面问题还是业务问题。如果团队里没人熟悉目标系统的业务逻辑,界面层反而是最快验证 manus-manus 思路的方式,但一定要在计划后面补足后置校验。
3.2 最小可复现:用一个脚本把导入流程跑通
下面这段演示如何用前面的骨架搭一个真实的导入任务。execute_primitive 里接入层怎么实现,这里先不展开,但计划的结构是完整的。
def import_orders(csv_path: str) -> list[dict]: plan = [ Operation("assert", {"condition": file_exists(csv_path)}), Operation("upload", {"file": csv_path, "form_id": "upload-form"}), Operation("wait", {"seconds": 2}), Operation("assert", {"condition": result_panel_ok()}), ] return run_plan(plan, stop_on_error=True)这个 plan 一共四个原语,前两个是动作,后两个是校验。关键不在写得多复杂,而是“先断言再动手”:第一步先确认文件存在,避免拿一个空路径跑到上传才发现,白白把计划状态打成 FAILED。wait 这里为了演示用了固定等待,但实际项目里不要学,下一章会说为什么。upload 的 form_id 是界面层的选择器;如果走接口层,params 应该换成 endpoint、payload、headers 之类。
把界面层切换成接口层时,只需要重写 execute_primitive 的适配器,plan 本身基本不用动。比如 upload 原语,界面层实现是找到上传按钮并提交文件,接口层实现是构造 multipart 请求并校验响应;上层只关心 upload 这个原语最终是 SUCCESS 还是 FAILED。这种隔离价值很大,意味着前期原型用界面层跑通逻辑,后期性能要求上来后,可以平滑迁到接口层,业务计划不用重写。
3.3 日志与可观测性:操作不是执行完就结束
自动化任务的日志如果只是 print 一堆字符串,出问题时会非常痛苦。我习惯每个原语执行完都落一条结构化日志,一条 JSON 对应一次原语执行,方便按 trace_id 聚合查找。
import json, logging logger = logging.getLogger("manus_manus") def emit_trace(op: Operation, status: Status, trace_id: str): row = { "trace_id": trace_id, "op": op.op, "params": mask_sensitive(op.params), "status": status.name, "timeout": op.timeout, "retries": op.retries, } logger.info(json.dumps(row, ensure_ascii=False, sort_keys=True))mask_sensitive 这步不能省。很多操作原语里会带登录 token、接口密钥、支付密码,如果不过滤,日志就是泄漏出口。trace_id 要在任务开始时生成,贯穿同一批 plan 的所有原语,否则你只能看到“第三步失败”,看不到这一个任务整体跑了什么。每条日志里还应该带上执行实例的唯一标识,比如批次号、表单编号,这样和业务数据能对得上。
可观测性做到位之后,回放才可行。我一般会把每次执行的结果快照保存下来,保存的不是操作人的手速,而是每一步的入参、状态、耗时。下一次跑任务时,就能拿上一轮的快照做对比,这比肉眼盯屏幕可靠得多。
4. manus-manus 的五个高频踩坑点与排查方式
4.1 动作执行成功不等于业务目标达成
现象:脚本所有原语都返回 SUCCESS,但业务系统里根本没有新增记录。原因:把“请求已发送”或者“页面提示成功”当成了“业务处理完成”。上传接口可能只是把文件收下了,异步任务在后台抛了异常,前端照样显示成功。解决:每个关键动作后面必须跟一个业务后置断言,断言对象是数据库或者业务结果,不是页面文案。
Operation("upload", {"endpoint": "/orders", "payload": payload}, idempotent=False), Operation("assert", {"condition": order_count_increased()})这里的 assert 条件由接入层实现:查当前订单数,等它比执行前加一,并设超时。界面反馈可以当参考,不能当结论。
4.2 超时和重试会把“慢”变成“错”
现象:正常八秒跑完的任务,某天突然花了二十五秒,脚本自己重试了三次,最后把下游数据库连接池打满。原因:timeout 与 retries 配置过松,同时动作没标成非幂等。解决:非幂等动作必须把 idempotent 设为 False,重试前做退避,总超时按 P95 加三成。
| 场景 | timeout | retries | 说明 |
|---|---|---|---|
| 查询读取操作 | 5s | 2 | 可幂等,可放心重试 |
| 创建订单 | 30s | 0 | 非幂等,宁可失败后人工介入 |
| 批量导入 | 120s | 1 | 需要任务级幂等键,不能只靠系统判断 |
一个重要经验:把每次运行的实际耗时记录下来,定期看 P95。如果 P95 已经逼近 timeout,要调的首先是系统性能,而不是再加 timeout。
4.3 固定等待是最隐蔽的翻车点
现象:用sleep(5)当轮询,测的时候偶发失败,把等待加到八秒还是偶发失败。原因:固定等待不感知任务状态,页面渲染慢一秒就撞上,快一秒还在等。解决:改成轮询等待,条件满足立刻返回,整个等待有总超时兜底。
def wait_until(cond, timeout=30.0, interval=1.0): deadline = time.time() + timeout while time.time() < deadline: if cond(): return True time.sleep(interval) raise TimeoutError(f"wait_until timeout after {timeout}s")轮询间隔设一秒是通用选择;如果目标是本地接口,可以缩到 0.2 秒;如果目标是跨机房系统,建议放到两秒,减少无意义的请求。总超时仍然沿用 P95 加三成的原则。要注意 cond 函数内部不要自己再加 sleep,否则统计出来的实际耗时会被放大。
4.4 界面层级一变化,断言就成了黑匣子
现象:前一天全绿,今天全红,截图里元素只是挪了位置或者文案细微变化。原因:断言绑死了界面选择器和显示文案,界面一改就抓瞎。解决:给断言分等级。强断言查业务数据,比如数据库有没有新增记录;弱断言只查“结果面板是否出现”,不关心内容。强断言必须存在,弱断言只做烟雾验证。
如果只能部署一套环境做回归,尽量把强断言多放几个。界面层的弱断言即便全部通过,也只能证明“页面看起来正常”,不能证明“业务做对了”。这一点和 4.1 是同一类问题的两种表现,本质都是用了错误的校验信号。
4.5 手工基线不可比,回归等于白做
现象:优化了 timeout 和重试参数,拿旧的手工执行记录做对比,得出“新方案不如旧方案”的结论。原因:手工执行的时间、步骤顺序、等待时长都不可控,数据没有可比性。解决:建立自动化基线。同一组数据、同一环境、同一时间段,跑三次取中位数,作为回归对比基准。参数优化前后,都用这条自动化基线说话,不跟手工记录比。
我见过最典型的坑是:手工记录里包含人犹豫的 20 秒,自动化跑到 10 秒,但对比时把调试期间的异常重试也算进去,导致结论说自动化更慢。调参之前,先把基线的统计口径定义好:从哪个时刻开始计时,重试耗时不入基线,永远只看稳定态的中位数。
5. 让 manus-manus 学会回放:录制器与手动终止开关
前面的章节已经能跑通单次任务,但真正要长期用,还需要两件进阶工具:录制器和手动终止开关。录制器不是记录屏幕,而是记录“每个原语执行前后的状态”。人工操作时,把动作点一遍,同时把操作前的初始条件和操作后的业务状态抓下来,形成一组 Operation。回放时,先跑前置断言,再执行动作,最后跑后置断言。这和单元测试的 Arrange-Act-Assert 是同一个思路,只是对象从函数变成了真实业务操作。
class KillSwitch: def __init__(self): self._flag = False def stop(self): self._flag = True @property def armed(self) -> bool: return self._flag KILL_SWITCH = KillSwitch() def run_operation(op: Operation) -> Status: if KILL_SWITCH.armed: raise RuntimeError("manual stop requested") # ... 其余逻辑同之前手动终止开关必须放在每个原语执行前。它的价值是让操作员在发现“本轮数据不对劲”的时候,能立刻切断后续动作,而不是等计划把所有步骤跑完。我经历过一次线上数据被误批量修改,就是因为流程里校验全过了,但明显是源数据错了,却找不到入口让脚本停下。从那之后,每个 manus-manus 风格的任务我都默认带一个 kill switch,并且要求操作员知道它怎么触发。另一个习惯是每次上线前,把上一次成功跑通的任务快照作为预期,回放三遍取中位数。希望帮到你。
本文还有配套的精品资源,点击获取