AI搜索 工作流自动化教程｜附源码

在信息爆炸的时代，传统搜索引擎已经无法完全满足企业和个人对“精准、可追溯、可自动处理”的信息获取需求。我们不再只是想要一串网页链接，而是希望系统能够自动理解问题、检索资料、提取重点、生成答案，甚至把结果推送到飞书、钉钉、企业微信、邮件或数据库中。

这类能力通常被称为 AI搜索工作流自动化。

本文将从零开始，带你搭建一个简单但完整的 AI 搜索自动化工作流。它可以完成以下任务：

• 接收用户输入的问题；
• 自动生成搜索关键词；
• 调用搜索接口获取网页结果；
• 抓取网页正文内容；
• 使用大模型进行总结与问答；
• 输出带引用来源的答案；
• 可扩展为定时任务、知识库更新、日报生成、竞品监控等自动化流程。

本文附带完整 Python 示例源码，适合有一定编程基础的开发者学习和二次改造。

一、什么是 AI 搜索工作流自动化？

传统搜索流程通常是：

输入关键词 → 查看搜索结果 → 打开网页 → 人工筛选 → 阅读内容 → 总结结论

而 AI 搜索工作流自动化的流程是：

输入问题 → AI 理解意图 → 自动搜索 → 自动读取网页 → 自动总结 → 生成结构化答案 → 自动推送或存储

简单来说，它把“搜索 + 阅读 + 总结 + 分发”这几个步骤自动化了。

例如，你可以输入：

最近一周有哪些关于 OpenAI、Claude、Gemini 的重要更新？

系统会自动搜索相关新闻、官方博客、社交媒体文章，然后汇总成一份中文简报，并附上来源链接。

再比如，你可以输入：

帮我分析一下国内 AI 搜索产品的竞争格局。

系统可以搜索相关资料，提取每个产品的功能、商业模式、优缺点，并输出结构化分析报告。

二、典型应用场景

AI 搜索自动化并不是一个单一产品，而是一套可以嵌入到业务中的能力。常见场景包括：

1. 行业资讯日报

每天早上自动搜索指定行业关键词，例如“人工智能”“新能源”“跨境电商”“AIGC 工具”等，然后生成日报推送到微信群、飞书群或邮箱。

2. 竞品监控

定时搜索竞品名称、融资消息、新功能发布、招聘动态、官网更新等内容，帮助企业快速掌握市场变化。

3. 学术资料检索

输入研究主题后，自动搜索论文、博客、技术报告，并总结研究现状、代表观点和参考链接。

4. 企业知识助手

将搜索结果与企业内部文档结合，构建一个既能查内部知识，又能查外部信息的智能问答系统。

5. 内容创作辅助

自动搜索选题资料，提取观点、数据、案例，生成文章大纲、短视频脚本或营销文案。

三、整体架构设计

本文实现的 AI 搜索工作流主要包含以下模块：

用户问题
   ↓
问题理解与关键词生成
   ↓
搜索引擎 API
   ↓
网页内容抓取
   ↓
内容清洗与截断
   ↓
大模型总结与回答
   ↓
结构化结果输出

如果要用于生产环境，还可以增加：

任务队列
定时调度
数据库存储
向量检索
权限控制
日志监控
消息推送

为了便于学习，本文采用相对轻量的实现方式：

• 编程语言：Python
• Web 框架：FastAPI
• 搜索接口：SerpAPI 或自定义搜索 API
• 网页解析：BeautifulSoup
• HTTP 请求：requests
• 大模型接口：OpenAI API 兼容格式
• 配置管理：dotenv

你也可以将大模型替换为国内模型，例如通义千问、智谱、DeepSeek、Moonshot、豆包等，只要接口兼容即可。

四、准备工作

1. 安装 Python 环境

建议使用 Python 3.10 及以上版本。

创建项目目录：

mkdir ai-search-workflow
cd ai-search-workflow

创建虚拟环境：

python -m venv venv

激活虚拟环境：

macOS / Linux：

source venv/bin/activate

Windows：

venv\Scripts\activate

2. 安装依赖

创建 requirements.txt：

fastapi
uvicorn
requests
beautifulsoup4
python-dotenv
openai
pydantic

安装依赖：

pip install -r requirements.txt

3. 配置环境变量

创建 .env 文件：

OPENAI_API_KEY=你的大模型API密钥
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-4o-mini

SERPAPI_KEY=你的SerpAPI密钥

如果你使用的是兼容 OpenAI 格式的国内模型，只需要修改 OPENAI_BASE_URL 和 MODEL_NAME 即可。

例如：

OPENAI_API_KEY=你的DeepSeek密钥
OPENAI_BASE_URL=https://api.deepseek.com
MODEL_NAME=deepseek-chat

五、项目目录结构

推荐目录如下：

ai-search-workflow/
├── main.py
├── search_engine.py
├── crawler.py
├── llm.py
├── workflow.py
├── requirements.txt
└── .env

各文件作用：

六、搜索模块源码

新建 search_engine.py：

import os
import requests
from typing import List, Dict
from dotenv import load_dotenv

load_dotenv()

SERPAPI_KEY = os.getenv("SERPAPI_KEY")


def search_web(query: str, num_results: int = 5) -> List[Dict]:
    """
    调用 SerpAPI 进行网页搜索。
    返回结果包含标题、链接和摘要。
    """
    if not SERPAPI_KEY:
        raise ValueError("缺少 SERPAPI_KEY，请在 .env 文件中配置")

    url = "https://serpapi.com/search.json"

    params = {
        "engine": "google",
        "q": query,
        "api_key": SERPAPI_KEY,
        "num": num_results,
        "hl": "zh-cn"
    }

    response = requests.get(url, params=params, timeout=20)
    response.raise_for_status()

    data = response.json()
    organic_results = data.get("organic_results", [])

    results = []

    for item in organic_results[:num_results]:
        results.append({
            "title": item.get("title", ""),
            "link": item.get("link", ""),
            "snippet": item.get("snippet", "")
        })

    return results

如果你不想使用 SerpAPI，也可以替换成 Bing Search API、Google Custom Search API，或者自建搜索服务。

七、网页抓取模块源码

新建 crawler.py：

import requests
from bs4 import BeautifulSoup


def fetch_page_text(url: str, max_chars: int = 5000) -> str:
    """
    抓取网页正文文本。
    为了避免输入过长，这里限制最大字符数。
    """
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/120.0 Safari/537.36"
        )
    }

    try:
        response = requests.get(url, headers=headers, timeout=15)
        response.raise_for_status()
    except Exception as e:
        return f"网页抓取失败：{str(e)}"

    response.encoding = response.apparent_encoding

    soup = BeautifulSoup(response.text, "html.parser")

    # 移除无关标签
    for tag in soup(["script", "style", "noscript", "header", "footer", "nav"]):
        tag.decompose()

    text = soup.get_text(separator="\n")

    # 清洗空行
    lines = []
    for line in text.splitlines():
        line = line.strip()
        if line:
            lines.append(line)

    clean_text = "\n".join(lines)

    return clean_text[:max_chars]

这个抓取方式适合普通网页。对于强反爬网站、动态渲染页面，可能需要使用 Playwright、Selenium 或专门的内容解析服务。

八、大模型调用模块源码

新建 llm.py：

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
MODEL_NAME = os.getenv("MODEL_NAME", "gpt-4o-mini")

client = OpenAI(
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL
)


def call_llm(prompt: str, system_prompt: str = "你是一个严谨的中文AI搜索助手。") -> str:
    """
    调用大模型生成回答。
    """
    if not OPENAI_API_KEY:
        raise ValueError("缺少 OPENAI_API_KEY，请在 .env 文件中配置")

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[
            {
                "role": "system",
                "content": system_prompt
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.2
    )

    return response.choices[0].message.content

这里将 temperature 设置为 0.2，是为了让回答更稳定、更少发散。对于搜索问答类任务，建议不要设置得太高。

九、工作流编排模块源码

新建 workflow.py：

from typing import Dict, List
from search_engine import search_web
from crawler import fetch_page_text
from llm import call_llm


def generate_search_query(user_question: str) -> str:
    """
    使用大模型将用户问题改写为更适合搜索引擎的关键词。
    """
    prompt = f"""
请将下面的用户问题改写为适合搜索引擎检索的关键词。
要求：
1. 保留核心语义；
2. 不要输出解释；
3. 只输出一行搜索关键词。

用户问题：
{user_question}
"""

    query = call_llm(prompt)
    return query.strip().replace('"', "")


def build_context(search_results: List[Dict]) -> str:
    """
    根据搜索结果抓取网页正文，并构造上下文。
    """
    context_parts = []

    for index, item in enumerate(search_results, start=1):
        title = item.get("title", "")
        link = item.get("link", "")
        snippet = item.get("snippet", "")

        page_text = fetch_page_text(link, max_chars=4000)

        context = f"""
资料 [{index}]
标题：{title}
链接：{link}
搜索摘要：{snippet}
网页正文节选：
{page_text}
"""
        context_parts.append(context)

    return "\n\n".join(context_parts)


def answer_with_sources(user_question: str, context: str) -> str:
    """
    基于搜索上下文生成带来源的答案。
    """
    prompt = f"""
你是一个严谨的 AI 搜索助手。请基于给定资料回答用户问题。

要求：
1. 必须使用中文回答；
2. 不要编造资料中没有的信息；
3. 如果资料不足，请明确说明“不足以判断”；
4. 回答要结构化，可以使用标题、列表和表格；
5. 重要结论后请标注来源编号，例如：[1]、[2]；
6. 最后输出“参考来源”列表，包含标题和链接。

用户问题：
{user_question}

可用资料：
{context}
"""

    return call_llm(prompt)


def run_ai_search_workflow(user_question: str, num_results: int = 5) -> Dict:
    """
    AI搜索自动化工作流主函数。
    """
    search_query = generate_search_query(user_question)

    search_results = search_web(search_query, num_results=num_results)

    context = build_context(search_results)

    answer = answer_with_sources(user_question, context)

    return {
        "question": user_question,
        "search_query": search_query,
        "results": search_results,
        "answer": answer
    }

这个文件是整个系统的核心。它将“问题改写、搜索、抓取、总结”串联起来，形成一个可复用的自动化工作流。

十、FastAPI 服务入口源码

新建 main.py：

from fastapi import FastAPI
from pydantic import BaseModel, Field
from workflow import run_ai_search_workflow

app = FastAPI(
    title="AI Search Workflow API",
    description="一个简单的 AI 搜索工作流自动化示例",
    version="1.0.0"
)


class SearchRequest(BaseModel):
    question: str = Field(..., description="用户问题")
    num_results: int = Field(5, description="搜索结果数量")


@app.get("/")
def health_check():
    return {
        "message": "AI Search Workflow API is running."
    }


@app.post("/ai-search")
def ai_search(request: SearchRequest):
    result = run_ai_search_workflow(
        user_question=request.question,
        num_results=request.num_results
    )
    return result

启动服务：

uvicorn main:app --reload --port 8000

启动成功后，浏览器访问：

http://127.0.0.1:8000/docs

你会看到 FastAPI 自动生成的接口文档。

十一、接口调用示例

使用 curl 调用：

curl -X POST "http://127.0.0.1:8000/ai-search" \
-H "Content-Type: application/json" \
-d '{
  "question": "最近有哪些值得关注的AI搜索产品？请比较它们的特点。",
  "num_results": 5
}'

返回结果结构大致如下：

{
  "question": "最近有哪些值得关注的AI搜索产品？请比较它们的特点。",
  "search_query": "AI搜索产品 特点 对比 Perplexity Kimi 秘塔 纳米搜索",
  "results": [
    {
      "title": "某某AI搜索产品介绍",
      "link": "https://example.com/article",
      "snippet": "这里是搜索摘要"
    }
  ],
  "answer": "以下是基于搜索资料整理的分析……"
}

十二、如何扩展为自动化工作流？

上面的示例是一个同步接口：用户发起请求，系统实时搜索并返回结果。实际业务中，我们经常需要把它扩展为自动运行的工作流。

1. 定时生成日报

可以使用 APScheduler 或 Linux cron 定时执行搜索任务。

安装：

pip install apscheduler

示例代码：

from apscheduler.schedulers.blocking import BlockingScheduler
from workflow import run_ai_search_workflow

scheduler = BlockingScheduler()


@scheduler.scheduled_job("cron", hour=8, minute=30)
def daily_ai_news():
    result = run_ai_search_workflow(
        user_question="过去24小时全球人工智能行业有哪些重要新闻？",
        num_results=8
    )

    print(result["answer"])


if __name__ == "__main__":
    scheduler.start()

这样每天早上 8:30 就会自动生成 AI 行业简报。

2. 推送到飞书或企业微信

如果要推送到企业微信机器人，可以增加如下函数：

import requests


def send_wecom_message(webhook_url: str, content: str):
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": content
        }
    }

    response = requests.post(webhook_url, json=payload, timeout=10)
    response.raise_for_status()

    return response.json()

然后在任务完成后调用：

send_wecom_message(
    webhook_url="你的企业微信机器人Webhook",
    content=result["answer"]
)

3. 保存到数据库

如果希望保存每次搜索结果，可以使用 SQLite、MySQL、PostgreSQL 或 MongoDB。

简单 SQLite 示例：

import sqlite3
import json
from datetime import datetime


def init_db():
    conn = sqlite3.connect("ai_search.db")
    cursor = conn.cursor()

    cursor.execute("""
    CREATE TABLE IF NOT EXISTS search_logs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        question TEXT,
        search_query TEXT,
        answer TEXT,
        raw_results TEXT,
        created_at TEXT
    )
    """)

    conn.commit()
    conn.close()


def save_search_log(result: dict):
    conn = sqlite3.connect("ai_search.db")
    cursor = conn.cursor()

    cursor.execute("""
    INSERT INTO search_logs (
        question,
        search_query,
        answer,
        raw_results,
        created_at
    ) VALUES (?, ?, ?, ?, ?)
    """, (
        result["question"],
        result["search_query"],
        result["answer"],
        json.dumps(result["results"], ensure_ascii=False),
        datetime.now().isoformat()
    ))

    conn.commit()
    conn.close()

通过存储历史搜索结果，你可以进一步做数据分析、趋势追踪和知识沉淀。

十三、关键优化点

虽然上面的系统已经可以运行，但如果要在真实环境中使用，还需要关注以下优化点。

1. 搜索结果质量

搜索质量决定了最终答案质量。建议：

• 使用更稳定的搜索 API；
• 对搜索结果进行去重；
• 优先选择官方文档、权威媒体、论文、白皮书；
• 降低低质量内容站点权重；
• 根据时间、语言、地区进行过滤。

2. 网页正文提取

普通网页可以用 BeautifulSoup，但复杂网页可能出现正文缺失、广告过多、动态渲染等问题。可以考虑：

• 使用 trafilatura 提取正文；
• 使用 readability-lxml；
• 使用 Playwright 渲染后抓取；
• 使用第三方内容解析 API。

3. 长文本处理

如果抓取到的网页太长，直接塞给大模型会超出上下文限制。常见方案包括：

• 对每篇文章先单独总结；
• 使用 Map-Reduce 总结模式；
• 先切块，再筛选与问题最相关的片段；
• 使用向量检索挑选内容。

4. 防止幻觉

AI 搜索系统最大的风险之一是模型“看似合理地编造”。降低幻觉的方法包括：

• 明确要求模型只基于资料回答；
• 输出引用来源；
• 对关键事实进行二次校验；
• 不足以判断时明确说明；
• 保留原始链接方便人工追溯。

5. 异步与并发

当前示例是串行抓取网页，速度较慢。可以改为异步并发抓取，例如使用：

• httpx
• asyncio
• aiohttp

这样 5 个网页可以同时抓取，响应速度会明显提升。

十四、进阶版：加入向量检索

如果你希望系统不只是“实时搜索网页”，还可以“沉淀知识”，就可以加入向量数据库。

典型流程如下：

搜索网页
   ↓
抓取正文
   ↓
切分文本
   ↓
生成向量
   ↓
存入向量数据库
   ↓
用户提问时检索相关片段
   ↓
大模型生成答案

可选向量数据库包括：

• Chroma
• Milvus
• Qdrant
• Weaviate
• pgvector

这种方式适合构建长期知识库，例如企业竞品库、行业研究库、政策法规库等。

十五、常见问题

Q1：为什么搜索出来的答案不准确？

通常有三个原因：

第一，搜索关键词不够好；第二，网页资料本身质量较低；第三，模型没有严格按照资料回答。

可以优化问题改写 Prompt，增加来源筛选逻辑，并要求模型输出引用。

Q2：可以不用搜索 API，直接爬搜索引擎吗？

不建议直接爬 Google、百度、Bing 的搜索结果页面，因为容易触发反爬，也可能违反服务条款。生产环境建议使用正规搜索 API。

Q3：可以接入本地大模型吗？

可以。只要本地模型提供 OpenAI 兼容接口，就可以直接替换 OPENAI_BASE_URL。例如使用 Ollama、vLLM、LM Studio 等工具。

Q4：如何让系统输出固定格式？

可以在 Prompt 中明确要求输出 JSON、Markdown 或表格。例如：

请按以下JSON格式输出：
{
  "summary": "",
  "key_points": [],
  "sources": []
}

如果用于程序处理，建议使用 JSON 格式，并增加解析失败重试机制。

十六、完整运行流程回顾

最后，我们回顾一下完整步骤：

1. 创建 Python 项目；
2. 安装 FastAPI、requests、BeautifulSoup、openai 等依赖；
3. 配置大模型 API 和搜索 API；
4. 编写搜索模块；
5. 编写网页抓取模块；
6. 编写大模型调用模块；
7. 编写工作流编排逻辑；
8. 通过 FastAPI 暴露接口；
9. 调用接口测试 AI 搜索效果；
10. 根据业务扩展定时任务、消息推送、数据库存储和向量检索。

十七、总结

AI 搜索工作流自动化的核心价值，不是简单地把搜索结果交给大模型总结，而是将“问题理解、信息检索、资料筛选、内容阅读、答案生成、结果分发”串成一条稳定可靠的流程。

本文提供的示例虽然简洁，但已经具备一个 AI 搜索系统的基础形态。你可以在此基础上继续扩展：

• 做一个行业资讯自动简报系统；
• 做一个竞品监控机器人；
• 做一个学术资料搜索助手；
• 做一个企业内部知识问答平台；
• 做一个支持来源引用的 AI 搜索 API。

真正有价值的 AI 应用，往往不是单次对话，而是能融入业务流程、持续自动运行、持续积累数据的系统。AI 搜索工作流正是这种应用形态的典型代表。