利用 B 站评论区数据构建大语言模型微调数据集

该数据格式符合 Aquila 等开源模型微调所需的对话格式要求。更多数据样例可参考项目源码仓库。

前言

在构建大语言模型微调数据集时，往往面临中文高质量对话数据稀缺的问题。许多公开数据集基于 GPT 系列模型且多为英文。作为视频平台用户，B 站的视频评论区提供了大量贴近生活的真实对话场景，且热门视频评论数众多，适合构建中文对话数据集。

注：本文主要展示数据构建流程，实际微调效果需结合具体任务验证。

基本思路

视频评论区的结构通常呈现为树状层级。每一层可以作为一个主题，其下每一条评论可以作为一个对话节点。默认情况下，评论是对上一层回复，但也存在跨层交流的情况。使用数据结构表示，大致形成一棵树：

graph LR
A[主题 1] --> B[评论 1]
A --> C[评论 2]
A --> D[评论 3]
B --> E[评论 4]
B --> F[评论 5]
C --> G[评论 6]
C --> H[评论 7]
D --> I[评论 8]

从树的根节点开始，遍历至叶子结点，每一条路径即构成一段完整对话。基本思路是爬取评论数据，构建树结构，提取所有根到叶子的路径，筛选后保存为对话数据集。

爬虫实现

接口分析

打开视频评论区，通过开发者工具（F12）观察网络请求。评论数据通常位于分页请求的 data.replies 字段中，每条代表一层，包含 rpid 字段标识评论 ID。子评论同样嵌套在 replies 字段中，支持多层级嵌套。

由于需要获取两层分页（第一层为视频下的评论页，第二层为每条评论下的回复页），且数据量较大，建议使用异步并发方式提升效率。

协程与协程池

Python 3.4+ 通过 asyncio 模块支持协程。相比线程，协程更轻量，切换无需操作系统参与。使用 async 和 await 定义协程，通过 asyncio.run() 运行。

为避免对目标网站造成过大压力，需限制并发数量。可使用 asyncio.Semaphore 实现协程池，控制同时运行的协程数。

Cookie 认证

B 站 API 通常需要登录态。需先登录获取 SESSDATA, BILI_JCT 等 Cookie，并保存在 .env 文件中，使用 python-dotenv 读取。

from dotenv import load_dotenv
import os

load_dotenv()
env = os.environ

SESSDATA = env.get('SESSDATA', '')
BILI_JCT = env.get('BILI_JCT', '')
BUVID3 = env.get('BUVID3', '')
DEDE_USER_ID = env.get('DEDE_USER_ID', '')
AT_TIME_VALUE = env.get('AT_TIME_VALUE', '')

代码实现

以下代码展示了如何使用 httpx 进行异步请求，并结合协程池爬取评论。

import asyncio
import json
from typing import Any, Dict, List, Optional, TypeAlias

import httpx
from bilibili_api.credential import Credential

# config 中包含了必要的 Cookie 参数
from config import *

JSON_TYPE: TypeAlias = Dict[str, Any]

COMMON_HEADERS = {
    "Origin": "https://www.bilibili.com",
    "Authority": "api.bilibili.com",
    "Sec-Ch-Ua": '"Chromium";v="116", "Not)A;Brand";v="24", "Microsoft Edge";v="116"',
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko)"
}

if not (SESSDATA and BILI_JCT and BUVID3 and DEDE_USER_ID and AT_TIME_VALUE):
    raise ValueError("请在 .env 中填写 SESSDATA, BILI_JCT, BUVID3, DEDE_USER_ID, AT_TIME_VALUE")

credential = Credential(sessdata=SESSDATA, bili_jct=BILI_JCT,
                        buvid3=BUVID3, dedeuserid=DEDE_USER_ID, ac_time_value=AT_TIME_VALUE)
print("credential: ", credential.get_cookies())

# 假设已导入 AsyncPool 类
from async_pool import AsyncPool
pool = AsyncPool(maxsize=16)

async def get_html(url: str, params: Dict = None, headers: Dict = None, cookies: Dict = None, timeout: int = 30, client: httpx.AsyncClient = None):
    m_client = client
    try:
        if client is None:
            m_client = httpx.AsyncClient()
        r = await m_client.get(url, timeout=timeout, params=params, headers=headers, cookies=cookies)
        r.raise_for_status()
        return r.text
    except Exception as e:
        print(f"Request error: {e}")
        return "Error"
    finally:
        if client is None and m_client is not None:
            await m_client.aclose()

async def get_one_page(oid: int, pagination_str: str, client: httpx.AsyncClient = None):
    """获取范围：一个回复页"""
    params = {
        "type": 1,
        "oid": oid,
        "mode": 2,
        "pagination_str": '{"offset":"%s"}' % pagination_str.replace('"', r"\""),
    }
    url = "https://api.bilibili.com/x/v2/reply/main"
    text = await get_html(url, params, COMMON_HEADERS, cookies=credential.get_cookies(), client=client)
    obj = json.loads(text)
    return obj

async def crawl_one_video(oid: int):
    """爬取一个视频的所有评论"""
    print(f"- 开始爬取视频 {oid} 的评论")
    url = "https://api.bilibili.com/x/v2/reply/count"
    params = {"type": 1, "oid": oid}
    text = await get_html(url, params, COMMON_HEADERS)
    obj = json.loads(text)
    total_page: int = obj["data"]["count"] // 20 + 1
    print(f"- 视频 {oid} 一共有 {total_page} 页评论")
    
    pagination = ''
    async with httpx.AsyncClient() as client:
        for page in range(1, total_page + 1):
            # 此处简化调用逻辑，实际需处理分页字符串更新
            next_page = await crawl_one_page_video(oid, page, pagination_str=pagination, client=client)
            if next_page is None:
                break
            pagination = next_page
            await asyncio.sleep(0.1)

async def main():
    await refresh_cookie_if_necessary()
    await crawl_one_video(2)

if __name__ == "__main__":
    asyncio.run(main())

构建数据集

数据结构解析

爬取到的 replies 列表中包含大量信息，关键字段如下：

• rpid: 评论 ID
• parent / root: 父评论 ID 和根评论 ID，用于构建树关系
• member.uname: 用户名
• content.message: 评论内容

提取核心信息并整理为字典结构：

{
    476670: {'parent': 0, 'content': '貌似没人来', 'uname': '残星什么的就是残星'},
    214198179: {'parent': 476670, 'content': '可怜的二楼 (=・ω・=)', 'uname': '初音ハク'},
    # ... 更多评论
}

树构建与对话提取

根据 parent 字段构建邻接表形式的树结构，然后从根节点（parent=0）开始深度优先搜索（DFS），遍历所有叶子节点路径。每条路径即为一段对话。

from typing import List, Dict

def build_conv_from_replies(replies_dict) -> List[List[Dict]]:
    conv_tree = {}
    
    # 构建对话树
    for reply_id, reply in replies_dict.items():
        parent_id = reply['parent']
        if parent_id in conv_tree:
            conv_tree[parent_id].append(reply_id)
        else:
            conv_tree[parent_id] = [reply_id]

    longest_paths = []
    path = []

    # DFS 遍历
    def dfs(node):
        nonlocal path
        path.append(node)
        if node not in conv_tree:
            longest_paths.append(path.copy())
        else:
            for child in conv_tree[node]:
                dfs(child)
        path.pop()

    # 从每个根节点开始
    for root in conv_tree.get(0, []):
        dfs(root)

    conversations = []
    for path in longest_paths:
        conversation = []
        for node in path:
            item = replies_dict[node]
            content = item['content']
            # 清理回复标记
            if content.startswith('回复 @'):
                content = content.split(':')[1] if ':' in content else content
            conversation.append({
                'from': item['uname'],
                'value': content
            })
        # 过滤短对话
        if len(conversation) >= 3:
            conversations.append(conversation)
    
    return conversations

数据质量优化建议

为了提升微调效果，建议在原始数据基础上增加以下处理步骤：

1. 去重：移除内容完全相同的评论，避免模型过拟合。
2. 敏感词过滤：去除包含违规、广告或无意义字符的对话。
3. 长度控制：设定最大 token 数，防止单条对话过长导致训练显存溢出。
4. 角色分离：确保对话中的发言者身份明确，避免混淆。
5. 噪声清洗：移除表情符号、特殊链接等非文本干扰项。

总结

本文介绍了一种利用 B 站视频评论区构建中文对话数据集的方法。核心流程包括：通过异步爬虫获取评论数据，依据父子关系构建树状结构，提取根到叶的路径生成对话序列，并进行必要的清洗与格式化。该方法生成的数据集可用于大语言模型的指令微调（Instruction Tuning）或对话生成任务。实际应用中，建议结合具体业务需求进一步调整数据过滤策略，以确保数据集的质量与多样性。