agentscope.app.rag.index_worker

A worker process owns:

• a :class:~agentscope.app._service.IndexWorker instance, which runs the parse → chunk → embed pipeline for one document at a time;
• an :class:~agentscope.app._service.IndexTaskConsumer, which subscribes to the shared task channel on the message bus and feeds the worker.

It does NOT own the sweeper — the API process keeps running the sweeper so that documents stuck in pending (because the API crashed between the storage write and the dispatcher publish) still recover, even if every worker process happens to be offline at the moment the publish was attempted.

This module is a library: a deployment wires its concrete backends through :func:run_worker from whatever bootstrap script it uses (systemd unit, Kubernetes Deployment + container entrypoint, docker-compose command, ...). The shape mirrors :func:agentscope.app.create_app — pass already-constructed backend instances, the worker manages their lifecycle through an :class:AsyncExitStack.

Example::

import asyncio
import socket
import uuid

from agentscope.app.rag.blob_store import S3BlobStore
from agentscope.app.rag.knowledge_base_manager import (
    DefaultKnowledgeBaseManager,
)
from agentscope.app.message_bus import RedisMessageBus
from agentscope.app.storage import RedisStorage
from agentscope.app.rag.index_worker import run_worker
from agentscope.rag import TextParser, ...

async def main() -> None:
    storage = RedisStorage(url=os.environ["REDIS_URL"])
    message_bus = RedisMessageBus(url=os.environ["REDIS_URL"])
    blob_store = S3BlobStore(
        bucket=os.environ["S3_BUCKET"],
        endpoint_url=os.environ.get("S3_ENDPOINT"),
    )
    kb_manager = DefaultKnowledgeBaseManager(...)
    parsers = [TextParser()]
    await run_worker(
        storage=storage,
        message_bus=message_bus,
        blob_store=blob_store,
        knowledge_base_manager=kb_manager,
        parsers=parsers,
    )

if __name__ == "__main__":
    asyncio.run(main())