import os import json import requests import asyncio from abc import abstractmethod from zhipuai import ZhipuAI from langchain_huggingface import HuggingFaceEmbeddings from src import config from src.utils import hashstr, logger, get_docker_safe_url class BaseEmbeddingModel: embed_state = {} def __init__(self, model_id): self.model_id = model_id self.info = config.embed_model_names[model_id] self.model = self.info["name"] self.dimension = self.info.get("dimension", None) self.url = get_docker_safe_url(self.info["base_url"]) self.api_key = os.getenv(self.info["api_key"], self.info["api_key"]) @abstractmethod def predict(self, message): raise NotImplementedError("Subclasses must implement this method") def encode(self, message): return self.predict(message) def encode_queries(self, queries): return self.predict(queries) async def aencode(self, message): return await asyncio.to_thread(self.encode, message) async def aencode_queries(self, queries): return await asyncio.to_thread(self.encode_queries, queries) async def abatch_encode(self, messages, batch_size=20): return await asyncio.to_thread(self.batch_encode, messages, batch_size) def batch_encode(self, messages, batch_size=20): logger.info(f"Batch encoding {len(messages)} messages") data = [] if len(messages) > batch_size: task_id = hashstr(messages) self.embed_state[task_id] = { 'status': 'in-progress', 'total': len(messages), 'progress': 0 } for i in range(0, len(messages), batch_size): group_msg = messages[i:i+batch_size] logger.info(f"Encoding {i} to {i+batch_size} with {len(messages)} messages") response = self.encode(group_msg) # logger.debug(f"Response: {len(response)=}, {len(group_msg)=}, {len(response[0])=}") data.extend(response) if len(messages) > batch_size: self.embed_state[task_id]['progress'] = len(messages) self.embed_state[task_id]['status'] = 'completed' return data class OllamaEmbedding(BaseEmbeddingModel): """ Ollama Embedding Model """ def __init__(self, model_id) -> None: super().__init__(model_id) self.url = self.url or get_docker_safe_url("http://localhost:11434/api/embed") def predict(self, message: list[str] | str): if isinstance(message, str): message = [message] payload = { "model": self.model, "input": message, } response = requests.request("POST", self.url, json=payload) response = json.loads(response.text) assert response.get("embeddings"), f"Ollama Embedding failed: {response}" return response["embeddings"] class OtherEmbedding(BaseEmbeddingModel): def __init__(self, model_id) -> None: super().__init__(model_id) self.headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } def predict(self, message): payload = self.build_payload(message) response = requests.request("POST", self.url, json=payload, headers=self.headers) response = json.loads(response.text) assert response["data"], f"Other Embedding failed: {response}" data = [a["embedding"] for a in response["data"]] return data def build_payload(self, message): return { "model": self.model, "input": message, } def get_embedding_model(model_id): provider, model_name = model_id.split('/', 1) if model_id else ("", "") support_embed_models = config.embed_model_names.keys() assert model_id in support_embed_models, f"Unsupported embed model: {model_id}, only support {support_embed_models}" logger.debug(f"Loading embedding model {model_id}") if provider == "local": raise ValueError("Local embedding model is not supported, please use other embedding models") elif provider == "ollama": model = OllamaEmbedding(model_id) else: model = OtherEmbedding(model_id) return model