Spaces:
Sleeping
Sleeping
File size: 3,123 Bytes
257bc0d 5ea412d 257bc0d 5ea412d 9ecd254 5ea412d 257bc0d 9ecd254 5ea412d 9ecd254 5ea412d 257bc0d 5ea412d 257bc0d 5ea412d 257bc0d b404642 bd59653 257bc0d b404642 257bc0d 01d6691 9ecd254 5ea412d bd59653 5ea412d | 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 | import enum
import os
from core.helper import LifecycleHelper
from xpipe_wiki.robot_manager import XPipeWikiRobotManager, AzureXPipeWikiRobotManager
from multiprocessing import Lock
lock = Lock()
class XPipeRobotRevision(enum.Enum):
SIMPLE_OPENAI_VERSION_0 = 1
HUGGINGFACE_VERSION_0 = 2
class XPipeRobotManagerFactory:
"""
CAPABLE: Dict[XPipeRobotRevision, XPipeWikiRobotManager] =
{XPipeRobotRevision.SIMPLE_OPENAI_VERSION_0: XPipeWikiRobotManager()}
"""
CAPABLE = dict() # type: dict[XPipeRobotRevision, XPipeWikiRobotManager]
@classmethod
def get_or_create(cls, revision: XPipeRobotRevision) -> XPipeWikiRobotManager:
with lock:
if cls.CAPABLE.get(revision) is not None:
return cls.CAPABLE[revision]
if revision == XPipeRobotRevision.SIMPLE_OPENAI_VERSION_0:
manager = cls.create_simple_openai_version_0()
elif revision == XPipeRobotRevision.HUGGINGFACE_VERSION_0:
manager = cls.create_huggingface_version_0()
cls.CAPABLE[revision] = manager
return manager
@classmethod
def create_simple_openai_version_0(cls) -> AzureXPipeWikiRobotManager:
from llama.service_context import AzureServiceContextManager
from langchain_manager.manager import LangChainAzureManager
service_context_manager = AzureServiceContextManager(
lc_manager=LangChainAzureManager()
)
from llama.storage_context import LocalStorageContextManager
dataset_path = os.getenv("XPIPE_WIKI_DATASET_PATH", "./dataset")
storage_context_manager = LocalStorageContextManager(
dataset_path=dataset_path, service_context_manager=service_context_manager
)
robot_manager = AzureXPipeWikiRobotManager(
service_context_manager=service_context_manager,
storage_context_manager=storage_context_manager,
)
LifecycleHelper.initialize_if_possible(robot_manager)
LifecycleHelper.start_if_possible(robot_manager)
return robot_manager
@classmethod
def create_huggingface_version_0(cls) -> AzureXPipeWikiRobotManager:
from llama.service_context import HuggingFaceChineseOptServiceContextManager
from langchain_manager.manager import LangChainAzureManager
service_context_manager = HuggingFaceChineseOptServiceContextManager(
lc_manager=LangChainAzureManager()
)
from llama.storage_context import LocalStorageContextManager
dataset_path = os.getenv("XPIPE_WIKI_DATASET_PATH", "./dataset")
storage_context_manager = LocalStorageContextManager(
dataset_path=dataset_path, service_context_manager=service_context_manager
)
robot_manager = AzureXPipeWikiRobotManager(
service_context_manager=service_context_manager,
storage_context_manager=storage_context_manager,
)
LifecycleHelper.initialize_if_possible(robot_manager)
LifecycleHelper.start_if_possible(robot_manager)
return robot_manager
|