「Mac畅玩AIGC与多模态40」开发篇35 - 用 Python 开发服务对接 SearxNG 与本地知识库

一、概述

本篇介绍如何使用 Python 构建一个集成本地聚合搜索引擎 SearxNG 与本地知识库的双通道服务接口，返回标准结构化 JSON 数据，并用于对接智能体插件系统。该接口适用于本地 Agent 应用开发与 Dify 插件集成场景。

二、目标说明

使用 Flask 实现搜索服务端接口
/search_web 路由封装本地 SearxNG 请求
/search_kb 路由封装本地知识库检索接口
/search 路由整合聚合结果并返回统一格式
返回格式包含 title、content、url 字段

三、目录结构

search_service/
├── app.py               # 主服务入口
├── kb_docs.json         # 本地知识库内容
├── kb_index.faiss       # FAISS 索引文件（由构建脚本生成）
├── build_index.py       # 索引构建脚本
└── requirements.txt     # Python 依赖清单

四、安装依赖

pip install flask requests faiss-cpu sentence-transformers

或者使用 requirements.txt 安装：

requirements.txt

flask
requests
sentence-transformers
faiss-cpu

五、准备知识库文件（kb_docs.json）

[{"title": "HarmonyOS 系统概览","content": "HarmonyOS 是一个面向全场景的分布式操作系统，适用于智能手机、智能家居、车载终端等多设备融合。","url": "https://example.com/harmonyos-overview"},{"title": "HarmonyOS 的微内核架构","content": "HarmonyOS 使用微内核架构，提高系统安全性和模块独立性，支持硬件隔离和实时调度。","url": "https://example.com/harmonyos-kernel"},{"title": "鸿蒙系统与 Android 的区别","content": "HarmonyOS 与 Android 最大的区别在于系统底层架构和跨设备能力，前者强调分布式协同。","url": "https://example.com/harmonyos-vs-android"},{"title": "分布式任务调度机制","content": "HarmonyOS 支持多设备之间的任务迁移与协同执行，实现一次开发、多端部署。","url": "https://example.com/harmonyos-scheduler"},{"title": "HarmonyOS 对开发者的影响","content": "该系统为开发者提供统一 IDE、统一语言和一次编写多端运行的开发体验。","url": "https://example.com/harmonyos-developers"}
]

六、构建向量索引（build_index.py）

from sentence_transformers import SentenceTransformer
import faiss
import json
import numpy as np# 加载文档
with open("kb_docs.json", "r", encoding="utf-8") as f:docs = json.load(f)# 生成 embeddings
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
corpus = [doc["content"] for doc in docs]
embeddings = model.encode(corpus, convert_to_numpy=True)# 构建并保存索引
index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(np.array(embeddings))
faiss.write_index(index, "kb_index.faiss")

七、主服务接口实现（app.py）

from flask import Flask, request, jsonify
import requests
import json
import faiss
import numpy as np
from sentence_transformers import SentenceTransformerapp = Flask(__name__)# 初始化知识库模型和索引
model = SentenceTransformer('paraphrase-MiniLM-L6-v2')
index = faiss.read_index("kb_index.faiss")
with open("kb_docs.json", "r", encoding="utf-8") as f:kb_docs = json.load(f)@app.route("/search_web", methods=["GET"])
def search_web():q = request.args.get("q", "")if not q:return jsonify({"error": "Missing query"}), 400try:resp = requests.get("http://localhost:8090/search", params={"q": q, "format": "json"})raw = resp.json()results = []for item in raw.get("results", [])[:3]:results.append({"title": item.get("title", ""),"content": item.get("content", ""),"url": item.get("url", "")})return jsonify({"results": results})except Exception as e:return jsonify({"error": str(e)}), 500@app.route("/search_kb", methods=["GET"])
def search_kb():q = request.args.get("q", "").strip()if not q:return jsonify({"results": [], "error": "Missing query"})try:vec = model.encode([q])D, I = index.search(np.array(vec), k=3)results = []for idx in I[0]:doc = kb_docs[idx]results.append({"title": doc.get("title", "Untitled"),"content": doc.get("content", ""),"url": doc.get("url", "")})return jsonify({"results": results})except Exception as e:return jsonify({"results": [], "error": str(e)})@app.route("/search", methods=["GET"])
def search_combined():q = request.args.get("q", "")if not q:return jsonify({"error": "Missing query"}), 400try:from concurrent.futures import ThreadPoolExecutordef get_web():resp = requests.get("http://localhost:5001/search_web", params={"q": q})return resp.json().get("results", [])def get_kb():resp = requests.get("http://localhost:5001/search_kb", params={"q": q})return resp.json().get("results", [])with ThreadPoolExecutor() as pool:web_future = pool.submit(get_web)kb_future = pool.submit(get_kb)web_results = web_future.result()kb_results = kb_future.result()return jsonify({"results": web_results + kb_results})except Exception as e:return jsonify({"error": str(e)}), 500if __name__ == "__main__":app.run(host="0.0.0.0", port=5001)

八、启动服务

python app.py

九、接口测试示例

curl "http://localhost:5001/search?q=HarmonyOS"

返回格式：

{"results": [{ "title": "HarmonyOS 系统概览", "content": "...", "url": "..." },{ "title": "HarmonyOS 的微内核架构", "content": "...", "url": "..." },{ "title": "SearxNG 聚合结果", "content": "...", "url": "..." }]
}