feat: 知识库朴素语义搜索(输入→检索→结果列表)

This commit is contained in:
simonkoson
2026-05-27 19:25:41 +08:00
parent 0c7d2d7400
commit 783e212bb1
4 changed files with 193 additions and 8 deletions
+27 -1
View File
@@ -1,8 +1,9 @@
"""
知识库 API — 上传 / 列表 / 删除 / 来源筛选
知识库 API — 上传 / 列表 / 删除 / 来源筛选 / 语义搜索
"""
from typing import Optional
from pydantic import BaseModel
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query, status
from sqlmodel import Session
@@ -104,3 +105,28 @@ def get_grouped_knowledge_items(
"""
svc = KnowledgeService()
return svc.get_grouped_items()
class SearchRequest(BaseModel):
query: str
top_k: int = 5
@router.post("/search")
def search_knowledge(
body: SearchRequest,
session: Session = Depends(get_session),
current_user: User = Depends(require_role(UserRole.zhipianren, UserRole.zebian, UserRole.biandao)),
):
"""
语义检索:输入一段文字,返回最相关的知识库条目及相似度。
查询向量用 type="query"(区分于存入时的 type="db")。
三角色均可读。
"""
svc = KnowledgeService()
results = svc.search_similar(query_text=body.query, top_k=body.top_k)
return {
"results": results,
"query": body.query,
"count": len(results),
}
+21 -5
View File
@@ -232,7 +232,10 @@ class KnowledgeService:
def search_similar(self, query_text: str, top_k: int = 5) -> list[dict]:
"""
语义检索:查询句转为向量,用 SQL 余弦距离(<=>)在数据库层检索
返回 top_k 条相似笔记,含相似度分数
返回 top_k 条相似笔记,含相似度分数 + 原文片段(SQL 端截断前 200 字)。
注意:当前取前 200 字是已知妥协(整篇向量检索无法定位中段命中点),
Phase 4a 做切块检索(chunk)时可优化为取最相关片段。
"""
query_vector = self.embedder.embed_single(query_text, embed_type="query")
vec_str = "[" + ",".join(str(v) for v in query_vector) + "]"
@@ -243,6 +246,9 @@ class KnowledgeService:
ki.id,
ki.title,
ki.source_type,
ki.author,
ki.tags,
SUBSTRING(ki.content_md, 1, 200) AS snippet,
1 - (ke.embedding <=> '{vec_str}'::vector) AS similarity
FROM knowledge_embeddings ke
JOIN knowledge_items ki ON ke.knowledge_id = ki.id
@@ -252,10 +258,20 @@ class KnowledgeService:
"""
stmt = text(sql)
rows = session.execute(stmt).all()
return [
{"id": r.id, "title": r.title, "source_type": r.source_type, "similarity": round(r.similarity, 4)}
for r in rows
]
results = []
for r in rows:
tags = r.tags or {}
source_detail = tags.get("source_detail") if isinstance(tags, dict) else None
results.append({
"id": r.id,
"title": r.title,
"source_type": r.source_type,
"author": r.author,
"source_detail": source_detail,
"snippet": r.snippet,
"similarity": round(r.similarity, 4),
})
return results
def get_item_count(self) -> int:
with Session(engine) as session: