from __future__ import annotations
from typing import Any, Dict, Optional
import requests
from langchain_core.callbacks import CallbackManagerForToolRun
from langchain_core.tools import BaseTool
[docs]class CogniswitchKnowledgeRequest(BaseTool):
"""使用Cogniswitch服务来回答问题的工具。
名称: str = "cogniswitch_knowledge_request"
描述: str = (
"一个用于调用cogniswitch服务来从知识库中回答问题的包装器。"
"输入应该是一个搜索查询。"
)"""
name: str = "cogniswitch_knowledge_request"
description: str = """A wrapper around cogniswitch service to
answer the question from the knowledge base."""
cs_token: str
OAI_token: str
apiKey: str
api_url = "https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeRequest"
def _run(
self,
query: str,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> Dict[str, Any]:
"""使用该工具来回答一个查询。
参数:
query (str): 自然语言查询，你想问知识图谱的内容。
run_manager (Optional[CallbackManagerForChainRun]):
用于链式运行回调的管理器。
返回:
Dict[str, Any]: 包含来自服务的 'response' 的输出字典。
"""
response = self.answer_cs(self.cs_token, self.OAI_token, query, self.apiKey)
return response
[docs] def answer_cs(self, cs_token: str, OAI_token: str, query: str, apiKey: str) -> dict:
"""向Cogniswitch服务发送查询并获取响应。
参数:
cs_token (str): Cogniswitch令牌。
OAI_token (str): OpenAI令牌。
apiKey (str): OAuth令牌。
query (str): 待回答的查询。
返回:
dict: Cogniswitch服务的响应JSON。
"""
if not cs_token:
raise ValueError("Missing cs_token")
if not OAI_token:
raise ValueError("Missing OpenAI token")
if not apiKey:
raise ValueError("Missing cogniswitch OAuth token")
if not query:
raise ValueError("Missing input query")
headers = {
"apiKey": apiKey,
"platformToken": cs_token,
"openAIToken": OAI_token,
}
data = {"query": query}
response = requests.post(self.api_url, headers=headers, verify=False, data=data)
return response.json()
[docs]class CogniswitchKnowledgeStatus(BaseTool):
"""使用Cogniswitch服务获取文档或上传的URL的状态的工具。
名称：str = "cogniswitch_knowledge_status"
描述：str = (
"一个包装器，用于获取从URL或文件上传的文档的状态的Cogniswitch服务。"
"输入应为文件名或URL链接"
)"""
name: str = "cogniswitch_knowledge_status"
description: str = """A wrapper around cogniswitch services to know
the status of the document uploaded from a url or a file."""
cs_token: str
OAI_token: str
apiKey: str
knowledge_status_url = (
"https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeSource/status"
)
def _run(
self,
document_name: str,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> Dict[str, Any]:
"""使用该工具来了解上传文档的状态。
参数:
document_name (str): 文档名称或
已上传的url
run_manager (Optional[CallbackManagerForChainRun]):
用于链式运行回调的管理器。
返回:
Dict[str, Any]: 包含来自服务的 'response' 的输出字典。
"""
response = self.knowledge_status(document_name)
return response
[docs] def knowledge_status(self, document_name: str) -> dict:
"""使用此函数来了解已上传文档或URL的状态
参数:
document_name (str): 已上传的文档名称或URL。
返回:
dict: Cogniswitch服务的响应JSON。
"""
params = {"docName": document_name, "platformToken": self.cs_token}
headers = {
"apiKey": self.apiKey,
"openAIToken": self.OAI_token,
"platformToken": self.cs_token,
}
response = requests.get(
self.knowledge_status_url,
headers=headers,
params=params,
verify=False,
)
if response.status_code == 200:
source_info = response.json()
source_data = dict(source_info[-1])
status = source_data.get("status")
if status == 0:
source_data["status"] = "SUCCESS"
elif status == 1:
source_data["status"] = "PROCESSING"
elif status == 2:
source_data["status"] = "UPLOADED"
elif status == 3:
source_data["status"] = "FAILURE"
elif status == 4:
source_data["status"] = "UPLOAD_FAILURE"
elif status == 5:
source_data["status"] = "REJECTED"
if "filePath" in source_data.keys():
source_data.pop("filePath")
if "savedFileName" in source_data.keys():
source_data.pop("savedFileName")
if "integrationConfigId" in source_data.keys():
source_data.pop("integrationConfigId")
if "metaData" in source_data.keys():
source_data.pop("metaData")
if "docEntryId" in source_data.keys():
source_data.pop("docEntryId")
return source_data
else:
# error_message = response.json()["message"]
return {
"message": response.status_code,
}
[docs]class CogniswitchKnowledgeSourceFile(BaseTool):
"""工具使用Cogniswitch服务将数据存储到文件中。
name: str = "cogniswitch_knowledge_source_file"
description: str = (
"这调用CogniSwitch服务来分析和存储来自文件的数据。
如果输入类似文件路径，则将该字符串值分配给文件键。
仅在输入中提供时分配文档名称和描述。"
)"""
name: str = "cogniswitch_knowledge_source_file"
description: str = """
This calls the CogniSwitch services to analyze & store data from a file.
If the input looks like a file path, assign that string value to file key.
Assign document name & description only if provided in input.
"""
cs_token: str
OAI_token: str
apiKey: str
knowledgesource_file = (
"https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeSource/file"
)
def _run(
self,
file: Optional[str] = None,
document_name: Optional[str] = None,
document_description: Optional[str] = None,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> Dict[str, Any]:
"""执行工具以存储从文件中提供的数据。
这将调用CogniSwitch服务来分析并存储来自文件的数据。
如果输入看起来像是文件路径，则将该字符串值分配给文件键。
仅在输入中提供时，分配文档名称和描述。
参数:
file Optional[str]: 您的知识文件路径
document_name Optional[str]: 知识文档的名称
document_description Optional[str]: 知识文档的描述
run_manager (Optional[CallbackManagerForChainRun]):
链式运行回调的管理器。
返回:
Dict[str, Any]: 包含来自服务的 'response' 的输出字典。
"""
if not file:
return {
"message": "No input provided",
}
else:
response = self.store_data(
file=file,
document_name=document_name,
document_description=document_description,
)
return response
[docs] def store_data(
self,
file: Optional[str],
document_name: Optional[str],
document_description: Optional[str],
) -> dict:
"""使用Cogniswitch服务存储数据。
调用CogniSwitch服务来分析和存储文件中的数据。
如果输入看起来像是文件路径，则将该字符串值分配给文件键。
仅在输入中提供文档名称和描述时分配。
参数:
file (Optional[str]): 文件路径。
当前支持的文件类型包括 .txt, .pdf, .docx, .doc, .html
document_name (Optional[str]): 正在上传的文档名称。
document_description (Optional[str]): 文档的描述。
返回:
dict: Cogniswitch服务的响应JSON。
"""
headers = {
"apiKey": self.apiKey,
"openAIToken": self.OAI_token,
"platformToken": self.cs_token,
}
data: Dict[str, Any]
if not document_name:
document_name = ""
if not document_description:
document_description = ""
if file is not None:
files = {"file": open(file, "rb")}
data = {
"documentName": document_name,
"documentDescription": document_description,
}
response = requests.post(
self.knowledgesource_file,
headers=headers,
verify=False,
data=data,
files=files,
)
if response.status_code == 200:
return response.json()
else:
return {"message": "Bad Request"}
[docs]class CogniswitchKnowledgeSourceURL(BaseTool):
"""工具，使用Cogniswitch服务从URL存储数据。
name: str = "cogniswitch_knowledge_source_url"
description: str = (
"这调用CogniSwitch服务来分析和存储来自URL的数据。
输入中提供URL，将该值分配给url键。
只有在输入中提供文档名称和描述时才分配。"
)"""
name: str = "cogniswitch_knowledge_source_url"
description: str = """
This calls the CogniSwitch services to analyze & store data from a url.
the URL is provided in input, assign that value to the url key.
Assign document name & description only if provided in input"""
cs_token: str
OAI_token: str
apiKey: str
knowledgesource_url = (
"https://api.cogniswitch.ai:8243/cs-api/0.0.1/cs/knowledgeSource/url"
)
def _run(
self,
url: Optional[str] = None,
document_name: Optional[str] = None,
document_description: Optional[str] = None,
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> Dict[str, Any]:
"""执行工具以存储从URL提供的数据。
调用CogniSwitch服务分析并存储来自URL的数据。
如果提供了输入，则将URL分配给url键。
仅在输入中提供文档名称和描述时分配。
参数：
url Optional[str]: 您的知识网站/URL链接
document_name Optional[str]: 您的知识文档名称
document_description Optional[str]: 您的知识文档描述
run_manager (Optional[CallbackManagerForChainRun]):
链式运行回调的管理器。
返回：
Dict[str, Any]: 包含服务“响应”的输出字典。
"""
if not url:
return {
"message": "No input provided",
}
response = self.store_data(
url=url,
document_name=document_name,
document_description=document_description,
)
return response
[docs] def store_data(
self,
url: Optional[str],
document_name: Optional[str],
document_description: Optional[str],
) -> dict:
"""使用Cogniswitch服务存储数据。
这调用CogniSwitch服务来分析并存储来自URL的数据。
URL是以输入形式提供的，将该值分配给url键。
只有在输入中提供时，才分配文档名称和描述。
参数：
url（可选[str]）：URL链接。
document_name（可选[str]）：要上传的文档名称。
document_description（可选[str]）：文档描述。
返回：
字典：来自Cogniswitch服务的响应JSON。
"""
headers = {
"apiKey": self.apiKey,
"openAIToken": self.OAI_token,
"platformToken": self.cs_token,
}
data: Dict[str, Any]
if not document_name:
document_name = ""
if not document_description:
document_description = ""
if not url:
return {
"message": "No input provided",
}
else:
data = {"url": url}
response = requests.post(
self.knowledgesource_url,
headers=headers,
verify=False,
data=data,
)
if response.status_code == 200:
return response.json()
else:
return {"message": "Bad Request"}