尧图网站建设 
基于 MLflow 的 Haystack 自动追踪Autologging Tracing集成实战指南【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow导读本文围绕 MLflow 官方 API 文档中 mlflow.haystack 模块展开深入讲解如何将 Haystack 构建的 RAG 流水线、Agent 与 LLM 应用接入 MLflow 的追踪Tracing体系自动采集每次 Pipeline 运行的完整调用链、组件输入输出、Token 用量与成本信息。读完本文你将掌握mlflow.haystack.autolog()的完整用法与参数语义理解其基于 OpenTelemetry 的底层实现原理并能够查看、分析和导出 Haystack 应用的 Trace 数据。模块定位与 API 总览mlflow.haystack是 MLflow 为 Haystack 框架提供的官方集成模块其 API 文档页由automodule指令从模块源码自动生成真实内容由两部分组成公开 APIautolog()、setup_haystack_tracing()、teardown_haystack_tracing()定义于 mlflow/haystack/init.py 与 mlflow/haystack/autolog.py实现细节包括自动注册到 MLflow 总入口的 LazyLoader见 mlflow/init.py 与mlflow.autolog()的集成声明以及常量定义Span 类型、Token 用量键等位于 mlflow/entities/span.py 与 mlflow/tracing/constant.py。模块将 Flavor 名称固定为haystackFLAVOR_NAME haystack这是所有 MLflow 集成共用的命名约定用于标记和去重自动化日志行为。快速开始一行代码开启 Haystack 追踪使用方式极为简单只需在构建 Haystack Pipeline 之前调用import mlflow mlflow.haystack.autolog() # 之后正常构建并运行 Haystack Pipeline 即可 pipe Pipeline() pipe.add_component(retriever, retriever) pipe.add_component(llm, llm) result pipe.run({...})参数说明autolog()的三个参数定义于 mlflow/haystack/init.py参数类型默认值作用log_tracesboolTrue是否采集 Haystack 的 Trace。设为False时与disableTrue行为等价直接拆除追踪并返回disableboolFalse是否禁用该集成。True时调用teardown_haystack_tracing()移除已注册的 Span Processor后续运行不再产生 TracesilentboolFalse是否静默 MLflow 在自动日志过程中的事件与警告输出其逻辑为当disableTrue或log_tracesFalse时调用teardown_haystack_tracing()否则调用setup_haystack_tracing()。同时模块通过autolog.integration_name haystack标记集成名称并由autologging_integration(FLAVOR_NAME)装饰器包裹_autolog函数使其能被 mlflow/init.py 中的统一mlflow.autolog()入口识别——即你也可以直接mlflow.autolog()MLflow 会自动找到并启用 Haystack 集成。完整可运行示例仓库中提供了完整的 Agentic RAG 示例OpenAI 生成器 BM25 检索 条件路由 SerperDev 网页搜索见 examples/haystack/tracing.py。核心流程mlflow.set_experiment(Haystack Tracing) mlflow.haystack.autolog() # 构建包含 retriever / prompt_builder / llm / router / websearch 的流水线 agentic_rag_pipe Pipeline() agentic_rag_pipe.add_component(retriever, retriever) ... result agentic_rag_pipe.run({retriever: {query: query}, ...}) # 获取并检查最近一次 Trace last_trace_id mlflow.get_last_active_trace_id() trace mlflow.get_trace(trace_idlast_trace_id) # 汇总 Token 用量 total_usage trace.info.token_usage print(fInput tokens: {total_usage[input_tokens]}) print(fOutput tokens: {total_usage[output_tokens]}) print(fTotal tokens: {total_usage[total_tokens]}) # 逐 Span 查看每个 LLM 调用的用量 for span in trace.data.spans: if usage : span.get_attribute(mlflow.chat.tokenUsage): print(f{span.name}: input{usage[input_tokens]} output{usage[output_tokens]} total{usage[total_tokens]})自动捕获的追踪内容开启 autolog 后每次Pipeline.run()或AsyncPipeline.run()都会生成一条 Trace其结构由 mlflow/haystack/autolog.py 中的HaystackSpanProcessor构建包含两类 SpanPipeline 根 SpanCHAIN 类型对应 Haystack 的haystack.pipeline.run/haystack.async_pipeline.run被标记为SpanType.CHAIN流水线级调用链携带Pipeline 名称取自haystack.pipeline.name属性Pipeline 输入输出取自haystack.pipeline.input/haystack.pipeline.output属性并以 JSON 解析后写入 span 的 inputs/outputs聚合子组件 IO若 Pipeline 自身未提供输入输出则把各子组件 span 的输入输出按组件别名聚合填充。组件 Span对应haystack.component.runSpan 类型由 mlflow/haystack/autolog.py 中的_infer_span_type_from_haystack()依据组件类型/别名启发式推断匹配关键词大小写不敏感推断的 Span 类型llm、chat、generator、completion、textgen、chatgenerator、openai、anthropic、mistral、cohere、geminiLLMembedderEMBEDDINGretrieverRETRIEVERrankerRERANKERagentAGENT其余默认TOOL组件 Span 还会将 span 名称改为组件类型或别名如InMemoryBM25Retriever、Add同时同步更新底层 OTel span 与 MLflow span 的_original_name避免在去重流程中被覆盖见 autolog.py解析haystack.component.input/haystack.component.output属性并写入 inputs/outputs含 JSON 容错解析从 inputs 中提取模型名写入mlflow.llm.model属性set_span_model_attribute定义于 mlflow/tracing/utils/init.py从 outputs 中解析 Token 用量见下节将子组件的输入输出按别名聚合到父 Pipeline span实现层级化信息汇总。上述行为均有对应测试验证见 tests/haystack/test_haystack_tracing.py例如test_haystack_autolog_single_trace断言根 span 为 CHAIN 且名称为haystack.pipeline.run、组件 span 为 TOOL 且名称为组件类名test_in_memory_retriever_component_traced验证检索器被标记为RETRIEVERtest_multiple_components_in_pipeline_reranker验证RETRIEVER与RERANKER的混合流水线。Token 用量与成本追踪Token 用量解析_parse_token_usage()autolog.py从组件 outputs 中提取 usage 信息支持的两种输出结构replies[0][meta][usage]LLM 生成类组件meta[0][usage]整体元数据。提取prompt_tokens、completion_tokens、total_tokens后映射为 MLflow 标准 Token 用量键见 mlflow/tracing/constant.py{ input_tokens: 1, output_tokens: 2, total_tokens: 3 }写入 span 属性mlflow.chat.tokenUsageSpanAttributeKey.CHAT_USAGE。测试 tests/haystack/test_haystack_tracing.py 验证了 LLM 组件 span 的该属性以及model_name gpt-4的模型名提取。成本计算当should_compute_cost_client_side()返回 True 时即客户端侧成本估算开启集成会在设置 Token 用量后调用set_span_cost_attribute()根据 Token 用量与模型单价计算mlflow.llm.cost属性键定义见 mlflow/tracing/constant.py。测试示例中按“输入 token 1.0、输出 token 2.0”的模拟单价验证了input_cost1.0、output_cost4.0、total_cost5.0的计算结果。底层实现原理基于 OpenTelemetry 的集成依赖与版本兼容_get_opentelemetry_tracer_class()autolog.py按优先级选择 OTel Tracer 类Haystack 3优先从haystack_integrations.tracing.opentelemetry导入OpenTelemetryTracer若缺失抛出MlflowException并提示pip install opentelemetry-haystackHaystack 3回退到核心包中的haystack.tracing.OpenTelemetryTracer。对应测试 tests/haystack/test_haystack_tracing.py 模拟两者皆不可用验证报错信息包含安装指引。启动流程setup_haystack_tracingsetup_haystack_tracing()的执行步骤autolog.py开启 Haystack 的内容追踪开关hs_tracing.tracer.is_content_tracing_enabled True确保组件输入输出等详细内容被记录获取全局 Tracer Provider。若当前为NoOpTracerProvider或ProxyTracerProvider尚未显式配置则创建SDKTracerProvider并注册HaystackSpanProcessor后设为全局 Provider若已存在 SDK Provider则检查是否已注册HaystackSpanProcessor避免重复注册未注册则追加通过enable_tracing(OpenTelemetryTracer(tracer))将 MLflow 的 tracer 接入 Haystack使 Haystack 产生的 span 流经 MLflow 的 span processor。Span 生命周期处理HaystackSpanProcessor继承 OTel 的SimpleSpanProcessor重写on_start与on_endon_start调用 MLflow 内部tracer.span_processor.on_start()传递 span生成 MLflow Trace IDgenerate_trace_id_v3通过create_mlflow_span()工厂函数见 mlflow/entities/span.py创建 MLflow span 并注册到InMemoryTraceManageron_end通过get_mlflow_span_for_otel_span()关联回 MLflow span按 span 名称分支处理 Pipeline 信息或组件信息最后调用内部tracer.span_processor.on_end()完成落盘。递归保护机制一个值得注意的实现细节当环境变量MLFLOW_USE_DEFAULT_TRACER_PROVIDERfalse共享 Tracer Provider 场景时on_start/on_end会经由复合 processor 再次路由回自身若不处理将导致无限递归。为此实现使用了线程局部变量threading.local()作为重入标志见 autolog.py。对应测试test_haystack_autolog_shared_provider_no_recursiontests/haystack/test_haystack_tracing.py验证了共享 Provider 下不会抛出 RecursionError。关闭追踪teardown_haystack_tracingteardown_haystack_tracing()从当前 SDK Tracer Provider 的活跃 span processor 列表中移除所有HaystackSpanProcessor实例autolog.py此后 Haystack 运行不再产生 MLflow Trace。查看与分析 Trace开启 autolog 后可通过 MLflow 的标准 Tracing API 消费数据mlflow.get_last_active_trace_id()/mlflow.get_trace(trace_id...)获取最近一次运行或指定 Tracetrace.info.token_usage获取整条 Trace 的汇总 Token 用量trace.data.spans遍历所有 Span读取span.name、span.span_type、span.inputs、span.outputs、span.get_attribute(mlflow.chat.tokenUsage)、span.llm_cost、span.model_name等字段。此外MLflow 提供了 MLflow Tracing 的 UImlflow ui用于可视化 Trace 瀑布图可以直观查看 Pipeline 各组件的调用顺序、耗时与输入输出。小结使用mlflow.haystack.autolog()或统一入口mlflow.autolog()可零侵入地为 Haystack Pipeline 启用自动追踪每条 Pipeline 运行生成一条 Trace根 Span 为 CHAIN 类型各组件按启发式规则分类为 LLM / EMBEDDING / RETRIEVER / RERANKER / AGENT / TOOL集成自动采集组件输入输出、模型名、Token 用量与成本支撑应用的可观测性与成本控制底层基于 OpenTelemetry 标准实现兼容 Haystack 2.x 与 3.x3.x 需安装opentelemetry-haystack并针对共享 Tracer Provider 场景做了递归防护可通过 examples/haystack/tracing.py 复现完整 Agentic RAG 追踪示例通过 tests/haystack/test_haystack_tracing.py 深入理解集成行为边界。【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 基于 MLflow 的 Haystack 自动追踪（Autologging Tracing）集成实战指南
基于 MLflow 的 Haystack 自动追踪Autologging Tracing集成实战指南【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow导读本文围绕 MLflow 官方 API 文档中 mlflow.haystack 模块展开深入讲解如何将 Haystack 构建的 RAG 流水线、Agent 与 LLM 应用接入 MLflow 的追踪Tracing体系自动采集每次 Pipeline 运行的完整调用链、组件输入输出、Token 用量与成本信息。读完本文你将掌握mlflow.haystack.autolog()的完整用法与参数语义理解其基于 OpenTelemetry 的底层实现原理并能够查看、分析和导出 Haystack 应用的 Trace 数据。模块定位与 API 总览mlflow.haystack是 MLflow 为 Haystack 框架提供的官方集成模块其 API 文档页由automodule指令从模块源码自动生成真实内容由两部分组成公开 APIautolog()、setup_haystack_tracing()、teardown_haystack_tracing()定义于 mlflow/haystack/init.py 与 mlflow/haystack/autolog.py实现细节包括自动注册到 MLflow 总入口的 LazyLoader见 mlflow/init.py 与mlflow.autolog()的集成声明以及常量定义Span 类型、Token 用量键等位于 mlflow/entities/span.py 与 mlflow/tracing/constant.py。模块将 Flavor 名称固定为haystackFLAVOR_NAME haystack这是所有 MLflow 集成共用的命名约定用于标记和去重自动化日志行为。快速开始一行代码开启 Haystack 追踪使用方式极为简单只需在构建 Haystack Pipeline 之前调用import mlflow mlflow.haystack.autolog() # 之后正常构建并运行 Haystack Pipeline 即可 pipe Pipeline() pipe.add_component(retriever, retriever) pipe.add_component(llm, llm) result pipe.run({...})参数说明autolog()的三个参数定义于 mlflow/haystack/init.py参数类型默认值作用log_tracesboolTrue是否采集 Haystack 的 Trace。设为False时与disableTrue行为等价直接拆除追踪并返回disableboolFalse是否禁用该集成。True时调用teardown_haystack_tracing()移除已注册的 Span Processor后续运行不再产生 TracesilentboolFalse是否静默 MLflow 在自动日志过程中的事件与警告输出其逻辑为当disableTrue或log_tracesFalse时调用teardown_haystack_tracing()否则调用setup_haystack_tracing()。同时模块通过autolog.integration_name haystack标记集成名称并由autologging_integration(FLAVOR_NAME)装饰器包裹_autolog函数使其能被 mlflow/init.py 中的统一mlflow.autolog()入口识别——即你也可以直接mlflow.autolog()MLflow 会自动找到并启用 Haystack 集成。完整可运行示例仓库中提供了完整的 Agentic RAG 示例OpenAI 生成器 BM25 检索 条件路由 SerperDev 网页搜索见 examples/haystack/tracing.py。核心流程mlflow.set_experiment(Haystack Tracing) mlflow.haystack.autolog() # 构建包含 retriever / prompt_builder / llm / router / websearch 的流水线 agentic_rag_pipe Pipeline() agentic_rag_pipe.add_component(retriever, retriever) ... result agentic_rag_pipe.run({retriever: {query: query}, ...}) # 获取并检查最近一次 Trace last_trace_id mlflow.get_last_active_trace_id() trace mlflow.get_trace(trace_idlast_trace_id) # 汇总 Token 用量 total_usage trace.info.token_usage print(fInput tokens: {total_usage[input_tokens]}) print(fOutput tokens: {total_usage[output_tokens]}) print(fTotal tokens: {total_usage[total_tokens]}) # 逐 Span 查看每个 LLM 调用的用量 for span in trace.data.spans: if usage : span.get_attribute(mlflow.chat.tokenUsage): print(f{span.name}: input{usage[input_tokens]} output{usage[output_tokens]} total{usage[total_tokens]})自动捕获的追踪内容开启 autolog 后每次Pipeline.run()或AsyncPipeline.run()都会生成一条 Trace其结构由 mlflow/haystack/autolog.py 中的HaystackSpanProcessor构建包含两类 SpanPipeline 根 SpanCHAIN 类型对应 Haystack 的haystack.pipeline.run/haystack.async_pipeline.run被标记为SpanType.CHAIN流水线级调用链携带Pipeline 名称取自haystack.pipeline.name属性Pipeline 输入输出取自haystack.pipeline.input/haystack.pipeline.output属性并以 JSON 解析后写入 span 的 inputs/outputs聚合子组件 IO若 Pipeline 自身未提供输入输出则把各子组件 span 的输入输出按组件别名聚合填充。组件 Span对应haystack.component.runSpan 类型由 mlflow/haystack/autolog.py 中的_infer_span_type_from_haystack()依据组件类型/别名启发式推断匹配关键词大小写不敏感推断的 Span 类型llm、chat、generator、completion、textgen、chatgenerator、openai、anthropic、mistral、cohere、geminiLLMembedderEMBEDDINGretrieverRETRIEVERrankerRERANKERagentAGENT其余默认TOOL组件 Span 还会将 span 名称改为组件类型或别名如InMemoryBM25Retriever、Add同时同步更新底层 OTel span 与 MLflow span 的_original_name避免在去重流程中被覆盖见 autolog.py解析haystack.component.input/haystack.component.output属性并写入 inputs/outputs含 JSON 容错解析从 inputs 中提取模型名写入mlflow.llm.model属性set_span_model_attribute定义于 mlflow/tracing/utils/init.py从 outputs 中解析 Token 用量见下节将子组件的输入输出按别名聚合到父 Pipeline span实现层级化信息汇总。上述行为均有对应测试验证见 tests/haystack/test_haystack_tracing.py例如test_haystack_autolog_single_trace断言根 span 为 CHAIN 且名称为haystack.pipeline.run、组件 span 为 TOOL 且名称为组件类名test_in_memory_retriever_component_traced验证检索器被标记为RETRIEVERtest_multiple_components_in_pipeline_reranker验证RETRIEVER与RERANKER的混合流水线。Token 用量与成本追踪Token 用量解析_parse_token_usage()autolog.py从组件 outputs 中提取 usage 信息支持的两种输出结构replies[0][meta][usage]LLM 生成类组件meta[0][usage]整体元数据。提取prompt_tokens、completion_tokens、total_tokens后映射为 MLflow 标准 Token 用量键见 mlflow/tracing/constant.py{ input_tokens: 1, output_tokens: 2, total_tokens: 3 }写入 span 属性mlflow.chat.tokenUsageSpanAttributeKey.CHAT_USAGE。测试 tests/haystack/test_haystack_tracing.py 验证了 LLM 组件 span 的该属性以及model_name gpt-4的模型名提取。成本计算当should_compute_cost_client_side()返回 True 时即客户端侧成本估算开启集成会在设置 Token 用量后调用set_span_cost_attribute()根据 Token 用量与模型单价计算mlflow.llm.cost属性键定义见 mlflow/tracing/constant.py。测试示例中按“输入 token 1.0、输出 token 2.0”的模拟单价验证了input_cost1.0、output_cost4.0、total_cost5.0的计算结果。底层实现原理基于 OpenTelemetry 的集成依赖与版本兼容_get_opentelemetry_tracer_class()autolog.py按优先级选择 OTel Tracer 类Haystack 3优先从haystack_integrations.tracing.opentelemetry导入OpenTelemetryTracer若缺失抛出MlflowException并提示pip install opentelemetry-haystackHaystack 3回退到核心包中的haystack.tracing.OpenTelemetryTracer。对应测试 tests/haystack/test_haystack_tracing.py 模拟两者皆不可用验证报错信息包含安装指引。启动流程setup_haystack_tracingsetup_haystack_tracing()的执行步骤autolog.py开启 Haystack 的内容追踪开关hs_tracing.tracer.is_content_tracing_enabled True确保组件输入输出等详细内容被记录获取全局 Tracer Provider。若当前为NoOpTracerProvider或ProxyTracerProvider尚未显式配置则创建SDKTracerProvider并注册HaystackSpanProcessor后设为全局 Provider若已存在 SDK Provider则检查是否已注册HaystackSpanProcessor避免重复注册未注册则追加通过enable_tracing(OpenTelemetryTracer(tracer))将 MLflow 的 tracer 接入 Haystack使 Haystack 产生的 span 流经 MLflow 的 span processor。Span 生命周期处理HaystackSpanProcessor继承 OTel 的SimpleSpanProcessor重写on_start与on_endon_start调用 MLflow 内部tracer.span_processor.on_start()传递 span生成 MLflow Trace IDgenerate_trace_id_v3通过create_mlflow_span()工厂函数见 mlflow/entities/span.py创建 MLflow span 并注册到InMemoryTraceManageron_end通过get_mlflow_span_for_otel_span()关联回 MLflow span按 span 名称分支处理 Pipeline 信息或组件信息最后调用内部tracer.span_processor.on_end()完成落盘。递归保护机制一个值得注意的实现细节当环境变量MLFLOW_USE_DEFAULT_TRACER_PROVIDERfalse共享 Tracer Provider 场景时on_start/on_end会经由复合 processor 再次路由回自身若不处理将导致无限递归。为此实现使用了线程局部变量threading.local()作为重入标志见 autolog.py。对应测试test_haystack_autolog_shared_provider_no_recursiontests/haystack/test_haystack_tracing.py验证了共享 Provider 下不会抛出 RecursionError。关闭追踪teardown_haystack_tracingteardown_haystack_tracing()从当前 SDK Tracer Provider 的活跃 span processor 列表中移除所有HaystackSpanProcessor实例autolog.py此后 Haystack 运行不再产生 MLflow Trace。查看与分析 Trace开启 autolog 后可通过 MLflow 的标准 Tracing API 消费数据mlflow.get_last_active_trace_id()/mlflow.get_trace(trace_id...)获取最近一次运行或指定 Tracetrace.info.token_usage获取整条 Trace 的汇总 Token 用量trace.data.spans遍历所有 Span读取span.name、span.span_type、span.inputs、span.outputs、span.get_attribute(mlflow.chat.tokenUsage)、span.llm_cost、span.model_name等字段。此外MLflow 提供了 MLflow Tracing 的 UImlflow ui用于可视化 Trace 瀑布图可以直观查看 Pipeline 各组件的调用顺序、耗时与输入输出。小结使用mlflow.haystack.autolog()或统一入口mlflow.autolog()可零侵入地为 Haystack Pipeline 启用自动追踪每条 Pipeline 运行生成一条 Trace根 Span 为 CHAIN 类型各组件按启发式规则分类为 LLM / EMBEDDING / RETRIEVER / RERANKER / AGENT / TOOL集成自动采集组件输入输出、模型名、Token 用量与成本支撑应用的可观测性与成本控制底层基于 OpenTelemetry 标准实现兼容 Haystack 2.x 与 3.x3.x 需安装opentelemetry-haystack并针对共享 Tracer Provider 场景做了递归防护可通过 examples/haystack/tracing.py 复现完整 Agentic RAG 追踪示例通过 tests/haystack/test_haystack_tracing.py 深入理解集成行为边界。【免费下载链接】mlflowThe open source AI engineering platform for agents, LLMs, and ML models. MLflow enables teams of all sizes to debug, evaluate, monitor, and optimize production-quality AI applications while controlling costs and managing access to models and data.项目地址: https://gitcode.com/GitHub_Trending/ml/mlflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考