实时新闻流的智能去重与
事件聚类系统

一、系统架构总览

整个系统分为四个核心层次，每层职责清晰、可独立扩展：

• 采集层：RSS/API 订阅原始新闻流，推送到 AWS Kinesis Data Streams
• 预处理层：Lambda 函数触发，完成分词、去停用词、NER 提取
• 聚类层：向量化 → 相似度计算 → 凝聚层次聚类（AHC），产出事件簇
• 存储&展示层：聚类结果写入 DynamoDB / Elasticsearch，提供搜索与摘要 API

二、环境准备与依赖安装

# 创建并激活虚拟环境
python3 -m venv news-cluster-env
source news-cluster-env/bin/activate  # Windows: .\news-cluster-env\Scripts\activate

# 升级 pip
pip install --upgrade pip

# 创建 requirements.txt
cat > requirements.txt << 'EOF'
scikit-learn==1.4.2
nltk==3.8.1
spacy==3.7.4
sentence-transformers==3.1.1
numpy==1.26.4
pandas==2.2.2
boto3==1.34.100
elasticsearch==8.13.0
python-dotenv==1.0.1
EOF

pip install -r requirements.txt

# 下载 NLTK 语料库
python -c "import nltk; nltk.download('stopwords'); nltk.download('punkt'); nltk.download('punkt_tab')"

# 下载 spaCy 英文模型（中文新闻用 zh_core_web_sm）
python -m spacy download en_core_web_sm
python -m spacy download zh_core_web_sm

# .env 文件
AWS_REGION=us-east-1
KINESIS_STREAM_NAME=news-stream
DYNAMODB_TABLE=news-clusters
ES_HOST=https://your-es-host:9200
ES_API_KEY=your_api_key_here

# 聚类参数
SIMILARITY_THRESHOLD=0.35      # 余弦相似度阈值（越小聚类越粗）
CLUSTER_TIME_WINDOW=86400      # 时间窗口（秒），默认 24 小时
MAX_CLUSTER_SIZE=50            # 单簇最大文章数

三、文本预处理模块

预处理是整个系统的基石。新闻标题和正文中存在大量噪音：HTML 标签、特殊符号、广告词等。 我们需要提取干净的文本并标准化。

# preprocessor.py
import re
import nltk
import spacy
from nltk.corpus import stopwords

nlp_en = spacy.load("en_core_web_sm")
nlp_zh = spacy.load("zh_core_web_sm")

STOP_WORDS_EN = set(stopwords.words('english'))
# 中文停用词（精简版，可换成更完整的词表）
STOP_WORDS_ZH = {"的", "了", "在", "是", "我", "有", "和", "就", "不", "人",
                  "都", "一", "一个", "上", "也", "很", "到", "说", "要", "去",
                  "你", "会", "着", "没有", "看", "好", "自己", "这"}

def clean_html(text: str) -> str:
    """去除 HTML 标签和多余空白"""
    text = re.sub(r'<[^>]+>', ' ', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

def preprocess_en(text: str) -> str:
    """英文预处理：分词、去停用词、词形还原"""
    text = clean_html(text.lower())
    doc = nlp_en(text)
    tokens = [
        token.lemma_ for token in doc
        if token.is_alpha
        and not token.is_stop
        and token.lemma_ not in STOP_WORDS_EN
        and len(token.lemma_) > 1
    ]
    return ' '.join(tokens)

def preprocess_zh(text: str) -> str:
    """中文预处理：分词、去停用词"""
    text = clean_html(text)
    doc = nlp_zh(text)
    tokens = [
        token.text for token in doc
        if not token.is_space
        and token.text not in STOP_WORDS_ZH
        and len(token.text) > 1
    ]
    return ' '.join(tokens)

def detect_language(text: str) -> str:
    """简单语言检测：中文字符占比超过 30% 判断为中文"""
    zh_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
    return 'zh' if zh_chars / max(len(text), 1) > 0.3 else 'en'

def preprocess(text: str) -> str:
    lang = detect_language(text)
    return preprocess_zh(text) if lang == 'zh' else preprocess_en(text)

四、向量化与相似度计算

向量化是决定聚类质量的关键。我们提供两种方案：传统 TF-IDF（速度快、资源少） 和 Sentence-BERT（语义更精准但需 GPU 加速）。对于大多数新闻去重场景，TF-IDF 已足够。

方案 A：TF-IDF 向量化（推荐生产使用）

# vectorizer.py
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class TFIDFVectorizer:
    def __init__(self, max_features: int = 10000, ngram_range=(1, 2)):
        self.vectorizer = TfidfVectorizer(
            max_features=max_features,
            ngram_range=ngram_range,     # 同时使用单词和双词组合
            sublinear_tf=True,           # 对词频取对数，减少高频词权重
            min_df=2,                    # 至少出现2篇文章的词才纳入词典
            max_df=0.95                  # 超过95%文章都有的词（太常见）忽略
        )
        self._fitted = False

    def fit_transform(self, texts: list[str]) -> np.ndarray:
        matrix = self.vectorizer.fit_transform(texts)
        self._fitted = True
        return matrix.toarray()

    def transform(self, texts: list[str]) -> np.ndarray:
        if not self._fitted:
            raise RuntimeError("Vectorizer not fitted yet. Call fit_transform first.")
        return self.vectorizer.transform(texts).toarray()

    def similarity(self, vec_a: np.ndarray, vec_b: np.ndarray) -> float:
        """计算两个向量的余弦相似度"""
        return float(cosine_similarity(vec_a.reshape(1, -1), vec_b.reshape(1, -1))[0][0])

方案 B：Sentence-BERT 语义向量（精度更高）

# semantic_vectorizer.py
from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticVectorizer:
    """
    推荐模型：
    - 英文：'all-MiniLM-L6-v2'（快速，384维）
    - 多语言：'paraphrase-multilingual-MiniLM-L12-v2'
    - 中文：'shibing624/text2vec-base-chinese'
    """
    def __init__(self, model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2'):
        self.model = SentenceTransformer(model_name)

    def encode(self, texts: list[str], batch_size: int = 32) -> np.ndarray:
        return self.model.encode(
            texts,
            batch_size=batch_size,
            show_progress_bar=True,
            normalize_embeddings=True   # 归一化后余弦相似度等价于点积，更快
        )

五、凝聚层次聚类实现

凝聚层次聚类（AHC）的核心优势是无需预先指定簇数——通过距离阈值自动决定分多少组， 非常适合新闻聚类场景（我们事先无法知道当天会有多少个独立事件）。

# clusterer.py
import numpy as np
from sklearn.cluster import AgglomerativeClustering
from sklearn.metrics.pairwise import cosine_similarity
from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class NewsArticle:
    id: str
    title: str
    content: str
    published_at: datetime
    source: str
    vector: np.ndarray = field(default=None, repr=False)
    entities: dict = field(default_factory=dict)  # NER 抽取的实体

@dataclass
class NewsCluster:
    cluster_id: str
    articles: list[NewsArticle] = field(default_factory=list)
    representative: NewsArticle = None     # 代表性文章（最早发布）
    main_entities: dict = field(default_factory=dict)

class NewsClusterer:
    def __init__(self, distance_threshold: float = 0.65, linkage: str = 'average'):
        """
        distance_threshold: 1 - cosine_similarity，值越小聚类越严格
          - 0.3: 只聚合几乎相同的报道（近似去重）
          - 0.65: 聚合同一事件的相关报道（推荐）
          - 1.0: 主题级别聚合（过于宽松）
        linkage: 'average'（组间平均距离）最适合新闻场景
        """
        self.distance_threshold = distance_threshold
        self.linkage = linkage

    def cluster(self, articles: list[NewsArticle]) -> list[NewsCluster]:
        if len(articles) == 0:
            return []
        if len(articles) == 1:
            return [self._make_cluster([articles[0]])]

        # 构建向量矩阵
        vectors = np.array([a.vector for a in articles])

        # 余弦距离矩阵（距离 = 1 - 相似度）
        similarity_matrix = cosine_similarity(vectors)
        distance_matrix = 1 - similarity_matrix
        np.fill_diagonal(distance_matrix, 0)

        # 执行 AHC
        model = AgglomerativeClustering(
            n_clusters=None,
            distance_threshold=self.distance_threshold,
            linkage=self.linkage,
            metric='precomputed'   # 使用预计算的距离矩阵
        )
        labels = model.fit_predict(distance_matrix)

        # 按簇标签分组
        clusters_map: dict[int, list] = {}
        for article, label in zip(articles, labels):
            clusters_map.setdefault(label, []).append(article)

        return [self._make_cluster(group) for group in clusters_map.values()]

    def _make_cluster(self, articles: list[NewsArticle]) -> NewsCluster:
        # 按发布时间排序，最早的作为代表文章
        sorted_articles = sorted(articles, key=lambda a: a.published_at)
        cluster_id = f"cluster-{sorted_articles[0].id}-{len(articles)}"

        # 合并实体统计（出现次数最多的实体权重最高）
        entity_counts: dict = {}
        for a in articles:
            for etype, names in a.entities.items():
                for name in names:
                    key = (etype, name)
                    entity_counts[key] = entity_counts.get(key, 0) + 1

        # 取每类前3个高频实体
        main_entities: dict = {}
        for (etype, name), count in sorted(entity_counts.items(), key=lambda x: -x[1]):
            main_entities.setdefault(etype, [])
            if len(main_entities[etype]) < 3:
                main_entities[etype].append(name)

        return NewsCluster(
            cluster_id=cluster_id,
            articles=sorted_articles,
            representative=sorted_articles[0],
            main_entities=main_entities
        )

六、命名实体识别增强聚类

纯文本相似度聚类的致命弱点：两篇讲"苹果公司"和"苹果水果"的文章可能有较高词汇重叠， 但实际属于完全不同的事件。引入命名实体识别（NER）后， 我们可以将人名、机构名、地名等关键实体纳入相似度计算，大幅提升聚类的事件相关性。

# ner_extractor.py
import spacy
from collections import defaultdict

nlp_en = spacy.load("en_core_web_sm")
nlp_zh = spacy.load("zh_core_web_sm")

# 关注的实体类型
RELEVANT_TYPES = {
    "PERSON",   # 人名
    "ORG",      # 组织机构
    "GPE",      # 地缘政治实体（国家、城市）
    "LOC",      # 地点
    "EVENT",    # 事件名称
    "MONEY",    # 金额（财经新闻重要）
    "DATE",     # 日期（辅助时效过滤）
}

def extract_entities(text: str, lang: str = 'en') -> dict[str, list[str]]:
    nlp = nlp_zh if lang == 'zh' else nlp_en
    doc = nlp(text[:512])  # spaCy 对超长文本较慢，取前512字符即可

    entities = defaultdict(list)
    for ent in doc.ents:
        if ent.label_ in RELEVANT_TYPES:
            normalized = ent.text.strip().lower()
            if normalized not in entities[ent.label_]:
                entities[ent.label_].append(normalized)

    return dict(entities)

def entity_overlap_score(entities_a: dict, entities_b: dict) -> float:
    """
    计算两篇文章的实体重叠分数
    返回值 0-1，越高说明涉及相同事件实体的概率越大
    """
    if not entities_a or not entities_b:
        return 0.0

    all_types = set(entities_a) | set(entities_b)
    total_score = 0.0
    weights = {"PERSON": 2.0, "ORG": 1.5, "GPE": 1.2, "EVENT": 2.0}

    for etype in all_types:
        a_set = set(entities_a.get(etype, []))
        b_set = set(entities_b.get(etype, []))
        if not a_set or not b_set:
            continue
        jaccard = len(a_set & b_set) / len(a_set | b_set)
        weight = weights.get(etype, 1.0)
        total_score += jaccard * weight

    max_possible = sum(weights.get(t, 1.0) for t in all_types)
    return min(total_score / max(max_possible, 1e-8), 1.0)

def combined_similarity(text_sim: float, entity_sim: float,
                         text_weight: float = 0.6, entity_weight: float = 0.4) -> float:
    """融合文本相似度和实体重叠分数"""
    return text_sim * text_weight + entity_sim * entity_weight

七、实时流处理架构接入

生产环境中，新闻不是批量到达的，而是实时涌入的。我们使用 AWS Kinesis + Lambda 构建事件驱动架构， 每条新闻到达时自动触发去重与聚类。

# lambda_handler.py — 部署到 AWS Lambda
import json
import os
import boto3
import numpy as np
from datetime import datetime, timezone
from preprocessor import preprocess, detect_language
from ner_extractor import extract_entities
from vectorizer import TFIDFVectorizer
from clusterer import NewsArticle, NewsClusterer

# 全局变量（Lambda 复用容器时可复用）
vectorizer = TFIDFVectorizer()
clusterer = NewsClusterer(distance_threshold=float(os.environ.get('SIMILARITY_THRESHOLD', 0.65)))
dynamodb = boto3.resource('dynamodb', region_name=os.environ['AWS_REGION'])
table = dynamodb.Table(os.environ['DYNAMODB_TABLE'])

def lambda_handler(event, context):
    """
    Kinesis 触发器调用此函数
    event['Records'] 包含一批新闻记录（最多 100 条）
    """
    articles = []

    for record in event['Records']:
        # Kinesis 数据是 Base64 编码的
        payload = json.loads(
            __import__('base64').b64decode(record['kinesis']['data']).decode('utf-8')
        )

        # 合并标题+摘要作为主要文本
        full_text = f"{payload.get('title', '')} {payload.get('summary', '')}"
        lang = detect_language(full_text)
        processed_text = preprocess(full_text)
        entities = extract_entities(full_text, lang)

        article = NewsArticle(
            id=payload['id'],
            title=payload.get('title', ''),
            content=processed_text,
            published_at=datetime.fromisoformat(payload.get('published_at', datetime.now(timezone.utc).isoformat())),
            source=payload.get('source', 'unknown'),
            entities=entities
        )
        articles.append((article, processed_text))

    if not articles:
        return {'statusCode': 200, 'body': 'No articles to process'}

    # 批量向量化
    texts = [a[1] for a in articles]
    vectors = vectorizer.fit_transform(texts)
    for i, (article, _) in enumerate(articles):
        article[0].vector = vectors[i]

    article_objects = [a[0] for a in articles]

    # 执行聚类
    clusters = clusterer.cluster(article_objects)

    # 写入 DynamoDB
    with table.batch_writer() as batch:
        for cluster in clusters:
            batch.put_item(Item={
                'cluster_id': cluster.cluster_id,
                'article_count': len(cluster.articles),
                'representative_title': cluster.representative.title,
                'representative_source': cluster.representative.source,
                'published_at': cluster.representative.published_at.isoformat(),
                'main_entities': json.dumps(cluster.main_entities, ensure_ascii=False),
                'article_ids': [a.id for a in cluster.articles],
                'ttl': int(datetime.now(timezone.utc).timestamp()) + 7 * 86400  # 7天后自动过期
            })

    return {
        'statusCode': 200,
        'body': json.dumps({
            'processed': len(article_objects),
            'clusters_formed': len(clusters)
        })
    }

# 本地测试脚本 test_pipeline.py
import json
import base64
from datetime import datetime, timezone

def make_kinesis_event(articles: list[dict]) -> dict:
    """模拟 Kinesis 触发 Lambda 的 event 格式"""
    records = []
    for article in articles:
        data = base64.b64encode(json.dumps(article, ensure_ascii=False).encode()).decode()
        records.append({
            'kinesis': {'data': data},
            'eventSource': 'aws:kinesis'
        })
    return {'Records': records}

# 测试数据
test_articles = [
    {"id": "001", "title": "特斯拉Q1财报超预期，营收同比增长30%", "summary": "特斯拉发布2026年第一季度财报，营收达到300亿美元", "published_at": "2026-02-28T10:00:00Z", "source": "财联社"},
    {"id": "002", "title": "马斯克：特斯拉一季度业绩强劲，电动车交付量创新高", "summary": "特斯拉CEO马斯克表示Q1表现优于市场预期", "published_at": "2026-02-28T10:30:00Z", "source": "Reuters"},
    {"id": "003", "title": "苹果发布iOS 20正式版，带来全新AI功能", "summary": "苹果公司宣布iOS 20系统更新，新增多项AI辅助功能", "published_at": "2026-02-28T11:00:00Z", "source": "MacRumors"},
]

from lambda_handler import lambda_handler
result = lambda_handler(make_kinesis_event(test_articles), None)
print(json.dumps(result, indent=2))
# 期望：文章001和002被聚合为同一个特斯拉财报事件簇，文章003独立成簇

八、常见问题排查