大模型应用：TextRank+混元大模型：轻量化算法与大模型协同的文本摘要实践.99

import re

def clean_text(text):
    """
    文本清洗：去除特殊符号、多余空格、HTML标签
    """
    # 去除HTML标签
    text = re.sub(r'<[^>]+>', '', text)
    # 去除特殊符号（保留中文、英文、数字、常见标点）
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9，。！？；：""''()（）、]', '', text)
    # 去除多余空格和换行
    text = re.sub(r'\s+', ' ', text).strip()
    return text

# 测试
raw_text = "<p> 这是一篇测试文本★！包含HTML标签、特殊符号，还有多余的空格  </p>"
cleaned_text = clean_text(raw_text)
print(cleaned_text)  

import re

def split_sentences(text):
    """
    拆分句子：按中文标点拆分，返回句子列表
    """
    # 按。！？；拆分，保留标点
    sentences = re.split(r'(。|！|？|；)', text)
    # 重组句子（避免标点单独成句）
    sentences = [s1 + s2 for s1, s2 in zip(sentences[0::2], sentences[1::2])]
    # 过滤空句子
    sentences = [s.strip() for s in sentences if s.strip()]
    return sentences

# 测试
text = "这是第一句话。这是第二句话！这是第三句话？"
sentences = split_sentences(text)
print(sentences)  

import jieba

# 通用停用词列表（简化版，可扩展）
STOP_WORDS = {'的', '了', '是', '我', '你', '他', '她', '它', '们', '在', '有', '就', '都', '而', '及', '与', '着', '过', '也', '还', '吧', '吗', '呢', '啊', '哦', '哈', '哎', '嗨', '和', '或', '但', '却', '能', '能够', '可以', '将', '要', '会', '一个', '一些', '这种', '这个', '那个', '等', '使', '被', '由', '对', '为', '于', '以'}

# 自定义关键词词典（强制保留的词汇）
CUSTOM_KEYWORDS = {'人工智能', '机器学习', '深度学习', '自然语言处理', '计算机视觉', '神经网络', '算法', '数据', '模型'}

# 将自定义关键词加载到jieba词典中，确保正确分词
for word in CUSTOM_KEYWORDS:
    jieba.add_word(word, freq=1000)  # 设置较高频率确保优先匹配

def tokenize_text(text, custom_keywords=None):
    """
    分词并过滤停用词
    :param text: 待分词文本
    :param custom_keywords: 自定义关键词集合（强制保留）
    """
    # 如果没有传入自定义词典，使用全局默认词典
    if custom_keywords is None:
        custom_keywords = CUSTOM_KEYWORDS
    
    # 分词（精确模式）
    words = jieba.lcut(text)
    # 过滤停用词、空字符串和标点符号，但保留自定义关键词
    words = [word for word in words 
             if (word in custom_keywords) or 
                (word not in STOP_WORDS and word.strip() and word not in '，。！？；：""''()（）、')]
    return words

# 测试
text = "我喜欢吃苹果，苹果是一种很甜的水果"
words = tokenize_text(text)
print(words) 

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

class TextRankSentence:
    def __init__(self, sentences, custom_keywords=None, damping=0.85, max_iter=100, tol=1e-4):
        """
        句子级TextRank初始化
        :param sentences: 句子列表
        :param custom_keywords: 自定义关键词集合（用于增强句子权重）
        :param damping: 阻尼系数
        :param max_iter: 最大迭代次数
        :param tol: 收敛阈值
        """
        # 如果没有传入自定义词典，使用全局默认词典
        if custom_keywords is None:
            custom_keywords = CUSTOM_KEYWORDS
        self.custom_keywords = custom_keywords
        self.sentences = sentences
        self.damping = damping
        self.max_iter = max_iter
        self.tol = tol
        # 构建TF-IDF向量器（用于句子向量化）
        self.vectorizer = TfidfVectorizer(stop_words='english')  # 中文停用词可自定义
        # 句子相似度矩阵
        self.sim_matrix = self._build_sim_matrix()
        # 句子权重
        self.sentence_weights = self._calculate_weights()

    def _build_sim_matrix(self):
        """
        构建句子相似度矩阵
        """
        # 将句子列表转为TF-IDF向量
        tfidf_matrix = self.vectorizer.fit_transform(self.sentences)
        # 计算余弦相似度矩阵
        sim_matrix = cosine_similarity(tfidf_matrix)
        # 将相似度矩阵的对角线置0（句子不与自身相似）
        np.fill_diagonal(sim_matrix, 0)
        return sim_matrix

    def _calculate_weights(self):
        """
        迭代计算句子权重
        """
        n = len(self.sentences)
        # 初始化权重：所有句子权重为1
        weights = np.ones(n) / n
        # 迭代计算
        for _ in range(self.max_iter):
            new_weights = np.ones(n) * (1 - self.damping) / n
            for i in range(n):
                # 计算所有指向i的节点的加权权重
                sum_weight = 0
                for j in range(n):
                    if self.sim_matrix[j][i] > 0:
                        # 分母：j指向的所有节点的相似度之和
                        sum_sim_j = np.sum(self.sim_matrix[j])
                        if sum_sim_j == 0:
                            continue
                        sum_weight += self.sim_matrix[j][i] / sum_sim_j * weights[j]
                new_weights[i] += self.damping * sum_weight
            # 判断是否收敛
            if np.linalg.norm(new_weights - weights) < self.tol:
                break
            weights = new_weights
        return weights

    def get_top_sentences(self, top_k=5):
        """
        获取权重最高的top_k个句子
        :param top_k: 返回的句子数量
        :return: 核心句子列表（按原文顺序排列）
        """
        # 计算每个句子中自定义关键词的出现次数，作为额外的权重因子
        keyword_boost = []
        for sentence in self.sentences:
            # 统计句子中包含的自定义关键词数量
            count = sum(1 for kw in self.custom_keywords if kw in sentence)
            # 权重提升因子：每包含一个自定义关键词，权重提升20%
            boost = 1 + count * 0.2
            keyword_boost.append(boost)
        
        # 将TextRank权重与自定义关键词权重相乘
        boosted_weights = self.sentence_weights * np.array(keyword_boost)
        
        # 按提升后的权重排序，获取索引
        top_indices = np.argsort(boosted_weights)[-top_k:][::-1]
        # 返回核心句子（按原文顺序排列）
        top_sentences = [self.sentences[i] for i in sorted(top_indices)]
        return top_sentences

# 测试
if __name__ == "__main__":
    # 示例文本
    text = """
    人工智能是一门旨在使计算机系统能够模拟、延伸和扩展人类智能的技术科学。
    它涵盖了机器学习、深度学习、自然语言处理、计算机视觉等多个子领域。
    机器学习是人工智能的核心，它使计算机能够从数据中学习并改进性能，而无需显式编程。
    深度学习是机器学习的一个分支，基于神经网络，能够处理复杂的非线性关系。
    自然语言处理则专注于使计算机理解和生成人类语言，是人工智能落地的重要方向。
    随着算力的提升和数据量的增长，人工智能在医疗、金融、教育等领域的应用越来越广泛。
    未来，人工智能将继续推动产业升级，改变人们的生活方式，但也需要关注伦理和安全问题。
    """
    # 预处理
    cleaned_text = clean_text(text)
    sentences = split_sentences(cleaned_text)
    # 初始化TextRank
    textrank = TextRankSentence(sentences)
    # 获取核心句子（粗摘）
    top_sentences = textrank.get_top_sentences(top_k=3)
    print("TextRank粗摘核心句子：")
    for i, sent in enumerate(top_sentences, 1):
        print(f"{i}. {sent}")

import numpy as np
from collections import defaultdict

class TextRankKeyword:
    def __init__(self, words, custom_keywords=None, window_size=5, damping=0.85, max_iter=100, tol=1e-4):
        """
        词汇级TextRank初始化
        :param words: 分词后的词汇列表
        :param custom_keywords: 自定义关键词集合（优先返回）
        :param window_size: 共现窗口大小
        :param damping: 阻尼系数
        :param max_iter: 最大迭代次数
        :param tol: 收敛阈值
        """
        # 如果没有传入自定义词典，使用全局默认词典
        if custom_keywords is None:
            custom_keywords = CUSTOM_KEYWORDS
        self.custom_keywords = custom_keywords
        self.words = words
        self.window_size = window_size
        self.damping = damping
        self.max_iter = max_iter
        self.tol = tol
        # 构建词汇共现图
        self.graph = self._build_graph()
        # 词汇权重
        self.word_weights = self._calculate_weights()

    def _build_graph(self):
        """
        构建词汇共现图（邻接表形式）
        """
        graph = defaultdict(dict)
        # 遍历词汇，构建共现关系
        for i in range(len(self.words)):
            # 窗口范围：[i-window_size, i+window_size]
            start = max(0, i - self.window_size)
            end = min(len(self.words), i + self.window_size + 1)
            # 遍历窗口内的其他词汇
            for j in range(start, end):
                if i == j:
                    continue
                word_i = self.words[i]
                word_j = self.words[j]
                # 更新共现权重（共现一次，权重+1）
                if word_j in graph[word_i]:
                    graph[word_i][word_j] += 1
                else:
                    graph[word_i][word_j] = 1
        return graph

    def _calculate_weights(self):
        """
        迭代计算词汇权重
        """
        # 词汇列表（去重）
        unique_words = list(set(self.words))
        word2idx = {word: idx for idx, word in enumerate(unique_words)}
        n = len(unique_words)
        # 初始化权重：所有词汇权重为1
        weights = np.ones(n) / n
        # 构建邻接矩阵
        adj_matrix = np.zeros((n, n))
        for word_i in self.graph:
            for word_j in self.graph[word_i]:
                i = word2idx[word_i]
                j = word2idx[word_j]
                adj_matrix[i][j] = self.graph[word_i][word_j]

        # 迭代计算
        for _ in range(self.max_iter):
            new_weights = np.ones(n) * (1 - self.damping) / n
            for i in range(n):
                sum_weight = 0
                for j in range(n):
                    if adj_matrix[j][i] > 0:
                        # 分母：j指向的所有节点的权重之和
                        sum_adj_j = np.sum(adj_matrix[j])
                        if sum_adj_j == 0:
                            continue
                        sum_weight += adj_matrix[j][i] / sum_adj_j * weights[j]
                new_weights[i] += self.damping * sum_weight
            # 判断是否收敛
            if np.linalg.norm(new_weights - weights) < self.tol:
                break
            weights = new_weights

        # 转换为{词汇: 权重}的字典
        word_weights = {unique_words[i]: weights[i] for i in range(n)}
        return word_weights

    def get_top_keywords(self, top_k=10, only_custom=False):
        """
        获取权重最高的top_k个关键词
        :param top_k: 返回的关键词数量
        :param only_custom: 是否只返回自定义词典中的关键词
        """
        # 按权重排序
        sorted_keywords = sorted(self.word_weights.items(), key=lambda x: x[1], reverse=True)
        
        if only_custom:
            # 只返回自定义词典中的关键词
            custom_sorted = [(word, weight) for word, weight in sorted_keywords if word in self.custom_keywords]
            top_keywords = [word for word, weight in custom_sorted[:top_k]]
        else:
            # 返回所有关键词
            top_keywords = [word for word, weight in sorted_keywords[:top_k]]
        
        return top_keywords

# 测试
if __name__ == "__main__":
    # 示例文本
    text = """
    人工智能是一门旨在使计算机系统能够模拟、延伸和扩展人类智能的技术科学。
    它涵盖了机器学习、深度学习、自然语言处理、计算机视觉等多个子领域。
    机器学习是人工智能的核心，它使计算机能够从数据中学习并改进性能，而无需显式编程。
    深度学习是机器学习的一个分支，基于神经网络，能够处理复杂的非线性关系。
    自然语言处理则专注于使计算机理解和生成人类语言，是人工智能落地的重要方向。
    随着算力的提升和数据量的增长，人工智能在医疗、金融、教育等领域的应用越来越广泛。
    未来，人工智能将继续推动产业升级，改变人们的生活方式，但也需要关注伦理和安全问题。
    """
    # 预处理
    cleaned_text = clean_text(text)
    words = tokenize_text(cleaned_text)
    # 初始化TextRank
    textrank = TextRankKeyword(words)
    # 获取核心关键词（粗提）
    top_keywords = textrank.get_top_keywords(top_k=5)
    print("TextRank粗提核心关键词：")
    for i, keyword in enumerate(top_keywords, 1):
        print(f"{i}. {keyword}")

def format_rough_result(sentences, keywords):
    """
    格式化粗摘结果，用于构造大模型提示词
    """
    formatted = f"""
    以下是文本的核心句子（粗摘）：
    {chr(10).join([f"{i+1}. {sent}" for i, sent in enumerate(sentences)])}

    以下是文本的核心关键词（粗提）：
    {chr(10).join([f"{i+1}. {kw}" for i, kw in enumerate(keywords)])}
    """
    return formatted.strip()

# 测试
# top_sentences = [
#     "人工智能是一门旨在使计算机系统能够模拟、延伸和扩展人类智能的技术科学。",
#     "机器学习是人工智能的核心，它使计算机能够从数据中学习并改进性能，而无需显式编程。",
#     "随着算力的提升和数据量的增长，人工智能在医疗、金融、教育等领域的应用越来越广泛。"
# ]
# top_keywords = ["人工智能", "机器学习", "深度学习", "自然语言处理", "领域"]
formatted_text = format_rough_result(top_sentences, top_keywords)
print("格式化后的粗摘结果：")
print(formatted_text)

def build_summary_prompt(formatted_rough, style="学术", length="200字"):
    """
    构造长文摘要提示词
    :param formatted_rough: 格式化的粗摘结果
    :param style: 输出风格（学术/口语化/分点）
    :param length: 输出长度
    """
    prompt = f"""
    请基于以下文本的核心句子和关键词，生成一篇{length}左右的{style}风格摘要：
    {formatted_rough}

    要求：
    1. 严格保留核心信息，不遗漏关键内容；
    2. 语言流畅，逻辑连贯，符合{style}风格；
    3. 避免冗余，不添加无关信息；
    4. 若风格为分点，需分3-5点总结核心内容。
    """
    return prompt.strip()

# 测试
prompt = build_summary_prompt(formatted_text, style="学术", length="200字")
print("大模型摘要提示词：")
print(prompt)

def build_keyword_prompt(formatted_rough, need_explain=True):
    """
    构造关键词润色提示词
    :param formatted_rough: 格式化的粗摘结果
    :param need_explain: 是否需要为关键词添加简短解释
    """
    explain_text = "并为每个关键词添加1-2句话的简短解释（说明该关键词在文本中的含义）" if need_explain else ""
    prompt = f"""
    请基于以下文本的核心关键词，优化并整理关键词列表{explain_text}：
    {formatted_rough}

    要求：
    1. 保留核心关键词，可合并语义相近的词汇；
    2. 关键词列表控制在8-10个；
    3. 解释需简洁、准确，贴合文本语境。
    """
    return prompt.strip()

# 测试
keyword_prompt = build_keyword_prompt(formatted_text, need_explain=True)
print("大模型关键词提示词：")
print(keyword_prompt)

from openai import OpenAI

# 配置API Key（替换为你的混元API Key）
api_key = 'sk-bWl**********************vZ5NP8Ze'

client = OpenAI(
    api_key=api_key,
    base_url="https://api.hunyuan.cloud.tencent.com/v1",
)

def call_llm(prompt, model="hunyuan-lite", temperature=0.3):
    """
    调用混元大模型API
    :param prompt: 提示词
    :param model: 模型名称（hunyuan-lite/hunyuan-pro等）
    :param temperature: 生成温度（越低越精准）
    :return: 大模型输出结果
    """
    try:
        completion = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "你是一个专业的文本处理助手"},
                {"role": "user", "content": prompt}
            ],
            temperature=temperature,
            max_tokens=1000
        )
        # 提取输出内容
        result = completion.choices[0].message.content.strip()
        return result
    except Exception as e:
        print(f"调用混元大模型失败：{e}")
        return None

# 测试
if __name__ == "__main__":
    # 构造提示词
    prompt = build_summary_prompt(formatted_text, style="学术", length="200字")
    # 调用大模型
    summary_result = call_llm(prompt)
    if summary_result:
        print("大模型生成的精摘：")
        print(summary_result)

def post_process_result(result):
    """
    结果后处理：去除多余空格、换行，格式化输出
    """
    # 去除多余空格和换行
    result = re.sub(r'\s+', ' ', result).strip()
    # 按句子拆分，换行显示（提升可读性）
    sentences = split_sentences(result)
    final_result = chr(10).join(sentences)
    return final_result

# 测试
if summary_result:
    final_summary = post_process_result(summary_result)
    print("最终精摘结果：")
    print(final_summary)