forked from innovacion/Mayacontigo
52 lines
1.3 KiB
Python
52 lines
1.3 KiB
Python
from typing import Protocol
|
|
|
|
|
|
class BaseAda:
|
|
def __init__(
|
|
self, model: str | None = None, *, endpoint: str, key: str, version: str
|
|
) -> None:
|
|
self.model = model
|
|
|
|
class Config(Protocol):
|
|
embedding_model: str
|
|
azure_endpoint: str
|
|
openai_api_key: str
|
|
openai_api_version: str
|
|
|
|
@classmethod
|
|
def from_config(cls, c: Config):
|
|
return cls(
|
|
model=c.embedding_model,
|
|
endpoint=c.azure_endpoint,
|
|
key=c.openai_api_key,
|
|
version=c.openai_api_version,
|
|
)
|
|
|
|
@classmethod
|
|
def from_vault(
|
|
cls,
|
|
vault: str,
|
|
*,
|
|
model: str | None = None,
|
|
url: str | None = None,
|
|
token: str | None = None,
|
|
mount_point: str = "secret",
|
|
):
|
|
from hvac import Client
|
|
|
|
client = Client(url=url or "https://vault.ia-innovacion.work", token=token)
|
|
|
|
if not client.is_authenticated():
|
|
raise Exception("Vault authentication failed")
|
|
|
|
secret_map = client.secrets.kv.v2.read_secret_version(
|
|
path=vault, mount_point=mount_point
|
|
)["data"]["data"]
|
|
|
|
return cls(
|
|
model=model,
|
|
endpoint=secret_map["azure_endpoint"],
|
|
key=secret_map["openai_api_key"],
|
|
version=secret_map["openai_api_version"],
|
|
)
|