open-webui

Paused

open-webui / backend /open_webui /apps /webui /routers /knowledge.py

github-actions[bot]

GitHub deploy: 8744a12abb640ebf589b497d1c4f32cb04e0eb84

8050797 almost 2 years ago

15.4 kB

import json
from typing import Optional, Union
from pydantic import BaseModel
from fastapi import APIRouter, Depends, HTTPException, status, Request
import logging

from open_webui.apps.webui.models.knowledge import (
    Knowledges,
    KnowledgeForm,
    KnowledgeResponse,
    KnowledgeUserResponse,
)
from open_webui.apps.webui.models.files import Files, FileModel
from open_webui.apps.retrieval.vector.connector import VECTOR_DB_CLIENT
from open_webui.apps.retrieval.main import process_file, ProcessFileForm


from open_webui.constants import ERROR_MESSAGES
from open_webui.utils.utils import get_admin_user, get_verified_user
from open_webui.utils.access_control import has_access, has_permission


from open_webui.env import SRC_LOG_LEVELS


log = logging.getLogger(__name__)
log.setLevel(SRC_LOG_LEVELS["MODELS"])

router = APIRouter()

############################
# getKnowledgeBases
############################


@router.get("/", response_model=list[KnowledgeUserResponse])
async def get_knowledge(user=Depends(get_verified_user)):
    knowledge_bases = []

    if user.role == "admin":
        knowledge_bases = Knowledges.get_knowledge_bases()
    else:
        knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "read")

    # Get files for each knowledge base
    knowledge_with_files = []
    for knowledge_base in knowledge_bases:
        files = []
        if knowledge_base.data:
            files = Files.get_file_metadatas_by_ids(
                knowledge_base.data.get("file_ids", [])
            )

            # Check if all files exist
            if len(files) != len(knowledge_base.data.get("file_ids", [])):
                missing_files = list(
                    set(knowledge_base.data.get("file_ids", []))
                    - set([file.id for file in files])
                )
                if missing_files:
                    data = knowledge_base.data or {}
                    file_ids = data.get("file_ids", [])

                    for missing_file in missing_files:
                        file_ids.remove(missing_file)

                    data["file_ids"] = file_ids
                    Knowledges.update_knowledge_data_by_id(
                        id=knowledge_base.id, data=data
                    )

                    files = Files.get_file_metadatas_by_ids(file_ids)

        knowledge_with_files.append(
            KnowledgeUserResponse(
                **knowledge_base.model_dump(),
                files=files,
            )
        )

    return knowledge_with_files


@router.get("/list", response_model=list[KnowledgeUserResponse])
async def get_knowledge_list(user=Depends(get_verified_user)):
    knowledge_bases = []

    if user.role == "admin":
        knowledge_bases = Knowledges.get_knowledge_bases()
    else:
        knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "write")

    # Get files for each knowledge base
    knowledge_with_files = []
    for knowledge_base in knowledge_bases:
        files = []
        if knowledge_base.data:
            files = Files.get_file_metadatas_by_ids(
                knowledge_base.data.get("file_ids", [])
            )

            # Check if all files exist
            if len(files) != len(knowledge_base.data.get("file_ids", [])):
                missing_files = list(
                    set(knowledge_base.data.get("file_ids", []))
                    - set([file.id for file in files])
                )
                if missing_files:
                    data = knowledge_base.data or {}
                    file_ids = data.get("file_ids", [])

                    for missing_file in missing_files:
                        file_ids.remove(missing_file)

                    data["file_ids"] = file_ids
                    Knowledges.update_knowledge_data_by_id(
                        id=knowledge_base.id, data=data
                    )

                    files = Files.get_file_metadatas_by_ids(file_ids)

        knowledge_with_files.append(
            KnowledgeUserResponse(
                **knowledge_base.model_dump(),
                files=files,
            )
        )
    return knowledge_with_files


############################
# CreateNewKnowledge
############################


@router.post("/create", response_model=Optional[KnowledgeResponse])
async def create_new_knowledge(
    request: Request, form_data: KnowledgeForm, user=Depends(get_verified_user)
):
    if user.role != "admin" and not has_permission(
        user.id, "workspace.knowledge", request.app.state.config.USER_PERMISSIONS
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.UNAUTHORIZED,
        )

    knowledge = Knowledges.insert_new_knowledge(user.id, form_data)

    if knowledge:
        return knowledge
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.FILE_EXISTS,
        )


############################
# GetKnowledgeById
############################


class KnowledgeFilesResponse(KnowledgeResponse):
    files: list[FileModel]


@router.get("/{id}", response_model=Optional[KnowledgeFilesResponse])
async def get_knowledge_by_id(id: str, user=Depends(get_verified_user)):
    knowledge = Knowledges.get_knowledge_by_id(id=id)

    if knowledge:

        if (
            user.role == "admin"
            or knowledge.user_id == user.id
            or has_access(user.id, "read", knowledge.access_control)
        ):

            file_ids = knowledge.data.get("file_ids", []) if knowledge.data else []
            files = Files.get_files_by_ids(file_ids)

            return KnowledgeFilesResponse(
                **knowledge.model_dump(),
                files=files,
            )
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )


############################
# UpdateKnowledgeById
############################


@router.post("/{id}/update", response_model=Optional[KnowledgeFilesResponse])
async def update_knowledge_by_id(
    id: str,
    form_data: KnowledgeForm,
    user=Depends(get_verified_user),
):
    knowledge = Knowledges.get_knowledge_by_id(id=id)
    if not knowledge:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    if knowledge.user_id != user.id and user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
        )

    knowledge = Knowledges.update_knowledge_by_id(id=id, form_data=form_data)
    if knowledge:
        file_ids = knowledge.data.get("file_ids", []) if knowledge.data else []
        files = Files.get_files_by_ids(file_ids)

        return KnowledgeFilesResponse(
            **knowledge.model_dump(),
            files=files,
        )
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ID_TAKEN,
        )


############################
# AddFileToKnowledge
############################


class KnowledgeFileIdForm(BaseModel):
    file_id: str


@router.post("/{id}/file/add", response_model=Optional[KnowledgeFilesResponse])
def add_file_to_knowledge_by_id(
    id: str,
    form_data: KnowledgeFileIdForm,
    user=Depends(get_verified_user),
):
    knowledge = Knowledges.get_knowledge_by_id(id=id)

    if not knowledge:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    if knowledge.user_id != user.id and user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
        )

    file = Files.get_file_by_id(form_data.file_id)
    if not file:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )
    if not file.data:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.FILE_NOT_PROCESSED,
        )

    # Add content to the vector database
    try:
        process_file(ProcessFileForm(file_id=form_data.file_id, collection_name=id))
    except Exception as e:
        log.debug(e)
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )

    if knowledge:
        data = knowledge.data or {}
        file_ids = data.get("file_ids", [])

        if form_data.file_id not in file_ids:
            file_ids.append(form_data.file_id)
            data["file_ids"] = file_ids

            knowledge = Knowledges.update_knowledge_data_by_id(id=id, data=data)

            if knowledge:
                files = Files.get_files_by_ids(file_ids)

                return KnowledgeFilesResponse(
                    **knowledge.model_dump(),
                    files=files,
                )
            else:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail=ERROR_MESSAGES.DEFAULT("knowledge"),
                )
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=ERROR_MESSAGES.DEFAULT("file_id"),
            )
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )


@router.post("/{id}/file/update", response_model=Optional[KnowledgeFilesResponse])
def update_file_from_knowledge_by_id(
    id: str,
    form_data: KnowledgeFileIdForm,
    user=Depends(get_verified_user),
):
    knowledge = Knowledges.get_knowledge_by_id(id=id)
    if not knowledge:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    if knowledge.user_id != user.id and user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
        )

    file = Files.get_file_by_id(form_data.file_id)
    if not file:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    # Remove content from the vector database
    VECTOR_DB_CLIENT.delete(
        collection_name=knowledge.id, filter={"file_id": form_data.file_id}
    )

    # Add content to the vector database
    try:
        process_file(ProcessFileForm(file_id=form_data.file_id, collection_name=id))
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        )

    if knowledge:
        data = knowledge.data or {}
        file_ids = data.get("file_ids", [])

        files = Files.get_files_by_ids(file_ids)

        return KnowledgeFilesResponse(
            **knowledge.model_dump(),
            files=files,
        )
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )


############################
# RemoveFileFromKnowledge
############################


@router.post("/{id}/file/remove", response_model=Optional[KnowledgeFilesResponse])
def remove_file_from_knowledge_by_id(
    id: str,
    form_data: KnowledgeFileIdForm,
    user=Depends(get_verified_user),
):
    knowledge = Knowledges.get_knowledge_by_id(id=id)
    if not knowledge:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    if knowledge.user_id != user.id and user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
        )

    file = Files.get_file_by_id(form_data.file_id)
    if not file:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    # Remove content from the vector database
    VECTOR_DB_CLIENT.delete(
        collection_name=knowledge.id, filter={"file_id": form_data.file_id}
    )

    result = VECTOR_DB_CLIENT.query(
        collection_name=knowledge.id,
        filter={"file_id": form_data.file_id},
    )

    Files.delete_file_by_id(form_data.file_id)

    if knowledge:
        data = knowledge.data or {}
        file_ids = data.get("file_ids", [])

        if form_data.file_id in file_ids:
            file_ids.remove(form_data.file_id)
            data["file_ids"] = file_ids

            knowledge = Knowledges.update_knowledge_data_by_id(id=id, data=data)

            if knowledge:
                files = Files.get_files_by_ids(file_ids)

                return KnowledgeFilesResponse(
                    **knowledge.model_dump(),
                    files=files,
                )
            else:
                raise HTTPException(
                    status_code=status.HTTP_400_BAD_REQUEST,
                    detail=ERROR_MESSAGES.DEFAULT("knowledge"),
                )
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=ERROR_MESSAGES.DEFAULT("file_id"),
            )
    else:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )


############################
# DeleteKnowledgeById
############################


@router.delete("/{id}/delete", response_model=bool)
async def delete_knowledge_by_id(id: str, user=Depends(get_verified_user)):
    knowledge = Knowledges.get_knowledge_by_id(id=id)
    if not knowledge:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    if knowledge.user_id != user.id and user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
        )

    try:
        VECTOR_DB_CLIENT.delete_collection(collection_name=id)
    except Exception as e:
        log.debug(e)
        pass
    result = Knowledges.delete_knowledge_by_id(id=id)
    return result


############################
# ResetKnowledgeById
############################


@router.post("/{id}/reset", response_model=Optional[KnowledgeResponse])
async def reset_knowledge_by_id(id: str, user=Depends(get_verified_user)):
    knowledge = Knowledges.get_knowledge_by_id(id=id)
    if not knowledge:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.NOT_FOUND,
        )

    if knowledge.user_id != user.id and user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
        )

    try:
        VECTOR_DB_CLIENT.delete_collection(collection_name=id)
    except Exception as e:
        log.debug(e)
        pass

    knowledge = Knowledges.update_knowledge_data_by_id(id=id, data={"file_ids": []})

    return knowledge

	import json
	from typing import Optional, Union
	from pydantic import BaseModel
	from fastapi import APIRouter, Depends, HTTPException, status, Request
	import logging

	from open_webui.apps.webui.models.knowledge import (
	Knowledges,
	KnowledgeForm,
	KnowledgeResponse,
	KnowledgeUserResponse,
	)
	from open_webui.apps.webui.models.files import Files, FileModel
	from open_webui.apps.retrieval.vector.connector import VECTOR_DB_CLIENT
	from open_webui.apps.retrieval.main import process_file, ProcessFileForm


	from open_webui.constants import ERROR_MESSAGES
	from open_webui.utils.utils import get_admin_user, get_verified_user
	from open_webui.utils.access_control import has_access, has_permission


	from open_webui.env import SRC_LOG_LEVELS


	log = logging.getLogger(__name__)
	log.setLevel(SRC_LOG_LEVELS["MODELS"])

	router = APIRouter()

	############################
	# getKnowledgeBases
	############################


	@router.get("/", response_model=list[KnowledgeUserResponse])
	async def get_knowledge(user=Depends(get_verified_user)):
	knowledge_bases = []

	if user.role == "admin":
	knowledge_bases = Knowledges.get_knowledge_bases()
	else:
	knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "read")

	# Get files for each knowledge base
	knowledge_with_files = []
	for knowledge_base in knowledge_bases:
	files = []
	if knowledge_base.data:
	files = Files.get_file_metadatas_by_ids(
	knowledge_base.data.get("file_ids", [])
	)

	# Check if all files exist
	if len(files) != len(knowledge_base.data.get("file_ids", [])):
	missing_files = list(
	set(knowledge_base.data.get("file_ids", []))
	- set([file.id for file in files])
	)
	if missing_files:
	data = knowledge_base.data or {}
	file_ids = data.get("file_ids", [])

	for missing_file in missing_files:
	file_ids.remove(missing_file)

	data["file_ids"] = file_ids
	Knowledges.update_knowledge_data_by_id(
	id=knowledge_base.id, data=data
	)

	files = Files.get_file_metadatas_by_ids(file_ids)

	knowledge_with_files.append(
	KnowledgeUserResponse(
	**knowledge_base.model_dump(),
	files=files,
	)
	)

	return knowledge_with_files


	@router.get("/list", response_model=list[KnowledgeUserResponse])
	async def get_knowledge_list(user=Depends(get_verified_user)):
	knowledge_bases = []

	if user.role == "admin":
	knowledge_bases = Knowledges.get_knowledge_bases()
	else:
	knowledge_bases = Knowledges.get_knowledge_bases_by_user_id(user.id, "write")

	# Get files for each knowledge base
	knowledge_with_files = []
	for knowledge_base in knowledge_bases:
	files = []
	if knowledge_base.data:
	files = Files.get_file_metadatas_by_ids(
	knowledge_base.data.get("file_ids", [])
	)

	# Check if all files exist
	if len(files) != len(knowledge_base.data.get("file_ids", [])):
	missing_files = list(
	set(knowledge_base.data.get("file_ids", []))
	- set([file.id for file in files])
	)
	if missing_files:
	data = knowledge_base.data or {}
	file_ids = data.get("file_ids", [])

	for missing_file in missing_files:
	file_ids.remove(missing_file)

	data["file_ids"] = file_ids
	Knowledges.update_knowledge_data_by_id(
	id=knowledge_base.id, data=data
	)

	files = Files.get_file_metadatas_by_ids(file_ids)

	knowledge_with_files.append(
	KnowledgeUserResponse(
	**knowledge_base.model_dump(),
	files=files,
	)
	)
	return knowledge_with_files


	############################
	# CreateNewKnowledge
	############################


	@router.post("/create", response_model=Optional[KnowledgeResponse])
	async def create_new_knowledge(
	request: Request, form_data: KnowledgeForm, user=Depends(get_verified_user)
	):
	if user.role != "admin" and not has_permission(
	user.id, "workspace.knowledge", request.app.state.config.USER_PERMISSIONS
	):
	raise HTTPException(
	status_code=status.HTTP_401_UNAUTHORIZED,
	detail=ERROR_MESSAGES.UNAUTHORIZED,
	)

	knowledge = Knowledges.insert_new_knowledge(user.id, form_data)

	if knowledge:
	return knowledge
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.FILE_EXISTS,
	)


	############################
	# GetKnowledgeById
	############################


	class KnowledgeFilesResponse(KnowledgeResponse):
	files: list[FileModel]


	@router.get("/{id}", response_model=Optional[KnowledgeFilesResponse])
	async def get_knowledge_by_id(id: str, user=Depends(get_verified_user)):
	knowledge = Knowledges.get_knowledge_by_id(id=id)

	if knowledge:

	if (
	user.role == "admin"
	or knowledge.user_id == user.id
	or has_access(user.id, "read", knowledge.access_control)
	):

	file_ids = knowledge.data.get("file_ids", []) if knowledge.data else []
	files = Files.get_files_by_ids(file_ids)

	return KnowledgeFilesResponse(
	**knowledge.model_dump(),
	files=files,
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_401_UNAUTHORIZED,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)


	############################
	# UpdateKnowledgeById
	############################


	@router.post("/{id}/update", response_model=Optional[KnowledgeFilesResponse])
	async def update_knowledge_by_id(
	id: str,
	form_data: KnowledgeForm,
	user=Depends(get_verified_user),
	):
	knowledge = Knowledges.get_knowledge_by_id(id=id)
	if not knowledge:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	if knowledge.user_id != user.id and user.role != "admin":
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
	)

	knowledge = Knowledges.update_knowledge_by_id(id=id, form_data=form_data)
	if knowledge:
	file_ids = knowledge.data.get("file_ids", []) if knowledge.data else []
	files = Files.get_files_by_ids(file_ids)

	return KnowledgeFilesResponse(
	**knowledge.model_dump(),
	files=files,
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ID_TAKEN,
	)


	############################
	# AddFileToKnowledge
	############################


	class KnowledgeFileIdForm(BaseModel):
	file_id: str


	@router.post("/{id}/file/add", response_model=Optional[KnowledgeFilesResponse])
	def add_file_to_knowledge_by_id(
	id: str,
	form_data: KnowledgeFileIdForm,
	user=Depends(get_verified_user),
	):
	knowledge = Knowledges.get_knowledge_by_id(id=id)

	if not knowledge:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	if knowledge.user_id != user.id and user.role != "admin":
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
	)

	file = Files.get_file_by_id(form_data.file_id)
	if not file:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)
	if not file.data:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.FILE_NOT_PROCESSED,
	)

	# Add content to the vector database
	try:
	process_file(ProcessFileForm(file_id=form_data.file_id, collection_name=id))
	except Exception as e:
	log.debug(e)
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=str(e),
	)

	if knowledge:
	data = knowledge.data or {}
	file_ids = data.get("file_ids", [])

	if form_data.file_id not in file_ids:
	file_ids.append(form_data.file_id)
	data["file_ids"] = file_ids

	knowledge = Knowledges.update_knowledge_data_by_id(id=id, data=data)

	if knowledge:
	files = Files.get_files_by_ids(file_ids)

	return KnowledgeFilesResponse(
	**knowledge.model_dump(),
	files=files,
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.DEFAULT("knowledge"),
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.DEFAULT("file_id"),
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)


	@router.post("/{id}/file/update", response_model=Optional[KnowledgeFilesResponse])
	def update_file_from_knowledge_by_id(
	id: str,
	form_data: KnowledgeFileIdForm,
	user=Depends(get_verified_user),
	):
	knowledge = Knowledges.get_knowledge_by_id(id=id)
	if not knowledge:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	if knowledge.user_id != user.id and user.role != "admin":
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
	)

	file = Files.get_file_by_id(form_data.file_id)
	if not file:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	# Remove content from the vector database
	VECTOR_DB_CLIENT.delete(
	collection_name=knowledge.id, filter={"file_id": form_data.file_id}
	)

	# Add content to the vector database
	try:
	process_file(ProcessFileForm(file_id=form_data.file_id, collection_name=id))
	except Exception as e:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=str(e),
	)

	if knowledge:
	data = knowledge.data or {}
	file_ids = data.get("file_ids", [])

	files = Files.get_files_by_ids(file_ids)

	return KnowledgeFilesResponse(
	**knowledge.model_dump(),
	files=files,
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)


	############################
	# RemoveFileFromKnowledge
	############################


	@router.post("/{id}/file/remove", response_model=Optional[KnowledgeFilesResponse])
	def remove_file_from_knowledge_by_id(
	id: str,
	form_data: KnowledgeFileIdForm,
	user=Depends(get_verified_user),
	):
	knowledge = Knowledges.get_knowledge_by_id(id=id)
	if not knowledge:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	if knowledge.user_id != user.id and user.role != "admin":
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
	)

	file = Files.get_file_by_id(form_data.file_id)
	if not file:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	# Remove content from the vector database
	VECTOR_DB_CLIENT.delete(
	collection_name=knowledge.id, filter={"file_id": form_data.file_id}
	)

	result = VECTOR_DB_CLIENT.query(
	collection_name=knowledge.id,
	filter={"file_id": form_data.file_id},
	)

	Files.delete_file_by_id(form_data.file_id)

	if knowledge:
	data = knowledge.data or {}
	file_ids = data.get("file_ids", [])

	if form_data.file_id in file_ids:
	file_ids.remove(form_data.file_id)
	data["file_ids"] = file_ids

	knowledge = Knowledges.update_knowledge_data_by_id(id=id, data=data)

	if knowledge:
	files = Files.get_files_by_ids(file_ids)

	return KnowledgeFilesResponse(
	**knowledge.model_dump(),
	files=files,
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.DEFAULT("knowledge"),
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.DEFAULT("file_id"),
	)
	else:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)


	############################
	# DeleteKnowledgeById
	############################


	@router.delete("/{id}/delete", response_model=bool)
	async def delete_knowledge_by_id(id: str, user=Depends(get_verified_user)):
	knowledge = Knowledges.get_knowledge_by_id(id=id)
	if not knowledge:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	if knowledge.user_id != user.id and user.role != "admin":
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
	)

	try:
	VECTOR_DB_CLIENT.delete_collection(collection_name=id)
	except Exception as e:
	log.debug(e)
	pass
	result = Knowledges.delete_knowledge_by_id(id=id)
	return result


	############################
	# ResetKnowledgeById
	############################


	@router.post("/{id}/reset", response_model=Optional[KnowledgeResponse])
	async def reset_knowledge_by_id(id: str, user=Depends(get_verified_user)):
	knowledge = Knowledges.get_knowledge_by_id(id=id)
	if not knowledge:
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.NOT_FOUND,
	)

	if knowledge.user_id != user.id and user.role != "admin":
	raise HTTPException(
	status_code=status.HTTP_400_BAD_REQUEST,
	detail=ERROR_MESSAGES.ACCESS_PROHIBITED,
	)

	try:
	VECTOR_DB_CLIENT.delete_collection(collection_name=id)
	except Exception as e:
	log.debug(e)
	pass

	knowledge = Knowledges.update_knowledge_data_by_id(id=id, data={"file_ids": []})

	return knowledge