Search for textual memories across all registered MemCubes. Args: query (str): The search query. user_id (str, optional): The identifier of the user to search for. If None, the default user is used. install_cube_ids (list[str], op
(
self,
query: str,
user_id: str | None = None,
install_cube_ids: list[str] | None = None,
top_k: int | None = None,
mode: Literal["fast", "fine"] = "fast",
internet_search: bool = False,
moscube: bool = False,
session_id: str | None = None,
**kwargs,
)
| 544 | raise ValueError(f"MemCube with ID {mem_cube_id} does not exist.") |
| 545 | |
| 546 | def search( |
| 547 | self, |
| 548 | query: str, |
| 549 | user_id: str | None = None, |
| 550 | install_cube_ids: list[str] | None = None, |
| 551 | top_k: int | None = None, |
| 552 | mode: Literal["fast", "fine"] = "fast", |
| 553 | internet_search: bool = False, |
| 554 | moscube: bool = False, |
| 555 | session_id: str | None = None, |
| 556 | **kwargs, |
| 557 | ) -> MOSSearchResult: |
| 558 | """ |
| 559 | Search for textual memories across all registered MemCubes. |
| 560 | |
| 561 | Args: |
| 562 | query (str): The search query. |
| 563 | user_id (str, optional): The identifier of the user to search for. |
| 564 | If None, the default user is used. |
| 565 | install_cube_ids (list[str], optional): The list of MemCube IDs to install. |
| 566 | If None, all MemCube for the user is used. |
| 567 | |
| 568 | Returns: |
| 569 | MemoryResult: A dictionary containing the search results. |
| 570 | """ |
| 571 | target_session_id = session_id if session_id is not None else self.session_id |
| 572 | target_user_id = user_id if user_id is not None else self.user_id |
| 573 | |
| 574 | self._validate_user_exists(target_user_id) |
| 575 | # Get all cubes accessible by the target user |
| 576 | accessible_cubes = self.user_manager.get_user_cubes(target_user_id) |
| 577 | user_cube_ids = [cube.cube_id for cube in accessible_cubes] |
| 578 | |
| 579 | logger.info( |
| 580 | f"User {target_user_id} has access to {len(user_cube_ids)} cubes: {user_cube_ids}" |
| 581 | ) |
| 582 | if target_user_id not in self.chat_history_manager: |
| 583 | self._register_chat_history(target_user_id) |
| 584 | chat_history = self.chat_history_manager[target_user_id] |
| 585 | |
| 586 | # Create search filter if session_id is provided |
| 587 | search_filter = None |
| 588 | if session_id is not None: |
| 589 | search_filter = {"session_id": session_id} |
| 590 | |
| 591 | result: MOSSearchResult = { |
| 592 | "text_mem": [], |
| 593 | "act_mem": [], |
| 594 | "para_mem": [], |
| 595 | "pref_mem": [], |
| 596 | } |
| 597 | if install_cube_ids is None: |
| 598 | install_cube_ids = user_cube_ids |
| 599 | # create exist dict in mem_cubes and avoid one search slow |
| 600 | tmp_mem_cubes = {} |
| 601 | time_start_cube_get = time.time() |
| 602 | for mem_cube_id in install_cube_ids: |
| 603 | if mem_cube_id in self.mem_cubes: |