Python RAG 噪声注入

作者：追风剑情发布于：2026-4-28 15:57 分类：AI

噪声注入是 RAG 系统中的一种鲁棒性增强策略。它通过在检索到的上下文或用户查询中人为添加各种干扰信息，来训练或测试模型在真实混乱环境下的表现。

核心目的是让模型学会：

过滤无关信息
识别并忽略虚假内容
从杂乱文本中提取关键信息
容错处理拼写错误

示例

import random
import re
import jieba
from langchain_ollama import ChatOllama
from langchain_core.messages import HumanMessage

# ========== 1. 初始化本地 Ollama DeepSeek 模型 ==========
llm = ChatOllama(base_url="http://localhost:11434", model="deepseek-r1:7b")


# ========== 2. 定义各类噪声注入函数 ==========

def add_typo_noise(text, noise_rate=0.1):
    """
    拼写噪声：随机替换、删除或交换字符
    noise_rate: 噪声比例（针对每个字符的概率）
    """
    if not text:
        return text
    chars = list(text)
    for i in range(len(chars)):
        if random.random() < noise_rate:
            noise_type = random.choice(['replace', 'delete', 'swap'])
            if noise_type == 'replace' and chars[i].isalpha():
                # 替换为相邻拼音或随机相近字符（模拟错别字）
                common_errors = {
                    '机': '积', '器': '气', '学': '穴', '习': '席',
                    '人': '入', '工': '功', '智': '志', '能': '农',
                    '深': '身', '度': '渡', '神': '申', '经': '径',
                    '网': '往', '络': '洛'
                }
                if chars[i] in common_errors:
                    chars[i] = common_errors[chars[i]]
            elif noise_type == 'delete':
                chars[i] = ''
            elif noise_type == 'swap' and i < len(chars) - 1:
                chars[i], chars[i+1] = chars[i+1], chars[i]
    return ''.join(chars)


def add_irrelevant_sentence_noise(original_text):
    """
    支持性噪声/语义噪声：插入不相关的句子
    """
    irrelevant_sentences = [
        "今天天气真好啊。",
        "我昨天吃了一碗面条。",
        "Python是一种编程语言。",
        "这个例子展示了噪声注入技术。",
        "明天的会议取消了。"
    ]
    noise_sentence = random.choice(irrelevant_sentences)
    # 随机位置插入
    insert_pos = random.randint(0, len(original_text))
    return original_text[:insert_pos] + noise_sentence + original_text[insert_pos:]


def add_special_char_noise(text):
    """
    数据类型噪声：插入特殊字符、数字、URL等
    """
    special_noises = [
        " <!----> ",
        " &nbsp; ",
        " 123456 ",
        " https://example.com ",
        " 【广告】 ",
        " *** ",
        " {}[]() "
    ]
    noise = random.choice(special_noises)
    insert_pos = random.randint(0, len(text))
    return text[:insert_pos] + noise + text[insert_pos:]


def add_incomplete_sentence_noise(original_text):
    """
    非法句子噪声：插入不完整的片段
    """
    incomplete_fragments = [
        "因为所以而且",
        "虽然但是然而",
        "如果那么否则",
        "一方面另一方面",
        "首先然后最后"
    ]
    fragment = random.choice(incomplete_fragments)
    insert_pos = random.randint(0, len(original_text))
    return original_text[:insert_pos] + fragment + original_text[insert_pos:]


def add_counterfact_noise(original_text):
    """
    反事实噪声：插入明显错误的事实
    """
    counterfact = random.choice([
        "（注意：机器学习不需要任何数据就能学习。）",
        "（补充：深度学习的发明时间是1960年。）",
        "（提示：神经网络只有一层。）",
        "（科普：人工智能比人类笨得多。）"
    ])
    insert_pos = random.randint(0, len(original_text))
    return original_text[:insert_pos] + counterfact + original_text[insert_pos:]


def add_all_noises(text):
    """
    组合多种噪声：随机应用1-3种噪声
    """
    noise_functions = [
        lambda t: add_typo_noise(t, noise_rate=0.08),
        add_irrelevant_sentence_noise,
        add_special_char_noise,
        add_incomplete_sentence_noise,
        add_counterfact_noise
    ]
    # 随机选择1-3种噪声
    num_noises = random.randint(1, 3)
    selected = random.sample(noise_functions, num_noises)
    noisy_text = text
    for noise_func in selected:
        noisy_text = noise_func(noisy_text)
    return noisy_text


# ========== 3. 原始中文测试文本 ==========
original_text = """
机器学习是人工智能的一个分支，它使计算机能够从数据中学习和改进。
深度学习是机器学习的一个子领域，使用多层神经网络来处理复杂任务。
"""

# 清理文本（去除多余换行和空格）
original_text = ' '.join(original_text.strip().split())
print("=" * 60)
print("【原始文本】")
print(original_text)
print("=" * 60)

# ========== 4. 生成噪声文本 ==========
noisy_text = add_all_noises(original_text)
print("\n【噪声注入后的文本】")
print(noisy_text)
print("=" * 60)


# ========== 5. 构建增强提示词 ==========
prompt = f"""你是一个智能问答助手。下面是一段包含噪声的文本（可能有拼写错误、不相关句子、特殊字符或错误信息）。
请完成以下任务：
1. 识别并忽略噪声内容
2. 提取出真正有价值的信息
3. 基于有效信息，对"机器学习"和"深度学习"进行清晰、准确的解释

注意：只输出清理后的正确答案，不要输出你的分析过程。

带噪声的文本：
{noisy_text}

请输出清理后的正确解释："""

# ========== 6. 调用本地 LLM 生成增强文本 ==========
print("\n【正在调用本地 DeepSeek 处理噪声文本...】")
try:
    response = llm.invoke([HumanMessage(content=prompt)])
    print("\n【模型输出（清理后）】")
    print(response.content)
    print("=" * 60)
except Exception as e:
    print(f"LLM 调用失败: {e}")

# ========== 7. 可选：对比实验（使用原始文本调用模型） ==========
print("\n【对比实验：直接使用原始文本（无噪声）】")
prompt_clean = f"请解释什么是机器学习和深度学习：\n{original_text}"
try:
    response_clean = llm.invoke([HumanMessage(content=prompt_clean)])
    print(response_clean.content)
    print("=" * 60)
except Exception as e:
    print(f"LLM 调用失败: {e}")

运行测试