141 lines
4.3 KiB
Python
141 lines
4.3 KiB
Python
from dataclasses import dataclass, is_dataclass, fields, MISSING
|
|
from typing import Any, Tuple, Type
|
|
import yaml
|
|
from pathlib import Path
|
|
from typing import Dict
|
|
import os
|
|
|
|
from loguru import logger
|
|
|
|
## NOTE: base classes taken from nerfstudio
|
|
class PrintableConfig:
|
|
"""Printable Config defining str function"""
|
|
|
|
def __str__(self):
|
|
lines = [self.__class__.__name__ + ":"]
|
|
for key, val in vars(self).items():
|
|
|
|
if self.is_secrete(key):
|
|
val = str(val)
|
|
val = val[:3] + "*"*(len(val) - 6) + val[-3:]
|
|
|
|
if isinstance(val, Tuple):
|
|
flattened_val = "["
|
|
for item in val:
|
|
flattened_val += str(item) + "\n"
|
|
flattened_val = flattened_val.rstrip("\n")
|
|
val = flattened_val + "]"
|
|
lines += f"{key}: {str(val)}".split("\n")
|
|
return "\n" + "\n ".join(lines)
|
|
|
|
def is_secrete(self, inp:str):
|
|
sec_list = ["secret", "api_key"]
|
|
for sec in sec_list:
|
|
if sec in inp:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
# Base instantiate configs
|
|
@dataclass
|
|
class InstantiateConfig(PrintableConfig):
|
|
"""Config class for instantiating an the class specified in the _target attribute."""
|
|
|
|
_target: Type
|
|
|
|
def setup(self, **kwargs) -> Any:
|
|
"""Returns the instantiated object using the config."""
|
|
return self._target(self, **kwargs)
|
|
|
|
def save_config(self, filename: str) -> None:
|
|
"""Save the config to a YAML file."""
|
|
with open(filename, 'w') as f:
|
|
yaml.dump(self, f)
|
|
logger.info(f"[yellow]config saved to: {filename}[/yellow]")
|
|
|
|
|
|
|
|
|
|
@dataclass
|
|
class KeyConfig(InstantiateConfig):
|
|
|
|
api_key:str = None
|
|
"""api key for llm"""
|
|
|
|
def __post_init__(self):
|
|
if self.api_key == "wrong-key" or self.api_key is None:
|
|
self.api_key = os.environ.get("ALI_API_KEY")
|
|
if self.api_key is None:
|
|
logger.error(f"no ALI_API_KEY provided for embedding")
|
|
else:
|
|
logger.info("ALI_API_KEY loaded from environ")
|
|
|
|
|
|
@dataclass
|
|
class ToolConfig(InstantiateConfig):
|
|
use_tool:bool = True
|
|
"""specify to use tool or not"""
|
|
|
|
def load_tyro_conf(filename: str, inp_conf = None) -> InstantiateConfig:
|
|
"""load and overwrite config from file"""
|
|
config = yaml.load(Path(filename).read_text(), Loader=yaml.Loader)
|
|
|
|
config = ovewrite_config(config, inp_conf) if inp_conf is not None else config
|
|
return config
|
|
|
|
def is_default(instance, field_):
|
|
"""
|
|
Check if the value of a field in a dataclass instance is the default value.
|
|
"""
|
|
value = getattr(instance, field_.name)
|
|
|
|
if field_.default is not MISSING:
|
|
# Compare with default value
|
|
return value == field_.default
|
|
elif field_.default_factory is not MISSING:
|
|
# Compare with value generated by the default factory
|
|
return value == field_.default_factory()
|
|
else:
|
|
# No default value specified
|
|
return False
|
|
|
|
def ovewrite_config(loaded_conf, inp_conf):
|
|
"""for non-default values in inp_conf, overwrite the corresponding values in loaded_conf"""
|
|
if not (is_dataclass(loaded_conf) and is_dataclass(inp_conf)):
|
|
return loaded_conf
|
|
|
|
for field_ in fields(loaded_conf):
|
|
field_name = field_.name
|
|
# if field_name in inp_conf:
|
|
current_value = getattr(inp_conf, field_name)
|
|
new_value = getattr(inp_conf, field_name) #inp_conf[field_name]
|
|
|
|
if is_dataclass(current_value):
|
|
# Recurse for nested dataclasses
|
|
merged_value = ovewrite_config(current_value, new_value)
|
|
setattr(loaded_conf, field_name, merged_value)
|
|
elif not is_default(inp_conf, field_):
|
|
# Overwrite only if the current value is not default
|
|
setattr(loaded_conf, field_name, new_value)
|
|
|
|
return loaded_conf
|
|
|
|
|
|
def mcp_langchain_to_ws_config(conf:Dict[str, Dict[str, str]]):
|
|
serv_conf = {}
|
|
|
|
for k, v in conf.items():
|
|
|
|
if v["transport"] == "stdio":
|
|
serv_conf[k] = {
|
|
"type" : v["transport"],
|
|
"command": v["command"],
|
|
"args": v["args"],
|
|
}
|
|
else:
|
|
logger.warning(f"Unsupported transport {v['transport']} for MCP {k}. Skipping...")
|
|
continue
|
|
|
|
return {"mcpServers":serv_conf}
|