Add telemetry (#2260)

Author: rhlbhatnagar

src/ragas/_analytics.py

@@ -236,6 +236,33 @@ class IsCompleteEvent(BaseEvent):
     is_completed: bool = True  # True if the event was completed, False otherwise
 
 
+class LLMUsageEvent(BaseEvent):
+    provider: str  # "openai", "anthropic", "langchain", etc.
+    model: t.Optional[str] = None  # Model name (if available)
+    llm_type: str  # "instructor", "langchain_wrapper", "factory"
+    num_requests: int = 1  # Number of API calls
+    is_async: bool = False  # Sync vs async usage
+    event_type: str = "llm_usage"
+
+
+class EmbeddingUsageEvent(BaseEvent):
+    provider: str  # "openai", "google", "huggingface", etc.
+    model: t.Optional[str] = None  # Model name (if available)
+    embedding_type: str  # "modern", "legacy", "factory"
+    num_requests: int = 1  # Number of embed calls
+    is_async: bool = False  # Sync vs async usage
+    event_type: str = "embedding_usage"
+
+
+class PromptUsageEvent(BaseEvent):
+    prompt_type: str  # "pydantic", "few_shot", "simple", "dynamic"
+    has_examples: bool = False  # Whether prompt has few-shot examples
+    num_examples: int = 0  # Number of examples (if applicable)
+    has_response_model: bool = False  # Whether it has a structured response model
+    language: str = "english"  # Prompt language
+    event_type: str = "prompt_usage"
+
+
 @silent
 def track_was_completed(
     func: t.Callable[P, T],

src/ragas/embeddings/base.py

@@ -12,6 +12,7 @@
 from pydantic.dataclasses import dataclass
 from pydantic_core import CoreSchema, core_schema
 
+from ragas._analytics import EmbeddingUsageEvent, track
 from ragas.cache import CacheInterface, cacher
 from ragas.embeddings.utils import run_async_in_current_loop, validate_texts
 from ragas.run_config import RunConfig, add_async_retry, add_retry
@@ -310,25 +311,73 @@ def embed_query(self, text: str) -> t.List[float]:
         """
         Embed a single query text.
         """
-        return self.embeddings.embed_query(text)
+        result = self.embeddings.embed_query(text)
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="langchain",
+                model=getattr(self.embeddings, "model", None),
+                embedding_type="legacy",
+                num_requests=1,
+                is_async=False,
+            )
+        )
+        return result
 
     def embed_documents(self, texts: t.List[str]) -> t.List[t.List[float]]:
         """
         Embed multiple documents.
         """
-        return self.embeddings.embed_documents(texts)
+        result = self.embeddings.embed_documents(texts)
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="langchain",
+                model=getattr(self.embeddings, "model", None),
+                embedding_type="legacy",
+                num_requests=len(texts),
+                is_async=False,
+            )
+        )
+        return result
 
     async def aembed_query(self, text: str) -> t.List[float]:
         """
         Asynchronously embed a single query text.
         """
-        return await self.embeddings.aembed_query(text)
+        result = await self.embeddings.aembed_query(text)
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="langchain",
+                model=getattr(self.embeddings, "model", None),
+                embedding_type="legacy",
+                num_requests=1,
+                is_async=True,
+            )
+        )
+        return result
 
     async def aembed_documents(self, texts: t.List[str]) -> t.List[t.List[float]]:
         """
         Asynchronously embed multiple documents.
         """
-        return await self.embeddings.aembed_documents(texts)
+        result = await self.embeddings.aembed_documents(texts)
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="langchain",
+                model=getattr(self.embeddings, "model", None),
+                embedding_type="legacy",
+                num_requests=len(texts),
+                is_async=True,
+            )
+        )
+        return result
 
     def set_run_config(self, run_config: RunConfig):
         """
@@ -638,10 +687,34 @@ def embedding_factory(
             openai_embeddings.request_timeout = run_config.timeout
         else:
             run_config = RunConfig()
-        return LangchainEmbeddingsWrapper(openai_embeddings, run_config=run_config)
+        result = LangchainEmbeddingsWrapper(openai_embeddings, run_config=run_config)
+
+        # Track factory usage (legacy)
+        track(
+            EmbeddingUsageEvent(
+                provider="openai",
+                model=model_name,
+                embedding_type="factory_legacy",
+                num_requests=1,
+                is_async=False,
+            )
+        )
+        return result
 
     # Modern interface
-    return _create_modern_embedding(provider, model, client, **kwargs)
+    result = _create_modern_embedding(provider, model, client, **kwargs)
+
+    # Track factory usage (modern)
+    track(
+        EmbeddingUsageEvent(
+            provider=provider,
+            model=model,
+            embedding_type="factory_modern",
+            num_requests=1,
+            is_async=False,
+        )
+    )
+    return result
 
 
 def _is_legacy_embedding_call(

src/ragas/embeddings/openai_provider.py

@@ -1,5 +1,7 @@
 import typing as t
 
+from ragas._analytics import EmbeddingUsageEvent, track
+
 from .base import BaseRagasEmbedding
 from .utils import validate_texts
 
@@ -26,12 +28,24 @@ def embed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
         For async clients, this will run the async method in the appropriate event loop.
         """
         if self.is_async:
-            return self._run_async_in_current_loop(self.aembed_text(text, **kwargs))
+            result = self._run_async_in_current_loop(self.aembed_text(text, **kwargs))
         else:
             response = self.client.embeddings.create(
                 input=text, model=self.model, **kwargs
             )
-            return response.data[0].embedding
+            result = response.data[0].embedding
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="openai",
+                model=self.model,
+                embedding_type="modern",
+                num_requests=1,
+                is_async=self.is_async,
+            )
+        )
+        return result
 
     async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
         """Asynchronously embed a single text using OpenAI."""
@@ -43,7 +57,19 @@ async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
         response = await self.client.embeddings.create(
             input=text, model=self.model, **kwargs
         )
-        return response.data[0].embedding
+        result = response.data[0].embedding
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="openai",
+                model=self.model,
+                embedding_type="modern",
+                num_requests=1,
+                is_async=True,
+            )
+        )
+        return result
 
     def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[float]]:
         """Embed multiple texts using OpenAI's batch API for optimization."""
@@ -52,13 +78,25 @@ def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[floa
             return []
 
         if self.is_async:
-            return self._run_async_in_current_loop(self.aembed_texts(texts, **kwargs))
+            result = self._run_async_in_current_loop(self.aembed_texts(texts, **kwargs))
         else:
             # OpenAI supports batch embedding natively
             response = self.client.embeddings.create(
                 input=texts, model=self.model, **kwargs
             )
-            return [item.embedding for item in response.data]
+            result = [item.embedding for item in response.data]
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="openai",
+                model=self.model,
+                embedding_type="modern",
+                num_requests=len(texts),
+                is_async=self.is_async,
+            )
+        )
+        return result
 
     async def aembed_texts(
         self, texts: t.List[str], **kwargs: t.Any
@@ -76,7 +114,19 @@ async def aembed_texts(
         response = await self.client.embeddings.create(
             input=texts, model=self.model, **kwargs
         )
-        return [item.embedding for item in response.data]
+        result = [item.embedding for item in response.data]
+
+        # Track usage
+        track(
+            EmbeddingUsageEvent(
+                provider="openai",
+                model=self.model,
+                embedding_type="modern",
+                num_requests=len(texts),
+                is_async=True,
+            )
+        )
+        return result
 
     def _get_client_info(self) -> str:
         """Get client type and async status information."""