File size: 3,081 Bytes
395201c |
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 |
#### What this does ####
# On success, logs events to Langfuse
import dotenv, os
import requests
import requests
from datetime import datetime
dotenv.load_dotenv() # Loading env variables using dotenv
import traceback
class LangFuseLogger:
# Class variables or attributes
def __init__(self):
try:
from langfuse import Langfuse
except Exception as e:
raise Exception(f"\033[91mLangfuse not installed, try running 'pip install langfuse' to fix this error: {e}\033[0m")
# Instance variables
self.secret_key = os.getenv("LANGFUSE_SECRET_KEY")
self.public_key = os.getenv("LANGFUSE_PUBLIC_KEY")
self.langfuse_host = os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
self.langfuse_release = os.getenv("LANGFUSE_RELEASE")
self.langfuse_debug = os.getenv("LANGFUSE_DEBUG")
self.Langfuse = Langfuse(
public_key=self.public_key,
secret_key=self.secret_key,
host=self.langfuse_host,
release=self.langfuse_release,
debug=self.langfuse_debug
)
def log_event(self, kwargs, response_obj, start_time, end_time, print_verbose):
# Method definition
from langfuse.model import InitialGeneration, Usage
try:
print_verbose(
f"Langfuse Logging - Enters logging function for model {kwargs}"
)
litellm_params = kwargs.get("litellm_params", {})
metadata = litellm_params.get("metadata", {})
prompt = [kwargs.get('messages')]
optional_params = kwargs.get("optional_params", {})
# langfuse only accepts str, int, bool, float for logging
for param, value in optional_params.items():
if not isinstance(value, (str, int, bool, float)):
try:
optional_params[param] = str(value)
except:
# if casting value to str fails don't block logging
pass
# end of processing langfuse ########################
self.Langfuse.generation(InitialGeneration(
name=metadata.get("generation_name", "litellm-completion"),
startTime=start_time,
endTime=end_time,
model=kwargs['model'],
modelParameters=optional_params,
prompt=prompt,
completion=response_obj['choices'][0]['message'],
usage=Usage(
prompt_tokens=response_obj['usage']['prompt_tokens'],
completion_tokens=response_obj['usage']['completion_tokens']
),
metadata=metadata
))
self.Langfuse.flush()
print_verbose(
f"Langfuse Layer Logging - final response object: {response_obj}"
)
except:
# traceback.print_exc()
print_verbose(f"Langfuse Layer Error - {traceback.format_exc()}")
pass
|