diff --git a/langfuse/model.py b/langfuse/model.py index 30c6fe0f3..fa4d8e5d0 100644 --- a/langfuse/model.py +++ b/langfuse/model.py @@ -157,10 +157,28 @@ def get_langchain_prompt(self) -> Any: pass @staticmethod - def _get_langchain_prompt_string(content: str) -> str: - json_escaped_content = BasePromptClient._escape_json_for_langchain(content) - - return re.sub(r"{{\s*(\w+)\s*}}", r"{\g<1>}", json_escaped_content) + def _get_langchain_prompt_string( + content: str, data: Optional[Dict[str, Any]] = None + ) -> str: + supplied = data or {} + values = { + name: "" if supplied[name] is None else str(supplied[name]) + for name in TemplateParser.find_variable_names(content) + if name in supplied + } + marker = "\ue000" + while marker in content or any(marker in value for value in values.values()): + marker = chr(ord(marker) + 1) + # Keep supplied braces opaque to template parsing, while leaving quotes + # visible for JSON examples completed by a supplied value. + protected_values = { + name: value.replace("{", marker + "open").replace("}", marker + "close") + for name, value in values.items() + } + compiled = TemplateParser.compile_template(content, protected_values) + json_escaped_content = BasePromptClient._escape_json_for_langchain(compiled) + converted = re.sub(r"{{\s*(\w+)\s*}}", r"{\g<1>}", json_escaped_content) + return converted.replace(marker + "open", "{{").replace(marker + "close", "}}") @staticmethod def _escape_json_for_langchain(text: str) -> str: @@ -266,13 +284,7 @@ def get_langchain_prompt(self, **kwargs: Union[str, Any]) -> str: Returns: str: The string that can be plugged into Langchain's PromptTemplate. """ - prompt = ( - TemplateParser.compile_template(self.prompt, kwargs) - if kwargs - else self.prompt - ) - - return self._get_langchain_prompt_string(prompt) + return self._get_langchain_prompt_string(self.prompt, kwargs) class ChatPromptClient(BasePromptClient): @@ -443,7 +455,14 @@ def get_langchain_prompt( List of messages in the format expected by Langchain's ChatPromptTemplate: (role, content) tuples for regular messages or MessagesPlaceholder objects for unresolved placeholders. """ - compiled_messages = self.compile(**kwargs) + placeholder_names = { + message["name"] + for message in self.prompt + if message["type"] == "placeholder" + } + compiled_messages = self.compile( + **{key: value for key, value in kwargs.items() if key in placeholder_names} + ) langchain_messages: List[Union[Tuple[str, str], Any]] = [] for msg in compiled_messages: @@ -464,7 +483,7 @@ def get_langchain_prompt( langchain_messages.append( ( msg["role"], # type: ignore - self._get_langchain_prompt_string(msg["content"]), # type: ignore + self._get_langchain_prompt_string(msg["content"], kwargs), # type: ignore ), ) diff --git a/tests/unit/test_prompt_compilation.py b/tests/unit/test_prompt_compilation.py index 1b96a14dd..0c3fae821 100644 --- a/tests/unit/test_prompt_compilation.py +++ b/tests/unit/test_prompt_compilation.py @@ -932,3 +932,157 @@ def test_tool_calls_preservation_in_message_placeholder(): # Final user message with compiled variable assert compiled_messages[4]["role"] == "user" assert compiled_messages[4]["content"] == "Help me with weather inquiry" + + +@pytest.mark.parametrize("prompt_type", ["text", "chat"]) +@pytest.mark.parametrize( + "context", + [ + "Use {name}", + "Use {{name}}", + "Compare {question}", + "An unmatched { brace", + "Private \ue000 and NUL \0 with {name}", + 'JSON: {"name": "Ada"}', + "", + None, + 42, + ], +) +def test_precompiled_values_remain_literal_in_langchain(prompt_type, context): + source = "Context: {{context}}\nQuestion: {{question}}" + fields = dict(name="literal_context", version=1, config={}, tags=[], labels=[]) + if prompt_type == "text": + client = TextPromptClient(Prompt_Text(type="text", prompt=source, **fields)) + template = PromptTemplate.from_template( + client.get_langchain_prompt(context=context) + ) + rendered = template.format(question="Hello") + else: + client = ChatPromptClient( + Prompt_Chat( + type="chat", prompt=[ChatMessage(role="user", content=source)], **fields + ) + ) + template = ChatPromptTemplate.from_messages( + client.get_langchain_prompt(context=context) + ) + rendered = template.format_messages(question="Hello")[0].content + assert template.input_variables == ["question"] + literal_context = "" if context is None else str(context) + assert rendered == f"Context: {literal_context}\nQuestion: Hello" + + +def test_precompiled_history_values_remain_literal_in_langchain(): + client = ChatPromptClient( + Prompt_Chat( + type="chat", + name="history_literals", + version=1, + config={}, + tags=[], + labels=[], + prompt=[ + {"type": "placeholder", "name": "history"}, + ChatMessage(role="user", content="{{question}}"), + ], + ) + ) + history = [{"role": "system", "content": "Context: {{context}}"}] + messages = client.get_langchain_prompt(history=history, context="Keep {{question}}") + template = ChatPromptTemplate.from_messages(messages) + rendered = template.format_messages(question="Hello") + assert rendered[0].content == "Context: Keep {{question}}" + assert rendered[1].content == "Hello" + assert history == [{"role": "system", "content": "Context: {{context}}"}] + + +@pytest.mark.parametrize("prompt_type", ["text", "chat"]) +def test_precompiled_non_word_names_keep_working(prompt_type): + fields = dict(name="known_names", version=1, config={}, tags=[], labels=[]) + source = "Context: {{context-data}} / {{user name}} / {{question}}" + values = {"context-data": "Keep {name}", "user name": "Ada"} + if prompt_type == "text": + client = TextPromptClient(Prompt_Text(type="text", prompt=source, **fields)) + template = PromptTemplate.from_template(client.get_langchain_prompt(**values)) + rendered = template.format(question="Hello") + else: + client = ChatPromptClient( + Prompt_Chat( + type="chat", prompt=[ChatMessage(role="user", content=source)], **fields + ) + ) + template = ChatPromptTemplate.from_messages( + client.get_langchain_prompt(**values) + ) + rendered = template.format_messages(question="Hello")[0].content + assert rendered == "Context: Keep {name} / Ada / Hello" + + +@pytest.mark.parametrize("prompt_type", ["text", "chat"]) +def test_precompiled_values_can_complete_json_example(prompt_type): + fields = dict(name="json_context", version=1, config={}, tags=[], labels=[]) + source = "{ {{context}} }" + if prompt_type == "text": + client = TextPromptClient(Prompt_Text(type="text", prompt=source, **fields)) + template = PromptTemplate.from_template( + client.get_langchain_prompt(context='"name": "Ada"') + ) + rendered = template.format() + else: + client = ChatPromptClient( + Prompt_Chat( + type="chat", prompt=[ChatMessage(role="user", content=source)], **fields + ) + ) + template = ChatPromptTemplate.from_messages( + client.get_langchain_prompt(context='"name": "Ada"') + ) + rendered = template.format_messages()[0].content + assert rendered == '{ "name": "Ada" }' + + +@pytest.mark.parametrize("prompt_type", ["text", "chat"]) +def test_unfilled_variable_next_to_json_close_keeps_working(prompt_type): + fields = dict(name="json_variable", version=1, config={}, tags=[], labels=[]) + source = '{"answer": {{question}}}' + if prompt_type == "text": + client = TextPromptClient(Prompt_Text(type="text", prompt=source, **fields)) + template = PromptTemplate.from_template(client.get_langchain_prompt()) + rendered = template.format(question=42) + else: + client = ChatPromptClient( + Prompt_Chat( + type="chat", prompt=[ChatMessage(role="user", content=source)], **fields + ) + ) + template = ChatPromptTemplate.from_messages(client.get_langchain_prompt()) + rendered = template.format_messages(question=42)[0].content + assert rendered == '{"answer": 42}' + + +@pytest.mark.parametrize("prompt_type", ["text", "chat"]) +def test_unused_precompiled_values_are_not_stringified(prompt_type): + class Unused: + def __str__(self): + raise AssertionError("an unused value must not be evaluated") + + fields = dict(name="unused_context", version=1, config={}, tags=[], labels=[]) + source = "Question: {{question}}" + if prompt_type == "text": + client = TextPromptClient(Prompt_Text(type="text", prompt=source, **fields)) + template = PromptTemplate.from_template( + client.get_langchain_prompt(unused=Unused()) + ) + rendered = template.format(question="Hello") + else: + client = ChatPromptClient( + Prompt_Chat( + type="chat", prompt=[ChatMessage(role="user", content=source)], **fields + ) + ) + template = ChatPromptTemplate.from_messages( + client.get_langchain_prompt(unused=Unused()) + ) + rendered = template.format_messages(question="Hello")[0].content + assert rendered == "Question: Hello"