Python: small fix for CH serialization (#5738)

### Motivation and Context

<!-- Thank you for your contribution to the semantic-kernel repo!
Please help reviewers and future users, providing the following
information:
  1. Why is this change required?
  2. What problem does it solve?
  3. What scenario does it contribute to?
  4. If it fixes an open issue, please link to the issue here.
-->
Noticed a small issue when ChatHistory is created with different types
of messages, and then serialized.
This fixed that and adds a test for that case.

### Description

<!-- Describe your changes, the overall approach, the underlying design.
These notes will help understanding how your code works. Thanks! -->

### Contribution Checklist

<!-- Before submitting this PR, please make sure: -->

- [x] The code builds clean without any errors or warnings
- [x] The PR follows the [SK Contribution
Guidelines](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md)
and the [pre-submission formatting
script](https://github.com/microsoft/semantic-kernel/blob/main/CONTRIBUTING.md#development-scripts)
raises no violations
- [x] All unit tests pass, and I have added new tests where possible
- [x] I didn't break anyone 😄
This commit is contained in:
Eduard van Valkenburg
2024-04-02 20:20:22 +00:00
committed by GitHub
parent c65644a175
commit 290f44d593
3 changed files with 33 additions and 3 deletions
@@ -41,7 +41,10 @@ class OpenAIChatMessageContent(ChatMessageContent):
if isinstance(tool_calls, list):
for index, call in enumerate(tool_calls):
if not isinstance(call, ToolCall):
tool_calls[index] = ToolCall.model_validate_json(call)
if isinstance(call, dict):
tool_calls[index] = ToolCall.model_validate(call)
else:
tool_calls[index] = ToolCall.model_validate_json(call)
return tool_calls
if isinstance(tool_calls, str):
return [ToolCall.model_validate_json(call) for call in tool_calls.split("|")]
@@ -53,6 +56,8 @@ class OpenAIChatMessageContent(ChatMessageContent):
return None
if isinstance(function_call, FunctionCall):
return function_call
if isinstance(function_call, dict):
return FunctionCall.model_validate(function_call)
return FunctionCall.model_validate_json(function_call)
@staticmethod
@@ -6,6 +6,7 @@ from typing import Any, Iterator, List
from xml.etree.ElementTree import Element, tostring
from defusedxml.ElementTree import XML, ParseError
from pydantic import field_validator
from semantic_kernel.contents.chat_message_content import ChatMessageContent
from semantic_kernel.contents.chat_message_content_base import ChatMessageContentBase
@@ -34,7 +35,7 @@ class ChatHistory(KernelBaseModel):
messages (List[ChatMessageContent]): The list of chat messages in the history.
"""
messages: list["ChatMessageContent"]
messages: list[ChatMessageContent]
message_type: TYPES_CHAT_MESSAGE_CONTENT = CHAT_MESSAGE_CONTENT
def __init__(self, **data: Any):
@@ -75,6 +76,19 @@ class ChatHistory(KernelBaseModel):
data["messages"] = []
super().__init__(**data)
@field_validator("messages", mode="before")
@classmethod
def _validate_messages(cls, messages: List[ChatMessageContent]) -> List[ChatMessageContent]:
if not messages:
return messages
out_msgs: List[ChatMessageContent] = []
for message in messages:
if isinstance(message, dict):
out_msgs.append(ChatMessageContentBase.from_dict(message))
else:
out_msgs.append(message)
return out_msgs
def add_system_message(self, content: str, **kwargs: Any) -> None:
"""Add a system message to the chat history."""
self.add_message(message=self._prepare_for_add(ChatRole.SYSTEM, content, **kwargs))
@@ -49,7 +49,7 @@ def test_add_system_message(chat_history: ChatHistory):
assert chat_history.messages[-1].role == ChatRole.SYSTEM
def test_add_system_message_at_init(chat_history: ChatHistory):
def test_add_system_message_at_init():
content = "System message"
chat_history = ChatHistory(system_message=content)
assert chat_history.messages[-1].content == content
@@ -190,6 +190,17 @@ def test_serialize(): # ignore: E501
)
def test_serialize_and_deserialize_to_chat_history_mixed_content():
system_msg = "a test system prompt"
msgs = [ChatMessageContent(role=ChatRole.USER, content=f"Message {i}") for i in range(3)]
msgs.extend([OpenAIChatMessageContent(role=ChatRole.USER, content=f"Message {i}") for i in range(3)])
msgs.extend([AzureChatMessageContent(role=ChatRole.USER, content=f"Message {i}") for i in range(3)])
chat_history = ChatHistory(messages=msgs, system_message=system_msg)
json_str = chat_history.serialize()
new_chat_history = ChatHistory.restore_chat_history(json_str)
assert new_chat_history == chat_history
def test_serialize_and_deserialize_to_chat_history():
system_msg = "a test system prompt"
msgs = [ChatMessageContent(role=ChatRole.USER, content=f"Message {i}") for i in range(3)]