RagUser

def get_custom_embedding_function(
    self,
    name_prefix: Optional[str] = None,
    name_suffix: Optional[str] = None,
) -> Tuple[str, str]:
    """Generate the custom embedding function.

    Parameters
    ----------
    name_prefix : str
        The function name prefix.
    name_suffix : str
        The function name suffix.

    Returns
    -------
    Tuple[str, str]
        The custom embedding function and the function name.
    """
    function_name = CUSTOM_EMBEDDING_FUNCTION
    if name_prefix:
        function_name = f"{name_prefix}_{function_name}"
    if name_suffix:
        function_name = f"{function_name}_{name_suffix}"
    return (
        generate_function(
            function_name=function_name,
            function_args=CUSTOM_EMBEDDING_FUNCTION_ARGS,
            function_types=CUSTOM_EMBEDDING_FUNCTION_TYPES,
            function_body=self.embedding_function_string or "",
        ),
        function_name,
    )

def get_custom_text_split_function(
    self,
    name_prefix: Optional[str] = None,
    name_suffix: Optional[str] = None,
) -> Tuple[str, str]:
    """Generate the custom text split function.

    Parameters
    ----------
    name_prefix : str
        The function name prefix.
    name_suffix : str
        The function name suffix.

    Returns
    -------
    Tuple[str, str]
        The custom text split function and the function name.
    """
    function_name = CUSTOM_TEXT_SPLIT_FUNCTION
    if name_prefix:
        function_name = f"{name_prefix}_{function_name}"
    if name_suffix:
        function_name = f"{function_name}_{name_suffix}"
    return (
        generate_function(
            function_name=function_name,
            function_args=CUSTOM_TEXT_SPLIT_FUNCTION_ARGS,
            function_types=CUSTOM_TEXT_SPLIT_FUNCTION_TYPES,
            function_body=self.text_split_function_string or "",
        ),
        function_name,
    )

def get_custom_token_count_function(
    self,
    name_prefix: Optional[str] = None,
    name_suffix: Optional[str] = None,
) -> Tuple[str, str]:
    """Generate the custom token count function.

    Parameters
    ----------
    name_prefix : str
        The function name prefix.
    name_suffix : str
        The function name suffix.

    Returns
    -------
    Tuple[str, str]
        The custom token count function and the function name.
    """
    function_name = CUSTOM_TOKEN_COUNT_FUNCTION
    if name_prefix:
        function_name = f"{name_prefix}_{function_name}"
    if name_suffix:
        function_name = f"{function_name}_{name_suffix}"
    return (
        generate_function(
            function_name=function_name,
            function_args=CUSTOM_TOKEN_COUNT_FUNCTION_ARGS,
            function_types=CUSTOM_TOKEN_COUNT_FUNCTION_TYPES,
            function_body=self.token_count_function_string or "",
        ),
        function_name,
    )

def validate_custom_embedding_function(self) -> None:
    """Validate the custom embedding function.

    Raises
    ------
    ValueError
        If the validation fails.
    """
    if self.use_custom_embedding:
        if not self.embedding_function:
            raise ValueError(
                "The embedding_function is required "
                "if use_custom_embedding is True."
            )
        valid, error_or_content = check_function(
            code_string=self.embedding_function,
            function_name=CUSTOM_EMBEDDING_FUNCTION,
            function_args=CUSTOM_EMBEDDING_FUNCTION_ARGS,
        )
        if not valid:
            raise ValueError(error_or_content)
        self._embedding_function_string = error_or_content

def validate_custom_text_split_function(self) -> None:
    """Validate the custom text split function.

    Raises
    ------
    ValueError
        If the validation fails.
    """
    if self.use_custom_text_split:
        if not self.custom_text_split_function:
            raise ValueError(
                "The custom_text_split_function is required "
                "if use_custom_text_split is True."
            )
        valid, error_or_content = check_function(
            code_string=self.custom_text_split_function,
            function_name=CUSTOM_TEXT_SPLIT_FUNCTION,
            function_args=CUSTOM_TEXT_SPLIT_FUNCTION_ARGS,
        )
        if not valid:
            raise ValueError(error_or_content)
        self._text_split_function_string = error_or_content

def validate_custom_token_count_function(self) -> None:
    """Validate the custom token count function.

    Raises
    ------
    ValueError
        If the validation fails.
    """
    if self.use_custom_token_count:
        if not self.custom_token_count_function:
            raise ValueError(
                "The custom_token_count_function is required "
                "if use_custom_token_count is True."
            )
        valid, error_or_content = check_function(
            code_string=self.custom_token_count_function,
            function_name=CUSTOM_TOKEN_COUNT_FUNCTION,
            function_args=CUSTOM_TOKEN_COUNT_FUNCTION_ARGS,
        )
        if not valid:
            raise ValueError(error_or_content)
        self._token_count_function_string = error_or_content

def validate_docs_path(self) -> None:
    """Validate the docs path.

    Raises
    ------
    ValueError
        If the validation fails.
    """
    if not self.docs_path:
        return
    # if urls or directories ok, if files they should resolve
    doc_paths = (
        [self.docs_path]
        if isinstance(self.docs_path, str)
        else self.docs_path
    )
    paths: List[str] = []
    for path in doc_paths:
        resolved = path
        is_remote, is_raw = is_remote_path(resolved)
        if is_remote:
            if not is_raw:
                resolved = f'r"{resolved}"'
            if resolved not in paths:
                paths.append(resolved)
            continue
        resolved = remove_file_scheme(resolved)
        is_raw = resolved.startswith(("r'", 'r"'))
        maybe_folder = string_represents_folder(resolved)
        if maybe_folder:
            if not is_raw:
                resolved = f'r"{resolved}"'
            if resolved not in paths:
                paths.append(resolved)
            continue
        resolved = resolve_path(resolved, is_raw, not maybe_folder)
        if resolved not in paths:
            paths.append(resolved)
    self.docs_path = paths

@model_validator(mode="after")
def validate_rag_user_data(self) -> Self:
    """Validate the RAG user data.

    Raises
    ------
    ValueError
        If the validation fails.

    Returns
    -------
    WaldiezRagUserData
        The validated RAG user data.
    """
    self.validate_custom_embedding_function()
    self.validate_custom_token_count_function()
    self.validate_custom_text_split_function()
    self.validate_docs_path()
    if not self.db_config.model:
        self.db_config.model = WaldiezRagUserModels[self.vector_db]
    if isinstance(self.n_results, int) and self.n_results < 1:
        self.n_results = None
    return self

@model_validator(mode="after")
def validate_vector_db_config(self) -> Self:
    """Validate the vector db config.

    if local storage is used, make sure the path is provided,
    and make it absolute if not already.

    Returns
    -------
    WaldiezRagUserVectorDbConfig
        The vector db config.

    Raises
    ------
    ValueError
        If the validation fails.
    """
    if self.use_local_storage:
        if self.local_storage_path is None:
            raise ValueError(
                "The local storage path must be provided if local storage is used."
            )
        as_path = Path(self.local_storage_path)
        if not as_path.is_absolute():
            self.local_storage_path = str(as_path.resolve())
    return self