| """Embedding helpers with OpenAI or local fallback.""" |
|
|
| from __future__ import annotations |
|
|
| import hashlib |
| import re |
| from typing import Sequence |
|
|
| import numpy as np |
|
|
| from app.config import get_settings |
|
|
| _DIM = 384 |
|
|
|
|
| def _tokenize(text: str) -> list[str]: |
| return re.findall(r"[\w\-]+", text.lower(), flags=re.UNICODE) |
|
|
|
|
| def _local_embed(text: str) -> list[float]: |
| """Deterministic bag-of-tokens embedding for demo mode.""" |
| vec = np.zeros(_DIM, dtype=np.float32) |
| tokens = _tokenize(text) |
| if not tokens: |
| return vec.tolist() |
| for token in tokens: |
| digest = hashlib.sha256(token.encode("utf-8")).digest() |
| idx = int.from_bytes(digest[:4], "little") % _DIM |
| sign = 1.0 if digest[4] % 2 == 0 else -1.0 |
| vec[idx] += sign |
| norm = float(np.linalg.norm(vec)) |
| if norm > 0: |
| vec /= norm |
| return vec.tolist() |
|
|
|
|
| async def embed_texts( |
| texts: Sequence[str], |
| *, |
| force_local: bool = False, |
| ) -> list[list[float]]: |
| """Embed texts via OpenAI or local fallback.""" |
| settings = get_settings() |
| use_local = ( |
| force_local |
| or settings.seed_local_embeddings |
| or settings.use_json_store |
| or not settings.has_openai |
| ) |
| if use_local: |
| return [_local_embed(t) for t in texts] |
|
|
| from openai import AsyncOpenAI |
|
|
| client = AsyncOpenAI(api_key=settings.openai_api_key) |
| response = await client.embeddings.create( |
| model=settings.openai_embedding_model, |
| input=list(texts), |
| ) |
| return [item.embedding for item in response.data] |
|
|
|
|
| def cosine_similarity(a: Sequence[float], b: Sequence[float]) -> float: |
| """Cosine similarity between two vectors.""" |
| va = np.asarray(a, dtype=np.float32) |
| vb = np.asarray(b, dtype=np.float32) |
| denom = float(np.linalg.norm(va) * np.linalg.norm(vb)) |
| if denom == 0: |
| return 0.0 |
| return float(np.dot(va, vb) / denom) |
|
|
|
|
| def token_overlap_score(query: str, document: str) -> float: |
| """Simple keyword overlap for offline demo retrieval.""" |
| q_tokens = { |
| t for t in _tokenize(query) if len(t) > 2 |
| } |
| if not q_tokens: |
| return 0.0 |
| d_tokens = set(_tokenize(document)) |
| if not d_tokens: |
| return 0.0 |
| hits = q_tokens & d_tokens |
| return len(hits) / len(q_tokens) |
|
|
|
|
| def chunk_text( |
| text: str, |
| chunk_size: int = 700, |
| overlap: int = 120, |
| ) -> list[str]: |
| """Split text into overlapping chunks.""" |
| cleaned = re.sub(r"\s+", " ", text).strip() |
| if not cleaned: |
| return [] |
| if len(cleaned) <= chunk_size: |
| return [cleaned] |
|
|
| chunks: list[str] = [] |
| start = 0 |
| while start < len(cleaned): |
| end = min(len(cleaned), start + chunk_size) |
| chunk = cleaned[start:end].strip() |
| if chunk: |
| chunks.append(chunk) |
| if end >= len(cleaned): |
| break |
| start = max(0, end - overlap) |
| return chunks |
|
|
