def analyze()

in aiops/MicroAgents/layers/system/system_analyzer.py [0:0]

• 25 lines of code
• 11 McCabe index (conditional complexity)

    def analyze(self, graph, modules, organization_mode:Literal["forward", "backward", "forward_and_backward", "convolution", "isolation", "full"]="backward", k:int=1, layers=1):
        '''
        根据系统的依赖关系，分析整个系统的异常机器根因，可以从模块依赖图前向分析、后向分析、先前向再后向分析或者类似图神经网络进行图扩散分析（同时考虑前继节点和后继节点）
        Args:
            graph (dict): A networkx graph of the system indicating the dependencies between modules. The graph must be acyclic.
            modules (dict): A dict of modules. Each module is a dict with keys: "module_name", "module_function", "symptom", "module_dependency"
        '''
        if organization_mode in ["forward", "backward", "isolation", "full"]:
            decisions = self.directional_analyze(graph, modules, direction=organization_mode, decisions=None, k=k)

        elif organization_mode =="forward_and_backward":
            decisions = self.forward_and_backward_analyze(graph, modules, k=k, layers=layers)
        elif organization_mode =="convolution":
            decisions = self.convolution_analyze(graph, modules, k=k, layers=layers)
        else:
            raise NotImplementedError(f"organization mode {organization_mode} is not implemented !!!")
            decisions = None

        if self.human_input_mode:
            for node in decisions.keys():
                if decisions[node]:
                    human_response = self.huamn_expert.reply(decisions[node])
                    if self.intervention_mode.lower() == 'cover' and human_response:
                        decisions[node] = human_response
                    elif self.intervention_mode.lower() == 'append' and human_response:
                        decisions[node] += f"\nHuman expert's analysis: {human_response}"
        return decisions