✨ feat(langfuse.py): add temporary implementation of LangfuseInstance class

The LangfuseInstance class is a temporary implementation that will be replaced by a plugin system in the future. This class provides methods for creating, getting, updating, and tearing down an instance of Langfuse. It uses the Langfuse library and credentials from the settings module to create the instance. If no credentials are found, the instance is set to None. The update method resets the instance and creates a new one. The teardown method flushes the instance and sets it to None.
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-08-30 09:49:38 -03:00
commit 9559f1d684

View file

@ -0,0 +1,44 @@
from langflow.utils.logger import logger
### Temporary implementation
# This will be replaced by a plugin system once merged into 0.5.0
class LangfuseInstance:
_instance = None
@classmethod
def get(cls):
logger.debug("Getting Langfuse instance")
if cls._instance is None:
cls.create()
return cls._instance
@classmethod
def create(cls):
logger.debug("Creating Langfuse instance")
from langflow.settings import settings
from langfuse import Langfuse
if settings.LANGFUSE_PUBLIC_KEY and settings.LANGFUSE_SECRET_KEY:
logger.debug("Langfuse credentials found")
cls._instance = Langfuse(
public_key=settings.LANGFUSE_PUBLIC_KEY,
secret_key=settings.LANGFUSE_SECRET_KEY,
)
else:
logger.debug("No Langfuse credentials found")
cls._instance = None
@classmethod
def update(cls):
logger.debug("Updating Langfuse instance")
cls._instance = None
cls.create()
@classmethod
def teardown(cls):
logger.debug("Tearing down Langfuse instance")
if cls._instance is not None:
cls._instance.flush()
cls._instance = None