Enh: Integration of prompts from Langfuse

[flefevre]

is there any example of component / code able to retrieve a prompt managed by Langfuse?
there a module in Langfuse that is able to make "prompt management"
the idea is to have an example where the system prompt or the message prompt comes from a call of langfuse prompt

see: https://langfuse.com/docs/prompts/get-started

[chicale]

Hi @flefevre I've used the PromptHub component as reference to do the Langfuse prompt management integration:

import re

from langchain_core.prompts import HumanMessagePromptTemplate

from langflow.custom import Component
from langflow.inputs import DefaultPromptField, SecretStrInput, StrInput
from langflow.io import Output
from langflow.schema.message import Message


class LangfusePromptComponent(Component):
    display_name: str = "Prompt Hub"
    description: str = "Prompt Component that uses Langfuse prompts"
    beta = True
    icon = "prompts"
    trace_type = "prompt"
    name = "Langfuse Prompt"

    inputs = [
        StrInput(
            name="langfuse_prompt_key",
            display_name="Langfuse Prompt Key",
            info="The Langfuse prompt to use, i.e., 'prompt-key'",
            refresh_button=True,
            required=True,
        ),
    ]

    outputs = [
        Output(display_name="Output Prompt", name="prompt", method="build_prompt"),
    ]

    def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None):
        # If the field is not langfuse_prompt_key or the value is empty, return the build config as is
        if field_name != "langfuse_prompt_key" or not field_value:
            return build_config

        # Fetch the template
        template = self._fetch_langfuse_template()

        # Get the template's messages
        if hasattr(template, "messages"):
            template_messages = template.messages
        else:
            template_messages = [HumanMessagePromptTemplate(prompt=template)]

        # Extract the messages from the prompt data
        prompt_template = [message_data.prompt for message_data in template_messages]

        # Regular expression to find all instances of {<string>}
        pattern = r"\{(.*?)\}"

        # Get all the custom fields
        custom_fields: list[str] = []
        full_template = ""
        for message in prompt_template:
            # Find all matches
            matches = re.findall(pattern, message.template)
            custom_fields += matches

            # Create a string version of the full template
            full_template = full_template + "\n" + message.template

        # No need to reprocess if we have them already
        if all("param_" + custom_field in build_config for custom_field in custom_fields):
            return build_config

        # Easter egg: Show template in info popup
        build_config["langfuse_prompt_key"]["info"] = full_template

        # Remove old parameter inputs if any
        for key in build_config.copy():
            if key.startswith("param_"):
                del build_config[key]

        # Now create inputs for each
        for custom_field in custom_fields:
            new_parameter = DefaultPromptField(
                name=f"param_{custom_field}",
                display_name=custom_field,
                info="Fill in the value for {" + custom_field + "}",
            ).to_dict()

            # Add the new parameter to the build config
            build_config[f"param_{custom_field}"] = new_parameter

        return build_config

    async def build_prompt(
        self,
    ) -> Message:
        # Fetch the template
        template = self._fetch_langfuse_template()

        # Get the parameters from the attributes
        params_dict = {param: getattr(self, "param_" + param, f"{{{param}}}") for param in template.input_variables}
        original_params = {k: v.text if hasattr(v, "text") else v for k, v in params_dict.items() if v is not None}
        prompt_value = template.invoke(original_params)

        # Update the template with the new value
        original_params["template"] = prompt_value.to_string()

        # Now pass the filtered attributes to the function
        prompt = Message.from_template(**original_params, metadata={"langfuse_prompt": self.langfuse_prompt_key})

        self.status = prompt.text

        return prompt

    def _fetch_langfuse_template(self):
        from langfuse import Langfuse
        
        langfuse_prompt = Langfuse().get_prompt(self.langfuse_prompt_key)
        langchain_prompt = langfuse_prompt.get_langchain_prompt()
        langflow_prompt = PromptTemplate.from_template(langchain_prompt, metadata={"langfuse_prompt": langfuse_prompt})
        
        return langflow_prompt

[flefevre]

Thanks I will try your proposal.
But I do not see how you have been able to configure the Langfuse server url and access key.

[OUANZOUGUIAbdelhak]

Hello everyone!
I hope you're all doing well! 😊

I used the code below to fetch prompts from Langfuse, and you can also use it to add prompts to Langfuse. All you need to do is replace the public key and the host inside the code, and add the secret key in the input field of the component.

If you want to explore more functionalities with Langfuse, I recommend checking out the documentation here: Langfuse API Reference.

the code :

---

---

`import ssl
import requests
from langflow.custom import Component
from langflow.inputs import DropdownInput, StrInput, BoolInput
from langflow.io import Output
from langflow.schema.message import Message
import base64

ssl._create_default_https_context = ssl._create_unverified_context

class LangfusePromptComponent(Component):
display_name = "Langfuse Prompt Manager"
description = "Fetch and manage Langfuse prompts with version control"
icon = "prompts"
PUBLIC_KEY = "your_public_key_here"
HOST = "http://localhost:3000"

inputs = [
    DropdownInput(
        name="selected_prompt",
        display_name="Select Prompt",
        options=[],
        refresh_button=True,
        advanced=False
    ),
    StrInput(
        name="new_prompt_name",
        display_name="New Prompt Name",
        advanced=False
    ),
    StrInput(
        name="new_prompt_content",
        display_name="Prompt Content",
        advanced=False,
    ),
    DropdownInput(
        name="label",
        display_name="Label",
        options=["production", "staging", "latest"],
        value="production",
        advanced=False
    ),
    BoolInput(
        name="save_trigger",
        display_name="Save to Langfuse",
        advanced=False
    ),
    StrInput(
        name="secret_key",
        display_name="Secret Key",
        advanced=False
    )
]

outputs = [
    Output(display_name="Prompt Content", name="content", method="get_prompt_content"),
    Output(display_name="Save Result", name="save_result", method="handle_save")
]

def _get_auth_header(self):
    """Generate the Basic Auth header using PUBLIC_KEY and SECRET_KEY."""
    credentials = f"{self.PUBLIC_KEY}:{self.secret_key}"
    encoded_credentials = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
    return f"Basic {encoded_credentials}"

def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None):
    """Update the dropdown options for the selected prompt."""
    if field_name == "selected_prompt":
        build_config["selected_prompt"]["options"] = self._fetch_prompts_list()
    return build_config

def _fetch_prompts_list(self):
    """Fetch the list of prompts from the Langfuse API."""
    try:
        response = requests.get(
            f"{self.HOST}/api/public/v2/prompts",
            headers={"Authorization": self._get_auth_header()}
        )
        response.raise_for_status()  # Raise an exception for HTTP errors
        prompts = response.json().get("data", [])
        return [p["name"] for p in prompts] or ["No prompts found"]
    except requests.exceptions.RequestException as e:
        print(f"Prompt fetch error: {e}")
        return ["Error loading prompts"]

def _get_prompt_content(self, promptName: str):
    """Fetch the content of a specific prompt from the Langfuse API."""
    try:
        response = requests.get(
            f"{self.HOST}/api/public/v2/prompts/{promptName}",
            headers={"Authorization": self._get_auth_header()},
            params={"label": self.label, "type": "chat"}  # Specify the prompt type as 'chat'
        )
        response.raise_for_status()  # Raise an exception for HTTP errors
        prompt_data = response.json()
        print(f"Raw prompt data: {prompt_data}")  # Log the raw response for debugging

        # Check if the response contains the expected structure for a chat prompt
        if "prompt" in prompt_data and isinstance(prompt_data["prompt"], list):
            # Extract the chat messages
            chat_messages = prompt_data["prompt"]
            formatted_messages = []
            for message in chat_messages:
                role = message.get("role", "unknown")
                content = message.get("content", "")
                formatted_messages.append(f"{role}: {content}")
            return "\n".join(formatted_messages)  # Return the chat messages as a single string
        else:
            print(f"Unexpected prompt data structure: {prompt_data}")
            return "Error: Unexpected prompt data structure"
    except requests.exceptions.RequestException as e:
        print(f"Content fetch error: {e}")
        return "Error loading content"
    except Exception as e:
        print(f"Unexpected error: {e}")
        return "Error loading content"

def _save_prompt(self, name: str, content: str):
    """Save a new prompt to the Langfuse API."""
    try:
        response = requests.post(
            f"{self.HOST}/api/public/v2/prompts",
            headers={"Authorization": self._get_auth_header()},
            json={
                "name": name,
                "prompt": [{"role": "system", "content": content}],  # Save as a chat prompt
                "labels": [self.label],
                "config": {"temperature": 0.7},
                "type": "chat"  # Specify the prompt type as 'chat'
            }
        )
        response.raise_for_status()  # Raise an exception for HTTP errors
        return f"Prompt saved successfully! Version: {response.json().get('version')}"
    except requests.exceptions.RequestException as e:
        return f"Save failed: {str(e)}"

async def get_prompt_content(self) -> Message:
    """Return the content of the selected prompt as a Message."""
    content = self._get_prompt_content(self.selected_prompt)
    return Message(text=content)

async def handle_save(self) -> Message:
    """Handle saving a new prompt to Langfuse."""
    if self.save_trigger:
        result = self._save_prompt(
            self.new_prompt_name,
            self.new_prompt_content
        )
        return Message(text=result)
    return Message(text="Save not triggered")

`