Repository: lm-sys/FastChat
File: fastchat/serve/monitor/clean_chat_data.py (Python)

Code before:
            "ct_network_error": 1,
        }
    user_id = hashlib.md5(row["ip"].encode()).hexdigest()

    # Prepare the result data
    result = dict(
        conversation_id=conversation_id,
        model=model,
        conversation=conversation,
        turn=len(conversation) // 2,
        language=lang_code,
        user_id=user_id,
        tstamp=row["tstamp"],
    )

    return {
        "result": result,
        "model": model,
    }


def clean_chat_data(log_files, action_type):
    with mp.Pool() as pool:
        # Use partial to pass action_type to get_action_type_data
        func = partial(get_action_type_data, action_type=action_type)
        file_data = pool.map(func, log_files, chunksize=1)
    # filter out Nones as some files may not contain any data belong to action_type
    raw_data = []
    for data in file_data:
        raw_data.extend(data)
    raw_data = [r for r in raw_data if r is not None]

    # Use the multiprocessing Pool
    with mp.Pool() as pool:
        func = partial(process_data, action_type=action_type)
        results = pool.map(func, raw_data, chunksize=1)

    # Aggregate results from child processes
    ct_invalid_conv_id = sum(
        [data["ct_invalid_conv_id"] for data in results if "ct_invalid_conv_id" in data]
    )
    ct_invalid = sum([data["ct_invalid"] for data in results if "ct_invalid" in data])
    ct_network_error = sum(
        [data["ct_network_error"] for data in results if "ct_network_error" in data]
    )
    all_models = set([data["model"] for data in results if "model" in data])
    chats = [data["result"] for data in results if "result" in data]

    chats.sort(key=lambda x: x["tstamp"])
    last_updated_tstamp = chats[-1]["tstamp"]
    last_updated_datetime = datetime.fromtimestamp(

Review comment (suggestion):
chunksize shouldn't be 1. otherwise we will introduce unnecessary overhead. we probably should do something like `chunksize = len(log_files)/len(num_parallel)` and user can specify `num_parallel` in argument.

Code after:
    if NETWORK_ERROR_MSG in messages:
        return {
            "ct_network_error": 1,
        }
    user_id = hashlib.md5(row["ip"].encode()).hexdigest()

    # Prepare the result data
    result = dict(
        conversation_id=conversation_id,
        model=model,
        conversation=conversation,
        turn=len(conversation) // 2,
        language=lang_code,
        user_id=user_id,
        tstamp=row["tstamp"],
    )

    return {
        "result": result,
        "model": model,
    }


def clean_chat_data(log_files, action_type, num_parallel):
    with mp.Pool(num_parallel) as pool:
        # Use partial to pass action_type to get_action_type_data
        func = partial(get_action_type_data, action_type=action_type)
        file_data = list(
            tqdm(
                pool.imap(
                    func, log_files, chunksize=ceil(len(log_files) / len(pool._pool))
                ),
                total=len(log_files),
                desc="Processing Log Files",
            )
        )
    # filter out Nones as some files may not contain any data belong to action_type
    raw_data = []
    for data in file_data:
        raw_data.extend(data)
    raw_data = [r for r in raw_data if not (r is None)]

    # Use the multiprocessing Pool
    with mp.Pool(num_parallel) as pool:
        func = partial(process_data, action_type=action_type)
        results = list(
            tqdm(
                pool.imap(
                    func, raw_data, chunksize=ceil(len(raw_data) / len(pool._pool))
                ),
                total=len(raw_data),