使用 Python 制作一个属于自己的 AI 搜索引擎

news2024/9/20 18:29:46

1. 使用到技术

OpenAI KEY
Serper KEY
Bing Search

2. 原理解析

使用Google和Bing的搜搜结果交由OpenAI处理并给出回答。

3. 代码实现

import requests
from lxml import etree
import os
from openai import OpenAI

# 从环境变量中加载 API 密钥
os.environ["OPENAI_API_KEY"] = "sk-xxxx-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
os.environ["SERPER_API_KEY"] = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# 确保在执行代码前已经设置了环境变量
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
SERPER_API_KEY = os.getenv("SERPER_API_KEY")

def search_bing(query):
    headers = {
        'Referer': 'https://www.bing.com/',
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.0.0 Safari/537.36',
    }
    params = {
        'q': query,
        'mkt': 'zh-CN'
    }

    response = requests.get('https://www.bing.com/search', params=params, headers=headers)

    html = etree.HTML(response.text)

    li_list = html.xpath("//li[@class='b_algo']")

    result = []

    for index in range(len(li_list)):
        title = ";".join(li_list[index].xpath("./h2/a/text()"))
        link = li_list[index].xpath("./h2/a/@href")[0]
        snippet = ";".join(li_list[index].xpath("./div/p/text()"))
        position = index
        print(title, link, snippet, position)
        result.append({
            'title': title,
            'link': link,
            'snippet': snippet,
            'position': position,
        })
    return result

def search_serper(query):
    """使用Serper API进行搜索并返回结果。"""
    url = "https://google.serper.dev/search"
    headers = {
        "X-API-KEY": SERPER_API_KEY,
        "Content-Type": "application/json",
    }
    params = {
        'q': query,
        'gl': "cn",
        'hl': "zh-cn",
    }

    try:
        response = requests.post(url, headers=headers, json=params)
        response.raise_for_status()  # 检查HTTP请求状态
        serper_data = response.json()
        if not serper_data:
            return "无法获取搜索结果", []
        google_context = serper_data.get('organic', [])
        google_other = serper_data.get('relatedSearches', [])
        return google_context, google_other
    except requests.RequestException as e:
        print(f"请求失败: {e}")
        return None

def search_openai(query, context):
    """利用OpenAI API回答问题并引用相关上下文，并使用流的方式输出。"""
    context_template = (
        "你是GinLynn构建的大型语言AI助手。给你一个用户问题，请正确、简洁、准确的讲述这个问题的答案。"
        "你会得到一组与问题相关的上下文，其中每个对象都是一个json字符串，"
        "'snippet'字段表示片段，'title'字段表示标题，'link'字段表示链接，'position'字段表示位置。"
        "请使用这些上下文并在每个句子的末尾引用上下文(如果适用)。"
        "你的答案必须是正确、准确的，由专家以公正和专业的语气撰写。请限制为2048token。"
        "不要给出任何与问题无关的信息，也不要重复。如果给定的上下文没有提供足够的信息，"
        "那么在相关主题后面加上“information is missing on”。请以[position]的格式注明出处和参考编号。"
        "以下是一组上下文:"
    )

    client = OpenAI(api_key=OPENAI_API_KEY)

    try:
        completion = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": context_template + context},
                {"role": "user", "content": query}
            ],
            stream=True  # 启用流式响应
        )

        # 逐条打印流式输出的结果
        for chunk in completion:
            if chunk.choices[0].delta.content is not None:
                print(chunk.choices[0].delta.content, end="")

        print()  # 输出换行
        return "完成输出"

    except Exception as e:
        print(f"OpenAI API request failed: {e}")
        return "无法完成请求", []


if __name__ == '__main__':
    query = input("请输入查询: ")

    if query.strip() == "":
        query = "最新俄乌局势信息"

    print("正在搜索...")
    serper_context, other_queries = search_serper(query)
    bing_context = search_bing(query)

    context = []
    if bing_context:
        context.extend(bing_context)
    if serper_context:
        # 为Serper上下文的每个条目重置 position 值，以防止重复
        for index, item in enumerate(serper_context, start=len(bing_context)):
            item['position'] = index  # 从当前Bing结果的数量开始
        context.extend(serper_context)

    print("搜索结果:", context)
    search_openai(query, str(context))

    if other_queries:
        print("相关搜索:", other_queries)