Rag user

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

    # Normalize to list
    doc_paths = (
        [self.docs_path]
        if isinstance(self.docs_path, str)
        else self.docs_path
    )

    validated_paths: list[str] = []

    for path in doc_paths:
        # Skip duplicates
        if path in validated_paths:
            continue

        # Check if it's a remote path
        is_remote = is_remote_path(path)
        if is_remote:
            # Remote paths: ensure proper raw string wrapping if needed
            content = extract_raw_string_content(path)
            validated_paths.append(f'r"{content}"')
            continue

        # Handle local paths
        # First remove any file:// scheme
        cleaned_path = remove_file_scheme(path)
        content = extract_raw_string_content(cleaned_path)

        # Determine if it's likely a folder
        is_folder = string_represents_folder(content)

        if is_folder:
            validated_paths.append(f'r"{content}"')
        else:
            # Files: resolve and validate existence
            try:
                resolved_path = resolve_path(cleaned_path, must_exist=True)
                validated_paths.append(resolved_path)
            except ValueError as e:
                raise ValueError(f"Invalid file path '{path}': {e}") from e

    # remove dupes (but keep order)
    validated_paths = list(dict.fromkeys(validated_paths))
    self.docs_path = [path for path in validated_paths if path]

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

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

    Returns
    -------
    WaldiezRagUserProxyData
        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 = WaldiezRagUserProxyModels[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
    -------
    WaldiezRagUserProxyVectorDbConfig
        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