openassistant/testr2r.py

from typing import Optional, Any

from r2r import R2RClient

from app.libs.util import verify_jwt_expiration
from config.llm import tool_settings

import inspect

import r2r

print(r2r.__version__)

class R2R:
    client: R2RClient

    def init(self):
        self.client = R2RClient(tool_settings.R2R_BASE_URL)
        
        self.auth_enabled = tool_settings.R2R_USERNAME and tool_settings.R2R_PASSWORD
        self._login()

    def ingest_file(self, file_path: str, metadata: Optional[dict]):
        self._check_login()
        ingest_response = self.client.ingest_files(
            file_paths=[file_path],
            metadatas=[metadata] if metadata else None
        )
        return ingest_response.get("results")

    def search(self, query: str, filters: dict[str, Any]):
        self._check_login()
        search_response = self.client.search(
            query=query,
            vector_search_settings={
                "filters": filters,
                "search_limit": tool_settings.R2R_SEARCH_LIMIT,
                "do_hybrid_search": True
            },
        )
        return search_response.get("results").get("vector_search_results")

    def _login(self):
        if not self.auth_enabled:
            return
        print("test")
        print(self.client)
        print(self.client.users)
        print(self.client._wrap_v3_methods)
        # 获取实例方法
        instance_methods = inspect.getmembers(self.client._wrap_v3_methods, predicate=inspect.isfunction)
        print("实例方法:", [method[0] for method in instance_methods])

        # 获取类方法（包括静态方法）
        class_methods = inspect.getmembers(self.client._wrap_v3_methods, predicate=inspect.ismethod)
        print("类方法:", [method[0] for method in class_methods])
        #self.client.user.login(tool_settings.R2R_USERNAME, tool_settings.R2R_PASSWORD)

    def _check_login(self):
        if not self.auth_enabled:
            return
        if verify_jwt_expiration(self.client.access_token):
            return
        if verify_jwt_expiration(self.client._refresh_token):
            self.client.refresh_access_token()
        else:
            self._login()

def get_method_parameters(method):
    params = inspect.signature(method).parameters
    for name, param in params.items():
        default_value = param.default if param.default is not param.empty else None
        print(f"参数名: {name}, 默认值: {default_value}")

# 获取实例方法
instance_methods = inspect.getmembers(R2RClient, predicate=inspect.isfunction)
print("实例方法:", [method[0] for method in instance_methods])

# 获取类方法（包括静态方法）
class_methods = inspect.getmembers(R2RClient, predicate=inspect.ismethod)
print("类方法:", [method[0] for method in class_methods])

r2r1 = R2R()

r2r1.init()