Agent File

An agent is a singluar entity with a predefined set of actions to take.

• Read Why the Fury? Building a new flow engine from scratch to understand more about the architecture behind what an Agent is.

Agent

We follow registry pattern for models and actions.

class chainfury.agent.AIAction(node_id: str, model: Model, model_params: Dict[str, Any], fn: object, action_name: str)

Bases: object

This class is a callable for all the AI actions.

Parameters:

• node_id (str) – The id of the node

• model (Model) – The model that is used for this action

• model_params (Dict[str, Any]) – The parameters for the model

• fn (object) – The function that is used for this action

FUNC = 'python-function'

constant for Python function type

JTYPE = 'jinja-template'

constant for Jinja template type

__call__(**data: Dict[str, Any]) → Tuple[Any, Exception | None]

This is a callable that takes in all the arguments that the underlying models take.

Parameters:

**data (Dict[str, Any]) – The data that is passed to the model

Returns:

The output of the model and the exception if any

Return type:

Tuple[Any, Optional[Exception]]

classmethod from_dict(data: Dict[str, Any])

Deserialize the AIAction object from a dict.

to_dict(no_vars: bool = False) → Dict[str, Any]

Serialize the AIAction object to a dict.

class chainfury.agent.AIActionsRegistry

Bases: object

This class is a registry for all the AI actions.

DB_REGISTER = 'db'

get(node_id: str) → Node | None

Get the node for the given node id

Parameters:

node_id (str) – The node id for this action

Returns:

The node object

Return type:

Optional[Node]

get_count_for_nodes(node_id: str) → int

Get number of times a particular node is called

Parameters:

node_id (str) – The node id for this action

Returns:

The number of times the node is called

Return type:

int

get_nodes(tag: str = '') → Dict[str, Dict[str, Any]]

Get all the nodes that are registered

Parameters:

tag (str, optional) – The tag to filter the nodes. Defaults to “”.

Returns:

The dict of nodes

Return type:

Dict[str, Dict[str, Any]]

get_tags() → List[str]

Get all the tags that are registered

Returns:

The list of tags

Return type:

List[str]

register(node_id: str, model_id: str, model_params: Dict[str, Any], fn: object, outputs: Dict[str, Any], action_name: str = '', description: str = '', tags: List[str] = []) → Node

This function will register this action in the local AI registry so it is accesible everywhere. Use this when you are hosting your own chainfury server and want to serve private functions not available in the public DB. If you are using this just for your local usecase and have no interest in the serving capabilities then use to_action instead.

NOTE: If you do not pass node_id then this will create a uudi4. This behaviour is important when dev wants to play with the action without commiting it anywhere.

Parameters:

• node_id (str) – The node id for this action

• model_id (str) – The model id that is to be used for this action

• model_params (Dict[str, Any]) – The model params that are to be used for this action

• fn (object) – The function that is to be used as a preprocessor for this action

• outputs – This is a dict like {‘x’: (-1, ‘b’, ‘c’)}, if provided function returns a dictionary with key x and value automatically extracted from the model output at location (-1, ‘b’, ‘c’).

• description (str, optional) – The description for this action. Defaults to “”.

• tags (List[str], optional) – The tags for this action. Defaults to [].

register_node(node: Node) → Node

to_action(action_name: str, model_id: str, model_params: Dict[str, Any], fn: object, outputs: Dict[str, Any], node_id: str = '', description: str = '') → Node

function to create an “Action” aka. chainfury.Node.

NOTE: If you do not pass node_id then this will create a uudi4. This behaviour is important when dev wants to play with the action without commiting it anywhere.

Parameters:

• model_id (str) – The model id that is to be used for this action

• model_params (Dict[str, Any]) – The model params that are to be used for this action

• fn (object) – The function that is to be used as a preprocessor for this action

• outputs – This is a dict like {‘x’: (-1, ‘b’, ‘c’)}, if provided function returns a dictionary with key x and value automatically extracted from the model output at location (-1, ‘b’, ‘c’).

• node_id (str, optional) – The node id for this action. Defaults to “”.

• description (str, optional) – The description for this action. Defaults to “”.

Returns:

The node object that can be used to create a chain

Return type:

Node

unregister(node_id: str)

Unregister an ai-node

Parameters:

node_id (str) – The node id for this action

Raises:

ValueError – If the node is not found

class chainfury.agent.Memory(node_id: str, fn: object, vector_key: str, read_mode: bool = False)

Bases: object

Class to wrap the DB functions as a callable.

Parameters:

• node_id (str) – The id of the node

• fn (object) – The function that is used for this action

• vector_key (str) – The key for the vector in the DB

• read_mode (bool, optional) – If the function is a read function, if False then this is a write function.

__call__(**data: Dict[str, Any]) → Any

Call self as a function.

fields_model = [Var(*name='items', type='[Var(name='', type='string', items=[], additionalProperties=[]), Var(name='', type='array', items=[Var(name='', type='string', items=[], additionalProperties=[])], additionalProperties=[])]', items=[], additionalProperties=[]), Var(*name='embedding_model', type='string', items=[], additionalProperties=[]), Var(name='embedding_model_params', type='object', items=[], additionalProperties=Var(name='', type='string', items=[], additionalProperties=[])), Var(name='embedding_model_key', type='string', items=[], additionalProperties=[]), Var(name='translation_layer', type='object', items=[], additionalProperties=Var(name='', type='string', items=[], additionalProperties=[]))]

These are the fields that are used to map the input items to the embedding model, do not use directly

classmethod from_dict(data: Dict[str, Any])

Deserialize the Memory object from a dict.

to_dict() → Dict[str, Any]

Serialize the Memory object to a dict.

class chainfury.agent.MemoryRegistry

Bases: object

get_nodes()

get_read(node_id: str) → Node | None

get_write(node_id: str) → Node | None

register_read(component_name: str, fn: object, outputs: Dict[str, Any], vector_key: str, description: str = '', tags: List[str] = []) → Node

register_write(component_name: str, fn: object, outputs: Dict[str, Any], vector_key: str, description: str = '', tags: List[str] = []) → Node

class chainfury.agent.ModelRegistry

Bases: object

Model registry contains metadata for all the models that are provided in the components

get(id: str) → Model

Get a model from the registry

Parameters:

id (str) – Id of the model

Returns:

Model

Return type:

Model

get_count_for_model(id: str) → int

Get the number of times a model is used

Parameters:

id (str) – Id of the model

Returns:

Number of times the model is used

Return type:

int

get_models(tag: str = '') → Dict[str, Dict[str, Any]]

Get all the models that are registered in the registry

Parameters:

tag (str, optional) – Filter models by tag. Defaults to “”.

Returns:

Dictionary of models

Return type:

Dict[str, Dict[str, Any]]

get_tags() → List[str]

Get all the tags that are registered in the registry

Returns:

List of tags

Return type:

List[str]

has(id: str)

A helper function to check if a model is registered or not

register(model: Model)

Register a model in the registry

Parameters:

model (Model) – Model to register

class chainfury.agent.ProgramaticActionsRegistry

Bases: object

Programatic actions registry contains metadata for all the programatic actions that are provided in the components

get(node_id: str) → Node | None

Get a node from the registry

Parameters:

node_id (str) – Id of the node

Returns:

Node

Return type:

Node

get_count_for_nodes(node_id: str) → int

Get the number of times a node is used

Parameters:

node_id (str) – Id of the node

Returns:

Number of times the node is used

Return type:

int

get_nodes(tag: str = '') → Dict[str, Dict[str, Any]]

Get all the nodes that are registered in the registry

Parameters:

tag (str, optional) – Filter nodes by tag. Defaults to “”.

Returns:

Dictionary of nodes

Return type:

Dict[str, Dict[str, Any]]

get_tags() → List[str]

Get all the tags that are registered in the registry

Returns:

List of tags

Return type:

List[str]

register(fn: object, outputs: Dict[str, Tuple[int]], node_id: str = '', description: str = '', tags: List[str] = []) → Node

Register a programatic action in the registry

Parameters:

• fn (object) – Function to register

• node_id (str) – Id of the node

• description (str) – Description of the node

• outputs ([type], optional) – [description]. Defaults to None.

• tags (List[str], optional) – List of tags. Defaults to [].

Raises:

Exception – If the node is already registered

Returns:

Node

Return type:

Node

to_action(fn: object, outputs: Dict[str, Tuple[int]], node_id: str = '', description: str = '', tags: List[str] = []) → Node

chainfury.agent.ai_actions_registry = <chainfury.agent.AIActionsRegistry object>

ai_actions_registry is a global instance of AIActionsRegistry class. This is used to register and unregister AIAction instances. This is used by the server to serve the registered actions.

chainfury.agent.memory_registry = <chainfury.agent.MemoryRegistry object>

memory_registry is a global instance of MemoryRegistry class. This is used to register and unregister Memory instances. This is what the user should use when they want to use the memory elements in their chain.

chainfury.agent.model_registry = <chainfury.agent.ModelRegistry object>

model_registry is a global variable that is used to register models. It is an instance of ModelRegistry class.

chainfury.agent.programatic_actions_registry = <chainfury.agent.ProgramaticActionsRegistry object>

programatic_actions_registry is a global variable that is used to register programatic nodes. It is an instance of ProgramaticActionsRegistry class.