embedding模块代码整理

tangwang
1 parent 5bac9649
Showing 17 changed files with 54 additions and 99 deletions Show diff stats
config/__init__.py
config/config.yaml
config/env_config.py
config/services_config.py
docs/DEVELOPER_GUIDE.md
docs/QUICKSTART.md
docs/搜索API对接指南.md
embeddings/README.md
embeddings/clip_as_service_encoder.py
embeddings/clip_model.py
embeddings/config.py
embeddings/image_encoder.py
embeddings/protocols.py
embeddings/text_encoder.py
scripts/service_ctl.sh
scripts/start_embedding_service.sh
scripts/trace_indexer_calls.sh
@@ -26,7 +26,6 @@ from .services_config import (
     get_embedding_backend_config,
     get_rerank_backend_config,
     get_translation_base_url,
-    get_embedding_base_url,
     get_embedding_text_base_url,
     get_embedding_image_base_url,
     get_rerank_service_url,
@@ -54,7 +53,6 @@ __all__ = [
     'get_embedding_backend_config',
     'get_rerank_backend_config',
     'get_translation_base_url',
-    'get_embedding_base_url',
     'get_embedding_text_base_url',
     'get_embedding_image_base_url',
     'get_rerank_service_url',
@@ -195,10 +195,8 @@ services:
         use_cache: true
   embedding:
     provider: "http"  # http
-    base_url: "http://127.0.0.1:6005"
     providers:
       http:
-        base_url: "http://127.0.0.1:6005"
         text_base_url: "http://127.0.0.1:6005"
         image_base_url: "http://127.0.0.1:6008"
     # 服务内文本后端（embedding 进程启动时读取）
@@ -59,11 +59,12 @@ API_PORT = int(os.getenv(&#39;API_PORT&#39;, 6002))
 INDEXER_HOST = os.getenv('INDEXER_HOST', '0.0.0.0')
 INDEXER_PORT = int(os.getenv('INDEXER_PORT', 6004))
 # Optional dependent services
+# EMBEDDING_HOST / EMBEDDING_PORT are only used by the optional combined embedding mode.
 EMBEDDING_HOST = os.getenv('EMBEDDING_HOST', '127.0.0.1')
 EMBEDDING_PORT = int(os.getenv('EMBEDDING_PORT', 6005))
-EMBEDDING_TEXT_HOST = os.getenv('EMBEDDING_TEXT_HOST', EMBEDDING_HOST)
-EMBEDDING_TEXT_PORT = int(os.getenv('EMBEDDING_TEXT_PORT', EMBEDDING_PORT))
-EMBEDDING_IMAGE_HOST = os.getenv('EMBEDDING_IMAGE_HOST', EMBEDDING_HOST)
+EMBEDDING_TEXT_HOST = os.getenv('EMBEDDING_TEXT_HOST', '127.0.0.1')
+EMBEDDING_TEXT_PORT = int(os.getenv('EMBEDDING_TEXT_PORT', 6005))
+EMBEDDING_IMAGE_HOST = os.getenv('EMBEDDING_IMAGE_HOST', '127.0.0.1')
 EMBEDDING_IMAGE_PORT = int(os.getenv('EMBEDDING_IMAGE_PORT', 6008))
 TRANSLATION_HOST = os.getenv('TRANSLATION_HOST', '127.0.0.1')
 TRANSLATION_PORT = int(os.getenv('TRANSLATION_PORT', 6006))
@@ -77,7 +78,6 @@ if not API_BASE_URL:
 INDEXER_BASE_URL = os.getenv('INDEXER_BASE_URL') or (
     f'http://localhost:{INDEXER_PORT}' if INDEXER_HOST == '0.0.0.0' else f'http://{INDEXER_HOST}:{INDEXER_PORT}'
 )
-EMBEDDING_SERVICE_URL = os.getenv('EMBEDDING_SERVICE_URL') or f'http://{EMBEDDING_HOST}:{EMBEDDING_PORT}'
 EMBEDDING_TEXT_SERVICE_URL = os.getenv('EMBEDDING_TEXT_SERVICE_URL') or (
     f'http://{EMBEDDING_TEXT_HOST}:{EMBEDDING_TEXT_PORT}'
 )
@@ -78,18 +78,20 @@ def _resolve_embedding() -&gt; ServiceConfig:
     if provider != "http":
         raise ValueError(f"Unsupported embedding provider: {provider}")
  
-    env_url = os.getenv("EMBEDDING_SERVICE_URL")
     env_text_url = os.getenv("EMBEDDING_TEXT_SERVICE_URL")
     env_image_url = os.getenv("EMBEDDING_IMAGE_SERVICE_URL")
-    if (env_url or env_text_url or env_image_url) and provider == "http":
+    if provider == "http":
         providers = dict(providers)
-        providers["http"] = dict(providers.get("http", {}))
-        if env_url:
-            providers["http"]["base_url"] = env_url.rstrip("/")
+        http_cfg = dict(providers.get("http", {}))
         if env_text_url:
-            providers["http"]["text_base_url"] = env_text_url.rstrip("/")
+            http_cfg["text_base_url"] = env_text_url.rstrip("/")
         if env_image_url:
-            providers["http"]["image_base_url"] = env_image_url.rstrip("/")
+            http_cfg["image_base_url"] = env_image_url.rstrip("/")
+        if not http_cfg.get("text_base_url"):
+            raise ValueError("services.embedding.providers.http.text_base_url is required")
+        if not http_cfg.get("image_base_url"):
+            raise ValueError("services.embedding.providers.http.image_base_url is required")
+        providers["http"] = http_cfg
  
     return ServiceConfig(provider=provider, providers=providers)
  
@@ -171,27 +173,9 @@ def get_translation_cache_config() -&gt; Dict[str, Any]:
     return get_translation_cache(get_translation_config())
  
  
-def get_embedding_base_url() -> str:
-    provider_cfg = get_embedding_config().providers.get("http", {})
-    base = (
-        os.getenv("EMBEDDING_SERVICE_URL")
-        or provider_cfg.get("base_url")
-        or provider_cfg.get("text_base_url")
-        or provider_cfg.get("image_base_url")
-    )
-    if not base:
-        raise ValueError("Embedding HTTP base_url is not configured")
-    return str(base).rstrip("/")
-
-
 def get_embedding_text_base_url() -> str:
     provider_cfg = get_embedding_config().providers.get("http", {})
-    base = (
-        os.getenv("EMBEDDING_TEXT_SERVICE_URL")
-        or provider_cfg.get("text_base_url")
-        or os.getenv("EMBEDDING_SERVICE_URL")
-        or provider_cfg.get("base_url")
-    )
+    base = os.getenv("EMBEDDING_TEXT_SERVICE_URL") or provider_cfg.get("text_base_url")
     if not base:
         raise ValueError("Embedding text HTTP base_url is not configured")
     return str(base).rstrip("/")
@@ -199,12 +183,7 @@ def get_embedding_text_base_url() -&gt; str:
  
 def get_embedding_image_base_url() -> str:
     provider_cfg = get_embedding_config().providers.get("http", {})
-    base = (
-        os.getenv("EMBEDDING_IMAGE_SERVICE_URL")
-        or provider_cfg.get("image_base_url")
-        or os.getenv("EMBEDDING_SERVICE_URL")
-        or provider_cfg.get("base_url")
-    )
+    base = os.getenv("EMBEDDING_IMAGE_SERVICE_URL") or provider_cfg.get("image_base_url")
     if not base:
         raise ValueError("Embedding image HTTP base_url is not configured")
     return str(base).rstrip("/")
@@ -293,7 +293,7 @@ services:
  
 ### 6.3 环境变量（常用）
  
-- 能力 URL：`EMBEDDING_SERVICE_URL`、`RERANKER_SERVICE_URL`
+- 能力 URL：`EMBEDDING_TEXT_SERVICE_URL`、`EMBEDDING_IMAGE_SERVICE_URL`、`RERANKER_SERVICE_URL`
 - 能力选择：`EMBEDDING_PROVIDER`、`EMBEDDING_BACKEND`、`RERANK_PROVIDER`、`RERANK_BACKEND`
 - 翻译服务行为：统一查看 `config/config.yaml -> services.translation`
 - 环境与索引：`ES_HOST`、`ES_INDEX_NAMESPACE`、`RUNTIME_ENV`、DB 与 Redis 等
@@ -346,14 +346,14 @@ services:
  
 **重排后端协议（服务内）**：所有在 reranker 服务内加载的后端须实现 `score_with_meta(query, docs, normalize=True) -> (scores: List[float], meta: dict)`。返回的 `scores[i]` 与 `docs[i]` 一一对应；meta 至少含 `input_docs`、`usable_docs`、`elapsed_ms` 等。对外 HTTP 契约固定：`POST /rerank` 请求体 `{ "query": str, "docs": [str] }`，响应体 `{ "scores": [float], "meta": object }`；`GET /health` 返回 `status`、`model`、`backend` 等。
  
-**向量化后端协议（服务内）**：文本后端需支持 `encode(sentences: Union[str, List[str]], batch_size, device) -> ndarray | List[ndarray]`，单条与批量输入统一通过一个接口处理；图片后端实现 `embeddings/protocols.ImageEncoderProtocol`：`encode_image_urls(urls, batch_size) -> List[Optional[ndarray]]`，与 urls 等长。
+**向量化后端协议（服务内）**：文本后端需支持 `encode(sentences: Union[str, List[str]], batch_size, device) -> ndarray | List[ndarray]`，单条与批量输入统一通过一个接口处理；图片后端实现 `embeddings/protocols.ImageEncoderProtocol`：`encode_image_urls(urls, batch_size) -> List[ndarray]`，与 urls 等长，异常直接抛出。
  
 **配置速查**：
  
 | 层次 | 配置键 | 重排 | 向量化 |
 |------|--------|------|--------|
 | 调用方 | `services.<capability>.provider` | http | http |
-| 调用方 | `services.<capability>.providers.http.base_url` | 6007 | 6005 |
+| 调用方 | `services.<capability>.providers.http.*_base_url` | 6007 | text=6005 / image=6008 |
 | 服务内 | `services.<capability>.backend` | qwen3_vllm / qwen3_transformers / bge / dashscope_rerank | tei / local_st |
 | 服务内 | `services.<capability>.backends.<name>` | 模型名、batch、vLLM 参数 | 模型名、device 等 |
  
@@ -370,7 +370,7 @@ services:
  
 - **单一路径**：Provider 和 backend 必须由 `config/config.yaml` 的 `services` 块显式指定；未知配置应直接报错。
 - **无兼容回退**：不保留“旧配置自动推导/兜底默认值”机制，避免静默行为偏差。
-- **环境变量覆盖**：允许环境变量覆盖（如 `RERANKER_SERVICE_URL`、`RERANK_BACKEND`、`RERANK_DASHSCOPE_API_KEY_CN`/`RERANK_DASHSCOPE_API_KEY_US`、`RERANK_DASHSCOPE_ENDPOINT`、`EMBEDDING_SERVICE_URL`、`EMBEDDING_BACKEND`、`TEI_BASE_URL`），但覆盖后仍需满足合法性校验。
+- **环境变量覆盖**：允许环境变量覆盖（如 `RERANKER_SERVICE_URL`、`RERANK_BACKEND`、`RERANK_DASHSCOPE_API_KEY_CN`/`RERANK_DASHSCOPE_API_KEY_US`、`RERANK_DASHSCOPE_ENDPOINT`、`EMBEDDING_TEXT_SERVICE_URL`、`EMBEDDING_IMAGE_SERVICE_URL`、`EMBEDDING_BACKEND`、`TEI_BASE_URL`），但覆盖后仍需满足合法性校验。
  
 ---
  
@@ -422,7 +422,7 @@ services:
     provider: "http"
     backend: "tei"
     providers:
-      http: { base_url: "http://127.0.0.1:6005" }
+      http: { text_base_url: "http://127.0.0.1:6005", image_base_url: "http://127.0.0.1:6008" }
     backends:
       tei: { base_url: "http://127.0.0.1:8080", timeout_sec: 60, model_id: "Qwen/Qwen3-Embedding-0.6B" }
   rerank:
@@ -434,7 +434,8 @@ services:
  
 环境变量覆盖（优先级更高）：
  
-- `EMBEDDING_SERVICE_URL`
+- `EMBEDDING_TEXT_SERVICE_URL`
+- `EMBEDDING_IMAGE_SERVICE_URL`
 - `EMBEDDING_BACKEND`
 - `TEI_BASE_URL`
 - `RERANKER_SERVICE_URL`
@@ -1707,7 +1707,6 @@ curl -X POST &quot;http://localhost:6004/indexer/enrich-content&quot; \
 - **启动**:
   - 文本：`./scripts/start_embedding_text_service.sh`
   - 图片：`./scripts/start_embedding_image_service.sh`
-  - 兼容入口：`./scripts/start_embedding_service.sh text|image|all`
 - **依赖**:
   - 文本向量后端默认走 TEI（`http://127.0.0.1:8080`）
   - 图片向量依赖 `cnclip`（`grpc://127.0.0.1:51000`）
@@ -16,7 +16,7 @@
 - **统一配置**：`config.py`
 - **接口契约**：`protocols.ImageEncoderProtocol`（图片编码统一为 `encode_image_urls(urls, batch_size, normalize_embeddings)`，本地 CN-CLIP 与 clip-as-service 均实现该接口）
  
-说明：历史上的云端 embedding 试验实现（DashScope）已从主仓库移除。当前默认部署为**文本服务 6005** 与**图片服务 6008** 两条链路，也兼容 `all` 模式单进程启动。
+说明：历史上的云端 embedding 试验实现（DashScope）已从主仓库移除。当前默认部署为**文本服务 6005** 与**图片服务 6008** 两条独立链路；`all` 模式仅作为单进程调试入口。
  
 ### 文本向量后端（默认）
  
@@ -29,22 +29,16 @@
  
 - 文本服务（默认 `6005`）
   - `POST /embed/text`
-  - `GET /health`
-  - `GET /ready`
+  - 请求体：`["文本1", "文本2", ...]`
+  - 可选 query 参数：`normalize=true|false`
+  - 返回：`[[...], [...], ...]`
+  - 健康接口：`GET /health`、`GET /ready`
 - 图片服务（默认 `6008`）
   - `POST /embed/image`
-  - `GET /health`
-  - `GET /ready`
-
-- `POST /embed/text`
-  - 入参：`["文本1", "文本2", ...]`
-  - 可选 query 参数：`normalize=true|false`（不传则使用服务端默认）
-  - 出参：`[[...], [...], ...]`（与输入按 index 对齐，失败直接报错）
-
-- `POST /embed/image`
-  - 入参：`["url或本地路径1", ...]`
-  - 可选 query 参数：`normalize=true|false`（不传则使用服务端默认）
-  - 出参：`[[...], [...], ...]`（与输入按 index 对齐，失败直接报错）
+  - 请求体：`["url或本地路径1", ...]`
+  - 可选 query 参数：`normalize=true|false`
+  - 返回：`[[...], [...], ...]`
+  - 健康接口：`GET /health`、`GET /ready`
  
 ### Redis 向量缓存
  
@@ -105,17 +99,13 @@ TEI_DEVICE=cpu ./scripts/start_tei_service.sh
  
 ./scripts/start_embedding_text_service.sh
 ./scripts/start_embedding_image_service.sh
-
-# 或兼容入口
-./scripts/start_embedding_service.sh text
-./scripts/start_embedding_service.sh image
 ```
  
 ### 修改配置
  
 编辑 `embeddings/config.py`：
  
-- `PORT`: combined 模式默认端口（默认 6005）
+- `PORT`: `all` 模式单进程端口（默认 6005）
 - `TEXT_MODEL_ID`, `TEXT_DEVICE`, `TEXT_BATCH_SIZE`, `TEXT_NORMALIZE_EMBEDDINGS`
 - `IMAGE_NORMALIZE_EMBEDDINGS`（默认 true）
 - `USE_CLIP_AS_SERVICE`, `CLIP_AS_SERVICE_SERVER`, `CLIP_AS_SERVICE_MODEL_NAME`：图片向量（clip-as-service）
@@ -121,7 +121,9 @@ class ClipAsServiceImageEncoder:
             out.append(vec)
         return out
  
-    def encode_image_from_url(self, url: str, normalize_embeddings: bool = True) -> Optional[np.ndarray]:
-        """Encode a single image URL. Returns 1024-dim vector or None."""
+    def encode_image_from_url(self, url: str, normalize_embeddings: bool = True) -> np.ndarray:
+        """Encode a single image URL and return one 1024-dim vector."""
         results = self.encode_image_urls([url], batch_size=1, normalize_embeddings=normalize_embeddings)
-        return results[0] if results else None
+        if not results:
+            raise RuntimeError("clip-as-service returned empty result for single image URL")
+        return results[0]
@@ -86,7 +86,7 @@ class ClipImageModel(object):
                 image_features /= image_features.norm(dim=-1, keepdim=True)
         return image_features.cpu().numpy().astype("float32")[0]
  
-    def encode_image_from_url(self, url: str, normalize_embeddings: bool = True) -> Optional[np.ndarray]:
+    def encode_image_from_url(self, url: str, normalize_embeddings: bool = True) -> np.ndarray:
         image_data = self.download_image(url)
         image = self.validate_image(image_data)
         image = self.preprocess_image(image)
@@ -97,7 +97,7 @@ class ClipImageModel(object):
         urls: List[str],
         batch_size: Optional[int] = None,
         normalize_embeddings: bool = True,
-    ) -> List[Optional[np.ndarray]]:
+    ) -> List[np.ndarray]:
         """
         Encode a list of image URLs to vectors. Same interface as ClipAsServiceImageEncoder.
  
@@ -106,7 +106,7 @@ class ClipImageModel(object):
             batch_size: batch size for internal batching (default 8).
  
         Returns:
-            List of vectors (or None for failed items), same length as urls.
+            List of vectors, same length as urls.
         """
         return self.encode_batch(
             urls,
@@ -119,8 +119,8 @@ class ClipImageModel(object):
         images: List[Union[str, Image.Image]],
         batch_size: int = 8,
         normalize_embeddings: bool = True,
-    ) -> List[Optional[np.ndarray]]:
-        results: List[Optional[np.ndarray]] = []
+    ) -> List[np.ndarray]:
+        results: List[np.ndarray] = []
         for i in range(0, len(images), batch_size):
             batch = images[i : i + batch_size]
             for img in batch:
@@ -129,6 +129,5 @@ class ClipImageModel(object):
                 elif isinstance(img, Image.Image):
                     results.append(self.encode_image(img, normalize_embeddings=normalize_embeddings))
                 else:
-                    results.append(None)
+                    raise ValueError(f"Unsupported image input type: {type(img)!r}")
         return results
-
@@ -18,7 +18,7 @@ class EmbeddingConfig(object):
  
     # Text backend defaults
     TEXT_MODEL_ID = os.getenv("TEXT_MODEL_ID", "Qwen/Qwen3-Embedding-0.6B")
-    # Backward-compatible alias for old naming in docs/scripts.
+    # Keep TEXT_MODEL_DIR as an alias so code can refer to one canonical text model value.
     TEXT_MODEL_DIR = TEXT_MODEL_ID
     TEXT_DEVICE = os.getenv("TEXT_DEVICE", "cuda")  # "cuda" or "cpu"
     TEXT_BATCH_SIZE = int(os.getenv("TEXT_BATCH_SIZE", "32"))
 """Image embedding client for the local embedding HTTP service."""
  
-import os
 import logging
 from typing import Any, List, Optional, Union
  
@@ -24,12 +23,7 @@ class CLIPImageEncoder:
     """
  
     def __init__(self, service_url: Optional[str] = None):
-        resolved_url = (
-            service_url
-            or os.getenv("EMBEDDING_IMAGE_SERVICE_URL")
-            or os.getenv("EMBEDDING_SERVICE_URL")
-            or get_embedding_image_base_url()
-        )
+        resolved_url = service_url or get_embedding_image_base_url()
         self.service_url = str(resolved_url).rstrip("/")
         self.endpoint = f"{self.service_url}/embed/image"
         # Reuse embedding cache prefix, but separate namespace for images to avoid collisions.
@@ -18,11 +18,12 @@ class ImageEncoderProtocol(Protocol):
         urls: List[str],
         batch_size: Optional[int] = None,
         normalize_embeddings: bool = True,
-    ) -> List[Optional[np.ndarray]]:
+    ) -> List[np.ndarray]:
         """
         Encode a list of image URLs to vectors.
  
         Returns:
-            List of vectors (or None for failed items), same length as urls.
+            List of vectors, same length as urls. Invalid inputs should raise instead
+            of returning partial None placeholders.
         """
         ...
 """Text embedding client for the local embedding HTTP service."""
  
 import logging
-import os
 from datetime import timedelta
 from typing import Any, List, Optional, Union
  
@@ -24,12 +23,7 @@ class TextEmbeddingEncoder:
     """
  
     def __init__(self, service_url: Optional[str] = None):
-        resolved_url = (
-            service_url
-            or os.getenv("EMBEDDING_TEXT_SERVICE_URL")
-            or os.getenv("EMBEDDING_SERVICE_URL")
-            or get_embedding_text_base_url()
-        )
+        resolved_url = service_url or get_embedding_text_base_url()
         self.service_url = str(resolved_url).rstrip("/")
         self.endpoint = f"{self.service_url}/embed/text"
         self.expire_time = timedelta(days=REDIS_CONFIG.get("cache_expire_days", 180))
@@ -30,7 +30,7 @@ get_port() {
     backend) echo "${API_PORT:-6002}" ;;
     indexer) echo "${INDEXER_PORT:-6004}" ;;
     frontend) echo "${FRONTEND_PORT:-6003}" ;;
-    embedding) echo "${EMBEDDING_TEXT_PORT:-${EMBEDDING_PORT:-6005}}" ;;
+    embedding) echo "${EMBEDDING_TEXT_PORT:-6005}" ;;
     embedding-image) echo "${EMBEDDING_IMAGE_PORT:-6008}" ;;
     translator) echo "${TRANSLATION_PORT:-6006}" ;;
     reranker) echo "${RERANKER_PORT:-6007}" ;;
@@ -593,7 +593,7 @@ start_one() {
         return 1
       fi
       ;;
-    backend|indexer|frontend|embedding|translator|reranker)
+    backend|indexer|frontend|embedding|embedding-image|translator|reranker)
       echo "[start] ${service}"
       nohup "${cmd}" >> "${lf}" 2>&1 &
       local pid=$!
@@ -60,7 +60,7 @@ fi
  
 EMBEDDING_SERVICE_HOST="${EMBEDDING_HOST:-${DEFAULT_EMBEDDING_SERVICE_HOST}}"
 if [[ "${SERVICE_KIND}" == "text" ]]; then
-  EMBEDDING_SERVICE_PORT="${EMBEDDING_TEXT_PORT:-${EMBEDDING_PORT:-${DEFAULT_EMBEDDING_SERVICE_PORT}}}"
+  EMBEDDING_SERVICE_PORT="${EMBEDDING_TEXT_PORT:-6005}"
 elif [[ "${SERVICE_KIND}" == "image" ]]; then
   EMBEDDING_SERVICE_PORT="${EMBEDDING_IMAGE_PORT:-6008}"
 else
@@ -147,7 +147,7 @@ if [[ &quot;${SERVICE_KIND}&quot; == &quot;text&quot; ]]; then
 elif [[ "${SERVICE_KIND}" == "image" ]]; then
   echo "  - Clients can set EMBEDDING_IMAGE_SERVICE_URL=http://localhost:${EMBEDDING_SERVICE_PORT}"
 else
-  echo "  - Clients can set EMBEDDING_SERVICE_URL=http://localhost:${EMBEDDING_SERVICE_PORT}"
+  echo "  - All mode serves both /embed/text and /embed/image on port ${EMBEDDING_SERVICE_PORT}"
 fi
 echo
  
@@ -14,7 +14,7 @@ echo &quot;索引服务调用方排查&quot;
 echo "=========================================="
  
 INDEXER_PORT="${INDEXER_PORT:-6004}"
-EMBEDDING_TEXT_PORT="${EMBEDDING_TEXT_PORT:-${EMBEDDING_PORT:-6005}}"
+EMBEDDING_TEXT_PORT="${EMBEDDING_TEXT_PORT:-6005}"
 EMBEDDING_IMAGE_PORT="${EMBEDDING_IMAGE_PORT:-6008}"
  
 echo ""