Opik 隔离子进程执行器IsolatedSubprocessExecutor完全指南环境隔离、并发安全与日志流式收集【免费下载链接】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导读本文围绕 Opik 开源项目仓库根目录 apps/opik-python-backend中的IsolatedSubprocessExecutor隔离子进程执行器展开它是 Opik Optimization Studio 等场景下执行用户自定义 Python 代码如评估指标、评分函数的底层基础设施。读完本文你将掌握为什么在多租户、多配置并发执行场景需要每个执行一个全新子进程的隔离模型如何通过execute()完成 JSON 数据传递与环境变量作用域隔离如何借助BatchLogCollector/RedisBatchLogCollector实现零日志丢失的流式收集以及超时、资源限制、生命周期管理与故障排查的完整实战方案。一、Executive Summary它解决了什么问题问题背景Opik 后端此前基于 executor_process.py 中的ProcessExecutor它维护一个可复用的 worker 进程池多个执行共享同一环境。这在多租户系统中会带来一个致命问题——环境变量泄漏租户 A 的API_KEY、TENANT_ID等环境变量可能被并发执行的租户 B 读到造成数据串租户与安全隐患。解决方案IsolatedSubprocessExecutorexecutor_isolated.py为每次执行创建全新的子进程并对环境变量做完全作用域化隔离——无泄漏、无冲突天然适配多租户系统。核心特性一览特性说明环境变量隔离每次执行拥有作用域化、隔离的环境变量子进程生命周期管理自动创建与清理Teardown 回调注册清理函数在 teardown 时按注册顺序执行上下文管理器支持with语句自动释放资源线程安全并发可与 ThreadPoolExecutor / AsyncIO 安全配合资源限制每个子进程栈内存限制 20MBRLIMIT_STACK日志流式收集可选 HTTP 方式回传后端或 Redis 中转OpenTelemetry 指标创建与执行延迟、活跃进程数追踪完整错误处理所有错误路径均优雅处理超时支持防止失控执行零共享状态执行之间完全独立适用场景✅ 适合❌ 不适合多租户系统极致高吞吐100 次/秒每次执行配置不同实时流式10ms 延迟需要环境变量隔离资源受限环境安全敏感操作每次执行使用不同 API Key性能画像文档声明值指标数值吞吐量5-10 次/秒单次执行开销约 150ms子进程创建单子进程内存约 20MB 栈上限线程安全✅并发安全✅自动清理✅二、Quick Start60 秒集成1. 导入from opik_backend.executor_isolated import IsolatedSubprocessExecutor2. 创建实例executor IsolatedSubprocessExecutor(timeout_secs30)3. 编写待执行 Python 文件# metric.py import json from opik.evaluation.metrics import base_metric, score_result result { scores: [{ value: 0.95, name: my_metric, reason: Works! }] } print(json.dumps(result))4. 执行文件result executor.execute(file_path/path/to/metric.py, data{}) # Output: {scores: [{value: 0.95, name: my_metric, reason: Works!}]}5. 携带环境变量执行env_vars { TENANT_ID: tenant_123, API_KEY: secret_key, } result executor.execute( file_path/path/to/metric.py, data{}, env_varsenv_vars ) # 环境变量仅作用于此执行不会泄漏6. 上下文管理器自动清理with IsolatedSubprocessExecutor() as executor: result executor.execute(file_path/path/to/metric.py, data{}) # 退出 with 块时自动 teardown三、核心架构三个模块的分工1.executor_isolated.py— 主执行器类路径apps/opik-python-backend/src/opik_backend/executor_isolated.py职责为每次代码执行创建隔离子进程通过 stdin/stdout 以 JSON 传递数据每次执行作用域化环境变量强制 20MB 栈内存限制提供进程生命周期管理kill、teardown 回调与BatchLogCollector集成实现可选日志流式收集关键方法与源码签名一致# 执行代码注意当前源码签名的首个参数是 file_path result executor.execute( file_path..., # 待执行 Python 文件路径 data{...}, # 通过 stdin 传入的数据字典 env_vars{...}, # 作用域化的环境变量 timeout_secs30, # 可选覆盖默认超时 payload_typeNone, # 可选payload 类型如 trace_thread optimization_idopt-123,# 可选日志关联 ID job_idjob-456 # 可选日志关联 ID ) # 注册清理回调 executor.register_teardown_callback(cleanup_func) # 手动进程管理 executor.kill_process(pid) executor.kill_all_processes() executor.teardown() # 上下文管理器 with executor: result executor.execute(...)需要说明的是文档 FAQ 中提到code参数同时支持文件路径与内联代码字符串而当前仓库源码中execute()的实际签名以file_path为第一参数见 executor_isolated.py测试也全部基于临时.py文件路径执行因此以文件路径方式为准。2.subprocess_logger.py— 日志收集与流式传输路径apps/opik-python-backend/src/opik_backend/subprocess_logger.py关键类SubprocessLogRecord表示单条日志timestamp、level、logger_name、message、attributes并提供to_dict()序列化与size_bytes()大小估算BatchLogCollector继承logging.Handler在内存缓冲中收集日志并按时间/大小批量 HTTP 上报RedisBatchLogCollector将原始日志行写入 Redis LIST保留 Rich 富文本/ANSI 颜色供 Java 后端同步到 S3用于 Optimization Studio特性捕获子进程 stdout/stderr解析 JSON 格式日志回退为纯文本按时间默认 1 秒或大小默认 10MB批量支持 gzip 压缩的 HTTP POST注当前源码中 HTTP 上报为普通 JSON请求头含 Authorization 与 Comet-Workspace线程安全的后台 flush 线程优雅错误处理记录 warning不崩溃HTTP 收集器用法logger BatchLogCollector( backend_urlhttp://api.example.com/logs, optimization_idopt-123, job_idjob-456, api_keysecret-key, workspaceworkspace-id ) # 处理来自子进程的日志 logger.process_subprocess_output(stdout, stderr)注意process_subprocess_output是文档中的示意 API当前源码提供的真实接口为start_stream_from_process(process)在后台线程中并发读取 stdout/stderr逐行解析后 emit配合wait_for_reader_threads()与get_last_lines()取回最终输出。3.subprocess_log_config.py— 集中式配置路径apps/opik-python-backend/src/opik_backend/subprocess_log_config.py职责集中读取环境变量单一事实来源配置校验与默认值无副作用仅 getenv 调用方法get_backend_url()— 日志后端 HTTP 端点is_enabled()— 日志是否启用get_flush_interval_ms()— 基于时间的 flush 间隔get_max_size_bytes()— 基于大小的 flush 阈值get_request_timeout_secs()— HTTP 请求超时get_log_reader_timeout_secs()— 日志读取线程超时is_fully_configured()— 是否所有必需配置就绪backend_url 且 enabled四、架构图与代码执行流程整体架构┌─────────────────────────────────────────────────────────────┐ │ Parent Process (IsolatedSubprocessExecutor) │ │ │ │ ┌─────────────────────────────────────────────────────────┐│ │ │ execute(file_path, data, env_vars, ...) ││ │ └──────────────────┬──────────────────────────────────────┘│ │ │ │ │ ┌────────────┴────────────┐ │ │ │ │ │ │ ┌───▼──────┐ ┌───────▼────────┐ │ │ │ 准备 │ │ Prepare │ │ │ │ 环境 │ │ Environment │ │ │ └───┬──────┘ └────────────────┘ │ │ │ │ │ └──────────────┬──────────────────┐ │ │ │ │ │ │ ┌────────▼────────┐ ┌──────▼──────┐ │ │ │ json.dumps │ │ Popen │ │ │ │ 输入数据 │ │ python -u │ │ │ └────────┬────────┘ │ file │ │ │ │ └──────┬───────┘ │ │ ┌───▼──────────────────▼───┐ │ │ │ subprocess.Popen() │ │ │ │ python -u file │ │ │ └───┬──────────────────┬───┘ │ │ │ │ │ └───────────────────────┼──────────────────┼─────────────────┘ │ │ ┌───────────────▼────────────────▼─────────────────┐ │ Child Process (Subprocess) │ │ stdin: ◄── json data │ │ Read JSON input │ │ exec(user_code) │ │ print(json.dumps(result)) to stdout ──► stdout │ │ Logger output ──────────────────────► stderr │ └──────────────────────────────────────────────────┘ │ │ ┌───────────────▼────────────────▼─────────────────┐ │ Parent Process Continues │ │ communicate() 或 日志流式模式 获取 stdout/stderr │ │ 若开启日志: BatchLogCollector 解析/批量/上报 │ │ 从 stdout 最后一行 JSON 解析结果并返回 │ └──────────────────────────────────────────────────┘实现细节代码执行流程1. 输入准备# 用户代码示例 code from opik.evaluation.metrics import base_metric, score_result result {scores: [{value: 0.8, name: quality}]} print(json.dumps(result)) # 传给代码的数据 data {text: Hello world} # 作用域化环境变量 env_vars {CUSTOM_VAR: value, OPIK_API_KEY: key, OPIK_WORKSPACE: ws}2. 环境准备源码_prepare_environment源码以os.environ.copy()为基底叠加env_vars中的覆盖项值为None的变量会被跳过并告警因为subprocess.Popen要求环境值全部为字符串最后强制写入PYTHONUNBUFFERED1与LOG_FORMATjson保证输出不缓冲且日志以 JSON 格式输出。3. 子进程执行python -u file_path # stdin: {data: {text: Hello}, payload_type: null} # stdout: {scores: [{value: 0.8, name: quality}]} # stderr: 代码产生的任何日志子进程创建时通过preexec_fn_set_memory_limit注入 20MB 栈内存限制并用-uunbuffered与bufsize1保证行级流式输出。4. 结果解析源码_parse_last_json_line执行完成后父进程从 stdout 中取最后一条非空行解析为 JSON 作为结果若解析失败返回{code: 500, error: ...}。若子进程退出码非 0则优先尝试从 stdout 中解析错误 JSONOptimization 场景下 runner 会把错误以 JSON 打在 stdout 上否则返回{code: 500, error: stderr}。相关边界用例多行输出取末行、空白行忽略、非法 JSON、空输出可参见 test_executor_isolated.py。5. 日志收集可选# 若 logging 完全配置 if SubprocessLogConfig.is_fully_configured(): from opik_backend.subprocess_logger import BatchLogCollector # 为当前 PID 创建收集器并启动实时流式读取 self._log_collectors[process.pid] BatchLogCollector( backend_urlSubprocessLogConfig.get_backend_url(), optimization_idoptimization_id or , job_idjob_id or , api_keyenv_vars.get(OPIK_API_KEY, ), workspaceenv_vars.get(OPIK_WORKSPACE, ), ) log_collector.start_stream_from_process(process)资源限制内存限制类型仅栈内存RLIMIT_STACK限制大小每子进程 20MB源码常量SUBPROCESS_MEMORY_LIMIT_BYTES 20 * 1024 * 1024效果防止无限递归与栈溢出不影响堆分配、运行时数据结构理由与 ProcessExecutor 行为一致同时保证 Python 解释器与运行时堆正常工作五、并发执行每进程独立日志收集器架构每个子进程拥有独立日志收集器父进程用字典以 PID 为键管理# 内部结构 _log_collectors { 1234: BatchLogCollector(...), # 进程 1 的日志 1235: BatchLogCollector(...), # 进程 2 的日志 1236: BatchLogCollector(...), # 进程 3 的日志 }收益✅ 完全并发支持多进程可同时运行 ✅ 独立日志流每个进程独立流式上报 ✅ 零干扰关闭一个进程的日志不影响其他进程 ✅ 线程安全增删操作由_process_lock保护 ✅ 零日志丢失规范的关闭序列 signal → flush → cleanup并发执行示例import concurrent.futures executor IsolatedSubprocessExecutor() def execute_with_tenant(tenant_id): return executor.execute( file_path/path/to/metric.py, data{tenant_id: tenant_id}, env_vars{TENANT_ID: tenant_id}, optimization_idfopt_{tenant_id}, job_idfjob_{tenant_id}, ) # 并发执行 10 次 with concurrent.futures.ThreadPoolExecutor(max_workers4) as pool: futures [pool.submit(execute_with_tenant, ftenant_{i}) for i in range(10)] results [f.result() for f in concurrent.futures.as_completed(futures)]仓库中的并发验证测试test_executor_isolated.py用 3 个 worker 并发执行 3 个不同TENANT_ID的执行断言每个结果中的租户 ID 正确无串扰。线程安全保证操作线程安全保护机制execute()✅进程隔离kill_process()✅_process_lock_log_collectors 访问✅_process_lock日志流式读取✅每进程独立的 ThreadPoolExecutormax_workers2 并发读 stdout/stderr关闭✅signal → Executor.shutdown(waitTrue) → 最终 flush关闭序列Executor 同时运行 3 个进程: ├─ Process A (PID 1000) │ └─ _log_collectors[1000] → 流式日志 ├─ Process B (PID 1001) │ └─ _log_collectors[1001] → 流式日志 └─ Process C (PID 1002) └─ _log_collectors[1002] → 流式日志 调用 teardown() 时: ├─ 并行 terminate 所有进程kill_all_processes ├─ 等待全部退出 ├─ 对每个进程: │ ├─ 置 should_stop True │ ├─ 关闭 flush executor等待待处理任务 │ ├─ 最终 flush全部日志上报/写入 │ └─ 清理读取线程 └─ 所有日志被捕获零丢失保证 ✓六、配置参考环境变量由SubprocessLogConfig统一读取# 日志后端配置 SUBPROCESS_LOG_ENABLEDtrue/false # 启用日志默认: false OPIK_SUBPROCESS_LOG_BACKEND_URL... # 日志后端 HTTP 端点 SUBPROCESS_LOG_FLUSH_INTERVAL1000 # flush 间隔 ms默认: 1000 SUBPROCESS_LOG_MAX_SIZE10485760 # 缓冲大小上限 bytes默认: 10MB SUBPROCESS_LOG_REQUEST_TIMEOUT60 # HTTP 请求超时秒默认: 60 SUBPROCESS_LOG_READER_TIMEOUT5 # 日志读取线程超时秒源码补充默认: 5SubprocessLogConfig对每个读取都做了int()解析容错非法值自动回退默认见 subprocess_log_config.py。日志凭据通过 env_vars 参数传递这些凭据不是通过环境变量配置而是随execute()的env_vars参数传入用于构造 HTTP 请求头executor.execute( file_path..., # 源码签名file_path 而非 code datadata, env_vars{ OPIK_API_KEY: your-api-key, # 用于 Authorization 请求头 OPIK_WORKSPACE: workspace-id, # 用于 Comet-Workspace 请求头 } )错误处理模式优雅模式默认SUBPROCESS_LOG_FAIL_ON_MISSING_BACKENDfalse若 backend_url 未配置记录 warning跳过日志继续执行即使日志失败执行也成功严格模式SUBPROCESS_LOG_FAIL_ON_MISSING_BACKENDtrue若 backend_url 未配置抛出 ValueError执行以清晰错误信息失败源码现状说明当前 subprocess_log_config.py 中is_fully_configured()实现为backend_url 存在 且 is_enabled()HTTP 收集器BatchLogCollector.__init__在缺少backend_url时直接抛出ValueError(backend_url is required for BatchLogCollector)。上述两种模式是文档中定义的设计目标。七、使用模式Usage Patterns模式 1多租户评分executor IsolatedSubprocessExecutor() for tenant in tenants: result executor.execute( file_path, # 按当前源码签名使用 file_path data, env_vars{ TENANT_ID: tenant.id, OPIK_API_KEY: tenant.api_key, OPIK_WORKSPACE: tenant.workspace, }, optimization_idfopt_{tenant.id}, job_idfjob_{tenant.id}, ) process_result(result)模式 2并发执行import concurrent.futures executor IsolatedSubprocessExecutor() with concurrent.futures.ThreadPoolExecutor(max_workers4) as pool: futures [ pool.submit( executor.execute, file_path, # 源码签名首参为 file_path data, {TENANT_ID: ftenant_{i}}, ) for i in range(10) ] results [f.result() for f in concurrent.futures.as_completed(futures)]模式 3上下文管理器自动清理with IsolatedSubprocessExecutor(timeout_secs30) as executor: result executor.execute(file_path, data, env_vars) # 退出时自动 teardown模式 4开启日志import os # 配置日志 os.environ[SUBPROCESS_LOG_ENABLED] true os.environ[OPIK_SUBPROCESS_LOG_BACKEND_URL] http://api.example.com/logs os.environ[SUBPROCESS_LOG_FLUSH_INTERVAL] 500 # 500ms os.environ[SUBPROCESS_LOG_MAX_SIZE] str(5 * 1024 * 1024) # 5MB executor IsolatedSubprocessExecutor() result executor.execute( file_pathfile_path, datadata, env_vars{OPIK_API_KEY: key, OPIK_WORKSPACE: ws}, optimization_idopt-123, job_idjob-456, ) # 日志自动上报后端八、生命周期管理上下文管理器模式with IsolatedSubprocessExecutor() as executor: # 设置阶段 executor.register_teardown_callback(lambda: print(Cleanup 1)) executor.register_teardown_callback(lambda: print(Cleanup 2)) # 执行 result executor.execute(file_path, data) # 退出 with 块自动 teardown # teardown 回调按注册顺序执行源码为顺序遍历手动生命周期executor IsolatedSubprocessExecutor() # 注册 teardown 回调 def cleanup(): print(Cleaning up...) executor.register_teardown_callback(cleanup) # 执行 result executor.execute(file_path, data) # 手动 teardown executor.teardown() # 所有 teardown 回调被调用进程终止executor IsolatedSubprocessExecutor() # 终止指定进程先 terminate超时后 kill executor.kill_process(pid, timeout2) # 并行终止所有活跃进程 executor.kill_all_processes(timeout2)源码细节executor_isolated.pykill_process先在_process_lock下原子地查找并移除进程避免锁长时间持有再在锁外terminate()等待timeout秒后仍存活则kill()强杀随后关闭该进程的日志收集器以捕获最终日志kill_all_processes先快照 PID 列表再用ThreadPoolExecutor并行执行 killteardown()依次执行kill 全部进程 → 关闭残留日志收集器 → 执行全部 teardown 回调单个回调异常不影响其余回调见测试 test_teardown_callback_exception_handling。九、日志与监控日志条目格式{ timestamp: 1697539200000, level: INFO, logger_name: task, message: Task started, attributes: {step: 1} }支持的日志来源Python logging 模块— 输出到 stderr 的 JSON 格式日志print 到 stdout— 纯文本行print 到 stderr— 纯文本行JSON 到 stdout/stderr— 结构化日志读取线程_read_stream的解析规则见 subprocess_logger.py每行先尝试json.loads若同时含level与logger_name字段则作为结构化日志原样 emit否则包装为subprocess.streamlogger 的 INFO 记录JSON 解析失败则按纯文本消息处理。日志批量策略基于时间每 1 秒 flush可配置后台_periodic_flush线程基于大小缓冲达 10MB 即 flush可配置基于事件关闭/teardown 时 flushHTTP 请求POST /logs HTTP/1.1 Content-Type: application/json Authorization: api_key Comet-Workspace: workspace { optimization_id: opt-123, job_id: job-456, logs: [ {timestamp: ..., level: INFO, message: ...}, ... ] }BatchLogCollector使用requests.post上报带timeout默认 60s失败仅记录 warning 不崩溃buffer 保留以待下次 flush。该链路被集成测试 test_subprocess_logging.py 用本地 mock HTTP server 完整验证包括 print→stderr、print→stdout、logging 模块、JSON 日志四种来源的捕获以及Authorization/Comet-Workspace请求头断言见 test_subprocess_logging.py。Optimization Studio 的 Redis 日志通道在真实业务中jobs/optimizer.py优化任务先创建RedisBatchLogCollector通过create_optimization_log_collector(workspace_id, optimization_id)工厂函数再以 PID 0 为占位键注册到 executor 的_log_collectors字典执行开始时_execute_in_subprocess弹出该预注册收集器并挂到真实子进程 PID 下。RedisBatchLogCollector将原始文本行保留 Rich 富文本与 ANSI 颜色以 Redis pipeline 原子地rpush到opik:logs:{workspace_id}:{optimization_id}LIST同时更新...:metaHASH 的last_append_ts并刷新 TTL默认 86400 秒可由OPTSTUDIO_LOG_REDIS_TTL覆盖供 Java 后端周期同步至 S3。OpenTelemetry 指标源码在模块加载时创建executor_isolated.py- isolated_subprocess_creation_latency # 子进程创建耗时 (ms) —— histogram - isolated_subprocess_execution_latency # 代码执行总耗时 (ms) —— histogram - isolated_subprocess_active_count # 当前活跃子进程数 —— up_down_counter十、生产环境检查清单✅ 无可变默认参数陷阱源码execute()中data默认值在函数内重新赋值为{}env_vars为 None 时重置为空 dict✅ 无静默失败显式错误处理✅ 全链路适当错误日志✅ 锁保护的线程安全操作✅ 每进程日志收集器PID → BatchLogCollector 字典映射✅ 完全并发执行支持仓库测试以 3 并行进程验证✅ 零日志丢失保证signal → flush → cleanup 序列✅ 配置错误优雅降级✅ 资源限制生效20MB 栈✅ 全面测试覆盖executor 17 项 日志集成 4 项仓库内单元/集成测试全绿✅ 清晰的配置接口单一来源 SubprocessLogConfig✅ ThreadPoolExecutor 后台线程清理✅ 内存高效的日志批量✅ 自动进程清理✅ 超时处理✅ 基于 JSON 的进程间通信IPC✅ OpenTelemetry 集成✅ 上下文管理器 自动 teardown十一、故障排查与 FAQ常见问题问题Subprocess logging enabled but backend_url not configured原因SUBPROCESS_LOG_ENABLEDtrue但未设置OPIK_SUBPROCESS_LOG_BACKEND_URL解决# 要么禁用日志 export SUBPROCESS_LOG_ENABLEDfalse # 要么设置后端 URL export OPIK_SUBPROCESS_LOG_BACKEND_URLhttp://api.example.com/logs问题requests library not available for log posting原因未安装requests库解决pip install requests仓库的 requirements.txt 已声明该依赖作为后端服务部署时通常已具备。问题子进程超时原因代码执行时长超过超时值解决# 调大超时 executor IsolatedSubprocessExecutor(timeout_secs60)超时路径细节源码_execute_in_subprocess捕获subprocess.TimeoutExpired后先process.kill()并最多等待 2 秒仍未退出则再 kill 一次后向上抛出外层execute()捕获后返回{code: 500, error: Execution timed out after N seconds}见 [executor_isolated.py](https://link.gitcode.com/i/abaae1a77769a5442f249df0f7b57b97#L355-L361, L173-L180)并由测试 test_execute_timeout 验证。FAQQ: 执行之间可以共享状态吗A: 不可以。每次执行完全隔离这是设计使然。Q: 子进程中的环境变量会发生什么A: 它们仅作用于该次执行父进程不受影响。源码基于os.environ.copy()构造子进程环境叠加项只对该子进程生效。Q: 可以修改被执行的代码吗A: 可以。当前源码execute()以file_path接收待执行文件FAQ 中提到的内联代码字符串形式是文档描述的目标能力若需内联执行可自行将代码写入临时.py文件仓库测试即用tempfile.NamedTemporaryFile实现。Q: 它是线程安全的吗A: 是的完全线程安全。多个线程可并发调用execute()并发测试已覆盖test_concurrent_execution。Q: 内存限制是多少A: 每个子进程 20MB 栈内存防止无限递归。使用resource.setrlimit(resource.RLIMIT_STACK, ...)实现仅限制栈不影响堆。Q: 子进程可以访问文件吗A: 可以。子进程可访问文件系统OS 级资源共享这也是为什么多租户场景必须依赖环境变量隔离而非文件系统隔离。十二、文件参考文件用途路径executor_isolated.py主执行器类apps/opik-python-backend/src/opik_backend/executor_isolated.pysubprocess_logger.py日志收集与流式传输apps/opik-python-backend/src/opik_backend/subprocess_logger.pysubprocess_log_config.py配置管理apps/opik-python-backend/src/opik_backend/subprocess_log_config.pytest_executor_isolated.py执行器单元测试17 项apps/opik-python-backend/tests/unit/test_executor_isolated.pytest_subprocess_logging.py日志集成测试4 项apps/opik-python-backend/tests/unit/test_subprocess_logging.pyjobs/optimizer.pyOptimization Studio 真实调用示例apps/opik-python-backend/src/opik_backend/jobs/optimizer.py总结IsolatedSubprocessExecutor是 Opik 后端在必须安全执行不可信/多租户用户代码场景下的关键基础设施它以每执行一个新子进程 环境变量作用域 JSON IPC 20MB 栈限制 可插拔日志收集器的组合换取了 ProcessExecutor 池化复用模式所不具备的强隔离性。对于多租户评分、按执行注入不同 API Key、Optimization Studio 优化任务等场景它是开箱即用的首选方案而文档声明的吞吐量5-10 次/秒与单次约 150ms 的子进程创建开销也决定了它更适合对吞吐不极端的业务路径。若你的业务符合配置随执行变化、需要严格环境隔离的特征即可直接按本文的 Quick Start 与使用模式接入。文档状态Production Ready | 版本 2.1 | 核心特性每进程日志收集器并发执行零日志丢失保证【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考