Add chat sessions related command (#13268)

### What problem does this PR solve?

1. Create / Drop / List chat sessions
2. Chat with LLM and datasets

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

---------

Signed-off-by: Jin Hai <haijin.chn@gmail.com>

admin/client/parser.py

@@ -88,6 +88,10 @@ sql_command: login_user
            | parse_dataset_async
            | import_docs_into_dataset
            | search_on_datasets
+           | create_chat_session
+           | drop_chat_session
+           | list_chat_sessions
+           | chat_on_session
            | benchmark
 
 // meta command definition
@@ -170,6 +174,8 @@ ASYNC: "ASYNC"i
 SYNC: "SYNC"i
 BENCHMARK: "BENCHMARK"i
 PING: "PING"i
+SESSION: "SESSION"i
+SESSIONS: "SESSIONS"i
 
 login_user: LOGIN USER quoted_string ";"
 list_services: LIST SERVICES ";"
@@ -246,6 +252,10 @@ user_statement: ping_server
                 | list_user_default_models
                 | import_docs_into_dataset
                 | search_on_datasets
+                | create_chat_session
+                | drop_chat_session
+                | list_chat_sessions
+                | chat_on_session
 
 ping_server: PING ";"
 show_current_user: SHOW CURRENT USER ";"
@@ -274,6 +284,10 @@ list_user_agents: LIST AGENTS ";"
 list_user_chats: LIST CHATS ";"
 create_user_chat: CREATE CHAT quoted_string ";"
 drop_user_chat: DROP CHAT quoted_string ";"
+create_chat_session: CREATE CHAT quoted_string SESSION quoted_string ";"
+drop_chat_session: DROP CHAT quoted_string SESSION quoted_string ";"
+list_chat_sessions: LIST CHAT quoted_string SESSIONS ";"
+chat_on_session: CHAT quoted_string ON quoted_string SESSION quoted_string ";"
 list_user_model_providers: LIST MODEL PROVIDERS ";"
 list_user_default_models: LIST DEFAULT MODELS ";"
 import_docs_into_dataset: IMPORT quoted_string INTO DATASET quoted_string ";"
@@ -575,6 +589,26 @@ class RAGFlowCLITransformer(Transformer):
         dataset_name = items[2].children[0].strip("'\"")
         return {"type": "parse_dataset", "dataset_name": dataset_name, "method": "async"}
 
+    def create_chat_session(self, items):
+        chat_name = items[2].children[0].strip("'\"")
+        session_name = items[4].children[0].strip("'\"")
+        return {"type": "create_chat_session", "chat_name": chat_name, "session_name": session_name}
+
+    def drop_chat_session(self, items):
+        chat_name = items[2].children[0].strip("'\"")
+        session_name = items[4].children[0].strip("'\"")
+        return {"type": "drop_chat_session", "chat_name": chat_name, "session_name": session_name}
+
+    def list_chat_sessions(self, items):
+        chat_name = items[2].children[0].strip("'\"")
+        return {"type": "list_chat_sessions", "chat_name": chat_name}
+
+    def chat_on_session(self, items):
+        message = items[1].children[0].strip("'\"")
+        chat_name = items[3].children[0].strip("'\"")
+        session_id = items[5].children[0].strip("'\"")
+        return {"type": "chat_on_session", "message": message, "chat_name": chat_name, "session_id": session_id}
+
     def import_docs_into_dataset(self, items):
         document_list_str = items[1].children[0].strip("'\"")
         document_paths = document_list_str.split(",")

admin/client/ragflow_client.py

@@ -13,7 +13,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 #
-
+import json
 import time
 from typing import Any, List, Optional
 import multiprocessing as mp
@@ -881,6 +881,142 @@ class RAGFlowClient:
         else:
             print(f"Fail to drop chat {chat_name}, code: {res_json['code']}, message: {res_json['message']}")
 
+    def _get_chat_id_by_name(self, chat_name):
+        """Get chat (dialog) ID by name."""
+        res_json = self._list_chats({})
+        if res_json is None:
+            return None
+        for elem in res_json:
+            if elem["name"] == chat_name:
+                return elem["id"]
+        print(f"Chat '{chat_name}' not found")
+        return None
+
+    def _list_chat_sessions(self, dialog_id):
+        """List all sessions (conversations) for a given dialog."""
+        response = self.http_client.request("GET", f"/conversation/list?dialog_id={dialog_id}", use_api_base=False,
+                                            auth_kind="web")
+        res_json = response.json()
+        if response.status_code == 200 and res_json["code"] == 0:
+            return res_json["data"]
+        else:
+            print(f"Fail to list chat sessions, code: {res_json['code']}, message: {res_json['message']}")
+            return None
+
+    def create_chat_session(self, command):
+        if self.server_type != "user":
+            print("This command is only allowed in USER mode")
+        chat_name = command["chat_name"]
+        session_name = command["session_name"]
+        dialog_id = self._get_chat_id_by_name(chat_name)
+        if dialog_id is None:
+            return
+        payload = {
+            "conversation_id": "",
+            "is_new": True,
+            "name": session_name,
+            "dialog_id": dialog_id
+        }
+        response = self.http_client.request("POST", "/conversation/set", json_body=payload, use_api_base=False,
+                                            auth_kind="web")
+        res_json = response.json()
+        if response.status_code == 200 and res_json["code"] == 0:
+            print(f"Success to create chat session '{session_name}' for chat: {chat_name}")
+        else:
+            print(f"Fail to create chat session '{session_name}' for chat {chat_name}, code: {res_json['code']}, message: {res_json['message']}")
+
+    def drop_chat_session(self, command):
+        if self.server_type != "user":
+            print("This command is only allowed in USER mode")
+        chat_name = command["chat_name"]
+        session_name = command["session_name"]
+        dialog_id = self._get_chat_id_by_name(chat_name)
+        if dialog_id is None:
+            return
+        sessions = self._list_chat_sessions(dialog_id)
+        if sessions is None:
+            return
+        to_drop_session_ids = []
+        for session in sessions:
+            if session["name"] == session_name:
+                to_drop_session_ids.append(session["id"])
+        if not to_drop_session_ids:
+            print(f"Chat session '{session_name}' not found in chat '{chat_name}'")
+            return
+        payload = {"conversation_ids": to_drop_session_ids}
+        response = self.http_client.request("POST", "/conversation/rm", json_body=payload, use_api_base=False,
+                                            auth_kind="web")
+        res_json = response.json()
+        if response.status_code == 200 and res_json["code"] == 0:
+            print(f"Success to drop chat session '{session_name}' from chat: {chat_name}")
+        else:
+            print(f"Fail to drop chat session '{session_name}' from chat {chat_name}, code: {res_json['code']}, message: {res_json['message']}")
+
+    def list_chat_sessions(self, command):
+        if self.server_type != "user":
+            print("This command is only allowed in USER mode")
+        chat_name = command["chat_name"]
+        dialog_id = self._get_chat_id_by_name(chat_name)
+        if dialog_id is None:
+            return
+        sessions = self._list_chat_sessions(dialog_id)
+        if sessions is None:
+            return
+        # Add chat_name to each session for display
+        for session in sessions:
+            session["chat_name"] = chat_name
+        if "iterations" in command:
+            # for benchmark
+            return sessions
+        self._print_table_simple(sessions)
+
+    def chat_on_session(self, command):
+        if self.server_type != "user":
+            print("This command is only allowed in USER mode")
+        message = command["message"]
+        session_id = command["session_id"]
+
+        # Prepare payload for completion API
+        # Note: stream parameter is not sent, server defaults to stream=True
+        payload = {
+            "conversation_id": session_id,
+            "messages": [{"role": "user", "content": message}]
+        }
+
+        response = self.http_client.request("POST", "/conversation/completion", json_body=payload,
+                                            use_api_base=False, auth_kind="web", stream=True)
+
+        if response.status_code != 200:
+            print(f"Fail to chat on session, status code: {response.status_code}")
+            return
+
+        print("Assistant: ", end="", flush=True)
+        full_answer = ""
+        for line in response.iter_lines():
+            if not line:
+                continue
+            line_str = line.decode('utf-8')
+            if not line_str.startswith('data:'):
+                continue
+            data_str = line_str[5:].strip()
+            if data_str == '[DONE]':
+                break
+            try:
+                data_json = json.loads(data_str)
+                if data_json.get("code") != 0:
+                    print(f"\nFail to chat on session, code: {data_json.get('code')}, message: {data_json.get('message', '')}")
+                    return
+                # Check if it's the final message
+                if data_json.get("data") is True:
+                    break
+                answer = data_json.get("data", {}).get("answer", "")
+                if answer:
+                    print(answer, end="", flush=True)
+                    full_answer += answer
+            except json.JSONDecodeError:
+                continue
+        print()  # Final newline
+
     def list_user_model_providers(self, command):
         if self.server_type != "user":
             print("This command is only allowed in USER mode")
@@ -1368,6 +1504,14 @@ def run_command(client: RAGFlowClient, command_dict: dict):
             client.create_user_chat(command_dict)
         case "drop_user_chat":
             client.drop_user_chat(command_dict)
+        case "create_chat_session":
+            client.create_chat_session(command_dict)
+        case "drop_chat_session":
+            client.drop_chat_session(command_dict)
+        case "list_chat_sessions":
+            return client.list_chat_sessions(command_dict)
+        case "chat_on_session":
+            client.chat_on_session(command_dict)
         case "list_user_model_providers":
             client.list_user_model_providers(command_dict)
         case "list_user_default_models":