feature(dspy): dspy-integration-with-langfuse

docs/api/language_model_clients/about_prompt_visible.md

@@ -0,0 +1,46 @@
+# Prompt Visiblity through Langfuse
+
+We have now integrated Langfuse as one of the trackers.
+
+How to configure Langfuse?[Langfuse details](https://langfuse.com/docs/deployment/self-host) .
+
+### Install Langfuse.
+
+```shell
+pip install langfuse
+```
+
+Additionally, configure the following environment variables: `LANGFUSE_SECRET_KEY`、`LANGFUSE_PUBLIC_KEY` and `LANGFUSE_HOST`.
+
+If you are using **openai** or **azure_openai**, your LMs are configured.
+
+For other LM providers, you need to configure the Langfuse tracker and call it manually.
+
+### example
+
+```python
+import dspy
+from dsp.trackers.langfuse_tracker import LangfuseTracker
+# e.g:
+# Assuming the environment variables have been set
+langfuse = LangfuseTracker()
+turbo = dspy.OllamaLocal()
+dspy.settings.configure(lm=turbo)
+
+completions =  turbo("Hi,how's it going today？")
+turbo.tracker_call(tracker=langfuse)
+```
+
+## Custom Tracker
+
+We also provide  `BaseTracker`: simply inherit it and override the `call()` method
+```python
+# custom_tracker.py
+from dsp.trackers.base import BaseTracker
+
+class CustomTracker(BaseTracker):
+
+    def call(self, *args, **kwargs):
+        pass
+
+```

dsp/modules/azure_openai.py

@@ -4,7 +4,16 @@
 from typing import Any, Callable, Literal, Optional, cast
 
 import backoff
-import openai
+
+try:
+    """
+    If there is any error in the langfuse configuration, it will turn to request the real address(openai or azure endpoint)
+    """
+    import langfuse
+    from langfuse.openai import openai
+    logging.info(f"You are using Langfuse,version{langfuse.__version__}")
+except:
+    import openai
 
 from dsp.modules.cache_utils import CacheMemory, NotebookCacheMemory, cache_turn_on
 from dsp.modules.lm import LM

dsp/modules/gpt3.py

@@ -2,10 +2,17 @@
 import json
 import logging
 from typing import Any, Literal, Optional, cast
-
 import backoff
-import openai
 import httpx
+try:
+    """
+    If there is any error in the langfuse configuration, it will turn to request the real address(openai or azure endpoint)
+    """
+    import langfuse
+    from langfuse.openai import openai
+    logging.info(f"You are using Langfuse,version{langfuse.__version__}")
+except:
+    import openai
 
 from dsp.modules.cache_utils import CacheMemory, NotebookCacheMemory, cache_turn_on
 from dsp.modules.lm import LM

dsp/modules/lm.py

@@ -4,7 +4,7 @@
 class LM(ABC):
     """Abstract class for language models."""
 
-    def __init__(self, model):
+    def __init__(self, model, tracker=None):
         self.kwargs = {
             "model": model,
             "temperature": 0.0,
@@ -15,6 +15,7 @@ def __init__(self, model):
             "n": 1,
         }
         self.provider = "default"
+        self.tracker = tracker
 
         self.history = []
 
@@ -135,9 +136,23 @@ def inspect_history(self, n: int = 1, skip: int = 0, color_format: bool = True):
     def __call__(self, prompt, only_completed=True, return_sorted=False, **kwargs):
         pass
 
+    def tracker_call(self, tracker, prompt=None, output=None, name=None, **kwargs):
+        from dsp.trackers.base import BaseTracker
+        assert issubclass(tracker.__class__, BaseTracker), "tracker must be a subclass of BaseTracker"
+        assert self.history, "tracker.call() requires a previous request"
+
+        last_req = self.history[-1]
+        if not prompt:
+            prompt = last_req.get('prompt', None)
+        if not output:
+            output = last_req.get('response', None)
+        kwargs = {**self.kwargs, **kwargs}
+        name = name if name else self.__class__.__name__
+        tracker.call(i=prompt, o=output, name=name, **kwargs)
+
     def copy(self, **kwargs):
         """Returns a copy of the language model with the same parameters."""
         kwargs = {**self.kwargs, **kwargs}
         model = kwargs.pop("model")
 
-        return self.__class__(model=model, **kwargs)
+        return self.__class__(model=model, **kwargs)

dsp/modules/ollama.py

@@ -92,18 +92,18 @@ def basic_request(self, prompt: str, **kwargs):
             "options": {k: v for k, v in kwargs.items() if k not in ["n", "max_tokens"]},
             "stream": False,
         }
-        
+
         # Set the format if it was defined
         if self.format:
             settings_dict["format"] = self.format
-            
+
         if self.model_type == "chat":
             settings_dict["messages"] = [{"role": "user", "content": prompt}]
         else:
             # Overwrite system prompt defined in modelfile
             if self.system:
                 settings_dict["system"] = self.system
-    
+
             settings_dict["prompt"] = prompt
 
         urlstr = f"{self.base_url}/api/chat" if self.model_type == "chat" else f"{self.base_url}/api/generate"
@@ -195,7 +195,7 @@ def __call__(
         completions = [self._get_choice_text(c) for c in choices]
 
         return completions
-    
+
     def copy(self, **kwargs):
         """Returns a copy of the language model with the same parameters."""
         kwargs = {**self.kwargs, **kwargs}
@@ -206,4 +206,4 @@ def copy(self, **kwargs):
             base_url=self.base_url,
             timeout_s=self.timeout_s,
             **kwargs,
-        )
+        )

dsp/trackers/base.py

@@ -0,0 +1,8 @@
+
+class BaseTracker:
+    def __init__(self):
+        pass
+
+    @classmethod
+    def call(cls, *args, **kwargs):
+        pass

dsp/trackers/langfuse_tracker.py

@@ -0,0 +1,87 @@
+from typing import Optional, Union, List, Any
+import httpx
+import logging
+import os
+from langfuse.client import Langfuse, StatefulTraceClient, StatefulSpanClient, StateType
+from dsp.trackers.base import BaseTracker
+
+
+class LangfuseTracker(BaseTracker):
+    log = logging.getLogger("langfuse")
+
+    def __init__(self, *, public_key: Optional[str] = None, secret_key: Optional[str] = None,
+                 host: Optional[str] = None, debug: bool = False, stateful_client: Optional[
+                Union[StatefulTraceClient, StatefulSpanClient]
+            ] = None, update_stateful_client: bool = False, version: Optional[str] = None,
+                 session_id: Optional[str] = None, user_id: Optional[str] = None, trace_name: Optional[str] = None,
+                 release: Optional[str] = None, metadata: Optional[Any] = None, tags: Optional[List[str]] = None,
+                 threads: Optional[int] = None, flush_at: Optional[int] = None, flush_interval: Optional[int] = None,
+                 max_retries: Optional[int] = None, timeout: Optional[int] = None, enabled: Optional[bool] = None,
+                 httpx_client: Optional[httpx.Client] = None, sdk_integration: str = "default") -> None:
+        super().__init__()
+        self.version = version
+        self.session_id = session_id
+        self.user_id = user_id
+        self.trace_name = trace_name
+        self.release = release
+        self.metadata = metadata
+        self.tags = tags
+
+        self.root_span = None
+        self.update_stateful_client = update_stateful_client
+        self.langfuse = None
+
+        prio_public_key = public_key or os.environ.get("LANGFUSE_PUBLIC_KEY")
+        prio_secret_key = secret_key or os.environ.get("LANGFUSE_SECRET_KEY")
+        prio_host = host or os.environ.get(
+            "LANGFUSE_HOST", "https://cloud.langfuse.com"
+        )
+
+        if stateful_client and isinstance(stateful_client, StatefulTraceClient):
+            self.trace = stateful_client
+            self._task_manager = stateful_client.task_manager
+            return
+
+        elif stateful_client and isinstance(stateful_client, StatefulSpanClient):
+            self.root_span = stateful_client
+            self.trace = StatefulTraceClient(
+                stateful_client.client,
+                stateful_client.trace_id,
+                StateType.TRACE,
+                stateful_client.trace_id,
+                stateful_client.task_manager,
+            )
+            self._task_manager = stateful_client.task_manager
+            return
+
+        args = {
+            "public_key": prio_public_key,
+            "secret_key": prio_secret_key,
+            "host": prio_host,
+            "debug": debug,
+        }
+
+        if release is not None:
+            args["release"] = release
+        if threads is not None:
+            args["threads"] = threads
+        if flush_at is not None:
+            args["flush_at"] = flush_at
+        if flush_interval is not None:
+            args["flush_interval"] = flush_interval
+        if max_retries is not None:
+            args["max_retries"] = max_retries
+        if timeout is not None:
+            args["timeout"] = timeout
+        if enabled is not None:
+            args["enabled"] = enabled
+        if httpx_client is not None:
+            args["httpx_client"] = httpx_client
+        args["sdk_integration"] = sdk_integration
+
+        self.langfuse = Langfuse(**args)
+        self.trace: Optional[StatefulTraceClient] = None
+        self._task_manager = self.langfuse.task_manager
+
+    def call(self, i, o, name=None, **kwargs):
+        self.langfuse.trace(input=i, output=o, name=name, metadata=kwargs)

requirements.txt

@@ -10,4 +10,5 @@ requests
 structlog
 tqdm
 ujson
+httpx
 magicattr~=0.1.6

setup.py

@@ -38,7 +38,8 @@
         "snowflake": ["snowflake-snowpark-python"],
         "fastembed": ["fastembed"],
         "groq": ["groq~=0.8.0"],
-    },	
+        "langfuse": ["langfuse~=2.36.1"]
+    },
     classifiers=[	
         "Development Status :: 3 - Alpha",	
         "Intended Audience :: Science/Research",