Skip to content

Cogniswitch

CogniswitchToolSpec #

Bases: BaseToolSpec

Cogniswitch工具规范。 一个工具规范,具有store_data和query_knowledge作为工具,用于存储来自文件或URL的数据,并分别从存储的知识中回答问题。

Source code in llama_index/tools/cogniswitch/base.py
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class CogniswitchToolSpec(BaseToolSpec):
    """Cogniswitch工具规范。
    一个工具规范,具有store_data和query_knowledge作为工具,用于存储来自文件或URL的数据,并分别从存储的知识中回答问题。"""

    spec_functions = ["store_data", "query_knowledge", "knowledge_status"]

    def __init__(
        self,
        cs_token: str,
        apiKey: str,
        OAI_token: Optional[str] = None,
    ) -> None:
        """Args:
    cs_token(str):Cogniswitch令牌。
    OAI_token(str):OpenAI令牌。
    apiKey(str):Oauth令牌。
"""
        self.cs_token = cs_token
        if OAI_token:
            self.OAI_token = OAI_token
        elif os.environ["OPENAI_API_KEY"]:
            self.OAI_token = os.environ["OPENAI_API_KEY"]
        else:
            raise ValueError("Please provide the OpenAI token")
        self.apiKey = apiKey
        self.source_URL_endpoint = (
            "https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeSource/url"
        )
        self.source_file_endpoint = (
            "https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeSource/file"
        )
        self.knowledge_request_endpoint = (
            "https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeRequest"
        )
        self.knowledge_status_endpoint = (
            "https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeSource/status"
        )
        self.headers = {
            "apiKey": self.apiKey,
            "platformToken": self.cs_token,
            "openAIToken": self.OAI_token,
        }

    def store_data(
        self,
        url: Optional[str] = None,
        file: Optional[str] = None,
        document_name: Optional[str] = None,
        document_description: Optional[str] = None,
    ) -> dict:
        """使用Cogniswitch服务存储数据。

Args:
    url(可选[str]):URL链接。
    file(可选[str]):文件路径。
    目前支持的文件类型为.txt, .pdf, .docx, .doc, .html
    document_name(可选[str]):上传的文档名称。
    document_description(可选[str]):文档描述。

Returns:
    dict: Cogniswitch服务的响应JSON。
"""
        if not file and not url:
            return {
                "message": "No input provided",
            }
        elif file and url:
            return {
                "message": "Too many inputs, please provide either file or url",
            }
        elif url:
            api_url = self.source_URL_endpoint
            headers = self.headers
            files = None
            data = {
                "url": url,
                "documentName": document_name,
                "documentDescription": document_description,
            }
            response = requests.post(api_url, headers=headers, data=data, files=files)

        elif file:
            api_url = self.source_file_endpoint

            headers = self.headers
            if file is not None:
                files = {"file": open(file, "rb")}
            else:
                files = None
            data = {
                "url": url,
                "documentName": document_name,
                "documentDescription": document_description,
            }
            response = requests.post(api_url, headers=headers, data=data, files=files)
        if response.status_code == 200:
            return response.json()
        else:
            # error_message = response.json()["message"]
            return {
                "message": "Bad Request",
            }

    def query_knowledge(self, query: str) -> dict:
        """向Cogniswitch服务发送查询并检索响应。

Args:
    query (str): 待回答的查询。

Returns:
    dict: 来自Cogniswitch服务的响应JSON。
"""
        api_url = self.knowledge_request_endpoint

        headers = self.headers

        data = {"query": query}
        response = requests.post(api_url, headers=headers, data=data)
        if response.status_code == 200:
            return response.json()
        else:
            # error_message = response.json()["message"]
            return {
                "message": "Bad Request",
            }

    def knowledge_status(self, document_name: str) -> dict:
        """使用此函数来了解已上传文档或URL的状态
Args:
    document_name (str): 已上传的文档名称或URL。

Returns:
    dict: Cogniswitch服务的响应JSON。
"""
        params = {"docName": document_name, "platformToken": self.cs_token}
        response = requests.get(
            self.knowledge_status_endpoint,
            headers=self.headers,
            params=params,
        )
        if response.status_code == 200:
            source_info = response.json()
            return source_info[-1]
        else:
            # error_message = response.json()["message"]
            return {
                "message": "Bad Request",
            }

store_data #

store_data(
    url: Optional[str] = None,
    file: Optional[str] = None,
    document_name: Optional[str] = None,
    document_description: Optional[str] = None,
) -> dict

使用Cogniswitch服务存储数据。

Returns:

Name Type Description
dict dict

Cogniswitch服务的响应JSON。

Source code in llama_index/tools/cogniswitch/base.py
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
    def store_data(
        self,
        url: Optional[str] = None,
        file: Optional[str] = None,
        document_name: Optional[str] = None,
        document_description: Optional[str] = None,
    ) -> dict:
        """使用Cogniswitch服务存储数据。

Args:
    url(可选[str]):URL链接。
    file(可选[str]):文件路径。
    目前支持的文件类型为.txt, .pdf, .docx, .doc, .html
    document_name(可选[str]):上传的文档名称。
    document_description(可选[str]):文档描述。

Returns:
    dict: Cogniswitch服务的响应JSON。
"""
        if not file and not url:
            return {
                "message": "No input provided",
            }
        elif file and url:
            return {
                "message": "Too many inputs, please provide either file or url",
            }
        elif url:
            api_url = self.source_URL_endpoint
            headers = self.headers
            files = None
            data = {
                "url": url,
                "documentName": document_name,
                "documentDescription": document_description,
            }
            response = requests.post(api_url, headers=headers, data=data, files=files)

        elif file:
            api_url = self.source_file_endpoint

            headers = self.headers
            if file is not None:
                files = {"file": open(file, "rb")}
            else:
                files = None
            data = {
                "url": url,
                "documentName": document_name,
                "documentDescription": document_description,
            }
            response = requests.post(api_url, headers=headers, data=data, files=files)
        if response.status_code == 200:
            return response.json()
        else:
            # error_message = response.json()["message"]
            return {
                "message": "Bad Request",
            }

query_knowledge #

query_knowledge(query: str) -> dict

向Cogniswitch服务发送查询并检索响应。

Parameters:

Name Type Description Default
query str

待回答的查询。

required

Returns:

Name Type Description
dict dict

来自Cogniswitch服务的响应JSON。

Source code in llama_index/tools/cogniswitch/base.py
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
    def query_knowledge(self, query: str) -> dict:
        """向Cogniswitch服务发送查询并检索响应。

Args:
    query (str): 待回答的查询。

Returns:
    dict: 来自Cogniswitch服务的响应JSON。
"""
        api_url = self.knowledge_request_endpoint

        headers = self.headers

        data = {"query": query}
        response = requests.post(api_url, headers=headers, data=data)
        if response.status_code == 200:
            return response.json()
        else:
            # error_message = response.json()["message"]
            return {
                "message": "Bad Request",
            }

knowledge_status #

knowledge_status(document_name: str) -> dict

使用此函数来了解已上传文档或URL的状态 Args: document_name (str): 已上传的文档名称或URL。

Returns:

Name Type Description
dict dict

Cogniswitch服务的响应JSON。

Source code in llama_index/tools/cogniswitch/base.py
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
    def knowledge_status(self, document_name: str) -> dict:
        """使用此函数来了解已上传文档或URL的状态
Args:
    document_name (str): 已上传的文档名称或URL。

Returns:
    dict: Cogniswitch服务的响应JSON。
"""
        params = {"docName": document_name, "platformToken": self.cs_token}
        response = requests.get(
            self.knowledge_status_endpoint,
            headers=self.headers,
            params=params,
        )
        if response.status_code == 200:
            source_info = response.json()
            return source_info[-1]
        else:
            # error_message = response.json()["message"]
            return {
                "message": "Bad Request",
            }