`Pkg.AIAssistant` — API for using Large Language Models¶

Overview¶

The Pkg.AIAssistant module contains the API for using Large Language Models (LLMs).

This module offers both an abstract API to process tasks using a queue system as well a way to provide AI-driven analysis for file formats.

Processing AI Tasks¶

The following code example demonstrates how to process a simple task.

from Pkg.AIAssistant import *

import time

def simpleAIQueue():
    queue = AIQueue()
    task = AITask(1, "Test task")
    task.setInstructions("Answer the question.")
    task.addInput("How are you?")
    queue.addTask(task)
    while not queue.process():
        time.sleep(0.1)
    output = queue.getOutput()
    print(output)

AI Analysis¶

The package allows to install custom handlers for formats not internally supported.

In order to create a custom handler it is necessary to create an entry in the ‘ai_analysis.cfg’ configuration file.

This is the entry for the “LNK” format:

[Windows LNK]
format = LNK
file = Pkg.AIAssistant.Handlers.Lnk.pyc
handler = WindowsLnkAIAnalysisHandler

‘handler’ specifies the name of the handler class in ‘file’.

The following is the handler for the LNK format:

from Pro.Core import *

from Pkg.AIAssistant import AIAnalysisHandler, AITask

class WindowsLnkAIAnalysisHandler(AIAnalysisHandler):

    def initTasks(self, *, wo=None):
        t = AITask(1, "Analyze Windows link")
        return [t]

    def loadTask(self, tasks, task, *, wo=None):
        id = task.getId()
        if id == 1:
            self.setDefaultInstructions(task)
            out = NTTextBuffer()
            self.getObject().Dump(out)
            task.addInput("Analyze the provided Windows Link for potential malware.")
            task.addInput(out.buffer)
            return True
        return False

The LNK format only requires a single analysis task to be completed. However, many file formats require several tasks to be completed and summarized. Here is an abridged example demonstrating how to summarize multiple tasks:

class ComplexAIAnalysisHandler(AIAnalysisHandler):

    First_TaskId = 1
    Second_TaskId = 2
    # etc.

    Report_TaskId = 100

    def initTasks(self, *, wo=None):
        l = []
        l.append(AITask(self.First_TaskId, "Analyze first"))
        l.append(AITask(self.Second_TaskId, "Analyze second", optional=True))
        # etc.
        l.append(AITask(self.Report_TaskId, "Create report", priority=1000))
        return l

    def loadTask(self, tasks, task, *, wo=None):
        id = task.getId()
        if id == self.First_TaskId:
            self.setDefaultInstructions(task, mode=self.PARTIAL)
            # TODO: add input
            return True
        elif id == self.Second_TaskId:
            self.setDefaultInstructions(task, mode=self.PARTIAL)
            # TODO: add input
            return True
        elif id == self.Report_TaskId:
            self.setDefaultInstructions(task, mode=self.SUMMARY)
            return self.createSummary(tasks, task)
        return False

Module API¶

Pkg.AIAssistant module API.

Classes:

AIAnalysisHandler()

Base class for analysis handlers that define AI-driven task workflows for a scanned object.

AIAnalysisHandlerInfo()

Describes a registered AI analysis handler for a specific format.

AIQueue(*, user_data, manager, model, …)

A queue that manages and executes AITask instances using an AI model.

AIQueueView(queue, *, helper)

View to display an AIQueue.

AIQueueViewHelper()

Helper interface used by AIQueueView.

AIService()

This class describes an AI service provider (name, credentials, default model).

AITask(id, label, *, priority, optional, enabled)

Represents a single AI task to be executed by a queue (prompt + inputs).

Functions:

getAIAnalysisHanlders()

Retrieves the registry of available AI analysis handlers.

getAIServiceCredentials(service)

Retrieves credentials (API key) for a given AI service.

getAutoAIModel()

Selects an appropriate model automatically based on available services.

getAvailableAIModels(*[, service, wo])

Lists available AI models, optionally filtered by service.

getAvailableAIServices()

Returns the names of AI services that are currently usable (credentials present).

getDefaulAIModel()

Returns the global default AI model if one is available; otherwise returns None.

getDefaultAIModelForService(service)

Retrieves the default model for a specific AI service.

getSupportedAIServices()

Returns a list of supported AI services.

hasAIAnalysisHandler(handlers, format)

Checks whether a handler exists for the specified format.

initAIAnalysisHandler(handlers, sprovider)

Initializes an AIAnalysisHandler instance suitable for the given ScanProvider.

isAIServiceAvailable(service)

Checks whether the given AI service can be used (e.g., credentials present).

class AIAnalysisHandler¶

Base class for analysis handlers that define AI-driven task workflows for a scanned object.

Attributes:

PARTIAL

Partial analysis task mode (one of many contributing tasks).

REPORT

Single-task mode producing a complete report.

SUMMARY

Summary mode that aggregates outputs of partial tasks.

Methods:

addJSONInput(task, js)

Adds a JSON payload as input to a task.

createSummary(tasks, task, *[, minimum_priority])

Builds a summary output from a set of partial tasks into the provided summary task.

getObject()

Returns the object being analyzed.

getReport(tasks, *[, wo])

Produces a final report from task results.

getScanProvider()

Returns the scan provider instance.

getTasks(*[, wo])

Creates the set of tasks to be executed for this analysis.

initTasks(*[, wo])

Initializes and returns the base set of tasks for this handler.

loadTask(tasks, task, *[, wo])

Loads data and context into a specific task before processing.

setDefaultInstructions(task, *[, mode])

Applies default instructions to a task based on the chosen mode.

structToJSON(s)

Serializes a CFFStruct to a JSON-serializable dictionary.

PARTIAL¶

Partial analysis task mode (one of many contributing tasks).

REPORT¶

Single-task mode producing a complete report.

SUMMARY¶

Summary mode that aggregates outputs of partial tasks.

addJSONInput(task: Pkg.AIAssistant.AITask, js: Union[Dict[Any, Any], str])¶

Adds a JSON payload as input to a task.

Parameters

task (AITask) – The target task.

js (Union[Dict[Any, Any], str]) – The JSON-like mapping or string to attach.

See also structToJSON().

createSummary(tasks: List[Pkg.AIAssistant.AITask], task: Pkg.AIAssistant.AITask, *, minimum_priority: int = 0) → bool¶

Builds a summary output from a set of partial tasks into the provided summary task.

Parameters

tasks (List[AITask]) – The full task set.

task (AITask) – The summary task to receive the aggregated input.

minimum_priority (int) – Only include tasks with at least this priority.

Returns

Returns True if the summary was created; otherwise returns False.

Return type

bool

getObject() → Pro.Core.CFFObject ¶

Returns

Returns the object being analyzed.

Return type

CFFObject

See also getScanProvider().

getReport(tasks: List[Pkg.AIAssistant.AITask], *, wo: Optional[Pro.Core.NTIWait] = None) → Optional[str]¶

Produces a final report from task results.

Note

Most handlers can rely on the default implementation of this method, unless they need to customize the behavior.

Parameters

tasks (List[AITask]) – The tasks whose outputs should be consolidated.

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns the report text if available; otherwise returns None.

Return type

Optional[str]

getScanProvider() → Pro.Core.ScanProvider ¶

Returns

Returns the scan provider instance.

Return type

ScanProvider

See also getObject().

getTasks(*, wo: Optional[Pro.Core.NTIWait] = None) → List[Pkg.AIAssistant.AITask]¶

Creates the set of tasks to be executed for this analysis.

Hint

This method internally calls initTasks().

Parameters

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns a list of prepared AITask instances.

Return type

List[AITask]

See also initTasks().

initTasks(*, wo: Optional[Pro.Core.NTIWait] = None) → List[Pkg.AIAssistant.AITask]¶

Initializes and returns the base set of tasks for this handler.

Parameters

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns list of newly created AITask instances.

Return type

List[AITask]

See also loadTask().

loadTask(tasks: List[Pkg.AIAssistant.AITask], task: Pkg.AIAssistant.AITask, *, wo: Optional[Pro.Core.NTIWait] = None) → bool¶

Loads data and context into a specific task before processing.

Parameters

tasks (List[AITask]) – The full task set (for cross-references if needed).

task (AITask) – The task to populate.

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns True if the task was prepared; otherwise returns False.

Return type

bool

See also initTasks().

setDefaultInstructions(task: Pkg.AIAssistant.AITask, *, mode: int = REPORT)¶

Applies default instructions to a task based on the chosen mode.

Parameters

task (AITask) – The task to configure.

mode (int) – One of REPORT, PARTIAL or SUMMARY.

structToJSON(s: Pro.Core.CFFStruct) → Dict[str, Any]¶

Serializes a CFFStruct to a JSON-serializable dictionary.

Parameters

s (CFFStruct) – The structure to serialize.

Returns

Returns a dictionary suitable for JSON encoding.

Return type

Dict[str, Any]

See also addJSONInput().

class AIAnalysisHandlerInfo¶

Describes a registered AI analysis handler for a specific format.

Attributes:

file

The file that defines the handler.

format

The format handled by this handler.

handler

The fully qualified class name implementing AIAnalysisHandler.

name

A human-friendly handler name.

file¶

The file that defines the handler.

format¶

The format handled by this handler.

handler¶

The fully qualified class name implementing AIAnalysisHandler.

name¶

A human-friendly handler name.

class AIQueue(*, user_data: Optional[object] = None, manager: Optional[Pro.MP.ProManager] = None, model: Optional[object] = None, api_key: Optional[str] = None, verbose: bool = True)¶

A queue that manages and executes AITask instances using an AI model.

Parameters

user_data (Optional[object]) – Optional caller-provided context object.

manager (Optional[ProManager]) – Optional user-provided ProManager for multi-processing.

model (Optional[object]) – Optional model descriptor to use initially.

api_key (Optional[str]) – Optional API key override.

verbose (bool) – Enable verbose logging.

Methods:

abort()

Requests cancellation of the current processing session.

addTask(task)

Adds a single task the list.

addTasks(tasks)

Adds multiple tasks the list.

getModel()

Returns the current model descriptor.

getOutput(*[, wo])

Retrieves the final output produced by the queue.

getTasks()

Returns the list of tasks.

getUserData()

Returns the user data associated with this queue.

hasFinished()

Returns True if processing is complete; otherwise returns False.

isProcessing()

Returns True if processing is ongoing; otherwise returns False.

prepareTasks(*[, wo, priority, max_count])

Prepares tasks for processing.

process(*[, wo])

Processes tasks asynchronously, honoring their priority.

reset()

Resets the queue to its initial state, clearing results and statuses.

setModel(model)

Sets the model descriptor to be used for subsequent processing.

abort()¶

Requests cancellation of the current processing session.

See also reset().

addTask(task: Pkg.AIAssistant.AITask)¶

Adds a single task the list.

Parameters

task (AITask) – The task to add.

See also addTasks().

addTasks(tasks: List[Pkg.AIAssistant.AITask])¶

Adds multiple tasks the list.

Parameters

tasks (List[AITask]) – The tasks to add.

See also addTask().

getModel() → object¶

Returns

Returns the current model descriptor.

Return type

object

See also setModel().

getOutput(*, wo: Optional[Pro.Core.NTIWait] = None) → Optional[str]¶

Retrieves the final output produced by the queue.

Parameters

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns the output if available; otherwise returns None.

Return type

Optional[str]

See also process().

getTasks() → List[Pkg.AIAssistant.AITask]¶

Returns

Returns the list of tasks.

Return type

List[AITask]

See also addTask() and addTasks().

getUserData() → object¶

Returns

Returns the user data associated with this queue.

Return type

object

hasFinished() → bool¶

Returns

Returns True if processing is complete; otherwise returns False.

Return type

bool

See also process() and isProcessing().

isProcessing() → bool¶

Returns

Returns True if processing is ongoing; otherwise returns False.

Return type

bool

See also process() and hasFinished().

prepareTasks(*, wo: Optional[Pro.Core.NTIWait] = None, priority: int = 0, max_count: int = - 1)¶

Prepares tasks for processing.

Note

Tasks are prepared when process() is called. This method offers the possibility to prepare some tasks before calling process().

Parameters

wo (Optional[NTIWait]) – Optional wait object.

priority (int) – Minimum priority threshold to include.

max_count (int) – Maximum number of tasks to prepare, or -1 for all.

process(*, wo: Optional[Pro.Core.NTIWait] = None) → bool¶

Processes tasks asynchronously, honoring their priority.

Hint

This method is designed to be called repeatedly in a loop until it returns True

Parameters

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns False while there is still work to do, and True once all tasks have finished.

Return type

bool

See also isProcessing(), hasFinished() and getOutput().

reset()¶

Resets the queue to its initial state, clearing results and statuses.

See also abort().

setModel(model: object)¶

Sets the model descriptor to be used for subsequent processing.

Parameters

model (object) – The model object to configure.

See also getModel().

class AIQueueView(queue: Pkg.AIAssistant.AIQueue, *, helper: Optional[Pkg.AIAssistant.AIQueueViewHelper] = None)¶

View to display an AIQueue.

Initializes an AIQueueView.

Parameters

queue (AIQueue) – The underlying queue instance.

helper (Optional[AIQueueViewHelper]) – Optional helper to manage persistence and lifecycle events.

Methods:

setup(sdata)

Binds the view to the provided ScanViewData.

setup(sdata: Pro.Core.ScanViewData) → bool¶

Binds the view to the provided ScanViewData.

Parameters

sdata (ScanViewData) – The scan view data object.

Returns

Returns True if successful; otherwise returns False.

Return type

bool

class AIQueueViewHelper¶

Helper interface used by AIQueueView.

Methods:

closing()

Called when the view is closing.

fetchOutput(model)

Allows the helper to provide preexisting output for the given model (if any).

saveOutput(model, output)

Allows the helper to persist output produced by the queue.

closing()¶

Called when the view is closing.

fetchOutput(model: object) → Optional[str]¶

Allows the helper to provide preexisting output for the given model (if any).

Parameters

model (object) – The model descriptor currently selected.

Returns

Returns the output if available; otherwise returns None.

Return type

Optional[str]

See also saveOutput().

saveOutput(model: object, output: str)¶

Allows the helper to persist output produced by the queue.

Parameters

model (object) – The model descriptor currently selected.

output (str) – The output text to save.

See also fetchOutput().

class AIService¶

This class describes an AI service provider (name, credentials, default model).

Attributes:

default_model

The service’s default model identifier (if any).

env_key_name

The environment variable name used to retrieve the API key for this service.

name

The display name of the AI service (e.g., "OpenAI", "Claude").

Methods:

getAPIKey()

Retrieves the API key associated with this service (typically from an environment variable).

default_model¶

The service’s default model identifier (if any).

env_key_name¶

The environment variable name used to retrieve the API key for this service.

getAPIKey() → str¶

Retrieves the API key associated with this service (typically from an environment variable).

Returns

The API key string, or an empty string if not available.

Return type

str

name¶

The display name of the AI service (e.g., "OpenAI", "Claude").

class AITask(id: int, label: str, *, priority: int = 0, optional: bool = False, enabled: bool = True)¶

Represents a single AI task to be executed by a queue (prompt + inputs).

Parameters

id (int) – Unique identifier for the task.

label (str) – Human-readable label for display/logging.

priority (int) – Optional priority (higher values indicate higher priority).

optional (bool) – If True, makes the task optional.

enabled (bool) – If the task is optional, it can be enabled or disabled using this parameter.

Attributes:

ASSISTANT

Input role indicating assistant-provided content.

FAILED

Task failed with an error.

NONE

No status set.

PENDING

Task is queued but not yet running.

RUNNING

Task is currently being processed.

SKIPPED

Task was skipped (e.g., due to a load error).

SUCCEEDED

Task completed successfully.

TIMEDOUT

Task exceeded the configured time limit.

USER

Input role indicating user-provided content.

Methods:

addInput(content, *[, type, role])

Adds an input message to the task.

getElapsed()

Returns the elapsed processing time for this task in milliseconds.

getError()

Returns the error message, if any; otherwise returns an empty string.

getId()

Returns the task id.

getLabel()

Returns the task label.

getOutput()

Returns the task output produced by the AI model if the status is SUCCEEDED; otherwise returns an empty string.

getPriority()

Returns the task priority.

getStatus()

Returns the current task status (e.g., PENDING, RUNNING).

isEnabled()

Returns True if the task is enabled; otherwise returns False.

isOptional()

Returns True if the task is optional; otherwise returns False.

setEnabled(enabled)

Sets whether the task is enabled or not.

setInstructions(text)

Sets the system/instruction prompt for this task.

ASSISTANT¶

Input role indicating assistant-provided content.

FAILED¶

Task failed with an error.

NONE¶

No status set.

PENDING¶

Task is queued but not yet running.

RUNNING¶

Task is currently being processed.

SKIPPED¶

Task was skipped (e.g., due to a load error).

SUCCEEDED¶

Task completed successfully.

TIMEDOUT¶

Task exceeded the configured time limit.

USER¶

Input role indicating user-provided content.

addInput(content: str, *, type: str = 'text', role: int = USER)¶

Adds an input message to the task.

Parameters

content (str) – Message content (text or serialized data).

type (str) – Content type hint (e.g., "text", "json").

role (int) – Message role (USER or ASSISTANT).

See also setInstructions().

getElapsed() → int¶

Returns

Returns the elapsed processing time for this task in milliseconds.

Return type

int

getError() → str¶

Returns

Returns the error message, if any; otherwise returns an empty string.

Return type

str

getId() → int¶

Returns

Returns the task id.

Return type

int

getLabel() → str¶

Returns

Returns the task label.

Return type

str

getOutput() → str¶

Returns

Returns the task output produced by the AI model if the status is SUCCEEDED; otherwise returns an empty string.

Return type

str

getPriority() → int¶

Returns

Returns the task priority.

Return type

int

getStatus() → int¶

Returns

Returns the current task status (e.g., PENDING, RUNNING).

Return type

int

isEnabled() → bool¶

Returns

Returns True if the task is enabled; otherwise returns False.

Return type

bool

See also setEnabled().

isOptional() → bool¶

Returns

Returns True if the task is optional; otherwise returns False.

Return type

bool

setEnabled(enabled: bool)¶

Sets whether the task is enabled or not.

Note

Only optional tasks can be disabled.

Parameters

enabled (bool) – If True enables the task; otherwise disables it.

See also isEnabled().

setInstructions(text: str)¶

Sets the system/instruction prompt for this task.

Parameters

text (str) – Instruction text guiding model behavior.

See also addInput().

getAIAnalysisHanlders() → Dict[str, Pkg.AIAssistant.AIAnalysisHandlerInfo]¶

Retrieves the registry of available AI analysis handlers.

Returns

Returns a mapping of format identifiers to AIAnalysisHandlerInfo.

Return type

Dict[str, AIAnalysisHandlerInfo]

See also hasAIAnalysisHandler() and initAIAnalysisHandler().

getAIServiceCredentials(service: Union[str, Pkg.AIAssistant.AIService]) → Optional[str]¶

Retrieves credentials (API key) for a given AI service.

Parameters

service (Union[str, AIService]) – The service name or AIService instance.

Returns

Returns the API key string if available; otherwise returns None.

Return type

Optional[str]

See also isAIServiceAvailable() and AIService.getAPIKey().

getAutoAIModel() → Optional[object]¶

Selects an appropriate model automatically based on available services.

Returns

Returns a model descriptor if a suitable model is found; otherwise returns None.

Return type

Optional[object]

See also getAvailableAIModels().

getAvailableAIModels(*, service: Optional[str] = None, wo: Optional[Pro.Core.NTIWait] = None) → List[object]¶

Lists available AI models, optionally filtered by service.

Parameters

service (Optional[str]) – Optional service name to filter by.

wo (Optional[NTIWait]) – Optional wait object.

Returns

Returns a list of model descriptors (opaque objects).

Return type

List[object]

See also getDefaultAIModelForService() and getAutoAIModel().

getAvailableAIServices() → List[str]¶

Returns

Returns the names of AI services that are currently usable (credentials present).

Return type

List[str]

See also isAIServiceAvailable() and getSupportedAIServices().

getDefaulAIModel() → Optional[object]¶

Returns

Returns the global default AI model if one is available; otherwise returns None.

Return type

Optional[object]

See also getAutoAIModel() and getDefaultAIModelForService().

getDefaultAIModelForService(service: str) → Optional[object]¶

Retrieves the default model for a specific AI service.

Parameters

service (str) – The service name.

Returns

Returns a model descriptor if available; otherwise returns None.

Return type

Optional[object]

getSupportedAIServices() → List[Pkg.AIAssistant.AIService]¶

Returns

Returns a list of supported AI services.

Return type

List[AIService]

See also getAvailableAIServices() and isAIServiceAvailable()

hasAIAnalysisHandler(handlers: Dict[str, Pkg.AIAssistant.AIAnalysisHandlerInfo], format: str) → bool¶

Checks whether a handler exists for the specified format.

Parameters

handlers (Dict[str, AIAnalysisHandlerInfo]) – The handler registry returned by getAIAnalysisHanlders().

format (str) – Format identifier to look up.

Returns

Returns True if a handler is available; otherwise returns False.

Return type

bool

See also getAIAnalysisHanlders() and initAIAnalysisHandler().

initAIAnalysisHandler(handlers: Dict[str, Pkg.AIAssistant.AIAnalysisHandlerInfo], sprovider: Pro.Core.ScanProvider) → Optional[Pkg.AIAssistant.AIAnalysisHandler]¶

Initializes an AIAnalysisHandler instance suitable for the given ScanProvider.

Parameters

handlers (Dict[str, AIAnalysisHandlerInfo]) – The handler registry returned by getAIAnalysisHanlders().

sprovider (ScanProvider) – The scan provider.

Returns

Returns an initialized analysis handler if successful; otherwise returns None.

Return type

Optional[AIAnalysisHandler]

See also getAIAnalysisHanlders() and hasAIAnalysisHandler().

isAIServiceAvailable(service: Union[str, Pkg.AIAssistant.AIService]) → bool¶

Checks whether the given AI service can be used (e.g., credentials present).

Parameters

service (Union[str, AIService]) – The service name or AIService instance.

Returns

Returns True if the service is available; otherwise returns False.

Return type

bool

See also getAvailableAIServices().

Pkg.AIAssistant — API for using Large Language Models¶