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

1. 使用到技术

1. OpenAI KEY
2. Serper KEY
3. 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)

4. 运行结果