#!/usr/bin/env python3 """aria-embed-providers — Multi-Provider Embedding-Adapter für aria-brain. KAR-180 (Phase-1 gbrain-Adoption). 14 Provider-Recipes vorbereitet, 3 Top-Provider initial implementiert: OpenAI, Voyage, Ollama. Adoptiert aus gbrain's `docs/integrations/embedding-providers.md`. Usage als Library: from aria_embed_providers import embed, get_recipe vectors = embed(["hello world"], provider="openai") vectors = embed(["def foo():"], provider="voyage") # code-strong vectors = embed(["geheim"], provider="ollama") # local, free Usage als CLI: aria-embed-providers.py --provider openai --text "hello" aria-embed-providers.py --provider voyage --file note.md aria-embed-providers.py --list Recipe-Format pro Provider (Dict): - model: str - dimensions: int - api_key_env: str | None (None für lokale) - endpoint: str - batch_size: int - cost_per_1m_tokens: float - auto_detect: callable | None (Heuristik: lohnt sich dieser Provider für diesen Text?) Quelle: gbrain repo, audit `02-Wissen/garrytan-github-audit-2026-05-14.md` §3.4 """ from __future__ import annotations import argparse import json import os import re import sys import time import urllib.request import urllib.error from pathlib import Path from typing import Callable, Any # --- 14 Recipes (3 active, 11 vorbereitet) ---------------------------------- RECIPES: dict[str, dict[str, Any]] = { # --- Active (Phase-1 Top-3, Kais-Decision G) --- "openai": { "model": "text-embedding-3-large", "dimensions": 3072, "api_key_env": "OPENAI_API_KEY", "endpoint": "https://api.openai.com/v1/embeddings", "batch_size": 100, "cost_per_1m_tokens": 0.13, "active": True, "auto_detect": lambda t: True, # Default-Provider }, "voyage": { "model": "voyage-code-2", "dimensions": 1536, "api_key_env": "VOYAGE_API_KEY", "endpoint": "https://api.voyageai.com/v1/embeddings", "batch_size": 128, "cost_per_1m_tokens": 0.12, "active": True, # Code-Heuristik: enthält Code-Marker "auto_detect": lambda t: bool(re.search(r"```|def\s|function\s|class\s|import\s|from\s+\w+\s+import", t)), }, "ollama": { "model": "nomic-embed-text", "dimensions": 768, "api_key_env": None, "endpoint": "http://localhost:11434/api/embeddings", "batch_size": 1, # Ollama ist single-text per call "cost_per_1m_tokens": 0.0, "active": True, "auto_detect": lambda t: False, # nur explizit, als Privacy-Fallback }, # --- Vorbereitet (nicht aktiv, zur späteren Integration) --- "azure-openai": { "model": "text-embedding-3-large", "dimensions": 3072, "api_key_env": "AZURE_OPENAI_API_KEY", "endpoint": "https://YOUR-RESOURCE.openai.azure.com/openai/deployments/YOUR-DEPLOY/embeddings?api-version=2024-02-01", "batch_size": 16, "cost_per_1m_tokens": 0.13, "active": False, }, "gemini": { "model": "gemini-embedding-001", "dimensions": 3072, "api_key_env": "GEMINI_API_KEY", "endpoint": "https://generativelanguage.googleapis.com/v1beta/models/gemini-embedding-001:embedContent", "batch_size": 100, "cost_per_1m_tokens": 0.15, "active": False, }, "cohere": { "model": "embed-english-v3.0", "dimensions": 1024, "api_key_env": "COHERE_API_KEY", "endpoint": "https://api.cohere.com/v2/embed", "batch_size": 96, "cost_per_1m_tokens": 0.10, "active": False, }, "mistral": { "model": "mistral-embed", "dimensions": 1024, "api_key_env": "MISTRAL_API_KEY", "endpoint": "https://api.mistral.ai/v1/embeddings", "batch_size": 100, "cost_per_1m_tokens": 0.10, "active": False, }, "minimax": { "model": "embo-01", "dimensions": 1536, "api_key_env": "MINIMAX_API_KEY", "endpoint": "https://api.minimax.chat/v1/embeddings", "batch_size": 32, "cost_per_1m_tokens": 0.05, "active": False, }, "dashscope-alibaba": { "model": "text-embedding-v2", "dimensions": 1536, "api_key_env": "DASHSCOPE_API_KEY", "endpoint": "https://dashscope.aliyuncs.com/api/v1/services/embeddings/text-embedding/text-embedding", "batch_size": 25, "cost_per_1m_tokens": 0.05, "active": False, }, "zhipu": { "model": "embedding-3", "dimensions": 2048, "api_key_env": "ZHIPU_API_KEY", "endpoint": "https://open.bigmodel.cn/api/paas/v4/embeddings", "batch_size": 64, "cost_per_1m_tokens": 0.05, "active": False, }, "llamacpp-server": { "model": "local-embed", "dimensions": 768, "api_key_env": None, "endpoint": "http://localhost:8080/embedding", "batch_size": 1, "cost_per_1m_tokens": 0.0, "active": False, }, "litellm-proxy": { "model": "configurable-via-litellm", "dimensions": 3072, "api_key_env": "LITELLM_API_KEY", "endpoint": "http://localhost:4000/v1/embeddings", "batch_size": 50, "cost_per_1m_tokens": 0.0, "active": False, }, "huggingface-tei": { "model": "BAAI/bge-large-en-v1.5", "dimensions": 1024, "api_key_env": None, "endpoint": "http://localhost:8081/embed", "batch_size": 32, "cost_per_1m_tokens": 0.0, "active": False, }, "jina": { "model": "jina-embeddings-v3", "dimensions": 1024, "api_key_env": "JINA_API_KEY", "endpoint": "https://api.jina.ai/v1/embeddings", "batch_size": 100, "cost_per_1m_tokens": 0.10, "active": False, }, } def get_recipe(provider: str) -> dict[str, Any]: if provider not in RECIPES: raise ValueError(f"unknown provider: {provider}. Available: {sorted(RECIPES.keys())}") return RECIPES[provider] def _load_api_key(env_name: str) -> str: key = os.environ.get(env_name, "") if key: return key # Fallback: /root/.aria-secrets/.env, then /root/aria/.env candidates = [ f"/root/.aria-secrets/{env_name.lower().replace('_api_key','').replace('_','-')}.env", f"/root/.aria-secrets/{env_name.lower().split('_')[0]}.env", "/root/aria/.env", ] for c in candidates: if not Path(c).exists(): continue for line in Path(c).read_text().splitlines(): if line.startswith(f"{env_name}="): return line.split("=", 1)[1].strip().strip('"').strip("'") return "" def auto_provider(text: str, available: list[str] | None = None) -> str: """Heuristic: pick best provider based on text. Priority: specialized provider (e.g. voyage for code) > OpenAI default.""" if available is None: available = [k for k, v in RECIPES.items() if v.get("active")] # Check specialized providers FIRST (skip openai which is general-default) for p in available: if p == "openai": continue detect = RECIPES[p].get("auto_detect") if detect and detect(text): return p # General-default return "openai" if "openai" in available else available[0] def embed(texts: list[str], provider: str = "openai", retries: int = 3, timeout: int = 60) -> list[list[float]]: """Embed texts via specified provider. Returns list of vectors. Raises: ValueError on missing API key or unknown provider. RuntimeError on API failure after retries. """ if isinstance(texts, str): texts = [texts] recipe = get_recipe(provider) if not recipe.get("active"): raise ValueError(f"provider {provider} is prepared but not active") api_key = "" if recipe["api_key_env"]: api_key = _load_api_key(recipe["api_key_env"]) if not api_key: raise ValueError(f"{recipe['api_key_env']} not set " f"(env or /root/.aria-secrets/)") if provider == "ollama": # Ollama API: single text per call, simpler shape vectors = [] for t in texts: payload = json.dumps({"model": recipe["model"], "prompt": t}).encode() req = urllib.request.Request(recipe["endpoint"], data=payload, headers={"Content-Type": "application/json"}, method="POST") with urllib.request.urlopen(req, timeout=timeout) as r: data = json.loads(r.read()) vectors.append(data["embedding"]) return vectors # OpenAI-compatible API (OpenAI, Voyage, most others) payload = json.dumps({"model": recipe["model"], "input": texts}).encode() headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } last_err = None delay = 1.0 for attempt in range(retries + 1): try: req = urllib.request.Request(recipe["endpoint"], data=payload, headers=headers, method="POST") with urllib.request.urlopen(req, timeout=timeout) as r: data = json.loads(r.read()) return [item["embedding"] for item in data["data"]] except urllib.error.HTTPError as e: last_err = e if e.code in (429, 500, 502, 503, 504) and attempt < retries: time.sleep(delay) delay *= 2 continue raise RuntimeError(f"{provider} embed-fail: HTTP {e.code} {e.read()[:200]!r}") except Exception as e: last_err = e if attempt < retries: time.sleep(delay) delay *= 2 continue raise RuntimeError(f"{provider} embed-fail: {e}") raise RuntimeError(f"{provider} embed-fail after {retries} retries: {last_err}") def main(argv: list[str]) -> int: ap = argparse.ArgumentParser(description="Multi-provider embedding adapter") ap.add_argument("--provider", default=None, help="openai|voyage|ollama|... (default: auto-detect)") ap.add_argument("--text", default=None, help="Single text to embed") ap.add_argument("--file", default=None, help="File path to embed (full content)") ap.add_argument("--list", action="store_true", help="List all available recipes") ap.add_argument("--head", type=int, default=8, help="Show first N dims of vector (default 8)") args = ap.parse_args(argv[1:]) if args.list: print(f"{'Provider':<22} {'Model':<32} {'Dim':>5} {'$/1M':>6} {'Active':>8}") print("-" * 78) for name, r in sorted(RECIPES.items()): print(f"{name:<22} {r['model']:<32} {r['dimensions']:>5} " f"{r['cost_per_1m_tokens']:>6.3f} {'Y' if r.get('active') else 'N':>8}") return 0 if args.text: text = args.text elif args.file: text = Path(args.file).read_text(encoding="utf-8") else: print("ERROR: --text, --file, or --list required", file=sys.stderr) return 1 provider = args.provider or auto_provider(text) print(f"[embed-providers] using provider={provider}", file=sys.stderr) vecs = embed([text], provider=provider) print(f"[embed-providers] vector dim={len(vecs[0])}", file=sys.stderr) print(json.dumps({"provider": provider, "dim": len(vecs[0]), "head": vecs[0][:args.head]})) return 0 if __name__ == "__main__": sys.exit(main(sys.argv))