Aszinkron feladatok

12.1   Mikor válaszd az IAsyncTaskTool-t?

Az eddig megismert tool-típusok mindegyike egy adott pipeline-hoz kötött: az IChatbotTool a felhasználói kérdésre válaszol, az IIngestionTool a dokumentum-feltöltés pipeline végén fut. Az IAsyncTaskTool egyik pipeline-hoz sem tartozik – egy külső backend integrátor a Platform API-n keresztül, tetszőleges időpontban indítja el, a feladat típusazonosítójának megnevezésével.

Tipikus felhasználási eset: egy már betöltött tudásgráf utólagos, célzott bővítése (pl. új entitás- vagy relációtípus pótlása a teljes ingestion újrafuttatása nélkül), vagy bármilyen más, a chat-kérésektől független, percekig-órákig futó feldolgozás, amelynek állapotát a hívó fél időről időre lekérdezi.

12.2   A szerződés – IAsyncTaskTool<Request>

Az egyenként feldolgozott elemek közötti, elszigetelt hibákat a végrehajtás saját maga kezeli – naplózza, a progressReporter hiba-számlálóján jelzi –, és a feldolgozást a többi elemmel folytatja. Ha viszont az executeTask() kivétellel tér vissza, azt a hívó a teljes feladat meghiúsulásaként kezeli, és újrapróbálkozást indíthat.

@Component
public class ExampleBackfillTask implements IAsyncTaskTool<ExampleBackfillRequest> {

    @Override
    public String getTaskType() {
        return "example_backfill";
    }

    @Override
    public String getDomain() {
        return "EXAMPLE";
    }

    @Override
    public Class<ExampleBackfillRequest> getRequestClass() {
        return ExampleBackfillRequest.class;
    }

    @Override
    public void executeTask(String taskId, String domain, ExampleBackfillRequest request,
            AsyncTaskProgressReporter progressReporter) throws Exception {
        progressReporter.reportPhase("PROCESSING");
        // ... a feladat saját feldolgozása
    }
}

Ha a feladat indításához nincs szükség semmilyen bemeneti paraméterre, a getRequestClass() az EmptyAsyncTaskRequest közös, mezők nélküli típust adhatja vissza – nem kell minden paraméter nélküli feladathoz külön, azonos szerepű, üres kérés-osztályt létrehozni.

A getDomain() az ALL állandót is visszaadhatja: az ezt visszaadó feladat-végrehajtó minden domain-hez érkező indítási kéréssel elindítható, függetlenül attól, milyen domain-t ad meg a kérés – az érintendő konkrét domain-t ilyenkor az executeTask() domain paramétere hordozza.

12.3   Regisztráció

Az IAsyncTaskTool nem terjeszti ki az IChatbotTool-t, ezért a discovery is elkülönül a chat tool-okétól: a plugin.jar-ban egy önálló META-INF/services fájl szükséges hozzá, saját sorral minden regisztrálandó osztályhoz:

META-INF/services/hu.medical.rag.common.tool.intf.IAsyncTaskTool

A betöltés menete megegyezik a többi plugin tool-éval: a platform @Autowired-del feltölti a mezőket, majd lefuttatja a @PostConstruct-tal jelölt metódusokat – ezért az IAsyncTaskTool implementációnak is publikus, paraméter nélküli konstruktorral kell rendelkeznie, a domain-specifikus függőségeket (pl. egy NER ingestion tool referenciáját) @Autowired mezőn és @PostConstruct init()-en keresztül kell bekötni.

12.4   AsyncTaskProgressReporter – élő előrehaladás-jelzés

Az executeTask() egy AsyncTaskProgressReporter csatornát kap paraméterként – ezen keresztül teszi közzé a feladat, hol tart, anélkül hogy ismernie kellene, hogyan és hová kerül ez az információ tárolásra.

12.5   AbstractGraphEnrichmentTask – kész alap tudásgráf-bővítéshez

Egy gyakori aszinkron feladat-fajta: egy már betöltött domain tudásgráfjának utólagos, célzott bővítése – új entitás- és/vagy relációtípus pótlása a teljes ingestion újrafuttatása nélkül, a domain már tárolt szövegén. Erre a platform egy kész abstract alaposztályt biztosít:

A forrás-szöveg betöltését, az LLM-alapú kinyerést, a kereszt-oldal elemzést és a gráfba-írást az executeTask() már implementálva végzi el, a beállított NER ingestion tool-on keresztül – a leszármazottnak csak azt kell megadnia, milyen entitás- és relációtípusokat keressen ebben a futásban.

A bekötés a többi plugin tool típusnál megszokott no-arg konstruktor + @PostConstruct init() mintát követi:

@Component
public class SupplierRelationBackfillTask
        extends AbstractGraphEnrichmentTask<EmptyAsyncTaskRequest> {

    private static final Set<String> TARGET_ENTITY_TYPES = Set.of("SUPPLIER");
    private static final Set<String> TARGET_RELATION_TYPES = Set.of("SUPPLIES");

    @Autowired
    private MyDomainNERIngestionTool nerIngestionTool;

    public SupplierRelationBackfillTask() {
    }

    @PostConstruct
    public void init() {
        setAbstractNERIngestionTool(nerIngestionTool);
    }

    @Override
    protected List<EntityTypeDescriptor> defineEntityTypes() {
        return nerIngestionTool.defineEntityTypes().stream()
            .filter(d -> TARGET_ENTITY_TYPES.contains(d.type()))
            .collect(Collectors.toList());
    }

    @Override
    protected List<RelationTypeDescriptor> defineRelationTypes() {
        return nerIngestionTool.defineRelationTypes().stream()
            .filter(d -> TARGET_RELATION_TYPES.contains(d.type()))
            .collect(Collectors.toList());
    }

    @Override
    public String getTaskType() {
        return "supplier_relation_backfill";
    }

    @Override
    public Class<EmptyAsyncTaskRequest> getRequestClass() {
        return EmptyAsyncTaskRequest.class;
    }
}

12.5b   AbstractGraphEnrichmentTask – újraosztályozási mód

Ha a defineReclassificationSourceTypes() hook felülírva egy nem üres entitástípus-halmazzal tér vissza, a feladat futáskor nem a fenti, szövegből kinyerő módban fut, hanem a tudásgráfban már meglévő, a megadott típusú fogalmakat vizsgálja végig: minden ilyen fogalom jelenlegi megnevezését és típusát egy LLM elé tárja a defineReclassificationTargetTypes() által felkínált típuslistával együtt, majd a kapott döntés alapján a fogalom típusát a helyén (azonos ArangoDB kulcs mellett) frissíti - a korábbi típust a fogalom previousType mezőjében, a változás időpontját a retypedAt mezőben megőrizve, hogy egy elhibázott átosztályozás utólag visszakereshető/visszaállítható legyen (egy fogalom ismételt átosztályozása esetén ez a két mező csak a legutóbbi változást tükrözi, nem egy teljes előzménylistát). A fogalom aliasai, mention-jei és a hozzá kapcsolódó élek túlnyomó része érintetlen marad - a _key nem változik, ezért a rá mutató kapcsolatok automatikusan a retípusozott csomópontra mutatnak tovább, migrálás nélkül.

@Component
public class AnatomyOtherReclassificationTask
        extends AbstractGraphEnrichmentTask<EmptyAsyncTaskRequest> {

    private static final Set<String> SOURCE_TYPES = Set.of("ANATOMY", "OTHER");

    private static final List<RelationTypeMigration> MIGRATIONS = List.of(
        new RelationTypeMigration("AFFECTS", "PHARMACOLOGICAL_TARGET", "MODULATES")
    );

    @Autowired
    private MyDomainNERIngestionTool nerIngestionTool;

    @PostConstruct
    public void init() {
        setAbstractNERIngestionTool(nerIngestionTool);
    }

    @Override
    protected Set<String> defineReclassificationSourceTypes() {
        return SOURCE_TYPES;
    }

    @Override
    protected List<EntityTypeDescriptor> defineReclassificationTargetTypes() {
        return nerIngestionTool.defineEntityTypes();
    }

    @Override
    protected List<RelationTypeMigration> defineRelationTypeMigrations() {
        return MIGRATIONS;
    }

    // defineEntityTypes()/defineRelationTypes() (a kinyerési mód hookjai) ilyenkor is
    // implementálandók a fordításhoz, de amíg defineReclassificationSourceTypes() nem üres,
    // futáskor nem kerülnek meghívásra.
}

12.6   Élő progress Spark partíciókból – PartitionProgressCounter

Ha a feladat saját Spark transformer-en keresztül, sok partíción párhuzamosan dolgozik, a PartitionProgressCounter statikus segédosztály ad módot arra, hogy minden partíció a saját, elkülönített számlálóját írja – anélkül hogy a partíciók egymást felülírnák vagy versenyhelyzetbe kerülnének.

A basePath és a phase paraméternek meg kell egyeznie azzal, amit a feladat a progressReporter.beginLiveProgress(phase, progressBasePath, ...) hívásban megadott – a jelentő csatorna ez alapján találja meg és olvassa fel a partíciónkénti számlálókat.

dataset.foreachPartition(rows -> {
    int partitionId = TaskContext.getPartitionId();
    for (Row row : rows) {
        // ... egy elem feldolgozása
        PartitionProgressCounter.increment(progressBasePath, "PROCESSING", partitionId, "succeeded");
    }
});