mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-24 04:31:02 +02:00
Logging strategy updates
This commit is contained in:
Cyber MacGeddon
parent
905e17ac36
commit
4b50abbf5a
6 changed files with 77 additions and 53 deletions
|
|
@ -5,6 +5,10 @@ from .. exceptions import RequestError
|
|||
from minio import Minio
|
||||
import time
|
||||
import io
|
||||
import logging
|
||||
|
||||
# Module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class BlobStore:
|
||||
|
||||
|
|
@ -23,7 +27,7 @@ class BlobStore:
|
|||
|
||||
self.bucket_name = bucket_name
|
||||
|
||||
print("Connected to minio", flush=True)
|
||||
logger.info("Connected to MinIO")
|
||||
|
||||
self.ensure_bucket()
|
||||
|
||||
|
|
@ -33,9 +37,9 @@ class BlobStore:
|
|||
found = self.minio.bucket_exists(self.bucket_name)
|
||||
if not found:
|
||||
self.minio.make_bucket(self.bucket_name)
|
||||
print("Created bucket", self.bucket_name, flush=True)
|
||||
logger.info(f"Created bucket {self.bucket_name}")
|
||||
else:
|
||||
print("Bucket", self.bucket_name, "already exists", flush=True)
|
||||
logger.debug(f"Bucket {self.bucket_name} already exists")
|
||||
|
||||
async def add(self, object_id, blob, kind):
|
||||
|
||||
|
|
@ -48,7 +52,7 @@ class BlobStore:
|
|||
content_type = kind,
|
||||
)
|
||||
|
||||
print("Add blob complete", flush=True)
|
||||
logger.debug("Add blob complete")
|
||||
|
||||
async def remove(self, object_id):
|
||||
|
||||
|
|
@ -58,7 +62,7 @@ class BlobStore:
|
|||
object_name = "doc/" + str(object_id),
|
||||
)
|
||||
|
||||
print("Remove blob complete", flush=True)
|
||||
logger.debug("Remove blob complete")
|
||||
|
||||
|
||||
async def get(self, object_id):
|
||||
|
|
|
|||
|
|
@ -5,9 +5,13 @@ from .. exceptions import RequestError
|
|||
from .. tables.library import LibraryTableStore
|
||||
from . blob_store import BlobStore
|
||||
import base64
|
||||
import logging
|
||||
|
||||
import uuid
|
||||
|
||||
# Module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class Librarian:
|
||||
|
||||
def __init__(
|
||||
|
|
@ -70,7 +74,7 @@ class Librarian:
|
|||
|
||||
async def remove_document(self, request):
|
||||
|
||||
print("Removing doc...")
|
||||
logger.debug("Removing document...")
|
||||
|
||||
if not await self.table_store.document_exists(
|
||||
request.user,
|
||||
|
|
@ -92,7 +96,7 @@ class Librarian:
|
|||
request.document_id
|
||||
)
|
||||
|
||||
print("Remove complete", flush=True)
|
||||
logger.debug("Remove complete")
|
||||
|
||||
return LibrarianResponse(
|
||||
error = None,
|
||||
|
|
@ -104,7 +108,7 @@ class Librarian:
|
|||
|
||||
async def update_document(self, request):
|
||||
|
||||
print("Updating doc...")
|
||||
logger.debug("Updating document...")
|
||||
|
||||
# You can't update the document ID, user or kind.
|
||||
|
||||
|
|
@ -116,7 +120,7 @@ class Librarian:
|
|||
|
||||
await self.table_store.update_document(request.document_metadata)
|
||||
|
||||
print("Update complete", flush=True)
|
||||
logger.debug("Update complete")
|
||||
|
||||
return LibrarianResponse(
|
||||
error = None,
|
||||
|
|
@ -128,14 +132,14 @@ class Librarian:
|
|||
|
||||
async def get_document_metadata(self, request):
|
||||
|
||||
print("Get doc...")
|
||||
logger.debug("Getting document metadata...")
|
||||
|
||||
doc = await self.table_store.get_document(
|
||||
request.user,
|
||||
request.document_id
|
||||
)
|
||||
|
||||
print("Get complete", flush=True)
|
||||
logger.debug("Get complete")
|
||||
|
||||
return LibrarianResponse(
|
||||
error = None,
|
||||
|
|
@ -147,7 +151,7 @@ class Librarian:
|
|||
|
||||
async def get_document_content(self, request):
|
||||
|
||||
print("Get doc content...")
|
||||
logger.debug("Getting document content...")
|
||||
|
||||
object_id = await self.table_store.get_document_object_id(
|
||||
request.user,
|
||||
|
|
@ -158,7 +162,7 @@ class Librarian:
|
|||
object_id
|
||||
)
|
||||
|
||||
print("Get complete", flush=True)
|
||||
logger.debug("Get complete")
|
||||
|
||||
return LibrarianResponse(
|
||||
error = None,
|
||||
|
|
@ -170,7 +174,7 @@ class Librarian:
|
|||
|
||||
async def add_processing(self, request):
|
||||
|
||||
print("Add processing")
|
||||
logger.debug("Adding processing metadata...")
|
||||
|
||||
if await self.table_store.processing_exists(
|
||||
request.processing_metadata.user,
|
||||
|
|
@ -192,13 +196,13 @@ class Librarian:
|
|||
object_id
|
||||
)
|
||||
|
||||
print("Got content")
|
||||
logger.debug("Retrieved content")
|
||||
|
||||
print("Add processing...")
|
||||
logger.debug("Adding processing to table...")
|
||||
|
||||
await self.table_store.add_processing(request.processing_metadata)
|
||||
|
||||
print("Invoke document processing...")
|
||||
logger.debug("Invoking document processing...")
|
||||
|
||||
await self.load_document(
|
||||
document = doc,
|
||||
|
|
@ -218,7 +222,7 @@ class Librarian:
|
|||
|
||||
async def remove_processing(self, request):
|
||||
|
||||
print("Removing processing...")
|
||||
logger.debug("Removing processing metadata...")
|
||||
|
||||
if not await self.table_store.processing_exists(
|
||||
request.user,
|
||||
|
|
@ -232,7 +236,7 @@ class Librarian:
|
|||
request.processing_id
|
||||
)
|
||||
|
||||
print("Remove complete", flush=True)
|
||||
logger.debug("Remove complete")
|
||||
|
||||
return LibrarianResponse(
|
||||
error = None,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,9 @@
|
|||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
# Module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LABEL="http://www.w3.org/2000/01/rdf-schema#label"
|
||||
|
||||
|
|
@ -18,12 +22,12 @@ class Query:
|
|||
async def get_vector(self, query):
|
||||
|
||||
if self.verbose:
|
||||
print("Compute embeddings...", flush=True)
|
||||
logger.debug("Computing embeddings...")
|
||||
|
||||
qembeds = await self.rag.embeddings_client.embed(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Done.", flush=True)
|
||||
logger.debug("Embeddings computed")
|
||||
|
||||
return qembeds
|
||||
|
||||
|
|
@ -32,7 +36,7 @@ class Query:
|
|||
vectors = await self.get_vector(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Get docs...", flush=True)
|
||||
logger.debug("Getting documents...")
|
||||
|
||||
docs = await self.rag.doc_embeddings_client.query(
|
||||
vectors, limit=self.doc_limit,
|
||||
|
|
@ -40,9 +44,9 @@ class Query:
|
|||
)
|
||||
|
||||
if self.verbose:
|
||||
print("Docs:", flush=True)
|
||||
logger.debug("Documents:")
|
||||
for doc in docs:
|
||||
print(doc, flush=True)
|
||||
logger.debug(f" {doc}")
|
||||
|
||||
return docs
|
||||
|
||||
|
|
@ -60,7 +64,7 @@ class DocumentRag:
|
|||
self.doc_embeddings_client = doc_embeddings_client
|
||||
|
||||
if self.verbose:
|
||||
print("Initialised", flush=True)
|
||||
logger.debug("DocumentRag initialized")
|
||||
|
||||
async def query(
|
||||
self, query, user="trustgraph", collection="default",
|
||||
|
|
@ -68,7 +72,7 @@ class DocumentRag:
|
|||
):
|
||||
|
||||
if self.verbose:
|
||||
print("Construct prompt...", flush=True)
|
||||
logger.debug("Constructing prompt...")
|
||||
|
||||
q = Query(
|
||||
rag=self, user=user, collection=collection, verbose=self.verbose,
|
||||
|
|
@ -78,9 +82,9 @@ class DocumentRag:
|
|||
docs = await q.get_docs(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Invoke LLM...", flush=True)
|
||||
print(docs)
|
||||
print(query)
|
||||
logger.debug("Invoking LLM...")
|
||||
logger.debug(f"Documents: {docs}")
|
||||
logger.debug(f"Query: {query}")
|
||||
|
||||
resp = await self.prompt_client.document_prompt(
|
||||
query = query,
|
||||
|
|
@ -88,7 +92,7 @@ class DocumentRag:
|
|||
)
|
||||
|
||||
if self.verbose:
|
||||
print("Done", flush=True)
|
||||
logger.debug("Query processing complete")
|
||||
|
||||
return resp
|
||||
|
||||
|
|
|
|||
|
|
@ -4,12 +4,16 @@ Simple RAG service, performs query using document RAG an LLM.
|
|||
Input is query, output is response.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from ... schema import DocumentRagQuery, DocumentRagResponse, Error
|
||||
from . document_rag import DocumentRag
|
||||
from ... base import FlowProcessor, ConsumerSpec, ProducerSpec
|
||||
from ... base import PromptClientSpec, EmbeddingsClientSpec
|
||||
from ... base import DocumentEmbeddingsClientSpec
|
||||
|
||||
# Module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
default_ident = "document-rag"
|
||||
|
||||
class Processor(FlowProcessor):
|
||||
|
|
@ -81,7 +85,7 @@ class Processor(FlowProcessor):
|
|||
# Sender-produced ID
|
||||
id = msg.properties()["id"]
|
||||
|
||||
print(f"Handling input {id}...", flush=True)
|
||||
logger.info(f"Handling input {id}...")
|
||||
|
||||
if v.doc_limit:
|
||||
doc_limit = v.doc_limit
|
||||
|
|
@ -98,13 +102,13 @@ class Processor(FlowProcessor):
|
|||
properties = {"id": id}
|
||||
)
|
||||
|
||||
print("Done.", flush=True)
|
||||
logger.info("Request processing complete")
|
||||
|
||||
except Exception as e:
|
||||
|
||||
print(f"Exception: {e}")
|
||||
logger.error(f"Document RAG service exception: {e}", exc_info=True)
|
||||
|
||||
print("Send error response...", flush=True)
|
||||
logger.debug("Sending error response...")
|
||||
|
||||
await flow("response").send(
|
||||
DocumentRagResponse(
|
||||
|
|
|
|||
|
|
@ -1,5 +1,9 @@
|
|||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
# Module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LABEL="http://www.w3.org/2000/01/rdf-schema#label"
|
||||
|
||||
|
|
@ -22,12 +26,12 @@ class Query:
|
|||
async def get_vector(self, query):
|
||||
|
||||
if self.verbose:
|
||||
print("Compute embeddings...", flush=True)
|
||||
logger.debug("Computing embeddings...")
|
||||
|
||||
qembeds = await self.rag.embeddings_client.embed(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Done.", flush=True)
|
||||
logger.debug("Done.")
|
||||
|
||||
return qembeds
|
||||
|
||||
|
|
@ -36,7 +40,7 @@ class Query:
|
|||
vectors = await self.get_vector(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Get entities...", flush=True)
|
||||
logger.debug("Getting entities...")
|
||||
|
||||
entities = await self.rag.graph_embeddings_client.query(
|
||||
vectors=vectors, limit=self.entity_limit,
|
||||
|
|
@ -49,9 +53,9 @@ class Query:
|
|||
]
|
||||
|
||||
if self.verbose:
|
||||
print("Entities:", flush=True)
|
||||
logger.debug("Entities:")
|
||||
for ent in entities:
|
||||
print(" ", ent, flush=True)
|
||||
logger.debug(f" {ent}")
|
||||
|
||||
return entities
|
||||
|
||||
|
|
@ -126,7 +130,7 @@ class Query:
|
|||
entities = await self.get_entities(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Get subgraph...", flush=True)
|
||||
logger.debug("Getting subgraph...")
|
||||
|
||||
subgraph = set()
|
||||
|
||||
|
|
@ -157,12 +161,12 @@ class Query:
|
|||
sg2 = sg2[0:self.max_subgraph_size]
|
||||
|
||||
if self.verbose:
|
||||
print("Subgraph:", flush=True)
|
||||
logger.debug("Subgraph:")
|
||||
for edge in sg2:
|
||||
print(" ", str(edge), flush=True)
|
||||
logger.debug(f" {str(edge)}")
|
||||
|
||||
if self.verbose:
|
||||
print("Done.", flush=True)
|
||||
logger.debug("Done.")
|
||||
|
||||
return sg2
|
||||
|
||||
|
|
@ -183,7 +187,7 @@ class GraphRag:
|
|||
self.label_cache = {}
|
||||
|
||||
if self.verbose:
|
||||
print("Initialised", flush=True)
|
||||
logger.debug("GraphRag initialized")
|
||||
|
||||
async def query(
|
||||
self, query, user = "trustgraph", collection = "default",
|
||||
|
|
@ -192,7 +196,7 @@ class GraphRag:
|
|||
):
|
||||
|
||||
if self.verbose:
|
||||
print("Construct prompt...", flush=True)
|
||||
logger.debug("Constructing prompt...")
|
||||
|
||||
q = Query(
|
||||
rag = self, user = user, collection = collection,
|
||||
|
|
@ -205,14 +209,14 @@ class GraphRag:
|
|||
kg = await q.get_labelgraph(query)
|
||||
|
||||
if self.verbose:
|
||||
print("Invoke LLM...", flush=True)
|
||||
print(kg)
|
||||
print(query)
|
||||
logger.debug("Invoking LLM...")
|
||||
logger.debug(f"Knowledge graph: {kg}")
|
||||
logger.debug(f"Query: {query}")
|
||||
|
||||
resp = await self.prompt_client.kg_prompt(query, kg)
|
||||
|
||||
if self.verbose:
|
||||
print("Done", flush=True)
|
||||
logger.debug("Query processing complete")
|
||||
|
||||
return resp
|
||||
|
||||
|
|
|
|||
|
|
@ -4,12 +4,16 @@ Simple RAG service, performs query using graph RAG an LLM.
|
|||
Input is query, output is response.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from ... schema import GraphRagQuery, GraphRagResponse, Error
|
||||
from . graph_rag import GraphRag
|
||||
from ... base import FlowProcessor, ConsumerSpec, ProducerSpec
|
||||
from ... base import PromptClientSpec, EmbeddingsClientSpec
|
||||
from ... base import GraphEmbeddingsClientSpec, TriplesClientSpec
|
||||
|
||||
# Module logger
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
default_ident = "graph-rag"
|
||||
default_concurrency = 1
|
||||
|
||||
|
|
@ -102,7 +106,7 @@ class Processor(FlowProcessor):
|
|||
# Sender-produced ID
|
||||
id = msg.properties()["id"]
|
||||
|
||||
print(f"Handling input {id}...", flush=True)
|
||||
logger.info(f"Handling input {id}...")
|
||||
|
||||
if v.entity_limit:
|
||||
entity_limit = v.entity_limit
|
||||
|
|
@ -139,13 +143,13 @@ class Processor(FlowProcessor):
|
|||
properties = {"id": id}
|
||||
)
|
||||
|
||||
print("Done.", flush=True)
|
||||
logger.info("Request processing complete")
|
||||
|
||||
except Exception as e:
|
||||
|
||||
print(f"Exception: {e}")
|
||||
logger.error(f"Graph RAG service exception: {e}", exc_info=True)
|
||||
|
||||
print("Send error response...", flush=True)
|
||||
logger.debug("Sending error response...")
|
||||
|
||||
await flow("response").send(
|
||||
GraphRagResponse(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue