Main classes

Agentic · Source · AgenticGraph · MultiGraph · Formalizer · EvidenceGraph · Agent · Endpoint

mercury.graph.evidence.Agentic(my_class, schema=None, endpoint=None, logger=None)

Bases: ABC

This is the parent of any class that is called by an Agent including the Agent itself.

Overview

This class provides the layer that connects tools with agents. It is not a protocol but a method to discover a protocol the class itself implements and validates. It takes inspiration from the Model Context Protocol (MCP) while being lighter, oriented towards classes that coexist within the same process although they can also represent remote services.

It provides a simple interface with four main methods:

• meta for the object's metadata: What the class can do, what input it expects and what output it produces, what state the object is in, ...
• run for the actual execution of a "query", i.e., the request is a valid dictionary created according to the meta.
• dry_run for simulating the execution of a query, without actually running it. This validates the input and returns fast and descriptive feedback on errors.
• pilot for piloting the object to a desired state. Typically, an Agentic object must be in a READY state (after having set up services, loaded data, etc.) before it can accept queries. The pilot() method "drives" the object to a desired state.

Parts of a Agentic object

ID

Architecturally, all Agentic descendants form a graph inside some Endpoint. Each Agentic has an ID that is composed of the IDs of its parents, and of its own name joined by a /. Its name has the format: class_schema (schema is optional if there is only one instance of the class). Agentics can use other Agentics within the same Endpoint, but they must require specific ids in the architecture. When that requirement can be satisfied, the Endpoint will provide the Agentic with the tool by calling its add_tool method.

Capabilities

Think of capabilities as calling tools. Each tool has a unique function name, it expects input and returns output. That is specified in the object's .meta. There is a key called "capabilities" which is a list of dictionaries in the format: "name": {"description": ..., "parameters": ..., "returns": ...}. The "parameters" is a dictionary with the format: {"type": "object", "properties": ..., "required": [-- the names of the required properties --]}. Each property is a dictionary with the format: {"name": {"type": "...", "description": "..."}}. "returns" is a dictionary with the same format as "parameters" except it does not have a "required" key.

State

State is managed internally and exposed in the .meta attribute as the key "state" containing an integer number. Additionally, the number can have text descriptions in the .states attribute which is an Enum class. The name of the state can be obtained using the method state_name(). By convention, negative numbers represent non recoverable errors, zero is the initial state, and positive are sorted up to 100 which is the READY state. The states 1..99 represent intermediate states that are specific to each class.

Intent and piloting

Intent is a desired state for the object. Piloting is the process of taking the object to a desired state. This is done typically using the mge cli. An Agentic that is "always ready" does not need to define its own pilot() method. The mge will pilot a complete Endpoint and the Endpoint will pilot its Agentics. A class that overrides the pilot() method must set the state according to the success or failure of the piloting process.

Running queries

This is done using the run() and dry_run() methods. Their arguments are identical, but their logic and return values are different.

In all cases, the dictionary defines a capability (see Agentic.run() for details.) that must contain a valid function name in the object's capabilities.

In the case of a non-AI Agentic, the capability will be used to forward the request to the appropriate method. In the case of an Agent, the capability provides the function's description in the message list. In the case of an Endpoint, the capability is used to find who will handle the request.

The run() method executes and raises exceptions on errors. The dry_run() checks the request and the state of the object and returns a dictionary with a status code and a description. (See the docstrings of the methods for details.)

Closing the Agentic

The agentic has a method close() that is called just once when the Endpoint is closing. The Agentic can track its own state to know if it was modified and is informed by the Endpoint if it was locked for writing. The Endpoint is locked during the pilot and serve phases. It that case, if the Agentic was modified, it should persist its state to disk or a database. (see the docstring of close() for details.)

Validation and Debugging

The descendants are responsible for validating the input and returning/logging errors. Additionally, they can use the method log_error() to provide further details via the logger.

Logger

The class can log events, errors, and other function calls and responses. The logger is optional can be used for debugging and can be as simple as a python list. I must provide an append() method to add new events. A custom method can filter events or add extra fields to the event.

API:

Attributes:

Name	Type	Description
id	str	the ID of the Agentic, composed of the IDs of its endpoint, class, its own name and an optional schema.
logger		the logger to use for logging events. It must provide an append() method to add new events. It is optional.
endpoint	Agentic	the Endpoint at the root of the architecture.
tools	dict	a dictionary of the Agentics it can use as tools, keyed by their IDs.
states	Enum	an optional Enum class that defines names for the states of an Agentic. It is used to improve readability and cli argument parsing.

Parameters:

Name	Type	Description	Default
my_class	str	the name of the class, used to build the ID. It must ba a string of letters, numbers, and underscores. Typically, it is the name of the class in lowercase.	required
schema	str	an optional string to distinguish different instances of the same class. It is a schema (like a database, a graph, ontology, etc.) that the class is serving.	None
endpoint	str	an optional endpoint Agentic. If not provided, the Agentic is the Endpoint itself.	None
logger	list	an optional logger. If not provided, no logging will be done.	None

Source code in mercury/graph/evidence/agentic.py

def __init__(self, my_class, schema = None, endpoint = None, logger = None):
	self.id		  = my_class
	self.logger	  = logger
	self.seq_num  = 0
	self.endpoint = self
	self.tools	  = {}
	self.states	  = AlwaysReadyState
	self._meta_	  = None

	if schema is not None:
		self.id += '_' + schema

	if endpoint is not None:
		self.id		  = endpoint.id + '/' + self.id
		self.endpoint = endpoint

meta property

This the object's metadata as a dictionary.

It is cached after the first call, but classes can modify the .meta dictionary directly to change the metadata.

Returns:

Type	Description
dict	the metadata of the object. This contains, at the minimum, the keys "state" (see pilot()) and "capabilities" (see Agent).

_dry_run(request) abstractmethod

This is the method that simulates the execution of a query. It MUST be implemented by the descendants.

The method Agentic.dry_run() is a wrapper that logs the request and response, this method does the actual work.

Source code in mercury/graph/evidence/agentic.py

@abstractmethod
def _dry_run(self, request):
	""" This is the method that simulates the execution of a query. It MUST be implemented by the descendants.

	The method [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run] is a wrapper that logs the request and response, this
	method does the actual work.
	"""

	pass

_meta() abstractmethod

This is the method that returns the metadata of the class. It MUST be implemented by the descendants.

Source code in mercury/graph/evidence/agentic.py

@abstractmethod
def _meta(self):
	""" This is the method that returns the metadata of the class. It MUST be implemented by the descendants. """

	pass

_normalize_name(name) staticmethod

Normalizes a name to be used as an ID.

That replaces spaces with underscores and removes any character that is not a letter, number, or underscore.

Parameters:

Name	Type	Description	Default
name	str	the name to normalize.	required

Returns:

Type	Description
str	the normalized name.

Source code in mercury/graph/evidence/agentic.py

@staticmethod
def _normalize_name(name):
	""" Normalizes a name to be used as an ID.

	That replaces spaces with underscores and removes any character that is not a letter, number, or underscore.

	Args:
		name (str): the name to normalize.

	Returns:
		(str): the normalized name.
	"""

	name = name.replace(' ', '_')
	name = re.sub('[^a-zA-Z0-9_]', '', name)

	return name

_now() staticmethod

Returns the current time as a formatted string.

Returns:

Type	Description
str	the current time as %Y-%m-%d %H:%M:%S.

Source code in mercury/graph/evidence/agentic.py

@staticmethod
def _now():
	""" Returns the current time as a formatted string.

	Returns:
		(str): the current time as %Y-%m-%d %H:%M:%S.
	"""

	return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

_run(request) abstractmethod

This is the method that actually runs the query. It MUST be implemented by the descendants.

The method Agentic.run() is a wrapper that logs the request and response, this method does the actual work.

Source code in mercury/graph/evidence/agentic.py

@abstractmethod
def _run(self, request):
	""" This is the method that actually runs the query. It MUST be implemented by the descendants.

	The method [`Agentic.run()`][mercury.graph.evidence.Agentic.run] is a wrapper that logs the request and response, this method
	does the actual work.
	"""

	pass

add_tool(agentic)

Adds a tool (another Agentic this one can use) to the Agentic.

Parameters:

Name	Type	Description	Default
agentic	Agentic	the tool to add. It must be an Agentic and its endpoint must be the same as the current Agentic's.	required

Source code in mercury/graph/evidence/agentic.py

def add_tool(self, agentic):
	""" Adds a tool (another Agentic this one can use) to the Agentic.

	Args:
		agentic (Agentic): the tool to add. It must be an Agentic and its endpoint must be the same as the current Agentic's.
	"""

	self.tools[agentic.id] = agentic

close(endpoint_locked)

Closes the Agentic.

This method is called just once when the Endpoint is closing. The Agentic can track its own state to know if it was modified and is informed by the Endpoint if it was locked for writing. The Endpoint is locked during the pilot and serve phases. It that case, if the Agentic was modified, it should persist its state to disk or a database.

Parameters:

Name	Type	Description	Default
endpoint_locked	bool	True if the Endpoint is locked for writing, False otherwise.	required

Source code in mercury/graph/evidence/agentic.py

def close(self, endpoint_locked):
	""" Closes the Agentic.

	This method is called just once when the Endpoint is closing. The Agentic can track its own state to know if it was modified and
	is informed by the Endpoint if it was locked for writing. The Endpoint is locked during the pilot and serve phases. It that case,
	if the Agentic was modified, it should persist its state to disk or a database.

	Args:
		endpoint_locked (bool): True if the Endpoint is locked for writing, False otherwise.
	"""

	pass

dry_run(request)

Simulates the execution of a query.

The request is a valid argument identical to the one expected by Agentic.run().

This method does not raise exceptions, it provides feedback. It should not just validate the request, but also anticipate the readiness of the Agentic whenever possible to prevent an AgenticRunInvalidState on run().

The return value is a

Parameters:

Name	Type	Description	Default
request	dict	the request to simulate.	required

Returns:

Type	Description
dict	{'status': 0, 'description': 'Ok.'} dictionary. The status is 0 for success. 1 for busy, 2 for invalid request with an appropriate description.

Source code in mercury/graph/evidence/agentic.py

def dry_run(self, request):
	""" Simulates the execution of a query.

	The request is a valid argument identical to the one expected by [`Agentic.run()`][mercury.graph.evidence.Agentic.run].

	This method does not raise exceptions, it provides feedback. It should not just validate the request, but also anticipate
	the readiness of the Agentic whenever possible to prevent an AgenticRunInvalidState on run().

	The return value is a

	Args:
		request (dict): the request to simulate.

	Returns:
		(dict): {'status': 0, 'description': 'Ok.'} dictionary. The status is 0 for success. 1 for busy, 2 for invalid
			request with an appropriate description.
	"""

	return self._dry_run(request)

log_error(message)

Logs an error message.

The class can use this to introduce events in the logger.

Parameters:

Name	Type	Description	Default
message	str	the error message to log.	required

Source code in mercury/graph/evidence/agentic.py

def log_error(self, message):
	""" Logs an error message.

	The class can use this to introduce events in the logger.

	Args:
		message (str): the error message to log.
	"""

	if self.logger is not None:
		event = {'type': 'error', 'timestamp': self._now(), 'id': self.id, 'seq_num': self.seq_num, 'error': message}

		self.logger.append(event)
		self.seq_num += 1

pilot(intent, just_once=False)

Pilots the object to a desired state.

In the parent class, this method returns the object as AlwaysReadyState.READY unless it is in an error (negative) state. The classes that need piloting must override it.

(See Intent and piloting for details.)

Parameters:

Name	Type	Description	Default
intent	str	the desired state to pilot to. It must be a valid state name in the object's meta.	required
just_once	bool	An optional argument to break without necessarily reaching the desired state after completing one iteration. The iteration is Agentic specific and can be anything, (E.g., a complete file chunked and processed, ...) That parameter is intended for when piloting can take hours or days and the user wants a finer control of the process.	False

Returns:

Type	Description
bool	True if the object is in the desired state (or advanced towards it one step when just_once is True), False otherwise.

Source code in mercury/graph/evidence/agentic.py

def pilot(self, intent, just_once = False):
	""" Pilots the object to a desired state.

	In the parent class, this method returns the object as AlwaysReadyState.READY unless it is in an error (negative) state.
	The classes that need piloting must override it.

	(See [`Intent and piloting`][mercury.graph.evidence.Agentic] for details.)

	Args:
		intent (str): the desired state to pilot to. It must be a valid state name in the object's meta.
		just_once (bool): An optional argument to break without necessarily reaching the desired state after completing one iteration.
			The iteration is Agentic specific and can be anything, (E.g., a complete file chunked and processed, ...) That parameter is
			intended for when piloting can take hours or days and the user wants a finer control of the process.

	Returns:
		(bool): True if the object is in the desired state (or advanced towards it one step when just_once is True), False otherwise.
	"""

	if self._meta_ is None:
		self._meta_ = self._meta()

	if self._meta_['state'] >= 0:
		self._meta_['state'] = AlwaysReadyState.READY.value

run(request)

Runs a query.

Expected input

The request is a valid (ChatGPT/litellm function call) that can be any of the following (from simplest to most complex):

1. A pure function call {"arguments": (mandatory), "name": (mandatory), "id": optional, "type": "function" (not used)}
2. A unique message {"content": (mandatory), "role": (mandatory), "id": optional, "tool_calls": optional}
3. A list of messages [ ... ]

NOTES:

1. When the Agentic has multiple capabilities, the request must be a function call since the capability is the function name.
2. When an id is provided, it must be returned in the response. It is used to match requests and responses in history.

Expected output

This method does not provide user feedback on errors, other than raising exceptions that are descendants of AgenticRunException (e.g., AgenticRunInvalidRequest, AgenticRunInvalidState, AgenticRunFailed). Use the dry_run() method to validate the request.

The returned value does not provide any status key, it just assumes the request was successful.

The returned value is a dictionary similar to a response.choices[0] in the OpenAI API, with these keys:

• "finish_reason": A string in {'stop', 'tool_calls', 'error'}. (Anything is an ending condition, unless its lower case form starts with 'tool' or 'error'.) Errors are just a "normal" ending that will be highlighted and the content will typically be an error message. Tool calls are a special condition that will be handled by the Endpoint. Only Agentics with a single capability can use tool calls, since the Endpoint will call them with the result in a list of messages.
• "message": A dictionary with {"content" (mandatory), "role" (optional), "id" (for tool calls), "tool_calls" (optional)}}

Parameters:

Name	Type	Description	Default
request	dict	the request to run.	required

Returns:

Type	Description
dict	the response to the request.

Source code in mercury/graph/evidence/agentic.py

def run(self, request):
	""" Runs a query.

	## Expected input

	The request is a valid (ChatGPT/litellm function call) that can be any of the following (from simplest to most complex):

	1. A pure function call {"arguments": (mandatory), "name": (mandatory), "id": optional, "type": "function" (not used)}
	2. A unique message {"content": (mandatory), "role": (mandatory), "id": optional, "tool_calls": optional}
	3. A list of messages [ ... ]

	###NOTES:
	1. When the Agentic has multiple capabilities, the request must be a function call since the capability is the function name.
	2. When an id is provided, it must be returned in the response. It is used to match requests and responses in history.

	## Expected output

	This method does not provide user feedback on errors, other than raising exceptions that are descendants of AgenticRunException
	(e.g., AgenticRunInvalidRequest, AgenticRunInvalidState, AgenticRunFailed). Use the `dry_run()` method to validate the request.

	The returned value does not provide any status key, it just assumes the request was successful.

	The returned value is a dictionary similar to a response.choices[0] in the OpenAI API, with these keys:

	* "finish_reason": A string in {'stop', 'tool_calls', 'error'}. (Anything is an ending condition, unless its lower case form starts
		with 'tool' or 'error'.) Errors are just a "normal" ending that will be highlighted and the content will typically be an error
		message. Tool calls are a special condition that will be handled by the Endpoint. Only Agentics with a single capability can
		use tool calls, since the Endpoint will call them with the result in a list of messages.
	* "message": A dictionary with {"content" (mandatory), "role" (optional), "id" (for tool calls), "tool_calls" (optional)}}

	Args:
		request (dict): the request to run.

	Returns:
		(dict): the response to the request.
	"""

	if self.logger is not None:
		event = {'type': 'request', 'timestamp': self._now(), 'id': self.id, 'seq_num': self.seq_num, 'request': request}

		self.logger.append(event)

	ret = self._run(request)

	if self.logger is not None:
		event = {'type': 'response', 'timestamp': self._now(), 'id': self.id, 'seq_num': self.seq_num, 'response': ret}

		self.logger.append(event)
		self.seq_num += 1

	return ret

state_name(state)

Returns the name of a state given its integer value.

Parameters:

Name	Type	Description	Default
state	int	the integer value of the state.	required

Returns:

Type	Description
str	the name of the state, or None if the state is not defined in the object's states Enum.

Source code in mercury/graph/evidence/agentic.py

def state_name(self, state):
	""" Returns the name of a state given its integer value.

	Args:
		state (int): the integer value of the state.

	Returns:
		(str): the name of the state, or None if the state is not defined in the object's states Enum.
	"""

	if self.states is not None:
		try:
			ret = self.states(int(state)).name

		except ValueError:
			ret = None

		return ret

mercury.graph.evidence.Source(schema, extra_args, endpoint=None, logger=None)

Bases: Agentic

The Source is an Agentic interface to a corpus of documents.

Overview

The Source can contain different types of documents and possibly call format conversion tools. Documents can be text or code.

The Source provides:

• A "chunking" interface to break documents into smaller pieces for easier processing.
• A hierarchy that divides a corpus into: a collection (that manages many files), a document (a file) a section (which can be nested).
• An indexing system that provides unique identifiers for each component.
• A persistence backend that possibly includes vectorization and embedding of the chunks for later retrieval.
• An Agentic interface to everything above.

Source Components

The Source uses the following components to manage file conversion, chunking, indexing and to represent the parts of a document:

• SourceNode: The base class to manage the index logic of all components.
• SourceMaker: The root SourceNode responsible for managing a tree of markdown files.
• SourceFile: Each individual file as a SourceNode.
• SourceEntity: Each section, subsection, paragraph, table, figure, text, table cell or link in a markdown file as a SourceNode.

Known Limitations

See

• Warning
• Limitations
• The SourceNode tree (all the SourceEntity objects inside each SourceFile, all the SourceFiles inside the SourceMaker created on demand) can become very large and the Source has yet not mechanisms to limit the size of the tree. We postpone until a later release exploring how to do that to make Sources production ready. For now, Sources typically load very fast since everything is stored in files and can be re-loaded when they grow too much.
• The vector database is configured, created by the Source, and not used. This requires exposing new capabilities by the Source. It is postponed until a later release, since it is not part of the MVP.

Parameters:

Name	Type	Description	Default
schema	str	a schema (a unique name) to use for the Source's ID.	required
endpoint	Agentic	an optional Endpoint. It becomes part of the Source's ID and is available via self.endpoint. If not provided, the Source becomes its own Endpoint.	None
logger	list	an optional logger to use for logging events. It must provide an append() method to add new events.	None
extra_args	dict	the configuration for the Source.	required

Source code in mercury/graph/evidence/source.py

def __init__(self, schema, extra_args, endpoint = None, logger = None):
	super().__init__(my_class = 'source', schema = schema, endpoint = endpoint, logger = logger)

	if schema is None:			# This allows finding out the class name (to link tools) without actually instantiating a full object.
		return

	self.states = SourceState

	self.conf = extra_args
	self.name = schema

	self._maker	 = None
	self._chroma = None

	self._meta_	 = self._meta()	# Just to make .meta reflect the initial state.

_capabilities()

Returns the capabilities of the Source.

Returns:

Type

Description

list

A list of capabilities, each represented as a dictionary with the following keys: 'type': The type of capability (e.g., 'function'). 'function': A dictionary containing details about the function The value of 'function' is: 'name': The name of the function. 'description': A brief description of what the function does. 'parameters': A dictionary with 'type', 'properties', and 'required' 'returns': A dictionary with 'type' and 'items'

Source code in mercury/graph/evidence/source.py

def _capabilities(self):
	""" Returns the capabilities of the Source.

	Returns:
		(list): A list of capabilities, each represented as a dictionary with the following keys:

			- 'type': The type of capability (e.g., 'function').
			- 'function': A dictionary containing details about the function

			The value of 'function' is:

			* 'name': The name of the function.
			* 'description': A brief description of what the function does.
			* 'parameters': A dictionary with 'type', 'properties', and 'required'
			* 'returns': A dictionary with 'type' and 'items'
	"""

	name_get_children_idx = 'children_by_idx_%s' % self.name
	name_child			  = 'object_by_idx_%s' % self.name

	self.call = {name_get_children_idx: self.get_children_idx, name_child: self.child}

	return [
		{
			'type': 'function',
			'function': {
				'name': name_get_children_idx,
				'description': 'Get indices of the children of an index. Indices are either sources, files, sections or chunks.',
				'parameters': {
					'type': 'object',
					'properties': {
						'index': {
							'type': 'string',
							'description': 'Index whose children indices are required.'
						}
					},
					'required': ['index']
				},
				'returns': {
					'type': 'array',
					'items': {
						'type': 'string'
					}
				}
			}
		},
		{
			'type': 'function',
			'function': {
				'name': name_child,
				'description': 'Get the text at a given index.',
				'parameters': {
					'type': 'object',
					'properties': {
						'index': {
							'type': 'string',
							'description': 'Index of the text component.'
						}
					},
					'required': ['index']
				},
				'returns': {
					'type': 'dict'
				}
			}
		}
	]

_dry_run(request)

Simulates running the Source with the given request.

(See Agentic.dry_run().)

NOTE:

The Endpoint takes care of validating the request according to the capabilities exposed by the Source. It is not necessary to validate again here and the Endpoint does not forward the dry_run() request to the Source. This method is provided as a requirement of the Agentic interface, but it is only used when you use Sources directly outside of an Endpoint.

Source code in mercury/graph/evidence/source.py

def _dry_run(self, request):
	""" Simulates running the Source with the given request.

	(See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

	## NOTE:

	The Endpoint takes care of validating the request according to the capabilities exposed by the Source. It is not necessary to
	validate again here and the Endpoint does not forward the dry_run() request to the Source. This method is provided as a
	requirement of the Agentic interface, but it is only used when you use Sources directly outside of an Endpoint.
	"""

	return {'status': 0, 'description': 'Valid request.'}

_meta()

Returns the metadata of the Source.

(See Agentic.meta().)

Source code in mercury/graph/evidence/source.py

def _meta(self):
	""" Returns the metadata of the Source.

	(See [`Agentic.meta()`][mercury.graph.evidence.Agentic.meta].)
	"""

	meta = {}
	meta['state'] = SourceState.INITIAL.value

	meta['description'] = self.conf.get('description', '')
	if type(meta['description']) is list:
		meta['description'] = '\n'.join(meta['description'])

	meta['capabilities'] = self._capabilities()

	return meta

_run(request)

Runs the Source with the given request.

(See Agentic.run().)

Source code in mercury/graph/evidence/source.py

def _run(self, request):
	""" Runs the Source with the given request.

	(See [`Agentic.run()`][mercury.graph.evidence.Agentic.run].)
	"""

	call = self.call.get(request['name'], None)

	if call is None:
		self.log_error('Source does not have a function named "%s".' % request['name'])
		raise AgenticRunInvalidRequest

	index = request['arguments'].get('index', None)

	if index is None:
		self.log_error('Source function "%s" requires an "index" argument.' % request['name'])
		raise AgenticRunInvalidRequest

	ret = {'finish_reason': 'stop', 'message': call(index)}

	return ret

_setup_chroma_db()

Sets up the Chroma vector database for the Source.

The Chroma vector database is used to store embeddings of chunks for later retrieval. It is configured in the Source's configuration.

Returns:

Type	Description
bool	True if the Chroma vector database was set up successfully or is not used, False if setup failed.

Source code in mercury/graph/evidence/source.py

def _setup_chroma_db(self):
	""" Sets up the Chroma vector database for the Source.

	The Chroma vector database is used to store embeddings of chunks for later retrieval. It is configured in the Source's
	configuration.

	Returns:
		(bool): True if the Chroma vector database was set up successfully or is not used, False if setup failed.
	"""

	self._chroma = None

	chroma_path = self.conf.get('chroma_path', None)

	if chroma_path is None:
		return True			# This is the neat way to disable ChromaDB.

	try:
		self._chroma = chroma.PersistentClient(path = chroma_path)

	except:
		self.log_error('Source failed to create Chroma client at path: "%s".' % (chroma_path))

		return False

	try:
		name = self.conf['chroma_descriptions_collection_name']

		self._chroma_descr = self._chroma.get_or_create_collection(name)

		name = self.conf['chroma_chunks_collection_name']

		self._chroma_chunks = self._chroma.get_or_create_collection(name)

	except:
		self.log_error('Source failed to create Chroma collections at path: "%s".' % (chroma_path))

		return False

	return True

child(index)

Returns the corresponding SourceNode object following the SourceNode interface and serializes it to a dictionary.

(See SourceNode.child().)

Source code in mercury/graph/evidence/source.py

def child(self, index):
	""" Returns the corresponding SourceNode object following the SourceNode interface and serializes it to a dictionary.

	(See [`SourceNode.child()`][mercury.graph.evidence.source_parts.SourceNode.child].)
	"""

	if (   self._maker is None
		or self._maker.state != self.states.MAKER_READY_OK.value
		or self._meta_['state'] < self.states.READY.value):

		self.log_error('Source is not ready for child("%s").' % index)

		return None

	child = self._maker.child(index)

	if child is None:
		return None

	if type(child) is str:					# The SourceMaker returned the index of a SourceFile that understands the index.
		child = self._maker.child(child)
		child = child.child(index)

		if child is None:
			return None

	if type(child) is not SourceEntity:
		return {'type': 'object', 'class': str(type(child)), 'description': child.description}

	if child._children is None:
		return {'type': str(child.entity_type), 'content': child.content}

	return {'type': 'SourceEntity: %s' % child.entity_type, 'description': child.description}

close(endpoint_locked)

Closes the Source and releases any resources it holds.

(See Agentic.close().)

Source code in mercury/graph/evidence/source.py

def close(self, endpoint_locked):
	""" Closes the Source and releases any resources it holds.

	(See [`Agentic.close()`][mercury.graph.evidence.Agentic.close].)
	"""

	self._chroma = None		# There is no need to explicitly .close(), .flush() ... That persists changes.
	self._maker	 = None

get_children_idx(index=None)

Returns the children indices following the SourceNode interface.

(See SourceNode.get_children_idx().)

Source code in mercury/graph/evidence/source.py

def get_children_idx(self, index = None):
	""" Returns the children indices following the SourceNode interface.

	(See [`SourceNode.get_children_idx()`][mercury.graph.evidence.source_parts.SourceNode.get_children_idx].)
	"""

	if (   self._maker is None
		or self._maker.state != self.states.MAKER_READY_OK.value
		or self._meta_['state'] < self.states.READY.value):

		self.log_error('Source is not ready for get_children_idx("%s").' % index)

		return None

	if index is None or index == '':
		index = self._maker.index

	ret = self._maker.get_children_idx(index)

	if type(ret) is list or ret is None:		# The SourceMaker provided the children indices or an error.
		return ret

	# Now, ret is a string that is the index of the SourceFile.

	file = self._maker.child(ret)
	if type(file) is not SourceFile:
		self.log_error('SourceMaker could not find a SourceFile for index "%s".' % index)
		return None

	return file.get_children_idx(index)

pilot(intent, just_once=False)

Pilots the Source to a new state based on the given intent.

(See Agentic.pilot().)

Source code in mercury/graph/evidence/source.py

def pilot(self, intent, just_once = False):
	""" Pilots the Source to a new state based on the given intent.

	(See [`Agentic.pilot()`][mercury.graph.evidence.Agentic.pilot].)
	"""

	if self.meta['state'] < 0:
		self.log_error('Source is in error state %d' % self._meta_['state'])
		return

	while self._meta_['state'] < intent:
		if self._meta_['state'] == self.states.INITIAL.value:
			try:
				typ = self.conf['type']
				src = self.conf['src_path']
				dst = self.conf['dst_path']
				siz = self.conf.get('cluster_size', 256)
				ext = self.conf.get('extensions', None)
				pdf = self.conf.get('pdf_to_markdown', None)
				self._maker = SourceMaker(self.name, typ, src, dst, siz, ext, pdf)

			except:
				self.log_error('SourceMaker could not be created and initialized for Source "%s".' % self.name)
				self._meta_['state'] = self.states.ERR_MAKER_INIT.value
				break

			self._meta_['state'] = self.states.MAKER_INIT_OK.value

			if just_once:
				break

		if self._meta_['state'] == self.states.MAKER_INIT_OK.value:
			if self._maker.build_indices():
				self._meta_['state'] = self.states.MAKER_READY_OK.value
			else:
				self.log_error('SourceMaker could not build indices for Source "%s".' % self.name)
				self._meta_['state'] = self.states.ERR_MAKER_INDEX.value
				break

			if just_once:
				break

		if self._meta_['state'] == self.states.MAKER_READY_OK.value:
			if self._setup_chroma_db():
				self._meta_['state'] = self.states.READY.value

			else:
				self.log_error('Source could not setup the vector database for Source "%s".' % self.name)
				self._meta_['state'] = self.states.ERR_DB_SETUP.value

			break

mercury.graph.evidence.AgenticGraph(schema, extra_args, endpoint=None, logger=None)

Bases: Agentic

AgenticGraph is the class that exposes a mercury.graph.Graph using the Agentic interface.

AgenticGraphs are typically persisted

A graph is a persisted storage of anything: the storage of nodes is a key/value store with any properties. On top of that, nodes can be connected by edges that also have properties. AgenticGraph objects are typically persisted (that is part of their configuration to store such things as hierarchical ontologies for both entities and relationships). They can read from .csv files that contain ontologies, for initialization, but do not write to .csv files to keep project management simple. Once the graph is initialized, it stores itself as a pickle file. When initializing the object, if the pickle file exists, it will be loaded rather than initializing the graph from .csv files.

IDs and hierarchical structure

The ontology is not only the place where concepts and relationships are defined to be used by the formalizer, but also the place where the IDs of every node and edge in the EvidenceGraph live. Despite the number of entities being potentially very large and AgenticGraphs typically living in RAM, the importance of creating concepts and maintaining the hierarchical structure dynamically from text by Agentic decisions makes this "all in one place" design recommendable. Also, the mechanism used to structure text entities using SourceNode is exactly the mechanism used in the hierarchy of the IDs.

How AgenticGraphs are used to build EvidenceGraphs

There a three components:

• The Ontologies are just graphs with IDs and definitions for concepts. Typically entities and relationships live in different graphs (relationships may not be used, all that is configurable).
• The Formalizer takes concepts defined in the Ontology (E.g., person, product, country, etc.) also possibly relationships (E.g., is_owned_by, is_located_in, etc.) and finds instances of those concepts in natural language text.
• The EvidenceGraph creates an aggregated graph of all that information handling contradictions, reinforcements and confidence.

What the AgenticGraph exposes via the Agentic interface

All the internals of a graph are basically understood by the Formalizer and the EvidenceGraph. The AgenticGraph is an understandable storage for concepts, entity indices and relationships and, via the Agentic interface that is what is accessible: A mechanism similar to that of a source to retrieve, via unique indices, concepts, entities and relationships.

Known Limitations

There is no "graph language" exposed via the Agentic interface that allows for arbitrary graph oriented queries. Note that Agentic objects have a mechanism to access the objects directly and will typically bypass the Agentic interface for intense computations such as crawling entire corpora. In the future, Agents, besides discovering the hierarchy tree may also have to modify it by creating or updating concepts and relationships.

Parameters:

Name	Type	Description	Default
schema	str	a schema (a unique name) to use for the AgenticGraph's ID.	required
extra_args	dict	the configuration for the AgenticGraph. See the configuration for the examples in ontologies.jsonc for reference on how to configure the AgenticGraph's persistence.	required
endpoint	Agentic	an optional Endpoint. It becomes part of the AgenticGraph's ID and is available via self.endpoint. If not provided, the AgenticGraph becomes its own Endpoint.	None
logger	list	an optional logger to use for logging events. It must provide an append() method to add new events.	None

Source code in mercury/graph/evidence/agentic_graph.py

def __init__(self, schema, extra_args, endpoint = None, logger = None):
	super().__init__(my_class = 'agentic_graph', schema = schema, endpoint = endpoint, logger = logger)

	if schema is None:			# This allows finding out the class name (to link tools) without actually instantiating a full object.
		return

	self.states = GraphState

	self.conf = extra_args
	self.name = schema

	self._graph	= None

	self._meta_ = self._meta()	# Just to make .meta reflect the initial state.

_add_index_to_tree(index)

Adds an index to the hierarchical tree structure. It traverses the tree rooted at self._indices and adds dictionaries as required. The same tree contains node and edge indices, but the latter have their index prefixed by _edge_ which places them in a dictionary under the _edge_ key.

Parameters:

Name	Type	Description	Default
index	str	The index to add, represented as a string with components separated by '|'.	required

Source code in mercury/graph/evidence/agentic_graph.py

def _add_index_to_tree(self, index):
	""" Adds an index to the hierarchical tree structure. It traverses the tree rooted at self._indices and adds dictionaries
	as required. The same tree contains node and edge indices, but the latter have their index prefixed by `_edge_` which
	places them in a dictionary under the `_edge_` key.

	Args:
		index (str): The index to add, represented as a string with components separated by '|'.
	"""

	ii = index.split('|')
	last = len(ii) - 1

	tree = self._indices

	for i, ix in enumerate(ii):
		if i == last:
			tree[ix] = None

		else:
			if ix not in tree or tree[ix] is None:
				tree[ix] = {}

			tree = tree[ix]

_build_indices()

Builds the indices for the AgenticGraph for the first time after it is loaded.

This builds a hierarchical tree structure of dictionaries, rooted at self._indices.

Source code in mercury/graph/evidence/agentic_graph.py

def _build_indices(self):
	""" Builds the indices for the AgenticGraph for the first time after it is loaded.

	This builds a hierarchical tree structure of dictionaries, rooted at self._indices.
	"""

	self._indices = {}

	for id, _ in self._graph.networkx.nodes.data('id'):
		self._add_index_to_tree(id)

	for src, dst, key in self._graph.networkx.edges.data(keys = True, data = False):
		self._add_index_to_tree('_edge_|%s||%s||%s' % (src, dst, key))

_capabilities()

Returns the capabilities of the AgenticGraph.

Returns:

Type

Description

list

A list of capabilities, each represented as a dictionary with the following keys: 'type': The type of capability (e.g., 'function'). 'function': A dictionary containing details about the function The value of 'function' is: 'name': The name of the function. 'description': A brief description of what the function does. 'parameters': A dictionary with 'type', 'properties', and 'required' 'returns': A dictionary with 'type' and 'items'

Source code in mercury/graph/evidence/agentic_graph.py

def _capabilities(self):
	""" Returns the capabilities of the AgenticGraph.

	Returns:
		(list): A list of capabilities, each represented as a dictionary with the following keys:

			- 'type': The type of capability (e.g., 'function').
			- 'function': A dictionary containing details about the function

			The value of 'function' is:

			* 'name': The name of the function.
			* 'description': A brief description of what the function does.
			* 'parameters': A dictionary with 'type', 'properties', and 'required'
			* 'returns': A dictionary with 'type' and 'items'
	"""

	name_children_idx = 'children_by_idx_%s' % self.name
	name_node_by_idx  = 'node_by_idx_%s' % self.name

	self.call = {name_children_idx: self.get_children_idx, name_node_by_idx: self.child}

	return [
		{
			'type': 'function',
			'function': {
				'name': name_children_idx,
				'description': 'Get indices of the children of an index.',
				'parameters': {
					'type': 'object',
					'properties': {
						'index': {
							'type': 'string',
							'description': 'Index whose children indices are required.'
						}
					},
					'required': ['index']
				},
				'returns': {
					'type': 'array',
					'items': {
						'type': 'string'
					}
				}
			}
		},
		{
			'type': 'function',
			'function': {
				'name': name_node_by_idx,
				'description': 'Get the properties of a node by its index.',
				'parameters': {
					'type': 'object',
					'properties': {
						'index': {
							'type': 'string',
							'description': 'Index of the node.'
						}
					},
					'required': ['index']
				},
				'returns': {
					'type': 'dict'
				}
			}
		}
	]

_dry_run(request)

Simulates running the AgenticGraph with the given request.

(See Agentic.dry_run().)

NOTE:

The Endpoint takes care of validating the request according to the capabilities exposed by the AgenticGraph. It is not necessary to validate again here and the Endpoint does not forward the dry_run() request to the AgenticGraph. This method is provided as a requirement of the Agentic interface, but it is only used when you use AgenticGraphs directly outside of an Endpoint.

Source code in mercury/graph/evidence/agentic_graph.py

def _dry_run(self, request):
	""" Simulates running the AgenticGraph with the given request.

	(See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

	## NOTE:

	The Endpoint takes care of validating the request according to the capabilities exposed by the AgenticGraph. It is not necessary to
	validate again here and the Endpoint does not forward the dry_run() request to the AgenticGraph. This method is provided as a
	requirement of the Agentic interface, but it is only used when you use AgenticGraphs directly outside of an Endpoint.
	"""

	return {'status': 0, 'description': 'Valid request.'}

_meta()

Returns the metadata of the AgenticGraph.

(See Agentic.meta().)

Source code in mercury/graph/evidence/agentic_graph.py

def _meta(self):
	""" Returns the metadata of the AgenticGraph.

		(See [`Agentic.meta()`][mercury.graph.evidence.Agentic.meta].)
	"""

	meta = {}
	meta['state'] = GraphState.INITIAL.value

	meta['description'] = self.conf.get('description', '')
	if type(meta['description']) is list:
		meta['description'] = '\n'.join(meta['description'])

	meta['capabilities'] = self._capabilities()

	return meta

_run(request)

Runs the AgenticGraph with the given request.

(See Agentic.run().)

Source code in mercury/graph/evidence/agentic_graph.py

def _run(self, request):
	""" Runs the AgenticGraph with the given request.

		(See [`Agentic.run()`][mercury.graph.evidence.Agentic.run].)
	"""

	call = self.call.get(request['name'], None)

	if call is None:
		self.log_error('AgenticGraph does not have a function named "%s".' % request['name'])
		raise AgenticRunInvalidRequest

	index = request['arguments'].get('index', None)

	if index is None:
		self.log_error('AgenticGraph function "%s" requires an "index" argument.' % request['name'])
		raise AgenticRunInvalidRequest

	ret = {'finish_reason': 'stop', 'message': call(index)}

	return ret

child(index)

Returns the corresponding SourceNode object following the SourceNode interface and serializes it to a dictionary.

(See SourceNode.child().)

Source code in mercury/graph/evidence/agentic_graph.py

def child(self, index):
	""" Returns the corresponding SourceNode object following the SourceNode interface and serializes it to a dictionary.

	(See [`SourceNode.child()`][mercury.graph.evidence.source_parts.SourceNode.child].)
	"""

	if self._meta_['state'] != self.states.READY.value:
		self.log_error('AgenticGraph is not ready for child.')

		return None

	idx_stack = index.split('|')

	if idx_stack.pop(0) != self.name:
		return None

	d = self._indices
	for i in idx_stack:
		if i not in d:
			return None

		d = d[i]

	ntx = self._graph.networkx

	if idx_stack[0] != '_edge_':
		ix = '|'.join(idx_stack)

		try:
			ret = dict(ntx.nodes(data = True))[ix]

			return ret

		except:
			return None

	else:
		idx_stack.pop(0)

		try:
			src, dst, key = '|'.join(idx_stack).split('||')
			ret = dict(ntx.edges[src, dst, key])

			return ret

		except:
			return None

close(endpoint_locked)

Closes the AgenticGraph, persists it to disk and releases any resources it holds.

(See Agentic.close().)

Source code in mercury/graph/evidence/agentic_graph.py

def close(self, endpoint_locked):
	""" Closes the AgenticGraph, persists it to disk and releases any resources it holds.

	(See [`Agentic.close()`][mercury.graph.evidence.Agentic.close].)
	"""

	if endpoint_locked and self._graph is not None and self._fname is not None:
		ntx = self._graph.networkx
		with open(self._fname, 'wb') as f:
			pickle.dump(ntx, f)

	self._graph = None

get_children_idx(index=None)

Returns the children indices following the SourceNode interface.

(See SourceNode.get_children_idx().)

Source code in mercury/graph/evidence/agentic_graph.py

def get_children_idx(self, index = None):
	""" Returns the children indices following the SourceNode interface.

	(See [`SourceNode.get_children_idx()`][mercury.graph.evidence.source_parts.SourceNode.get_children_idx].)
	"""

	if self._meta_['state'] != self.states.READY.value:
		self.log_error('AgenticGraph is not ready for get_children_idx.')

		return None

	if index is None or index == '':
		index = self.name

	idx_stack = index.split('|')

	if idx_stack.pop(0) != self.name:
		return None

	d = self._indices
	for i in idx_stack:
		if i not in d:
			return None

		d = d[i]

	if type(d) is dict:
		return ['%s|%s' % (index, k) for k in d.keys()]

	return None

pilot(intent, just_once=False)

Pilots the AgenticGraph to a new state based on the given intent.

(See Agentic.pilot().)

Source code in mercury/graph/evidence/agentic_graph.py

def pilot(self, intent, just_once = False):
	""" Pilots the AgenticGraph to a new state based on the given intent.

	(See [`Agentic.pilot()`][mercury.graph.evidence.Agentic.pilot].)
	"""

	def new_graph():
		""" Creates a new graph when there is no persisted graph to load, but possibly initial_nodes and/or initial_edges. """

		keys = self.conf.get('file_format', None)
		if keys is None:
			keys = {'src': 'src', 'dst': 'dst', 'id': 'id', 'directed': True, 'sep': '\t'}

		nodes = None
		fn_nodes = self.conf.get('initial_nodes', None)
		if fn_nodes is not None:
			if fn_nodes['type'] == 'csv':
				sep = ',' if keys is None or 'sep' not in keys else keys['sep']
				nodes = pd.read_csv(fn_nodes['path'], sep = sep)
			elif fn_nodes['type'] == 'pickle':
				nodes = pd.read_pickle(fn_nodes['path'])

		fn_edges = self.conf.get('initial_edges', None)
		if fn_edges is not None:
			if fn_edges['type'] == 'csv':
				sep = ',' if keys is None or 'sep' not in keys else keys['sep']
				edges = pd.read_csv(fn_edges['path'], sep = sep)
			elif fn_edges['type'] == 'pickle':
				edges = pd.read_pickle(fn_edges['path'])
		else:
			edges = pd.DataFrame({keys['src']: pd.Series(dtype='str'), keys['dst']: pd.Series(dtype='str')})

		return MultiGraph(data = edges, keys = keys, nodes = nodes)

	if self.meta['state'] < 0:
		self.log_error('AgenticGraph is in error state %d' % self._meta_['state'])

		return

	while self._meta_['state'] < intent:
		if self._meta_['state'] == self.states.INITIAL.value:
			try:
				self._fname = self.conf.get('persistence', None)
				if self._fname is None:
					self._graph = new_graph()
				else:
					self._fname = self._fname['path']
					parent_dir	= os.path.dirname(self._fname)

					if parent_dir:
						os.makedirs(parent_dir, exist_ok = True)

					if os.path.isfile(self._fname):
						with open(self._fname, 'rb') as f:
							ntx = pickle.load(f)			# A NetworkX graph object saved by this class.
						self._graph = MultiGraph(data = ntx)
					else:
						self._graph = new_graph()

			except:
				self.log_error('Graph could not be created and initialized for AgenticGraph "%s".' % self.name)
				self._meta_['state'] = self.states.ERR_GRAPH_INIT.value
				break

			self._meta_['state'] = self.states.GRAPH_LOADED_OK.value

			if just_once:
				break

		if self._meta_['state'] == self.states.GRAPH_LOADED_OK.value:
			self._build_indices()
			self._meta_['state'] = self.states.READY.value

			break

mercury.graph.evidence.MultiGraph(data=None, keys=None, nodes=None)

Bases: Graph

MultiGraph is a subclass of Graph, implemented only for AgenticGraph.

It demonstrates how to extend the original Graph class to support MultiDiGraph structures, but only supports NetworkX MultiDiGraph instead of all the technologies supported by the original Graph class. Its interactions with graph-specific functions and attributes in the original class are not guaranteed to work correctly.

Source code in mercury/graph/evidence/agentic_graph.py

def __init__(self, data = None, keys = None, nodes = None):
	super().__init__(data = data, keys = keys, nodes = nodes)

_calculate_edges_colnames()

This internal method is overridden to add the key column for MultiDiGraph edges.

Source code in mercury/graph/evidence/agentic_graph.py

def _calculate_edges_colnames(self):
	""" This internal method is overridden to add the key column for MultiDiGraph edges. """

	l = ['src', 'dst', 'key']
	k = self._as_networkx.edges.keys()
	if len(k) > 0:
		l.extend(list(self._as_networkx.edges[list(self._as_networkx.edges.keys())[0]].keys()))

	return l

_from_pandas(edges, nodes, keys)

This internal method is overridden to construct a NetworkX MultiDiGraph (directed, not weighted) instead of the DiGraph() or Graph() of the original Graph class.

Source code in mercury/graph/evidence/agentic_graph.py

def _from_pandas(self, edges, nodes, keys):
	""" This internal method is overridden to construct a NetworkX MultiDiGraph (directed, not weighted) instead of the DiGraph() or
	Graph() of the original Graph class.
	"""

	src = keys.get('src', 'src')
	dst = keys.get('dst', 'dst')
	id  = keys.get('id', 'id')

	directed = keys.get('directed', True)

	if directed:
		g = nx.MultiDiGraph()
	else:
		raise NotImplementedError('Only directed graphs are currently supported.')

	for _, row in edges.iterrows():
		attr = row.drop([src, dst, id]).to_dict()
		g.add_edge(row[src], row[dst], key = row[id], **attr)

	if nodes is not None:
		for _, row in nodes.iterrows():
			attr = row.drop([id]).to_dict()
			g.add_node(row[id], **attr)

	self._from_networkx(g)

mercury.graph.evidence.Formalizer(schema, extra_args, endpoint=None, logger=None)

Bases: Agentic

The Formalizer is the class that takes in natural language and produces structured data in the form of subgraphs.

Overview

The Formalizer has a double role:

1. While building the evidence graph from text in corpora, providing nodes and edges for the EvidenceGraph to be merged into a coherent structure.
2. While querying the evidence graph in natural language, identifying nodes and edges in the text of a query verifying matches between them and corresponding nodes and edges in the evidence graph.

The level of "intelligence" of the Formalizer

The Formalizer "speaks" the language of EvidenceGraph, has both an Agentic interface (to be used as a tool by an Agent or just exposed in an Endpoint) and direct access (via the parent) to its ontologies and from the EvidenceGraph. It also abstracts the functionality of some "smart" model that does the heavy lifting.

It is smart enough to suggest potential nodes and edges or score how strongly the definition of one node matches another (making both actually the same entity), but the "intelligence" about how that becomes a final EvidenceGraph is handled by either an Agent or the EvidenceGraph itself.

This functionality is often called (SIE) Structured Information Extracting.

How GLiNER2 fits into this workflow

Unlike in the PdfToMarkdown conversion, where a lot of valid "turn-key" solutions exist, and therefore we do not favor any specific tool and provide a simple configuration mechanism to integrate any tool, we consider GLiNER2 as the mature, efficient, open-source, CPU-friendly solution for, at least, this early stage of this project.

Possible alternatives to GLiNER2 could be

• A completely LLM-based solution using a model that could reach similar performance without disproportionate computational cost.
• A SpaCy based solution for named entity recognition and relationship extraction.
• Something like NuExtract3 which for the moment has high GPU requirements, but works directly from PDF (similar to Unlimited-OCR for the PDF stage).

The functions used in GLiNER2 terms

• AutoExtractor (Basic Entity Extraction): Is used for node identification form text. It is provided a schema (from a selection of the Ontology or custom) and the output is provided with autogenerated temporary node identifiers that require further grounding in the EvidenceGraph.
• AutoExtractor (Structure Extraction): Is given an extract_json() format created by the Formalizer to match the definitions of both the relationship and its elements from the Ontology.
• AutoExtractor (Classification): With a classification schema provided. This is used to classify the entire text as containing the information or to determine if nodes belong to the same entity.

Capabilities exposed by the Formalizer

The Formalizer exposes three capabilities:

1. Entity Extraction
2. Relationship Extraction
3. Classification

Known Limitations

The functionality is still very exploratory and minimalistic. Specifically, the following limitations exist:

• Confidence scoring is not implemented.
• There is not mechanism to specify "threshold"
• The key 'definition' is hardcoded, it should be included in the config to make it easier to find and modify.
• hint_edges() requires a different logic since it is forcing to return all possible edges, not just those
• is_a() is not implemented. It is still unclear how to merge entities and if text classification should play a role in it.

In general, the whole system is very early-stage, released for developer use and experimentation. The best way of formalizing entities and relationships from text is still to be determined. It requires a functioning architecture to experiment, and that is what we are to-some-extent providing with this early-stage implementation.

Parameters:

Name	Type	Description	Default
schema	str	a schema (a unique name) to use for the Formalizer's ID.	required
extra_args	dict	the configuration for the Formalizer.	required
endpoint	Agentic	an optional Endpoint. It becomes part of the Formalizer's ID and is available via self.endpoint. If not provided, the Formalizer becomes its own Endpoint.	None
logger	list	an optional logger to use for logging events. It must provide an append() method to add new events.	None

Source code in mercury/graph/evidence/formalizer.py

def __init__(self, schema, extra_args, endpoint = None, logger = None):
	super().__init__(my_class = 'formalizer', schema = schema, endpoint = endpoint, logger = logger)

	if schema is None:			# This allows finding out the class name (to link tools) without actually instantiating a full object.
		return

	self.states = FormalizerState

	self.conf = extra_args
	self.name = schema

	self._entities = None
	self._relation = None
	self._known_id = None
	self._model	   = None

	self._meta_ = self._meta()	# Just to make .meta reflect the initial state.

_capabilities()

Returns the capabilities of the Formalizer.

Returns:

Type

Description

list

A list of capabilities, each represented as a dictionary with the following keys: 'type': The type of capability (e.g., 'function'). 'function': A dictionary containing details about the function The value of 'function' is: 'name': The name of the function. 'description': A brief description of what the function does. 'parameters': A dictionary with 'type', 'properties', and 'required' 'returns': A dictionary with 'type' and 'items'

Source code in mercury/graph/evidence/formalizer.py

def _capabilities(self):
	""" Returns the capabilities of the Formalizer.

	Returns:
		(list): A list of capabilities, each represented as a dictionary with the following keys:

			- 'type': The type of capability (e.g., 'function').
			- 'function': A dictionary containing details about the function

			The value of 'function' is:

			* 'name': The name of the function.
			* 'description': A brief description of what the function does.
			* 'parameters': A dictionary with 'type', 'properties', and 'required'
			* 'returns': A dictionary with 'type' and 'items'
	"""

	name_hint_nodes = 'hint_nodes_%s' % self.name
	name_hint_edges = 'hint_edges_%s' % self.name
	name_is_a		= 'is_a_%s' % self.name

	self.call = {name_hint_nodes: self.hint_nodes, name_hint_edges: self.hint_edges, name_is_a: self.is_a}

	return [
		{
			'type': 'function',
			'function': {
				'name': name_hint_nodes,
				'description': 'Identify nodes in a text from the ontology or some concepts in it.',
				'parameters': {
					'type': 'object',
					'properties': {
						'text': {
							'type': 'string',
							'description': 'Text from which to identify nodes.'
						},
						'concepts': {
							'type': 'array',
							'items': {
								'type': 'string'
							},
							'description': 'List of concepts to identify in the text. If empty, all concepts will be considered.'
						}
					},
					'required': ['text']
				},
				'returns': {
					'type': 'dict',
					'items': {
						'key': 'description'
					}
				}
			}
		},
		{
			'type': 'function',
			'function': {
				'name': name_hint_edges,
				'description': 'Identify edges in a text from the ontology or some concepts in it.',
				'parameters': {
					'type': 'object',
					'properties': {
						'text': {
							'type': 'string',
							'description': 'Text from which to identify edges.'
						},
						'concepts': {
							'type': 'array',
							'items': {
								'type': 'string'
							},
							'description': 'List of concepts to identify in the text. If empty, all concepts will be considered.'
						}
					},
					'required': ['text']
				},
				'returns': {
					'type': 'dict',
					'items': {
						'key': 'description'
					}
				}
			}
		},
		{
			'type': 'function',
			'function': {
				'name': name_is_a,
				'description': 'Determine if a given concept is a type of another concept in the ontology.',
				'parameters': {
					'type': 'object',
					'properties': {
						'text': {
							'type': 'string',
							'description': 'Text from which to determine the relationship between concepts.'
						},
						'concept': {
							'type': 'string',
							'description': 'The concept to check.'
						},
						'parent_concept': {
							'type': 'string',
							'description': 'The parent concept to check against.'
						}
					},
					'required': ['text', 'concept', 'parent_concept']
				},
				'returns': {
					'type': 'number',
					'description': 'A numerical score (0..1) indicating the confidence.',
					'score': {
						'type': 'number'
					}
				}
			}
		}
	]

_dry_run(request)

Simulates running the Formalizer with the given request.

    (See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

NOTE:

The Endpoint takes care of validating the request according to the capabilities exposed by the Formalizer. It is not necessary to validate again here and the Endpoint does not forward the dry_run() request to the Formalizer. This method is provided as a requirement of the Agentic interface, but it is only used when you use Formalizers directly outside of an Endpoint.

Source code in mercury/graph/evidence/formalizer.py

def _dry_run(self, request):
	""" Simulates running the Formalizer with the given request.

		(See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

	## NOTE:

	The Endpoint takes care of validating the request according to the capabilities exposed by the Formalizer. It is not necessary to
	validate again here and the Endpoint does not forward the dry_run() request to the Formalizer. This method is provided as a
	requirement of the Agentic interface, but it is only used when you use Formalizers directly outside of an Endpoint.
	"""

	return {'status': 0, 'description': 'Valid request.'}

_meta()

Returns the metadata of the Formalizer.

(See Agentic.meta().)

Source code in mercury/graph/evidence/formalizer.py

def _meta(self):
	""" Returns the metadata of the Formalizer.

		(See [`Agentic.meta()`][mercury.graph.evidence.Agentic.meta].)
	"""
	meta = {}
	meta['state'] = FormalizerState.INITIAL.value

	meta['description'] = self.conf.get('description', '')
	if type(meta['description']) is list:
		meta['description'] = '\n'.join(meta['description'])

	meta['capabilities'] = self._capabilities()

	return meta

_run(request)

Runs the Formalizer with the given request.

(See Agentic.run().)

Source code in mercury/graph/evidence/formalizer.py

def _run(self, request):
	""" Runs the Formalizer with the given request.

		(See [`Agentic.run()`][mercury.graph.evidence.Agentic.run].)
	"""
	call = self.call.get(request['name'], None)

	if call is None:
		self.log_error('Formalizer does not have a function named "%s".' % request['name'])
		raise AgenticRunInvalidRequest

	ret = {'finish_reason': 'stop', 'message': call(request['arguments'])}

	return ret

close(endpoint_locked)

Closes the Formalizer, persists it to disk and releases any resources it holds.

(See Agentic.close().)

Source code in mercury/graph/evidence/formalizer.py

def close(self, endpoint_locked):
	""" Closes the Formalizer, persists it to disk and releases any resources it holds.

	(See [`Agentic.close()`][mercury.graph.evidence.Agentic.close].)
	"""

	# There is no state to persist, just freeing resources.
	self._entities = None
	self._relation = None
	self._known_id = None
	self._model	   = None

hint_edges(arguments)

Identifies edges in a text from the ontology or some concepts in it.

Parameters:

Name	Type	Description	Default
arguments	dict	A dictionary containing the following keys: - 'text' (str): The text from which to identify edges. - 'concepts' (list): A list of concepts to identify in the text. If empty, all concepts will be considered.	required

Returns:

Type	Description
dict	A dictionary where keys are identified edges and values are their descriptions.

Source code in mercury/graph/evidence/formalizer.py

def hint_edges(self, arguments):
	""" Identifies edges in a text from the ontology or some concepts in it.

	Args:
		arguments (dict): A dictionary containing the following keys:
			- 'text' (str): The text from which to identify edges.
			- 'concepts' (list): A list of concepts to identify in the text. If empty, all concepts will be considered.

	Returns:
		(dict): A dictionary where keys are identified edges and values are their descriptions.
	"""

	if self._meta_['state'] != self.states.READY.value:
		self.log_error('Formalizer is not ready for hint_edges.')

		return None

	text = arguments.get('text', None)

	if text is None:
		self.log_error('Formalizer.hint_edges() %s called without "text" argument.' % self.id)

		return None

	ntx = self._relation._graph.networkx

	format = {}
	concepts = arguments.get('concepts', None)
	if concepts is None:
		for id, _ in ntx.nodes.data('id'):
			node = dict(ntx.nodes(data = True))[id]
			src = node.get('src', None)
			dst = node.get('dest', None)
			key = node.get('definition', None)
			format[key] = [src, dst]

	else:
		for id, _ in ntx.nodes.data('id'):
			if id in concepts:
				node = dict(ntx.nodes(data = True))[id]
				src = node.get('src', None)
				dst = node.get('dest', None)
				key = node.get('definition', None)
				format[key] = [src, dst]

	result = self._model.extract_json(text, format)

	return result

hint_nodes(arguments)

Identifies nodes in a text from the ontology or some concepts in it.

Parameters:

Name	Type	Description	Default
arguments	dict	A dictionary containing the following keys: - 'text' (str): The text from which to identify nodes. - 'concepts' (list): A list of concepts to identify in the text. If empty, all concepts will be considered.	required

Returns:

Type	Description
dict	A dictionary where keys are identified nodes and values are their descriptions.

Source code in mercury/graph/evidence/formalizer.py

def hint_nodes(self, arguments):
	""" Identifies nodes in a text from the ontology or some concepts in it.

	Args:
		arguments (dict): A dictionary containing the following keys:
			- 'text' (str): The text from which to identify nodes.
			- 'concepts' (list): A list of concepts to identify in the text. If empty, all concepts will be considered.

	Returns:
		(dict): A dictionary where keys are identified nodes and values are their descriptions.
	"""

	if self._meta_['state'] != self.states.READY.value:
		self.log_error('Formalizer is not ready for hint_nodes.')

		return None

	text = arguments.get('text', None)

	if text is None:
		self.log_error('Formalizer.hint_nodes() %s called without "text" argument.' % self.id)

		return None

	ntx = self._entities._graph.networkx
	schema = {}
	concepts = arguments.get('concepts', None)
	if concepts is None:
		for id, _ in ntx.nodes.data('id'):
			node = dict(ntx.nodes(data = True))[id]
			schema[id] = node.get('definition', '')
	else:
		for id in concepts:
			node = dict(ntx.nodes(data = True)).get(id, {})
			schema[id] = node.get('definition', '')

	schema = self._model.create_schema().entities(schema)

	result = self._model.extract(text, schema)

	return result.get('entities', {})

is_a(arguments)

Checks if a given concept is a subclass of another concept in the ontology.

Parameters:

Name	Type	Description	Default
arguments	dict	A dictionary containing the following keys: - 'child' (str): The child concept to check. - 'parent' (str): The parent concept to check against.	required

Returns:

Type	Description
float	Returns a [0..1] score indicating the likelihood that the child is a subclass of the parent.

Source code in mercury/graph/evidence/formalizer.py

def is_a(self, arguments):
	""" Checks if a given concept is a subclass of another concept in the ontology.

	Args:
		arguments (dict): A dictionary containing the following keys:
			- 'child' (str): The child concept to check.
			- 'parent' (str): The parent concept to check against.

	Returns:
		(float): Returns a [0..1] score indicating the likelihood that the child is a subclass of the parent.
	"""

	if self._meta_['state'] != self.states.READY.value:
		self.log_error('Formalizer is not ready for is_a.')

		return None

	# TODO: Implement is_a().
	return 0

pilot(intent, just_once=False)

Pilots the Formalizer to a new state based on the given intent.

(See Agentic.pilot().)

Source code in mercury/graph/evidence/formalizer.py

def pilot(self, intent, just_once = False):
	""" Pilots the Formalizer to a new state based on the given intent.

	(See [`Agentic.pilot()`][mercury.graph.evidence.Agentic.pilot].)
	"""

	if self.meta['state'] < 0:
		self.log_error('Formalizer is in error state %d' % self._meta_['state'])

		return

	while self._meta_['state'] < intent:
		if self._meta_['state'] == self.states.INITIAL.value:
			model_config = self.conf.get('model', None)
			if model_config is None:
				self.log_error('Model configuration is missing while piloting Formalizer %s.' % self.id)

				self._meta_['state'] = self.states.ERR_MODEL_INIT.value

				break

			library = model_config.get('library', None)
			if library != 'gliner2' or AutoExtractor is None:
				self.log_error('Missing gliner2.AutoExtractor or unsupported model library while piloting Formalizer %s.' % self.id)

				self._meta_['state'] = self.states.ERR_MODEL_INIT.value

				break

			model = model_config.get('model', None)
			if model is None:
				self.log_error('Model specification is missing while piloting Formalizer %s.' % self.id)

				self._meta_['state'] = self.states.ERR_MODEL_INIT.value

				break

			map_location = model_config.get('map_location', None)
			quantize = model_config.get('quantize', None)
			compile = model_config.get('compile', None)

			args = {}
			if map_location is not None:
				args['map_location'] = map_location

			if quantize is not None:
				args['quantize'] = quantize

			if compile is not None:
				args['compile'] = compile

			try:
				with warnings.catch_warnings():
					warnings.simplefilter('ignore')
					self._model = AutoExtractor.from_pretrained(model, **args)

				self._meta_['state'] = self.states.MODEL_LOADED_OK.value

			except Exception as e:
				self.log_error('Failed to load model while piloting Formalizer %s: %s' % (self.id, str(e)))

				self._meta_['state'] = self.states.ERR_MODEL_INIT.value

				break

			if just_once:
				break

		if self._meta_['state'] == self.states.MODEL_LOADED_OK.value:

			graph = AgenticGraph(schema = None, extra_args = None)

			ontologies	  = self.conf.get('ontologies', None)
			endpoint_name = self.id.split('/')[0]
			graph_class	  = graph.id

			def find_tool(name):
				if ontologies is not None:
					if name not in ontologies:
						self.log_error('Ontologies is malformed in the configuration of %s (missing %s)' % (self.id, name))
						self._meta_['state'] = self.states.ERR_ONTOLOGY_INIT.value

						return None

					name = ontologies[name]

				if name is None:
					return None		# The option is just disabled via configuration

				key = '%s/%s_%s' % (endpoint_name, graph_class, name)

				if key not in self.tools:
					self.log_error('Tool with key "%s" not found while piloting Formalizer %s.' % (key, self.id))
					self._meta_['state'] = self.states.ERR_ONTOLOGY_INIT.value

					return None

				return self.tools[key]

			self._meta_['state'] = self.states.ONTOLOGY_LOADED_OK.value

			self._entities = find_tool('entities')
			self._relation = find_tool('relationships')
			self._known_id = find_tool('known_ids')

			if self._meta_['state'] < 0:
				break

			if self._entities is None:
				self.log_error('Configuration error: "entities" is mandatory while piloting Formalizer %s.' % self.id)
				self._meta_['state'] = self.states.ERR_ONTOLOGY_INIT.value

				break

			if just_once:
				break

		if self._meta_['state'] == self.states.ONTOLOGY_LOADED_OK.value:

			def remove_capability(name):
				self._meta_['capabilities'] = [c for c in self._meta_['capabilities'] if c['function']['name'] != name]

			if self._relation is None:
				remove_capability('hint_edges_%s' % self.name)

			if self._known_id is None:
				remove_capability('is_a_%s' % self.name)

			self._meta_['state'] = self.states.READY.value

			if just_once:
				break

mercury.graph.evidence.EvidenceGraph(schema, extra_args, endpoint=None, logger=None)

Bases: Agentic

The EvidenceGraph is the class that holds the knowledge about one or many sources in the form of a graph.

Overview

This is the core of the architecture.

It is essentially a multigraph, whose nodes and edges are strictly connected by their ids to ontologies and hold information about all the sources that contain every mention to the entities and relationships represented in it and evidence metrics associated with each piece of information.

It delegates, but ultimately controls, its own building process. It pilots itself only up to a point where all its tools are ready to be queried and previous persisted state is loaded, but piloting does not include processing all the sources available to it. That is done by ordering it to crawl from a starting source id.

It is the structured memory where the knowledge (defined and limited to by the concepts in the ontology that created the nodes and edges) is stored.

It is available for:

• Inspection and correction via deterministic (create, read, update, delete) operations
• Aggregation of more evidence as a continuous process
• Integration with upstream and downstream Agentic components

Storing the graph

The EvidenceGraph is a child of Agentic directly, not AgenticGraph. It does contain a MultiGraph to store the data just like AgenticGraph does. There are enough differences (initialization, capabilities, much larger size) and the commonalities are mostly contained in the class MultiGraph which both use to justify this decision.

The graph is persisted to disk using a pickle file and a configuration mechanism identical to that of AgenticGraph.

Connection to ontologies is via the Formalizer

An EvidenceGraph is connected to exactly one Formalizer. The tasks themselves the Formalizer does (id_nodes identifying nodes, id_edges identifying edges and is_same determining identity and category membership) are treated as separate functionality to make it possible to have them handled by an Agent or an agent that supervises another Agentic, etc.

The Formalizer must have its three ontology AgenticGraph objects (entities, relationships, known_ids) defined. And the EvidenceGraph will modify these ontology objects as it integrates new evidence into the graph. It is highly recommended not to share these AgenticGraph ontology objects with other EvidenceGraph instances.

Extraction of candidate entities and relationships

Extraction of candidate entities is controlled via the "id_nodes" configuration key that must provide the name of an Agentic object usable as a tool. In this early implementation, the "id_nodes" tool is expected to be the Formalizer itself, any replacement should provide a compatible capability.

The EvidenceGraph provides a similar key "id_edges", which must also expect a capability compatible with the corresponding capability in the Formalizer that still requires research to provide a reliable implementation.

Merging of entities and relationships inside the EvidenceGraph

Those are two different problems that are currently handled by the same functionality in the Formalizer.

It is currently managed as a text classification with definitions that are not yet fully formalized. Expect this to change.

Evidence weighting and contradiction handling

The EvidenceGraph must compute aggregated evidence weights for both entities and relationships. This is a function of

• an authority score for each source
• a confidence score that the relationship or entity is correctly attributed to the given text.
• a confidence measure of the correct identification of the entities involved in the relationship.

aggregated over each chunk of evidence.

Building the entire EvidenceGraph from Sources

This is done by "crawling" the sources. The mechanism is similar to "reading" by a human, instead of training over the entire corpus. We intentionally, want the system be able to manage the situation where the sources are immense and "reading" everything would require an impractical amount of time and resources to be handled gracefully.

In the case of large corpora, like wikipedia, it is possible to parse them at a high level, just extracting top level titles, index those and their corresponding embeddings. The Sources provide that functionality. It is still an open problem to determine how "answering a question" can have a budget of resources allocated to read new content in the source and become more knowledgeable as a continuous process and not just as a static snapshot of a trained model.

Downstream Agentic Integration

For now, we are researching ways to enforce functionality that is not well supported in the Formalizer (GLiNER2) to be done by Agents. Agents, as currently implemented, should be flexible enough to provide functionality compatible with that of the Formalizer, although at a much higher computational cost. Expect this to evolve as the integration between Agents and the Formalizer matures.

Capabilities (Upstream Agentic Integration)

The first capability implemented is the ability to crawl a source, extracting all the text content, formalizing it and aggregating it. The full implementation of this capability requires proper entity and relationship recognition and aggregation mechanisms.

The EvidenceGraph, once it can reliably aggregate evidence, should not provide any "guessing" functionality, it must strictly behave like a database providing functionality to create, read, update, and delete entities and relationships by a supervisor to maintain quality EvidenceGraph as valuable data assets.

Querying EvidenceGraph

The EvidenceGraph is defined as much by what it does as by what it does not do: It does not answer queries in natural language. It is just a multigraph with downstream solution to gather text and formalize it into structured evidence.

To answer questions about the content of the EvidenceGraph, we use a specialized Agent. The Agent has access to: the EvidenceGraph, the Formalizer and the Sources that the EvidenceGraph has. The Agent can use the Formalizer to identify the entities and relationships in the query and relate them to the ontologies defined in the EvidenceGraph, possibly including user confirmation.

Finding the best way to query the EvidenceGraph through the Agent is an ongoing research topic.

Known Limitations

• This class lacks the merging mechanism for both entities and relationships.
• The size of the EvidenceGraph is limited by the current implementation to a NetworkX multigraph that fits in memory.
• Some mechanism that allows distributing the computation of one large EvidenceGraph across multiple machines will be necessary.
• The full implementation of the Source crawling functionality requires the missing mechanisms.
• Continuous and incomplete (as in the percentage of the source that has been crawled) knowledge acquisition is still an open challenge.

Parameters:

Name	Type	Description	Default
schema	str	a schema (a unique name) to use for the EvidenceGraph's ID.	required
extra_args	dict	the configuration for the EvidenceGraph.	required
endpoint	Agentic	an optional Endpoint. It becomes part of the EvidenceGraph's ID and is available via self.endpoint. If not provided, the EvidenceGraph becomes its own Endpoint.	None
logger	list	an optional logger to use for logging events. It must provide an append() method to add new events.	None

Source code in mercury/graph/evidence/evidence_graph.py

def __init__(self, schema, extra_args, endpoint = None, logger = None):
	super().__init__(my_class = 'evidence_graph', schema = schema, endpoint = endpoint, logger = logger)

	if schema is None:			# This allows finding out the class name (to link tools) without actually instantiating a full object.
		return

	self.states = GraphState

	self.conf = extra_args
	self.name = schema

	self._graph	= None

	self._formalizer = None

	self._entities = None
	self._relation = None
	self._known_id = None

	self._id_nodes = None
	self._id_edges = None
	self._is_same  = None

	self._sources = None

	self._meta_ = self._meta()	# Just to make .meta reflect the initial state.

_capabilities()

Returns the capabilities of the EvidenceGraph.

Returns:

Type

Description

list

A list of capabilities, each represented as a dictionary with the following keys: 'type': The type of capability (e.g., 'function'). 'function': A dictionary containing details about the function The value of 'function' is: 'name': The name of the function. 'description': A brief description of what the function does. 'parameters': A dictionary with 'type', 'properties', and 'required' 'returns': A dictionary with 'type' and 'items'

Source code in mercury/graph/evidence/evidence_graph.py

def _capabilities(self):
	""" Returns the capabilities of the EvidenceGraph.

	Returns:
		(list): A list of capabilities, each represented as a dictionary with the following keys:

			- 'type': The type of capability (e.g., 'function').
			- 'function': A dictionary containing details about the function

			The value of 'function' is:

			* 'name': The name of the function.
			* 'description': A brief description of what the function does.
			* 'parameters': A dictionary with 'type', 'properties', and 'required'
			* 'returns': A dictionary with 'type' and 'items'
	"""

	name_crawl = 'crawl_%s' % self.name

	self.call = {name_crawl: self.crawl}

	return [
		{
			'type': 'function',
			'function': {
				'name': name_crawl,
				'description': 'Crawl the text starting from a given source index.',
				'parameters': {
					'type': 'object',
					'properties': {
						'index': {
							'type': 'string',
							'description': 'Source index from which to start crawling the text.'
						}
					},
					'required': ['index']
				},
				'returns': {
					'type': 'dict',
					'items': {
						'key': 'description'
					}
				}
			}
		}
	]

_connect_downstream()

Connects the EvidenceGraph to downstream components or systems.

This parses the configuration and locates every necessary downstream Agentic: the Formalizer, its ontologies, Agentics with capabilities to do entity extraction, etc. and the Sources.

Returns:

Type	Description
bool	True if the connection was successful, False otherwise.

Source code in mercury/graph/evidence/evidence_graph.py

def _connect_downstream(self):
	""" Connects the EvidenceGraph to downstream components or systems.

	This parses the configuration and locates every necessary downstream Agentic: the Formalizer, its ontologies, Agentics with
	capabilities to do entity extraction, etc. and the Sources.

	Returns:
		(bool): True if the connection was successful, False otherwise.
	"""

	agentics = self.conf.get('agentics', None)

	if agentics is None:
		self.log_error('No agentics configuration found in EvidenceGraph "%s".' % self.name)

		return False

	try:
		formalizer	= agentics['formalizer']

		assert type(formalizer) is str and formalizer != ''

		id_nodes = agentics['id_nodes']
		id_edges = agentics['id_edges']
		is_same	 = agentics['is_same']
		sources	 = agentics['sources']

		if type(sources) is not list:
			assert type(sources) is str
			sources = [sources]

	except:
		self.log_error('Error parsing agentics configuration in EvidenceGraph "%s".' % self.name)

		return False

	endpoint_name = self.id.split('/')[0]

	ag = Formalizer(schema = None, extra_args = None)
	class_name_formalizer = ag.id

	key = '%s/%s_%s' % (endpoint_name, class_name_formalizer, formalizer)

	if key not in self.tools:
		self.log_error('Formalizer "%s" not found in tools for EvidenceGraph "%s".' % (formalizer, self.name))

		return False

	self._formalizer = self.tools[key]

	self._entities = self._formalizer._entities
	self._relation = self._formalizer._relation
	self._known_id = self._formalizer._known_id

	ag = Agent(schema = None, extra_args = None)
	class_name_agent = ag.id

	def get_tool(name):
		key = '%s/%s_%s' % (endpoint_name, class_name_formalizer, name)

		if key in self.tools:
			return self.tools[key]

		key = '%s/%s_%s' % (endpoint_name, class_name_agent, name)
		if key in self.tools:
			return self.tools[key]

		return None

	if id_nodes is not None:
		self._id_nodes = get_tool(id_nodes)

		if self._id_nodes is None:
			self.log_error('"id_nodes" tool "%s" not found for EvidenceGraph "%s".' % (id_nodes, self.name))
			return False

	if id_edges is not None:
		self._id_edges = get_tool(id_edges)

		if self._id_edges is None:
			self.log_error('"id_edges" tool "%s" not found for EvidenceGraph "%s".' % (id_edges, self.name))
			return False

	if is_same is not None:
		self._is_same = get_tool(is_same)

		if self._is_same is None:
			self.log_error('"is_same" tool "%s" not found for EvidenceGraph "%s".' % (is_same, self.name))
			return False

	ag = Source(schema = None, extra_args = None)
	class_name_source = ag.id

	self._sources = []
	for source in sources:
		key = '%s/%s_%s' % (endpoint_name, class_name_source, source)

		if key in self.tools:
			self._sources.append(self.tools[key])
		else:
			self.log_error('"sources" tool "%s" not found for EvidenceGraph "%s".' % (source, self.name))
			return False

	return True

_dry_run(request)

Simulates running the EvidenceGraph with the given request.

    (See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

NOTE:

The Endpoint takes care of validating the request according to the capabilities exposed by the EvidenceGraph. It is not necessary to validate again here and the Endpoint does not forward the dry_run() request to the EvidenceGraph. This method is provided as a requirement of the Agentic interface, but it is only used when you use EvidenceGraphs directly outside of an Endpoint.

Source code in mercury/graph/evidence/evidence_graph.py

def _dry_run(self, request):
	""" Simulates running the EvidenceGraph with the given request.

		(See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

	## NOTE:

	The Endpoint takes care of validating the request according to the capabilities exposed by the EvidenceGraph. It is not necessary to
	validate again here and the Endpoint does not forward the dry_run() request to the EvidenceGraph. This method is provided as a
	requirement of the Agentic interface, but it is only used when you use EvidenceGraphs directly outside of an Endpoint.
	"""

	return {'status': 0, 'description': 'Valid request.'}

_meta()

Returns the metadata of the EvidenceGraph.

(See Agentic.meta().)

Source code in mercury/graph/evidence/evidence_graph.py

def _meta(self):
	""" Returns the metadata of the EvidenceGraph.

	(See [`Agentic.meta()`][mercury.graph.evidence.Agentic.meta].)
	"""

	meta = {}
	meta['state'] = GraphState.INITIAL.value

	meta['description'] = self.conf.get('description', '')
	if type(meta['description']) is list:
		meta['description'] = '\n'.join(meta['description'])

	meta['capabilities'] = self._capabilities()

	return meta

_run(request)

Runs the EvidenceGraph with the given request.

(See Agentic.run().)

Source code in mercury/graph/evidence/evidence_graph.py

def _run(self, request):
	""" Runs the EvidenceGraph with the given request.

	(See [`Agentic.run()`][mercury.graph.evidence.Agentic.run].)
	"""

	call = self.call.get(request['name'], None)

	if call is None:
		self.log_error('EvidenceGraph does not have a function named "%s".' % request['name'])
		raise AgenticRunInvalidRequest

	index = request['arguments'].get('index', None)

	if index is None:
		self.log_error('EvidenceGraph function "%s" requires an "index" argument.' % request['name'])
		raise AgenticRunInvalidRequest

	ret = {'finish_reason': 'stop', 'message': call(index)}

	return ret

close(endpoint_locked)

Closes the EvidenceGraph, persists it to disk and releases any resources it holds.

(See Agentic.close().)

Source code in mercury/graph/evidence/evidence_graph.py

def close(self, endpoint_locked):
	""" Closes the EvidenceGraph, persists it to disk and releases any resources it holds.

	(See [`Agentic.close()`][mercury.graph.evidence.Agentic.close].)
	"""

	if endpoint_locked and self._graph is not None and self._fname is not None:
		ntx = self._graph.networkx
		with open(self._fname, 'wb') as f:
			pickle.dump(ntx, f)

	self._graph	= None

	self._formalizer = None
	self._entities = None
	self._relation = None
	self._known_id = None

	self._id_nodes = None
	self._id_edges = None
	self._is_same  = None

	self._sources = None

pilot(intent, just_once=False)

Pilots the EvidenceGraph to a new state based on the given intent.

(See Agentic.pilot().)

Source code in mercury/graph/evidence/evidence_graph.py

def pilot(self, intent, just_once = False):
	""" Pilots the EvidenceGraph to a new state based on the given intent.

	(See [`Agentic.pilot()`][mercury.graph.evidence.Agentic.pilot].)
	"""

	def new_graph():
		""" Creates an empty new graph when there is no persisted graph to load. The pandas dataframes are empty with just
		column headers as required by a nx.digraph. The class `MultiGraph` overrides the behavior and converts it into a multigraph.
		"""

		keys = {'src': 'src', 'dst': 'dst', 'id': 'id', 'directed': True, 'sep': '\t'}

		nodes = pd.DataFrame({keys['id']: pd.Series(dtype='str')})

		edges = pd.DataFrame({keys['src']: pd.Series(dtype='str'), keys['dst']: pd.Series(dtype='str')})

		return MultiGraph(data = edges, keys = keys, nodes = nodes)

	if self.meta['state'] < 0:
		self.log_error('EvidenceGraph is in error state %d' % self._meta_['state'])

		return

	while self._meta_['state'] < intent:
		if self._meta_['state'] == self.states.INITIAL.value:
			try:
				self._fname = self.conf.get('persistence', None)
				if self._fname is None:
					self._graph = new_graph()
				else:
					self._fname = self._fname['path']
					parent_dir	= os.path.dirname(self._fname)

					if parent_dir:
						os.makedirs(parent_dir, exist_ok = True)

					if os.path.isfile(self._fname):
						with open(self._fname, 'rb') as f:
							ntx = pickle.load(f)			# A NetworkX graph object saved by this class.
						self._graph = MultiGraph(data = ntx)
					else:
						self._graph = new_graph()

			except:
				self.log_error('Graph could not be created and initialized for EvidenceGraph "%s".' % self.name)
				self._meta_['state'] = self.states.ERR_GRAPH_INIT.value
				break

			self._meta_['state'] = self.states.GRAPH_LOADED_OK.value

			if just_once:
				break

		if self._meta_['state'] == self.states.GRAPH_LOADED_OK.value:
			if self._connect_downstream():
				self._meta_['state'] = self.states.READY.value

			else:
				self._meta_['state'] = self.states.ERR_BUILDING.value

			break

mercury.graph.evidence.Agent(schema, extra_args, endpoint=None, logger=None)

Bases: Agentic

Agent is a class that connects to an external LLM using litellm and exposes it in the Agentic tree.

Overview

This connects the LLM model in two ways:

1. Upstream: It exposes the Agent's capabilities (defined in the Agents configuration) via an Agentic interface to any Agentic in its Endpoint that can use it as a tool.

2. Downstream: Making the Agent aware of what tools it has access to, and how they work so it can create proper arguments.

Focus and Scope

Agents solve minimal problems, ideally they would be simpler models than LLMs if such models could produce correct output from inputs.

Agents are not aware of the Endpoint architecture, they only "see" what tools have been provided to them via add_tool().

Agents do not validate, prepare, plan their own tasks. The Endpoint is responsible for the "higher-level" orchestration.

These agents do not always communicate through natural language. They use instructions and metadata in natural language but produce structured output that may or may not include natural language.

Interfacing with Agents

Unlike other Agentic, each Agent has only one capability, but you can create as many Agents as needed.

Agents do not directly call a tool. Instead, they produce a litellm (OpenAI-style) answer with finish_reason='tool_calls' and tool_calls=[ChatCompletionMessageToolCall(function=Function(arguments='{"input": 16}', name='my_tool_for_sqrt'). The Endpoint calls the tools if the "accounting" of resources is valid and calls the Agent back with result properly appended to the Agent's conversation. Any accounting of resources for answering a query belongs to the Endpoint.

Defining an Agent

Agents are defined entirely by their configuration (model, capability and tools). Agents are intentionally "narrow" in scope. You can use as many as you want, either exposing them in the Endpoint or letting another Agent use them as a tool.

Configuration of an Agent

(See the file agents.jsonc of an Endpoint newly created using the mge CLI for a working example.)

To connect to any external LLM, possibly providing credentials, note that anything inside the "completion" dictionary will be passed as arguments to the litellm completion() method in addition to messages and tools (if applicable).

Known Limitations

• For now, calls to Agents to not use parallel execution.
• The management of tool calls is super-simplistic, just a counter to prevent infinite usage.
• There is no safe management of credentials. (You can restrict access to the configuration manually.)

Parameters:

Name	Type	Description	Default
schema	str	a schema (a unique name) to use for the Agent's ID.	required
extra_args	dict	the configuration for the Agent.	required
endpoint	Agentic	an optional Endpoint. It becomes part of the Agent's ID and is available via self.endpoint. If not provided, the Agent becomes its own Endpoint.	None
logger	list	an optional logger to use for logging events. It must provide an append() method to add new events.	None

Source code in mercury/graph/evidence/agent.py

def __init__(self, schema, extra_args, endpoint = None, logger = None):
	super().__init__(my_class = 'agent', schema = schema, endpoint = endpoint, logger = logger)

	if schema is None:			# This allows finding out the class name (to link tools) without actually instantiating a full object.
		return

	self.states = AgentState

	self.conf = extra_args

	self._meta_ = self._meta()	# Just to make .meta reflect the initial state.

_capability(name, description)

Defines a capability for the agent with the given name and description.

Parameters:

Name	Type	Description	Default
name	str	The name of the capability.	required
description	str	A brief description of what the capability does.	required

Returns:

Type	Description
dict	A dictionary representing the capability in the required format.

Source code in mercury/graph/evidence/agent.py

def _capability(self, name, description):
	""" Defines a capability for the agent with the given name and description.

	Args:
		name (str): The name of the capability.
		description (str): A brief description of what the capability does.

	Returns:
		(dict): A dictionary representing the capability in the required format.
	"""

	return {
		'type': 'function',
		'function': {
			'name': name,
			'description': description,
			'parameters': {
				'type': 'object',
				'properties': {
					'content': {
						'type': 'string',
						'description': 'The content to the user prompt.'
					}
				},
				'required': ['content']
			},
			'returns': {
				'type': 'dict'
			}
		}
	}

_dry_run(request)

Simulates running the Agent with the given request.

    (See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)

NOTE:

The Endpoint takes care of validating the request according to the capabilities exposed by the Agent. It is not necessary to validate again here and the Endpoint does not forward the dry_run() request to the Agent. This method is provided as a requirement of the Agentic interface, but it is only used when you use Formalizers directly outside of an Endpoint.

Source code in mercury/graph/evidence/agent.py

def _dry_run(self, request):
	""" Simulates running the Agent with the given request.

		(See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)
	## NOTE:

	The Endpoint takes care of validating the request according to the capabilities exposed by the Agent. It is not necessary to
	validate again here and the Endpoint does not forward the dry_run() request to the Agent. This method is provided as a
	requirement of the Agentic interface, but it is only used when you use Formalizers directly outside of an Endpoint.
	"""

	return {'status': 0, 'description': 'Valid request.'}

_meta()

Returns the metadata of the Agent.

(See Agentic.meta().)

Source code in mercury/graph/evidence/agent.py

def _meta(self):
	""" Returns the metadata of the Agent.

		(See [`Agentic.meta()`][mercury.graph.evidence.Agentic.meta].)
	"""

	meta = {}
	meta['state'] = AgentState.INITIAL.value

	meta['description'] = self.conf.get('description', '')
	if type(meta['description']) is list:
		meta['description'] = '\n'.join(meta['description'])

	meta['capabilities'] = []

	upstream = self.conf.get('upstream', None)
	if upstream is None or 'name' not in upstream or 'description' not in upstream:
		self.log_error('Upstream configuration is missing or incomplete for Agent %s' % self.id)
		meta['state'] = self.states.ERR_SETUP.value

		return meta

	self.name = upstream['name']
	meta['capabilities'].append(self._capability(self.name, upstream['description']))

	return meta

_run(request)

Runs the Agent with the given request.

(See Agentic.run().)

Source code in mercury/graph/evidence/agent.py

def _run(self, request):
	""" Runs the Agent with the given request.

		(See [`Agentic.run()`][mercury.graph.evidence.Agentic.run].)
	"""

	if self._meta_['state'] != self.states.READY.value:
		self.log_error('Agent %s is not ready for ._run.' % self.id)

		raise AgenticRunInvalidState

	if type(request) == list:
		messages = []

		if self.you_are is not None:
			messages.append(self.you_are)

		if self.you_must is not None:
			messages.append(self.you_must)

		for msg in request:
			if 'role' not in msg or 'content' not in msg:
				self.log_error('Invalid message format in request.')
				raise AgenticRunInvalidRequest

			messages.append(msg)

	else:
		if request['name'] != self.name:
			self.log_error('Agent does not have a function named "%s".' % request['name'])
			raise AgenticRunInvalidRequest

		args = request['arguments']

		messages = args.get('messages', None)
		if messages is None:
			if type(args) is dict and len(args) == 1:
				args = next(iter(args.values()))

			if type(args) is str:
				messages = []

				if self.you_are is not None:
					messages.append(self.you_are)

				if self.you_must is not None:
					messages.append(self.you_must)

				messages.append({'role': 'user', 'content': args})

			else:
				self.log_error('Agent _run received invalid arguments.')
				raise AgenticRunInvalidRequest

	try:
		ret = completion(messages = messages, **self.completion)

	except Exception as e:
		self.log_error('Agent encountered an error during completion: %s' % str(e))
		self._meta_['state'] = AgentState.ERR_COMPLETION.value
		raise AgenticRunFailed

	return ret.choices[0]

pilot(intent, just_once=False)

Pilots the Agent to a new state based on the given intent.

(See Agentic.pilot().)

Source code in mercury/graph/evidence/agent.py

def pilot(self, intent, just_once = False):
	""" Pilots the Agent to a new state based on the given intent.

	(See [`Agentic.pilot()`][mercury.graph.evidence.Agentic.pilot].)
	"""

	if self.meta['state'] < 0:
		self.log_error('Agent is in error state %d' % self._meta_['state'])

		return

	while self._meta_['state'] < intent:
		if self._meta_['state'] == self.states.INITIAL.value:
			self.completion = self.conf.get('completion', None)
			if self.completion is None:
				self.log_error('Completion configuration is missing for Agent %s' % self.id)
				self._meta_['state'] = self.states.ERR_SETUP.value

				break

			you_are = self.conf.get('you_are', None)
			if you_are is not None:
				if type(you_are) is list:
					you_are = '\n'.join(you_are)

				if you_are == '':
					you_are = None
				else:
					you_are = {'role': 'system', 'content': you_are}

			self.you_are = you_are

			you_must = self.conf.get('you_must', None)
			if you_must is not None:
				if type(you_must) is list:
					you_must = '\n'.join(you_must)

				if you_must == '':
					you_must = None
				else:
					you_must = {'role': 'developer', 'content': you_must}

			self.you_must = you_must

			self._meta_['state'] = self.states.SETUP_OK.value

			if just_once:
				break

		if self._meta_['state'] == self.states.SETUP_OK.value:
			if completion is None:
				self.log_error('Error importing completion from litellm')
				self._meta_['state'] = self.states.ERR_COMPLETION.value

				break

			self._meta_['state'] = self.states.COMPLETION_OK.value

			if just_once:
				break

		if self._meta_['state'] == self.states.COMPLETION_OK.value:
			tools = []
			for agentic in self.tools.values():
				capabilities = agentic.meta.get('capabilities', None)
				if capabilities is None:
					self.log_error('Error piloting Agent %s: capabilities missing for tool %s' % (self.id, agentic.id))
					self._meta_['state'] = self.states.ERR_BUILDING_TOOLS.value

					return

				for capability in capabilities:
					if 'type' in capability and capability['type'] == 'function' and 'function' in capability:
						tools.append(capability)

					else:
						self.log_error('Error piloting Agent %s: invalid capability format for tool %s' % (self.id, agentic.id))
						self._meta_['state'] = self.states.ERR_BUILDING_TOOLS.value

						return

			if len(tools) > 0:
				self.completion['tools'] = tools

			self._meta_['state'] = self.states.READY.value

			if just_once:
				break

mercury.graph.evidence.Endpoint(path=None, logger=None, auto_pilot=True)

Bases: Agentic

The Endpoint is the class that serves an entire Agentic architecture to the outside world using the Agentic interface itself.

Overview

An Endpoint is a full project that contains any number of Agentic objects. It's metadata lives in a folder that contains a mge_endpoint.jsonc file and a file named either mge_endpoint.free or mge_endpoint.locked that acts as a write mutex. Endpoint is the only object that owns the architecture of the tree.

For any "outside" Agentic user, an Endpoint is just another Agentic service. It is a graph of Agentic objects. There are two ways to do this: Create as many Endpoints as you want in your Python code and use them as Agentics or, more commonly, use the mge cli to maintain and serve the Endpoints persisted in the file system as a folder.

What the Endpoint manages

• The definition of the architecture. The Agentic objects are defined here but may be located anywhere. The architecture can have a life cycle defined by intent values. An intent is a desired state for the architecture. These intents can apply to each Agentic in the architecture. They will typically be ordered integer numbers with names such as "initial", "ready", "busy", etc. defined as an Enum to support human-friendly interfaces (E.g., `mge serve main_doc all_ready 8888'). Negative values represent non recoverable errors, zero is the initial state, and positive values are sorted. Each class can have up to 99 intermediate states below READY which corresponds to 100. Those intents 1..99 represent: building indices, chunking documents, setting up vector databases, formalizing chunks of text, merging into evidence graphs, etc. across multiple Agentic objects. Also, the arrival/update of new documents may require updating the Evidence Graph setting back the state of some of them. So the Endpoint is ALL_READY (100) only when all of them are READY (100).
• Saving, loading and parsing (manually edited) its own definition stored in a folder with its name and a mge_endpoint.jsonc file. That file also contains configuration of Agentics stored as separate files in the same folder.
• A mechanism to pilot the Endpoint's state up to a desired state. This is different from the run() method which runs queries using the Endpoint's Agentic interface. Piloting is a process typically done using the mge cli.
• The mutex providing exclusive write access to the objects. The Endpoint is locked when the cli either serves or pilots the Endpoint. It can also be done programmatically by calling the lock() method. This is mandatory when the metadata is modified.
• It's own Agentic API. This exposes a set of functions merged from the Agentics that are marked as "exposed" in the Endpoint's conf.
• The interdependencies within the Agentic objects.

Http interface

The cli provides a simple http interface to the Endpoint. The Endpoint exposes its Agentic API like any other Agentic object.

Self configuration

Endpoints can use Agents to complete and verify their configuration with or without human intervention.

Attributes:

Name	Type	Description
id	str	the Agentic ID of the Endpoint.
logger	list	the logger to use for logging events. It must provide an append() method to add new events. It is optional.
tools	dict	a dictionary of the Agentics in the Endpoint, keyed by their IDs.
ids	dict	a dictionary of the Agentics in the Endpoint, keyed by their types. This connects the Agentic from their category and name in the architecture to IDs of the loaded objects.
states	Enum	an optional Enum class that defines names for the states of an Agentic. It is used to improve readability and cli argument parsing.
meta	dict	a dictionary of metadata about the Agentic. It is used to store the current state of the Agentic and other information.

Parameters:

Name	Type	Description	Default
path	str	the path to the Endpoint's home directory. The final name (the folder inside whatever path) must be identical to its ._normalize_name() value, that is, a name with only letters, numbers or underscores.	None
logger	list	an optional logger. If not provided, no logging will be done.	None
auto_pilot	bool	an optional flag that completely disables auto-piloting for such things as forceful unlocking. By default, auto_pilot is set via configuration.	True

Source code in mercury/graph/evidence/endpoint.py

def __init__(self, path = None, logger = None, auto_pilot = True):
	if not os.path.isdir(path):
		raise ValueError('The path "%s" is not a valid directory.' % path)

	self.home		= os.path.abspath(path)
	schema			= self._normalize_name(os.path.basename(self.home))
	self.conf_fn	= os.path.join(self.home, 'mge_endpoint.jsonc')
	self.lock_fn	= os.path.join(self.home, 'mge_endpoint.locked')
	self.free_fn	= os.path.join(self.home, 'mge_endpoint.free')
	self.rex_remark = re.compile('^[ \\t]*//.*$')	# Regular expression to remove comments from JSONC files.

	if not os.path.isfile(self.conf_fn):
		raise ValueError('The path "%s" is not a valid Endpoint. The file "mge_endpoint.jsonc" is missing.' % self.conf_fn)

	self.conf = self._json_load(self.conf_fn)

	self.lock(LockState.INIT_IF_NONE)

	super().__init__(my_class = 'endpoint', schema = schema, endpoint = None, logger = logger)

	self.states = EndPointState
	self.ids = {'sources': {}, 'ontologies': {}, 'formalizers': {}, 'evidence_graphs': {}, 'agents': {}, 'custom_agentics': {}}

	if not auto_pilot:
		return

	auto_pilot = self.conf.get('auto_pilot', None)
	if auto_pilot is not None:
		path = os.path.abspath(os.path.join(self.home, auto_pilot['file_name']))

		if os.path.isfile(path):
			with open(path, 'rb') as f:
				obj = pickle.load(f)

			intent = obj.get('state', self.meta['state'])

			if intent > self.meta['state']:
				if self.logger is not None:
					msg	  = 'Auto-piloting Endpoint "%s" to state %d.' % (self.id, intent)
					event = {'type': 'message', 'timestamp': self._now(), 'id': self.id, 'seq_num': self.seq_num, 'message': msg}

					self.logger.append(event)
					self.seq_num += 1

				self.pilot(intent)

__str__()

Returns a console friendly summary of the Endpoint state.

Source code in mercury/graph/evidence/endpoint.py

def __str__(self):
	""" Returns a console friendly summary of the Endpoint state. """

	bold	 = '\033[1m'
	italic	 = '\033[3m'
	reset	 = '\033[0m'
	labels	 = ['name', 'creation_date', 'mge_version', 'description']
	sections = ['sources', 'ontologies', 'formalizers', 'evidence_graphs', 'agents', 'custom_agentics']
	no_name	 = '%sno name%s' % (italic, reset)
	no_obj	 = '%snot loaded%s' % (italic, reset)
	no_state = '%sno state%s' % (italic, reset)
	icons	 = {
		'endpoint': '🌐',
		'sources': '📚',
		'ontologies': '🏛️',
		'formalizers': '🧩',
		'evidence_graphs': '🕸️',
		'agents': '🤖',
		'custom_agentics': '🛠️'
	}

	txt = ['%s%s Endpoint%s' % (bold, icons['endpoint'], reset)]

	for label in labels:
		value = self.conf.get(label, '')

		txt.append('   %-14s: %s' % (label, value))

	state = self.meta['state']
	name  = self.state_name(state)
	if name is None:
		name = no_name

	txt.append('   %-14s: %s %s(%s)%s' % ('state', state, italic, name, reset))

	capabilities = self._meta_.get('capabilities', None)
	if capabilities is not None:
		capability_names = list(self.agentic_by_capability.keys())
		if len(capability_names) > 4:
			capability_names = capability_names[:3] + ['...'] + capability_names[-1:]

		txt.append('   %-14s: %s' % ('capabilities', '(%d total) %s' % (len(capabilities), capability_names)))

	for section_name in sections:
		items = self.conf.get(section_name, {})
		txt.append('')
		txt.append('  %s%s %s%s (%d total)' % (bold, icons[section_name], section_name, reset, len(items)))

		if len(items) == 0:
			txt.append('     %sempty%s' % (italic, reset))
			continue

		for item_name in sorted(items.keys()):
			id = self.ids[section_name].get(item_name, None)

			if id is None:
				txt.append('     - %s: %s' % (item_name, no_obj))
			else:
				agentic = self.tools[id]
				state   = agentic.meta.get('state', None)
				if state is None:
					state = no_state
					name  = no_name
				else:
					name = agentic.state_name(state)
					if name is None:
						name = no_name

				txt.append('     - %s: %s (%s) id: %s' % (item_name, state, name, id))

	return '\n'.join(txt)

_dry_run(request)

Runs a dry run of the Endpoint with the given request.

(See Agentic.dry_run().)

Source code in mercury/graph/evidence/endpoint.py

def _dry_run(self, request):
	""" Runs a dry run of the Endpoint with the given request.

		(See [`Agentic.dry_run()`][mercury.graph.evidence.Agentic.dry_run].)
	"""

	if self._meta_['state'] < self.states.ALL_READY.value:
		return {'status': 1, 'description': 'Not ready.'}

	issues = self._request_issues(request)

	if issues is None:
		return {'status': 0, 'description': 'Valid request.'}

	else:
		return {'status': 2, 'description': str(issues)}

_expose_api()

This method called by pilot() builds the capabilities of the Endpoint by merging the capabilities of all the Agentics that in the 'expose' list of the Endpoint's configuration.

It also checks that all the capabilities have unique names and builds two dictionaries one of capabilities by name and one of Agentic by capability name. These dictionaries capabilities_by_name and agentic_by_capability are used by the run() and dry_run() methods.

It also updates the Endpoint's meta with the capabilities.

The pilot() method calls this method when appropriate and sets the state according to the success.

Returns:

Type	Description
bool	True if no errors found exposing the capabilities, False otherwise.

Source code in mercury/graph/evidence/endpoint.py

def _expose_api(self):
	""" This method called by pilot() builds the capabilities of the Endpoint by merging the capabilities of all the Agentics that
	in the 'expose' list of the Endpoint's configuration.

	It also checks that all the capabilities have unique names and builds two dictionaries one of capabilities by name and one of
	Agentic by capability name. These dictionaries `capabilities_by_name` and `agentic_by_capability` are used  by the run() and
	dry_run() methods.

	It also updates the Endpoint's meta with the capabilities.

	The pilot() method calls this method when appropriate and sets the state according to the success.

	Returns:
		(bool): True if no errors found exposing the capabilities, False otherwise.
	"""

	expose = self.conf.get('expose', None)

	if type(expose) != list or len(expose) == 0:
		self.log_error('The Endpoint configuration must define a non-empty "expose" list.')
		return False

	capabilities = []
	agentic_by_capability = {}
	capabilities_by_name  = {}

	for agentic_name in expose:
		agentic = self.name_to_agentic.get(agentic_name, None)
		if agentic is None:
			self.log_error('Agentic "%s" in "expose" was not found in Endpoint architecture.' % agentic_name)
			return False

		agentic_capabilities = agentic.meta.get('capabilities', None)
		if agentic_capabilities is None:
			self.log_error('Agentic "%s" does not expose any capabilities.' % agentic.id)
			return False

		for capability in agentic_capabilities:
			function = capability.get('function', None)
			if type(function) != dict:
				self.log_error('Agentic "%s" has a capability without a valid "function" object.' % agentic_name)
				return False

			capability_name = function.get('name', None)
			if type(capability_name) != str or capability_name != self._normalize_name(capability_name):
				self.log_error('Agentic "%s" has a capability without a valid function name.' % agentic_name)
				return False

			if capability_name in agentic_by_capability:
				self.log_error('Duplicate exposed capability "%s".' % capability_name)
				return False

			capabilities.append(capability)
			agentic_by_capability[capability_name] = agentic
			capabilities_by_name[capability_name]  = capability

	self._meta_['capabilities']	= capabilities
	self.agentic_by_capability	= agentic_by_capability
	self.capabilities_by_name	= capabilities_by_name
	self.num_capabilities		= len(capabilities)

	return self.num_capabilities > 0

_json_load(fn, recursion_depth=0)

Loads a JSONC file and returns the corresponding object. It also removes comments and recursively loads any referenced JSONC files. The recursion depth is limited to 8 to avoid infinite loops.

Parameters:

Name	Type	Description	Default
fn	str	the path to the JSONC file to load.	required
recursion_depth	int	the current recursion depth used internally in recursive calls.	0

Returns:

Type	Description
any	The object loaded from the JSONC file.

Source code in mercury/graph/evidence/endpoint.py

def _json_load(self, fn, recursion_depth = 0):
	""" Loads a JSONC file and returns the corresponding object. It also removes comments and recursively loads any referenced
	JSONC files. The recursion depth is limited to 8 to avoid infinite loops.

	Args:
		fn (str): the path to the JSONC file to load.
		recursion_depth (int): the current recursion depth used internally in recursive calls.

	Returns:
		(any): The object loaded from the JSONC file.
	"""

	if recursion_depth > 8:
		raise ValueError('Recursion depth exceeded while loading JSON file "%s".' % fn)

	base_path = os.path.dirname(os.path.abspath(fn))

	# Load it as a list of string to remove comments.
	with open(fn, 'r') as f:
		txt = f.readlines()

	txt = [s for s in txt if not self.rex_remark.match(s)]

	ret = json.loads(''.join(txt))

	# Parse the object (top level only) to search for dictionaries that have "$ref" as their only key. When found, load the referenced
	# file and replace the corresponding value with the object loaded recursively.

	if type(ret) == dict:
		for key in ret.keys():
			o = ret[key]

			if (type(o) == dict) and (len(o) == 1) and ('$ref' in o):
				r_fn  = os.path.abspath(os.path.join(base_path, o['$ref']))
				r_ret = self._json_load(r_fn, recursion_depth + 1)
				ret[key] = r_ret

	return ret

_link_objects()

This method is called by pilot() when all the Agentic objects have been loaded.

It just looks for what tools each one requires as defined in the configuration field 'tools'. It checks that every tool is found and the resulting graph does not have cycles.

Once that is done, it calls the add_tool() method of each Agentic which required tools to make them available.

The pilot() method calls this method when appropriate and sets the state according to the success.

Returns:

Type	Description
bool	True if all objects were linked successfully, False otherwise.

Source code in mercury/graph/evidence/endpoint.py

def _link_objects(self):
	""" This method is called by pilot() when all the Agentic objects have been loaded.

	It just looks for what tools each one requires as defined in the configuration field 'tools'. It checks that every tool is
	found and the resulting graph does not have cycles.

	Once that is done, it calls the add_tool() method of each Agentic which required tools to make them available.

	The pilot() method calls this method when appropriate and sets the state according to the success.

	Returns:
		(bool): True if all objects were linked successfully, False otherwise.
	"""

	def _has_cycle(name, visiting, visited):
		if name in visited:
			return False

		if name in visiting:
			return True

		visiting.add(name)

		for tool_name in edges.get(name, []):
			if _has_cycle(tool_name, visiting, visited):
				return True

		visiting.remove(name)
		visited.add(name)

		return False

	self.name_to_agentic = {}
	for section in self.ids.keys():
		for name, id in self.ids[section].items():
			if name in self.name_to_agentic:
				self.log_error('Duplicate tool name (%s).' % name)
				return False

			self.name_to_agentic[name] = self.tools[id]

	edges = {}
	for section in self.ids.keys():
		for name in self.conf[section].keys():
			tool_names = self.conf[section][name].get('tools', [])
			edges[name] = []

			for tool_name in tool_names:
				tool = self.name_to_agentic.get(tool_name, None)

				if tool is None:
					self.log_error('Tool "%s" required by "%s" was not found.' % (tool_name, name))
					return False

				edges[name].append(tool_name)

	visited = set()
	for name in edges.keys():
		if _has_cycle(name, set(), visited):
			self.log_error('Cycle detected in tool dependencies involving "%s".' % name)
			return False

	for section in self.ids.keys():
		for name, id in self.ids[section].items():
			agentic = self.tools[id]

			for tool_name in edges.get(name, []):
				agentic.add_tool(self.name_to_agentic[tool_name])

	return True

_load_objects()

This method parses the self.conf dictionary, category by category: sources, ontologies, formalizers, evidence_graphs, agents and custom_agentics.

It creates and instance of each, passing extra arguments to the constructor if they are present in the configuration. These instances are verified to have unique IDs and are stored in self.ids.

The pilot() method calls this method when appropriate and sets the state according to the success.

Returns:

Type	Description
bool	True if all objects were loaded successfully, False otherwise.

Source code in mercury/graph/evidence/endpoint.py

def _load_objects(self):
	""" This method parses the self.conf dictionary, category by category: sources, ontologies, formalizers, evidence_graphs,
	agents and custom_agentics.

	It creates and instance of each, passing extra arguments to the constructor if they are present in the configuration.
	These instances are verified to have unique IDs and are stored in self.ids.

	The pilot() method calls this method when appropriate and sets the state according to the success.

	Returns:
		(bool): True if all objects were loaded successfully, False otherwise.
	"""

	def _resolve_conf_paths(arg):
		""" Resolve endpoint-relative paths inside configuration objects. """

		new_arg = {}

		for key, value in arg.items():
			if key == '$path':
				new_arg['path'] = os.path.abspath(os.path.join(self.home, value))
			elif type(value) == dict:
				new_arg[key] = _resolve_conf_paths(value)
			else:
				new_arg[key] = value

		return new_arg

	def _load_custom_agentic_class(class_name, file_path):
		""" Load a custom Agentic class from a Python source file. """

		module_name = os.path.splitext(os.path.basename(file_path))[0]

		spec = importlib.util.spec_from_file_location(module_name, file_path)

		if spec is None or spec.loader is None:
			return None

		module = importlib.util.module_from_spec(spec)
		spec.loader.exec_module(module)
		custom_class = getattr(module, class_name, None)

		if custom_class is None or not issubclass(custom_class, Agentic):
			return None

		return custom_class

	section_classes = {
		'sources': Source,
		'ontologies': AgenticGraph,
		'formalizers': Formalizer,
		'evidence_graphs': EvidenceGraph,
		'agents': Agent,
		'custom_agentics': None
	}
	for section in self.ids.keys():
		for value in self.conf[section].values():
			agentic_def = _resolve_conf_paths(value)

			name = agentic_def['name']
			extra_args = agentic_def.get('extra_args', {})
			tools = agentic_def.get('tools', [])

			agentic_class = section_classes[section]

			if agentic_class is None:
				agentic_class = _load_custom_agentic_class(agentic_def['class_name'], agentic_def['path'])

				if agentic_class is None:
					return False

			agentic = agentic_class(schema = name, extra_args = extra_args, endpoint = self, logger = self.logger)

			id = agentic.id

			if name in self.ids[section]:
				self.log_error('Duplicate name (%s) in section "%s"' % (name, section))
				return False

			self.ids[section][name] = id

			if id in self.tools:
				self.log_error('Duplicate ID (%s) in section "%s"' % (id, section))
				return False

			self.tools[id] = agentic

	return True

_meta()

Returns the metadata of the Endpoint.

(See Agentic.meta().)

Source code in mercury/graph/evidence/endpoint.py

def _meta(self):
	""" Returns the metadata of the Endpoint.

		(See [`Agentic.meta()`][mercury.graph.evidence.Agentic.meta].)
	"""

	return {'state' : 0}			# Anything else is created in the different stages of pilot()

_next_agentic_below(intent)

This method is called by pilot() to find the first Agentic in the Endpoint whose state is below the desired intent.

It may be and error state, in which case the Endpoint will set its state to ERR_PILOTING and stop piloting.

The pilot() method calls this method when appropriate and sets the state according to the success.

Returns:

Type	Description
Agentic or None	The first Agentic whose state is below the desired intent, or None if all Agentics are at or above.

Source code in mercury/graph/evidence/endpoint.py

def _next_agentic_below(self, intent):
	""" This method is called by pilot() to find the first Agentic in the Endpoint whose state is below the desired intent.

	It may be and error state, in which case the Endpoint will set its state to ERR_PILOTING and stop piloting.

	The pilot() method calls this method when appropriate and sets the state according to the success.

	Returns:
		(Agentic or None): The first Agentic whose state is below the desired intent, or None if all Agentics are at or above.
	"""

	for agentic in self.tools.values():
		if agentic.meta['state'] < intent:
			return agentic

	return None

_request_issues(request)

Matches the request against the Endpoint's capabilities following the formats described in: Agentic.run()

Parameters:

Name	Type	Description	Default
request	dict	the request to check.	required

Returns:

Type	Description
None or str	None if the request is valid, or a string describing the issues found.

Source code in mercury/graph/evidence/endpoint.py

def _request_issues(self, request):
	""" Matches the request against the Endpoint's capabilities following the formats described in:
	[`Agentic.run()`][mercury.graph.evidence.Agentic.run]

	Args:
		request (dict): the request to check.

	Returns:
		(None or str): None if the request is valid, or a string describing the issues found.
	"""

	pure_function_call = (self.num_capabilities > 1) or (type(request) == dict and 'name' in request and 'arguments' in request)

	if pure_function_call:
		if type(request) != dict:
			return 'Request must be a dictionary with a "name" (of a capability) and "arguments".'

		name = request.get('name', None)

		if type(name) != str:
			return 'Request must have a "name" key with a string value.'

		cap = self.capabilities_by_name.get(name, None)
		if cap is None:
			return 'Capability "%s" not found in Endpoint.' % name

		args = request.get('arguments', None)
		if type(args) != dict:
			return 'Request must have an "arguments" key with a dictionary value.'

		if 'arguments' not in request:
			return 'Request must have an "arguments" key.'

		fun = cap.get('function', None)
		if type(fun) != dict:
			return 'Definition of capability "%s" is malformed. No function details given. Edit its configuration to fix it.' % name

		par = fun.get('parameters', None)
		if type(par) != dict:
			return 'Definition of capability "%s" is malformed. No parameters given. Edit its configuration to fix it.' % name

		for key in par.get('required', []):
			if key not in args:
				return 'Request is missing required argument "%s".' % key

		return None					# No issues found.

	# From here on, the request can only be a message or a list of messages.

	if type(request) != list:
		request = [request]

	for msg in request:
		if type(msg) != dict or ('content' not in msg and 'role' not in msg):
			return 'Request must be a dictionary with "content" and "role".'

	return None						# No issues found.

_research_capabilities()

This method is called by pilot() when every tool is ready. It builds the new dictionary, self.tools_by_capability (similar to self.agentic_by_capability, but for all the capabilities of all the Agentics instead of just the exposed ones). And adds new entries to self.capabilities_by_name to find every capability of every tool by name. It fails if there are duplicate capability names.

Note: These capabilities are only available as tools calls from the Agentics, they are not exposed via meta['capabilities'].

The pilot() method calls this method when appropriate and sets the state according to the success.

Returns:

Type	Description
bool	True if all capabilities were researched successfully, False otherwise.

Source code in mercury/graph/evidence/endpoint.py

def _research_capabilities(self):
	""" This method is called by pilot() when every tool is ready. It builds the new dictionary, self.tools_by_capability
	(similar to self.agentic_by_capability, but for all the capabilities of all the Agentics instead of just the exposed ones).
	And adds new entries to self.capabilities_by_name to find every capability of every tool by name. It fails if there are duplicate
	capability names.

	Note: These capabilities are only available as tools calls from the Agentics, they are not exposed via meta['capabilities'].

	The pilot() method calls this method when appropriate and sets the state according to the success.

	Returns:
		(bool): True if all capabilities were researched successfully, False otherwise.
	"""

	tools_by_capability = {}

	for agentic in self.tools.values():
		agentic_capabilities = agentic.meta.get('capabilities', None)
		if agentic_capabilities is None:
			continue

		for capability in agentic_capabilities:
			function = capability.get('function', None)
			if type(function) != dict:
				self.log_error('Agentic "%s" has a capability without a valid "function" object.' % agentic.id)
				return False

			capability_name = function.get('name', None)
			if type(capability_name) != str or capability_name != self._normalize_name(capability_name):
				self.log_error('Agentic "%s" has a capability without a valid function name.' % agentic.id)
				return False

			if capability_name in tools_by_capability:
				self.log_error('Duplicate capability "%s" found in Agentics.' % capability_name)
				return False

			tools_by_capability[capability_name] = agentic
			self.capabilities_by_name[capability_name] = capability

	self.tools_by_capability = tools_by_capability

	return True

_response_loop(agentic, request, response)

This method is called by _run() to handle the case when an Agentic returns a response with finish_reason = 'tool_calls'.

First it examines if the tool call can be made

• There are enough remaining resources. (According to the configured "max_tool_calls_per_query")
• The tool call expects a tool that is available. (According to self.tools_by_capability)
• It does not check arguments, if it can make the call, the Agentic will check.

If if calling the tool is not allowed or failed, it returns a response with finish_reason = 'error' and a message history that includes the reason for the failure, immediately.

If the tool call works

• It builds or continues a message history as a list of messages.
• It appends the request as a message with the role of the Agent who required, and storing the id of the call if there is one.
• It appends the result of tool call with: 'role': 'tool', 'tool_call_id': , 'content':
• It passes the message history to calling agentic's run() method.
• It stops if the response is not another tool call, or continues the loop if it is.

Parameters:

Name	Type	Description	Default
agentic	Agentic	The Agentic that made the initial request.	required
request	dict	The initial request made to the Agentic.	required
response	dict	The initial response from the Agentic.	required

Returns:

Type	Description
dict	The final response from the Agentic after handling tool calls with possibly some error messages if the tool calls failed or were not allowed.

Source code in mercury/graph/evidence/endpoint.py

def _response_loop(self, agentic, request, response):
	""" This method is called by _run() to handle the case when an Agentic returns a response with finish_reason = 'tool_calls'.

	First it examines if the tool call can be made:
		- There are enough remaining resources. (According to the configured "max_tool_calls_per_query")
		- The tool call expects a tool that is available. (According to self.tools_by_capability)
		- It does not check arguments, if it can make the call, the Agentic will check.

	If if calling the tool is not allowed or failed, it returns a response with finish_reason = 'error' and a message history that
	includes the reason for the failure, immediately.

	If the tool call works:
		- It builds or continues a message history as a list of messages.
		- It appends the request as a message with the role of the Agent who required, and storing the id of the call if there is one.
		- It appends the result of tool call with: 'role': 'tool', 'tool_call_id': <id of the call>, 'content': <result of the call>
		- It passes the message history to calling agentic's run() method.
		- It stops if the response is not another tool call, or continues the loop if it is.

	Args:
		agentic (Agentic): The Agentic that made the initial request.
		request (dict): The initial request made to the Agentic.
		response (dict): The initial response from the Agentic.

	Returns:
		(dict): The final response from the Agentic after handling tool calls with possibly some error messages if the tool calls
			failed or were not allowed.
	"""

	if type(request) == dict:
		message = request['arguments']
		message['role'] = 'user'
		history = [message]

	else:
		history = request												# The Endpoint expects the request to be a conversation.

	n_calls	  = 1														# The initial call to agentic was already made by _run().
	max_calls = int(self.conf.get('max_tool_calls_per_query', 0))

	while True:
		try:
			message = response['message']
			if type(message) != dict:									# If not a dict, it must be a litellm.types.utils.Message
				message = message.model_dump()

		except Exception:
			raise AgenticFailedToParseOutput

		history.append(message)

		tool_calls = message.get('tool_calls', None)
		if type(tool_calls) != list or len(tool_calls) == 0:
			raise AgenticFailedToParseOutput

		for tool_call in tool_calls:
			try:
				function = dict(tool_call['function'])
				name	 = function['name']
				call_id  = tool_call.get('id', None)

			except Exception:
				reason = 'Tool call is missing a valid "function" object.'
				error_message = {'role': 'tool', 'content': reason}
				history.append(error_message)
				return {'finish_reason': 'error', 'message': error_message, 'history': history}

			if n_calls >= max_calls:
				reason = 'Maximum number of tool calls per query exceeded.'
				error_message = {'role': 'tool', 'tool_call_id': call_id, 'content': reason}
				history.append(error_message)
				return {'finish_reason': 'error', 'message': error_message, 'history': history}

			tool = self.tools_by_capability.get(name, None)
			if tool is None:
				reason = 'Tool "%s" was not found.' % name
				error_message = {'role': 'tool', 'tool_call_id': call_id, 'content': reason}
				history.append(error_message)
				return {'finish_reason': 'error', 'message': error_message, 'history': history}

			arguments = function.get('arguments', {})
			if type(arguments) == str:
				try:
					arguments = json.loads(arguments)
				except Exception as e:
					reason = 'Could not parse arguments for tool "%s": %s' % (name, e)
					error_message = {'role': 'tool', 'tool_call_id': call_id, 'content': reason}
					history.append(error_message)
					return {'finish_reason': 'error', 'message': error_message, 'history': history}

			tool_request = {'name': name, 'arguments': arguments}
			if call_id is not None:
				tool_request['id'] = call_id

			n_calls += 1
			try:
				tool_response = tool.run(tool_request)
			except Exception as e:
				reason = 'Tool "%s" failed: %s' % (name, e)
				error_message = {'role': 'tool', 'tool_call_id': call_id, 'content': reason}
				history.append(error_message)
				return {'finish_reason': 'error', 'message': error_message, 'history': history}

			history.append({'role': 'tool', 'tool_call_id': call_id, 'content': tool_response})

		if n_calls >= max_calls:
			reason = 'Maximum number of tool calls per query exceeded.'
			error_message = {'role': 'assistant', 'content': reason}
			history.append(error_message)
			return {'finish_reason': 'error', 'message': error_message, 'history': history}

		n_calls += 1
		response = agentic.run(history)
		finish_reason = response.get('finish_reason', None)
		if finish_reason is None:
			raise AgenticFailedToParseOutput

		if finish_reason == 'tool_calls' or finish_reason.lower().startswith('tool'):
			continue

		if self.conf.get('tool_call_history', False):
			response['history'] = history

		return response

_run(request)

Runs the Endpoint with the given request.

(See Agentic.run().)

This can raise its own AgenticRunException:

• AgenticFailedToFindCapability When the Endpoint could not identify which Agentic and/or which capability to call.
• AgenticFailedToParseOutput When the Endpoint could not parse the output.

Notes

• Unlike _dry_run(), _run() does not call _request_issues(). The Agentic, not the Endpoint validates the call.
• The Endpoint is responsible of handling Tool calls. If an Agentic (typically an Agent) calls a Tool, the Endpoint has to accept or reject the call based on resources (preventing infinite loops, etc.). If the call is rejected, the Endpoint provides a reason and a message history if possible. If accepted, the Endpoint, calls the tool and then calls the same Agentic with a message history that includes the result of the tool call.
• In an Endpoint with more than one capability (the typical case), the request must be a pure function to identify the Agent. When tool calls are made, conversation becomes a message history assuming Agentics that call tools can behave as (or are) Agents. This is normal behavior, the First tool call is actually passing a message to a function.

Source code in mercury/graph/evidence/endpoint.py

def _run(self, request):
	""" Runs the Endpoint with the given request.

	(See [`Agentic.run()`][mercury.graph.evidence.Agentic.run].)

	This can raise its own AgenticRunException:

	* AgenticFailedToFindCapability When the Endpoint could not identify which Agentic and/or which capability to call.
	* AgenticFailedToParseOutput When the Endpoint could not parse the output.

	Notes:
		* Unlike _dry_run(), _run() does not call _request_issues(). The Agentic, not the Endpoint validates the call.
		* The Endpoint is responsible of handling Tool calls. If an Agentic (typically an Agent) calls a Tool, the Endpoint has
			to accept or reject the call based on resources (preventing infinite loops, etc.). If the call is rejected, the Endpoint
			provides a reason and a message history if possible. If accepted, the Endpoint, calls the tool and then calls the same
			Agentic with a message history that includes the result of the tool call.
		* In an Endpoint with more than one capability (the typical case), the request must be a pure function to identify the Agent.
			When tool calls are made, conversation becomes a message history assuming Agentics that call tools can behave as (or are)
			Agents. This is normal behavior, the First tool call is actually passing a message to a function.
	"""

	if self._meta_['state'] < self.states.ALL_READY.value:
		raise AgenticRunInvalidState

	is_pure_call = type(request) == dict and 'name' in request and 'arguments' in request

	if is_pure_call:
		name = request['name']

		agentic = self.agentic_by_capability.get(name, None)

		if agentic is None:
			raise AgenticFailedToFindCapability
	else:
		# In this case, the Endpoint itself behaves like a single Agent capable of long conversations.

		if self.num_capabilities > 1:
			# The caller wants to run a long conversation (is_pure_call == False), but the Endpoint cannot know which Agent to call.
			raise AgenticFailedToFindCapability

		agentic = next(iter(self.agentic_by_capability.values()))		# There is only one capability

	response = agentic.run(request)

	finish_reason = response.get('finish_reason', None)

	if finish_reason is None:
		raise AgenticFailedToParseOutput

	if finish_reason == 'stop' or finish_reason == 'error':		# Canonical 'finish_reason' values first.
		return response

	if finish_reason != 'tool_calls':							# Try to guess other names
		if not finish_reason.lower().startswith('tool'):
			return response										# Non-canonical finish_reason, let the caller handle it.

	return self._response_loop(agentic, request, response)

close(endpoint_locked)

It calls the close() of each Agentic in the Endpoint. And persists its state to disk.

(See Agentic.close().)

Source code in mercury/graph/evidence/endpoint.py

def close(self, endpoint_locked):
	""" It calls the close() of each Agentic in the Endpoint. And persists its state to disk.

		(See [`Agentic.close()`][mercury.graph.evidence.Agentic.close].)
	"""

	for a in self.tools.values():
		if self.logger is not None:
			msg	  = 'Closing Agentic "%s" endpoint_locked = %s.' % (a.id, endpoint_locked)
			event = {'type': 'message', 'timestamp': self._now(), 'id': self.id, 'seq_num': self.seq_num, 'message': msg}

			self.logger.append(event)

		a.close(endpoint_locked)

	if self.logger is not None:
		msg	  = 'Closing Endpoint "%s".' % self.id
		event = {'type': 'message', 'timestamp': self._now(), 'id': self.id, 'seq_num': self.seq_num, 'message': msg}

		self.logger.append(event)
		self.seq_num += 1

	if endpoint_locked:
		auto_save = self.conf.get('auto_save', None)

		if auto_save is None:
			return

		path = os.path.abspath(os.path.join(self.home, self.conf['auto_save']['file_name']))

		obj = {}

		for key in self.conf['auto_save']['save']:
			obj[key] = self._meta_[key]

		with open(path, 'wb') as f:
			pickle.dump(obj, f)

lock(cmd)

Locks or unlocks the Endpoint's mutex. The mutex is used to provide exclusive write access to the Endpoint's metadata for piloting and serving.

Parameters:

Name	Type	Description	Default
cmd	LockState	the lock command. It must be one of the values of the LockState Enum. (See source code for details.)	required

Returns:

Type	Description
LockState	the final state of the mutex after the command is executed. A value in the LockState Enum. (See source code for details.)

Source code in mercury/graph/evidence/endpoint.py

def lock(self, cmd):
	""" Locks or unlocks the Endpoint's mutex. The mutex is used to provide exclusive write access to the Endpoint's metadata for
	piloting and serving.

	Args:
		cmd (LockState): the lock command. It must be one of the values of the `LockState` Enum. (See source code for details.)

	Returns:
		(LockState): the final state of the mutex after the command is executed. A value in the `LockState` Enum.
			(See source code for details.)
	"""

	if cmd == LockState.FREE:
		try:
			os.rename(self.lock_fn, self.free_fn)

			return LockState.FREE

		except Exception:
			if os.path.isfile(self.free_fn):
				return LockState.FREE
			else:
				return LockState.FREE_FAILED

	elif cmd == LockState.LOCK:
		try:
			os.rename(self.free_fn, self.lock_fn)

			return LockState.LOCK

		except Exception:
			return LockState.LOCK_FAILED

	elif cmd == LockState.INIT_IF_NONE:
		if os.path.isfile(self.lock_fn):
			return LockState.LOCK

		elif os.path.isfile(self.free_fn):
			return LockState.FREE

		open(self.free_fn, 'w').close()

		return LockState.FREE

	elif cmd == LockState.FORCE_FREE:
		try:
			os.path.remove(self.lock_fn)
		except Exception:
			pass

		open(self.free_fn, 'w').close()		# No need to check if it exists.

		return LockState.FREE

	else:
		raise ValueError('Invalid lock command "%s".' % cmd)

pilot(intent, just_once=False)

Pilots the Endpoint to a new state based on the given intent.

(See Agentic.pilot().)

Source code in mercury/graph/evidence/endpoint.py

def pilot(self, intent, just_once = False):
	""" Pilots the Endpoint to a new state based on the given intent.

	(See [`Agentic.pilot()`][mercury.graph.evidence.Agentic.pilot].)
	"""

	if self.meta['state'] < 0:		# Irrecoverable error.
		return

	try:
		intent = int(intent)

	except Exception:
		intent = self.states[intent.upper()].value

	while self._meta_['state'] < intent:
		if self._meta_['state'] == self.states.INITIAL.value:
			if self._load_objects():
				self._meta_['state'] = self.states.LOADED_OBJ.value
			else:
				self._meta_['state'] = self.states.ERR_LOADING_OBJ.value
				break

			if just_once:
				break

		if self._meta_['state'] == self.states.LOADED_OBJ.value:
			if self._link_objects():
				self._meta_['state'] = self.states.LINKED_OBJ.value
			else:
				self._meta_['state'] = self.states.ERR_LINKING.value
				break

			if just_once:
				break

		if self._meta_['state'] == self.states.LINKED_OBJ.value:
			if self._expose_api():
				self._meta_['state'] = self.states.EXPOSED_API.value
			else:
				self._meta_['state'] = self.states.ERR_EXPOSING.value
				break

			if just_once:
				break

		next_agentic = self._next_agentic_below(intent)
		if next_agentic is not None:
			if next_agentic.meta['state'] < 0:
				self._meta_['state'] = self.states.ERR_PILOTING.value
				break

			next_agentic.pilot(intent, just_once = just_once)

			if self._next_agentic_below(intent) is None:
				self._meta_['state'] = self.states.TOOLS_ARE_READY.value
			else:
				self._meta_['state'] = self.states.PILOT_REQUIRED.value

			if just_once:
				break

		if self._meta_['state'] == self.states.TOOLS_ARE_READY.value:
			if self._research_capabilities():
				self._meta_['state'] = self.states.ALL_READY.value
			else:
				self._meta_['state'] = self.states.ERR_TOOL_CAPS.value

			break