如何做四六级考试的段落匹配题¶

# 需要的库： pip install requests numpy scikit-learn nltk
# 在 passage.txt 内填入阅读原文，以 \n\n 分段
# 在 sentences.txt 内填入待处理的句子，用 \n\n 分开
# 然后运行！
import os
import requests
import json
import numpy as np
import nltk
from concurrent.futures import ThreadPoolExecutor
from sklearn.metrics.pairwise import cosine_similarity

# ================= 配置区域 =================
# 请在此处填入你的 API Key
API_KEY = "sk-*******"

# 请在此处填入模型名称 (例如: BAAI/bge-m3, BAAI/bge-large-en-v1.5 等)
MODEL_NAME = "Qwen/Qwen3-Embedding-8B"

# 并发数设置
CONCURRENCY = 50

# 兼容 OpenAI 格式 API 地址
API_URL = "https://api.siliconflow.cn/v1/embeddings"
# ===========================================

def download_nltk_data():
    """下载必要的分句数据"""
    try:
        nltk.data.find('tokenizers/punkt')
    except LookupError:
        print("正在下载 NLTK punkt 分词数据...")
        nltk.download('punkt')
        nltk.download('punkt_tab')

def get_embedding(text):
    """
    调用 SiliconFlow API 获取文本嵌入向量
    """
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    # 移除换行符，防止影响 Embedding 质量
    clean_text = text.replace("\n", " ")

    payload = {
        "model": MODEL_NAME,
        "input": clean_text
    }

    try:
        response = requests.post(API_URL, json=payload, headers=headers, timeout=30)
        response.raise_for_status()
        data = response.json()
        return data['data'][0]['embedding']
    except Exception as e:
        print(f"Error getting embedding for text: {clean_text[:30]}... : {e}")
        # 如果失败返回全0向量防止程序崩溃（实际生产中应重试）
        return None

def read_files():
    """读取文件内容"""
    try:
        with open('passage.txt', 'r', encoding='utf-8') as f:
            passage_text = f.read().strip()

        with open('sentences.txt', 'r', encoding='utf-8') as f:
            sentences_text = f.read().strip()

        return passage_text, sentences_text
    except FileNotFoundError as e:
        print(f"错误: 找不到文件 - {e}")
        exit(1)

def main():
    # 0. 准备工作
    print("Preparing...")
    download_nltk_data()
    if API_KEY == "YOUR_SILICONFLOW_API_KEY_HERE":
        print("请在代码中设置正确的 API_KEY")
        return

    # 1. 读取文件
    print("1. 读取文件...")
    passage_raw, sentences_raw = read_files()

    # 处理 passage.txt: 按 \n\n 分割段落
    paragraphs = [p.strip() for p in passage_raw.split('\n\n') if p.strip()]

    # 处理 sentences.txt: 按 \n\n 分割句子 (集合 T)
    t_sentences = [s.strip() for s in sentences_raw.split('\n\n') if s.strip()]

    # 2. 将段落分句得到集合 S，并记录所属段落索引
    print("2. 预处理段落与分句...")
    s_data = [] # 存储格式: {'text': sentence, 'para_idx': int}

    for idx, para in enumerate(paragraphs):
        # 使用 nltk 进行英文分句
        sents = nltk.sent_tokenize(para)
        for s in sents:
            s_data.append({'text': s, 'para_idx': idx})

    # 3. 准备所有需要 Embedding 的文本列表
    #我们需要 Embed: T 中的句子, S 中的句子, 以及原段落(用于步骤5)
    all_texts_to_embed = []

    # 映射索引范围
    t_range = range(0, len(t_sentences))
    start_s = len(t_sentences)
    s_range = range(start_s, start_s + len(s_data))
    start_p = start_s + len(s_data)
    p_range = range(start_p, start_p + len(paragraphs))

    all_texts_to_embed.extend(t_sentences)
    all_texts_to_embed.extend([item['text'] for item in s_data])
    all_texts_to_embed.extend(paragraphs)

    print(f"3. 开始并发获取 Embeddings (总数: {len(all_texts_to_embed)}, 并发数: {CONCURRENCY})...")

    embeddings_map = {}

    # 使用 ThreadPoolExecutor 并发调用接口
    with ThreadPoolExecutor(max_workers=CONCURRENCY) as executor:
        # 提交所有任务
        futures = {executor.submit(get_embedding, text): i for i, text in enumerate(all_texts_to_embed)}

        # 收集结果（保持顺序）
        results = [None] * len(all_texts_to_embed)
        for future in futures:
            idx = futures[future]
            res = future.result()
            if res is None:
                print("获取 Embedding 失败，程序终止。")
                return
            results[idx] = res

    # 分离 Embedding 向量
    t_embeddings = np.array(results[:len(t_sentences)])
    s_embeddings = np.array(results[start_s:start_s + len(s_data)])
    p_embeddings = np.array(results[start_p:])

    print("4 & 5. 计算相似度矩阵...")

    # 计算 T vs S 的相似度矩阵 (Shape: len(T) x len(S))
    sim_t_s = cosine_similarity(t_embeddings, s_embeddings)

    # 计算 T vs Paragraphs 的相似度矩阵 (Shape: len(T) x len(Paragraphs))
    sim_t_p = cosine_similarity(t_embeddings, p_embeddings)

    print("6. 综合评分并匹配段落...")

    final_results = []
    num_paras = len(paragraphs)

    for i in range(len(t_sentences)):
        # --- 步骤 4 逻辑 ---
        # 找出当前 T 句子与 S 集合中相似度最高的那个句子，并获取其所属段落
        # 这里我们不只取最大值，而是计算“基于句子的段落得分”
        # 策略：对于每个段落，找出该段落中与当前 T 句子相似度最高的那个句子的得分

        # 初始化该 T 句子针对每个段落的句子匹配最高分
        best_sentence_score_per_para = np.zeros(num_paras)

        # 遍历所有 S 句子及其相似度
        for j, score in enumerate(sim_t_s[i]):
            p_idx = s_data[j]['para_idx']
            if score > best_sentence_score_per_para[p_idx]:
                best_sentence_score_per_para[p_idx] = score

        # --- 步骤 5 逻辑 ---
        # 当前 T 句子与段落集合的直接相似度
        direct_para_scores = sim_t_p[i]

        # --- 步骤 6 综合逻辑 ---
        # 综合评分策略：直接相加 (或者加权相加)
        # Final Score(Para k) = Sim(T, Para k) + Max(Sim(T, Sentence in Para k))
        # 这种方法既考虑了宏观语义(段落级)，也考虑了微观细节(句子级)

        combined_scores = direct_para_scores + best_sentence_score_per_para

        # 找到综合得分最高的段落索引
        best_para_idx = np.argmax(combined_scores)

        # 转换为字母 (0->A, 1->B)
        para_letter = chr(ord('A') + best_para_idx)
        final_results.append(para_letter)

    # 7. 输出结果
    print("\n---------- 匹配结果 ----------")
    for res in final_results:
        print(res)
    print("-----------------------------")

if __name__ == "__main__":
    main()

做段落匹配题目，文章：{passage.txt} 句子：{sentence.txt} 给出答案。不要用数字进行段落编号，用英文字母，如A=1