langfuse.model

  1"""@private"""
  2
  3import re
  4from abc import ABC, abstractmethod
  5from typing import Any, Dict, List, Optional, Tuple, TypedDict, Union
  6
  7from langfuse.api.resources.commons.types.dataset import (
  8    Dataset,  # noqa: F401
  9)
 10
 11# these imports need to stay here, otherwise imports from our clients wont work
 12from langfuse.api.resources.commons.types.dataset_item import DatasetItem  # noqa: F401
 13
 14# noqa: F401
 15from langfuse.api.resources.commons.types.dataset_run import DatasetRun  # noqa: F401
 16
 17# noqa: F401
 18from langfuse.api.resources.commons.types.dataset_status import (  # noqa: F401
 19    DatasetStatus,
 20)
 21from langfuse.api.resources.commons.types.map_value import MapValue  # noqa: F401
 22from langfuse.api.resources.commons.types.observation import Observation  # noqa: F401
 23from langfuse.api.resources.commons.types.trace_with_full_details import (  # noqa: F401
 24    TraceWithFullDetails,
 25)
 26
 27# noqa: F401
 28from langfuse.api.resources.dataset_items.types.create_dataset_item_request import (  # noqa: F401
 29    CreateDatasetItemRequest,
 30)
 31from langfuse.api.resources.dataset_run_items.types.create_dataset_run_item_request import (  # noqa: F401
 32    CreateDatasetRunItemRequest,
 33)
 34
 35# noqa: F401
 36from langfuse.api.resources.datasets.types.create_dataset_request import (  # noqa: F401
 37    CreateDatasetRequest,
 38)
 39from langfuse.api.resources.prompts import ChatMessage, Prompt, Prompt_Chat, Prompt_Text
 40
 41
 42class ModelUsage(TypedDict):
 43    unit: Optional[str]
 44    input: Optional[int]
 45    output: Optional[int]
 46    total: Optional[int]
 47    input_cost: Optional[float]
 48    output_cost: Optional[float]
 49    total_cost: Optional[float]
 50
 51
 52class ChatMessageDict(TypedDict):
 53    role: str
 54    content: str
 55
 56
 57class TemplateParser:
 58    OPENING = "{{"
 59    CLOSING = "}}"
 60
 61    @staticmethod
 62    def _parse_next_variable(
 63        content: str, start_idx: int
 64    ) -> Optional[Tuple[str, int, int]]:
 65        """Returns (variable_name, start_pos, end_pos) or None if no variable found"""
 66        var_start = content.find(TemplateParser.OPENING, start_idx)
 67        if var_start == -1:
 68            return None
 69
 70        var_end = content.find(TemplateParser.CLOSING, var_start)
 71        if var_end == -1:
 72            return None
 73
 74        variable_name = content[
 75            var_start + len(TemplateParser.OPENING) : var_end
 76        ].strip()
 77        return (variable_name, var_start, var_end + len(TemplateParser.CLOSING))
 78
 79    @staticmethod
 80    def find_variable_names(content: str) -> List[str]:
 81        names = []
 82        curr_idx = 0
 83
 84        while curr_idx < len(content):
 85            result = TemplateParser._parse_next_variable(content, curr_idx)
 86            if not result:
 87                break
 88            names.append(result[0])
 89            curr_idx = result[2]
 90
 91        return names
 92
 93    @staticmethod
 94    def compile_template(content: str, data: Optional[Dict[str, Any]] = None) -> str:
 95        if data is None:
 96            return content
 97
 98        result_list = []
 99        curr_idx = 0
100
101        while curr_idx < len(content):
102            result = TemplateParser._parse_next_variable(content, curr_idx)
103
104            if not result:
105                result_list.append(content[curr_idx:])
106                break
107
108            variable_name, var_start, var_end = result
109            result_list.append(content[curr_idx:var_start])
110
111            if variable_name in data:
112                result_list.append(
113                    str(data[variable_name]) if data[variable_name] is not None else ""
114                )
115            else:
116                result_list.append(content[var_start:var_end])
117
118            curr_idx = var_end
119
120        return "".join(result_list)
121
122
123class BasePromptClient(ABC):
124    name: str
125    version: int
126    config: Dict[str, Any]
127    labels: List[str]
128    tags: List[str]
129    commit_message: Optional[str]
130
131    def __init__(self, prompt: Prompt, is_fallback: bool = False):
132        self.name = prompt.name
133        self.version = prompt.version
134        self.config = prompt.config
135        self.labels = prompt.labels
136        self.tags = prompt.tags
137        self.commit_message = prompt.commit_message
138        self.is_fallback = is_fallback
139
140    @abstractmethod
141    def compile(self, **kwargs) -> Union[str, List[ChatMessage]]:
142        pass
143
144    @property
145    @abstractmethod
146    def variables(self) -> List[str]:
147        pass
148
149    @abstractmethod
150    def __eq__(self, other):
151        pass
152
153    @abstractmethod
154    def get_langchain_prompt(self):
155        pass
156
157    @staticmethod
158    def _get_langchain_prompt_string(content: str):
159        return re.sub(r"{{\s*(\w+)\s*}}", r"{\g<1>}", content)
160
161
162class TextPromptClient(BasePromptClient):
163    def __init__(self, prompt: Prompt_Text, is_fallback: bool = False):
164        super().__init__(prompt, is_fallback)
165        self.prompt = prompt.prompt
166
167    def compile(self, **kwargs) -> str:
168        return TemplateParser.compile_template(self.prompt, kwargs)
169
170    @property
171    def variables(self) -> List[str]:
172        """Return all the variable names in the prompt template."""
173        return TemplateParser.find_variable_names(self.prompt)
174
175    def __eq__(self, other):
176        if isinstance(self, other.__class__):
177            return (
178                self.name == other.name
179                and self.version == other.version
180                and self.prompt == other.prompt
181                and self.config == other.config
182            )
183
184        return False
185
186    def get_langchain_prompt(self, **kwargs) -> str:
187        """Convert Langfuse prompt into string compatible with Langchain PromptTemplate.
188
189        This method adapts the mustache-style double curly braces {{variable}} used in Langfuse
190        to the single curly brace {variable} format expected by Langchain.
191
192        kwargs: Optional keyword arguments to precompile the template string. Variables that match
193                the provided keyword arguments will be precompiled. Remaining variables must then be
194                handled by Langchain's prompt template.
195
196        Returns:
197            str: The string that can be plugged into Langchain's PromptTemplate.
198        """
199        prompt = (
200            TemplateParser.compile_template(self.prompt, kwargs)
201            if kwargs
202            else self.prompt
203        )
204
205        return self._get_langchain_prompt_string(prompt)
206
207
208class ChatPromptClient(BasePromptClient):
209    def __init__(self, prompt: Prompt_Chat, is_fallback: bool = False):
210        super().__init__(prompt, is_fallback)
211        self.prompt = [
212            ChatMessageDict(role=p.role, content=p.content) for p in prompt.prompt
213        ]
214
215    def compile(self, **kwargs) -> List[ChatMessageDict]:
216        return [
217            ChatMessageDict(
218                content=TemplateParser.compile_template(
219                    chat_message["content"], kwargs
220                ),
221                role=chat_message["role"],
222            )
223            for chat_message in self.prompt
224        ]
225
226    @property
227    def variables(self) -> List[str]:
228        """Return all the variable names in the chat prompt template."""
229        return [
230            variable
231            for chat_message in self.prompt
232            for variable in TemplateParser.find_variable_names(chat_message["content"])
233        ]
234
235    def __eq__(self, other):
236        if isinstance(self, other.__class__):
237            return (
238                self.name == other.name
239                and self.version == other.version
240                and all(
241                    m1["role"] == m2["role"] and m1["content"] == m2["content"]
242                    for m1, m2 in zip(self.prompt, other.prompt)
243                )
244                and self.config == other.config
245            )
246
247        return False
248
249    def get_langchain_prompt(self, **kwargs):
250        """Convert Langfuse prompt into string compatible with Langchain ChatPromptTemplate.
251
252        It specifically adapts the mustache-style double curly braces {{variable}} used in Langfuse
253        to the single curly brace {variable} format expected by Langchain.
254
255        kwargs: Optional keyword arguments to precompile the template string. Variables that match
256                the provided keyword arguments will be precompiled. Remaining variables must then be
257                handled by Langchain's prompt template.
258
259        Returns:
260            List of messages in the format expected by Langchain's ChatPromptTemplate: (role, content) tuple.
261        """
262        return [
263            (
264                msg["role"],
265                self._get_langchain_prompt_string(
266                    TemplateParser.compile_template(msg["content"], kwargs)
267                    if kwargs
268                    else msg["content"]
269                ),
270            )
271            for msg in self.prompt
272        ]
273
274
275PromptClient = Union[TextPromptClient, ChatPromptClient]