AI联网搜索实现(SerpApi+crawl4ai)

2026年04月26日 aigc, python, ai大模型预计阅读 1 分钟

```python
import requests
import json
import openai
import asyncio
import os
import re
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import BM25ContentFilter, PruningContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

serp_api_key = os.getenv("SERP_API_KEY")
api_key = os.getenv("AI_API_KEY")
base_url = os.getenv("AI_BASE_URL")
model = os.getenv("AI_MODEL")
searxng_url = os.getenv("SEARXNG_URL")

# 搜索
def serp_search(query, search_num=10, country="us", language="en", exclude_images=False, exclude_videos=False):
    """
    使用SERP API执行搜索，并返回结果URL列表
    
    参数:
        query (str): 搜索查询词
        search_num (int): 希望返回的结果数量
        country (str): 搜索结果的国家代码，默认为'us'
        language (str): 搜索结果的语言代码，默认为'en'
        exclude_images (bool): 是否排除图片结果，默认为False
        exclude_videos (bool): 是否排除视频结果，默认为False
        
    返回:
        list: 搜索结果URL列表
    """
    url = "https://serpapi.com/search"
    params = {
        "api_key": serp_api_key,
        "q": query,
        "num": search_num,
        "gl": country,
        "hl": language,
        "engine": "google"  # 使用Google搜索引擎
    }

# 使用Google搜索操作符排除图片和视频
    if exclude_images or exclude_videos:
        exclude_operators = []
        if exclude_images:
            # 排除常见图片文件类型和图片相关网站
            exclude_operators.extend([
                "-filetype:jpg", "-filetype:jpeg", "-filetype:png", "-filetype:gif",
                "-filetype:webp", "-filetype:bmp", "-filetype:svg",
                "-site:images.google.com", "-site:pinterest.com", "-site:instagram.com"
            ])
        if exclude_videos:
            # 排除常见视频文件类型和视频网站
            exclude_operators.extend([
                "-filetype:mp4", "-filetype:avi", "-filetype:mov", "-filetype:wmv",
                "-filetype:flv", "-filetype:mkv", "-filetype:webm", "-filetype:video"
                                                                    "-site:youtube.com", "-site:youtu.be",
                "-site:vimeo.com",
                "-site:bilibili.com", "-site:tiktok.com", "-site:douyin.com"
            ])

# 将排除操作符添加到查询中
        if exclude_operators:
            params["q"] = f"{query} {' '.join(exclude_operators)}"

try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        search_results = response.json().get("organic_results", [])
        search_results_info = []
        for result in search_results:
            search_results_info.append({
                "link": result.get("link", ""),
                "title": result.get("title", ""),
                "snippet": result.get("snippet", "")
            })

return search_results_info
    except requests.exceptions.RequestException as e:
        print(f"SERP API请求出错: {e}")
        return []
    except (KeyError, json.JSONDecodeError) as e:
        print(f"解析SERP API结果出错: {e}")
        return []

# 爬虫
async def crawl(urls):
    # 浏览器配置
    browser_config = BrowserConfig(
        headless=True,  # 是否无头模式，True：不打卡浏览器
        viewport_width=1280,  # 视口宽度
        viewport_height=720,  # 视口高度
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        # 用户代理
        text_mode=True,  # 文本模式,禁用图片加载
    )
    # 爬虫运行配置
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.DISABLED,  # 禁用缓存模式，获取最新内容
        stream=False,  # 确保返回列表而不是异步生成器
        page_timeout=5000,  # 设置页面超时时间为5秒
        markdown_generator=DefaultMarkdownGenerator(
            content_filter=PruningContentFilter(
                # min_word_threshold = 10, # 丢掉少于10个单词的块，因为他们可能太短或者无用
                threshold=0.7,  # 丢掉重要度低于0.76的块,越高过滤越严格
                threshold_type="dynamic",  # 重要度阈值类型，fixed：固定值，dynamic：相对值
                # threshold_type = "fixed"
            ),
            options={
                "ignore_links": True,  # 是否在最终markdown中移除所有超链接
                "ignore_images": True,  # 是否在最终markdown中移除所有图片
            }
        )
    )  # 爬虫运行配置

async with AsyncWebCrawler(config=browser_config) as crawler:
        try:
            results = await crawler.arun_many(
                urls=urls,
                config=run_config
            )
        except Exception as e:
            print(f"爬取错误：{e}")
            return {"error": "爬取失败", "detail": str(e)}

results_list = []
        failed_urls = []
        success_urls = []
        try:
            for result in results:
                # 检查爬取是否成功
                if result.markdown is None or len(result.markdown.fit_markdown) < 50:
                    print(f"跳过失败的URL: {result.url}")
                    failed_urls.append({
                        "url": result.url,
                        "error": "内容为空或过短"
                    })
                    continue
                results_list.append({
                    "url": result.url,
                    "content": result.markdown.fit_markdown
                })
                print(f"成功爬取内容: {result.markdown.fit_markdown}")
                success_urls.append(result.url)
        except Exception as e:
            print(f"处理结果错误：{e}")
            return {"error": "处理爬取结果失败", "detail": str(e)}

return {
            "success_urls": success_urls,
            "markdown_results": results_list,
            "success_count": len(results_list),
            "failed_urls": failed_urls,
            "failed_count": len(failed_urls),
            "total_urls": len(results_list) + len(failed_urls)
        }

# AI回答的json格式修复
def AI_fix_json(answer: str):
    """
    修复AI回答的json格式
    优先提取```json ... ```代码块内容，如果存在则只保留代码块内内容，然后再去除```json和```，最后返回json.loads结果。
    """
    # 优先提取```json ... ```代码块
    match = re.search(r"```json(.*?)```", answer, re.DOTALL)
    if match:
        answer = match.group(1)
    # 去除多余空白字符
    answer = answer.strip()
    # 再去除可能残留的```json和```
    answer = answer.replace("```json", "").replace("```", "").strip()
    return json.loads(answer)

# AI选择爬取网页
def AI_choose(search_results_info, query, choose_num=3):
    """
    使用AI判断搜索结果是否足以回答用户问题，如果不够则返回需要爬取的URLs
    
    参数:
        search_results_info (list): 搜索结果列表
        query (str): 用户的原始问题
        
    返回:
        dict: 包含以下键值对:
            - sufficient (bool): 搜索结果是否足以回答用户问题
            - answer (str): 如果搜索结果足够，则为回答内容
            - urls_to_crawl (list): 如果搜索结果不够，则为需要爬取的URL列表
    """
    # 使用AI选择最佳结果
    client = openai.OpenAI(api_key=api_key, base_url=base_url)

system_prompt = f"""
    你是一名资深的信息检索与问答专家，需要判断【搜索结果摘要 snippet】是否已足够回答用户问题，决定是否还要爬取网页全文。

========== 判定标准 ==========
    一、可判定为"足够"("sufficient": true ) 的情形
    1. 用户问题只需单一事实或极短答案（数字、日期、地名、温度、价格、概念定义、是/否判断等）。  
    2. 至少有 1 条 snippet 同时满足：
        • 明确包含问题所求的关键信息；  
        • 信息完整、可直接复述给用户而不需补充上下文；  
        • 可信度高（来自政府、知名媒体、维基百科、权威数据站等）或多条 snippet 相互印证。

二、必须判定为"不足"("sufficient": false ) 的情形
    1. snippet 仅回答部分要点或缺乏具体数值 / 事实。  
    2. snippet 信息彼此矛盾，或来源可疑，需要阅读全文核实。  
    3. 用户问题含有 "详细 / 全面 / 教程 / 步骤 / 代码 / 对比 / 原理 / 示例" 等字眼。

========== URL 选择规则 ==========
    如果 sufficient = false，请从搜索结果中挑选最可能包含完整答案的前 {choose_num} 个链接填入 "urls_to_crawl"；若 sufficient = true，则 "urls_to_crawl": []。

========== 输出格式 ==========
    只返回以下 JSON，字段顺序保持一致，布尔值用首字母小写：
    {{
      "sufficient": true/false,
      "urls_to_crawl": ["url1", "url2"]
    }}
    """

try:
        response = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user",
                 "content": f"用户问题: {query}\n\n搜索结果: {json.dumps(search_results_info, ensure_ascii=False, indent=2)}"}
            ],
        )
        result = response.choices[0].message.content
        print("👌👌AI选择结果", result)
        result = AI_fix_json(result)
        print("👌👌AI选择结果修复后", result)
        return result
    except Exception as e:
        print(f"❌❌AI选择失败：{e}")

def choose_crawl(query, choose_num=3, search_num=10, country="cn", language="zh-CN", exclude_images=True,
                 exclude_videos=True):
    print("✅✅开始AI选择爬取网页")
    try:
        search_results_info = serp_search(query, search_num, country, language, exclude_images, exclude_videos)
        print("👌👌获取到联网搜索结果", search_results_info[:30])
    except Exception as e:
        return {"error": "❌❌SERP API搜索失败", "detail": str(e)}
    try:

ai_result = AI_choose(search_results_info, query, choose_num)
    except Exception as e:
        return {"error": "❌❌AI选择失败", "detail": str(e)}
    # 如果AI判断搜索结果足以回答问题，则返回搜索结果
    if ai_result["sufficient"]:
        print("=======================================================================================")
        print("👌👌AI判断搜索结果足以回答问题")
        return {"sufficient": True, "search_results_info": search_results_info}
    else:
        # 如果AI判断搜索结果不足以回答问题，则爬取网页
        print("=======================================================================================")
        print("🤔🤔AI判断搜索结果不足以回答问题")
        print("✅✅需要爬取的URLs", ai_result["urls_to_crawl"])
        final_result = {
            "sufficient": False,
            "search_results_info": search_results_info,
            "crawl_result": ""
        }

# 爬取网页
        result = asyncio.run(crawl(ai_result["urls_to_crawl"]))
        final_result["crawl_result"] = result

return final_result

if __name__ == "__main__":
    query = "当前最新的模型是那些，详细说明一点"
    result = choose_crawl(query)

client = openai.OpenAI(api_key=api_key, base_url=base_url)
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一名资深的美食专家，需要根据用户问题和爬取的网页内容，回答用户问题。"},
            {"role": "user", "content": f"用户问题: {query}\n\n搜索的网页内容: {result}"}
        ]
    )
    print("👌👌开始AI回答")
    print(response.choices[0].message.content)

# 安装依赖：pip install crawl4ai openai requests;  playwright install chromium
# 👌👌获取到联网搜索结果 [{'link': 'https://zhuanlan.zhihu.com/p/670574382', 'title': '国内外知名大模型及应用——模型/应用维度（2026/04/24）', 'snippet': '阿里通义团队发布的开源视频生成模型，最新版是2025年7月发布的v2.2，这是一个基于先进Wan2.2-VAE构建的5B和14B参数模型。该模型支持720P分辨率 ...'}, {'link': 'https://help.openai.com/zh-hans-cn/articles/9624314-model-release-notes', 'title': '模型发布说明', 'snippet': '我们开发了一系列新的AI 模型，设计目标是在响应前花更多时间思考。它们能够对复杂任务进行推理，并在科学、编码和数学方面解决比以往模型更难的问题。 今天 ...'}, {'link': 'https://hub.baai.ac.cn/view/51654', 'title': '2025大模型最全复盘：“中国开源”崛起、OpenAI“走下神坛”', 'snippet': '2025 年，是“中国开源”与“美国闭源”的竞争之年。这一年，GLM-4.7、Kimi K2 Thinking 等国产模型开始在全球范围内赢得更多关注。'}, {'link': 'https://help.aliyun.com/zh/model-studio/models', 'title': '模型列表 - 阿里云文档', 'snippet': '阿里云百炼提供了丰富多样的模型选择，它集成了千问系列大模型和第三方大模型，涵盖文本、图像、音视频等不同模态。'}, {'link': 'https://openai.xiniushu.com/docs/models', 'title': '模型（Model） | OpenAI 官方帮助文档中文版', 'snippet': '如果要使用最新的模型版本，请使用标准模型名称，如 gpt-4 或 gpt-3.5-turbo 。 Model name（GPT 模型）, Discontinuation date（停用时间）, Replacement model（替换的GPT 模型） ...'}, {'link': 'https://help.openai.com/zh-hans-cn/articles/6825453-chatgpt-%E5%8F%91%E5%B8%83%E8%AF%B4%E6%98%8E', 'title': 'ChatGPT — 发布说明', 'snippet': 'ChatGPT 在用户询问时仍会告知当前激活的是哪个模型。 GPT-5 Instant 的这项更新今天开始向ChatGPT 用户推出。我们将继续改进，并会持续更新该模型 ...'}, {'link': 'https://zhuanlan.zhihu.com/p/1971614245851493901', 'title': '开源大模型哪家强？9大模型架构演变历程一次性看明白', 'snippet': 'DeepSeek、Claude、Kimi2等系列模型的持续进化，标志着大模型技术已从“可用”迈向“高效可靠”，其背后隐藏的核心变化亟待系统梳理。 当前，主流开源模型的提升 ...'}, {'link': 'https://docs.github.com/zh/copilot/reference/ai-models/model-comparison', 'title': 'AI 模型比较- GitHub 文档', 'snippet': 'AI 模型的比较用于GitHub Copilot. GitHub Copilot 支持具有不同功能的多个AI 模型。 你选择的模型会影响副驾驶聊天和Copilot 内联建议的响应质量和相关性。'}, {'link': 'https://ai.google.dev/gemini-api/docs/gemini-3?hl=zh-cn', 'title': 'Gemini 3 开发者指南', 'snippet': 'Gemini 3 Flash 是我们最新的3 系列模型，具有Pro 级智能，但速度和价格与Flash 相同。 Nano Banana Pro（也称为Gemini 3 Pro Image）是我们质量最高的图片生成模型 ...'}, {'link': 'https://api-docs.deepseek.com/zh-cn/news/news250528', 'title': 'DeepSeek-R1 更新，思考更深，推理更强', 'snippet': 'DeepSeek R1 模型已完成小版本升级，当前版本为DeepSeek-R1-0528。用户通过官方网站、APP 或小程序进入对话界面后，开启“深度思考”功能即可体验最新版本 ...'}]
```

评论区