Refine Data Access using OpeaStore (#1916)

* Porting the data store layers of components to OpeaStore.
* Decoupling user-defined data structure from data access layer of MongoDB.
* Adding OpeaStore tools
* Add search by KV API
* Update READMEs

---------

Signed-off-by: Yi Yao <yi.a.yao@intel.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: joshuayao

comps/chathistory/README.md

@@ -6,78 +6,18 @@ It can be integrated into application by making HTTP requests to the provided AP
 
 ![Flow Chart](./assets/img/chathistory_flow.png)
 
----
-
 ## 🛠️ Features
 
 - **Store Chat Conversations**: Save chat messages user information, and metadata associated with each conversation.
 - **Retrieve Chat Histories**: Fetch chat histories for a specific user or retrieve a particular conversation by its unique identifier.
 - **Update Chat Conversations**: Modify existing chat conversations by adding new messages or updating existing ones.
 - **Delete Chat Conversations**: Remove chat conversations record from database.
 
----
-
-## 🤖 MCP (Model Context Protocol) Support
-
-The Chat History microservice supports MCP integration, allowing AI agents to discover and utilize chat history management capabilities as tools.
-
-### MCP Configuration
-
-#### Environment Variables
-
-- `ENABLE_MCP`: Set to `true`, `1`, or `yes` to enable MCP support (default: `false`)
-
-#### Docker Compose
-
-```yaml
-services:
-  chathistory-mongo:
-    environment:
-      ENABLE_MCP: true
-```
-
-#### Kubernetes
-
-```yaml
-chathistory:
-  ENABLE_MCP: true
-```
-
-### MCP Tools Available
-
-When MCP is enabled, the following tools are available for AI agents:
-
-1. **create_documents** - Create or update chat conversation history
-2. **get_documents** - Retrieve chat conversation history
-3. **delete_documents** - Delete chat conversation history
-
-### Usage with AI Agents
-
-```python
-from comps.cores.mcp import OpeaMCPToolsManager
-
-# Initialize MCP tools manager
-tools_manager = OpeaMCPToolsManager()
-
-# Add chathistory service
-tools_manager.add_service("http://chathistory-service:6012")
-
-# AI agents can now discover and use chathistory tools
-tools = await tools_manager.get_available_tools()
-```
-
-### MCP Endpoint
-
-When MCP is enabled, the service exposes an additional SSE endpoint:
-
-- `/sse` - Server-Sent Events endpoint for MCP communication
-
----
-
-## ⚙️ Implementation
-
-The Chat History microservice able to support various database backends for storing the chat conversations.
+## ⚙️ Deployment Options
 
-### Chat History with MongoDB
+To get detailed, step-by-step instructions on deploying the `chathistory` microservice, you should consult the deployment guide. This guide will walk you through all the necessary steps, from building the Docker images to configuring your environment and running the service.
 
-For more detail, please refer to this [README](src/README.md)
+| Platform | Deployment Method | Database | Link                                                      |
+| -------- | ----------------- | -------- | --------------------------------------------------------- |
+| CPU      | Docker            | MongoDB  | [Deployment Guide](./deployment/docker_compose/README.md) |
+| CPU      | Docker Compose    | MongoDB  | [Deployment Guide](./deployment/docker_compose/README.md) |

comps/chathistory/src/README.md …tory/deployment/docker_compose/README.mdcomps/chathistory/src/README.md renamed to comps/chathistory/deployment/docker_compose/README.md comps/chathistory/src/README.md renamed to comps/chathistory/deployment/docker_compose/README.md

@@ -9,6 +9,7 @@ This README provides setup guides and all the necessary information about the Ch
 ```bash
 export http_proxy=${your_http_proxy}
 export https_proxy=${your_http_proxy}
+export OPEA_STORE_NAME="mongodb"
 export MONGO_HOST=${MONGO_HOST}
 export MONGO_PORT=27017
 export DB_NAME=${DB_NAME}

comps/chathistory/deployment/docker_compose/compose.yaml

@@ -24,6 +24,7 @@ services:
       http_proxy: ${http_proxy}
       no_proxy: ${no_proxy}
       https_proxy: ${https_proxy}
+      OPEA_STORE_NAME: ${OPEA_STORE_NAME:-mongodb}
       MONGO_HOST: ${MONGO_HOST}
       MONGO_PORT: ${MONGO_PORT}
       COLLECTION_NAME: ${COLLECTION_NAME}

comps/chathistory/src/document_store.py

@@ -0,0 +1,136 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Optional
+
+from fastapi import HTTPException
+from pydantic import BaseModel
+
+from comps.cores.proto.api_protocol import ChatCompletionRequest
+from comps.cores.storages.models import ChatId, ChatMessage
+from comps.cores.storages.stores import column_to_id, get_store, id_to_column
+
+
+class ChatMessageDto(BaseModel):
+    data: ChatCompletionRequest
+    first_query: Optional[str] = None
+    doc_id: Optional[str] = None
+    user: Optional[str] = None
+
+
+def _prepersist(document: ChatMessage) -> dict:
+    """Converts a ChatMessage object to a dictionary suitable for persistence.
+
+    Args:
+        document (ChatMessage): The ChatMessage object to be converted.
+
+    Returns:
+        dict: A dictionary representation of the ChatMessage, ready for persistence.
+    """
+    data_dict = document.model_dump(by_alias=True, mode="json")
+    data_dict = column_to_id("id", data_dict)
+    return data_dict
+
+
+def _post_getby_id(rs: dict) -> dict:
+    """Post-processes a document retrieved by ID from the store.
+
+    Args:
+        rs (dict): The raw document dictionary from the store.
+
+    Returns:
+        dict: The processed document data, or None if the document doesn't exist.
+    """
+    rs = id_to_column("id", rs)
+    return rs.get("data") if rs else None
+
+
+def _post_getby_user(rss: list) -> list:
+    """Post-processes a list of documents retrieved by user from the store.
+
+    Args:
+        rss (list): A list of raw document dictionaries from the store.
+
+    Returns:
+        list: A list of processed documents with the 'data' field removed.
+    """
+    for rs in rss:
+        rs = id_to_column("id", rs)
+        rs.pop("data")
+    return rss
+
+
+def _check_user_info(document: ChatMessage | ChatId):
+    """Checks if the user information is provided in the document.
+
+    Args:
+        document (ChatMessage|ChatId): The document to be checked.
+    Raises:
+        HTTPException: If the user information is missing.
+    """
+    user = document.data.user if isinstance(document, ChatMessage) else document.user
+    if user is None:
+        raise HTTPException(status_code=400, detail="Please provide the user information")
+
+
+async def save_or_update(document: ChatMessage):
+    """Saves a new chat message or updates an existing one in the data store.
+
+    Args:
+        document (ChatMessage): The ChatMessage object to be saved or updated.
+                               If the document has an ID, it will be updated;
+                               otherwise, a new document will be created.
+
+    Returns:
+        The result of the save or update operation from the store.
+    """
+    _check_user_info(document)
+    store = get_store(document.data.user)
+    if document.id:
+        return await store.aupdate_document(_prepersist(document))
+    else:
+        return await store.asave_document(_prepersist(document))
+
+
+async def get(document: ChatId):
+    """Retrieves chat messages from the data store.
+
+    Args:
+        document (ChatId): The ChatId object containing user information and
+                           optionally a document ID. If document.id is None,
+                           retrieves all documents for the user; otherwise,
+                           retrieves the specific document by ID.
+
+    Returns:
+        Either a list of all documents for the user (if document.id is None) or
+        a specific document (if document.id is provided).
+    """
+    _check_user_info(document)
+    store = get_store(document.user)
+    if document.id is None:
+        rss = await store.asearch(key="data.user", value=document.user)
+        return _post_getby_user(rss)
+    else:
+        rs = await store.aget_document_by_id(document.id)
+        return _post_getby_id(rs)
+
+
+async def delete(document: ChatId):
+    """Deletes a specific chat message from the data store.
+
+    Args:
+        document (ChatId): The ChatId object containing user information and document ID.
+                          The document ID must be provided for deletion.
+
+    Returns:
+        The result of the delete operation from the store.
+
+    Raises:
+        Exception: If the document ID is not provided.
+    """
+    _check_user_info(document)
+    store = get_store(document.user)
+    if document.id is None:
+        raise Exception("Document id is required.")
+    else:
+        return await store.adelete_document(document.id)