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())