Cách tôi xây dựng AI cho dịch vụ phân tích

Trong bài viết này, tôi muốn chia sẻ kinh nghiệm phát triển dịch vụ AI cho nền tảng phân tích trang web, được gọi là Swetrix.

Mục đích của tôi là phát triển mô hình học máy có thể dự đoán lưu lượng truy cập trang web trong tương lai dựa trên dữ liệu hiển thị trên ảnh chụp màn hình sau

Mục tiêu cuối cùng là mang đến cho khách hàng một tầm nhìn rõ ràng về lưu lượng truy cập sẽ xuất hiện trên trang web của họ trong tương lai, từ đó cho phép họ hiểu rõ hơn và nâng cao kế hoạch kinh doanh nói chung.

2. Yêu cầu và kiến trúc

Trong quá trình lập kế hoạch, đã có quyết định tiếp tục Kiến trúc microservice với nhà môi giới tin nhắn RabbitMQ để liên lạc giữa các dịch vụ AI và API.

Trước hết, chúng ta cần thu thập dữ liệu bằng cron task hàng giờ vào một cơ sở dữ liệu riêng. Chúng tôi quyết định chọn ClickHouse vì dữ liệu gốc từ các trang web trên Swetrix được lưu trữ trong đó. Thông tin chi tiết về định dạng sẽ được đề cập trong các phần tiếp theo.

RabbitMQ được chọn làm nhà môi giới tin nhắn do tính đơn giản của nó và chúng tôi cần thiết lập liên lạc giữa các dịch vụ AI và API. Hãy phân tích mọi thứ và kiểm tra logic chính

Dịch vụ Swetrix-API:

• Thu thập số liệu thống kê dữ liệu hàng giờ thông qua Cron Task và gửi dữ liệu thô đến dịch vụ AI.
• Chèn và nhận dữ liệu được xử lý trước từ ClickHouse.

Dịch vụ Swetrix-AI:

• Xử lý dữ liệu thô và các tùy chọn đã chọn (khoảng thời gian và danh mục phụ) để dự báo.
• Chuyển đổi dữ liệu dự báo thành định dạng JSON và gửi dữ liệu đó trở lại dịch vụ API thông qua RabbitMQ.

Dịch vụ Swetrix-AI sẽ sử dụng khung NestJs cho phần phụ trợ và các tập lệnh Python để xử lý trước dữ liệu và dự đoán mô hình.

3. Tiền xử lý

Chúng tôi thu thập dữ liệu sau về các dự án vào bảng analytics . Bạn đã xem phiên bản hiển thị của dữ liệu này trong phần đầu tiên của bài viết.

Tôi đã có thể đạt được kết quả này (gần như chấp nhận được) bằng truy vấn sau:

 @Cron(CronExpression.EVERY_HOUR) async insertHourlyProjectData(): Promise<void> { const gatherProjectsData = ` INSERT INTO analytics.hourly_projects_data (UniqueID, projectID, statisticsGathered, br_keys, br_vals, os_keys, os_vals, lc_keys, lc_vals, ref_keys, ref_vals, so_keys, so_vals, me_keys, me_vals, ca_keys, ca_vals, cc_keys, cc_vals, dv_keys, dv_vals, rg_keys, rg_vals, ct_keys, ct_vals) SELECT generateUUIDv4() as UniqueID, pid as projectID, toStartOfHour(now()) as statisticsGathered, groupArray(br) as br_keys, groupArray(br_count) as br_vals, groupArray(os) as os_keys, groupArray(os_count) as os_vals, ... groupArray(ct) as ct_keys, groupArray(ct_count) as ct_vals FROM ( SELECT pid, br, count(*) as br_count, os, count(*) as os_count, ... ct, count(*) as ct_count FROM analytics.analytics GROUP BY pid, br, os, lc, ref, so, me, ca, cc, dv, rg, ct ) GROUP BY pid; ` try { await clickhouse.query(gatherProjectsData).toPromise() } catch (e) { console.error( `[CRON WORKER] Error whilst gathering hourly data for all projects: ${e}`, )

Chức năng này được lên lịch để chạy mỗi giờ bằng cách sử dụng Cron Job. Nó thu thập và chèn dữ liệu phân tích vào một clickhouse analytics.hourly_projects_data .

đầu ra

Do những hạn chế của ClickHouse, tôi không thể đạt được định dạng dữ liệu mong muốn. Vì vậy, tôi quyết định sử dụng pandas để hoàn thành quá trình tiền xử lý cần thiết cho quá trình đào tạo mô hình.

Cụ thể tôi đã sử dụng Python để làm như sau:

3.1 Kết hợp khóa và giá trị

Kết hợp các khóa và giá trị liên quan đến một danh mục vào một trường JSON, chẳng hạn như kết hợp các khóa và giá trị của thiết bị vào một đối tượng.

 os_keys = {“Windows”, ”MacOS”, ”MacOS”, ”MacOS”, ”Linux”} os_values = {1, 2, 2, 1, 5}

Vào trong:

 os = {“Windows”: 1, “MacOS”: 5, “Linux”: 5}

Đính kèm mã và đầu ra:

 def format_data(keys_list, vals_list, threshold): """ Format data by converting string representations of lists to actual lists, then sums up the counts for each key. Keys with counts below a specified threshold are aggregated into 'Other'. """ counts = defaultdict(int) for keys_str, vals_str in zip(keys_list, vals_list): keys = ast.literal_eval(keys_str) vals = ast.literal_eval(vals_str) for key, val in zip(keys, vals): counts[key] += val final_data = defaultdict(int) for value, count in counts.items(): final_data[value] = count return dict(final_data) def process_group(group): """ Combine specific groups by a group clause, and make a """ result = {} for col in group.columns: if col.endswith('_keys'): prefix = col.split('_')[0] # Extract prefix to identify the category (eg, 'br' for browsers) threshold = other_thresholds.get(prefix, 1) # Get the threshold for this category, default to 1 vals_col = col.replace('_keys', '_vals') keys_list = group[col].tolist() vals_list = group[vals_col].tolist() result[col.replace('_keys', '')] = format_data(keys_list, vals_list, threshold) return pd.Series(result)

Tôi có thể nói rằng định dạng dữ liệu này sẽ không được sử dụng cho chính dự đoán mà nó được sử dụng nhiều hơn để lưu trữ nó trong cơ sở dữ liệu và nhằm mục đích gỡ lỗi để xác minh rằng không có giá trị nào bị thiếu và hơn nữa, để kiểm tra kỹ xem mô hình có tạo ra kết quả chính xác hay không. kết quả.

đầu ra
3.2 Kết hợp khóa và giá trị

Để đào tạo một mô hình phù hợp, tôi quyết định xác định các nhóm khác cho nhiều danh mục khác nhau. Điều đó có nghĩa là nếu trên toàn cầu, số lượng phiên bản của một nhóm trong một danh mục cụ thể thấp hơn một tỷ lệ phần trăm (%) nhất định thì nó sẽ được thêm vào như một phần của nhóm kia.

Chẳng hạn, trong danh mục os , chúng ta có:

 {“MacOS”: 300, “Windows”: 400, “Linux”: 23 and “TempleOS”: 10}

Vì cả Linux và TempleOS trong trường hợp này đều cực kỳ hiếm nên chúng sẽ được kết hợp vào nhóm khác , do đó kết quả cuối cùng sẽ là:

 {“MacOS”: 300, “Windows”: 400, “other”: 33}.

Và “độ hiếm” được xác định khác nhau tùy thuộc vào danh mục và dựa trên ngưỡng được chỉ định cho danh mục này.

Nó có thể được cấu hình dựa trên sở thích và dữ liệu mong muốn của khách hàng

 other_thresholds = { 'br': 0.06, 'os': 0.04, 'cc': 0.02, 'lc': 0.02, 'ref': 0.02, 'so': 0.03, 'me': 0.03, 'ca': 0.03, 'cc': 0.02, 'dv': 0.02, 'rg': 0.01, 'ct': 0.01 }

Có 2 chức năng được thực hiện để đạt được điều này

 def get_groups_by_treshholds(df,column_name): """Calculate total values for all columns""" if column_name in EXCLUDED_COLUMNS: return counter = count_dict_values(df[column_name]) total = sum(counter.values()) list1 = [] for key, value in counter.items(): if not (value / total) < other_thresholds[column_name]: list1.append(key) return list1 def create_group_columns(df): column_values = [] for key in other_thresholds.keys(): groups = get_groups_by_treshholds(df, key) if not groups: continue for group in groups: column_values.append(f"{key}_{group}") column_values.append(f"{key}_other") return column_values column_values = create_group_columns(df) column_values

đầu ra

 ['br_Chrome', 'br_Firefox', 'os_Mac OS', 'os_other', 'cc_UA', 'cc_GB', 'cc_other', 'dv_mobile', 'dv_desktop', 'dv_other']

Khi làm việc với các mô hình học máy, điều quan trọng là dữ liệu đầu vào phải ở định dạng mà mô hình có thể hiểu được. Các mô hình học máy thường yêu cầu các giá trị số (số nguyên, số float) thay vì các cấu trúc dữ liệu phức tạp như JSON.

Do đó, một lần nữa, tốt hơn là nên xử lý trước dữ liệu của chúng tôi thêm một chút để phù hợp với yêu cầu này.

Tôi đã tạo một hàm create_exploded_df trong đó mỗi đối tượng được biểu diễn dưới dạng một cột riêng biệt và các hàng chứa các giá trị số tương ứng. (Nó chưa lý tưởng lắm, nhưng đó là giải pháp tốt nhất tôi có thể tạo ra)

 def create_exploded_df(df): """ Function which creates a new data set, iterates through the old one and fill in values according to their belongings (br_other, etc..) """ new_df = df[['projectID', 'statisticsGathered']] for group in column_values: new_df[group] = 0 new_df_cols = new_df.columns df_cols = df.columns for column in df_cols: if column in ['projectID', 'statisticsGathered']: continue for index, row in enumerate(df[column]): if column in EXCLUDED_COLUMNS: continue for key, value in row.items(): total = 0 if (a:=f"{column}_{key}") in new_df_cols: new_df[a][index] = value else: total += value new_df[f"{column}_other"][index] = total return new_df new_df = create_exploded_df(df) new_df.to_csv("2-weeks-exploded.csv") new_df

đầu ra

3.3 Điền giờ

Một vấn đề khác với định dạng dữ liệu mà chúng tôi gặp phải là nếu không có lưu lượng truy cập cho một dự án trong một giờ cụ thể thay vì tạo một hàng trống thì sẽ không có hàng nào cả, điều này thật bất tiện khi xét đến thực tế là mô hình được thiết kế để dự đoán dữ liệu cho khung thời gian sắp tới (ví dụ: giờ tiếp theo). Tuy nhiên, việc huấn luyện mô hình để đưa ra dự đoán là không khả thi nếu không có sẵn dữ liệu cho khung thời gian ban đầu.

Vì vậy, tôi đã viết một tập lệnh có thể tìm số giờ bị thiếu và chèn các hàng trống khi một giờ bị bỏ qua

3.4 Thêm và dịch chuyển cột mục tiêu

Về đào tạo mô hình, cách tiếp cận chính là sử dụng dữ liệu từ giờ trước làm mục tiêu cho mô hình. Điều này cho phép mô hình dự đoán lưu lượng truy cập trong tương lai dựa trên dữ liệu hiện tại.

 def sort_df_and_assign_targets(df): df = df.copy() df = df.sort_values(by=['projectID', 'statisticsGathered']) for column_name in df.columns: if not column_name.endswith('target'): continue df[column_name] = df.groupby('projectID')[column_name].shift(-1) return df new_df = sort_df_and_assign_targets(new_df)

đầu ra

3.5 Phân chia statisticsGathered thống kêTập hợp thành các cột riêng biệt

Lý do chính cho cách tiếp cận như vậy là statisticsGathered là một đối tượng datetime , những mô hình mà tôi đã cố gắng sử dụng (kiểm tra các phần tiếp theo) không thể xử lý nó và xác định đúng mẫu.

Điều đó dẫn đến số liệu MSE/MRSE khủng khiếp. Vì vậy, trong quá trình phát triển, người ta đã quyết định tách các tính năng theo day , month và hour để nâng cao kết quả một cách đáng kể.

 def split_statistic_gathered(df): df['Month'] = df['statisticsGathered'].dt.month.astype(int) # as int df['Day'] = df['statisticsGathered'].dt.day.astype(int) # as int df['Hour'] = df['statisticsGathered'].dt.hour df = df.drop('statisticsGathered', axis = 1) return df new_df = split_statistic_gathered(new_df) new_df

đầu ra

Và thế là xong! Hãy chuyển sang phần đào tạo chính nó! 🎉🎉🎉

4. Hồi quy tuyến tính

Chà, tôi đoán, dự đoán thực tế là phần thử thách nhất trong quá trình xây dựng ứng dụng này.

Điều đầu tiên tôi muốn thử là sử dụng mô hình LinearRegression :

Tôi đã thực hiện các chức năng sau:

 def create_model_for_target(train_df, target_series):    X_train, x_test, Y_train, y_test = train_test_split(train_df, target_series, test_size=0.3, shuffle=False)    reg = LinearRegression()    reg.fit(X_train, Y_train)    y_pred = reg.predict(x_test)    return {"y_test": y_test, "y_pred": y_pred} def create_models_for_targets(df):    models_data = dict()    df = df.dropna()    train_df = clear_df(df)    for target_name in df[[column_name for column_name in df.columns if column_name.endswith("target")]]:        models_data[target_name] = create_model_for_target(train_df, df[target_name])    return models_data

Giải trình

Đối với mỗi cột mục tiêu, chúng tôi chia dữ liệu thành tập huấn luyện và tập kiểm tra. Sau đó, chúng tôi huấn luyện mô hình LinearRegression trên dữ liệu huấn luyện và đưa ra dự đoán về dữ liệu thử nghiệm.

Để đánh giá kết quả đó có chính xác không, tôi đã thêm hàm thu thập các số liệu cần thiết và tạo ra kết quả đầu ra

 def evaluate_models(data):    evaluation = []    for target, results in data.items():        y_test, y_pred = results['y_test'], results['y_pred']        mse = mean_squared_error(y_test, y_pred)        rmse = mean_squared_error(y_test, y_pred) ** 0.5        mae = mean_absolute_error(y_test, y_pred)        mean_y = y_test.mean()        median_y = y_test.median()        evaluation.append({'target': target, 'mse': mse, 'rmse': rmse, 'mae': mae, 'mean_y': mean_y, 'median_y': median_y})    return pd.DataFrame(evaluation)

đầu ra

Tôi đã viết một tập lệnh tạo đầu ra và lưu nó vào tệp Excel, các giá trị kế toán mse , rmse , mae và mean_y

Như bạn có thể thấy, các số liệu không đạt yêu cầu và dữ liệu lưu lượng truy cập được dự đoán sẽ không chính xác và không phù hợp với mục tiêu dự báo lưu lượng truy cập của tôi.

Vì vậy, tôi đã quyết định dự đoán tổng số khách truy cập mỗi giờ để tạo ra các chức năng sau

 def add_target_column(df, by):  totals_series = df.apply(lambda x: sum(x[[column for column in df.columns if column.startswith(by)]]), axis=1)  df['total'] = totals_series  df[f'total_{by}_target'] = totals_series  return df def shift_target_column(df, by):  df = df.sort_values(by=['projectID', 'statisticsGathered'], ignore_index=True)  df['total_target'] = df.groupby('projectID')[f'total_{by}_target'].shift(-1)  return df new_df = add_target_column(new_df, 'br') new_df = shift_target_column(new_df, 'br') new_df[['total_br_target']]

đầu ra

Hàm này lấy một danh mục cụ thể và tính toán tổng số khách truy cập dựa trên danh mục đó. Điều này hoạt động vì tổng số giá trị Thiết bị sẽ bằng tổng số giá trị Hệ điều hành.

Với cách tiếp cận như vậy, mô hình đã cho kết quả tốt hơn gấp 10 lần so với trước đây .

5. Kết luận

Nếu chúng ta đang nói về trường hợp này, nó gần như có thể chấp nhận được và sẵn sàng sử dụng tính năng này. Giờ đây, khách hàng có thể lập kế hoạch phân bổ ngân sách và mở rộng quy mô máy chủ tùy thuộc vào kết quả của những dự đoán này

Dự đoán sai lệch so với giá trị thực tế khoảng 2,45 khách truy cập (vì RMSE = √MSE ) . Điều này không thể có bất kỳ tác động tiêu cực quan trọng nào đối với nhu cầu tiếp thị.

Vì bài viết này đã phát triển khá rộng rãi và ứng dụng vẫn đang được phát triển nên chúng tôi sẽ tạm dừng ở đây. Chúng tôi sẽ tiếp tục cải tiến phương pháp này trong tương lai và tôi sẽ cập nhật cho bạn!

Cảm ơn đã đọc và chú ý của bạn! Tôi mong muốn được nghe phản hồi và suy nghĩ của bạn trong phần bình luận. Tôi hy vọng thông tin này hữu ích cho mục tiêu của bạn!

Và chúc may mắn!