MLOps机器学习后端工作流自动化AI Agent【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址https://gitcode.com/GitHub_Trending/ze/zenml点击查看免费下载在 ZenML 的 API 参考文档中步骤日志端点文档 以 OpenAPI 引用块的形式声明了GET /api/v1/steps/{step_id}/logs这一用于按步Step拉取日志的 REST 接口。本文以该端点为骨架结合 ZenML 开源仓库的服务端路由、日志存储抽象与数据模型实现完整解析该接口的路径构成、请求参数、响应结构、鉴权规则、错误场景以及日志从 Pipeline/Step 运行中产生、被采集、落盘到最终可被 API 检索的完整链路并给出通过 Python Client 与 ZenML Dashboard 获取步骤日志的实操方案。端点概览从 OpenAPI 占位符到真实实现logs.md文档页本身是 ZenML OpenAPI 文档的自动渲染页其核心信息是声明了以下接口GET /api/v1/steps/{step_id}/logs该端点并非仅存在于文档声明中其真实实现位于服务端路由模块 steps_endpoints.py 的get_step_logs函数。从路由定义来看路径由以下常量拼接而成见 constants.pyAPI /apiconstants.pyVERSION_1 /v1constants.pySTEPS /stepsconstants.pyLOGS /logsconstants.py即GET /api/v1/steps/{step_id}/logs挂载在steps路由上用于返回某个 Step Run 关联的日志条目列表。它的典型使用场景是在 Pipeline Run 执行完成后按 Step 维度精确拉取其 stdout/stderr、Pythonlogging输出与print()语句用于调试、审计或接入外部系统。请求参数与调用规则从 get_step_logs 的函数签名可知该接口除路径参数外还接受两个互斥的查询参数参数类型必填说明step_id路径参数UUID是目标 Step Run 的 ID用于在 Zen Store 中定位对应的步骤运行source查询参数str与logs_id二选一日志来源标识例如client、orchestrator、step等用于在步骤的日志集合中按来源精确匹配logs_id查询参数UUID与source二选一日志记录LogsResponse的 ID用于直接定位某条日志记录调用规则互斥校验对应源码中的两段校验逻辑若source与logs_id均为空抛出ValueError响应400 Bad Request提示 Either source or logs_id must be provided.若两者同时提供同样抛出ValueError响应400 Bad Request提示 Only one of source or logs_id must be provided.。也就是说source与logs_id是“必须且只能提供一个”的关系。两者本质上对应日志集合中的两种索引方式source面向业务语义这个日志是哪一层产生的而logs_id面向精确 ID直接指向某条LogsResponse实体。服务端分别通过 search_logs_by_source 与 search_logs_by_id 在步骤的日志集合中进行匹配查找。响应结构List[LogEntry]接口返回的是List[LogEntry]。LogEntry是 ZenML 定义的结构化日志条目模型见 logging_utils.py既被 Artifact Log Store 用作 JSON 序列化格式也是服务端查询时返回给客户端的数据结构其字段如下字段类型说明messagestr日志消息正文nameOptional[str]产生该条日志的 logger 名称levelOptional[LoggingLevels]日志级别DEBUG / INFO / WARN / ERROR / CRITICALtimestampOptional[datetime]日志产生时间moduleOptional[str]产生该条日志的模块名filenameOptional[str]产生该条日志的源文件名linenoOptional[int]源文件中的行号chunk_indexint默认 0分块日志的块索引total_chunksint默认 1分块日志的块总数idUUID日志条目唯一标识用于重组分块消息分块机制对于超长日志消息约超过 5KBZenML 会自动将消息拆分为多个块各块共享同一个id并通过递增的chunk_index与固定的total_chunks标记顺序供客户端重组完整消息。这一约定同时体现在 Artifact Log Store 的 NDJSON 落盘格式中详见后文。此外单次请求返回的日志条目数量受上限约束服务端调用fetch_logs(..., limitLOGS_MAX_ENTRIES_PER_REQUEST)拉取日志LOGS_MAX_ENTRIES_PER_REQUEST默认值为50000见 constants.py可通过环境变量ZENML_LOGS_MAX_ENTRIES_PER_REQUEST调整。鉴权与错误场景该端点通过Security(authorize)强制鉴权并遵循 ZenML 的 RBAC 权限模型。其关键逻辑见 steps_endpoints.py先通过 Zen Store 获取stepget_run_step(step_id, hydrateTrue)拿到其所属的pipeline_run_id与project_id随后以ResourceType.PIPELINE_RUN资源、Action.READ动作进行权限校验——读取步骤日志的前提是调用者对日志所属的 Pipeline Run 拥有读权限。路由声明中注册的错误响应包括400、401、403、404、422结合实现可归纳如下状态码触发场景400source与logs_id均未提供或同时提供互斥校验失败401未携带有效认证凭据403已认证但对所属 Pipeline Run 无读权限404步骤不存在或步骤无日志集合No logs found for step {step_id}.或按source未匹配到日志或按logs_id未匹配到日志422参数校验失败如step_id非合法 UUID值得注意的细节当step.log_collection为空时直接抛出KeyError对应 404当按source或logs_id在集合中找不到匹配项时同样抛出KeyError错误信息中会包含具体的source值或logs_id值便于定位问题。数据模型LogsRequest / LogsResponse / LogsUpdate/api/v1/steps/{step_id}/logs端点背后承载的是 ZenML 的 Logs 实体。围绕该实体数据模型定义在 models/v2/core/logs.py分为三类LogsRequest创建请求关键字段id: UUID日志记录 ID默认uuid4()自动生成uri: Optional[str]日志文件 URI用于 Artifact Log Store 场景且受TEXT_FIELD_MAX_LENGTH长度约束source: str日志来源标识artifact_store_id/log_store_id分别指代日志由哪个 Artifact Store 或 Log Store 收集pipeline_run_id/step_run_id日志关联的 Pipeline Run 或 Step Run二者互斥不可同时设置hook_invocation_id日志关联的 Hook 调用。模型级校验model_validator保证artifact_store_id与log_store_id至少设置其一除非source为特殊的runner来源LOGS_RUNNER_SOURCEstep_run_id与pipeline_run_id不能同时存在。LogsResponse查询响应分为 body 与 metadata 两层——bodyuri、sourcemetadatastep_run_id、pipeline_run_id、artifact_store_id、log_store_id、hook_invocation_id。LogsUpdate更新请求支持重新关联pipeline_run_id或step_run_id同样受“二者不可同时设置”约束。这些模型通过 Step Run / Pipeline Run 的log_collection资源字段与运行实体关联见 step_run.py 中StepRunResponseResources.log_collection的定义。换言之一个 Step Run 可以挂载多条日志记录例如client、orchestrator、step等不同来源各有其一/api/v1/steps/{step_id}/logs的职责正是从这组集合中按source或logs_id挑选并返回。底层实现get_step_logs 的完整调用链从服务端视角一次GET /api/v1/steps/{step_id}/logs请求的处理流程见 steps_endpoints.py为校验source与logs_id的互斥关系从 Zen Store 获取 Step RunhydrateTrue并校验对所属 Pipeline Run 的读权限若step.log_collection存在提供source时调用search_logs_by_source匹配命中后调用fetch_logs拉取日志条目提供logs_id时调用search_logs_by_id匹配命中后同样调用fetch_logs未命中则抛KeyError若step.log_collection为空直接抛KeyError。其中fetch_logs是关键的跨层函数见 logging_utils.py它只能在服务端环境ZENML_SERVER环境变量已设置中调用避免在客户端实例化重复的 Log Store。其内部逻辑为若logs.log_store_id存在通过 Zen Store 获取对应的 Stack 组件模型校验其类型确为LOG_STORE再用StackComponent.from_model实例化具体的 Log Store调用其fetch()方法最后执行cleanup()释放资源若集成依赖缺失则抛NotImplementedError若logs.artifact_store_id存在实例化对应的 Artifact Store并构建ArtifactLogStore同样调用fetch()两者皆无时返回空列表[]。也就是说该 API 端点本身并不直接读取日志文件而是将“按模型定位后端并读取日志”的职责委托给对应 Log Store 的fetch抽象。这一设计使得服务端无需知道日志究竟存在 Artifact Store、OpenTelemetry 还是 Datadog实现了后端无关的日志检索。日志从产生到可查询LoggingContext 与 Log Store 机制要理解/api/v1/steps/{step_id}/logs为何能取到日志需要了解日志在运行时是如何被捕获与关联的。ZenML 的日志采集系统参见 Log Stores 组件指南包含三步stdout/stderr 包装ZenML 包装标准输出与错误流捕获所有print()及写入这两个流的消息根 logger handler为 Python 根 logger 挂载自定义 Handler捕获所有向上传播的logging消息含第三方库输出日志路由所有被捕获的消息经LoggingContext路由到当前 Stack 的活跃 Log Store。核心上下文管理器是 LoggingContext它通过ContextVar维护当前活跃的日志上下文并提供了完整的生命周期管理begin()在 Log Store 上注册一个origin即日志来源emit(record)将 LogRecord 交给 Log Store 的emit()处理并附带元数据如zenml.step.run.id、zenml.pipeline.run.id、zenml.project.id等end()注销origin若这是最后一个活跃 origin将触发 Log Store 的flush()退出上下文时若发生异常会额外以 ERROR 级别注入一条 An exception has occurred. 记录元数据标记为zenml.event.typeexception。Origin 的生命周期定义在 base_log_store.py流程为实例化 Log Store →register_origin()传入 log model 与元数据→emit()逐条发送日志 →deregister_origin()最后一个 origin 注销时 flush。setup_logging_contextlogging_utils.py负责组装这一切查找已存在的 LogsResponse按source匹配否则通过generate_logs_request生成LogsRequest并调用Client().zen_store.create_logs创建新的日志记录再把 Step Run / Pipeline Run 的 ID 通过LogsUpdate关联回日志实体——这正是步骤的log_collection集合被填充、最终可供 API 检索的底层机制。日志存储格式与位置默认情况下若 Stack 中未显式配置 Log StoreZenML 会自动使用Artifact Log Store零配置开箱即用详见 Artifact Log Store 指南将日志写入 Artifact Store 的logs目录。存储位置规则见 artifact_log_store.py 的prepare_logs_uri可变文件系统本地、S3、Azure 等日志以单文件形式追加写入路径为artifact_store_path/logs/log_id.log不可变文件系统如 GCS日志写入按时间戳命名的目录artifact_store_path/logs/log_id/其中包含多个*.log分片文件并在最终化时合并为timestamp_merged.log。落盘格式为换行分隔的 JSONNDJSON。每行对应一条日志字段包括message、level、timestamp、name、filename、lineno、module、chunk_index、total_chunks、id与LogEntry模型一一对应。读取时fetch_log_records 与_stream_logs_line_by_line会按行流式读取不把整个文件载入内存并通过 parse_log_entry 兼容两种格式以{开头且以}结尾的 JSON 行走model_validate_json完整解析普通文本行则降级为INFO级别、仅填充message字段的LogEntry同时兼容旧版[YYYY-MM-DD HH:MM:SS UTC]时间戳前缀格式并自动剥离 ANSI 转义序列remove_ansi_escape_codes。除默认的 Artifact Log Store 外ZenML 还提供otel通用 OpenTelemetry 后端不支持日志回拉Dashboard 会显示 Logs not available与datadog完整支持导出与回拉等 flavor可用zenml log-store flavor list查看全部可用 flavor。无论底层是哪种 Log Store/api/v1/steps/{step_id}/logs都会经由 Log Store 抽象的fetch()统一返回List[LogEntry]。相关 API/api/v1/logs 端点家族/api/v1/steps/{step_id}/logs是日志查询的“便捷入口”而完整的日志实体 CRUD 路由则定义在 logs_endpoints.py挂载前缀为API VERSION_1 LOGS即/api/v1/logs方法与路径功能说明POST /api/v1/logscreate_logs创建日志记录LogsRequest。创建前会校验对关联 Pipeline Run / Step Run 的UPDATE权限以及对 Artifact Store / Log Store 组件的READ权限GET /api/v1/logs/{logs_id}get_logs按 ID 获取单条日志记录。日志必须已关联 Pipeline Run 或 Step Run否则抛IllegalOperationError通过hydrate参数控制是否返回 metadataPUT /api/v1/logs/{logs_id}update_logs更新日志关联的pipeline_run_id/step_run_id同样需要所属运行的UPDATE权限其中update_logs的持久化逻辑可在 sql_zen_store.py 中看到若日志已关联了不同的实体再次修改关联会抛IllegalOperationErrorThe logs entry is already associated with a different entity.从而保证一条日志记录的归属一旦确定即不可转移。这也解释了为什么setup_logging_context是先创建、后attach()关联运行 ID 的异步两段式设计。通过 Python Client 与 Dashboard 获取步骤日志方式一Python Client 编程式获取虽然 REST 端点要求source与logs_id互斥传参但日常使用中更推荐通过 Python Client 直接读取见 Log Stores 组件指南 中的代码示例from zenml.client import Client client Client() # 获取 Pipeline Run或按步骤 ID 获取步骤 run client.get_pipeline_run(RUN_NAME_OR_ID) step client.get_run_step(step_run_idSTEP_RUN_ID) # 注意使用的 Log Store 必须与采集日志时一致 log_store client.active_stack.log_store # 拉取运行级日志 log_entries log_store.fetch(logs_modelrun.logs, limit1000) for entry in log_entries: print(f[{entry.level}] {entry.message})Client.get_run_step(step_run_id, hydrateTrue)会返回带log_collection资源的StepRunResponse见 client.py其中每一条LogsResponse对应一个日志来源。需要精确按来源过滤时可直接在集合中匹配source字段。方式二ZenML Dashboard在 ZenML Dashboard 中进入 Pipeline Run 页面即可按 Step 查看其日志Dashboard 后端调用的正是本文所述的/api/v1/steps/{step_id}/logs系列接口。对于datadog等外部 Log Store日志也会在对应平台的原生界面中展示。环境变量与调优围绕日志采集、批处理与检索ZenML 暴露了若干可调环境变量集中定义于 constants.py环境变量默认值作用ZENML_LOGS_MAX_ENTRIES_PER_REQUEST50000单次 API / fetch 请求返回的日志条目上限直接影响/api/v1/steps/{step_id}/logs的响应规模ZENML_LOGS_OTEL_MAX_QUEUE_SIZE100000OpenTelemetry 批量日志处理器的最大队列长度ZENML_LOGS_OTEL_SCHEDULE_DELAY_MILLIS5000批量导出之间的调度延迟毫秒ZENML_LOGS_OTEL_MAX_EXPORT_BATCH_SIZE5000单次导出的最大批大小ZENML_LOGS_OTEL_EXPORT_TIMEOUT_MILLIS15000单批导出的超时时间毫秒ZENML_LOGS_STORAGE_MAX_QUEUE_SIZE100000日志存储队列最大容量ZENML_LOGS_STORAGE_QUEUE_TIMEOUT-1日志存储队列超时-1 表示无限等待ZENML_LOGS_WRITE_INTERVAL_SECONDS1日志写入间隔秒ZENML_LOGS_MERGE_INTERVAL_SECONDS600不可变文件系统日志合并间隔秒ZENML_DISABLE_PIPELINE_LOGS_STORAGEFalse置为true可全局禁用 Pipeline 级日志存储ZENML_DISABLE_STEP_LOGS_STORAGEFalse置为true可全局禁用 Step 级日志存储ZENML_DISABLE_STEP_NAMES_IN_LOGSTrue控制在控制台日志中是否打印 Step 名称上述默认值针对大多数场景已足够仅在高并发、大日志量的流水线场景下建议适当调大批处理与队列相关参数并在 Pipeline / Step 配置层面通过enable_step_logs、enable_pipeline_logs显式控制日志存储的开关。禁用后/api/v1/steps/{step_id}/logs将返回 404 或空结果因此在排查API 取不到日志的问题时应优先检查这些开关以及实际使用的 Log Store flavor 是否支持日志回拉。总结GET /api/v1/steps/{step_id}/logs是 ZenML 面向 Step 粒度的日志检索接口它通过source或logs_id两个互斥查询参数在步骤的log_collection中精确挑选日志经 RBAC 校验后委托给底层 Log Store 的fetch()抽象返回结构化的LogEntry列表。从运行时的 stdout/stderr 包装、LoggingContext路由、Origin 注册与 flush到 Artifact Log Store 的 NDJSON 落盘与分块重组再到服务端fetch_logs的跨层装配整个链路环环相扣。掌握该端点及其背后的数据模型、日志存储机制与调优变量即可在自建运维、日志审计与集成第三方监控平台时精确、高效地获取每一步骤的执行日志。赞分享MLOps机器学习后端工作流自动化AI Agent【免费下载链接】zenmlZenML : One AI Platform from Pipelines to Agents. https://zenml.io.项目地址https://gitcode.com/GitHub_Trending/ze/zenml点击查看免费下载相关推荐AI Agent 浏览器怎么挑ego-lite 让你和 Agent 共用一个登录态的完整指南AI Agent 浏览器怎么挑ego lite 让你和 Agent 共用一个登录态的完整指南 ego lite 是一款为 AI Agent 打造的 ChromAI 技能浏览器控制GUI 自动化ZenML API 令牌端点深度解析Generic 与 Workload 两类 API Token 的生成规则、参数与源码实现ZenML API 令牌端点深度解析Generic 与 Workload 两类 API Token 的生成规则、参数与源码实现 本篇围绕 ZenML RESTMLOps机器学习后端工作流自动化AI AgentZenML 健康检查 API 详解/health 端点的实现原理与运维实践ZenML 健康检查 API 详解/health 端点的实现原理与运维实践 本篇文章围绕 ZenML ZenServer 提供的 GET /health 健康MLOps机器学习后端工作流自动化AI Agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
