Chat with the MOS. Args: query (str): The user's query. user_id (str, optional): The user ID for the chat session. Defaults to the user ID from the config. base_prompt (str, optional): A custom base prompt to use for the chat. It
(self, query: str, user_id: str | None = None, base_prompt: str | None = None)
| 249 | return documents |
| 250 | |
| 251 | def chat(self, query: str, user_id: str | None = None, base_prompt: str | None = None) -> str: |
| 252 | """ |
| 253 | Chat with the MOS. |
| 254 | |
| 255 | Args: |
| 256 | query (str): The user's query. |
| 257 | user_id (str, optional): The user ID for the chat session. Defaults to the user ID from the config. |
| 258 | base_prompt (str, optional): A custom base prompt to use for the chat. |
| 259 | It can be a template string with a `{memories}` placeholder. |
| 260 | If not provided, a default prompt is used. |
| 261 | |
| 262 | Returns: |
| 263 | str: The response from the MOS. |
| 264 | """ |
| 265 | target_user_id = user_id if user_id is not None else self.user_id |
| 266 | accessible_cubes = self.user_manager.get_user_cubes(target_user_id) |
| 267 | user_cube_ids = [cube.cube_id for cube in accessible_cubes] |
| 268 | if target_user_id not in self.chat_history_manager: |
| 269 | self._register_chat_history(target_user_id) |
| 270 | |
| 271 | chat_history = self.chat_history_manager[target_user_id] |
| 272 | |
| 273 | if self.config.enable_textual_memory and self.mem_cubes: |
| 274 | memories_all = [] |
| 275 | for mem_cube_id, mem_cube in self.mem_cubes.items(): |
| 276 | if mem_cube_id not in user_cube_ids: |
| 277 | continue |
| 278 | if not mem_cube.text_mem: |
| 279 | continue |
| 280 | |
| 281 | # submit message to scheduler |
| 282 | if self.enable_mem_scheduler and self.mem_scheduler is not None: |
| 283 | message_item = ScheduleMessageItem( |
| 284 | user_id=target_user_id, |
| 285 | mem_cube_id=mem_cube_id, |
| 286 | label=QUERY_TASK_LABEL, |
| 287 | content=query, |
| 288 | timestamp=datetime.utcnow(), |
| 289 | ) |
| 290 | self.mem_scheduler.submit_messages(messages=[message_item]) |
| 291 | |
| 292 | memories = mem_cube.text_mem.search( |
| 293 | query, |
| 294 | top_k=self.config.top_k, |
| 295 | info={ |
| 296 | "user_id": target_user_id, |
| 297 | "session_id": self.session_id, |
| 298 | "chat_history": chat_history.chat_history, |
| 299 | }, |
| 300 | ) |
| 301 | memories_all.extend(memories) |
| 302 | logger.info(f"🧠 [Memory] Searched memories:\n{self._str_memories(memories_all)}\n") |
| 303 | system_prompt = self._build_system_prompt(memories_all, base_prompt=base_prompt) |
| 304 | else: |
| 305 | system_prompt = self._build_system_prompt(base_prompt=base_prompt) |
| 306 | current_messages = [ |
| 307 | {"role": "system", "content": system_prompt}, |
| 308 | *chat_history.chat_history, |