apis/LangSmith.py

import requests
import json
import os 
from time import sleep
from dotenv import load_dotenv
from datetime import datetime


load_dotenv()

API_BASE_URL = "https://api.smith.langchain.com/api/v1"

class LangSmithAPI:

    def __init__(self):
        self.base_url = API_BASE_URL
        self.headers = {
            'X-API-Key': os.environ.get("LANGCHAIN_API_KEY")
        }

        self.getSessionId()

    def getSessionId(self):

        payload = {}

        response = requests.request("GET", f"{self.base_url}/sessions", headers=self.headers, data=payload)
        response_json = response.json()

        for session in response_json:
            session_id = session.get('id', None)
            name = session.get('name', 'Unnamed')
            if( name == os.environ.get("LANGCHAIN_PROJECT") ):
                self.session_id = session_id


    def getLastTraces(self, id_request: str = ''):
        payload = {
            "session": [self.session_id],
            "run_type":"chain",
            "select": [
                "id","trace_id","name","run_type","extra","start_time","inputs",
                "total_tokens","prompt_tokens","completion_tokens",
                "total_cost","prompt_cost","completion_cost","first_token_time"
            ],
            "order": "desc",
            "limit": 100
        }

        # Make request
        response = requests.request("POST", f"{self.base_url}/runs/query", headers=self.headers, data=json.dumps(payload))
        response_json = response.json()
           
        # filters for the traces that have the same id_request
        # filtered_traces = [trace for trace in response_json['runs'] if 'extra' in trace and 'metadata' in trace['extra'] and 'id_request' in trace['extra']['metadata'] and trace['extra']['metadata']['id_request'] == id_request]
        filtered_traces = [trace for trace in response_json['runs'] if trace.get('parent_run_ids') is not None and trace['parent_run_ids'] == []]
        
        return filtered_traces
        
    
    def getDailyMessages(self, id_request: str = ""):

        trace = self.getLastTraces(id_request)
        
        daily_counter = 0
        
        for t in trace:
            start_time = datetime.fromisoformat(t['start_time'])
            if start_time.date() == datetime.today().date():
                daily_counter += 1

        return daily_counter