Method query

methods/memos_agent.py:353–454

Query using MemOS's official search() method. Uses MemOS's native vector search for retrieval, then generates response. Args: question: User question to answer. system_message: Optional system prompt. **kwargs: Additional arguments (unused

(
        self,
        question: str,
        system_message: Optional[str] = None,
        **kwargs,
    )

        )

    def query(
        self,
        question: str,
        system_message: Optional[str] = None,
        **kwargs,
    ) -> AgentResponse:
        """Query using MemOS's official search() method.

        Uses MemOS's native vector search for retrieval, then generates response.

        Args:
            question: User question to answer.
            system_message: Optional system prompt.
            **kwargs: Additional arguments (unused).

        Returns:
            AgentResponse with answer and retrieval details.
        """
        context_id = self._get_context_id()
        memory_system = self._get_memory_system(context_id)

        # Truncate question
        bounded_question = self._truncate_to_tokens(question, self.max_question_tokens)

        start_time = time.time()

        # Use MemOS's official search() method
        # Request exactly retrieve_num results (no over-fetching)
        memory_items = memory_system.search(bounded_question, top_k=self.retrieve_num)

        search_time = time.time() - start_time

        # Calculate token budget for memory context
        system_tokens = self._llm_client.count_tokens(system_message) if system_message else 0
        reserved_tokens = self.max_tokens + 500
        available_tokens = max(self.max_context_tokens - reserved_tokens - system_tokens, 0)
        question_tokens = self._llm_client.count_tokens(bounded_question)
        memory_budget = max(available_tokens - question_tokens, 0)
        memory_budget = min(memory_budget, self.max_memory_tokens)

        # Build memory context with truncation
        retrieved_memories: List[Dict[str, Any]] = []
        memory_blocks: List[str] = []
        used_tokens = 0

        for item in memory_items:
            mem_text = str(getattr(item, "memory", "") or "").strip()
            if not mem_text:
                continue

            mem_tokens = self._llm_client.count_tokens(mem_text)
            if used_tokens + mem_tokens > memory_budget:
                # Truncate this memory to fit remaining budget
                remaining = memory_budget - used_tokens
                if remaining > 100:
                    truncated = self._truncate_to_tokens(mem_text, remaining)
                    memory_blocks.append(truncated)
                    retrieved_memories.append({