Skip to content

Configuration

xronai.config.agent_factory.AgentFactory

A factory class for creating hierarchical structures of Supervisors and Agents.

This class provides static methods to create a complete hierarchy of Supervisors and Agents based on a configuration dictionary. It handles the creation of tools for Agents and ensures proper nesting of Supervisors and Agents.

Source code in xronai/config/agent_factory.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class AgentFactory:
    """
    A factory class for creating hierarchical structures of Supervisors and Agents.

    This class provides static methods to create a complete hierarchy of
    Supervisors and Agents based on a configuration dictionary. It handles
    the creation of tools for Agents and ensures proper nesting of
    Supervisors and Agents.
    """

    @staticmethod
    async def create_from_config(config: Dict[str, Any],
                                 history_base_path: Optional[str] = None) -> Union[Supervisor, Agent]:
        """
        Create a Supervisor or a standalone Agent from a configuration dictionary.

        Args:
            config (Dict[str, Any]): The configuration dictionary containing
                the entire hierarchy structure.
            history_base_path (Optional[str]): The root directory for storing history logs.

        Returns:
            Union[Supervisor, Agent]: The root Supervisor or Agent of the created workflow.

        Raises:
            ConfigValidationError: If the configuration is invalid.
        """
        ConfigValidator.validate(config)
        workflow_id = config.get('workflow_id', str(uuid.uuid4()))

        if 'supervisor' in config:
            return await AgentFactory._create_supervisor(config['supervisor'],
                                                         is_root=True,
                                                         workflow_id=workflow_id,
                                                         history_base_path=history_base_path)
        elif 'agent' in config:
            return await AgentFactory._create_agent(config['agent'],
                                                    workflow_id=workflow_id,
                                                    history_base_path=history_base_path)

    @staticmethod
    async def _create_supervisor(supervisor_config: Dict[str, Any],
                                 is_root: bool = False,
                                 workflow_id: Optional[str] = None,
                                 history_base_path: Optional[str] = None) -> Supervisor:
        """
        Create a Supervisor instance and its children from a configuration dictionary.

        Args:
            supervisor_config (Dict[str, Any]): The configuration for this Supervisor.
            is_root (bool): Whether this Supervisor is the root of the hierarchy.
            workflow_id (Optional[str]): ID of the workflow (only for root supervisor).
            history_base_path (Optional[str]): The root directory for storing history logs.

        Returns:
            Supervisor: The created Supervisor instance with all its children.
        """
        supervisor = Supervisor(name=supervisor_config['name'],
                                llm_config=supervisor_config['llm_config'],
                                system_message=supervisor_config['system_message'],
                                workflow_id=workflow_id if is_root else None,
                                is_assistant=supervisor_config.get('is_assistant', False),
                                history_base_path=history_base_path)

        for child_config in supervisor_config.get('children', []):
            if child_config['type'] == 'supervisor':
                child = await AgentFactory._create_supervisor(child_config,
                                                              is_root=False,
                                                              workflow_id=None,
                                                              history_base_path=history_base_path)
            else:
                child = await AgentFactory._create_agent(child_config, history_base_path=history_base_path)
            supervisor.register_agent(child)

        return supervisor

    @staticmethod
    async def _create_agent(agent_config: Dict[str, Any],
                            workflow_id: Optional[str] = None,
                            history_base_path: Optional[str] = None) -> Agent:
        """
        Create an Agent instance from a configuration dictionary.

        Args:
            agent_config (Dict[str, Any]): The configuration for this Agent.
            workflow_id (Optional[str]): The ID for the workflow, if this is a root agent.
            history_base_path (Optional[str]): The root directory for storing history logs.

        Returns:
            Agent: The created Agent instance with its tools.
        """
        tools = AgentFactory._create_tools(agent_config.get('tools', []))

        agent_params = {
            'name': agent_config['name'],
            'llm_config': agent_config['llm_config'],
            'system_message': agent_config['system_message'],
            'workflow_id': workflow_id,
            'tools': tools,
            'use_tools': agent_config.get('use_tools',
                                          bool(tools) or bool(agent_config.get('mcp_servers'))),
            'keep_history': agent_config.get('keep_history', True),
            'output_schema': agent_config.get('output_schema'),
            'strict': agent_config.get('strict', False),
            'mcp_servers': agent_config.get('mcp_servers', []),
            'history_base_path': history_base_path
        }

        agent = Agent(**agent_params)
        await agent._load_mcp_tools()
        return agent

    @staticmethod
    def _create_tools(tools_config: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """
        Create a list of tool configurations from the provided tool configs.

        Args:
            tools_config (List[Dict[str, Any]]): List of tool configurations.

        Returns:
            List[Dict[str, Any]]: List of created tool configurations.
        """
        tools = []
        for tool_config in tools_config:
            imported_obj = AgentFactory._import_function(tool_config['python_path'])
            tool_function = None

            if isinstance(imported_obj, type):
                tool_init_config = tool_config.get('config', {})
                tool_instance = imported_obj(**tool_init_config)
                tool_function = tool_instance.execute
            else:
                tool_function = imported_obj

            metadata = {
                "type": "function",
                "function": {
                    "name": tool_config['name'],
                    "description": tool_config.get('description', ''),
                    "parameters": {
                        "type": "object",
                        "properties": tool_config.get('parameters', {}),
                        "required": list(tool_config.get('parameters', {}).keys())
                    }
                }
            }
            tools.append({"tool": tool_function, "metadata": metadata})
        return tools

    @staticmethod
    def _import_function(python_path: str):
        """
        Import a function or class from a given Python path.

        Args:
            python_path (str): The full path to the object, including module.

        Returns:
            Callable or type: The imported function or class.

        Raises:
            ImportError: If the module or object cannot be imported.
            AttributeError: If the specified object is not found in the module.
        """
        module_name, object_name = python_path.rsplit('.', 1)
        module = importlib.import_module(module_name)
        return getattr(module, object_name)

create_from_config(config, history_base_path=None) async staticmethod

Create a Supervisor or a standalone Agent from a configuration dictionary.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary containing the entire hierarchy structure.

 required
history_base_path Optional[str]

The root directory for storing history logs.

 None

Returns:

Type Description
Union[Supervisor, Agent]

Union[Supervisor, Agent]: The root Supervisor or Agent of the created workflow.

Raises:

Type Description
ConfigValidationError

If the configuration is invalid.
Source code in xronai/config/agent_factory.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@staticmethod
async def create_from_config(config: Dict[str, Any],
                             history_base_path: Optional[str] = None) -> Union[Supervisor, Agent]:
    """
    Create a Supervisor or a standalone Agent from a configuration dictionary.

    Args:
        config (Dict[str, Any]): The configuration dictionary containing
            the entire hierarchy structure.
        history_base_path (Optional[str]): The root directory for storing history logs.

    Returns:
        Union[Supervisor, Agent]: The root Supervisor or Agent of the created workflow.

    Raises:
        ConfigValidationError: If the configuration is invalid.
    """
    ConfigValidator.validate(config)
    workflow_id = config.get('workflow_id', str(uuid.uuid4()))

    if 'supervisor' in config:
        return await AgentFactory._create_supervisor(config['supervisor'],
                                                     is_root=True,
                                                     workflow_id=workflow_id,
                                                     history_base_path=history_base_path)
    elif 'agent' in config:
        return await AgentFactory._create_agent(config['agent'],
                                                workflow_id=workflow_id,
                                                history_base_path=history_base_path)

xronai.config.config_validator.ConfigValidator

A validator class for checking the structure and content of configuration dictionaries.

This class provides static methods to validate the entire configuration hierarchy, including Supervisors, Agents, LLM configurations, and tools.

Source code in xronai/config/config_validator.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class ConfigValidator:
    """
    A validator class for checking the structure and content of configuration dictionaries.

    This class provides static methods to validate the entire configuration
    hierarchy, including Supervisors, Agents, LLM configurations, and tools.
    """

    @staticmethod
    def validate(config: Dict[str, Any]) -> None:
        """
        Validate the entire configuration dictionary.

        Args:
            config (Dict[str, Any]): The configuration dictionary to validate.

        Raises:
            ConfigValidationError: If the configuration is invalid.
        """
        has_supervisor = 'supervisor' in config
        has_agent = 'agent' in config

        if not has_supervisor and not has_agent:
            raise ConfigValidationError("Configuration must contain a root 'supervisor' or 'agent' key.")

        if has_supervisor and has_agent:
            raise ConfigValidationError("Configuration cannot contain both a root 'supervisor' and 'agent' key.")

        if has_supervisor:
            ConfigValidator._validate_supervisor(config['supervisor'], is_root=True)
        elif has_agent:
            ConfigValidator._validate_agent(config['agent'])

    @staticmethod
    def _validate_supervisor(supervisor: Dict[str, Any], is_root: bool = False) -> None:
        """
        Validate a supervisor configuration.

        Args:
            supervisor (Dict[str, Any]): The supervisor configuration to validate.
            is_root (bool): Whether this supervisor is the root of the hierarchy.

        Raises:
            ConfigValidationError: If the supervisor configuration is invalid.
        """
        required_fields = ['name', 'type', 'llm_config', 'system_message']
        if is_root:
            if 'children' in supervisor and not isinstance(supervisor['children'], list):
                raise ConfigValidationError("'children' must be a list of agent/supervisor configurations.")

        for field in required_fields:
            if field not in supervisor:
                raise ConfigValidationError(f"Missing required field '{field}' in supervisor configuration")

        if supervisor['type'] != 'supervisor':
            raise ConfigValidationError(f"Invalid type for supervisor: {supervisor['type']}")

        if 'is_assistant' in supervisor and not isinstance(supervisor['is_assistant'], bool):
            raise ConfigValidationError("'is_assistant' must be a boolean value")

        if is_root and supervisor.get('is_assistant', False):
            raise ConfigValidationError("Root supervisor cannot be an assistant supervisor")

        ConfigValidator._validate_llm_config(supervisor['llm_config'])

        for child in supervisor.get('children', []):
            if child['type'] == 'supervisor':
                if not child.get('is_assistant', False):
                    raise ConfigValidationError("Child supervisors must be assistant supervisors")
                ConfigValidator._validate_supervisor(child)
            elif child['type'] == 'agent':
                ConfigValidator._validate_agent(child)
            else:
                raise ConfigValidationError(f"Invalid type for child: {child['type']}")

    @staticmethod
    def _validate_agent(agent: Dict[str, Any]) -> None:
        """
        Validate an agent configuration.

        Args:
            agent (Dict[str, Any]): The agent configuration to validate.

        Raises:
            ConfigValidationError: If the agent configuration is invalid.
        """
        required_fields = ['name', 'type', 'llm_config', 'system_message']
        for field in required_fields:
            if field not in agent:
                raise ConfigValidationError(f"Missing required field '{field}' in agent configuration")

        if agent['type'] != 'agent':
            raise ConfigValidationError(f"Invalid type for agent: {agent['type']}")

        bool_fields = ['keep_history', 'use_tools', 'strict']
        for field in bool_fields:
            if field in agent and not isinstance(agent[field], bool):
                raise ConfigValidationError(f"'{field}' must be a boolean value")

        if 'output_schema' in agent:
            if not isinstance(agent['output_schema'], dict):
                raise ConfigValidationError("output_schema must be a dictionary")
            if 'type' not in agent['output_schema']:
                raise ConfigValidationError("output_schema must have 'type' field")

        if 'mcp_servers' in agent:
            if not isinstance(agent['mcp_servers'], list):
                raise ConfigValidationError("mcp_servers must be a list")
            for server in agent['mcp_servers']:
                if 'type' not in server:
                    raise ConfigValidationError("Each MCP server must have 'type' field")
                if server['type'] not in ['sse', 'stdio']:
                    raise ConfigValidationError("MCP server type must be 'sse' or 'stdio'")
                if server['type'] == 'sse' and 'url' not in server:
                    raise ConfigValidationError("SSE MCP server must have 'url' field")
                if server['type'] == 'stdio' and 'script_path' not in server:
                    raise ConfigValidationError("stdio MCP server must have 'script_path' field")

        ConfigValidator._validate_llm_config(agent['llm_config'])
        ConfigValidator._validate_tools(agent.get('tools', []))

    @staticmethod
    def _validate_llm_config(llm_config: Dict[str, Any]) -> None:
        """
        Validate the LLM (Language Model) configuration.

        Args:
            llm_config (Dict[str, Any]): The LLM configuration to validate.

        Raises:
            ConfigValidationError: If the LLM configuration is invalid.
        """
        required_fields = ['model', 'api_key', 'base_url']
        for field in required_fields:
            if field not in llm_config:
                raise ConfigValidationError(f"Missing required field '{field}' in llm_config")

    @staticmethod
    def _validate_tools(tools: List[Dict[str, Any]]) -> None:
        """
        Validate the list of tools in an agent's configuration.

        Args:
            tools (List[Dict[str, Any]]): The list of tool configurations to validate.

        Raises:
            ConfigValidationError: If any tool configuration is invalid.
        """
        for tool in tools:
            required_fields = ['name', 'type', 'python_path']
            for field in required_fields:
                if field not in tool:
                    raise ConfigValidationError(f"Missing required field '{field}' in tool configuration")

            if tool['type'] not in ['function', 'class']:
                raise ConfigValidationError(f"Invalid tool type: {tool['type']}. Must be 'function' or 'class'.")

validate(config) staticmethod

Validate the entire configuration dictionary.

Parameters:

Name Type Description Default
config Dict[str, Any]

The configuration dictionary to validate.

 required

Raises:

Type Description
ConfigValidationError

If the configuration is invalid.
Source code in xronai/config/config_validator.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@staticmethod
def validate(config: Dict[str, Any]) -> None:
    """
    Validate the entire configuration dictionary.

    Args:
        config (Dict[str, Any]): The configuration dictionary to validate.

    Raises:
        ConfigValidationError: If the configuration is invalid.
    """
    has_supervisor = 'supervisor' in config
    has_agent = 'agent' in config

    if not has_supervisor and not has_agent:
        raise ConfigValidationError("Configuration must contain a root 'supervisor' or 'agent' key.")

    if has_supervisor and has_agent:
        raise ConfigValidationError("Configuration cannot contain both a root 'supervisor' and 'agent' key.")

    if has_supervisor:
        ConfigValidator._validate_supervisor(config['supervisor'], is_root=True)
    elif has_agent:
        ConfigValidator._validate_agent(config['agent'])