File size: 2,910 Bytes
582520a a6e51ba 582520a a6e51ba 582520a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """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
|