|
|
|
|
|
import dotenv, os |
|
import requests |
|
import requests |
|
from datetime import datetime |
|
|
|
dotenv.load_dotenv() |
|
import traceback |
|
|
|
class LangsmithLogger: |
|
|
|
def __init__(self): |
|
self.langsmith_api_key = os.getenv("LANGSMITH_API_KEY") |
|
|
|
|
|
def log_event(self, kwargs, response_obj, start_time, end_time, print_verbose): |
|
|
|
|
|
metadata = {} |
|
if "litellm_params" in kwargs: |
|
metadata = kwargs["litellm_params"].get("metadata", {}) |
|
|
|
|
|
|
|
|
|
project_name = metadata.get("project_name", "litellm-completion") |
|
run_name = metadata.get("run_name", "LLMRun") |
|
print_verbose(f"Langsmith Logging - project_name: {project_name}, run_name {run_name}") |
|
try: |
|
print_verbose( |
|
f"Langsmith Logging - Enters logging function for model {kwargs}" |
|
) |
|
import requests |
|
import datetime |
|
from datetime import timezone |
|
try: |
|
start_time = kwargs["start_time"].astimezone(timezone.utc).isoformat() |
|
end_time = kwargs["end_time"].astimezone(timezone.utc).isoformat() |
|
except: |
|
start_time = datetime.datetime.utcnow().isoformat() |
|
end_time = datetime.datetime.utcnow().isoformat() |
|
|
|
|
|
new_kwargs = {} |
|
for key in kwargs: |
|
value = kwargs[key] |
|
if key == "start_time" or key =="end_time": |
|
pass |
|
elif type(value) != dict: |
|
new_kwargs[key] = value |
|
|
|
requests.post( |
|
"https://api.smith.langchain.com/runs", |
|
json={ |
|
"name": run_name, |
|
"run_type": "llm", |
|
"inputs": { |
|
**new_kwargs |
|
}, |
|
"outputs": response_obj, |
|
"session_name": project_name, |
|
"start_time": start_time, |
|
"end_time": end_time, |
|
}, |
|
headers={ |
|
"x-api-key": self.langsmith_api_key |
|
} |
|
) |
|
print_verbose( |
|
f"Langsmith Layer Logging - final response object: {response_obj}" |
|
) |
|
except: |
|
|
|
print_verbose(f"Langsmith Layer Error - {traceback.format_exc()}") |
|
pass |
|
|