my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	from typing import Any, Optional
#2
#3	from embedchain.config.base_config import BaseConfig
#4	from embedchain.helpers.json_serializable import register_deserializable
#5
#6
#7	@register_deserializable
#8	class CacheSimilarityEvalConfig(BaseConfig):
#9	"""
#10	This is the evaluator to compare two embeddings according to their distance computed in embedding retrieval stage.
#11	In the retrieval stage, `search_result` is the distance used for approximate nearest neighbor search and have been
#12	put into `cache_dict`. `max_distance` is used to bound this distance to make it between [0-`max_distance`].
#13	`positive` is used to indicate this distance is directly proportional to the similarity of two entities.
#14	If `positive` is set `False`, `max_distance` will be used to subtract this distance to get the final score.
#15
#16	:param max_distance: the bound of maximum distance.
#17	:type max_distance: float
#18	:param positive: if the larger distance indicates more similar of two entities, It is True. Otherwise, it is False.
#19	:type positive: bool
#20	"""
#21
#22	def __init__(
#23	self,
#24	strategy: Optional[str] = "distance",
#25	max_distance: Optional[float] = 1.0,
#26	positive: Optional[bool] = False,
#27	):
#28	self.strategy = strategy
#29	self.max_distance = max_distance
#30	self.positive = positive
#31
#32	@staticmethod
#33	def from_config(config: Optional[dict[str, Any]]):
#34	if config is None:
#35	return CacheSimilarityEvalConfig()
#36	else:
#37	return CacheSimilarityEvalConfig(
#38	strategy=config.get("strategy", "distance"),
#39	max_distance=config.get("max_distance", 1.0),
#40	positive=config.get("positive", False),
#41	)
#42
#43
#44	@register_deserializable
#45	class CacheInitConfig(BaseConfig):
#46	"""
#47	This is a cache init config. Used to initialize a cache.
#48
#49	:param similarity_threshold: a threshold ranged from 0 to 1 to filter search results with similarity score higher \
#50	than the threshold. When it is 0, there is no hits. When it is 1, all search results will be returned as hits.
#51	:type similarity_threshold: float
#52	:param auto_flush: it will be automatically flushed every time xx pieces of data are added, default to 20
#53	:type auto_flush: int
#54	"""
#55
#56	def __init__(
#57	self,
#58	similarity_threshold: Optional[float] = 0.8,
#59	auto_flush: Optional[int] = 20,
#60	):
#61	if similarity_threshold < 0 or similarity_threshold > 1:
#62	raise ValueError(f"similarity_threshold {similarity_threshold} should be between 0 and 1")
#63
#64	self.similarity_threshold = similarity_threshold
#65	self.auto_flush = auto_flush
#66
#67	@staticmethod
#68	def from_config(config: Optional[dict[str, Any]]):
#69	if config is None:
#70	return CacheInitConfig()
#71	else:
#72	return CacheInitConfig(
#73	similarity_threshold=config.get("similarity_threshold", 0.8),
#74	auto_flush=config.get("auto_flush", 20),
#75	)
#76
#77
#78	@register_deserializable
#79	class CacheConfig(BaseConfig):
#80	def __init__(
#81	self,
#82	similarity_eval_config: Optional[CacheSimilarityEvalConfig] = CacheSimilarityEvalConfig(),
#83	init_config: Optional[CacheInitConfig] = CacheInitConfig(),
#84	):
#85	self.similarity_eval_config = similarity_eval_config
#86	self.init_config = init_config
#87
#88	@staticmethod
#89	def from_config(config: Optional[dict[str, Any]]):
#90	if config is None:
#91	return CacheConfig()
#92	else:
#93	return CacheConfig(
#94	similarity_eval_config=CacheSimilarityEvalConfig.from_config(config.get("similarity_evaluation", {})),
#95	init_config=CacheInitConfig.from_config(config.get("init_config", {})),
#96	)
#97

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public