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}{prompt_tag}>")
+ xml_prompt = XML(text=f"<{prompt_tag}>{xml_text}{prompt_tag}>")
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]