企业官网建设流程全解析

Opik Python Backend 是 Opik 平台中专司"安全执行用户 Python 评估代码"的独立服务：它将在线评测（Online Evaluation）中提交的 Python 指标/评分脚本放入沙箱执行，并以 JSON 形式返回评分结果。本文围绕 apps/opik-python-backend/README.md 展开，结合仓库源码，完整讲解该服务的执行策略、全部环境变量、本地调试与生产部署方式，以及底层容器池/进程池与沙箱安全机制，帮助你掌握如何安装、配置并运维这一执行引擎。

服务定位：为什么需要独立的 Python 执行后端

Opik 的整体架构中，用户会提交自定义的 Python 评估指标（如ScoreResult类型的打分函数）参与在线评测（Online Evaluation）。这些代码来自用户侧、运行在服务端，存在两条核心诉求：

1. 安全隔离：用户代码可能访问网络、读写文件系统、死循环或耗尽内存，必须与主服务隔离；
2. 标准化契约：执行结果必须是机器可读的结构化数据，便于 Java 后端统一消费。

Opik Python Backend 就是为此而生的服务。从 README 的定义看，它是一个"在沙箱环境中运行 Python 代码的服务"：默认通过 Docker 容器执行（生产形态），也可以在派生子进程中运行（用于开发或非受限环境）。它对外暴露一个 Flask 应用，核心接口为POST /v1/private/evaluators/python（见 evaluator.py）。

该接口的调用契约非常简洁：请求体包含code（用户编写的评估代码）和data（待评估数据），服务将其交给执行器（Executor）运行，返回{"scores": [...]}。若执行失败则返回对应的 HTTP 错误码。

快速开始：本地环境准备

根据 README，运行本服务需要以下前置条件：

• 安装 Docker；
• 安装 Python；
• 创建并启用 Python 虚拟环境；
• 从 requirements.txt 安装全部依赖；
• 运行测试时，额外安装 tests/test_requirements.txt 中的依赖。

依赖清单中的核心组件包括：Flask（Web 框架）、gunicorn（生产 WSGI 服务器）、docker（Docker SDK，驱动容器执行器）、opik与opik-optimizer（评估/优化 SDK，供被执行的用户代码调用）、redis与rq（可选的异步优化任务队列）、以及一整套opentelemetry-*组件（可观测性埋点）。

两种执行策略：Docker 容器 vs. 子进程

服务最关键的开关是环境变量PYTHON_CODE_EXECUTOR_STRATEGY，它决定代码由哪种执行器承载。初始化逻辑位于 evaluator.py：

• docker：使用 DockerExecutor，从镜像池拉取沙箱容器执行代码，生产环境默认值；
• process（或留空）：使用 ProcessExecutor，在可复用的子进程池中执行，适合本地开发或不受限环境；
• 其他取值：启动时直接抛出ValueError。

两个执行器都继承自 executor.py 中的CodeExecutorBase，共享并行度、执行超时与池获取超时的配置读取逻辑，并各自实现run_scoring(code, data, payload_type)抽象方法。这意味着切换策略不需要改动任何上层调用代码。

核心环境变量详解

README 完整列出了服务的核心环境变量，下面结合源码逐一展开其含义、默认值与取值影响。

补充说明：超出 README 的进阶配置

源码中还提供了若干 README 未展开但同样重要的变量，属于运维调优的进阶入口：

• PYTHON_CODE_EXECUTOR_POOL_ACQUIRE_TIMEOUT_IN_SECS：从池中获取空闲执行器的最长等待时间，默认0.0（快速失败，池满即返回 HTTP 503）。源码注释明确指出，突发流量应由 HTTP 层的重试退避来吸收，而不是让服务端请求线程长时间挂起等待（见 executor.py）。如果业务流量特征适合短暂等待，可以适当调高。
• PYTHON_CODE_EXECUTOR_POOL_CHECK_INTERVAL_IN_SECONDS：后台守护线程检查并补充池内容器/进程的周期，默认3秒（见 executor_docker.py）。
• 镜像相关变量：PYTHON_CODE_EXECUTOR_IMAGE_REGISTRY（默认ghcr.io/comet-ml/opik）、PYTHON_CODE_EXECUTOR_IMAGE_NAME（默认opik-sandbox-executor-python）、PYTHON_CODE_EXECUTOR_IMAGE_TAG，共同拼接出沙箱镜像的完整引用（见 executor_docker.py）。这些默认值也在 Dockerfile 中以 ENV 形式固化。

关键参数的影响面

• PARALLEL_NUM：同时决定容器池/进程池大小与并发能力。在 Docker 执行器里它同时作为scoring_executor线程池的max_workers（见 executor_docker.py）；在生产 entrypoint 中，gunicorn 的--threads也取该值，保证 Web 线程与执行器池容量匹配（见 entrypoint.sh）。
• EXEC_TIMEOUT_IN_SECS：Docker 执行器中通过future.result(timeout=...)强制限制单次exec_run的等待时间，超时返回 HTTP 504EXEC_TIMEOUT_ERROR（见 executor_docker.py）；Process 执行器中则通过connection.poll(timeout=...)实现等价的超时语义（见 executor_process.py）。
• CPU_LIMIT与CPU_SHARES的差异：前者是硬性上限（内部转换为 Docker SDK 的nano_cpus，如0.5转为500000000），后者是软性优先级权重；二者可以叠加使用（见 executor_docker.py）。

本地运行 Flask 服务（Debug 模式）

README 给出了本地开发的标准启动方式，需要在apps/opik-python-backend目录下执行：

flask --app src/opik_backend --debug run

• --app src/opik_backend指向模块入口，Flask 会调用模块内的create_app()工厂函数构建应用（见init.py）；
• --debug开启调试模式，代码修改后自动重载，适合开发期使用；
• 服务默认监听http://localhost:5000。

需要注意的是：调试模式下 Flask 的重载器会启动两个进程，为避免执行器（尤其是进程池）被重复初始化，init_executor对process策略做了保护——只有当WERKZEUG_RUN_MAIN=true或非 debug 模式时才调用start_services()（见 evaluator.py）。

应用启动时还会依次注册三个 Blueprint：健康检查（healthcheck）、评估执行（evaluator）与用户注册后处理（post_user_signup），并在OPIK_OTEL_SDK_ENABLED=true时初始化 OpenTelemetry（见init.py）。

手动验证一次评估请求

启动服务后，可以向核心接口发起一次评估调用：

curl -X POST http://localhost:5000/v1/private/evaluators/python \ -H "Content-Type: application/json" \ -d '{ "code": "from opik.evaluation.metrics import base_metric, score_result\nresult = {\"scores\": [{\"value\": 0.95, \"name\": \"my_metric\", \"reason\": \"ok\"}]}\nprint(json.dumps(result))", "data": {"input": "hello"} }'

返回{"scores": [...]}即表示沙箱链路打通。接口会对缺失字段、空评分结果等情况返回 400（见 evaluator.py）。

生产部署：Docker 镜像与 entrypoint 启动链路

生产环境以 Docker 镜像方式运行，Dockerfile 采用多阶段构建：

• 构建阶段：基于docker:29.5.1（Alpine）安装python3、gcc、rust/cargo等原生编译工具链，用uv将依赖安装进/opt/venv，随后把依赖编译为 bytecode-only（.pyc）布局以缩小镜像体积；
• 运行阶段：仅保留tini（PID 1 收割器）、python3与精简后的 venv，EXPOSE 8000，并将镜像内默认策略设为docker、并行度 5、超时 3 秒、禁止网络（见 Dockerfile）；
• 沙箱执行器镜像（opik-sandbox-executor-python）通过 tar 包COPY进镜像，运行时由 entrypoint 执行docker load导入；若未随镜像打包，则会在首次使用时按PYTHON_CODE_EXECUTOR_IMAGE_REGISTRY/NAME/TAG拉取。

entrypoint.sh 的启动流程分为三步：

1. 启动 Docker daemon：仅当策略为docker时，后台拉起dockerd-entrypoint.sh，并以 1 秒间隔最多重试 30 次等待docker info可用；若 30 次后仍失败则直接退出（见 entrypoint.sh）；
2. 导入沙箱镜像：若./images/${PYTHON_CODE_EXECUTOR_ASSET_NAME}.tar.gz非空则docker load（见 entrypoint.sh）；
3. 启动 gunicorn：单 worker +gthread线程模型，--threads取PYTHON_CODE_EXECUTOR_PARALLEL_NUM（默认 5），监听端口由PYTHON_BACKEND_PORT控制（默认 8000）；若开启OPIK_OTEL_SDK_ENABLED=true，则通过opentelemetry-instrument进行自动埋点（见 entrypoint.sh）。

执行流程深入：从 HTTP 到 JSON 结果

无论采用哪种策略，一次评估的完整链路都是：

1. evaluatorBlueprint 收到POST /v1/private/evaluators/python，校验code与data字段；
2. 调用get_executor().run_scoring(code, data, payload_type)；
3. 执行器从池中获取空闲容器/进程，传入代码与数据；
4. 用户代码运行并输出结果 JSON（打印到 stdout 的最后一行）；
5. 执行器解析结果并返回，evaluator提取scores数组返回给调用方。

Docker 执行器：预热的容器池

DockerExecutor 的核心机制是预热的容器池：

• 初始化时并行创建max_parallel个沙箱容器，容器以tail -f /dev/null常驻保持存活，并打上managed_by=<instance_id>标签以便统一管理；
• 容器创建参数包含mem_limit、cpu_shares、nano_cpus（可选）、network_disabled与security_opt=["no-new-privileges"]，共同构成沙箱的资源与安全边界（见 executor_docker.py）；
• 后台调度线程按POOL_CHECK_INTERVAL检查并补充池容量（ensure_pool_filled）；
• 每次执行通过container.exec_run运行scoring_runner.pyc并传入code、data_json、payload_type三个参数（见 executor_docker.py）；
• 执行结束后，旧容器被异步停掉并删除，同时立即创建一个新容器回填池中——保证每次执行都使用"干净"的容器状态，避免上一次运行的残留数据污染下一次评估。

Process 执行器：基于 Pipe 的进程池

ProcessExecutor 面向开发/受限环境，机制类似但基于多进程：

• 预热max_parallel个 worker 进程，每个进程通过multiprocessing.Pipe与父进程通信，子进程入口为 process_worker.py 的worker_process_main；
• 创建 worker 时会等待子进程发出READY信号（最长 10 秒）才入池，保证拿到的 worker 一定可用（见 executor_process.py）；
• 执行时通过connection.send({'code': ..., 'data': ..., 'payload_type': ...})发送任务，poll(timeout)等待结果，超时后异步终止该 worker 并返回 HTTP 504；
• 注册了 SIGINT/SIGTERM 信号处理器，实现优雅关停：先终止全部 worker 再sys.exit(0)（见 executor_process.py）。

结果解析契约

执行结果统一由parse_execution_result解析（见 executor.py），其约定值得每个评估代码作者注意：

• 退出码 0 时，取stdout 最后一行解析为 JSON 对象作为结果；
• 无输出 / 最后一行不是合法 JSON / 不是 JSON 对象，均按 400 处理（视为用户指标代码本身有误，而非服务故障）；
• 退出码非 0 时，尝试从最后一行 JSON 中提取error字段返回，否则返回通用错误文案。

因此，评估代码必须以print(json.dumps(result))的形式在最后输出一个 JSON 对象，其中应包含scores列表。

沙箱安全机制：网络与文件系统的双重隔离

沙箱的安全性由容器参数与测试用例共同保证。test_executor_docker.py 中有两组直接的验证用例：

• 网络访问被阻断：PYTHON_CODE_EXECUTOR_ALLOW_NETWORK默认false，测试代码尝试urllib.request.urlopen('http://example.com')，断言结果为失败且 reason 包含urlopen error（见 test_executor_docker.py）；
• 文件系统访问受限：容器中读取宿主机文件被拒绝（见 test_executor_docker.py）。

资源层面，mem_limit=256m与cpu_shares/nano_cpus限制了单容器资源占用，security_opt=["no-new-privileges"]禁止容器内提权；执行超时则兜底防止死循环拖垮服务。若确实需要联网（例如调用外部模型 API 的评估指标），需显式设置PYTHON_CODE_EXECUTOR_ALLOW_NETWORK=true。

隔离子进程执行器：更彻底的执行隔离方案

除 README 提到的两种策略外，仓库还提供了第三种执行器 IsolatedSubprocessExecutor，其完整设计文档见 docs/ISOLATED_EXECUTOR_COMPLETE.md。它的定位是解决ProcessExecutor复用 worker 池带来的环境变量泄漏问题：每次执行都创建全新子进程，环境变量按执行作用域完全隔离，不共享任何状态，适合多租户场景（每个租户携带不同的API_KEY/TENANT_ID）。

典型用法：

from opik_backend.executor_isolated import IsolatedSubprocessExecutor executor = IsolatedSubprocessExecutor(timeout_secs=30) # 环境变量仅对本次执行生效 result = executor.execute( file_path="/path/to/metric.py", data={}, env_vars={"TENANT_ID": "tenant_123", "API_KEY": "secret_key"}, )

其核心特性包括：环境变量作用域隔离、每次执行自动创建/清理子进程、teardown 回调注册、with上下文管理器自动释放、线程安全的并发执行、每个子进程 20MB 栈内存限制（RLIMIT_STACK，防止无限递归）、可选的 HTTP 日志流式收集（subprocess_logger.py 中的BatchLogCollector按时间 1 秒/大小 10MB 批量上报，支持 gzip 与鉴权头），以及基于 OpenTelemetry 的创建/执行延迟指标。相关配置项包括SUBPROCESS_LOG_ENABLED、OPIK_SUBPROCESS_LOG_BACKEND_URL、SUBPROCESS_LOG_FLUSH_INTERVAL、SUBPROCESS_LOG_MAX_SIZE、SUBPROCESS_LOG_REQUEST_TIMEOUT与SUBPROCESS_LOG_FAIL_ON_MISSING_BACKEND。

可观测性与错误语义

OpenTelemetry 指标

服务在OPIK_OTEL_SDK_ENABLED=true且配置了OTEL_EXPORTER_OTLP_ENDPOINT时启用 OTLP 导出（见init.py）。Docker 执行器暴露了丰富的指标，便于监控执行引擎的健康度（见 executor_docker.py）：

• container_creation_latency/container_stop_latency：容器创建/销毁耗时直方图；
• scoring_executor_latency：单次评估总耗时；
• container_pool_size/scoring_executor_queue_size：池容量与排队任务数（用于判断饱和度）；
• payload_code_size/payload_data_size：代码与数据负载大小直方图（自定义分桶覆盖 100B 至 100MB）；
• execution_outcome：执行结果计数器（success/timeout/invalid_code/saturated/error/serialization_error）；
• executor_container_cpu_cores/executor_container_memory_bytes：由指标采集线程按配置间隔通过 Docker stats API 计算并上报的单容器资源用量。

HTTP 错误语义

执行失败以结构化错误码返回，调用方应依据状态码而非错误文案分支处理：

错误常量定义于 executor.py，SATURATED_ERROR文案为 "Code executor is saturated, please retry"，提示调用方应带退避重试。

小结

Opik Python Backend 通过"策略化执行器 + 预热资源池 + 沙箱隔离 + 结构化结果契约"四层设计，为在线评测提供了安全、可控、可观测的代码执行能力。开发期使用flask --app src/opik_backend --debug run配合process策略即可快速迭代；生产环境则按 Dockerfile 构建镜像、以docker策略运行，并依据流量特征调优PYTHON_CODE_EXECUTOR_PARALLEL_NUM、EXEC_TIMEOUT_IN_SECS与POOL_ACQUIRE_TIMEOUT等参数。若要进一步了解执行器内部细节与隔离方案演进，可继续阅读 executor_docker.py、executor_process.py 与 docs/ISOLATED_EXECUTOR_COMPLETE.md。

【免费下载链接】comet-llmDebug, evaluate, and monitor your LLM applications, RAG systems, and agentic workflows with comprehensive tracing, automated evaluations, and production-ready dashboards.项目地址: https://gitcode.com/GitHub_Trending/co/comet-llm