How to serialize and deserialize ChatHistoryAgentThread Object into database (MongoDB) in python

[JasonL233]

Hi community,
The following is the current method I used to serialize and deserialize thread to MongoDB. However, I am writing a custom serializer and deserializer, it works but I don't think this is the proper way to do it. I am wondering if there is a better and connivence way to perform serialization and deserialization into NoSql database like MongoDB.

Also, I add message into the thread using the ChatMessageContent Class, is this the correct way to do it?

I would appreciate if I can get any advice and help. Thank you!

Serializer

async def save_thread(session_id: str, thread: ChatHistoryAgentThread):
    thread_id = thread._id
    messages = []

    async for msg in thread.get_messages():
        # Convert metadata to a serializable format

        messages.append({
            "thread_id": thread_id,
            "timestamp": datetime.now(UTC),
            "role": msg.role.name,
            "content": msg.content,
            "items": [
                {
                    "type": item.content_type,
                    "text": getattr(item, "text", None)  # only for TextContent
                }
                for item in msg.items
            ]
        })

    threads_collection.update_one(
        {"session_id": session_id},
        {"$set": {
            "thread_id": thread_id,
            "last_updated": datetime.now(UTC)
        }},
        upsert=True
    )

    # Overwrite existing messages for this thread
    msg_collection.delete_many({"thread_id": thread_id})
    if messages:
        msg_collection.insert_many(messages)

Deserializer

async def load_thread(session_id: str) -> ChatHistoryAgentThread:
    thread_meta = threads_collection.find_one({"session_id": session_id})
    if not thread_meta:
        return ChatHistoryAgentThread()

    thread_id = thread_meta["thread_id"]
    thread = ChatHistoryAgentThread(thread_id=thread_id)

    cursor = msg_collection.find({"thread_id": thread_id}).sort("timestamp", ASCENDING)
    for doc in cursor:
        role = AuthorRole[doc["role"].upper()]
        content = doc.get("content", "")
        items = []

        for item in doc.get("items", []):
            if item["type"] == "text":
                items.append(TextContent(text=item["text"]))

        msg = ChatMessageContent(
            role=role,
            content=content,
            items=items,
        )

        await thread._on_new_message(msg)

    return thread

[moonbox3]

Hi @JasonL233, thanks for your question. We have a sample of serializing a chat history to CosmosDB in the following sample. It should be something you could attempt to adapt to your use-case, while we work on (in the next few weeks) providing a more concrete solution to doing this with the thread abstraction: https://github.com/microsoft/semantic-kernel/blob/main/python/samples/concepts/chat_history/store_chat_history_in_cosmosdb.py.

The sample above isn't strictly for agents, but you are able to create a thread with a chat history object - you could use a similar construct. Let me know your thoughts, please.

[moonbox3]

Hi @JasonL233, I don't have a super useful update yet - been trying to get through some other priorities. I'll work on getting you something this week. Thanks for your patience.

[JasonL233]

Hi @JasonL233, I don't have a super useful update yet - been trying to get through some other priorities. I'll work on getting you something this week. Thanks for your patience.

For sure! I will look forward to any updates from you. Thank you!

[JasonL233]

Hi @JasonL233, I don't have a super useful update yet - been trying to get through some other priorities. I'll work on getting you something this week. Thanks for your patience.

Hi Evan,

I encountered a new problem on my AI Agent.

I was using the above serialization & deserialization but then I got an error saying

IndexError: list index out of range File "semantic_kernel/contents/chat_message_content.py", line 313, in to_dict assert isinstance(self.items[0], FunctionResultContent) # nosec ~~~~~~~~~~^^^

It occurs inside the SK internals after this call:

response = await agent.get_response(messages=user_input, thread=thread)

Below is my chat endpoint:

    @app.post("/chat")
    async def chat(userMessage: Message):
    try:
          user_input = userMessage.message.strip()
          session_id = userMessage.session_id

          thread = await load_thread(session_id)
          print(f"Loaded thread with ID: {thread._id}")
          
          user_msg = ChatMessageContent(role=AuthorRole.USER, content=user_input)
          await thread._on_new_message(user_msg)
          print("Added user message to thread")
  
          augmented_prompt = await get_augmented_prompt(user_input)
          system_msg = ChatMessageContent(role=AuthorRole.SYSTEM, content=f"Here is relevant information: {augmented_prompt}")
          await thread._on_new_message(system_msg)
          print("Added system message to thread")
  
          response = await agent.get_response(messages=user_input, thread=thread)
          print("Got response from agent")
  
          await save_thread(session_id, thread)
          print("Saved thread to database")
          
          print("------------------------------")
          print("Thread ID:", thread._id)
          print("LOG: ")
          try:
              messages = []
              async for msg in thread.get_messages():
                  messages.append(msg)
              
              if not messages:
                  print("No messages found in thread")
              else:
                  for i, msg in enumerate(messages, 1):
                      print(f"{i}. [Role: {msg.role.name}] Content: {msg.content}")
          except Exception as e:
              print(f"Error while logging messages: {str(e)}")
              
          return {"reply": response.message.content}

    except Exception as e:
          print(f"Error in chat endpoint: {str(e)}")
          import traceback
          print("Full traceback:")
          print(traceback.format_exc())
          return {"error": str(e)}

Do you might taking a look of this problem!! I was really frustrated because I've been debugging it for hours but still cannot find a solution. Would be really appreciated if I have get any help from you or your colleagues!

Thank you very much!

[moonbox3]

Hi there. I can help take a look on Monday. One quick comment: you shouldn’t be explicitly calling thread._on_new_message(). Any context (messages) you want to provide to the thread should be handled by passing into the messages kwarg of the get response or invoke method.

[JasonL233]

Hi there. I can help take a look on Monday. One quick comment: you shouldn’t be explicitly calling thread._on_new_message(). Any context (messages) you want to provide to the thread should be handled by passing into the messages kwarg of the get response or invoke method.

Thank you so much! I'll give a try on replacing ._on_new_message().