langchain llamaindex callback handler

portkey_ai/api_resources/apis/logger.py

@@ -0,0 +1,34 @@
+import json
+import os
+from typing import Optional
+import requests
+
+from portkey_ai.api_resources.global_constants import PORTKEY_BASE_URL
+
+
+class Logger:
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+    ) -> None:
+        api_key = api_key or os.getenv("PORTKEY_API_KEY")
+
+        self.headers = {
+            "Content-Type": "application/json",
+            "x-portkey-api-key": api_key,
+        }
+
+        self.url = PORTKEY_BASE_URL + "/logs"
+
+        if api_key is None:
+            raise ValueError("API key is required to use the Logger API")
+
+    def log(
+        self,
+        log_object: dict,
+    ):
+        response = requests.post(
+            url=self.url, data=json.dumps(log_object), headers=self.headers
+        )
+
+        return response

portkey_ai/llms/langchain/portkey_langchain_callback.py

@@ -0,0 +1,169 @@
+from datetime import datetime
+import time
+from typing import Any, Dict, List, Optional
+from langchain_core.callbacks import BaseCallbackHandler
+from langchain_core.outputs import LLMResult
+from langchain_core.agents import AgentFinish, AgentAction
+
+from portkey_ai.api_resources.apis.logger import Logger
+
+
+class PortkeyCallbackHandler(BaseCallbackHandler):
+    def __init__(
+        self,
+        api_key: str,
+    ) -> None:
+        super().__init__()
+        self.startTimestamp: float = 0
+        self.endTimestamp: float = 0
+
+        self.api_key = api_key
+
+        self.portkey_logger = Logger(api_key=api_key)
+
+        self.log_object: Dict[str, Any] = {}
+        self.prompt_records: Any = []
+
+        self.request: Any = {}
+        self.response: Any = {}
+
+        # self.responseHeaders: Dict[str, Any] = {}
+        self.responseBody: Any = None
+        self.responseStatus: int = 0
+
+        self.streamingMode: bool = False
+
+        if not api_key:
+            raise ValueError("Please provide an API key to use PortkeyCallbackHandler")
+
+    def on_llm_start(
+        self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
+    ) -> None:
+        for prompt in prompts:
+            messages = prompt.split("\n")
+            for message in messages:
+                role, content = message.split(":", 1)
+                self.prompt_records.append(
+                    {"role": role.lower(), "content": content.strip()}
+                )
+
+        self.startTimestamp = float(datetime.now().timestamp())
+
+        self.streamingMode = kwargs.get("invocation_params", False).get("stream", False)
+
+        self.request["method"] = "POST"
+        self.request["url"] = serialized.get("kwargs", "").get(
+            "base_url", "https://api.openai.com/v1/chat/completions"
+        )  # placeholder
+        self.request["provider"] = serialized["id"][2]
+        self.request["headers"] = serialized.get("kwargs", {}).get(
+            "default_headers", {}
+        )
+        self.request["headers"].update({"provider": serialized["id"][2]})
+        self.request["body"] = {"messages": self.prompt_records}
+        self.request["body"].update({**kwargs.get("invocation_params", {})})
+
+    def on_chain_start(
+        self,
+        serialized: Dict[str, Any],
+        inputs: Dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        """Run when chain starts running."""
+
+    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
+        self.endTimestamp = float(datetime.now().timestamp())
+        responseTime = self.endTimestamp - self.startTimestamp
+
+        usage = response.llm_output.get("token_usage", {})  # type: ignore[union-attr]
+
+        self.response["status"] = (
+            200 if self.responseStatus == 0 else self.responseStatus
+        )
+        self.response["body"] = {
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": response.generations[0][0].text,
+                    },
+                    "logprobs": response.generations[0][0].generation_info["logprobs"],  # type: ignore[index]
+                    "finish_reason": response.generations[0][0].generation_info["finish_reason"],  # type: ignore[index] # noqa: E501
+                }
+            ]
+        }
+        self.response["body"].update({"usage": usage})
+        self.response["body"].update({"id": str(kwargs.get("run_id", ""))})
+        self.response["body"].update({"created": int(time.time())})
+        self.response["body"].update({"model": response.llm_output.get("model_name", "")})  # type: ignore[union-attr] # noqa: E501
+        self.response["body"].update({"system_fingerprint": response.llm_output.get("system_fingerprint", "")})  # type: ignore[union-attr] # noqa: E501
+        self.response["time"] = int(responseTime * 1000)
+        self.response["headers"] = {}
+        self.response["streamingMode"] = self.streamingMode
+
+        self.log_object.update(
+            {
+                "request": self.request,
+                "response": self.response,
+            }
+        )
+
+        self.portkey_logger.log(log_object=self.log_object)
+
+    def on_chain_end(
+        self,
+        outputs: Dict[str, Any],
+        **kwargs: Any,
+    ) -> None:
+        """Run when chain ends running."""
+        pass
+
+    def on_chain_error(self, error: BaseException, **kwargs: Any) -> None:
+        self.responseBody = error
+        self.responseStatus = error.status_code  # type: ignore[attr-defined]
+        """Do nothing."""
+        pass
+
+    def on_llm_error(self, error: BaseException, **kwargs: Any) -> None:
+        self.responseBody = error
+        self.responseStatus = error.status_code  # type: ignore[attr-defined]
+        """Do nothing."""
+        pass
+
+    def on_tool_error(self, error: BaseException, **kwargs: Any) -> None:
+        self.responseBody = error
+        self.responseStatus = error.status_code  # type: ignore[attr-defined]
+        pass
+
+    def on_text(self, text: str, **kwargs: Any) -> None:
+        pass
+
+    def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> None:
+        pass
+
+    def on_llm_new_token(self, token: str, **kwargs: Any) -> None:
+        self.streamingMode = True
+        """Do nothing."""
+        pass
+
+    def on_tool_start(
+        self,
+        serialized: Dict[str, Any],
+        input_str: str,
+        **kwargs: Any,
+    ) -> None:
+        pass
+
+    def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
+        """Do nothing."""
+        pass
+
+    def on_tool_end(
+        self,
+        output: Any,
+        observation_prefix: Optional[str] = None,
+        llm_prefix: Optional[str] = None,
+        **kwargs: Any,
+    ) -> None:
+        pass

portkey_ai/llms/llama_index/__init__.py

@@ -1,3 +1,3 @@
-from .completions import PortkeyLLM
+# from .completions import PortkeyLLM
 
-__all__ = ["PortkeyLLM"]
+# __all__ = ["PortkeyLLM"]

portkey_ai/llms/llama_index/portkey_llama_callback.py

@@ -0,0 +1,146 @@
+import time
+from typing import Any, Dict, List, Optional
+from llama_index.core.callbacks.base_handler import (
+    BaseCallbackHandler as LlamaIndexBaseCallbackHandler,
+)
+
+from portkey_ai.api_resources.apis.logger import Logger
+from datetime import datetime
+from llama_index.core.callbacks.schema import CBEventType, EventPayload
+from llama_index.core.utilities.token_counting import TokenCounter
+
+
+class PortkeyCallbackHandler(LlamaIndexBaseCallbackHandler):
+    startTimestamp: int = 0
+    endTimestamp: float = 0
+
+    def __init__(
+        self,
+        api_key: str,
+    ) -> None:
+        super().__init__(
+            event_starts_to_ignore=[],
+            event_ends_to_ignore=[],
+        )
+
+        self.api_key = api_key
+
+        self.portkey_logger = Logger(api_key=api_key)
+
+        self._token_counter = TokenCounter()
+        self.token_llm = 0
+
+        self.log_object: Dict[str, Any] = {}
+        self.prompt_records: Any = []
+
+        self.request: Any = {}
+        self.response: Any = {}
+
+        self.responseTime: int = 0
+        self.streamingMode: bool = False
+
+        if not api_key:
+            raise ValueError("Please provide an API key to use PortkeyCallbackHandler")
+
+    def on_event_start(  # type: ignore[return]
+        self,
+        event_type: CBEventType,
+        payload: Optional[Dict[str, Any]] = None,
+        event_id: str = "",
+        parent_id: str = "",
+        **kwargs: Any,
+    ) -> str:
+        """Run when an event starts and return id of event."""
+
+        if event_type == CBEventType.LLM:
+            self.llm_event_start(payload)
+
+    def on_event_end(
+        self,
+        event_type: CBEventType,
+        payload: Optional[Dict[str, Any]] = None,
+        event_id: str = "",
+        **kwargs: Any,
+    ) -> None:
+        """Run when an event ends."""
+
+        if event_type == CBEventType.LLM:
+            self.llm_event_stop(payload, event_id)
+
+    def start_trace(self, trace_id: Optional[str] = None) -> None:
+        """Run when an overall trace is launched."""
+        self.startTimestamp = int(datetime.now().timestamp())
+
+    def end_trace(
+        self,
+        trace_id: Optional[str] = None,
+        trace_map: Optional[Dict[str, List[str]]] = None,
+    ) -> None:
+        """Run when an overall trace is exited."""
+
+    def llm_event_start(self, payload: Any) -> None:
+        if EventPayload.MESSAGES in payload:
+            messages = payload.get(EventPayload.MESSAGES, {})
+            self.prompt_records = [
+                {"role": m.role.value, "content": m.content} for m in messages
+            ]
+        self.request["method"] = "POST"
+        self.request["url"] = payload.get(EventPayload.SERIALIZED, {}).get(
+            "base_url", "https://api.openai.com/v1/chat/completions"
+        )
+        self.request["provider"] = "openai"  # placeholder
+        self.request["headers"] = {}
+        self.request["body"] = {"messages": self.prompt_records}
+        self.request["body"].update(
+            {"model": payload.get(EventPayload.SERIALIZED, {}).get("model", "")}
+        )
+        self.request["body"].update(
+            {
+                "temperature": payload.get(EventPayload.SERIALIZED, {}).get(
+                    "temperature", ""
+                )
+            }
+        )
+
+        return None
+
+    def llm_event_stop(self, payload: Any, event_id) -> None:
+        self.endTimestamp = float(datetime.now().timestamp())
+        responseTime = self.endTimestamp - self.startTimestamp
+
+        data = payload.get(EventPayload.RESPONSE, {})
+
+        chunks = payload.get(EventPayload.MESSAGES, {})
+        self.token_llm = self._token_counter.estimate_tokens_in_messages(chunks)
+
+        self.response["status"] = 200
+        self.response["body"] = {
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": data.message.role.value,
+                        "content": data.message.content,
+                    },
+                    "logprobs": data.logprobs,
+                    "finish_reason": "done",
+                }
+            ]
+        }
+        self.response["body"].update({"usage": {"total_tokens": self.token_llm}})
+        self.response["body"].update({"id": event_id})
+        self.response["body"].update({"created": int(time.time())})
+        self.response["body"].update({"model": data.raw.get("model", "")})
+        self.response["time"] = int(responseTime * 1000)
+        self.response["headers"] = {}
+        self.response["streamingMode"] = self.streamingMode
+
+        self.log_object.update(
+            {
+                "request": self.request,
+                "response": self.response,
+            }
+        )
+        self.portkey_logger.log(log_object=self.log_object)
+
+        return None