构建AI智能体：十三、大数据下的“搭积木”：N-Gram 如何实现更智能的语义搜索

import pandas as pd
from sklearn.metrics.pairwise import linear_kernel
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.feature_extraction.text import TfidfVectorizer
import re
pd.options.display.max_columns = 30
import matplotlib.pyplot as plt
# 支持中文
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
df = pd.read_csv('Seattle_Hotels.csv', encoding="latin-1")
# 数据探索
print(df.head())
print('数据集中的酒店个数：', len(df))
# 创建英文停用词列表
ENGLISH_STOPWORDS = {
    'i', 'me', 'my', 'myself', 'we', 'our', 'ours', 'ourselves', 'you', "you're", "you've", "you'll", "you'd", 'your', 
    'yours', 'yourself', 'yourselves', 'he', 'him', 'his', 'himself', 'she', "she's", 'her', 'hers', 'herself', 'it', 
    "it's", 'its', 'itself', 'they', 'them', 'their', 'theirs', 'themselves', 'what', 'which', 'who', 'whom', 'this', 
    'that', "that'll", 'these', 'those', 'am', 'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 
    'having', 'do', 'does', 'did', 'doing', 'a', 'an', 'the', 'and', 'but', 'if', 'or', 'because', 'as', 'until', 'while', 
    'of', 'at', 'by', 'for', 'with', 'about', 'against', 'between', 'into', 'through', 'during', 'before', 'after', 'above', 
    'below', 'to', 'from', 'up', 'down', 'in', 'out', 'on', 'off', 'over', 'under', 'again', 'further', 'then', 'once', 
    'here', 'there', 'when', 'where', 'why', 'how', 'all', 'any', 'both', 'each', 'few', 'more', 'most', 'other', 'some', 
    'such', 'no', 'nor', 'not', 'only', 'own', 'same', 'so', 'than', 'too', 'very', 's', 't', 'can', 'will', 'just', 'don', 
    "don't", 'should', "should've", 'now', 'd', 'll', 'm', 'o', 're', 've', 'y', 'ain', 'aren', "aren't", 'couldn', 
    "couldn't", 'didn', "didn't", 'doesn', "doesn't", 'hadn', "hadn't", 'hasn', "hasn't", 'haven', "haven't", 'isn', 
    "isn't", 'ma', 'mightn', "mightn't", 'mustn', "mustn't", 'needn', "needn't", 'shan', "shan't", 'shouldn', "shouldn't", 
    'wasn', "wasn't", 'weren', "weren't", 'won', "won't", 'wouldn', "wouldn't"
}
def print_description(index):
    example = df[df.index == index][['desc', 'name']].values[0]
    if len(example) > 0:
        print(example[0])
        print('Name:', example[1])
print('第10个酒店的描述：')
print_description(10)
# 得到酒店描述中n-gram特征中的TopK个
def get_top_n_words(corpus, n=1, k=None):
    # 统计ngram词频矩阵，使用自定义停用词列表
    vec = CountVectorizer(ngram_range=(n, n), stop_words=list(ENGLISH_STOPWORDS)).fit(corpus)
    bag_of_words = vec.transform(corpus)
    """
    print('feature names:')
    print(vec.get_feature_names())
    print('bag of words:')
    print(bag_of_words.toarray())
    """
    sum_words = bag_of_words.sum(axis=0)
    words_freq = [(word, sum_words[0, idx]) for word, idx in vec.vocabulary_.items()]
    # 按照词频从大到小排序
    words_freq =sorted(words_freq, key = lambda x: x[1], reverse=True)
    return words_freq[:k]
common_words = get_top_n_words(df['desc'], n=2, k=20)
#print(common_words)
df1 = pd.DataFrame(common_words, columns = ['desc' , 'count'])
df1.groupby('desc').sum()['count'].sort_values().plot(kind='barh', title='去掉停用词后，酒店描述中的Top20单词')
plt.show()
# 文本预处理
REPLACE_BY_SPACE_RE = re.compile('[/(){}\[\]\|@,;]')
BAD_SYMBOLS_RE = re.compile('[^0-9a-z #+_]')
# 使用自定义的英文停用词列表替代nltk的stopwords
STOPWORDS = ENGLISH_STOPWORDS
# 对文本进行清洗
def clean_text(text):
    # 全部小写
    text = text.lower()
    # 用空格替代一些特殊符号，如标点
    text = REPLACE_BY_SPACE_RE.sub(' ', text)
    # 移除BAD_SYMBOLS_RE
    text = BAD_SYMBOLS_RE.sub('', text)
    # 从文本中去掉停用词
    text = ' '.join(word for word in text.split() if word not in STOPWORDS) 
    return text
# 对desc字段进行清理，apply针对某列
df['desc_clean'] = df['desc'].apply(clean_text)
#print(df['desc_clean'])
# 建模
df.set_index('name', inplace = True)
# 使用TF-IDF提取文本特征，使用自定义停用词列表
tf = TfidfVectorizer(analyzer='word', ngram_range=(1, 3), min_df=0.01, stop_words=list(ENGLISH_STOPWORDS))
# 针对desc_clean提取tfidf
tfidf_matrix = tf.fit_transform(df['desc_clean'])
print('TFIDF feature names:')
#print(tf.get_feature_names_out())
print(len(tf.get_feature_names_out()))
#print('tfidf_matrix:')
print(tfidf_matrix.shape)
# 计算酒店之间的余弦相似度（线性核函数）
cosine_similarities = linear_kernel(tfidf_matrix, tfidf_matrix)
#print(cosine_similarities)
print(cosine_similarities.shape)
indices = pd.Series(df.index) #df.index是酒店名称
# 基于相似度矩阵和指定的酒店name，推荐TOP10酒店
def recommendations(name, cosine_similarities = cosine_similarities):
    recommended_hotels = []
    # 找到想要查询酒店名称的idx
    idx = indices[indices == name].index[0]
    print('idx=', idx)
    # 对于idx酒店的余弦相似度向量按照从大到小进行排序
    score_series = pd.Series(cosine_similarities[idx]).sort_values(ascending = False)
    # 取相似度最大的前10个（除了自己以外）
    top_10_indexes = list(score_series.iloc[1:11].index)
    # 放到推荐列表中
    for i in top_10_indexes:
        recommended_hotels.append(list(df.index)[i])
    return recommended_hotels
print(recommendations('Hilton Seattle Airport & Conference Center'))
print(recommendations('The Bacon Mansion Bed and Breakfast'))
#print(result)

from collections import defaultdict, Counter
import numpy as np
# 简单语料库
corpus = [
    ['今天', '天气', '真好'],
    ['今天', '心情', '真好'], 
    ['明天', '天气', '不错']
]
# 构建Bigram模型
def build_bigram_model(corpus):
    bigram_counts = defaultdict(Counter)
    unigram_counts = Counter()
    for sentence in corpus:
        for i in range(len(sentence)-1):
            current_word = sentence[i]
            next_word = sentence[i+1]
            bigram_counts[current_word][next_word] += 1
            unigram_counts[current_word] += 1
    # 计算概率
    bigram_probs = {}
    for prev_word, next_words in bigram_counts.items():
        total = unigram_counts[prev_word]
        bigram_probs[prev_word] = {
            next_word: count/total for next_word, count in next_words.items()
        }
    return bigram_probs
# 构建模型
model = build_bigram_model(corpus)
print("P(天气|今天) =", model['今天']['天气'])
print("P(心情|今天) =", model['今天']['心情'])

def generate_text(seed_word, model, length=10):
    current_word = seed_word
    generated_text = [current_word]
    for _ in range(length-1):
        if current_word not in model:
            break
        next_words = list(model[current_word].keys())
        probabilities = list(model[current_word].values())
        # 按概率选择下一个词
        next_word = np.random.choice(next_words, p=probabilities)
        generated_text.append(next_word)
        current_word = next_word
    return ''.join(generated_text)
# 生成文本
print("生成示例:", generate_text('今天', model, 5))

import numpy as np
import matplotlib.pyplot as plt
from scipy.optimize import minimize_scalar
# 设置matplotlib支持中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 用来正常显示中文标签
plt.rcParams['axes.unicode_minus'] = False  # 用来正常显示负号
# 观测数据：7次正面，3次反面
data = np.array([1, 1, 1, 1, 1, 1, 1, 0, 0, 0])
n_heads = np.sum(data)
n_tails = len(data) - n_heads
# 定义似然函数
def likelihood(p):
    return p**n_heads * (1-p)**n_tails
# 定义负对数似然函数（因为我们要最小化）
def neg_log_likelihood(p):
    return - (n_heads * np.log(p) + n_tails * np.log(1-p))
# 使用优化方法找到最大似然估计
result = minimize_scalar(neg_log_likelihood, bounds=(0.01, 0.99), method='bounded')
mle_p = result.x
print(f"最大似然估计值: p = {mle_p:.3f}")
# 可视化似然函数
p_values = np.linspace(0.01, 0.99, 100)
likelihood_values = [likelihood(p) for p in p_values]
plt.figure(figsize=(10, 6))
plt.plot(p_values, likelihood_values, label='似然函数')
plt.axvline(mle_p, color='r', linestyle='--', label=f'MLE估计值: p={mle_p:.3f}')
plt.xlabel('p (正面概率)')
plt.ylabel('似然值')
plt.title('硬币投掷问题的似然函数')
plt.legend()
plt.grid(True)
plt.show()