音频检索 Somni: Mongo 差异同步与 somni ES 索引检索适配

将数据源从 comm 扁平表切换为 Mongo Somni 集合，新增 index_mappings 定义 somni 双索引 mapping，同步脚本、检索流水线与配套单测一并更新。

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 2 people

.env.example

@@ -5,8 +5,12 @@ APP_DEBUG=false
 
 # Elasticsearch
 ES_NODE=http://localhost:9200
-ES_AUDIO_INDEX=audio_materials
-ES_TAG_VECTORS_INDEX=tag_vectors
+ES_AUDIO_INDEX=somni_audio_materials
+ES_TAG_VECTORS_INDEX=somni_audio_tag_dictionary
+
+# MongoDB（Somni 数据同步源）
+MONGO_URI=mongodb://user:password@host:27017/Fullive
+MONGO_DB=Fullive
 
 # comm-service gRPC
 COMM_GRPC_HOST=bionode-test.fulai.tech
@@ -16,21 +20,21 @@ COMM_GRPC_USE_TLS=true
 # 检索参数
 SIM_THRESHOLD=0.7
 DEFAULT_TOP_K=10
-# 检索步骤 1 是否按睡眠阶段过滤（暂时关闭可设为 false）
-SEARCH_SLEEP_STAGE_FILTER_ENABLED=false
+# 检索步骤 1 是否按 sleep_stage_tags 做 nested 精确过滤
+SEARCH_SLEEP_STAGE_FILTER_ENABLED=true
 
 # Embedding（bge-small-zh-v1.5，512 维；换模型须重建 ES 并全量重算向量）
 EMBEDDING_BACKEND=onnx
 EMBEDDING_MODEL=BAAI/bge-small-zh-v1.5
 EMBEDDING_DIM=512
 EMBEDDING_ONNX_DIR=models/onnx/bge-small-zh-v1.5
 
-# comm → ES 差异同步（服务内每周定时 + 可手动 scripts/sync_es_from_comm.py）
+# Mongo → ES 差异同步（服务内每周定时 + 可手动 scripts/sync_es_from_comm.py）
 SYNC_ENABLED=true
 SYNC_INTERVAL_DAYS=7
 SYNC_PAGE_SIZE=100
 SYNC_BACKUP_DIR=data/sync_backup
-SYNC_BACKUP_FILENAME=audio_materials_backup.json
+SYNC_BACKUP_FILENAME=somni_audio_materials_backup.json
 
 # 日志（控制台 + logs/uburnode.log 文件双写，每次请求自动记录）
 LOG_LEVEL=INFO

README.md

@@ -1,83 +1,92 @@
-# UburNode
+# UburPython
 
-BioNode 体系中的**中间层服务**：以**音频检索**为核心能力，同时承担算法端与底层存储之间的**数据结构转换**。
+BioNode 体系中的 **Somni 音频检索服务**：以三维度检索为核心，从 MongoDB 同步 Somni 原料与标签词典至 Elasticsearch，对外提供 HTTP 检索 API。
 
-- **核心**：三维度音频检索（ES 召回 + 进程内 Embedding + 四步精排流水线）
-- **中间层**：统一对外 HTTP/Pydantic 契约，对内经 gRPC 访问 comm-service；在 HTTP、Mongo（扁平标签）、ES（六维结构 + 向量）之间做形态互转，避免算法端直连 Mongo 或各自维护多套字段约定
+- **核心**：三维度音频检索（`somni_audio_materials` ES 召回 + 标签词典向量 + 四步精排流水线）
+- **数据源**：MongoDB `Fullive` 库（`somni_audio_materials`、`somni_audio_tag_dictionary`）
+- **索引**：Elasticsearch `somni_audio_materials`、`somni_audio_tag_dictionary`（字段含义见 mapping `meta.description`）
+- **写路径（遗留）**：HTTP CUD 仍经 comm-service gRPC；Somni 索引由同步脚本维护
 
 ## 架构
 
 ```text
 算法端 / 调用方
     │
     ▼
-对外 HTTP (FastAPI + Pydantic)     ← 中间层：契约统一 + 数据结构转换
+对外 HTTP (FastAPI + Pydantic)
     │
-    ├──读──► Elasticsearch + Embedding + RetrievalService
+    ├──读（检索）──► somni_audio_materials (ES)
+    │                  + somni_audio_tag_dictionary (ES 向量)
+    │                  + 进程内 Embedding (bge-small-zh-v1.5)
     │
-    └──写──► comm-service (gRPC) ──► MongoDB（真值库，扁平 tags）
-                  └── EsSync ──► Elasticsearch（索引副本，六维 tags + vector_id）
+    ├──写（CUD，遗留）──► comm-service (gRPC) ──► MongoDB（旧 comm 表）
+    │
+    └──同步（定时/手动）──► MongoDB Somni 集合 ──► Elasticsearch
 ```
 
-## 中间层数据转换
-
-UburNode 不只做检索，还在各存储边界维持**单一对外契约**并完成形态映射：
+## 数据流
 
-| 边界 | 入站形态 | 出站形态 | 负责模块 |
-|------|----------|----------|----------|
-| HTTP 写（CUD） | 六维标签对象 `AudioTagsInput` | comm/Mongo 扁平 `string[]`（带维度前缀） | `app/schemas/audio.py`、`app/core/tags.py` |
-| HTTP 写响应 | comm `AudioMaterialInfo`（gRPC） | HTTP `AudioMaterialData` | `app/schemas/audio.py` |
-| ES 同步 | 扁平 tags + 业务字段 | ES 六维 `tags` + `tag_vectors` embedding | `app/es/sync.py` |
-| HTTP 读（检索） | ES 六维 `TagItem`（含 `vector_id`） | 出参六维 label 字符串 | `app/schemas/audio.py`、`app/services/retrieval.py` |
+| 环节 | 来源 | 目标 | 模块 |
+|------|------|------|------|
+| Mongo → ES 同步 | `somni_audio_materials`、`somni_audio_tag_dictionary` | 同名 ES 索引 | `scripts/sync_es_from_comm.py` |
+| 标签向量 | 词典 `name` / `name_en` | `name_vector` / `name_en_vector` | 同步脚本 + `app/embedding/` |
+| HTTP 检索 | ES 原料文档 | `data.materials[]` 原样返回 | `app/services/retrieval.py` |
+| HTTP CUD（遗留） | 六维标签入参 | comm 扁平 tags | `app/services/audio.py` |
 
-字段命名全链路 **snake_case**；对外以 Pydantic + OpenAPI 为唯一 HTTP 契约，对内 comm 调用走同源 `bionode_comm.proto`。
+字段命名全链路 **snake_case**。Somni 表结构详见仓库内 `音频表结构.md`。
 
 ## 目录结构
 
 ```text
-UburNode/
+UburPython/
 ├── app/
-│   ├── main.py              # FastAPI 入口 + lifespan
-│   ├── core/                # 配置、日志
-│   ├── api/audio.py         # 4 个 HTTP 端点
-│   ├── schemas/audio.py     # Pydantic 模型
-│   ├── services/            # AudioService、RetrievalService
-│   ├── es/                  # EsSearch、EsSync
-│   ├── embedding/encoder.py # bge-small-zh-v1.5 向量编码
-│   ├── bionode_grpc_clients/  # BioNode 外部微服务 gRPC 客户端
-│   │   └── comm/            # comm-service（client.py + grpc_gen/）
-├── proto/                   # bionode_comm.proto（唯一真源）
-├── scripts/gen_proto.sh     # 生成 gRPC stub
+│   ├── main.py                 # FastAPI 入口 + lifespan
+│   ├── core/                   # 配置、日志、标签转换
+│   ├── api/audio.py            # 4 个 HTTP 端点
+│   ├── schemas/audio.py        # Pydantic 模型
+│   ├── services/               # AudioService、RetrievalService
+│   ├── es/
+│   │   ├── search.py           # EsSearch 读路径
+│   │   ├── sync.py             # EsSync 写路径（CUD 跳过 upsert）
+│   │   └── index_mappings.py   # ES 索引 mapping + 字段注释
+│   ├── embedding/encoder.py    # bge-small-zh-v1.5 向量编码
+│   └── bionode_grpc_clients/   # comm-service gRPC 客户端
+├── scripts/
+│   ├── sync_es_from_comm.py    # Mongo → ES 差异同步
+│   └── gen_proto.sh            # 生成 gRPC stub
+├── proto/                      # bionode_comm.proto
 ├── tests/
-├── .cursor/skills/          # 项目 Skill
 ├── pyproject.toml
 └── .env.example
 ```
 
 ## 快速开始
 
 ```bash
-# 1. 创建虚拟环境并安装依赖
-python3.12 -m venv .venv
-source .venv/bin/activate
-pip install -e ".[dev]"
+# 1. 安装依赖（推荐 uv）
+uv sync --extra dev
 
-# 2. 生成 comm gRPC stub
+# 2. 生成 comm gRPC stub（CUD 接口需要）
 chmod +x scripts/gen_proto.sh
 ./scripts/gen_proto.sh
 
-# 3. 本地 Elasticsearch（向量索引，需先就绪）
-# 未安装 Docker 时：brew install --cask docker-desktop，打开 Docker Desktop 等待就绪
+# 3. 本地 Elasticsearch
 docker compose -f docker-compose.es.yml up -d
-curl -s http://localhost:9200   # 应返回 cluster 信息
-# .env 默认 ES_NODE=http://localhost:9200；索引由应用启动时 ensure_indices 自动创建
+curl -s http://localhost:9200
 
 # 4. 配置环境变量
 cp .env.example .env
-# 编辑 ES_NODE、COMM_GRPC_HOST 等
+# 编辑 ES_NODE、MONGO_URI、EMBEDDING_ONNX_DIR、COMM_GRPC_* 等
+
+# 5. 导出 ONNX 模型（若 models/ 目录尚无模型）
+# 见 scripts/export_onnx_model.py
+
+# 6. Mongo → ES 全量同步
+uv run python scripts/sync_es_from_comm.py --dry-run
+uv run python scripts/sync_es_from_comm.py
 
-# 5. 启动服务
-uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
+# 7. 启动服务
+uv run uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
 ```
 
 开发模式（`APP_DEBUG=true`）跳过 Embedding 模型加载，便于本地调试 HTTP 路由。
@@ -86,20 +95,56 @@ uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
 
 | 端点 | 方法 | 说明 |
 |------|------|------|
-| `/api/audio` | POST | 创建音频并同步 ES |
-| `/api/audio/{id}` | PUT | 更新音频并同步 ES |
-| `/api/audio/{id}` | DELETE | 删除音频并同步 ES |
+| `/api/audio` | POST | 创建音频（comm gRPC，遗留） |
+| `/api/audio/{id}` | PUT | 更新音频（comm gRPC，遗留） |
+| `/api/audio/{id}` | DELETE | 删除音频 |
 | `/api/audio/search` | POST | 三维度检索 |
 
 OpenAPI 文档：启动后访问 `http://localhost:8080/docs`。
 
-### comm-service gRPC 连通探测
+### 检索接口
 
-```bash
-source .venv/bin/activate   # 或: .venv/bin/python scripts/test_grpc_connect.py
-python scripts/test_grpc_connect.py
-# 或集成测试（需可达的 COMM_GRPC_HOST）
-COMM_GRPC_INTEGRATION=1 pytest tests/test_comm_grpc.py -v
+**请求** `POST /api/audio/search`：
+
+```json
+{
+  "sleep_stage_tags": ["放松"],
+  "content_tags": ["慢钢琴"],
+  "disliked_tags": [],
+  "top_k": 10
+}
+```
+
+**响应** `data.materials` 为命中条目的 `somni_audio_materials` 索引文档（含 `id`，字段与 ES/Mongo 一致，暂不做裁剪）：
+
+```json
+{
+  "code": 200,
+  "msg": "检索成功",
+  "data": {
+    "materials": [
+      {
+        "id": "6a33a7928030d4cf420efeb6",
+        "audio_name": "专属冥想南极 助眠解压舒缓情绪",
+        "description": "...",
+        "status": true,
+        "audio_url": "https://cdn.fulai.tech/comm/audio/xxx.mp3",
+        "operation_type": 0,
+        "created_by": "qwen3.5-omni-plus",
+        "updated_by": "qwen3.5-omni-plus",
+        "sleep_stage_tags": [{ "tag_id": "...", "code": "unwind", "name": "放松" }],
+        "content_form_tags": [],
+        "mechanism_tags": [],
+        "audio_engineering_tags": [],
+        "medical_risk_tags": [],
+        "evidence_level_tags": [{ "tag_id": "...", "code": "B", "name": "中等证据" }],
+        "created_at": "2026-06-18T00:00:00.000Z",
+        "updated_at": "2026-06-18T00:00:00.000Z"
+      }
+    ]
+  },
+  "timestamp": "..."
+}
 ```
 
 ## 检索流水线
@@ -108,17 +153,41 @@ COMM_GRPC_INTEGRATION=1 pytest tests/test_comm_grpc.py -v
 睡眠阶段精确过滤 → 内容形态准入 → 厌恶剔除 + 粗排 → 精排
 ```
 
-## Proto 变更
+| 步骤 | 说明 |
+|------|------|
+| 1 | `sleep_stage_tags.name` nested 精确匹配（可配置跳过） |
+| 2 | `content_tags` 与内容/机制/工程标签精确或向量模糊命中 |
+| 3 | `disliked_tags` 向量相似则剔除 |
+| 4 | 按 `match_count` 降序，`top_k` 截断 |
 
-`comm-service` 修改 `proto/bionode_comm.proto` 后须重新生成 stub：
+## Mongo → ES 同步
 
 ```bash
-./scripts/gen_proto.sh
+uv run python scripts/sync_es_from_comm.py          # 正式同步
+uv run python scripts/sync_es_from_comm.py --dry-run # 仅比对统计
 ```
 
-## 日志
+- 先同步 `somni_audio_tag_dictionary`（写入 `name_vector`、`name_en_vector`）
+- 再同步 `somni_audio_materials`（1:1 镜像 Mongo 文档）
+- 启动时删除旧索引 `audio_materials`、`tag_vectors`
+
+服务内按 `SYNC_INTERVAL_DAYS` 定时执行（需配置 `MONGO_URI`）。
+
+## 环境变量（节选）
 
-每次 HTTP 请求自动写入日志文件（`RequestLogMiddleware`），包含 method、path、status、耗时、`request_id`。
+| 变量 | 默认值 | 说明 |
+|------|--------|------|
+| `ES_NODE` | `http://localhost:9200` | Elasticsearch 地址 |
+| `ES_AUDIO_INDEX` | `somni_audio_materials` | 原料索引名 |
+| `ES_TAG_VECTORS_INDEX` | `somni_audio_tag_dictionary` | 标签词典索引名 |
+| `MONGO_URI` | — | MongoDB 连接串（同步必填） |
+| `MONGO_DB` | `Fullive` | 数据库名 |
+| `SIM_THRESHOLD` | `0.7` | 向量模糊命中阈值 |
+| `EMBEDDING_ONNX_DIR` | `models/onnx/bge-small-zh-v1.5` | ONNX 模型目录 |
+
+完整列表见 [`.env.example`](.env.example)。
+
+## 日志
 
 | 配置项 | 默认值 | 说明 |
 |--------|--------|------|
@@ -127,36 +196,18 @@ COMM_GRPC_INTEGRATION=1 pytest tests/test_comm_grpc.py -v
 | `LOG_ROTATION` | `10 MB` | 单文件滚动大小 |
 | `LOG_RETENTION` | `7 days` | 历史日志保留 |
 
-日志同时输出到控制台和 `logs/uburnode.log`。响应头会回传 `X-Request-Id` 便于链路追踪。
-
-## Docker 部署（服务器）
+响应头回传 `X-Request-Id` 便于链路追踪。
 
-与 `intelligent_reimbursement` 相同：**服务器 git pull + docker compose build**，不依赖 GHCR。
+## Docker 部署
 
 ```bash
-# 1. 本机一键写入 GitHub Secrets（SSH_HOST / SSH_USER / SSH_PRIVATE_KEY）
-chmod +x scripts/setup_github_secrets.sh
-./scripts/setup_github_secrets.sh
-
-# 2. 服务器一次性准备
-#    - 将公钥写入 ~/.ssh/authorized_keys
-#    - 已安装 Docker 与 Compose；/etc/docker/daemon.json 镜像加速自行维护
-#    - 克隆仓库并配置 .env：
-mkdir -p /opt/uburnode
-git clone -b dev https://github.com/dwqnidq/UburNode.git /opt/uburnode
-cp /opt/uburnode/.env.example /opt/uburnode/.env   # 编辑 COMM_GRPC_* 等
-
-# 3. 首次手动启动（build 约 20～40 分钟）
-cd /opt/uburnode && docker compose up -d --build
-
-# 4. 以后：GitHub → Actions → Deploy UburNode → Run workflow
-#    或 push 到 dev 分支自动部署（git pull + compose up --build）
+cd /opt/uburpython && docker compose up -d --build
 ```
 
-生产访问：`http://<服务器IP>:8001/docs`（nginx 映射宿主机 8001 → 容器 80，避免与宝塔 80 冲突）。
+生产访问：`http://<服务器IP>:8001/docs`（nginx 映射宿主机 8001 → 容器 80）。
 
 ## 测试
 
 ```bash
-pytest
+uv run pytest
 ```

app/api/audio.py

@@ -64,6 +64,6 @@ async def search_audio(
     body: SearchAudioRequest,
     service: AudioService = Depends(get_audio_service),
 ) -> ApiResponse:
-    """三维度检索：只读 ES，不写 Mongo / ES（规范红线）。"""
+    """三维度检索：只读 ES，返回 somni_audio_materials 索引文档列表。"""
     result = await service.search_audio(body)
     return success(data=result.model_dump(), msg="检索成功")

app/core/config.py

@@ -21,15 +21,20 @@ class Settings(BaseSettings):
     app_debug: bool = False
 
     es_node: str = "http://localhost:9200"
-    es_audio_index: str = "audio_materials"
-    es_tag_vectors_index: str = "tag_vectors"
+    es_audio_index: str = "somni_audio_materials"
+    es_tag_vectors_index: str = "somni_audio_tag_dictionary"
+
+    mongo_uri: str = ""
+    mongo_db: str = "Fullive"
+    mongo_materials_collection: str = "somni_audio_materials"
+    mongo_tag_dictionary_collection: str = "somni_audio_tag_dictionary"
 
     comm_grpc_host: str = "bionode-test.fulai.tech"
     comm_grpc_port: int = 443
     comm_grpc_use_tls: bool = True  # 443 走 TLS；内网明文可设 false
 
     sim_threshold: float = 0.7  # 内容形态向量模糊命中阈值（规范 §五-2）
-    search_sleep_stage_filter_enabled: bool = False  # 检索步骤 1 是否按睡眠阶段过滤
+    search_sleep_stage_filter_enabled: bool = True  # 检索步骤 1 是否按睡眠阶段过滤
 
     embedding_backend: str = "onnx"  # onnx（生产）| torch（对比/回退，需 sentence-transformers）
     embedding_model: str = "BAAI/bge-small-zh-v1.5"
@@ -42,12 +47,13 @@ class Settings(BaseSettings):
     log_rotation: str = "10 MB"
     log_retention: str = "7 days"
 
-    # comm → ES 差异同步（服务内定时 + scripts/sync_es_from_comm.py 手动）
+    # Mongo → ES 差异同步（服务内定时 + scripts/sync_es_from_comm.py 手动）
     sync_enabled: bool = True
     sync_interval_days: int = 7
     sync_page_size: int = 100
     sync_backup_dir: str = "data/sync_backup"
-    sync_backup_filename: str = "audio_materials_backup.json"
+    sync_backup_filename: str = "somni_audio_materials_backup.json"
+    sync_tag_dictionary_backup_filename: str = "somni_audio_tag_dictionary_backup.json"
 
     @property
     def embedding_onnx_path(self) -> Path:
@@ -61,6 +67,10 @@ def embedding_tokenizer_dir(self) -> Path:
     def sync_backup_path(self) -> Path:
         return Path(self.sync_backup_dir) / self.sync_backup_filename
 
+    @property
+    def sync_tag_dictionary_backup_path(self) -> Path:
+        return Path(self.sync_backup_dir) / self.sync_tag_dictionary_backup_filename
+
     @property
     def comm_grpc_target(self) -> str:
         return f"{self.comm_grpc_host}:{self.comm_grpc_port}"