apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-24 04:31:02 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Logging strategy updates

Browse source
This commit is contained in:
Cyber MacGeddon 2025-07-30 22:09:18 +01:00
parent 82269b0614
commit 84598c50f6
10 changed files with 95 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

12
trustgraph-flow/trustgraph/chunking/recursive/chunker.py
View file

@ -4,12 +4,16 @@ Simple decoder, accepts text documents on input, outputs chunks from the
as text as separate output objects.
"""
import logging
from langchain_text_splitters import RecursiveCharacterTextSplitter
from prometheus_client import Histogram
from ... schema import TextDocument, Chunk
from ... base import FlowProcessor, ConsumerSpec, ProducerSpec
# Module logger
logger = logging.getLogger(__name__)
default_ident = "chunker"
class Processor(FlowProcessor):
@ -54,12 +58,12 @@ class Processor(FlowProcessor):
)
)
print("Chunker initialised", flush=True)
logger.info("Recursive chunker initialized")
async def on_message(self, msg, consumer, flow):
v = msg.value()
print(f"Chunking {v.metadata.id}...", flush=True)
logger.info(f"Chunking document {v.metadata.id}...")
texts = self.text_splitter.create_documents(
[v.text.decode("utf-8")]
@ -67,7 +71,7 @@ class Processor(FlowProcessor):
for ix, chunk in enumerate(texts):
print("Chunk", len(chunk.page_content), flush=True)
logger.debug(f"Created chunk of size {len(chunk.page_content)}")
r = Chunk(
metadata=v.metadata,
@ -80,7 +84,7 @@ class Processor(FlowProcessor):
await flow("output").send(r)
print("Done.", flush=True)
logger.debug("Document chunking complete")
@staticmethod
def add_args(parser):

12
trustgraph-flow/trustgraph/chunking/token/chunker.py
View file

@ -4,12 +4,16 @@ Simple decoder, accepts text documents on input, outputs chunks from the
as text as separate output objects.
"""
import logging
from langchain_text_splitters import TokenTextSplitter
from prometheus_client import Histogram
from ... schema import TextDocument, Chunk
from ... base import FlowProcessor
# Module logger
logger = logging.getLogger(__name__)
default_ident = "chunker"
class Processor(FlowProcessor):
@ -53,12 +57,12 @@ class Processor(FlowProcessor):
)
)
print("Chunker initialised", flush=True)
logger.info("Token chunker initialized")
async def on_message(self, msg, consumer, flow):
v = msg.value()
print(f"Chunking {v.metadata.id}...", flush=True)
logger.info(f"Chunking document {v.metadata.id}...")
texts = self.text_splitter.create_documents(
[v.text.decode("utf-8")]
@ -66,7 +70,7 @@ class Processor(FlowProcessor):
for ix, chunk in enumerate(texts):
print("Chunk", len(chunk.page_content), flush=True)
logger.debug(f"Created chunk of size {len(chunk.page_content)}")
r = Chunk(
metadata=v.metadata,
@ -79,7 +83,7 @@ class Processor(FlowProcessor):
await flow("output").send(r)
print("Done.", flush=True)
logger.debug("Document chunking complete")
@staticmethod
def add_args(parser):

7
trustgraph-flow/trustgraph/config/service/config.py
View file

@ -1,9 +1,14 @@
import logging
from trustgraph.schema import ConfigResponse
from trustgraph.schema import ConfigValue, Error
from ... tables.config import ConfigTableStore
# Module logger
logger = logging.getLogger(__name__)
class ConfigurationClass:
async def keys(self):
@ -228,7 +233,7 @@ class Configuration:
async def handle(self, msg):
print("Handle message ", msg.operation)
logger.debug(f"Handling config message: {msg.operation}")
if msg.operation == "get":

8
trustgraph-flow/trustgraph/config/service/flow.py
View file

@ -1,6 +1,10 @@
from trustgraph.schema import FlowResponse, Error
import json
import logging
# Module logger
logger = logging.getLogger(__name__)
class FlowConfig:
def __init__(self, config):
@ -41,7 +45,7 @@ class FlowConfig:
async def handle_delete_class(self, msg):
print(msg)
logger.debug(f"Flow config message: {msg}")
await self.config.get("flow-classes").delete(msg.class_name)
@ -218,7 +222,7 @@ class FlowConfig:
async def handle(self, msg):
print("Handle message ", msg.operation)
logger.debug(f"Handling flow message: {msg.operation}")
if msg.operation == "list-classes":
resp = await self.handle_list_classes(msg)

13
trustgraph-flow/trustgraph/config/service/service.py
View file

@ -3,6 +3,8 @@
Config service. Manages system global configuration state
"""
import logging
from trustgraph.schema import Error
from trustgraph.schema import ConfigRequest, ConfigResponse, ConfigPush
@ -20,6 +22,9 @@ from . flow import FlowConfig
from ... base import ProcessorMetrics, ConsumerMetrics, ProducerMetrics
from ... base import Consumer, Producer
# Module logger
logger = logging.getLogger(__name__)
# FIXME: How to ensure this doesn't conflict with other usage?
keyspace = "config"
@ -146,7 +151,7 @@ class Processor(AsyncProcessor):
self.flow = FlowConfig(self.config)
print("Service initialised.")
logger.info("Config service initialized")
async def start(self):
@ -172,7 +177,7 @@ class Processor(AsyncProcessor):
# Race condition, should make sure version & config sync
print("Pushed version ", await self.config.get_version())
logger.info(f"Pushed configuration version {await self.config.get_version()}")
async def on_config_request(self, msg, consumer, flow):
@ -183,7 +188,7 @@ class Processor(AsyncProcessor):
# Sender-produced ID
id = msg.properties()["id"]
print(f"Handling {id}...", flush=True)
logger.info(f"Handling config request {id}...")
resp = await self.config.handle(v)
@ -214,7 +219,7 @@ class Processor(AsyncProcessor):
# Sender-produced ID
id = msg.properties()["id"]
print(f"Handling {id}...", flush=True)
logger.info(f"Handling flow request {id}...")
resp = await self.flow.handle(v)

38
trustgraph-flow/trustgraph/cores/knowledge.py
View file

@ -8,6 +8,10 @@ from .. base import Publisher
import base64
import asyncio
import uuid
import logging
# Module logger
logger = logging.getLogger(__name__)
class KnowledgeManager:
@ -26,7 +30,7 @@ class KnowledgeManager:
async def delete_kg_core(self, request, respond):
print("Deleting core...", flush=True)
logger.info("Deleting knowledge core...")
await self.table_store.delete_kg_core(
request.user, request.id
@ -44,7 +48,7 @@ class KnowledgeManager:
async def get_kg_core(self, request, respond):
print("Get core...", flush=True)
logger.info("Getting knowledge core...")
async def publish_triples(t):
await respond(
@ -82,7 +86,7 @@ class KnowledgeManager:
publish_ge,
)
print("Get complete", flush=True)
logger.debug("Knowledge core retrieval complete")
await respond(
KnowledgeResponse(
@ -158,13 +162,13 @@ class KnowledgeManager:
async def core_loader(self):
print("Running...", flush=True)
logger.info("Knowledge background processor running...")
while True:
print("Wait for next load...", flush=True)
logger.debug("Waiting for next load...")
request, respond = await self.loader_queue.get()
print("Loading...", request.id, flush=True)
logger.info(f"Loading knowledge: {request.id}")
try:
@ -204,7 +208,7 @@ class KnowledgeManager:
except Exception as e:
print("Exception:", e, flush=True)
logger.error(f"Knowledge exception: {e}", exc_info=True)
await respond(
KnowledgeResponse(
error = Error(
@ -219,15 +223,15 @@ class KnowledgeManager:
)
print("Going to start loading...", flush=True)
logger.debug("Starting knowledge loading process...")
try:
t_pub = None
ge_pub = None
print(t_q, flush=True)
print(ge_q, flush=True)
logger.debug(f"Triples queue: {t_q}")
logger.debug(f"Graph embeddings queue: {ge_q}")
t_pub = Publisher(
self.flow_config.pulsar_client, t_q,
@ -238,7 +242,7 @@ class KnowledgeManager:
schema=GraphEmbeddings
)
print("Start publishers...", flush=True)
logger.debug("Starting publishers...")
await t_pub.start()
await ge_pub.start()
@ -246,7 +250,7 @@ class KnowledgeManager:
async def publish_triples(t):
await t_pub.send(None, t)
print("Publish triples...", flush=True)
logger.debug("Publishing triples...")
# Remove doc table row
await self.table_store.get_triples(
@ -258,7 +262,7 @@ class KnowledgeManager:
async def publish_ge(g):
await ge_pub.send(None, g)
print("Publish GEs...", flush=True)
logger.debug("Publishing graph embeddings...")
# Remove doc table row
await self.table_store.get_graph_embeddings(
@ -267,19 +271,19 @@ class KnowledgeManager:
publish_ge,
)
print("Completed that.", flush=True)
logger.debug("Knowledge loading completed")
except Exception as e:
print("Exception:", e, flush=True)
logger.error(f"Knowledge exception: {e}", exc_info=True)
finally:
print("Stopping publishers...", flush=True)
logger.debug("Stopping publishers...")
if t_pub: await t_pub.stop()
if ge_pub: await ge_pub.stop()
print("Done", flush=True)
logger.debug("Knowledge processing done")
continue

16
trustgraph-flow/trustgraph/cores/service.py
View file

@ -7,6 +7,7 @@ from functools import partial
import asyncio
import base64
import json
import logging
from .. base import AsyncProcessor, Consumer, Producer, Publisher, Subscriber
from .. base import ConsumerMetrics, ProducerMetrics
@ -21,6 +22,9 @@ from .. exceptions import RequestError
from . knowledge import KnowledgeManager
# Module logger
logger = logging.getLogger(__name__)
default_ident = "knowledge"
default_knowledge_request_queue = knowledge_request_queue
@ -96,7 +100,7 @@ class Processor(AsyncProcessor):
self.flows = {}
print("Initialised.", flush=True)
logger.info("Knowledge service initialized")
async def start(self):
@ -106,7 +110,7 @@ class Processor(AsyncProcessor):
async def on_knowledge_config(self, config, version):
print("config version", version)
logger.info(f"Configuration version: {version}")
if "flows" in config:
@ -115,14 +119,14 @@ class Processor(AsyncProcessor):
for k, v in config["flows"].items()
}
print(self.flows)
logger.debug(f"Flows: {self.flows}")
async def process_request(self, v, id):
if v.operation is None:
raise RequestError("Null operation")
print("request", v.operation)
logger.debug(f"Knowledge request: {v.operation}")
impls = {
"list-kg-cores": self.knowledge.list_kg_cores,
@ -150,7 +154,7 @@ class Processor(AsyncProcessor):
id = msg.properties()["id"]
print(f"Handling input {id}...", flush=True)
logger.info(f"Handling knowledge input {id}...")
try:
@ -187,7 +191,7 @@ class Processor(AsyncProcessor):
return
print("Done.", flush=True)
logger.debug("Knowledge input processing complete")
@staticmethod
def add_args(parser):

14
trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py
View file

@ -6,11 +6,15 @@ PDF document as text as separate output objects.
import tempfile
import base64
import logging
from langchain_community.document_loaders import PyPDFLoader
from ... schema import Document, TextDocument, Metadata
from ... base import FlowProcessor, ConsumerSpec, ProducerSpec
# Module logger
logger = logging.getLogger(__name__)
default_ident = "pdf-decoder"
class Processor(FlowProcessor):
@ -40,15 +44,15 @@ class Processor(FlowProcessor):
)
)
print("PDF inited", flush=True)
logger.info("PDF decoder initialized")
async def on_message(self, msg, consumer, flow):
print("PDF message received", flush=True)
logger.debug("PDF message received")
v = msg.value()
print(f"Decoding {v.metadata.id}...", flush=True)
logger.info(f"Decoding PDF {v.metadata.id}...")
with tempfile.NamedTemporaryFile(delete_on_close=False) as fp:
@ -62,7 +66,7 @@ class Processor(FlowProcessor):
for ix, page in enumerate(pages):
print("page", ix, flush=True)
logger.debug(f"Processing page {ix}")
r = TextDocument(
metadata=v.metadata,
@ -71,7 +75,7 @@ class Processor(FlowProcessor):
await flow("output").send(r)
print("Done.", flush=True)
logger.debug("PDF decoding complete")
@staticmethod
def add_args(parser):

7
trustgraph-flow/trustgraph/embeddings/fastembed/processor.py
View file

@ -4,10 +4,15 @@ Embeddings service, applies an embeddings model using fastembed
Input is text, output is embeddings vector.
"""
import logging
from ... base import EmbeddingsService
from fastembed import TextEmbedding
# Module logger
logger = logging.getLogger(__name__)
default_ident = "embeddings"
default_model="sentence-transformers/all-MiniLM-L6-v2"
@ -22,7 +27,7 @@ class Processor(EmbeddingsService):
**params | { "model": model }
)
print("Get model...", flush=True)
logger.info("Loading FastEmbed model...")
self.embeddings = TextEmbedding(model_name = model)
async def on_embeddings(self, text):

20
trustgraph-vertexai/trustgraph/model/text_completion/vertexai/llm.py
View file

@ -19,6 +19,7 @@ Google Cloud. Input is prompt, output is response.
from google.oauth2 import service_account
import google
import vertexai
import logging
# Why is preview here?
from vertexai.generative_models import (
@ -29,6 +30,9 @@ from vertexai.generative_models import (
from .... exceptions import TooManyRequests
from .... base import LlmService, LlmResult
# Module logger
logger = logging.getLogger(__name__)
default_ident = "text-completion"
default_model = 'gemini-2.0-flash-001'
@ -91,7 +95,7 @@ class Processor(LlmService):
),
]
print("Initialise VertexAI...", flush=True)
logger.info("Initializing VertexAI...")
if private_key:
credentials = (
@ -113,11 +117,11 @@ class Processor(LlmService):
location=region
)
print(f"Initialise model {model}", flush=True)
logger.info(f"Initializing model {model}")
self.llm = GenerativeModel(model)
self.model = model
print("Initialisation complete", flush=True)
logger.info("VertexAI initialization complete")
async def generate_content(self, system, prompt):
@ -137,16 +141,16 @@ class Processor(LlmService):
model = self.model
)
print(f"Input Tokens: {resp.in_token}", flush=True)
print(f"Output Tokens: {resp.out_token}", flush=True)
logger.info(f"Input Tokens: {resp.in_token}")
logger.info(f"Output Tokens: {resp.out_token}")
print("Send response...", flush=True)
logger.debug("Send response...")
return resp
except google.api_core.exceptions.ResourceExhausted as e:
print("Hit rate limit:", e, flush=True)
logger.warning(f"Hit rate limit: {e}")
# Leave rate limit retries to the base handler
raise TooManyRequests()
@ -154,7 +158,7 @@ class Processor(LlmService):
except Exception as e:
# Apart from rate limits, treat all exceptions as unrecoverable
print(f"Exception: {e}")
logger.error(f"VertexAI LLM exception: {e}", exc_info=True)
raise e
@staticmethod