oguzhankir/omnichunk

from omnichunk import chunk

code = """
import os

def hello(name: str) -> str:
    return f"hello {name}"
"""

chunks = chunk("example.py", code, max_chunk_size=128, size_unit="chars")

for c in chunks:
    print(c.index, c.byte_range, c.context.breadcrumb)
    print(c.contextualized_text)

from omnichunk import Chunker

chunker = Chunker(
    max_chunk_size=1024,
    min_chunk_size=80,
    tokenizer="cl100k_base",
    context_mode="full",
    overlap=0.1,
    overlap_lines=1,
)

chunks = chunker.chunk("api.py", source_code)

for c in chunker.stream("large.py", large_source):
    consume(c)

import asyncio
from omnichunk import Chunker

chunker = Chunker(max_chunk_size=1024, size_unit="tokens")

# Single file async
chunks = asyncio.run(chunker.achunk("api.py", source_code))

# Async streaming
async def process():
    async for chunk in chunker.astream("large.py", large_source):
        consume(chunk)

# Async batch (concurrent)
results = asyncio.run(chunker.abatch(
    [
        {"filepath": "a.py", "code": code_a},
        {"filepath": "b.ts", "code": code_b},
    ],
    concurrency=8,
))

batch_results = chunker.batch(
    [
        {"filepath": "a.py", "code": code_a},
        {"filepath": "b.ts", "code": code_b},
        {"filepath": "README.md", "code": readme_md},
    ],
    concurrency=8,
)

directory_results = chunker.chunk_directory(
    "./src",
    glob="**/*.py",
    exclude=["**/tests/**"],
    concurrency=8,
)

all_chunks = [chunk for result in directory_results for chunk in result.chunks]

jsonl_payload = chunker.to_jsonl(all_chunks)
csv_payload = chunker.to_csv(all_chunks)

stats = chunker.chunk_stats(all_chunks, size_unit="chars")
quality = chunker.quality_scores(
    all_chunks,
    min_chunk_size=80,
    max_chunk_size=1024,
    size_unit="chars",
)

langchain_docs = chunker.to_langchain_docs(all_chunks)
llamaindex_docs = chunker.to_llamaindex_docs(all_chunks)

# Vector DB–ready rows (you compute embeddings elsewhere)
from omnichunk import chunks_to_pinecone_vectors, chunks_to_supabase_rows

emb = [[0.1, 0.2, 0.3] for _ in all_chunks]  # same length as chunks
pinecone_batch = chunks_to_pinecone_vectors(all_chunks, emb, namespace="my_ns")
weaviate_batch = chunker.to_weaviate_objects(all_chunks, emb, class_name="Doc")
supabase_rows = chunks_to_supabase_rows(all_chunks, emb)

import numpy as np
from omnichunk import Chunker

def embed(texts):
    # Replace with your actual embedding model
    return np.random.default_rng(0).standard_normal((len(texts), 384))

chunker = Chunker(max_chunk_size=512, size_unit="tokens")
essay = "Your prose here…"
chunks = chunker.semantic_chunk("essay.md", essay, embed_fn=embed)

from omnichunk.semantic import detect_topic_shifts, split_sentences

text = "Your document…"
sentences_with_offsets = split_sentences(text)
sentences = [s for s, _, _ in sentences_with_offsets]
shifts = detect_topic_shifts(sentences, window=5, threshold=0.4)

from omnichunk import Chunker, build_chunk_graph

source = "class MyClass:\n    pass\n"
chunks = Chunker().chunk("repo.py", source)
graph = build_chunk_graph(chunks)
print(graph.entity_chunks("MyClass"))       # chunk indices containing MyClass
print(graph.chunk_neighbors(0))             # chunks sharing entities with chunk 0
data = graph.to_dict()                      # JSON-serializable

from omnichunk import Chunker

chunker = Chunker(size_unit="chars")
source = "..."  # your file contents
tree = chunker.hierarchical_chunk(
    "service.py", source,
    levels=[64, 256, 1024],   # leaf → root
)

small_chunks = tree.leaves()   # embed and index these
large_chunks = tree.roots()    # pass these to LLM as context
parent = tree.parent(small_chunks[0])  # navigate up

from omnichunk import Chunker

chunker = Chunker(max_chunk_size=512, size_unit="chars")
new_source = "..."  # updated file contents
diff = chunker.chunk_diff(
    "api.py",
    new_source,
    previous_chunks=old_chunks,
)
# diff.added        → upsert to vector DB
# diff.removed_ids  → delete from vector DB
# diff.unchanged    → skip re-embedding

from omnichunk.budget import TokenBudgetOptimizer

optimizer = TokenBudgetOptimizer(budget=4096, strategy="greedy")
result = optimizer.select(retrieved_chunks, scores=relevance_scores)
# result.selected → pass to LLM

from omnichunk import register_parser, register_formatter, Chunker

def my_parse(filepath: str, content: str):
    # Return a tree-sitter-like tree, or None to use the built-in parser.
    return None

register_parser("python", my_parse, overwrite=True)

def my_fmt(chunks):
    return str(len(chunks))

register_formatter("count", my_fmt)

from omnichunk import chunk_file

chunks = chunk_file("path/to/file.py")

from omnichunk import chunk_directory

results = chunk_directory("./src", glob="**/*.py", max_chunk_size=512, size_unit="chars")

for result in results:
    if result.error:
        print("error", result.filepath, result.error)
    else:
        print(result.filepath, len(result.chunks))

original_bytes = source.encode("utf-8")
for chunk in chunks:
    assert original_bytes[chunk.byte_range.start:chunk.byte_range.end].decode("utf-8") == chunk.text