Lychee Rerank与Python爬虫数据智能处理实战：多模态内容重排序应用-平芜编程栈

Lychee Rerank与Python爬虫数据智能处理实战：多模态内容重排序应用

你是不是也遇到过这样的问题？用Python爬虫辛辛苦苦抓了一大堆图文数据，结果发现内容质量参差不齐，有的图文完全不搭，有的内容重复冗余，想要从中筛选出真正有价值的信息，简直就像大海捞针。

传统的文本匹配方法只能看文字，对图片内容视而不见；而单纯用图片相似度，又忽略了文字描述的重要性。这种割裂的处理方式，往往导致推荐结果“货不对板”，用户体验大打折扣。

今天我要分享的这套方案，正好能解决这个痛点。我们结合Python爬虫和Lychee Rerank多模态重排序模型，打造一个智能的内容处理流水线。简单来说，就是让机器不仅能看懂文字，还能理解图片，然后根据你的需求，把最相关、质量最好的内容排在最前面。

1. 场景痛点：为什么需要多模态重排序？

想象一下，你正在开发一个内容聚合平台，每天要从几十个网站爬取上千条图文内容。这些内容可能是新闻资讯、产品介绍、教程文章，或者是社交媒体上的热门帖子。

传统做法会遇到哪些问题？

图文不匹配：标题说“最新款手机发布”，配图却是一张风景照
内容重复：同一事件被多个媒体转载，内容大同小异
质量参差：有的文章配图高清专业，有的却是模糊的截图
相关性弱：用户搜索“Python数据分析”，结果混入了大量不相关的编程教程

单纯依靠关键词匹配或者向量相似度，很难全面评估一篇图文内容的综合质量。这就是为什么我们需要引入多模态重排序——让AI同时理解文字和图片，做出更智能的判断。

Lychee Rerank这个模型，就是专门为这种场景设计的。它基于Qwen2.5-VL-Instruct开发，能够同时处理文本和图像信息，在召回结果的基础上进行精细化的重新排序。

2. 整体方案设计：从爬虫到智能排序

我们的方案分为四个主要步骤，形成一个完整的数据处理闭环：

graph TD A[Python爬虫数据采集] --> B[数据清洗与预处理] B --> C[多模态特征提取] C --> D[Lychee Rerank智能排序] D --> E[排序结果输出与应用]

2.1 第一步：Python爬虫数据采集

爬虫部分我们使用经典的requests和BeautifulSoup组合，配合selenium处理动态加载的内容。这里的关键是要同时抓取文本内容和图片信息。

import requests from bs4 import BeautifulSoup from urllib.parse import urljoin import os from PIL import Image from io import BytesIO class ContentCrawler: def __init__(self, base_url): self.base_url = base_url self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }) def fetch_article(self, url): """抓取单篇文章的图文内容""" try: response = self.session.get(url, timeout=10) response.raise_for_status() soup = BeautifulSoup(response.content, 'html.parser') # 提取文章标题 title = soup.find('h1').text.strip() if soup.find('h1') else '' # 提取正文内容 content_elements = soup.find_all(['p', 'h2', 'h3']) content = ' '.join([elem.text.strip() for elem in content_elements]) # 提取图片 images = [] for img in soup.find_all('img'): img_url = img.get('src') if img_url: full_url = urljoin(url, img_url) images.append(full_url) return { 'url': url, 'title': title, 'content': content[:1000], # 限制长度 'images': images[:5], # 最多取5张图片 'timestamp': datetime.now().isoformat() } except Exception as e: print(f"抓取失败 {url}: {e}") return None def download_image(self, image_url, save_path): """下载并保存图片""" try: response = self.session.get(image_url, timeout=5) if response.status_code == 200: img = Image.open(BytesIO(response.content)) img.save(save_path) return save_path except Exception as e: print(f"图片下载失败 {image_url}: {e}") return None

这个爬虫类会同时抓取文章的标题、正文内容和图片链接，为后续的多模态处理做好准备。

2.2 第二步：数据清洗与预处理

爬取到的原始数据往往包含大量噪音，我们需要进行清洗和标准化处理。

import re from typing import List, Dict import hashlib class DataCleaner: def __init__(self): self.stop_words = set(['的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这']) def clean_text(self, text: str) -> str: """清洗文本内容""" if not text: return "" # 移除HTML标签 text = re.sub(r'<[^>]+>', '', text) # 移除特殊字符和多余空格 text = re.sub(r'[^\w\u4e00-\u9fff\s]', '', text) text = re.sub(r'\s+', ' ', text).strip() # 移除停用词（可选，根据需求调整） words = text.split() filtered_words = [w for w in words if w not in self.stop_words] return ' '.join(filtered_words) def remove_duplicates(self, articles: List[Dict]) -> List[Dict]: """基于内容相似度去重""" unique_articles = [] content_hashes = set() for article in articles: # 生成内容指纹 content = article.get('title', '') + ' ' + article.get('content', '') content_hash = hashlib.md5(self.clean_text(content).encode()).hexdigest() if content_hash not in content_hashes: content_hashes.add(content_hash) unique_articles.append(article) return unique_articles def prepare_for_rerank(self, article: Dict) -> Dict: """准备重排序所需的数据格式""" return { 'id': article.get('url', ''), 'text': f"{article.get('title', '')} {article.get('content', '')}", 'image_paths': article.get('local_images', []), # 本地图片路径 'metadata': { 'source_url': article.get('url', ''), 'timestamp': article.get('timestamp', ''), 'image_count': len(article.get('images', [])) } }

数据清洗的关键是去除噪音、标准化格式，并为每篇文章生成唯一的标识。这样能确保后续处理的准确性和效率。

2.3 第三步：Lychee Rerank多模态重排序

这是整个方案的核心部分。我们将使用Lychee Rerank模型对爬取的内容进行智能排序。

import torch from transformers import AutoModel, AutoProcessor from typing import List, Dict import numpy as np class LycheeReranker: def __init__(self, model_name="lychee-rerank-mm"): """初始化Lychee Rerank模型""" self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") print(f"使用设备: {self.device}") # 加载模型和处理器 self.processor = AutoProcessor.from_pretrained(model_name) self.model = AutoModel.from_pretrained(model_name).to(self.device) self.model.eval() def encode_multimodal(self, text: str, image_path: str = None): """编码多模态输入""" try: inputs = {} # 处理文本 if text: text_inputs = self.processor( text=text, return_tensors="pt", padding=True, truncation=True, max_length=512 ) inputs.update({f"text_{k}": v.to(self.device) for k, v in text_inputs.items()}) # 处理图片 if image_path: from PIL import Image image = Image.open(image_path).convert("RGB") image_inputs = self.processor( images=image, return_tensors="pt" ) inputs.update({f"image_{k}": v.to(self.device) for k, v in image_inputs.items()}) return inputs except Exception as e: print(f"编码失败: {e}") return None def calculate_relevance(self, query: str, candidates: List[Dict]) -> List[Dict]: """计算查询与候选内容的相关性分数""" results = [] # 编码查询 query_inputs = self.encode_multimodal(query) if query_inputs is None: return candidates with torch.no_grad(): query_features = self.model(**query_inputs) for candidate in candidates: score = 0 valid_modalities = 0 # 文本相关性 if candidate.get('text'): candidate_inputs = self.encode_multimodal(candidate['text']) if candidate_inputs: candidate_features = self.model(**candidate_inputs) text_similarity = torch.cosine_similarity( query_features.last_hidden_state.mean(dim=1), candidate_features.last_hidden_state.mean(dim=1) ) score += text_similarity.item() valid_modalities += 1 # 图片相关性（如果有图片） image_paths = candidate.get('image_paths', []) if image_paths: image_scores = [] for img_path in image_paths[:3]: # 最多处理3张图片 candidate_inputs = self.encode_multimodal(None, img_path) if candidate_inputs: candidate_features = self.model(**candidate_inputs) image_similarity = torch.cosine_similarity( query_features.last_hidden_state.mean(dim=1), candidate_features.last_hidden_state.mean(dim=1) ) image_scores.append(image_similarity.item()) if image_scores: score += np.mean(image_scores) valid_modalities += 1 # 计算平均分数 final_score = score / valid_modalities if valid_modalities > 0 else 0 results.append({ **candidate, 'relevance_score': final_score, 'modalities_used': valid_modalities }) return results def rerank(self, query: str, candidates: List[Dict], top_k: int = 10) -> List[Dict]: """执行重排序并返回top_k结果""" scored_candidates = self.calculate_relevance(query, candidates) # 按相关性分数排序 sorted_candidates = sorted( scored_candidates, key=lambda x: x['relevance_score'], reverse=True ) return sorted_candidates[:top_k]

这个重排序器会同时考虑文本和图片的相关性，为每篇内容计算一个综合得分。你可以根据实际需求调整文本和图片的权重。

2.4 第四步：完整流水线整合

现在我们把所有组件整合起来，形成一个完整的工作流。

import json from datetime import datetime from pathlib import Path class ContentProcessingPipeline: def __init__(self, output_dir="./processed_data"): self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) # 初始化各个组件 self.crawler = ContentCrawler() self.cleaner = DataCleaner() self.reranker = LycheeReranker() # 创建图片存储目录 self.images_dir = self.output_dir / "images" self.images_dir.mkdir(exist_ok=True) def run(self, seed_urls: List[str], query: str, max_articles: int = 50): """运行完整的内容处理流水线""" print("开始爬取数据...") all_articles = [] # 1. 爬取数据 for url in seed_urls[:5]: # 限制种子URL数量 article = self.crawler.fetch_article(url) if article: # 下载图片到本地 local_images = [] for i, img_url in enumerate(article['images']): img_name = f"{hashlib.md5(img_url.encode()).hexdigest()[:8]}.jpg" img_path = self.images_dir / img_name if self.crawler.download_image(img_url, img_path): local_images.append(str(img_path)) article['local_images'] = local_images all_articles.append(article) if len(all_articles) >= max_articles: break print(f"爬取完成，共获取 {len(all_articles)} 篇文章") # 2. 数据清洗 print("开始数据清洗...") cleaned_articles = [] for article in all_articles: article['content'] = self.cleaner.clean_text(article['content']) article['title'] = self.cleaner.clean_text(article['title']) cleaned_articles.append(article) # 去重 unique_articles = self.cleaner.remove_duplicates(cleaned_articles) print(f"去重后剩余 {len(unique_articles)} 篇文章") # 3. 准备重排序数据 print("准备重排序数据...") candidates = [] for article in unique_articles: candidate = self.cleaner.prepare_for_rerank(article) candidates.append(candidate) # 4. 执行重排序 print(f"使用查询 '{query}' 进行重排序...") ranked_results = self.reranker.rerank(query, candidates, top_k=10) # 5. 保存结果 self.save_results(ranked_results, query) return ranked_results def save_results(self, results: List[Dict], query: str): """保存排序结果""" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"rerank_results_{query[:20]}_{timestamp}.json" filepath = self.output_dir / filename # 简化结果以便保存 simplified_results = [] for result in results: simplified_results.append({ 'id': result.get('id', ''), 'text_preview': result.get('text', '')[:200] + '...', 'relevance_score': round(result.get('relevance_score', 0), 4), 'modalities_used': result.get('modalities_used', 0), 'image_count': len(result.get('image_paths', [])), 'metadata': result.get('metadata', {}) }) with open(filepath, 'w', encoding='utf-8') as f: json.dump({ 'query': query, 'timestamp': datetime.now().isoformat(), 'total_results': len(results), 'results': simplified_results }, f, ensure_ascii=False, indent=2) print(f"结果已保存到: {filepath}") def print_results(self, results: List[Dict]): """打印排序结果""" print("\n" + "="*60) print("重排序结果:") print("="*60) for i, result in enumerate(results, 1): print(f"\n{i}. 分数: {result['relevance_score']:.4f}") print(f" 来源: {result['metadata'].get('source_url', 'N/A')}") print(f" 预览: {result['text'][:150]}...") print(f" 图片数: {len(result.get('image_paths', []))}") print(f" 使用模态: {result.get('modalities_used', 0)}") print("-"*40) # 使用示例 if __name__ == "__main__": # 配置参数 seed_urls = [ "https://example.com/article1", "https://example.com/article2", # 添加更多种子URL ] user_query = "Python数据分析教程" # 创建并运行流水线 pipeline = ContentProcessingPipeline() results = pipeline.run(seed_urls, user_query, max_articles=30) # 打印结果 pipeline.print_results(results)

这个完整的流水线从数据爬取开始，经过清洗、去重，最后使用Lychee Rerank进行智能排序，输出最相关的内容。