SiameseUIE代码实例：批量处理千条文本并统计人物地点共现频率

1. 项目概述

在日常的文本分析工作中，我们经常需要从大量文本中提取关键信息，特别是人物和地点的关联关系。传统的手工处理方式效率低下，而通用的信息抽取工具往往会产生冗余结果，需要二次清洗。

SiameseUIE模型正好解决了这个问题。这是一个专门针对中文文本优化的信息抽取模型，能够精准识别文本中的人物和地点实体，并且不会产生冗余信息。经过我们的镜像部署优化，现在你可以直接在云端环境中使用这个强大的工具。

本文将带你一步步实现一个实用场景：批量处理上千条文本数据，自动提取其中的人物和地点信息，并统计人物与地点的共现频率。这种分析可以帮助我们发现文本中隐藏的人物关系网络和地理分布模式。

2. 环境准备与快速部署

2.1 环境要求

使用我们预配置的镜像环境，你只需要：

• 一个系统盘≤50G的云实例
• 无需安装任何额外依赖包
• PyTorch版本保持为torch28（镜像已预装）
• 重启实例不会重置环境配置

2.2 快速启动步骤

登录你的云实例后，只需执行以下命令：

# 激活预配置环境
source activate torch28

# 进入模型工作目录
cd /nlp_structbert_siamese-uie_chinese-base

# 运行测试脚本验证环境
python test.py

如果看到模型加载成功信息和5个测试例子的抽取结果，说明环境配置正确。

3. 批量处理实现方案

3.1 数据处理架构设计

为了高效处理千条文本，我们设计了一个分批次处理的流水线：

import os
import json
from collections import defaultdict

class BatchProcessor:
    def __init__(self, batch_size=50):
        self.batch_size = batch_size
        self.entity_extractor = None
        self.co_occurrence = defaultdict(lambda: defaultdict(int))
    
    def load_model(self):
        """加载SiameseUIE模型"""
        # 模型加载代码将在下一节详细说明
        pass
    
    def process_batch(self, texts):
        """批量处理文本"""
        results = []
        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i + self.batch_size]
            batch_results = self._process_single_batch(batch)
            results.extend(batch_results)
        return results

3.2 核心抽取函数实现

基于test.py中的抽取逻辑，我们封装了批量处理函数：

def extract_entities_batch(texts, custom_entities=None):
    """
    批量提取文本中的人物和地点实体
    :param texts: 文本列表
    :param custom_entities: 自定义实体词典
    :return: 抽取结果列表
    """
    all_results = []
    
    for text in texts:
        try:
            # 调用SiameseUIE抽取函数
            result = extract_pure_entities(
                text=text,
                schema={"人物": None, "地点": None},
                custom_entities=custom_entities
            )
            all_results.append({
                "text": text,
                "persons": result.get("人物", []),
                "locations": result.get("地点", [])
            })
        except Exception as e:
            print(f"处理文本时出错: {str(e)}")
            all_results.append({"text": text, "persons": [], "locations": [], "error": str(e)})
    
    return all_results

4. 千条文本处理实战

4.1 数据准备与加载

假设我们有一个包含1000条文本的JSON文件，格式如下：

# 示例数据格式
[
    {
        "id": 1,
        "content": "李白在长安写下大量诗篇，杜甫则在成都创作了许多著名作品。"
    },
    {
        "id": 2, 
        "content": "马云在杭州创立了阿里巴巴，马化腾在深圳创办了腾讯。"
    }
    # ... 更多数据
]

加载数据的代码：

def load_texts_from_file(file_path, max_texts=1000):
    """从文件加载文本数据"""
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    
    texts = []
    for item in data[:max_texts]:
        texts.append(item['content'])
    
    return texts

# 加载数据
texts = load_texts_from_file('thousand_texts.json')
print(f"成功加载 {len(texts)} 条文本")

4.2 批量处理执行

现在执行批量处理：

# 初始化处理器
processor = BatchProcessor(batch_size=50)
processor.load_model()

# 处理所有文本
print("开始处理文本...")
results = processor.process_batch(texts)
print(f"处理完成，共处理 {len(results)} 条文本")

# 保存结果
with open('extraction_results.json', 'w', encoding='utf-8') as f:
    json.dump(results, f, ensure_ascii=False, indent=2)

5. 共现频率统计与分析

5.1 统计人物-地点共现关系

def calculate_co_occurrence(results):
    """统计人物和地点的共现频率"""
    co_occurrence = defaultdict(lambda: defaultdict(int))
    
    for result in results:
        persons = result['persons']
        locations = result['locations']
        
        # 统计共现关系
        for person in persons:
            for location in locations:
                co_occurrence[person][location] += 1
    
    return co_occurrence

# 执行统计
co_occurrence_data = calculate_co_occurrence(results)

5.2 生成统计报告

def generate_statistics_report(co_occurrence_data, top_n=20):
    """生成详细的统计报告"""
    report = {
        "summary": {
            "total_persons": len(co_occurrence_data),
            "total_relationships": sum(
                len(locations) for locations in co_occurrence_data.values()
            )
        },
        "top_relationships": [],
        "person_network": [],
        "location_network": []
    }
    
    # 提取前N个最频繁的共现关系
    all_relationships = []
    for person, locations in co_occurrence_data.items():
        for location, count in locations.items():
            all_relationships.append((person, location, count))
    
    # 按频率排序
    all_relationships.sort(key=lambda x: x[2], reverse=True)
    report["top_relationships"] = all_relationships[:top_n]
    
    return report

# 生成报告
stat_report = generate_statistics_report(co_occurrence_data)
print("统计报告生成完成")

5.3 可视化输出结果

将统计结果保存为易于阅读的格式：

def save_human_readable_report(report, output_path):
    """保存可读性强的统计报告"""
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write("人物-地点共现频率统计报告\n")
        f.write("=" * 50 + "\n\n")
        
        f.write(f"总统计概览：\n")
        f.write(f"- 出现人物总数: {report['summary']['total_persons']}\n")
        f.write(f"- 人物地点关系总数: {report['summary']['total_relationships']}\n\n")
        
        f.write("前20个最频繁的人物-地点关系：\n")
        f.write("-" * 50 + "\n")
        for i, (person, location, count) in enumerate(report['top_relationships'], 1):
            f.write(f"{i}. {person} - {location}: {count}次\n")
        
        f.write("\n详细数据已保存为JSON格式，可用于进一步分析")

# 保存报告
save_human_readable_report(stat_report, 'co_occurrence_report.txt')

6. 性能优化与实用技巧

6.1 处理速度优化

对于千条文本的处理，我们提供了多种优化方案：

# 优化建议1：调整批次大小
optimal_batch_size = 30  # 根据实例内存调整

# 优化建议2：启用缓存机制
def process_with_cache(texts, cache_file='processed_cache.json'):
    """使用缓存避免重复处理"""
    if os.path.exists(cache_file):
        with open(cache_file, 'r', encoding='utf-8') as f:
            return json.load(f)
    
    results = processor.process_batch(texts)
    
    # 保存缓存
    with open(cache_file, 'w', encoding='utf-8') as f:
        json.dump(results, f, ensure_ascii=False)
    
    return results

6.2 内存管理策略

在处理大量文本时，内存管理很重要：

# 内存友好的处理方式
def process_large_dataset(texts, output_file, batch_size=30):
    """分批处理并实时保存结果，减少内存占用"""
    all_results = []
    
    for i in range(0, len(texts), batch_size):
        batch_texts = texts[i:i + batch_size]
        batch_results = processor.process_batch(batch_texts)
        all_results.extend(batch_results)
        
        # 每处理10个批次保存一次
        if (i // batch_size) % 10 == 0:
            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(all_results, f, ensure_ascii=False)
            print(f"已处理 {i + batch_size} 条文本，结果已保存")
    
    return all_results

7. 实际应用案例

7.1 新闻数据分析

假设我们有一个新闻数据集，想要分析新闻中经常提到的人物和地点：

# 分析新闻数据中的关键人物和地点
news_texts = load_news_data()  # 加载新闻数据
results = process_with_cache(news_texts, 'news_cache.json')

# 生成热点人物地点报告
co_occurrence = calculate_co_occurrence(results)
report = generate_statistics_report(co_occurrence)

print("新闻数据分析完成！")
print("热点人物-地点关系：")
for person, location, count in report['top_relationships'][:10]:
    print(f"  {person} 在 {location} 被提及 {count} 次")

7.2 社交媒体内容分析

对于社交媒体文本，我们可以分析用户讨论的热点：

# 社交媒体内容分析
social_media_texts = load_social_media_posts()
results = extract_entities_batch(social_media_texts)

# 统计热门话题
trending_relations = []
for result in results:
    if result['persons'] and result['locations']:
        trending_relations.append({
            'content': result['text'][:100] + '...',  # 预览
            'persons': result['persons'],
            'locations': result['locations']
        })

# 保存热点话题
with open('trending_topics.json', 'w', encoding='utf-8') as f:
    json.dump(trending_relations, f, ensure_ascii=False, indent=2)

8. 总结

通过本文的实践，我们成功实现了使用SiameseUIE模型批量处理千条文本并统计人物地点共现频率的完整流程。这个方案有以下几个显著优势：

核心价值：

• 高效准确：利用SiameseUIE模型的精准抽取能力，避免了传统方法的冗余和错误
• 易于部署：基于预配置的镜像环境，无需复杂的环境配置
• 灵活扩展：代码结构清晰，可以轻松适配不同的数据源和分析需求
• 实用性强：提供的统计分析和可视化结果可以直接用于业务决策

适用场景：

• 媒体行业的热点人物和地点分析
• 社交媒体的话题趋势挖掘
• 历史文献的人物关系网络研究
• 商业情报的竞争对手地理分布分析

下一步建议：

1. 尝试处理更大规模的数据集，测试系统的稳定性
2. 扩展实体类型，加入时间、组织等更多维度的分析
3. 开发实时处理管道，用于流式数据的实时分析
4. 结合可视化工具，创建交互式的人物地点关系图谱

通过这种基于SiameseUIE的批量处理方法，你可以从海量文本中快速提取有价值的信息，发现数据背后隐藏的模式和关系，为各种应用场景提供数据支持。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。