File size: 3,293 Bytes
f5776d3 |
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 |
from contextlib import contextmanager
from dsp.utils.utils import dotdict
import threading
class Settings(object):
"""DSP configuration settings."""
_instance = None
def __new__(cls):
"""
Singleton Pattern. See https://python-patterns.guide/gang-of-four/singleton/
"""
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance.lock = threading.Lock()
cls._instance.main_tid = threading.get_ident()
cls._instance.main_stack = []
cls._instance.stack_by_thread = {}
cls._instance.stack_by_thread[
threading.get_ident()
] = cls._instance.main_stack
# TODO: remove first-class support for re-ranker and potentially combine with RM to form a pipeline of sorts
# eg: RetrieveThenRerankPipeline(RetrievalModel, Reranker)
# downstream operations like dsp.retrieve would use configs from the defined pipeline.
config = dotdict(
lm=None,
rm=None,
branch_idx=0,
reranker=None,
compiled_lm=None,
force_reuse_cached_compilation=False,
compiling=False,
skip_logprobs=False,
trace=None,
release=0,
log_openai_usage=False,
bypass_assert=False,
bypass_suggest=False,
assert_failures=0,
suggest_failures=0,
langchain_history=[]
)
cls._instance.__append(config)
return cls._instance
@property
def config(self):
thread_id = threading.get_ident()
if thread_id not in self.stack_by_thread:
self.stack_by_thread[thread_id] = [self.main_stack[-1].copy()]
return self.stack_by_thread[thread_id][-1]
def __getattr__(self, name):
if hasattr(self.config, name):
return getattr(self.config, name)
if name in self.config:
return self.config[name]
super().__getattr__(name)
def __append(self, config):
thread_id = threading.get_ident()
if thread_id not in self.stack_by_thread:
self.stack_by_thread[thread_id] = [self.main_stack[-1].copy()]
self.stack_by_thread[thread_id].append(config)
def __pop(self):
thread_id = threading.get_ident()
if thread_id in self.stack_by_thread:
self.stack_by_thread[thread_id].pop()
def configure(self, inherit_config: bool = True, **kwargs):
"""Set configuration settings.
Args:
inherit_config (bool, optional): Set configurations for the given, and use existing configurations for the rest. Defaults to True.
"""
if inherit_config:
config = {**self.config, **kwargs}
else:
config = {**kwargs}
self.__append(config)
@contextmanager
def context(self, inherit_config=True, **kwargs):
self.configure(inherit_config=inherit_config, **kwargs)
try:
yield
finally:
self.__pop()
def __repr__(self) -> str:
return repr(self.config)
settings = Settings()
|