[Feature Request]: Post-insert callback / hook support

[tavurth]

Do you need to file a feature request?

• I have searched the existing feature request and this feature request is not already filed.
• I believe this is a legitimate feature request, not just a question or bug.

Feature Request Description

What problem does this solve?
There's currently no way to register a callback that fires after all documents finish processing. The only option is to await immediately after ainsert / apipeline_process_enqueue_documents at the call site, which doesn't work well for server contexts (e.g. document_routes.py) or out-of-process consumers.

Proposed API

async def on_documents_processed(results: list[DocProcessingStatus]) -> None:
    # triggered once all enqueued docs reach PROCESSED or FAILED
    ...

rag = LightRAG(
    ...
    on_insert_complete=on_documents_processed,
)

Or alternatively, accept it as a parameter on apipeline_process_enqueue_documents itself:

await rag.apipeline_process_enqueue_documents(
    on_complete=on_documents_processed
)

Why not just await?
Awaiting works fine for direct library use, but the server's pipeline_index_file fires processing as a background task — there's no clean external hook point without patching the router or monkey-patching the rag instance.

Alternatives considered

• Polling doc_status storage — works but is wasteful
• Monkey-patching rag.apipeline_process_enqueue_documents — fragile across version updates

Additional Context

No response

[danielaskdd]

Currently, a track_id is returned upon file upload, and you can query the processing status using the /track_status endpoint with your track_id.

[tavurth]

The way I worked around this now, was to inject a custom LightRAG with overrides before calling _server.get_application()

rag.py

from somewhere import post_process

class OurRAG(LightRAG):
    async def apipeline_process_enqueue_documents(self, *args, **kwargs):
        result = await super().apipeline_process_enqueue_documents(*args, **kwargs)
        await post_process(self)
        return result

main.py

from rag import OurRAG
import lightrag.api.lightrag_server as _server

_server.LightRAG = OurRAG

def get_application() -> FastAPI:
    app = _server.get_application()
    ...

This seems to work well as a (monkey patch) workaround. However it would be great to have a bunch of hooks scattered around the application code that I can self register to on startup, or perhaps there's a cleaner way?

[danielaskdd]

Integrating a web-hook is a great suggestion; however, it is currently a lower priority. We expect significant pipeline improvements following the merger of LightRAG and Rag-Anything. We will consider implementing web-hook support once the system has stabilized.