From 7335b8388ca7dc6ed532cf9f229d3f72583776aa Mon Sep 17 00:00:00 2001 From: Wasim Date: Sat, 17 Oct 2026 17:33:17 +0530 Subject: [PATCH] Python: fix: decode entities once in ChatHistory.from_rendered_prompt Co-authored-by: VANDRANKI --- .../semantic_kernel/contents/chat_history.py | 16 +++-- .../contents/chat_message_content.py | 2 +- .../contents/reasoning_content.py | 3 +- .../semantic_kernel/contents/text_content.py | 3 +- .../tests/unit/contents/test_chat_history.py | 63 +++++++++++++++++++ 5 files changed, 78 insertions(+), 9 deletions(-) diff --git a/python/semantic_kernel/contents/chat_history.py b/python/semantic_kernel/contents/chat_history.py index a773c232c7bc..a314c7173dac 100644 --- a/python/semantic_kernel/contents/chat_history.py +++ b/python/semantic_kernel/contents/chat_history.py @@ -1,11 +1,13 @@ # Copyright (c) Microsoft. All rights reserved. import logging +import re from collections.abc import Generator, Iterable from functools import singledispatchmethod from html import unescape from typing import Any, TypeVar from xml.etree.ElementTree import Element, tostring # nosec +from xml.sax.saxutils import escape # nosec from defusedxml.ElementTree import XML, ParseError from pydantic import Field, field_validator, model_validator @@ -22,6 +24,8 @@ _T = TypeVar("_T", bound="ChatHistory") +_CDATA_SECTION = re.compile(r"", re.DOTALL) + class ChatHistory(KernelBaseModel): """This class holds the history of chat messages from a chat conversation. @@ -340,13 +344,17 @@ def from_rendered_prompt(cls: type[_T], rendered_prompt: str) -> _T: prompt_tag = "root" messages: list["ChatMessageContent"] = [] prompt = rendered_prompt.strip() + # The XML parser decodes entities in normal text but keeps a CDATA section literal, and the + # template engine escapes values wherever they land. Decode CDATA once here, so every text + # is decoded exactly once by the parser and not again afterwards. + xml_text = _CDATA_SECTION.sub(lambda match: escape(unescape(match.group(1))), prompt) try: - xml_prompt = XML(text=f"<{prompt_tag}>{prompt}") + xml_prompt = XML(text=f"<{prompt_tag}>{xml_text}") except ParseError as exc: logger.info(f"Could not parse prompt {prompt} as xml, treating as text, error was: {exc}") return cls(messages=[ChatMessageContent(role=AuthorRole.USER, content=unescape(prompt))]) if xml_prompt.text and xml_prompt.text.strip(): - messages.append(ChatMessageContent(role=AuthorRole.SYSTEM, content=unescape(xml_prompt.text.strip()))) + messages.append(ChatMessageContent(role=AuthorRole.SYSTEM, content=xml_prompt.text.strip())) for item in xml_prompt: if item.tag == CHAT_MESSAGE_CONTENT_TAG: messages.append(ChatMessageContent.from_element(item)) @@ -368,10 +376,10 @@ def from_rendered_prompt(cls: type[_T], rendered_prompt: str) -> _T: # For unknown tags the tail is part of the surrounding text, # so keep it in the same message instead of starting a new one. if item.tail: - messages[-1].content = (messages[-1].content or "") + unescape(item.tail) + messages[-1].content = (messages[-1].content or "") + item.tail continue if item.tail and item.tail.strip(): - messages.append(ChatMessageContent(role=AuthorRole.USER, content=unescape(item.tail.strip()))) + messages.append(ChatMessageContent(role=AuthorRole.USER, content=item.tail.strip())) if len(messages) == 1 and messages[0].role == AuthorRole.SYSTEM: messages[0].role = AuthorRole.USER return cls(messages=messages) diff --git a/python/semantic_kernel/contents/chat_message_content.py b/python/semantic_kernel/contents/chat_message_content.py index 13376106e8b1..cc467d7f45b6 100644 --- a/python/semantic_kernel/contents/chat_message_content.py +++ b/python/semantic_kernel/contents/chat_message_content.py @@ -268,7 +268,7 @@ def from_element(cls, element: Element) -> "ChatMessageContent": kwargs: dict[str, Any] = {key: value for key, value in element.items()} items: list[KernelContent] = [] if element.text: - items.append(TextContent(text=unescape(element.text))) + items.append(TextContent(text=element.text)) for child in element: if child.tag not in TAG_CONTENT_MAP: logger.warning('Unknown tag "%s" in ChatMessageContent, treating as text', child.tag) diff --git a/python/semantic_kernel/contents/reasoning_content.py b/python/semantic_kernel/contents/reasoning_content.py index 8d88be29ca0b..74b235db7932 100644 --- a/python/semantic_kernel/contents/reasoning_content.py +++ b/python/semantic_kernel/contents/reasoning_content.py @@ -2,7 +2,6 @@ """Reasoning content model.""" -from html import unescape from typing import ClassVar, Literal, TypeVar from xml.etree.ElementTree import Element # nosec @@ -52,7 +51,7 @@ def from_element(cls: type[_T], element: Element) -> _T: """Create an instance from an XML Element.""" if element.tag != cls.tag: raise ContentInitializationError(f"Element tag is not {cls.tag}") # pragma: no cover - return cls(text=unescape(element.text) if element.text else "") + return cls(text=element.text or "") def to_dict(self) -> dict[str, str]: """Convert the instance to a dictionary suitable for message serialization.""" diff --git a/python/semantic_kernel/contents/text_content.py b/python/semantic_kernel/contents/text_content.py index c3decbfa1a6d..a219278fd7c7 100644 --- a/python/semantic_kernel/contents/text_content.py +++ b/python/semantic_kernel/contents/text_content.py @@ -1,6 +1,5 @@ # Copyright (c) Microsoft. All rights reserved. -from html import unescape from typing import ClassVar, Literal, TypeVar from xml.etree.ElementTree import Element # nosec @@ -52,7 +51,7 @@ def from_element(cls: type[_T], element: Element) -> _T: if element.tag != cls.tag: raise ContentInitializationError(f"Element tag is not {cls.tag}") # pragma: no cover - return cls(text=unescape(element.text) if element.text else "", encoding=element.get("encoding", None)) + return cls(text=element.text or "", encoding=element.get("encoding", None)) def to_dict(self) -> dict[str, str]: """Convert the instance to a dictionary.""" diff --git a/python/tests/unit/contents/test_chat_history.py b/python/tests/unit/contents/test_chat_history.py index 3c1b92945296..b6b979c00ba9 100644 --- a/python/tests/unit/contents/test_chat_history.py +++ b/python/tests/unit/contents/test_chat_history.py @@ -670,3 +670,66 @@ def __str__(self) -> str: ]) chat_history.add_tool_message([FunctionResultContent(id="test1", result=custom_result)]) assert "CustomResultTestValue" in chat_history.serialize() + + +ENTITY_TEXTS = [ + "What does & mean in HTML?", + "Write <b> as text", + "Use : as a colon", + "Tom & Jerry <3", +] + + +@pytest.mark.parametrize("text", ENTITY_TEXTS) +async def test_template_message_input_is_decoded_once(text: str): + template = '{{$input}}' + rendered = await KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template) + ).render(kernel=Kernel(), arguments=KernelArguments(input=text)) + + chat_history = ChatHistory.from_rendered_prompt(rendered) + assert chat_history.messages[0].role == AuthorRole.USER + assert chat_history.messages[0].content == text + + +@pytest.mark.parametrize("text", ENTITY_TEXTS) +async def test_template_plain_input_is_decoded_once(text: str): + template = "{{$input}}" + rendered = await KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template) + ).render(kernel=Kernel(), arguments=KernelArguments(input=text)) + + chat_history = ChatHistory.from_rendered_prompt(rendered) + assert chat_history.messages[0].content == text + + +@pytest.mark.parametrize("text", ENTITY_TEXTS) +async def test_template_text_after_message_is_decoded_once(text: str): + template = 'rules{{$input}}' + rendered = await KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template) + ).render(kernel=Kernel(), arguments=KernelArguments(input=text)) + + chat_history = ChatHistory.from_rendered_prompt(rendered) + assert chat_history.messages[1].content == text + + +@pytest.mark.parametrize("text", ENTITY_TEXTS) +async def test_template_cdata_input_is_decoded_once(text: str): + template = '' + rendered = await KernelPromptTemplate( + prompt_template_config=PromptTemplateConfig(name="test", description="test", template=template) + ).render(kernel=Kernel(), arguments=KernelArguments(input=text)) + + chat_history = ChatHistory.from_rendered_prompt(rendered) + assert chat_history.messages[0].content == text + + +@pytest.mark.parametrize("text", ENTITY_TEXTS) +def test_chat_history_prompt_round_trip_keeps_entities(text: str): + chat_history = ChatHistory() + chat_history.add_user_message(text) + chat_history.add_assistant_message(text) + + restored = ChatHistory.from_rendered_prompt(chat_history.to_prompt()) + assert [message.content for message in restored.messages] == [text, text]