噪声注入是 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 = [
" <!----> ",
" ",
" 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}")