Agent Worker

AgentWorker(inventory: object, broker: str, worker_name: str, exit_event: object = None, init_done_event: object = None, log_level: str = 'WARNING', log_queue: object = None)

Bases: NFPWorker

This class represents a worker that interacts with a language model to handle various tasks such as chatting with users, retrieving inventory, and producing version reports of Python packages.

Parameters:

Name	Type	Description	Default
inventory	object	The inventory object to be used by the worker.	required
broker	str	The broker URL to connect to.	required
worker_name	str	The name of this worker.	required
exit_event	object	An event that, if set, indicates the worker needs to stop/exit.	None
init_done_event	object	An event to set when the worker has finished initializing.	None
log_level	str	The logging level of this worker. Defaults to "WARNING".	'WARNING'
log_queue	object	The logging queue object.	None

Attributes:

Name	Type	Description
agent_inventory		The inventory loaded from the broker.
llm_model	str	The language model to be used. Defaults to "llama3.1:8b".
llm_temperature	float	The temperature setting for the language model. Defaults to 0.5.
llm_base_url	str	The base URL for the language model. Defaults to "http://127.0.0.1:11434".
llm_flavour	str	The flavour of the language model. Defaults to "ollama".
llm	str	The language model instance.

Methods:

Name	Description
worker_exit	Placeholder method for worker exit logic.
get_version	Produces a report of the versions of Python packages.
get_inventory	Returns the agent's inventory.
get_status	Returns the status of the worker.
_chat_ollama	Handles the chat interaction with the Ollama LLM.
chat	Handles the chat interaction with the user by processing the input through a language model.

Source code in norfab\workers\agent_worker\agent_worker.py

def __init__(
    self,
    inventory: object,
    broker: str,
    worker_name: str,
    exit_event: object = None,
    init_done_event: object = None,
    log_level: str = "WARNING",
    log_queue: object = None,
) -> None:
    super().__init__(
        inventory, broker, SERVICE, worker_name, exit_event, log_level, log_queue
    )
    self.init_done_event = init_done_event

    # get inventory from broker
    self.agent_inventory = self.load_inventory()
    self.llms = {}

    self.init_done_event.set()
    log.info(f"{self.name} - Started")

get_llm(model: str, provider: str = None, **kwargs: object) -> object

Retrieve or create an LLM instance.

Parameters:

Name	Type	Description	Default
model	str | None	Name of the model to obtain.	required
provider	str	Model provider name, e.g. "ollama".	None
kwargs	dict	Any additional model parameters supported by LangChain.	{}

Returns:

Type	Description
object	object | None: The LLM instance (e.g. ChatOllama)

Source code in norfab\workers\agent_worker\agent_worker.py

def get_llm(self, model: str, provider: str = None, **kwargs: object) -> object:
    """
    Retrieve or create an LLM instance.

    Args:
        model (str | None): Name of the model to obtain.
        provider (str): Model provider name, e.g. "ollama".
        kwargs (dict): Any additional model parameters supported by LangChain.

    Returns:
        object | None: The LLM instance (e.g. ChatOllama)
    """
    model_data = {"model": model, **kwargs}

    # instantiate llm object
    if model in self.llms:
        llm = self.llms[model]
    elif provider == "ollama":
        from langchain_ollama import ChatOllama

        llm = ChatOllama(**model_data)
    elif provider == "groq":
        from langchain_groq import ChatGroq

        llm = ChatGroq(**model_data)
    else:
        log.error(f"Unsupported LLM provider '{provider}'")
        return None

    # store LLM for future references
    self.llms.setdefault(model, llm)

    return llm

get_version() -> Result

Generate a report of the versions of specific Python packages and system information. This method collects the version information of several Python packages and system details, including the Python version, platform, and a specified language model.

Returns:

Name	Type	Description
Result	Result	An object containing a dictionary with the package names as keys and their respective version numbers as values. If a package is not found, its version will be an empty string.

Source code in norfab\workers\agent_worker\agent_worker.py

@Task(
    input=GetVersionInput,
    output=GetVersionResult,
    fastapi={"methods": ["GET"]},
    mcp={
        "annotations": {
            "title": "Get Version",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": False,
        }
    },
)
def get_version(self) -> Result:
    """
    Generate a report of the versions of specific Python packages and system information.
    This method collects the version information of several Python packages and system details,
    including the Python version, platform, and a specified language model.

    Returns:
        Result: An object containing a dictionary with the package names as keys and their
                respective version numbers as values. If a package is not found, its version
                will be an empty string.
    """
    libs = {
        "norfab": "",
        "langchain": "",
        "langchain-community": "",
        "langchain-core": "",
        "langchain-ollama": "",
        "ollama": "",
        "python": sys.version.split(" ")[0],
        "platform": sys.platform,
    }
    # get version of packages installed
    for pkg in libs.keys():
        try:
            libs[pkg] = importlib.metadata.version(pkg)
        except importlib.metadata.PackageNotFoundError:
            pass

    return Result(result=libs)

get_inventory() -> Result

NorFab task to retrieve the agent's inventory.

Returns:

Name	Type	Description
Result	Result	An instance of the Result class containing the agent's inventory.

Source code in norfab\workers\agent_worker\agent_worker.py

@Task(
    input=GetInventoryInput,
    output=GetInventoryResult,
    fastapi={"methods": ["GET"]},
    mcp={
        "annotations": {
            "title": "Get Inventory",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": False,
        }
    },
)
def get_inventory(self) -> Result:
    """
    NorFab task to retrieve the agent's inventory.

    Returns:
        Result: An instance of the Result class containing the agent's inventory.
    """
    return Result(result=self.agent_inventory)

get_status() -> Result

NorFab Task that retrieves the status of the agent worker.

Returns:

Name	Type	Description
Result	Result	An object containing the status result with a value of "OK".

Source code in norfab\workers\agent_worker\agent_worker.py

@Task(
    input=GetStatusInput,
    output=GetStatusResult,
    fastapi={"methods": ["GET"]},
    mcp={
        "annotations": {
            "title": "Get Status",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": False,
        }
    },
)
def get_status(self) -> Result:
    """
    NorFab Task that retrieves the status of the agent worker.

    Returns:
        Result: An object containing the status result with a value of "OK".
    """
    return Result(result="OK")