Merge pull request #65 from fulai-tech/qlj/dev/0824-somni-audio-service

量产接口: 完善问卷答案与音频目录查询

Author: LuJie3

app/core/config.py

@@ -54,6 +54,8 @@ class Settings(BaseSettings):
     somni_mongo_answers_collection: str = "somni_quiz_answers"
 
     sim_threshold: float = 0.7  # 内容形态向量模糊命中阈值（规范 §五-2）
+    # GetAudio query_text 与根标签向量相似度下限
+    get_audio_root_tag_sim_threshold: float = 0.85
     # 多路文本检索厌恶硬剔除阈值；≥ 该值 penalty=1.0 丢弃候选
     strong_dislike_sim_threshold: float = 0.85
     search_sleep_stage_filter_enabled: bool = True  # 检索步骤 1 是否按睡眠阶段过滤
@@ -93,6 +95,7 @@ class Settings(BaseSettings):
     redis_max_connections: int = 512
     search_cache_max_size: int = 2048
     search_cache_ttl_sec: int = 604800  # 7 天
+    somni_audio_catalog_cache_ttl_sec: float = 60.0
     # CUD 后延时重建睡眠阶段候选缓存，窗口内多次写入只重建一次
     sleep_stage_cache_rewarm_delay_sec: float = 5.0
 

app/es/search.py

@@ -97,9 +97,20 @@ def _candidate_search_body(query: dict[str, Any], *, size: int = 1000) -> dict[s
 class EsSearch:
     """封装检索相关的 ES 查询与文档解析。"""
 
-    def __init__(self, client: AsyncElasticsearch, settings: Settings) -> None:
+    def __init__(
+        self,
+        client: AsyncElasticsearch,
+        settings: Settings,
+        *,
+        audio_index: str | None = None,
+        tag_dictionary_index: str | None = None,
+    ) -> None:
         self._client = client
         self._settings = settings
+        self._audio_index = audio_index or settings.es_audio_index
+        self._tag_dictionary_index = (
+            tag_dictionary_index or settings.es_tag_vectors_index
+        )
         self._content_tag_vectors_cache: list[dict[str, Any]] | None = None
         self._content_tag_vectors_lock = asyncio.Lock()
         # 按 tag_id 缓存 name_vector，避免每请求 mget（内容准入模糊路径）
@@ -108,11 +119,11 @@ def __init__(self, client: AsyncElasticsearch, settings: Settings) -> None:
 
     @property
     def audio_index(self) -> str:
-        return self._settings.es_audio_index
+        return self._audio_index
 
     @property
     def tag_dictionary_index(self) -> str:
-        return self._settings.es_tag_vectors_index
+        return self._tag_dictionary_index
 
     @property
     def tag_vectors_index(self) -> str:
@@ -321,6 +332,7 @@ async def _fetch_content_tag_vectors(self, *, size: int) -> list[dict[str, Any]]
                         "label": label,
                         "dimension": source.get("type", ""),
                         "vector": vector,
+                        "parent_tag_id": str(source.get("parent_tag_id") or ""),
                     }
                 )
         return tags
@@ -380,6 +392,18 @@ def content_tag_ids(tags: AudioTags) -> list[str]:
             ids.extend(item.vector_id for item in dim)
         return ids
 
+    async def list_audio_catalog_docs(self, *, size: int) -> list[dict[str, Any]]:
+        """量产 GetAudio：音频全量（不含 embedding），供内存过滤。"""
+        response = await self._client.search(
+            index=self.audio_index,
+            body={
+                "query": {"match_all": {}},
+                "size": max(1, size),
+                "_source": {"excludes": ["embedding", "description_vector"]},
+            },
+        )
+        return [_document_from_hit(hit) for hit in response["hits"]["hits"]]
+
     async def migrate_legacy_indices(self) -> None:
         """删除旧版 audio_materials / tag_vectors 索引。"""
         for index in LEGACY_INDICES:

app/main.py

@@ -63,6 +63,7 @@ def _bootstrap_dev_entry() -> None:
 from app.server.bootstrap import GrpcServers, start_grpc_servers, stop_grpc_servers
 from app.server.handboard.audio.service import AudioService
 from app.server.handboard.audio.store import MaterialsStore, create_materials_store
+from app.server.somni.audio.catalog import AudioCatalogService as SomniAudioService
 from app.server.somni.quiz.service import QuizService as SomniQuizService
 from app.server.somni.report.service import ReportService as SomniReportService
 from app.services.retrieval import RetrievalService
@@ -73,6 +74,7 @@ def _bootstrap_dev_entry() -> None:
 class AppState:
     settings: Settings
     es_client: AsyncElasticsearch | None = None
+    somni_es_client: AsyncElasticsearch | None = None
     encoder: Encoder | None = None
     materials_store: MaterialsStore | None = None
     somni_mongo_client: AsyncIOMotorClient | None = None
@@ -82,6 +84,7 @@ class AppState:
     audio_service: AudioService | None = None
     somni_quiz_service: SomniQuizService | None = None
     somni_report_service: SomniReportService | None = None
+    somni_audio_service: SomniAudioService | None = None
     search_cache: AudioSearchCache | None = None
     sleep_stage_cache: SleepStageCandidateCache | None = None
     grpc_servers: GrpcServers | None = None
@@ -171,10 +174,27 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         somni_mongo = AsyncIOMotorClient(settings.somni_mongo_uri)
         _app_state.somni_mongo_client = somni_mongo
     else:
-        logger.warning("未配置 SOMNI_MONGO_URI，量产问卷将不可用")
+        logger.warning("未配置 SOMNI_MONGO_URI，量产问卷与音频查询将不可用")
 
     _app_state.somni_quiz_service = SomniQuizService(somni_mongo, settings)
     _app_state.somni_report_service = SomniReportService()
+    somni_es_client = create_es_client(
+        settings,
+        node=settings.effective_somni_es_node,
+    )
+    _app_state.somni_es_client = somni_es_client
+    somni_es_search = EsSearch(
+        somni_es_client,
+        settings,
+        audio_index=settings.somni_es_audio_index,
+        tag_dictionary_index=settings.somni_es_tag_vectors_index,
+    )
+    _app_state.somni_audio_service = SomniAudioService(
+        somni_mongo,
+        settings,
+        es_search=somni_es_search,
+        encoder=encoder,
+    )
 
     start_sync_scheduler(_app_state, settings)
     _app_state.grpc_servers = await start_grpc_servers(_app_state, settings)
@@ -191,6 +211,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         materials_store.close()
     if somni_mongo is not None:
         somni_mongo.close()
+    await somni_es_client.close()
     await es_client.close()
 
 

app/server/bootstrap.py

@@ -6,13 +6,20 @@
 from typing import TYPE_CHECKING
 
 import grpc
+from grpc_reflection.v1alpha import reflection
 from loguru import logger
 
 from app.server.handboard.audio.rpc import AudioRpc as HandboardAudioRpc
 from app.server.handboard.quiz.rpc import QuizRpc as HandboardQuizRpc
+from app.server.somni.audio.rpc import AudioRpc as SomniAudioRpc
 from app.server.somni.quiz.rpc import QuizRpc as SomniQuizRpc
 from app.server.somni.report.rpc import ReportRpc as SomniReportRpc
-from app.uburnode_grpc.grpc_gen import uburnode_pb2_grpc, uburnode_somni_pb2_grpc
+from app.uburnode_grpc.grpc_gen import (
+    uburnode_pb2,
+    uburnode_pb2_grpc,
+    uburnode_somni_pb2,
+    uburnode_somni_pb2_grpc,
+)
 
 if TYPE_CHECKING:
     from app.core.config import Settings
@@ -55,6 +62,7 @@ async def _start_handboard(state: AppState, settings: Settings) -> grpc.aio.Serv
         server,
     )
     uburnode_pb2_grpc.add_QuizServiceServicer_to_server(HandboardQuizRpc(), server)
+    _enable_reflection(server, uburnode_pb2)
     bind = f"{settings.grpc_host}:{settings.grpc_port}"
     _bind(server, bind, "功能手板")
     await server.start()
@@ -71,12 +79,23 @@ async def _start_somni(state: AppState, settings: Settings) -> grpc.aio.Server:
         SomniReportRpc(getattr(state, "somni_report_service", None)),
         server,
     )
+    uburnode_somni_pb2_grpc.add_AudioServiceServicer_to_server(
+        SomniAudioRpc(getattr(state, "somni_audio_service", None)),
+        server,
+    )
+    _enable_reflection(server, uburnode_somni_pb2)
     bind = f"{settings.grpc_host}:{settings.somni_grpc_port}"
     _bind(server, bind, "量产")
     await server.start()
     return server
 
 
+def _enable_reflection(server: grpc.aio.Server, proto_module) -> None:
+    names = [reflection.SERVICE_NAME]
+    names.extend(svc.full_name for svc in proto_module.DESCRIPTOR.services_by_name.values())
+    reflection.enable_server_reflection(tuple(names), server)
+
+
 def _bind(server: grpc.aio.Server, bind: str, label: str) -> None:
     if server.add_insecure_port(bind) == 0:
         raise RuntimeError(f"{label} gRPC 无法绑定 {bind}")