| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859 |
- """火山方舟 Doubao-embedding-vision 文本向量化客户端（scope-link 用）。
- 契约见《数据接口与来源/embedding接口.md》：
- - Endpoint: POST .../api/v3/embeddings/multimodal（vision 版必须走 multimodal）
- - model = 预置推理接入点 ep-id
- - 输入 input=[{"type":"text","text":...}]，输出 data.embedding = 2048 维
- - 单条调用（一次请求一个向量），批量靠并发多次请求。
- 密钥/ep 全部从 .env 读，不硬编码。
- """
- from __future__ import annotations
- import time
- from dataclasses import dataclass
- from pathlib import Path
- import httpx
- from core.config import env_value, load_env_file
- _DEFAULT_URL = "https://ark.cn-beijing.volces.com/api/v3/embeddings/multimodal"
- @dataclass
- class ArkEmbedConfig:
- api_key: str
- endpoint: str # ep-id，作为 model 字段传入
- url: str = _DEFAULT_URL
- dim: int = 2048
- timeout: int = 30
- @classmethod
- def from_env(cls, env_file: str | Path = ".env") -> "ArkEmbedConfig":
- fe = load_env_file(env_file)
- return cls(
- api_key=env_value("ARK_API_KEY", fe, required=True),
- endpoint=env_value("ARK_EMBEDDING_EP", fe, required=True),
- url=env_value("ARK_EMBEDDING_URL", fe, _DEFAULT_URL),
- dim=int(env_value("ARK_EMBEDDING_DIM", fe, "2048")),
- )
- def embed_text(text: str, cfg: ArkEmbedConfig, retries: int = 4) -> list[float]:
- """文本 → 2048 维向量。失败指数退避重试，最终仍失败则抛出。"""
- payload = {"model": cfg.endpoint, "input": [{"type": "text", "text": text}]}
- headers = {
- "Authorization": f"Bearer {cfg.api_key}",
- "Content-Type": "application/json",
- }
- last_err: Exception | None = None
- for attempt in range(retries):
- try:
- resp = httpx.post(cfg.url, json=payload, headers=headers, timeout=cfg.timeout)
- resp.raise_for_status()
- return resp.json()["data"]["embedding"]
- except Exception as err: # noqa: BLE001 — 网络/限流统一退避重试
- last_err = err
- time.sleep(1.0 * (attempt + 1))
- raise RuntimeError(f"ark embed failed for {text!r}: {last_err}")
|
