https://us.cloud.langfuse.com/

langchain + langfuse

Langfuse가입 후 Project Setting을 진행할 수 있습니다.

API 키 획득 후 기존 langchain 에이전트에 연결해보겠습니다.

https://langfuse.com/docs/integrations/langchain/tracing

langchain과 비슷하게 환경변수만으로도 langchain 클라우드와 연결할 수 있습니다. 다만 한가지 절차가 더 필요한데, 핸들러 객체를 만들어 langchain의 invoke 메서드에 콜백으로 넘겨주면 됩니다.

프롬프트 관리

langfuse는 SDK를 이용해 프롬프트와 에이전트를 분리시킬 수도 있지만, GUI환경의 클라우드를 통해 더 쉽게 프롬프트를 관리할 수도 있습니다.

langfuse의 프롬프트 langchain에이전트에 연결하기

    # langfuse 핸들러 초기화
    langfuse_handler = CallbackHandler()
    langfuse = Langfuse()

    # langfuse 클라우드에서 production 라벨의 최신 프롬프트 가져오기
    langfuse_prompt = langfuse.get_prompt("jokerPrompt", label="production")
    langfuse_prompt_template = langfuse_prompt.get_langchain_prompt()

    print(f"주입받은 lanfuse 프롬프트: {langfuse_prompt_template}")

    # 가져온 프롬프트를 langchain 프롬프트로 변경
    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template(langfuse_prompt_template),
    ]
    )

위의 과정을 거쳐 만들어진 prompt 객체를 이전처럼 langchain 체인안에서 사용하면 된다.

langfuse + langflow

https://langfuse.com/blog/langflow?utm_source=chatgpt.com

from langflow.custom import Component
from langflow.io import MessageTextInput, Output, SecretStrInput, IntInput
from langflow.schema import Data, Message
from langfuse import Langfuse

class LangfusePromptComponent(Component):
    display_name = "Langfuse Prompt Fetcher"
    description = "Fetch prompts from Langfuse prompt management."
    documentation: str = "<https://langfuse.com/docs/prompts>"
    icon = "database"
    name = "LangfusePromptComponent"
    
    inputs = [
        SecretStrInput(
            name="langfuse_secret_key",
            display_name="Langfuse Secret Key",
            info="Your Langfuse secret key for authentication",
            required=True,
        ),
        MessageTextInput(
            name="langfuse_public_key",
            display_name="Langfuse Public Key",
            info="Your Langfuse public key for authentication",
            required=True,
        ),
        MessageTextInput(
            name="langfuse_host",
            display_name="Langfuse Host",
            info="Langfuse host URL (default: <https://cloud.langfuse.com>)",
            value="<https://cloud.langfuse.com>",
            required=False,
        ),
        MessageTextInput(
            name="prompt_name",
            display_name="Prompt Name",
            info="Name of the prompt to fetch from Langfuse",
            required=True,
        ),
        IntInput(
            name="prompt_version",
            display_name="Prompt Version",
            info="Version of the prompt (optional, latest if not specified)",
            required=False,
        ),
        MessageTextInput(
            name="prompt_label",
            display_name="Prompt Label",
            info="Label of the prompt (optional, e.g., 'production')",
            required=False,
        ),
    ]
    
    outputs = [
        Output(display_name="Prompt", name="prompt", method="build_prompt"),
        Output(display_name="Debug Info", name="debug", method="build_prompt")
    ]

    def build_prompt(self) -> Message:
        """Fetch prompt from Langfuse and return as Message."""
        try:
            self.log("fetching start")
            # Initialize Langfuse client
            langfuse = Langfuse(
                secret_key=self.langfuse_secret_key,
                public_key=self.langfuse_public_key,
                host=self.langfuse_host or "<https://cloud.langfuse.com>"
            )
            self.log(f"langfuse: {langfuse}")
            try:
                langfuse.auth_check()
                self.log("langfuse 연결 성공")
            except Exception as e :
                self.log("langfuse 연결 실패")
                
            
            # Prepare parameters for fetching prompt
            fetch_params = {"name": self.prompt_name}
            if hasattr(self, 'prompt_version') and self.prompt_version:
                fetch_params["version"] = self.prompt_version
            if hasattr(self, 'prompt_label') and self.prompt_label:
                fetch_params["label"] = self.prompt_label
            
            # Fetch prompt from Langfuse
            prompt = langfuse.get_prompt(fetch_params["name"], label=fetch_params["label"])
            self.log(f"langfuse prompt: {prompt.get_langchain_prompt()}")
            if not prompt:
                self.log("prompt fetching failed!")
                raise ValueError(f"Prompt '{self.prompt_name}' not found in Langfuse")
            
            # Set status message
            version_info = f"v{prompt.version}" if hasattr(prompt, 'version') else "latest"
            self.status = f"Successfully fetched prompt '{self.prompt_name}' ({version_info})"
            
            # For debug output, return the raw prompt info
            if self.current_output == "debug":
                return Message.from_template(
                    f"Prompt Info:\\nName: {prompt.name}\\nVersion: {version_info}\\nTemplate: {prompt.prompt}"
                )
            
            # For normal output, return the prompt template
            return Message.from_template(prompt.prompt)
            
        except Exception as e:
            error_msg = f"Error fetching prompt from Langfuse: {str(e)}"
            self.status = error_msg
            return Message.from_template(error_msg)