serpbase + Pinecone 向量数据库完整集成

背景

做 RAG 系统需要向量数据库,Pinecone 是最成熟的托管方案。serpbase 提供 SERP 数据做语义搜索,完整集成 5 步搭好。

1. 整体流程

[serpbase 拉 SERP 数据] → [OpenAI Embeddings 转成 1536 维向量] → [存到 Pinecone + metadata] → [用户 query] → [query 转成向量] → [Pinecone 查 top 5] → [LLM 生成答案]

2. 准备

pipinstallopenai pinecone-client requests

importosimportopenaifrompineconeimportPineconeimportrequests OPENAI_KEY=os.environ["OPENAI_API_KEY"]SERPBASE_KEY=os.environ["SERPBASE_KEY"]PINECONE_KEY=os.environ["PINECONE_KEY"]pc=Pinecone(api_key=PINECONE_KEY)openai.api_key=OPENAI_KEY index_name="serpbase-rag"# 创建 index(只需 1 次)ifindex_namenotinpc.list_indexes().names():pc.create_index(name=index_name,dimension=1536,metric="cosine",spec={"serverless":{"cloud":"aws","region":"us-east-1"}})index=pc.Index(index_name)

3. Index SERP 数据

defcall_serpbase(query,gl="us",num=5):r=requests.post("https://api.serpbase.dev/google/search",headers={"X-API-Key":SERPBASE_KEY},json={"q":query,"gl":gl,"hl":"en","num":num},timeout=10,)r.raise_for_status()returnr.json()defget_embedding(text):response=openai.embeddings.create(input=text,model="text-embedding-3-small",)returnresponse.data[0].embeddingdefindex_query_results(query):"""查 SERP + 索引到 Pinecone"""data=call_serpbase(query)vectors=[]fori,iteminenumerate(data.get("organic",[]),1):text=f"{item['title']}.{item.get('snippet','')}"embedding=get_embedding(text)vectors.append((f"{query}-{i}",# idembedding,{"query":query,"rank":i,"title":item["title"],"link":item["link"],"snippet":item.get("snippet",""),"ts":"2026-06-19T09:00:00Z",}))index.upsert(vectors=vectors)returnlen(vectors)# 索引count=index_query_results("SERP API 选型")print(f"Indexed{count}SERP results")

4. 语义搜索

defsemantic_search(query,top_k=5,filter_dict=None):"""语义搜索 SERP 数据"""q_embedding=get_embedding(query)results=index.query(vector=q_embedding,top_k=top_k,include_metadata=True,filter=filter_dict,)return[{"rank":r.metadata["rank"],"title":r.metadata["title"],"link":r.metadata["link"],"snippet":r.metadata["snippet"],"query":r.metadata["query"],"score":r.score,}forrinresults.matches]# 用法 1:语义搜索results=semantic_search("how to monitor SERP API")forrinresults:print(f"{r['title']}({r['score']:.2f})")# 用法 2:filter 检索results=semantic_search("SERP API",filter_dict={"query":"SERP API 选型"}# 只查特定 query)

5. 集成 LLM 生成答案

defanswer_with_rag(query):"""RAG:检索 + 生成"""# 1. 语义搜索sources=semantic_search(query,top_k=3)ifnotsources:return"No relevant data found."# 2. 拼 promptcontext="\n\n".join(f"[{i+1}]{s['title']}\n{s['link']}\n{s['snippet']}"fori,sinenumerate(sources))# 3. LLM 生成response=openai.chat.completions.create(model="gpt-4o",messages=[{"role":"system","content":f"Use these SERP results to answer:\n{context}"},{"role":"user","content":query},],)returnresponse.choices[0].message.content# 用法print(answer_with_rag("SERP API 选型要注意什么?"))

6. 定时同步(更新 Pinecone)

importschedule QUERIES=["SERP API 选型","SERP API 教程","SERP API 评测","SERP API 价格",]defdaily_sync():"""每天同步 SERP 数据到 Pinecone"""forqinQUERIES:count=index_query_results(q)print(f"{q}: indexed{count}")# 删除 30 天前的数据stats=index.describe_index_stats()fornamespace,statinstats.namespaces.items():ifstat["vector_count"]>100000:index.delete(delete_all=True,filter={"ts":{"$lt":"2026-05-01"}},namespace=namespace,)schedule.every().day.at("03:00").do(daily_sync)

7. 5 个工程细节

1. Namespace 区分不同主题

# 不同主题不同 namespaceindex.upsert(vectors=vectors,namespace="seo")index.upsert(vectors=vectors,namespace="ads")# 查询时只查相关 namespaceresults=index.query(vector=q_embedding,namespace="seo",top_k=5)

2. Metadata 索引(只查特定 query)

# 索引时 metadata 字段会被索引index.upsert(vectors=vectors)# 自动索引所有 metadata# 查询时 filterresults=index.query(vector=q_embedding,filter={"rank":{"$lt":5}},# 只查 top 5top_k=10,)

3. 批量 upsert(性能)

defbatch_index(queries,batch_size=100):vectors=[]forqinqueries:data=call_serpbase(q)fori,iteminenumerate(data.get("organic",[]),1):text=f"{item['title']}.{item.get('snippet','')}"vectors.append((f"{q}-{i}",get_embedding(text),{"query":q,"rank":i,"title":item["title"],"link":item["link"],"snippet":item.get("snippet","")}))# 批量 upsert(100 条/批)foriinrange(0,len(vectors),batch_size):batch=vectors[i:i+batch_size]index.upsert(vectors=batch)batch_index(QUERIES)# 100+ 条 1 次 upsert

4. 删除旧数据(节省成本)

# 7 天前数据清理fromdatetimeimportdatetime,timedelta cutoff=(datetime.now()-timedelta(days=7)).isoformat()index.delete(filter={"ts":{"$lt":cutoff}},namespace="seo",)

5. 多 region(不同 index)

# 不同 region 不同 indexforregion,glin[("us","us"),("cn","cn"),("uk","uk")]:pc.create_index(f"serpbase-{region}",dimension=1536,metric="cosine")

8. 实战数据(我项目 30 天)

9. 集成 N8N / Make / Slack

# 通过 webhook 让 Slack / Make 触发# /webhook/answer?query=xxx@app.route("/webhook/answer")defanswer():query=request.args.get("query")answer=answer_with_rag(query)return{"answer":answer}

Slack 机器人用 /answer 触发,RAG 答案直接发回 Slack。

10. 与纯 RAG 方案对比

serpbase + Pinecone 比自建快很多,精度稍低但够用。

小结

serpbase + Pinecone 5 步搭 RAG 系统:

1. 创建 Pinecone index(1 次)
2. 查 serpbase + 转 embedding + 存(每 query)
3. 语义搜索(用户 query)
4. 拼 prompt + LLM 生成
5. 定时同步更新(每天)

月成本 $1(serpbase $0.45 + embedding $0.50 + Pinecone 免费层),比自建向量库便宜 95%。