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

Token chunker

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-15 23:13:21 +01:00
parent cfdeadfb34
commit 510cfb1103
2 changed files with 15 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -21,7 +21,7 @@ class Processor(InputOutputProcessor):
def __init__(self, **params): def __init__(self, **params):
id = params.get("id") id = params.get("id")
subscriber = params.get("subscriber") subscriber = params.get("subscriber", default_subscriber)
chunk_size = params.get("chunk_size", 2000) chunk_size = params.get("chunk_size", 2000)
chunk_overlap = params.get("chunk_overlap", 100) chunk_overlap = params.get("chunk_overlap", 100)

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

@ -10,28 +10,23 @@ from prometheus_client import Histogram
from ... schema import TextDocument, Chunk, Metadata from ... schema import TextDocument, Chunk, Metadata
from ... schema import text_ingest_queue, chunk_ingest_queue from ... schema import text_ingest_queue, chunk_ingest_queue
from ... log_level import LogLevel from ... log_level import LogLevel
from ... base import ConsumerProducer from ... base import InputOutputProcessor
module = "chunker" module = "chunker"
default_input_queue = text_ingest_queue
default_output_queue = chunk_ingest_queue
default_subscriber = module default_subscriber = module
class Processor(ConsumerProducer): class Processor(InputOutputProcessor):
def __init__(self, **params): def __init__(self, **params):
input_queue = params.get("input_queue", default_input_queue) id = params.get("id")
output_queue = params.get("output_queue", default_output_queue)
subscriber = params.get("subscriber", default_subscriber) subscriber = params.get("subscriber", default_subscriber)
chunk_size = params.get("chunk_size", 250) chunk_size = params.get("chunk_size", 250)
chunk_overlap = params.get("chunk_overlap", 15) chunk_overlap = params.get("chunk_overlap", 15)
super(Processor, self).__init__( super(Processor, self).__init__(
**params | { **params | {
"input_queue": input_queue,
"output_queue": output_queue,
"subscriber": subscriber, "subscriber": subscriber,
"input_schema": TextDocument, "input_schema": TextDocument,
"output_schema": Chunk, "output_schema": Chunk,
@ -41,6 +36,7 @@ class Processor(ConsumerProducer):
if not hasattr(__class__, "chunk_metric"): if not hasattr(__class__, "chunk_metric"):
__class__.chunk_metric = Histogram( __class__.chunk_metric = Histogram(
'chunk_size', 'Chunk size', 'chunk_size', 'Chunk size',
["id", "flow"],
buckets=[100, 160, 250, 400, 650, 1000, 1600, buckets=[100, 160, 250, 400, 650, 1000, 1600,
2500, 4000, 6400, 10000, 16000] 2500, 4000, 6400, 10000, 16000]
) )
@ -51,7 +47,9 @@ class Processor(ConsumerProducer):
chunk_overlap=chunk_overlap, chunk_overlap=chunk_overlap,
) )
async def handle(self, msg): print("Chunker initialised", flush=True)
async def on_message(self, msg, consumer):
v = msg.value() v = msg.value()
print(f"Chunking {v.metadata.id}...", flush=True) print(f"Chunking {v.metadata.id}...", flush=True)
@ -62,24 +60,25 @@ class Processor(ConsumerProducer):
for ix, chunk in enumerate(texts): for ix, chunk in enumerate(texts):
print("Chunk", len(chunk.page_content), flush=True)
r = Chunk( r = Chunk(
metadata=v.metadata, metadata=v.metadata,
chunk=chunk.page_content.encode("utf-8"), chunk=chunk.page_content.encode("utf-8"),
) )
__class__.chunk_metric.observe(len(chunk.page_content)) __class__.chunk_metric.labels(
id=consumer.id, flow=consumer.flow
).observe(len(chunk.page_content))
await self.send(r) await consumer.q.output.send(r)
print("Done.", flush=True) print("Done.", flush=True)
@staticmethod @staticmethod
def add_args(parser): def add_args(parser):
ConsumerProducer.add_args( InputOutputProcessor.add_args(parser, default_subscriber)
parser, default_input_queue, default_subscriber,
default_output_queue,
)
parser.add_argument( parser.add_argument(
'-z', '--chunk-size', '-z', '--chunk-size',