自学内容网 自学内容网
自学内容网 > 正文

（RAG系列）联网搜索的探索及简单实现

🕗 发布于 2025-01-18 05:21 ai  python  自然语言处理  

（RAG系列）联网搜索的探索及简单实现

引言

建立完本地知识库，但是想通过联网搜索获取更多实时的信息，再交给大模型回答，优点是有实时性、丰富内容，缺点也很明显：延长系统回复时间、增加提示词的长度、筛选联网检索的内容的难度大。本文只是对联网搜索的探索及简单实现，想要具体落地还需要进一步优化或者找更优秀的第三方现成方案。

联网方案一：Google API

逻辑思路
  1. 调用谷歌的API*服务来获取数据，从中获取链接。
  2. 发起http请求链接，利用爬虫获取m个搜索结果的内容。
  3. 使用BM25算法进行相关度排序，然后从剩余内容中取出前n个文本，并且每个文本只取前k个字。
获取谷歌API密钥和自定义搜索引擎ID

自行搜索教程

代码实现

PS：把 hit_stopwords.txt （哈工大停用词）放到 google_search_api.py 同目录下，

google_search_api.py：

import requests
from flask import Flask, request, jsonify
import os
import requests
import re
from bs4 import BeautifulSoup
import jieba
from rank_bm25 import BM25Okapi


app = Flask(__name__)

# Google搜索函数
def google_search(query):
    # 你的API密钥和自定义搜索引擎ID
    api_key = ""
    cse_id = ""

    url = "https://www.googleapis.com/customsearch/v1"
    params = {
        "q": query,
        "key": api_key,
        "cx": cse_id,
        #"lr": "lang_zh-CN",
    }
    response = requests.get(url, params=params)

    return response.json()


def extract_content(query, num_links):
    #这里调用了上面的谷歌接口
    results = google_search(query)
    #print(results)
    #print("\n")
    texts = []

    # 提取指定数量的链接的信息
    for item in results["items"][:num_links]:
        url = item['link']
        print(url + "\n")
        try:
            timeout_seconds = 5
            # 使用requests库获取链接的内容，并设置超时时间
            response = requests.get(url, timeout=timeout_seconds)

            # 检查是否成功获取
            if response.status_code == 200:
                # 获取字符编码
                encoding = response.encoding if 'charset' in response.headers.get('content-type', '').lower() else None

                # 使用BeautifulSoup解析HTML
                soup = BeautifulSoup(response.content, 'html.parser', from_encoding=encoding)

                # 使用get_text()方法提取所有文本内容
                text_content = soup.get_text()
                # 对文本进行加工处理
                cleaned_text = clean_text(text_content)
                print(cleaned_text + "\n")
                texts.append(cleaned_text)
                # 打印提取的文本内容

            else:
                print(f"无法访问网页：{url}")
        except requests.Timeout:
            print(f"请求超时，超过了{timeout_seconds}秒的等待时间。链接：{url}")
            continue
        except BaseException as e:
            print(f"发生了一个异常：{e}")
            continue
    print("成功获取" + str(len(texts)) + "个链接内容", end="\n")

    return texts


#
def clean_text(text):

    # 使用正则表达式去除非中英文、数字的内容
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9，。？！；：”“（）]', ' ', text)
    # 将连续的空白字符替换为单个空格
    text = re.sub(r'\s+', ' ', text).strip()
    return text



def load_stopwords():
    # 加载哈工大停用词表
    stopwords_file_path = 'hit_stopwords.txt'

    with open(stopwords_file_path, 'r', encoding='utf-8') as file:
        stopwords = set(file.read().split())
    return stopwords


def get_top_texts(query, texts, top_n, top_chars):
    # 使用jieba进行中文分词
    tokenized_texts = [list(jieba.cut(text)) for text in texts]

    # 移除停用词
    hit_stopwords = load_stopwords()
    tokenized_texts = [[word for word in text if word not in hit_stopwords] for text in tokenized_texts]

    # 创建BM25模型
    bm25 = BM25Okapi(tokenized_texts)

    # 对查询进行分词
    tokenized_query = list(jieba.cut(query))

    # 使用BM25模型对查询进行打分
    scores = bm25.get_scores(tokenized_query)

    # 将打分结果与文本内容组合，并按分数降序排序
    scored_texts = sorted(
        [{"text": text, "score": score} for score, text in zip(scores, texts)],
        key=lambda x: x["score"],
        reverse=True
    )

    # 截取每个文本的前500个字符
    top_texts_truncated = [{"text": scored_text["text"][:top_chars], "score": scored_text["score"]} for scored_text in scored_texts[:top_n]]

    # 返回打分最高的top_n个数据结构
    return top_texts_truncated

@app.route('/search', methods=['GET'])
def get_advice():
    query = request.args.get('query', '')

    res = extract_content(query, 30)
    text = get_top_texts(query, res, 2, 500)
    #print(res)
    print(text)
    return jsonify({'text': text})


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)
效果展示

浏览器搜索 http://localhost:5000/search?query=硬盘

联网方案二：Bing API

逻辑思路
  1. 调用Bing的API服务来获取数据，从中获取链接内容。
  2. 使用BM25算法进行相关度排序，然后从剩余内容中取出前n个文本，并且每个文本只取前k个字。
获取Bing API密钥和自定义搜索引擎ID

自行搜索教程

代码实现

PS：把 hit_stopwords.txt （哈工大停用词）、cacert.pem（证书）放到 bing_search_api.py 同目录下

bing_search_api.py：

import json
import requests
from rank_bm25 import BM25Okapi
import jieba
from flask import Flask, request, jsonify

app = Flask(__name__)

# 定义全局变量
subscriptionKey = ""   # bing 密钥
endpoint = "https://api.bing.microsoft.com"
customConfigId = ""
mkt = "zh-CN"
setLang = "ZH"
count = "10"
max_length = 500


def load_stopwords():
    stopwords_file_path = 'hit_stopwords.txt'
    with open(stopwords_file_path, 'r', encoding='utf-8') as file:
        stopwords = set(file.read().split())
    return stopwords


def extract_snippet_part(text):
    return text.split('：', 1)[1]


def get_texts(query, texts, max_length):
    hit_stopwords = load_stopwords()
    tokenized_texts = [list(jieba.cut(extract_snippet_part(text))) for text in texts]
    tokenized_texts = [[word for word in text if word not in hit_stopwords] for text in tokenized_texts]
    bm25 = BM25Okapi(tokenized_texts)
    tokenized_query = list(jieba.cut(query))
    scores = bm25.get_scores(tokenized_query)
    scored_texts = sorted(
        [{"text": text, "score": score} for score, text in zip(scores, texts)],
        key=lambda x: x["score"],
        reverse=True
    )
    print(scored_texts)

    top_texts = ""
    count = 0
    for item in scored_texts:
        if len(top_texts) <= max_length or count < 3:
            top_texts += item['text'] + "；"
            count += 1

    return top_texts[:max_length]


def bing_search(query):
    searchTerm = "\"" + query + "\"" + " -广告"
    url = f"{endpoint}/v7.0/custom/search?q={searchTerm}&customconfig={customConfigId}&mkt={mkt}" \
          f"&setLang={setLang}&count={count}&responseFilter=webpages"

    r = requests.get(url, headers={'Ocp-Apim-Subscription-Key': subscriptionKey}, verify="cacert-2024-09-24.pem")
    data = json.loads(r.text)

    url_snippet_sets = [item['url'] + "：" + item['snippet'] for item in data['webPages']['value'][:int(count)]]

    return get_texts(query, url_snippet_sets, max_length)


@app.route('/search', methods=['GET'])
def get_search():
    query = request.args.get('query', '')
    res = bing_search(query)
    print(res)
    return jsonify({'text': res})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)
效果展示

浏览器搜索 http://localhost:5000/search?query=硬盘

总结

调用 API 缺点（待优化点）：

  • 搜索的内容相关度有限（跟用户输入的搜索词也有关），有些会包含广告等无关的内容

  • 搜索的内容文本字数有限，即提供给模型的信息量有限

爬虫的缺点（待优化点）：

  • 爬取的内容含有太多的无关信息，包括目录、广告等

  • 请求链接可能失败，外国的网页加载时间长

参考文档

如何为本地AI模型实现联网搜索功能：使用DuckDuckGo API的实现指南 – 老猫

定制基于 ChatGLM-6B 的本地知识库+联网检索能力的 LangChain Agent - 知乎

查询向量存储 | LangChain中文网:500页中文文档教程，助力大模型LLM应用开发从入门到精通

初识爬虫

是小白_鸭-CSDN博客收藏夹


原文地址：https://blog.csdn.net/qq_51047851/article/details/145181929

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

相关文章

自学内容网
Copyright © 2015-2025