刘金宝
35 lines
1.3 KiB
Python
35 lines
1.3 KiB
Python
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.models.knowledge import KnowledgeChunk
|
|
|
|
|
|
class KnowledgeRepository:
|
|
"""知识库仓储：负责评分参考指南的轻量检索。"""
|
|
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def search_chunks(
|
|
self,
|
|
department_id: int,
|
|
task_type: str,
|
|
keywords: list[str],
|
|
limit: int = 5,
|
|
) -> list[KnowledgeChunk]:
|
|
"""知识检索：按科室、任务类型和关键词检索知识片段。"""
|
|
stmt = (
|
|
select(KnowledgeChunk)
|
|
.options(selectinload(KnowledgeChunk.document))
|
|
.where(KnowledgeChunk.is_active.is_(True))
|
|
.where(or_(KnowledgeChunk.department_id == department_id, KnowledgeChunk.department_id.is_(None)))
|
|
.where(or_(KnowledgeChunk.task_type == task_type, KnowledgeChunk.task_type.is_(None)))
|
|
)
|
|
|
|
keyword_clauses = [KnowledgeChunk.chunk_text.contains(keyword) for keyword in keywords if keyword]
|
|
if keyword_clauses:
|
|
stmt = stmt.where(or_(*keyword_clauses))
|
|
|
|
stmt = stmt.order_by(KnowledgeChunk.weight.desc(), KnowledgeChunk.id.asc()).limit(limit)
|
|
return list(self.db.scalars(stmt).all())
|