from tikos import (
AddExtractionFile,
AddExtractionFiles,
AddExtractionFileStream,
AddExtractionFileStreams,
ProcessExtract,
ProcessExtractFile,
GetExtract,
GetExtractDocs,
GetGraphStructure,
GenerateGraph,
GetGraph,
GetGraphRetrieval,
GetGraphRetrievalWithDS,
)
from tikos.client import TikosClient
import json
# AddExtractionFile
def AddExtractionFileEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
file = [("<FILENAME.FILETYPE>", "<FILE PATH>")]
status_code, reason, text = AddExtractionFile(
requestId=requestId, authToken=authToken, fileObj=file
)
print(text)
# AddExtractionFiles
def AddExtractionFilesEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
files = [
("<FILENAME1.FILETYPE>", "<FILE PATH>"),
("<FILENAME2.FILETYPE>", "<FILE PATH>"),
]
status_code, reason, text = AddExtractionFiles(
requestId=requestId, authToken=authToken, fileObjs=files
)
print(text)
# AddExtractionFileStream
def AddExtractionFileStreamEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
file = [("<FILENAME.pdf>", open("<FILE PATH>", "rb"))]
status_code, reason, text = AddExtractionFileStream(
requestId=requestId, authToken=authToken, fileObj=file
)
print(text)
# AddExtractionFileStreams
def AddExtractionFileStreamsEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
files = [
("<FILENAME1.FILETYPE>", "<FILE PATH>", "rb"),
("<FILENAME2.FILETYPE>", "<FILE PATH>", "rb"),
]
status_code, reason, text = AddExtractionFileStreams(
requestId=requestId, authToken=authToken, fileObjs=files
)
print(text)
# ProcessExtract
def ProcessExtractEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
jqPattern = "<JQ PATTERN>"
# '.[].description' - to extract description elements of an array of JSON objects
# '.[]' - to extract all elements of an array of JSON objects
status_code, reason, text = ProcessExtract(
requestId=requestId, authToken=authToken, jq=jqPattern
)
print(text)
# ProcessExtractFile
def ProcessExtractFileEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
filesToExtract = ["<FILENAME1>", "<FILENAME2>"]
jqPattern = "<JQ PATTERN>"
# '.[].description' - to extract description elements of an array of JSON objects
# '.[]' - to extract all elements of an array of JSON objects
status_code, reason, text = ProcessExtractFile(
requestId=requestId, authToken=authToken, fileName=filesToExtract, jq=jqPattern
)
print(text)
# GetExtract
def GetExtractDocsEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
status_code, reason, text = GetExtract(requestId=requestId, authToken=authToken)
jsonResponse = json.loads(t)
for doc in jsonResponse:
print(doc.get("textContent"))
return text
# GetExtractDocs
def GetExtractDocsEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
status_code, reason, text = GetExtractDocs(requestId=requestId, authToken=authToken)
jsonResponse = json.loads(t)
for doc in jsonResponse:
print(doc.get("metadata").get("source"))
print(doc.get("page_content"))
return text
# GetGraphStructure
def GetGraphNEREX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
status_code, reason, text = GetGraphStructure(
requestId=requestId, authToken=authToken
)
if (status_code == 200) or (status_code == 201):
jsonResponse = json.loads(text)
jsonLLMResponse = str(jsonResponse.get("llmExtraction"))
jsonNLPResponse = str(jsonResponse.get("nerExtraction"))
print(f"LLM NER Response: {jsonLLMResponse}")
print(f"NLP NER Response: {jsonNLPResponse}")
return text
# GenerateGraph
def GenerateGraphEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
fileStr = str(open("<FILE PATH>").read())
status_code, reason, text = GenerateGraph(
requestId=requestId, authToken=authToken, fileObj=fileStr
)
if (status_code == 200) or (status_code == 201):
jsonResponse = json.loads(text)
print(f"Graph Stats: {jsonResponse}")
return text
# GetGraph
def GetGraphEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
status_code, reason, text = GetGraph(requestId=requestId, authToken=authToken)
if (status_code == 200) or (status_code == 201):
print(f"Graph Structure: {text}")
return text
# GetGraphRetrieval
def GetGraphRetrievalEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
retrieveQuery = "<QUERY?>"
status_code, reason, text = GetGraphRetrieval(
requestId=requestId,
authToken=authToken,
retrieveQuery=retrieveQuery,
)
if (status_code == 200) or (status_code == 201):
print(f"Response: {text}")
return text
# GetGraphRetrievalWithDS
def GetGraphRetrievalWithDSEX():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
retrieveQuery = "<QUERY?>"
status_code, reason, text = GetGraphRetrievalWithDS(
requestId=requestId,
authToken=authToken,
retrieveQuery=retrieveQuery,
)
if (status_code == 200) or (status_code == 201):
print(f"Graph Retrievel With Data Sets: {text}")
return text
def processModelClient_uploadEmbeddingModel():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
modelName = "<MODELNAME>"
modelPath = "<MODELPATH>"
objTC = TikosClient(requestId=requestId, authToken=authToken)
modelObj = (modelName, open(modelPath,"rb"))
rtnVal = objTC.uploadEmbeddingModel(modelObj=modelObj)
print(rtnVal)
print("===============================================================")
def processModelClient_uploadEmbeddingConfig():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
modelConfigName = "<MODELCONFIGNAME>"
modelConfigPath = "<MODELCONFIGPATH>"
modelSchemaName = "<MODELSCHEMANAME>"
modelSchemaPath = "<MODELSCHEMAPATH>"
objTC = TikosClient(requestId=requestId, authToken=authToken)
modelConfig = (modelConfigName, open(modelConfigPath, "rb"))
modelSchema = (modelSchemaName, open(modelSchemaPath, "rb"))
rtnVal = objTC.uploadEmbeddingConfig(modelConfig=modelConfig, modelSchema=modelSchema)
print(rtnVal)
print("===============================================================")
def processModelClient_uploadModelCaseData():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
caseDataSetName = "<CASEDATASETNAME>"
caseDataSetPath = "<CASEDATASETPATH>"
objTC = TikosClient(requestId=requestId, authToken=authToken)
caseDataSet = (caseDataSetName, open(caseDataSetPath, "rb"))
rtnVal = objTC.uploadModelCaseData(caseDataSet=caseDataSet)
print(rtnVal)
print("===============================================================")
def processModelClient_processEmbeddedModel():
requestId = "<REQUESTID>"
authToken = "<AUTHTOKEN>"
caseTypeName = "<CASETYPENAME>"
objTC = TikosClient(requestId=requestId, authToken=authToken)
rtnVal = objTC.processEmbeddedModel(caseType=caseTypeName)
print(rtnVal)
print("===============================================================")
if __name__ == "__main__":
appName = "functions_demo"
# AddExtractionFileEX()
# AddExtractionFilesEX()
# AddExtractionFileStreamEX()
# AddExtractionFileStreamsEX()
# ProcessExtractEX()
# ProcessExtractFileEX()
# GetExtractEX()
# GetExtractDocsEX()
# GetGraphStructureEX()
# GenerateGraphEX()
# GetGraphEX()
# GetGraphRetrievalEX()
# GetGraphRetrievalWithDSEX()
# processModelClient_uploadEmbeddingModel()
# processModelClient_uploadEmbeddingConfig()
# processModelClient_uploadModelCaseData()
# processModelClient_processEmbeddedModel()