mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-04-26 08:56:21 +02:00
Add Python support to calling the API (#169)
This commit is contained in:
cybermaggedon
GitHub
parent
7a64385a57
commit
ae1264f5c4
2 changed files with 339 additions and 0 deletions
3
trustgraph-base/trustgraph/api/__init__.py
Normal file
3
trustgraph-base/trustgraph/api/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
|
||||
from . api import *
|
||||
|
||||
336
trustgraph-base/trustgraph/api/api.py
Normal file
336
trustgraph-base/trustgraph/api/api.py
Normal file
|
|
@ -0,0 +1,336 @@
|
|||
|
||||
import requests
|
||||
import json
|
||||
import dataclasses
|
||||
import base64
|
||||
|
||||
from trustgraph.knowledge import hash
|
||||
|
||||
class ProtocolException(Exception):
|
||||
pass
|
||||
|
||||
class ApplicationException(Exception):
|
||||
pass
|
||||
|
||||
class Uri(str):
|
||||
def is_uri(self): return True
|
||||
def is_literal(self): return False
|
||||
|
||||
class Literal(str):
|
||||
def is_uri(self): return False
|
||||
def is_literal(self): return True
|
||||
|
||||
@dataclasses.dataclass
|
||||
class Triple:
|
||||
s : str
|
||||
p : str
|
||||
o : str
|
||||
|
||||
class Api:
|
||||
|
||||
def __init__(self, url="http://localhost:8088/"):
|
||||
|
||||
self.url = url
|
||||
|
||||
if not url.endswith("/"):
|
||||
self.url += "/"
|
||||
|
||||
self.url += "api/v1/"
|
||||
|
||||
def check_error(self, response):
|
||||
|
||||
if "error" in response:
|
||||
|
||||
try:
|
||||
msg = response["error"]["message"]
|
||||
tp = response["error"]["message"]
|
||||
except:
|
||||
raise ApplicationException(
|
||||
"Error, but the error object is broken"
|
||||
)
|
||||
|
||||
raise ApplicationException(f"{tp}: {msg}")
|
||||
|
||||
def text_completion(self, system, prompt):
|
||||
|
||||
# The input consists of system and prompt strings
|
||||
input = {
|
||||
"system": system,
|
||||
"prompt": prompt
|
||||
}
|
||||
|
||||
url = f"{self.url}text-completion"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
try:
|
||||
# Parse the response as JSON
|
||||
object = resp.json()
|
||||
except:
|
||||
raise ProtocolException(f"Expected JSON response")
|
||||
|
||||
self.check_error(resp)
|
||||
|
||||
try:
|
||||
return object["response"]
|
||||
except:
|
||||
raise ProtocolException(f"Response not formatted correctly")
|
||||
|
||||
def agent(self, question):
|
||||
|
||||
# The input consists of a question
|
||||
input = {
|
||||
"question": question
|
||||
}
|
||||
|
||||
url = f"{self.url}agent"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
try:
|
||||
# Parse the response as JSON
|
||||
object = resp.json()
|
||||
except:
|
||||
raise ProtocolException(f"Expected JSON response")
|
||||
|
||||
self.check_error(resp)
|
||||
|
||||
try:
|
||||
return object["answer"]
|
||||
except:
|
||||
raise ProtocolException(f"Response not formatted correctly")
|
||||
|
||||
def graph_rag(self, question):
|
||||
|
||||
# The input consists of a question
|
||||
input = {
|
||||
"query": question
|
||||
}
|
||||
|
||||
url = f"{self.url}graph-rag"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
try:
|
||||
# Parse the response as JSON
|
||||
object = resp.json()
|
||||
except:
|
||||
raise ProtocolException(f"Expected JSON response")
|
||||
|
||||
self.check_error(resp)
|
||||
|
||||
try:
|
||||
return object["response"]
|
||||
except:
|
||||
raise ProtocolException(f"Response not formatted correctly")
|
||||
|
||||
def embeddings(self, text):
|
||||
|
||||
# The input consists of a text block
|
||||
input = {
|
||||
"text": text
|
||||
}
|
||||
|
||||
url = f"{self.url}embeddings"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
try:
|
||||
# Parse the response as JSON
|
||||
object = resp.json()
|
||||
except:
|
||||
raise ProtocolException(f"Expected JSON response")
|
||||
|
||||
self.check_error(resp)
|
||||
|
||||
try:
|
||||
return object["vectors"]
|
||||
except:
|
||||
raise ProtocolException(f"Response not formatted correctly")
|
||||
|
||||
def prompt(self, id, variables):
|
||||
|
||||
# The input consists of system and prompt strings
|
||||
input = {
|
||||
"id": id,
|
||||
"variables": variables
|
||||
}
|
||||
|
||||
url = f"{self.url}prompt"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
try:
|
||||
# Parse the response as JSON
|
||||
object = resp.json()
|
||||
except:
|
||||
raise ProtocolException("Expected JSON response")
|
||||
|
||||
self.check_error(resp)
|
||||
|
||||
if "text" in object:
|
||||
return object["text"]
|
||||
|
||||
if "object" in object:
|
||||
try:
|
||||
return json.loads(object["object"])
|
||||
except Exception as e:
|
||||
raise ProtocolException(
|
||||
"Returned object not well-formed JSON"
|
||||
)
|
||||
|
||||
raise ProtocolException("Response not formatted correctly")
|
||||
|
||||
def triples_query(self, s=None, p=None, o=None, limit=10000):
|
||||
|
||||
# The input consists of system and prompt strings
|
||||
input = {
|
||||
"limit": limit
|
||||
}
|
||||
|
||||
if s: input["s"] = s
|
||||
if p: input["p"] = p
|
||||
if o: input["o"] = o
|
||||
|
||||
url = f"{self.url}triples-query"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
try:
|
||||
# Parse the response as JSON
|
||||
object = resp.json()
|
||||
except:
|
||||
raise ProtocolException("Expected JSON response")
|
||||
|
||||
self.check_error(resp)
|
||||
|
||||
if "response" not in object:
|
||||
raise ProtocolException("Response not formatted correctly")
|
||||
|
||||
def to_value(x):
|
||||
if x["e"]: return Uri(x["v"])
|
||||
return Literal(x["v"])
|
||||
|
||||
return [
|
||||
Triple(
|
||||
s=to_value(t["s"]),
|
||||
p=to_value(t["p"]),
|
||||
o=to_value(t["o"])
|
||||
)
|
||||
for t in object["response"]
|
||||
]
|
||||
|
||||
return object["response"]
|
||||
|
||||
def load_document(self, document, id=None, metadata=None):
|
||||
|
||||
if id is None:
|
||||
|
||||
if metadata is not None:
|
||||
|
||||
# Situation makes no sense. What can the metadata possibly
|
||||
# mean if the caller doesn't know the document ID.
|
||||
# Metadata should relate to the document by ID
|
||||
raise RuntimeError("Can't specify metadata without id")
|
||||
|
||||
id = hash(document)
|
||||
|
||||
triples = []
|
||||
|
||||
def emit(t):
|
||||
triples.append(t)
|
||||
|
||||
if metadata:
|
||||
metadata.emit(
|
||||
lambda t: triples.append({
|
||||
"s": t.s.value,
|
||||
"p": t.p.value,
|
||||
"o": t.o.value
|
||||
})
|
||||
)
|
||||
|
||||
input = {
|
||||
"id": id,
|
||||
"metadata": triples,
|
||||
"data": base64.b64encode(document).decode("utf-8"),
|
||||
}
|
||||
|
||||
url = f"{self.url}load/document"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
def load_text(self, text, id=None, metadata=None, charset="utf-8"):
|
||||
|
||||
if id is None:
|
||||
|
||||
if metadata is not None:
|
||||
|
||||
# Situation makes no sense. What can the metadata possibly
|
||||
# mean if the caller doesn't know the document ID.
|
||||
# Metadata should relate to the document by ID
|
||||
raise RuntimeError("Can't specify metadata without id")
|
||||
|
||||
id = hash(text)
|
||||
|
||||
triples = []
|
||||
|
||||
if metadata:
|
||||
metadata.emit(
|
||||
lambda t: triples.append({
|
||||
"s": t.s.value,
|
||||
"p": t.p.value,
|
||||
"o": t.o.value
|
||||
})
|
||||
)
|
||||
|
||||
input = {
|
||||
"id": id,
|
||||
"metadata": triples,
|
||||
"charset": charset,
|
||||
"text": base64.b64encode(text).decode("utf-8"),
|
||||
}
|
||||
|
||||
url = f"{self.url}load/text"
|
||||
|
||||
# Invoke the API, input is passed as JSON
|
||||
resp = requests.post(url, json=input)
|
||||
|
||||
# Should be a 200 status code
|
||||
if resp.status_code != 200:
|
||||
raise ProtocolException(f"Status code {resp.status_code}")
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue