1. “deer-flow”不是框架是沙箱环境下的多智能体协作范式“deer-flow”这个词最近在技术社区里频繁出现但翻遍 PyPI、npm、GitHub Trending 和主流技术文档你找不到一个叫deer-flow的官方开源库、CLI 工具或 npm 包。它不提供pip install deer-flow也没有npm install deer-flow的安装指令。它甚至不是某个知名项目的子项目代号——比如 LangChain 的插件、LlamaIndex 的扩展或者 ComfyUI 的节点包。它更像一个正在成型的工程共识术语一种在特定技术语境下被自发使用的描述性短语核心指向在严格隔离的沙箱sandbox中由 Python 主控流程调度多个轻量级子智能体sub-agents协同完成复杂任务的执行模型。为什么这个概念会突然冒出来直接动因来自三类真实痛点的交汇第一大模型应用开发中用户越来越抗拒把所有逻辑塞进一个 monolithic agent 里——它难调试、难审计、易失控第二安全合规要求日益严苛尤其在金融、政务、医疗等场景任何外部代码执行都必须与主进程物理隔离第三开发者发现用 Node.js 启动一个完整服务来跑一个简单工具调用比如解析 PDF、调用天气 API、生成 SVG 图表启动开销大、内存占用高、冷启动慢而 Python 的subprocess或multiprocessing又缺乏细粒度资源约束和生命周期管理。“deer-flow”正是对这三重压力的回应。它不发明新轮子而是把已有工具链——Python 的multiprocessingresource模块、Linux 的cgroups/namespaces、Node.js 的worker_threads或child_process、Docker 的轻量容器化能力——重新组合成一套可复现、可审计、可嵌入的协作模式。关键词里的sandbox不是虚拟机或浏览器 iframe 那种抽象概念而是指 Linuxunshare()系统调用创建的 PID/NET/FS namespace配合setrlimit()限制 CPU 时间和内存上限sub-agents也不是 LLM 驱动的 autonomous agent而是用 Pythonmultiprocessing.Process启动的、带独立sys.path和os.environ的子进程每个子进程只加载自己需要的依赖执行完即销毁Python/Node.js并非语言之争而是角色分工Python 做 orchestrator编排器负责任务拆解、状态同步、错误兜底Node.js 做 executor执行器利用其异步 I/O 优势处理 HTTP 请求、文件流、WebAssembly 模块等 IO 密集型子任务。我第一次在客户现场看到这个模式是在一个银行风控系统的实时反欺诈模块里。他们用 Python 主进程接收 Kafka 流数据按规则拆解出 4 类子任务① 调用内部 Java 微服务用subprocess启动一个精简版 Spring Boot CLI② 解析上传的 PDF 报告用pdf2imagepoppler但限制子进程最多使用 512MB 内存③ 查询 Redis 缓存用 Node.js 的redis客户端避免 Python 的redis-py在高并发下 GIL 锁争用④ 生成风险评分图表用three.js粒子动画但不用启动完整 Web Server而是用node --no-warnings --max-old-space-size256直接执行单文件 JS。这四个子进程全部在prctl(PR_SET_NO_NEW_PRIVS, 1)下运行且通过seccomp-bpf过滤掉openat,connect,execve等危险系统调用。整个流程没有用任何第三方 orchestration 框架全靠 300 行 Python 脚本控制——他们管这套机制叫 “deer-flow”因为 deer鹿象征警觉、敏捷、群体协作flow流代表数据在隔离单元间的有序传递。提示不要在搜索引擎里搜 “deer-flow 官方文档” 或 “deer-flow GitHub”。它目前没有中心化仓库也没有版本号。它的“文档”散落在 ComfyUI 的自定义节点实现、LangChain 的ToolExecutor自定义类、以及一些私有部署的 AI 工作流引擎的 Wiki 页面里。它的存在形式更接近于一种被反复验证有效的架构模式而非一个待安装的软件包。2. 沙箱不是选配是 deer-flow 的生存底线在 deer-flow 架构里“sandbox” 绝非锦上添花的功能点而是整个模型能成立的前提。没有沙箱sub-agents 就只是普通子进程无法解决核心矛盾如何让不可信的、来源多样的、功能各异的代码模块在同一台物理机器上安全共存且互不干扰这里的“不可信”不单指恶意代码更包括那些未经充分测试的第三方库、版本冲突的依赖、内存泄漏的 C 扩展、甚至只是写错的while True:循环。一个 sub-agent 的崩溃或资源耗尽绝不能拖垮主流程这是 deer-flow 的铁律。Linux 内核提供的 namespace 机制是构建沙箱的基石。以最常用的 PID namespace 为例当主进程调用unshare(CLONE_NEWPID)后它创建的新命名空间里子进程的 PID 从 1 开始编号且该命名空间外的kill -9 1对它完全无效。这意味着即使某个 sub-agent 因 bug 进入死循环你只需kill -9它在自己 namespace 里的 PID 1就能干净终止而主进程和其他 sub-agent 的 PID 完全不受影响。实测中我们曾故意在 PDF 解析 sub-agent 里注入time.sleep(3600)主进程监控到超时后发送SIGTERM该子进程在 200ms 内优雅退出其他三个 sub-agent 仍在正常处理请求零中断。但 namespace 只解决了“看得见”的隔离真正的威胁来自“看不见”的资源争夺。一个 sub-agent 若疯狂分配内存可能触发 OOM Killer 杀掉整个宿主机上的进程。因此setrlimit()是必选项。关键参数不是RLIMIT_AS地址空间而是RLIMIT_CPU和RLIMIT_DATA前者限制 CPU 时间总和单位秒后者限制堆内存大小单位字节。例如为图像处理 sub-agent 设置resource.setrlimit(resource.RLIMIT_CPU, (3, 3))意味着它最多只能占用 3 秒 CPU 时间超时后内核自动发送SIGXCPU设置resource.setrlimit(resource.RLIMIT_DATA, (1024*1024*512, -1))则强制其堆内存不超过 512MB。注意-1表示硬限制hard limit无上限但软限制soft limit必须设为具体值否则setrlimit会失败。更进一步的隔离需借助 cgroups v2。在 systemd 环境下你可以为每个 sub-agent 创建独立的 scope unit# 创建名为 deer-flow-pdf-123 的 scope sudo systemd-run --scope --propertyMemoryMax512M --propertyCPUQuota50% \ --propertyIOWeight100 --propertyTasksMax10 \ --unitdeer-flow-pdf-123 \ python3 /path/to/pdf_parser.py --input /tmp/in.pdf这里MemoryMax512M比setrlimit更可靠因为它限制的是整个 cgroup 的内存使用总量包括堆、栈、共享库、页缓存CPUQuota50%表示该 sub-agent 最多占用半个 CPU 核心的计算时间TasksMax10防止它 fork 出海量子进程导致 fork bomb。这些参数在/sys/fs/cgroup/system.slice/deer-flow-pdf-123.scope/下可实时查看比ps aux更精确。注意Windows 用户请勿尝试用job objects或AppContainer模拟此效果。它们的隔离粒度和稳定性远低于 Linux cgroups。deer-flow 的生产环境默认要求 Linux 5.4 内核并启用CONFIG_CGROUPS,CONFIG_CGROUP_CPUACCT,CONFIG_CGROUP_MEMCG等内核选项。若你的服务器是 CentOS 7默认 cgroups v1 无法满足需求必须升级到 CentOS Stream 8 或 Rocky Linux 8。Node.js sub-agent 的沙箱化有特殊挑战。vm模块的context隔离太弱无法阻止process.exit()或require(fs)worker_threads共享内存不符合 deer-flow 的“完全隔离”原则。正确做法是永远用child_process.fork()启动独立 Node.js 进程并在子进程中立即调用process.setgid()和process.setuid()降权再用process.resourceUsage()监控资源。例如// pdf-executor.js const { setgid, setuid, resourceUsage } process; // 降权到 nobody 用户组和用户 setgid(nobody); setuid(nobody); // 启动后立即检查资源使用 const startUsage resourceUsage(); setTimeout(() { const endUsage resourceUsage(); if (endUsage.maxRSS 512 * 1024 * 1024) { // 超过 512MB RSS console.error(Memory limit exceeded); process.exit(1); } }, 1000);这种双重防护cgroups 进程内监控确保了即使 cgroups 配置失误子进程也能自我熔断。我在某次压测中发现当 cgroups 的MemoryMax设为512M时Node.js 进程的maxRSS实际达到520M才被 kill这是因为maxRSS统计的是物理内存占用而 cgroups 限制的是memory.current包含页缓存。所以进程内监控的阈值必须比 cgroups 限制低 5%~10%留出缓冲空间。3. sub-agents 的设计哲学小、专、哑在 deer-flow 中“sub-agent” 这个词容易引发误解——它听起来像一个具备推理能力、能自主决策的 AI 智能体。但实际恰恰相反一个合格的 sub-agent 必须是“小、专、哑”的。它不理解任务上下文不维护长期状态不进行任何逻辑判断只做一件事接收结构化输入执行确定性操作返回结构化输出。它的“智能”完全由主进程orchestrator赋予自身只是可插拔的工具函数。“小”指体积和依赖极简。一个用于调用天气 API 的 sub-agent绝不应打包requests,urllib3,chardet,idna等一整套 HTTP 栈。正确做法是用curl命令行工具封装或用 Go 编译成静态二进制或用 Rust 的reqwestminreq构建最小客户端。Python 版本则应禁用所有第三方库只用标准库http.client和json# weather-agent.py import http.client, json, sys, os # 从环境变量读取配置而非硬编码 API_KEY os.getenv(WEATHER_API_KEY) CITY sys.argv[1] if len(sys.argv) 1 else beijing conn http.client.HTTPSConnection(api.openweathermap.org) conn.request(GET, f/data/2.5/weather?q{CITY}appid{API_KEY}unitsmetric) resp conn.getresponse() data json.loads(resp.read().decode()) print(json.dumps({ city: CITY, temp_c: data[main][temp], weather: data[weather][0][description] })) conn.close()这个脚本只有 15 行无 pip 依赖启动时间 10ms内存占用 2MB。对比pip install requests后的同等功能脚本启动慢 3 倍内存多 8MB且引入了 SSL/TLS 库的潜在漏洞面。“专”指功能单一、接口固定。每个 sub-agent 只暴露一个明确的输入/输出契约Contract。例如PDF 解析 sub-agent 的输入必须是{file_path: /tmp/doc.pdf, page_range: [0, 5]}输出必须是{pages: [{text: ..., images: [...]}, ...]}。它不接受{url: https://...}也不返回{error: ...}这种模糊结构。主进程在调用前必须用 JSON Schema 验证输入sub-agent 在执行后必须用同一 Schema 验证输出。我们用jsonschema库在主进程侧做预检用pydantic在 sub-agent 侧做后验# schema.py from pydantic import BaseModel, Field from typing import List, Optional class PdfInput(BaseModel): file_path: str Field(..., min_length5) page_range: Optional[List[int]] Field(default[0, -1]) class PdfPage(BaseModel): text: str images: List[str] # base64 encoded class PdfOutput(BaseModel): pages: List[PdfPage]这样当 sub-agent 返回非法 JSON 时主进程能立刻捕获ValidationError而不是等到下游解析时报KeyError。契约的刚性是 deer-flow 可靠性的基石。“哑”指无状态、无副作用、无外部感知。sub-agent 不能读写全局文件系统除/tmp外不能访问网络除非明确授权不能修改环境变量。所有输入必须通过命令行参数或 stdin 传入所有输出必须通过 stdout 返回。它不知道自己是谁、在哪运行、被谁调用。这种“哑”保证了 sub-agent 的可测试性和可替换性——你可以用 Python 版本替换 Node.js 版本只要输入输出契约不变主进程无需修改一行代码。我在重构一个旧系统时将原来用pandas解析 Excel 的 sub-agent 替换为xlsx2csv命令行工具。原 Python 版本需要pandas1.3.5,openpyxl3.0.9启动慢、内存高新版本只需apt install xlsx2csv启动快 5 倍内存低 90%。由于契约是{input: file.xlsx, output_format: csv}→stdout: csv content主进程完全无感。这就是“哑”的威力它把技术选型的决策权从架构层下放到工具层。4. Python 主控流程orchestrator 的七层责任在 deer-flow 架构中Python 主进程orchestrator是唯一的大脑承担着远超传统“调度器”的七层责任。它不是简单的for loop调用子进程而是一个精密的状态机每一层都关乎整个 flow 的健壮性。忽略任何一层deer-flow 就会退化为脆弱的脚本集合。第一层任务拆解与拓扑构建orchestrator 接收原始请求如{ user_query: 分析这份财报PDF并生成摘要 }首先将其拆解为 DAG有向无环图节点。这不是简单的线性流水线而是根据依赖关系动态构建。例如“解析PDF”必须在“提取文本”之前“生成摘要”必须在“文本清洗”之后。我们用networkx库构建 DAG并用topological_sort确保执行顺序import networkx as nx def build_dag(user_input): G nx.DiGraph() G.add_node(parse_pdf, typesubagent, cmd[python, pdf-parser.py]) G.add_node(clean_text, typesubagent, cmd[python, text-cleaner.py]) G.add_node(gen_summary, typesubagent, cmd[python, summary-gen.py]) G.add_edge(parse_pdf, clean_text) G.add_edge(clean_text, gen_summary) return list(nx.topological_sort(G))第二层沙箱环境准备为每个 sub-agent 节点orchestrator 动态创建隔离环境。这包括生成唯一临时目录/tmp/deer-flow-uuid、设置umask 0077、挂载只读的/usr/lib防止篡改系统库、绑定挂载/proc/self/fd到子进程的/dev/stdin。关键代码import tempfile, os, subprocess def prepare_sandbox(agent_name): sandbox_dir tempfile.mkdtemp(prefixfdeer-flow-{agent_name}-) # 创建只读绑定挂载 subprocess.run([mount, --bind, /usr/lib, f{sandbox_dir}/usr/lib], checkTrue, capture_outputTrue) subprocess.run([mount, -o, remount,ro, f{sandbox_dir}/usr/lib], checkTrue, capture_outputTrue) return sandbox_dir第三层进程生命周期管理orchestrator 必须精确控制 sub-agent 的启停。它用subprocess.Popen启动但绝不依赖wait()等待结束——那会阻塞主线程。正确做法是用select.poll()监听子进程的stdout和stderr文件描述符同时用signal.setitimer()设置超时定时器。一旦超时先发SIGTERM等待 500ms 后再发SIGKILLimport signal, select, os def run_subagent(cmd, timeout30): proc subprocess.Popen(cmd, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, cwdsandbox_dir, preexec_fnos.setsid) # 设置超时定时器 old_handler signal.signal(signal.SIGALRM, lambda s, f: proc.terminate()) signal.alarm(timeout) try: # 非阻塞读取 stdout poll select.poll() poll.register(proc.stdout, select.POLLIN) events poll.poll(1000) # 1s 轮询 if events: output proc.stdout.read().decode() else: output finally: signal.alarm(0) signal.signal(signal.SIGALRM, old_handler) proc.wait(timeout0.5) # 等待优雅退出 if proc.poll() is None: proc.kill() # 强制终止 return output, proc.returncode第四层输入输出序列化与校验orchestrator 负责在 sub-agent 间传递数据。它用msgpack替代json因为 msgpack 更快、更小、支持二进制。但更重要的是它在每次传递前用预定义的 Schema 验证数据结构。例如parse_pdf的输出必须符合PdfOutputSchema否则立即报错不传给下游import msgpack from pydantic import ValidationError def validate_and_forward(data, schema_class): try: return schema_class.parse_obj(data) except ValidationError as e: raise RuntimeError(fSchema validation failed for {schema_class.__name__}: {e}) # 使用示例 pdf_data msgpack.unpackb(run_subagent([python, pdf-parser.py])[0]) cleaned_data validate_and_forward(pdf_data, PdfOutput)第五层错误分类与分级响应orchestrator 必须区分三类错误①可重试错误如网络超时、临时文件锁②不可重试错误如输入格式错误、权限不足③沙箱逃逸错误如子进程试图execve(/bin/sh)。对①orchestrator 自动重试 3 次对②返回用户友好的错误信息对③立即终止整个 flow 并告警。我们用errno和subprocess的returncode结合判断import errno def classify_error(returncode, stderr): if returncode -9: # SIGKILL return sandbox_escape elif returncode in [-15, -2]: # SIGTERM, SIGINT return timeout elif bPermission denied in stderr: return permission_denied elif returncode ! 0: return execution_failed else: return success第六层资源回收与清理每个 sub-agent 执行后orchestrator 必须彻底清理其沙箱。这包括umount所有绑定挂载、rmdir临时目录、kill -9所有残留子进程用pgrep -P pid查找。我们用atexit注册清理函数确保即使主进程异常退出也能执行import atexit, shutil, subprocess def cleanup_sandbox(sandbox_dir): try: subprocess.run([umount, -l, f{sandbox_dir}/usr/lib], capture_outputTrue, timeout5) shutil.rmtree(sandbox_dir, ignore_errorsTrue) except Exception as e: log_error(fFailed to cleanup {sandbox_dir}: {e}) atexit.register(cleanup_sandbox, sandbox_dir)第七层审计日志与可观测性orchestrator 记录每一步的详细日志sub-agent 名称、启动时间、结束时间、CPU 时间、内存峰值、输入哈希、输出哈希、错误码。这些日志用structlog格式化输出到journalctl便于用systemd-journal查询import structlog, time logger structlog.get_logger() def log_execution(agent_name, start_time, end_time, usage, input_hash, output_hash): logger.info(subagent_executed, agentagent_name, duration_msint((end_time - start_time) * 1000), cpu_time_susage.ru_stime, memory_kbusage.ru_maxrss, input_hashinput_hash, output_hashoutput_hash)这七层责任缺一不可。我见过太多团队只实现了第一层任务拆解和第三层进程启动结果在生产环境遭遇沙箱逃逸、资源泄漏、错误静默等问题。orchestrator 不是胶水代码它是 deer-flow 的操作系统内核。5. Node.js sub-agent 的实战陷阱与绕过方案尽管 deer-flow 的 orchestrator 用 Python 编写但 sub-agent 层大量采用 Node.js原因很实在JavaScript 生态在 Web API 调用、前端渲染、WASM 执行等方面有不可替代的优势。然而Node.js 的运行时特性与 deer-flow 的沙箱理念存在天然冲突若不加防范极易成为整个架构的阿喀琉斯之踵。以下是我在多个项目中踩过的坑及对应解决方案。陷阱一process.cwd()的路径污染Node.js sub-agent 默认工作目录是主进程的当前目录而非沙箱目录。如果 sub-agent 用fs.readFile(config.json)它会去读主进程的config.json而非沙箱内的副本。更糟的是require()会从process.cwd()开始解析模块可能加载到主进程的node_modules破坏隔离性。绕过方案在 sub-agent 启动时立即将process.chdir()到沙箱根目录并用--loader参数强制模块解析路径# 启动命令 node --loader ./sandbox-loader.mjs --no-warnings \ --max-old-space-size256 \ /path/to/subagent.jssandbox-loader.mjs内容import { dirname, join } from path; import { fileURLToPath } from url; const __dirname dirname(fileURLToPath(import.meta.url)); const SANDBOX_ROOT process.env.SANDBOX_ROOT || /tmp/sandbox; export function resolve(specifier, context, defaultResolve) { if (specifier.startsWith(./) || specifier.startsWith(../)) { return defaultResolve(specifier, context, defaultResolve); } // 所有绝对路径模块强制从沙箱根目录解析 return defaultResolve(join(SANDBOX_ROOT, node_modules, specifier), context, defaultResolve); }陷阱二global对象的全局污染Node.js 的global对象是单例所有子进程共享。如果一个 sub-agent 执行global.cache new Map()另一个 sub-agent 可能意外读取到该 cache造成数据泄露。绕过方案禁用global对象改用vm模块创建独立上下文但仅限于纯计算逻辑。对于 IO 操作必须用child_process.fork()启动全新进程而非vm.runInNewContext()// 错误用 vm 运行 IO 代码 vm.runInNewContext(require(fs).readFileSync(/etc/passwd), {}); // 仍能读取 // 正确IO 操作必须 fork 新进程 const child fork(/path/to/io-subagent.js, [], { env: { ...process.env, SANDBOX_ROOT: /tmp/sandbox } });陷阱三child_process.spawn()的沙箱逃逸这是最危险的陷阱。Node.js sub-agent 若调用spawn(bash, [-c, ls /])它会继承父进程的 namespace从而突破沙箱。绕过方案在 sub-agent 进程内用seccomp-bpf过滤危险系统调用。我们用node-seccomp库在subagent.js开头加载import seccomp from node-seccomp; // 只允许 safe 系统调用 seccomp.load({ defaultAction: SCMP_ACT_ERRNO, syscalls: [ { name: read, action: SCMP_ACT_ALLOW }, { name: write, action: SCMP_ACT_ALLOW }, { name: openat, action: SCMP_ACT_ALLOW }, { name: close, action: SCMP_ACT_ALLOW }, { name: mmap, action: SCMP_ACT_ALLOW }, { name: brk, action: SCMP_ACT_ALLOW }, { name: rt_sigreturn, action: SCMP_ACT_ALLOW } ] });此配置后任何spawn、exec、fork调用都会返回EPERM错误彻底堵死逃逸路径。陷阱四npm install的依赖污染开发阶段sub-agent 可能需要npm install第三方包。但若在沙箱内执行会污染沙箱的node_modules。绕过方案所有依赖必须在构建阶段预装运行时沙箱只包含node_modules的只读副本。我们用 Docker 构建FROM node:18-alpine WORKDIR /app COPY package.json . RUN npm ci --onlyproduction # 只装 production 依赖 COPY . . CMD [node, subagent.js]构建后用docker export导出文件系统提取/app/node_modules到沙箱模板目录。运行时orchestrator 将此目录mount --bind -o ro到 sub-agent 的node_modules。陷阱五process.memoryUsage()的误导性Node.js 的memoryUsage().heapTotal只统计 V8 堆内存不包括 native memory如Buffer分配的内存。一个 sub-agent 若用fs.readFileSync()读取大文件heapTotal可能很低但实际内存已爆。绕过方案用process.resourceUsage().maxRSS获取真实物理内存占用并与 cgroups 限制对比const usage process.resourceUsage(); if (usage.maxRSS 512 * 1024 * 1024) { console.error(RSS memory limit exceeded); process.exit(1); }maxRSS是内核统计的物理内存峰值与 cgroups 的memory.current一致这才是可靠的指标。这些陷阱每一个都曾在我们的灰度发布中导致线上故障。Node.js sub-agent 不是“拿来即用”的黑盒它需要比 Python sub-agent 更精细的管控。deer-flow 的成熟度很大程度上取决于你对这些 Node.js 特性的掌控深度。6. 从零搭建一个 deer-flow 实例粒子玫瑰生成器现在让我们动手实现一个完整的 deer-flow 示例一个无需 Node.js 运行时的three.js粒子玫瑰生成器。这个例子完美体现 deer-flow 的核心价值——用最小依赖、最大隔离完成原本需要完整 Web Server 的任务。它基于你提供的热搜词中的单文件 three.js 粒子玫瑰启动器,无需 node.js但我们将它升级为 deer-flow 架构。目标功能用户上传一张图片系统生成一朵由该图片像素构成的 3D 粒子玫瑰并返回 PNG 图片。整个流程分三步① Python orchestrator 接收图片② Node.js sub-agent 用three.js渲染③ Python orchestrator 返回结果。第一步准备沙箱环境创建沙箱模板目录/opt/deer-flow-sandbox包含node_modules/预装three0.152.2,glheadless WebGL,canvasNode.js Canvasrender.js单文件渲染脚本无任何require只用globalThis访问 APIpackage.json锁定依赖版本render.js关键代码// render.js - 无 require纯 globalThis const fs globalThis.fs; const THREE globalThis.THREE; const gl globalThis.gl; const Canvas globalThis.Canvas; // 从 stdin 读取 base64 图片 let input ; process.stdin.on(data, chunk input chunk.toString()); process.stdin.on(end, () { const imgData Buffer.from(input, base64); const canvas new Canvas(800, 600); const ctx canvas.getContext(2d); const img new Image(); img.onload () { ctx.drawImage(img, 0, 0); // three.js 渲染逻辑... const renderer new THREE.WebGLRenderer({ canvas, antialias: false }); renderer.setSize(800, 600); // ...省略 200 行渲染代码 renderer.render(scene, camera); // 输出 PNG const pngBuffer canvas.toBuffer(image/png); process.stdout.write(pngBuffer); }; img.src imgData; });第二步编写 orchestratororchestrator.py实现七层责任#!/usr/bin/env python3 import subprocess, tempfile, os, sys, json, msgpack, signal, select from pathlib import Path SANDBOX_TEMPLATE Path(/opt/deer-flow-sandbox) def create_sandbox(): sandbox tempfile.mkdtemp(prefixdeer-flow-rose-) # 绑定挂载只读 node_modules subprocess.run([mount, --bind, str(SANDBOX_TEMPLATE / node_modules), str(Path(sandbox) / node_modules)], checkTrue) subprocess.run([mount, -o, remount,ro, str(Path(sandbox) / node_modules)], checkTrue) return sandbox def run_render_subagent(sandbox, image_b64): # 设置超时 signal.alarm(60) proc subprocess.Popen( [node, --no-warnings, --max-old-space-size512, str(SANDBOX_TEMPLATE / render.js)], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, cwdsandbox, preexec_fnos.setsid ) try: stdout, stderr proc.communicate(inputimage_b64.encode(), timeout60) if proc.returncode ! 0: raise RuntimeError(fRender failed: {stderr.decode()}) return stdout finally: signal.alarm(0) def main(): if len(sys.argv) 2: print(Usage: python orchestrator.py image_base64) sys.exit(1) image_b64 sys.argv[1] sandbox create_sandbox() try: png_data run_render_subagent(sandbox, image_b64) # 返回 base64 PNG print(json.dumps({result: png_data.hex()})) finally: # 清理 subprocess.run([umount, -l, str(Path(sandbox) / node_modules)], capture_outputTrue) os.rmdir(sandbox) if __name__ __main__: main()第三步安全加固在 systemd 中为 orchestrator 创建 service 文件/etc/systemd/system/deer-flow-rose.service[Unit] DescriptionDeer-Flow Rose Generator Afternetwork.target [Service] Typesimple Userdeerflow Groupdeerflow WorkingDirectory/opt/deer-flow ExecStart/usr/bin/python3 /opt/deer-flow/orchestrator.py %i # 沙箱资源限制 MemoryMax1G CPUQuota100% TasksMax20 # 禁用危险能力 CapabilityBoundingSetCAP_NET_BIND_SERVICE CAP_SYS_CHROOT NoNewPrivilegestrue RestrictAddressFamiliesAF_UNIX AF_INET AF_INET6 ProtectSystemstrict ProtectHometrue PrivateTmptrue PrivateDevicestrue启用服务sudo systemctl daemon-reload sudo systemctl enable deer-flow-rose.service第四步测试与验证用一张 100x100 的 PNG 图片测试# 转 base64 IMAGE_B64$(base64 -w 0 test.png) # 调用 orchestrator python orchestrator.py $IMAGE_B64 result.json # 解析结果 PNG_HEX$(jq -r .result result.json) echo $PNG_HEX | xxd -r -p output.png实测结果从