LLM微调数据集构建：垂直领域网页数据的自动化清洗与标注

import requests
import time
import json

class VerticalDataCollector:
    """垂直领域数据采集器"""

    def __init__(self, api_key):
        self.api_key = api_key

    def get_domestic_ip(self, area="110100", isp="电信", count=2):
        """获取国内代理IP"""
        url = f"http://api.fanproxy.com/?key={self.api_key}"
        params = {
            "area": area,
            "isp": isp,
            "count": count,
            "pattern": "json",
            "rreg": "true",
            "rcity": "true",
            "risp": "true",
            "rexp": "true"
        }
        resp = requests.get(url, params=params, timeout=10)
        return resp.json()

    def get_overseas_ip(self, cnt="AS", cty="JP", count=2):
        """获取海外代理IP"""
        url = f"http://api.fanproxy.com/?key={self.api_key}"
        params = {
            "cnt": cnt,
            "cty": cty,
            "count": count,
            "pattern": "json",
            "rcnt": "true",
            "rcty": "true",
            "rreg": "true",
            "risp": "true",
            "rexp": "true"
        }
        resp = requests.get(url, params=params, timeout=10)
        return resp.json()

    def collect_page(self, target_url, proxy_ip, proxy_port):
        """通过代理IP采集页面"""
        proxy_url = f"http://{proxy_ip}:{proxy_port}"
        proxies = {"http": proxy_url, "https": proxy_url}

        session = requests.Session()
        session.proxies = proxies
        session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                          "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36"
        })

        try:
            response = session.get(target_url, timeout=15)
            if response.status_code == 200:
                return {
                    "url": target_url,
                    "html": response.text,
                    "encoding": response.encoding,
                    "status": response.status_code,
                    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
                }
        except Exception as e:
            print(f"采集失败: {target_url} - {e}")
        return None

# 使用示例：采集国内医疗领域数据
collector = VerticalDataCollector("abc123")

# 获取北京电信代理IP
ip_data = collector.get_domestic_ip(area="110100", isp="电信", count=2)
for ip_info in ip_data.get("data", []):
    result = collector.collect_page(
        "https://medical-journal.example.com/article/123",
        ip_info["ip"],
        ip_info["port"]
    )
    if result:
        print(f"采集成功: {result['url']}, 内容长度: {len(result['html'])}")

import re
import hashlib
from bs4 import BeautifulSoup
from urllib.parse import urlparse

class DataCleaner:
    """网页数据自动化清洗"""

    def __init__(self):
        self.url_seen = set()
        self.content_hashes = set()

    def fix_encoding(self, html, declared_encoding=None):
        """Step 1: 编码修复"""
        if declared_encoding:
            try:
                html = html.decode(declared_encoding).encode('utf-8')
            except Exception:
                pass
        try:
            html = html.decode('utf-8')
        except Exception:
            html = html.decode('utf-8', errors='ignore')
        return html

    def strip_noise_tags(self, html):
        """Step 2: 剥离噪音标签"""
        soup = BeautifulSoup(html, 'html.parser')

        # 移除script、style、nav、footer、aside等噪音标签
        noise_tags = ['script', 'style', 'nav', 'footer', 'aside',
                      'header', 'noscript', 'iframe', 'form']
        for tag in noise_tags:
            for element in soup.find_all(tag):
                element.decompose()

        # 移除常见广告class
        ad_patterns = ['ad', 'advertisement', 'banner', 'popup',
                       'cookie', 'subscribe', 'newsletter']
        for pattern in ad_patterns:
            for element in soup.find_all(class_=re.compile(pattern, re.I)):
                element.decompose()

        return str(soup)

    def extract_main_content(self, html):
        """Step 3: 正文提取"""
        soup = BeautifulSoup(html, 'html.parser')

        # 优先提取article标签
        article = soup.find('article')
        if article:
            return article.get_text(separator='\n', strip=True)

        # 其次提取main标签
        main = soup.find('main')
        if main:
            return main.get_text(separator='\n', strip=True)

        # 回退策略：找文本密度最高的div
        divs = soup.find_all('div')
        if divs:
            best_div = max(divs, key=lambda d: len(d.get_text(strip=True)))
            return best_div.get_text(separator='\n', strip=True)

        return soup.get_text(separator='\n', strip=True)

    def filter_noise_text(self, text):
        """Step 4: 噪音文本过滤"""
        lines = text.split('\n')
        filtered = []

        for line in lines:
            line = line.strip()
            if not line:
                continue
            # 过滤过短的行（通常是导航或模板文字）
            if len(line) < 20:
                continue
            # 过滤cookie/隐私提示类文本
            noise_keywords = ['cookie', '隐私政策', '同意', '订阅',
                            'copyright', '©', 'all rights reserved']
            if any(kw in line.lower() for kw in noise_keywords):
                continue
            filtered.append(line)

        return '\n'.join(filtered)

    def deduplicate(self, url, content):
        """Step 6: 去重"""
        # URL去重
        normalized_url = urlparse(url).path
        if normalized_url in self.url_seen:
            return False
        self.url_seen.add(normalized_url)

        # 内容指纹去重
        content_hash = hashlib.md5(content.encode()).hexdigest()
        if content_hash in self.content_hashes:
            return False
        self.content_hashes.add(content_hash)

        return True

    def score_quality(self, text):
        """Step 7: 质量评分（0-100）"""
        score = 0

        # 文本长度（权重30分）
        if len(text) > 2000:
            score += 30
        elif len(text) > 500:
            score += 20
        elif len(text) > 200:
            score += 10

        # 信息密度：字母和数字占比（权重20分）
        if len(text) > 0:
            alpha_num_ratio = sum(c.isalnum() for c in text) / len(text)
            if alpha_num_ratio > 0.6:
                score += 20
            elif alpha_num_ratio > 0.4:
                score += 10

        # 段落结构：换行符密度（权重20分）
        paragraph_density = text.count('\n') / max(len(text), 1)
        if 0.005 < paragraph_density < 0.05:
            score += 20

        # 专业术语密度（权重15分）
        # 根据垂直领域自定义术语表
        domain_terms = ['诊断', '治疗', '症状', '临床', '患者',  # 医疗示例
                       '剂量', '禁忌', '不良反应', '适应症']
        term_count = sum(text.count(term) for term in domain_terms)
        if term_count > 5:
            score += 15
        elif term_count > 2:
            score += 8

        # 编码完整性（权重15分）
        try:
            text.encode('utf-8')
            score += 15
        except Exception:
            pass

        return score

    def clean(self, html, url):
        """完整清洗流程"""
        # Step 1: 编码修复
        html = self.fix_encoding(html)

        # Step 2: 标签剥离
        html = self.strip_noise_tags(html)

        # Step 3: 正文提取
        text = self.extract_main_content(html)

        # Step 4: 噪音过滤
        text = self.filter_noise_text(text)

        # Step 6: 去重
        is_unique = self.deduplicate(url, text)
        if not is_unique:
            return None

        # Step 7: 质量评分
        quality_score = self.score_quality(text)

        return {
            "url": url,
            "text": text,
            "quality_score": quality_score,
            "length": len(text)
        }

# 使用示例
cleaner = DataCleaner()
result = cleaner.clean(raw_html, "https://example.com/article/1")
if result and result["quality_score"] >= 60:
    print(f"清洗完成，质量分: {result['quality_score']}")

class InstructionGenerator:
    """指令数据自动构造器"""

    def __init__(self, llm_api_key):
        self.llm_api_key = llm_api_key
        self.llm_url = "https://api.llm-provider.com/v1/chat/completions"

    def generate_qa_pairs(self, source_text, domain="医疗", num_pairs=3):
        """从专业文本中生成问答对"""
        prompt = f"""你是一个{domain}领域的数据标注专家。
请从以下专业文本中提取{num_pairs}个高质量的问答对，用于大模型微调。

要求：
1. 问题应该具体、明确，模拟真实用户可能提出的疑问
2. 答案必须严格基于原文内容，不得编造
3. 答案应完整、准确，包含原文中的关键信息
4. 输出JSON格式

专业文本：
{source_text}

输出格式：
[{{"instruction": "问题", "output": "答案"}}, ...]
"""

        headers = {
            "Authorization": f"Bearer {self.llm_api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "your-model-name",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3
        }

        try:
            resp = requests.post(self.llm_url, json=payload,
                               headers=headers, timeout=30)
            result = resp.json()
            content = result["choices"][0]["message"]["content"]

            # 解析LLM输出的JSON
            import json
            qa_pairs = json.loads(content)
            return qa_pairs
        except Exception as e:
            print(f"标注失败: {e}")
            return []

    def validate_qa_pair(self, qa_pair, source_text):
        """校验问答对质量"""
        instruction = qa_pair.get("instruction", "")
        output = qa_pair.get("output", "")

        issues = []

        # 长度检查
        if len(instruction) < 10:
            issues.append("问题过短")
        if len(output) < 20:
            issues.append("答案过短")

        # 答案与原文的覆盖度
        output_chars = set(output)
        source_chars = set(source_text)
        overlap = len(output_chars & source_chars) / max(len(output_chars), 1)
        if overlap < 0.5:
            issues.append("答案与原文覆盖度低，可能含编造内容")

        # 问题是否为疑问句
        question_markers = ['？', '?', '什么是', '如何', '为什么', '哪些', '怎样']
        if not any(marker in instruction for marker in question_markers):
            issues.append("问题不是疑问句")

        return {
            "valid": len(issues) == 0,
            "issues": issues,
            "overlap_score": overlap
        }

    def build_dataset(self, cleaned_articles, domain="医疗"):
        """构建完整微调数据集"""
        dataset = []

        for article in cleaned_articles:
            if article["quality_score"] < 60:
                continue

            # 生成问答对
            qa_pairs = self.generate_qa_pairs(
                article["text"], domain=domain, num_pairs=3
            )

            for qa in qa_pairs:
                # 质量校验
                validation = self.validate_qa_pair(qa, article["text"])
                if validation["valid"]:
                    dataset.append({
                        "instruction": qa["instruction"],
                        "output": qa["output"],
                        "source_url": article["url"],
                        "quality_score": article["quality_score"],
                        "overlap_score": validation["overlap_score"]
                    })

        return dataset

# 使用示例
generator = InstructionGenerator("your_llm_api_key")

# 从清洗后的文章构建微调数据集
cleaned_articles = [
    {"url": "https://example.com/medical/1",
     "text": "阿司匹林是一种非甾体抗炎药...", "quality_score": 85}
]

dataset = generator.build_dataset(cleaned_articles, domain="医疗")
print(f"生成微调样本数: {len(dataset)}")

class FineTuningDatasetBuilder:
    """微调数据集构建完整流水线"""

    def __init__(self, proxy_api_key, llm_api_key):
        self.collector = VerticalDataCollector(proxy_api_key)
        self.cleaner = DataCleaner()
        self.generator = InstructionGenerator(llm_api_key)

    def build(self, url_list, domain="医疗", use_overseas=False,
              country=None, area=None):
        """从URL列表构建完整微调数据集"""
        raw_pages = []
        cleaned_articles = []

        # 阶段1：采集
        print(f"[阶段1] 采集 {len(url_list)} 个页面")
        for url in url_list:
            if use_overseas:
                ip_data = self.collector.get_overseas_ip(cty=country, count=1)
            else:
                ip_data = self.collector.get_domestic_ip(area=area, count=1)

            for ip_info in ip_data.get("data", []):
                page = self.collector.collect_page(
                    url, ip_info["ip"], ip_info["port"]
                )
                if page:
                    raw_pages.append(page)
                    break
            time.sleep(3)  # 请求间隔

        # 阶段2：清洗
        print(f"[阶段2] 清洗 {len(raw_pages)} 个页面")
        for page in raw_pages:
            result = self.cleaner.clean(page["html"], page["url"])
            if result and result["quality_score"] >= 60:
                cleaned_articles.append(result)

        # 阶段3：标注
        print(f"[阶段3] 标注 {len(cleaned_articles)} 篇文章")
        dataset = self.generator.build_dataset(cleaned_articles, domain)

        # 阶段4：格式化输出
        print(f"[阶段4] 输出 {len(dataset)} 条微调样本")
        return dataset

# 使用示例
builder = FineTuningDatasetBuilder(
    proxy_api_key="abc123",
    llm_api_key="your_llm_api_key"
)

# 采集国内医疗领域数据并构建微调数据集
medical_urls = [
    "https://medical-journal.example.com/article/1",
    "https://medical-journal.example.com/article/2",
    "https://medical-journal.example.com/article/3",
]

dataset = builder.build(
    url_list=medical_urls,
    domain="医疗",
    use_overseas=False,
    area="110100"  # 北京
)

# 保存为JSONL
import json
with open("medical_finetune_dataset.jsonl", "w", encoding="utf-8") as f:
    for sample in dataset:
        f.write(json.dumps(sample, ensure_ascii=False) + "\n")

print(f"数据集已保存: {len(dataset)} 条样本")