from typing import Optional, Any from r2r import R2RAsyncClient from r2r import R2RClient from fastapi import UploadFile from app.libs.util import verify_jwt_expiration from config.llm import tool_settings import nest_asyncio # 使得异步代码可以在已运行的事件循环中嵌套 nest_asyncio.apply() class R2R: client: R2RAsyncClient client_sync: R2RClient def __init__(self): self.auth_enabled = tool_settings.R2R_USERNAME and tool_settings.R2R_PASSWORD self.client = None self.client_sync = None def init_sync(self): if not self.auth_enabled: return if not self.client_sync: self.client_sync = R2RClient(tool_settings.R2R_BASE_URL, 300) self.client_sync.users.login( tool_settings.R2R_USERNAME, tool_settings.R2R_PASSWORD ) print( "1111111111111111111111111111111122222vvdgdfdf" + tool_settings.R2R_USERNAME ) print(tool_settings.R2R_USERNAME) print(tool_settings.R2R_PASSWORD) # print(self.client_sync.access_token) async def init(self): if not self.auth_enabled: return if not self.client: self.client = R2RAsyncClient(tool_settings.R2R_BASE_URL, 300) print( "1111111111111111111111111111111122222vvdgdfdf" + tool_settings.R2R_USERNAME ) print(tool_settings.R2R_USERNAME) print(tool_settings.R2R_PASSWORD) await self.client.users.login( tool_settings.R2R_USERNAME, tool_settings.R2R_PASSWORD ) print(self.client.access_token) async def ingest_file(self, file_path: str, metadata: Optional[dict]): await self._check_login() return await self.client.documents.create( file_path=file_path, metadata=metadata if metadata else None, ingestion_mode="fast", id=None, ) async def ingest_fileinfo(self, file: UploadFile, metadata: Optional[dict]): await self._check_login() return await self.client.documents.create( file=file, metadata=metadata if metadata else None, id=None, ) def search(self, query: str, filters: dict[str, Any]): self._check_login_sync() print( "aaaaaaaaaaaaaaaaaaaaaaaaaaaasssssssssssssssssssssssssssssssssssssssssgggggggggggggggggggg" ) search_response = self.client_sync.retrieval.search( query=query, search_mode="basic", search_settings={ "filters": filters, "limit": tool_settings.R2R_SEARCH_LIMIT, }, ) print("vvvvvvvvvvvvvvvvvvmmmmmmmmmmmmmmmmmmmmmmmmmmmmmm") # print(search_response) print(search_response.results) return search_response.results.chunk_search_results def list_chunks(self, ids: list[str] = []): self._check_login_sync() print( "retrieve_documentsretrieve_documentsretrieve_documentsretrieve_documentsretrieve_documents" ) print(ids) allfile = [] for id in ids: listed = self.client_sync.documents.list_chunks(id=id) allfile += listed.results return allfile def list_documents( self, id: Optional[str] = "", offset: Optional[int] = 0, limit: Optional[int] = 100, ): self._check_login_sync() """ docs = client.collections.list_documents(empty_coll_id).results assert len(docs) == 0, "Expected no documents in a new empty collection" """ print( "collectionscollectionscollectionscollectionscollectionscollectionscollectionscollectionscollectionscollectionscollectionscollections" ) if id != "": try: listed = self.client_sync.collections.list_documents( id=id, limit=limit, offset=offset ) print(listed.results) return listed.results except Exception as e: print(e) listed = [] return listed else: return [] async def _check_login(self): if not self.auth_enabled: return if not self.client.access_token and verify_jwt_expiration( self.client.access_token ): return else: await self.init() def _check_login_sync(self): if not self.auth_enabled: return try: if not self.client_sync.access_token and verify_jwt_expiration( self.client_sync.access_token ): return except Exception as e: print(e) self.init_sync() # 创建 R2R 实例 r2r = R2R() # 在您的应用程序启动时调用 initialize_r2r() async def initialize_r2r(): await r2r.init()