AgentClient

AgentClient is the main class utilized to interface with a GAIuS agent. It provides several helper functions and API calls to aid in data input, agent configuration, and debugging.

Quickstart

class ia.gaius.agent_client.AgentClient(bottle_info: dict, verify: bool = True, timeout: float | None = None)

Bases: object

Interface for interacting with agents.

__init__(bottle_info: dict, verify: bool = True, timeout: float | None = None)

Provide agent information in a dictionary.

Parameters:

• bottle_info (dict, required) – the info used to connect and authenticate with a GAIuS agent

• verify (bool, optional) – whether to enable SSL Verification on api commands to an agent

Example

from ia.gaius.agent_client import AgentClient

agent_info = {'api_key': 'ABCD-1234',
            'name': 'gaius-agent',
            'domain': 'intelligent-artifacts.com',
            'secure': False}

agent = AgentClient(agent_info)
agent.connect()

agent.set_ingress_nodes(['P1'])
agent.set_query_nodes(['P1'])

add_blacklisted_symbols(symbols_list: List, nodes: List | None = None)

Add a symbol to the blacklist manually. List all symbols currently in the blacklist using list_blacklisted_symbols(). Symbols may be removed from this list by using remove_blacklisted_symbols().

Parameters:

• symbols_list (List) – list of symbols to add to the blacklist

• nodes (List, optional) – list of nodes to add blacklisted symbols on. Defaults to None (all nodes).

Raises:

AgentQueryError – Error in Cognitive Processor handling

Returns:

Depicting result of adding blacklisted symbols to each CP in nodes

Return type:

dict or str

add_model(sequence: List, vector: List = None, frequency: int = 1, emotives: Dict = {}, metadata: List = [], nodes: List = None) → dict | Any

Function to explicitely add a model with the given sequence of symbols, frequency, and emotives. Returns the hash of the generated function

change_genes(gene_data: Dict, nodes: List = None) → dict | Any

Change the genes in gene_data to their associated values.

This will do live updates to an existing agent, rather than stopping an agent and starting a new one, as per ‘injectGenome’. gene_data of form:

{gene: value}

Example

agent.change_genes(gene_data={"recall_threshold" : 0.15})

clear_all_memory(clear_blacklist=True, nodes: List | None = None) → dict | Tuple[dict, str]

Clear both the Working Memory and persisted memory. Equivalent to re-initializing the nodes specified

Parameters:

• clear_blacklist (bool, optional) – Controls if symbol blacklist is also cleared or not. Defaults to False.

• nodes (List, optional) – List of nodes to call command on. Defaults to None.

>>> agent.clear_all_memory(nodes=["P1"])
'all-cleared'

Returns:

_description_

Return type:

Union[dict, Tuple[dict, str]]

clear_snapshots(nodes: List = None) → dict | Tuple[dict, str]

Clear the WM snapshots of the Agent. Will not delete any persisted memory from the KnowledgeBase

Example

>>> agent.clear_snapshots(nodes=['P1'])
'snapshots-cleared'

clear_target_class(nodes: List = None) → dict | Tuple[dict, str]

Clears target selection.

Example

agent.clear_target_class(nodes=["P1"])

clear_wm(nodes: List | None = None) → dict | Tuple[dict, str]

Clear the Working Memory of the Agent. Will not delete any persisted memory from the KnowledgeBase

Example

>>> agent.clear_wm(nodes=['P1'])
'wm-cleared'

connect() → Dict

Establishes initial connection to GAIuS agent and grabs the agent’s genome for node definitions.

Example

agent_info = {...}
agent = AgentClient(agent_info)
agent.connect()

delete_model(model_name: str, nodes: List = None) → dict | Any

Deletes model with name model_name.

Model name is unique, so it should not matter that we query all nodes, all its duplicates everywhere will be deleted.

get_all_genes(nodes: List | None = None) → dict | Tuple[dict, str]

Return all genes for nodes.

Example

genes = agent.get_all_genes(nodes=["P1"])

get_cluster_info(nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to start predicting.

Example

# will enable prediction generation on node P1
agent.start_predicting(nodes=["P1"])

get_cognition_data(unique_id: str | None = None, nodes: List | None = None) → dict | Tuple[dict, str]

Return cognition data.

Example

cog_data = agent.get_cognition_data(nodes=["P1"])

get_gene(gene: str, nodes: List = None) → dict | Dict[Any, Dict[str, Any]]

Return the value for the gene gene on nodes.

Example

agent.get_gene(gene="recall_threshold")

get_hypotheses(nodes: List = None) → dict | Any

Function to explicitely get the set hypotheses

get_info(nodes: List | None = None) → dict

Get debugging information from the nodes specified

Parameters:

nodes (List, optional) – Cognitive Processors to query. Defaults to all nodes.

Raises:

AgentQueryError – Error querying node

Returns:

GetInfo response

Return type:

dict

get_interface_node_config(interface_nodes: list | None = None) → dict

Produce copy of agent config with interface_node field for use with Thinkflux

Parameters:

interface_nodes (list) – nodes to use as interface nodes

get_kbs_as_json(nodes: List = None, directory='./', filename=None, lines=False, separated=False, ids=True, obj=False, compressed=False) → List[str] | dict

Returns KBs of specified nodes as JSON Objects

Parameters:

• nodes (list, optional) – Defaults to None.

• directory (str, optional) – Defaults to ‘./’.

• separated (bool, optional) – store each KB separately. Defaults to False

• ids (bool, optional) – use primitive_ids as keys for knowledgebase instead of node numbers. Defaults to True

• obj (bool, optional) – whether to save the kbs to a filepath or a python object returned from the function

• compressed (bool, optional) – whether to save the kbs to a zip file. Not compatible with “separated” or “obj” arguments

Returns:

List of files where KB is stored if obj = False, or dictionary where KB is to be stored if obj = True

Return type:

Union[List(str), Dict]

Example

# save the KBs in a specified directory, without using node IDs
# will use node names instead (e.g. P1, P2, P3)
kb_path = "../path/to/save/dir/"
agent.get_kbs_from_json(directory=kb_path, separated=False, ids=False)

#save the KBs to the active directory, using node IDs (e.g. p46514fa2, etc.)
agent.get_kbs_from_json(ids=True)

#save the KBs to a dictionary object, using node names
kb = agent.get_kbs_from_json(ids=False, obj=True)

get_model(model_name: str, nodes: List = None) → dict | Any

Returns model with name model_name.

Model name is unique, so it should not matter that we query all nodes, only one model will be found.

Parameters:

• model_name (str) – Name of model to retrieve

• nodes (List, optional) – list of nodes to retrieve model from. Defaults to None (all nodes).

Returns:

dictionary of {node_name : model} pairs

Return type:

Union[dict, Any]

Example

agent.get_model("MODEL|d85f28ebda228064299035bf2dbf58f5c6484108")

get_models_with_patterns(symbols_list: List = None, nodes: List = None) → dict | Any

Function which search the kbs on the nodes for models with certain regex patterns.

All regex patterns must be present in model.sequence, or the model will not be returned

get_models_with_symbols(symbols_list: List = None, nodes: List = None) → dict | Any

Function which search the kbs on the nodes for models with certain symbols. All symbols must be present in model.sequence, or the model will not be returned

get_name(nodes: List | None = None) → dict | Tuple[dict, str]

Return name of nodes.

Example

agent.get_name(nodes=["P1", "P2"])

get_percept_data(nodes: List | None = None) → dict | Tuple[dict, str]

Return percept data.

Example

percept_data = agent.get_percept_data(nodes=["P1"])

get_prediction_ensemble_for_model(model_name: str, nodes: List = None) → dict | Tuple[dict, str]

Get a prediction ensemble for the model with passed hash.

Example

>>> agent.get_prediction_ensemble_for_model(model_name="a2s51cas6541321htkmy56", nodes=['P1'])
'an ensemble or empty list'

get_predictions(unique_id: str | None = None, nodes: List | None = None) → dict | Tuple[dict, str]

Return prediction result data.

Example

predictions = agent.get_predictions(nodes=["P1"])

get_successor_model(model_name: str, nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to start predicting.

Example

# will enable prediction generation on node P1
agent.start_predicting(nodes=["P1"])

get_symbol(symbol_name: str, nodes: List = None) → dict | Any

Returns symbol with name symbol_name.

Parameters:

• symbol_name (str) – Name of symbol to retrieve

• nodes (List, optional) – list of nodes to retrieve symbol information from. Defaults to None (all nodes).

Returns:

dictionary of {node_name : symbol} pairs

Return type:

Union[dict, Any]

Example

agent.get_symbol("WEATHER|CLOUDY")

get_time(nodes: List | None = None) → dict | Tuple[dict, str]

Return time on nodes.

Example

>>> agent.get_time(nodes=['P1'])
'1'

get_vector(vector_name: str, nodes: List = None) → dict | Dict[Any, Dict[str, Any]]

Return the vector with vector_name on nodes (it will be present on at most one).

Example

agent.get_vector(vector_name="VECTOR|0b7f2aac43ae8a0d94c58be08a5b8b0ab28079de")

get_wm(nodes: List | None = None) → dict | Tuple[dict, str]

Return information about Working Memory.

Example

>>> gdf = {'strings':['hello'], 'vectors': [], 'emotives':{}}
>>> agent.observe(gdf, nodes=['P1'])
>>> agent.get_wm(nodes=['P1'])
[['hello']]

increment_recall_threshold(increment: float, nodes: List = None) → Dict[Any, Dict[str, Any]]

Increment recall threshold by increment on nodes.

Parameters:

increment (float) – value by which to increment the recall threshold by

Example

agent.increment_recall_threshold(increment=0.05, nodes=["P1"])

Note

This command with only update the recall threshold by a specified value. In order to set the recall threshold to a specific value, use change_genes() instead.

inject_genome(genome: dict | str) → str

Replace agent running in container with a different genome

Parameters:

genome (Union[dict, str]) – the genome to be loaded. As a filepath or dictionary

Raises:

AgentQueryError – If there is an error injecting the genome

Returns:

response depicting operation status

Return type:

str

Example

with open(genome_path) as f:
    genome = json.load(f)
agent.inject_genome(genome)

agent.connect() # required to trigger AgentClient to update node details
agent.show_status() # will now show updated content

Note

This command will only work on agents running in “single container mode”. This means that the agent must have been spawned using the “connectome” docker image. Specify single container mode when spawning an agent via AgentManager.

investigate(record: str) → dict

Higher level implementation of investigate_record() that does not require a node to be entered. Operates on abstracted symbols, models, etc. Useful when applying in a programatic manner, where the node a symbol comes from is unknown. May throw exception if unable to identify a starting point.

Parameters:

record (str) – the symbol to look for

Returns:

recursive structure containing a complete representation of the abstracted model

Return type:

dict

Example

record = 'PRIMITIVE|p124b23f9|004da01b0ef40d7c3e0965a6b4fd0f413672fbff|matches|KEY|VALUE'
traceback = agent.investigate(record=record)

investigate_record(node, record: str) → dict

Function for drilling down on a model recursively in a multi-node agent. Will retrieve all data “under” the abstracted model, giving a complete representation of the abstracted model

Parameters:

• node (list or str) – the node to start at/containing the model provided in record: e.g. [‘P4’] or ‘P4’

• record (str) – the model to search for

Returns:

recursive structure containing a complete representation of the abstracted model

Return type:

dict

learn(nodes: List | None = None) → dict | Tuple[dict, str]

Return the learn results.

Example

>>> agent.learn(nodes=["P1"])
'MODEL|004da01b0ef40d7c3e0965a6b4fd0f413672fbff'

list_blacklisted_symbols(nodes: List | None = None)

Display a list of all symbols currently ‘blacklisted’ on each node in nodes.

Symbols enter this list by being removed from a Cognitive Processor via remove_symbols_from_system() or remove_patterns_from_system().

Symbols may be removed from this list by using remove_blacklisted_symbols().

Parameters:

nodes (List, optional) – list of nodes to show blacklisted symbols on. Defaults to None (all nodes).

Example

from ia.gaius.kb_ops import list_symbols,
    get_models_containing_symbol

symbols_before = list_symbols(agent)
blacklisted_symbols = agent.list_blacklisted_symbols()
# ensure all blacklisted symbols do not appear in the system
assert all([sym not in symbols_before for sym in blacklisted_symbols])

# ensure no models contain the blacklisted symbols
models_containing_symbol = get_models_containing_symbol(agent, symbol_set=set(blacklisted_symbols))
assert all([len(val) == 0 for val in models_containing_symbol.values()])

load_kbs_from_json(path=None, obj=None)

Load KBs from a JSON file

Parameters:

• path (str, optional) – path to JSON file where KBs are stored.

• obj (dict, optional) – object to load KnowledgeBase from

Returns:

‘success’ or ‘failed’

Return type:

str

Example

kb_path = "../path/to/kb.json"
agent.load_kbs_from_json(path=kb_path)

load_kbs_from_json_iter(path)

Uses ijson library to load KBs more efficiently

Parameters:

path (str) – path to JSON file where KBs are stored.

Returns:

‘success’ or ‘failed’

Return type:

str

Example

kb_path = "../path/to/kb.json"
agent.load_kbs_from_json(path=kb_path)

observe(data: Dict, nodes: List | None = None) → dict | Tuple[dict, str]

Exclusively uses the ‘observe’ call. Will observe on nodes specified by set_ingress_nodes() unless nodes explicitly provided

Parameters:

• data (dict) – the GDF data to be observed

• nodes (list, optional) – Nodes to observe data on, defaults to ingress nodes

Example

from ia.gaius.utils import create_gdf
>>> gdf = create_gdf(strings=['hello'])
>>> agent.observe(data=gdf, nodes=["P1"])
'observed'

observe_classification(data=None, nodes: List = None)

Send a classification to all nodes as a singular symbol in the last event.

Sending the classification as a single symbol in the last event is the canonical way to classify a sequence.

ping(nodes: List = None) → dict | Any

Ping a node to ensure it’s up.

Example

agent.ping(nodes=["P1"])

receive_unique_ids(should_set: bool = True) → bool

remove_blacklisted_symbols(symbols_to_remove: List, nodes: List | None = None)

Remove symbols from blacklist on each Cognitive Processor in nodes

Parameters:

• symbols_to_remove (List) – list of symbols to remove from blacklisted symbols

• nodes (List, optional) – list of nodes to remove blacklisted symbols on. Defaults to None (all nodes).

Raises:

AgentQueryError – Error in Cognitive Processor handling

Returns:

‘removed-blacklisted-symbols’ or ‘failed-to-remove-some-symbols-from-blacklist’

Return type:

str

remove_patterns_from_system(symbols_list: List, nodes: List = None) → dict | Any

Function which search the kbs on the nodes for models with certain regex patterns. Removed symbols that matches patterns, but does not blacklist them. So patterns may show up in the future.

Parameters:

• symbols_list (List) – list of regex symbol patterns to remove.

• nodes (List, optional) – list of nodes to remove symbols on. Defaults to None (all nodes).

Raises:

AgentQueryError – Error in Cognitive Processor handling

Returns:

dict showing old model hash -> new model hash

(or ‘deleted’) pairs

Return type:

Union[dict, Any]

Example

from ia.gaius.kb_ops import list_symbols

symbols_before = list_symbols(agent)
agent.remove_patterns_from_system(['VECTOR|.+'])
symbols_after = list_symbols(agent)

remove_symbols_from_system(symbols_list: List, nodes: List = None) → dict | Any

Function which search the kbs on the nodes for models with certain symbols.

Symbols are removed, and the symbols are blacklisted, thus making it impossible fo them to be in the working memory in the future.

Parameters:

• symbols_list (List) – list of symbols to remove. Defaults to None.

• nodes (List, optional) – list of nodes to remove symbols from.

• None (Defaults to) –

Raises:

AgentQueryError – Error in Cognitive Processor handling

Returns:

dict showing old model hash -> new model hash (or ‘deleted’) pairs

Return type:

Union[dict, Any]

resolve_model(model_name: str, nodes: List = None) → dict | Any

Returns model with name model_name.

Model name is unique, so it should not matter that we query all nodes, only one model will be found.

set_hypotheses(hypotheses, nodes: List = None) → dict | Any

Function to explicitely set the hypotheses ranges for the specific node

set_ingress_nodes(nodes: List | None = None) → List

List of primitive names to define where data will be sent.

Example

# when observing data, it will be sent to only P1 by default
agent.set_ingress_nodes(nodes=["P1"])

set_query_nodes(nodes: List | None = None) → List

List of primitive names to define which nodes should return answers.

Example

# when getting predictions, it will be received only from P1 by default
agent.set_query_nodes(nodes=["P1"])

set_summarize_for_single_node(value: bool)

When True, queries against a single node return responses directly instead of in a dict key.

Example

# set summarize_for_single_node to True
>>> agent.set_summarize_for_single_node(value=True)
>>> gdf={'strings':['hello'], 'vectors': [], 'emotives':{}}
>>> agent.observe(gdf)
'observed'

# set summarize_for_single_node to False
>>> agent.set_summarize_for_single_node(value=False)
>>> agent.observe(gdf)
{'P1': 'observed'}

set_target_class(target_class: str, nodes: List = None) → dict | Tuple[dict, str]

Provide a target_class symbol for the searcher to look for.

The searcher will ignore all other classes. This is a symbol that is in the last event of a classification sequence.

Example

agent.set_target_class(target_class="apple", nodes=["P1"])

Note

This will result in the nodes specified only predicting on models that include the target class

set_timeout(timeout: float)

Redefine timeout for AgentClient API Calls. Requires re-instantiating requests.Session object

Parameters:

timeout (float) – The requested timeout (in seconds)

show_status(nodes: List | None = None) → dict | Tuple[dict, str]

Return the current status of the agent.

Example

>>> agent.show_status(nodes=['P1'])
{'PREDICT': True,
'SLEEPING': False,
'emotives': {},
'last_learned_model_name': '',
'models_kb': '{KB| objects: 0}',
'name': 'P1',
'size_WM': 0,
'target': '',
'time': 0,
'vectors_kb': '{KB| objects: 0}'}

start_autolearning(nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to start predicting.

Example

# will enable prediction generation on node P1
agent.start_predicting(nodes=["P1"])

start_predicting(nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to start predicting.

Example

# will enable prediction generation on node P1
agent.start_predicting(nodes=["P1"])

start_sleeping(nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to start sleeping.

Example

# will disable observing new data on node P1
agent.start_sleeping(nodes=["P1"])

stop_autolearning(nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to stop predicting.

Useful for faster training, but abstracted nodes will not learn.

Example

# will disable prediction generation on node P1
agent.stop_predicting(nodes=["P1"])

stop_predicting(nodes: List | None = None) → dict | Tuple[dict, str]

Tells nodes to stop predicting.

Useful for faster training, but abstracted nodes will not learn.

Example

# will disable prediction generation on node P1
agent.stop_predicting(nodes=["P1"])

stop_sleeping(nodes: List | None = None) → dict | Tuple[dict, str]

Wakes up sleeping nodes.

Example

# will enable observing new data on node P1
agent.stop_sleeping(nodes=["P1"])

switch_to_clusters(nodes: List = None) → dict | Tuple[dict, str]

Switch CognitiveProcessor to process vectors in clusters rather than by nearest_vectors

Example

>>> agent.switch_to_clusters(nodes=['P1'])
'snapshots-cleared'

update_model(model_name: str, model: Dict, nodes: List = None) → dict | Any

Update the model model_name with the data in model on update_model nodes

Note

This function is only capable of updating the emotives corresponding to a model. In order to update the sequence corresponding to a model, delete this model and reobserve the updated sequence

Example

# retrieve a model already learned on an agent
model_name = 'MODEL|d85f28ebda228064299035bf2dbf58f5c6484108'
model = agent.get_model(model_name=model_name, nodes=['P1'])

# update the emotives in the model
model["emotives"] = {'utility': [150, 200, 500]}
agent.update_model(model_name=model_name, model=model, nodes=['P1'])