alpha_tools/rpc_alpha_workflow/reference_01.py

# -*- coding: utf-8 -*-
import uuid
import sys
import httpx
from odoo import models, fields


class AlphaGenerateResearchDirection(models.Model):
    _name = 'alpha.generate.research.direction'
    _description = 'Alpha Generate Research Direction'
    _order = 'id desc'

    name = fields.Char(string='Name', required=True, default=lambda self: str(uuid.uuid4()))

    status = fields.Selection([
        ('draft', 'Draft'),
        ('generated_research', 'Generated Research'),
        ('generated_feature', 'Generated Feature'),
        ('done', 'Done'),
        ('failed', 'Failed'),
        ('cancel', 'Cancel')
    ], string='Status', default='draft')

    region = fields.Many2one('alpha.region.settings', string='Region', required=True)

    universe = fields.Many2one('alpha.universe.settings', string='Universe', required=True)

    research_direction_prompt = fields.Many2one('alpha.prompt.settings', string='Research Direction Prompt')

    feature_engineering_prompt = fields.Many2one('alpha.prompt.settings', string='Feature Engineering Prompt')  # 用于加上替换的研究方向, 组合新的工程特征提示词

    results = fields.Text(string='Results')

    message = fields.Char(string='Message')

    llm_settings_line_id = fields.Many2one('llm.settings.line', string='Model')

    models_name = fields.Char(string='Model Name', related='llm_settings_line_id.model_name')

    provider_name = fields.Char(string='Provider Name', related='llm_settings_line_id.llm_setting_id.name')

    research_direction = fields.Text(string='Research Direction')

    prepare_feature_engineering_prompt = fields.Text(string='Feature Engineering Prompt')  # 用于已组合完成, 准备生成工程特征的提示词

    feature_engineering = fields.Text(string='Feature Engineering')

    use_datasets = fields.Many2one('alpha.datasets', string='Use Datasets')

    datasets_line_ids = fields.One2many('alpha.research.datasets.line', 'research_direction_id', string='Datasets Lines')

    def _get_callback_url(self):
        """根据平台获取回调 URL"""
        platform_info = sys.platform
        if platform_info == "darwin":
            # Mac 开发环境使用本地地址
            base_url_odoo = self.env['ir.config_parameter'].sudo().get_param('web.base.url.local', 'http://127.0.0.1:8069')
        else:
            base_url_odoo = self.env['ir.config_parameter'].sudo().get_param('web.base.url')
        return f"{base_url_odoo}/api/alpha-generate-research-direction/result"

    def btn_generate_research(self):
        """
        生成研究方向按钮
        通过选择的基础提示词，推送到 LLM 微服务生成研究方向
        post 无需等待，LLM 会回写结果到 research_direction 字段
        成功后状态变为 generated_research，失败不修改状态，失败信息写到 message
        """
        self.ensure_one()

        # 如果research_direction字段已存在, 则跳过不生成
        if self.research_direction:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Research direction already exists, please delete it first.',
                    'sticky': False,
                },
            }

        # 校验必填字段
        if not self.research_direction_prompt:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please select a research direction prompt template.',
                    'type': 'danger',
                    'sticky': False
                }
            }
        if not self.llm_settings_line_id:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please select a model.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        model_name = self.llm_settings_line_id.model_name
        base_url = self.llm_settings_line_id.llm_setting_id.base_url
        api_key = self.llm_settings_line_id.llm_setting_id.api_key

        final_prompt = self.research_direction_prompt.prompt

        ms_config = self.get_ms_config()
        ms_url = ms_config.get('url')

        callback_url = self._get_callback_url()

        post_data = {
            "record_id": self.id,
            "odoo_model": self._name,
            "action_type": "research_direction",
            "prompt": final_prompt,
            "model_name": model_name,
            "base_url": base_url,
            "api_key": api_key,
            "callback_url": callback_url
        }

        if not ms_url:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Microservice URL is not available.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        # 发送请求，超短超时，不等待响应
        try:
            httpx.post(f'{ms_url}:32004/api_llm_generate', json=post_data, timeout=0.001)
        except httpx.TimeoutException:
            pass
        except Exception as e:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': f'Failed to send microservices: {e}',
                    'type': 'danger',
                    'sticky': False
                }
            }

        self.write({
            'status': 'draft',
            'message': 'Request sent to microservice, waiting for research direction...'
        })

    def btn_generate_prepare_feature_engineering_prompt(self):
        # 手动填入生成好的研究方向, 跳过第一步的 llm 生成
        self.ensure_one()

        # 如果没有研究方向, 需要生成
        if not self.research_direction or not self.feature_engineering_prompt:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Need to generate research direction and need select feature_engineering_prompt',
                    'sticky': False,
                },
            }

        # 将已有的研究方向, 和基础的feature_engineering_prompt组合
        # feature_engineering_prompt + research_direction
        feature_engineering_prompt = self.feature_engineering_prompt.prompt
        research_direction = self.research_direction

        replaced_feature_engineering_prompt = feature_engineering_prompt.replace('[#replace#]', research_direction)

        self.write({
            'status': 'generated_research',
            'prepare_feature_engineering_prompt': replaced_feature_engineering_prompt
        })

        return True

    def btn_generate_feature(self):
        """
        生成特征工程按钮
        通过 feature_engineering_prompt 作为提示词，推送到 LLM 微服务生成特征工程
        post 无需等待，LLM 会回写结果到 feature_engineering 字段
        成功后状态变为 done，失败状态变为 failed，失败信息写到 message
        """
        self.ensure_one()

        # 如果未生成prepare_feature_engineering_prompt, 则不允许使用此按钮
        if not self.prepare_feature_engineering_prompt:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please generate prepare_feature_engineering_prompt first',
                    'type': 'danger',
                    'sticky': False
                }
            }

        # 如果feature_engineering字段不为空, 则提示需要清空才可以使用这个按钮, 避免重复生成
        if self.feature_engineering:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please clear feature_engineering field first',
                    'type': 'danger',
                    'sticky': False
                }
            }

        # 校验状态
        if self.status != 'generated_research':
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please generate research direction first.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        # 校验必填字段
        if not self.research_direction:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Research direction is empty, please generate it first.',
                    'type': 'danger',
                    'sticky': False
                }
            }
        if not self.llm_settings_line_id:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please select a model.',
                    'type': 'danger',
                    'sticky': False
                }
            }
        if not self.feature_engineering_prompt:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Please select a feature engineering prompt template.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        model_name = self.llm_settings_line_id.model_name
        base_url = self.llm_settings_line_id.llm_setting_id.base_url
        api_key = self.llm_settings_line_id.llm_setting_id.api_key

        # 使用 feature_engineering_prompt 作为提示词
        final_prompt = self.feature_engineering_prompt.prompt

        ms_config = self.get_ms_config()
        ms_url = ms_config.get('url')

        callback_url = self._get_callback_url()

        post_data = {
            "record_id": self.id,
            "odoo_model": self._name,
            "action_type": "feature_engineering",
            "prompt": final_prompt,
            "model_name": model_name,
            "base_url": base_url,
            "api_key": api_key,
            "callback_url": callback_url
        }

        if not ms_url:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Microservice URL is not available.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        # 发送请求，超短超时，不等待响应
        try:
            httpx.post(f'{ms_url}:32004/api_llm_generate', json=post_data, timeout=0.001)
        except httpx.TimeoutException:
            pass
        except Exception as e:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Failed to send microservice.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        self.write({
            'status': 'generated_research',
            'message': 'Request sent to microservice, waiting for feature engineering...'
        })

    def get_ms_config(self):
        # TODO 先从 nacos 获取微服务 url
        nacos_url = ''

        platform_info = sys.platform
        if platform_info == "darwin":
            nacos_url = 'http://192.168.31.41:30848/nacos/v1/cs/configs?dataId=microservices_dev&group=quantify'
        elif platform_info.startswith("linux"):
            nacos_url = 'http://192.168.31.41:30848/nacos/v1/cs/configs?dataId=microservices&group=quantify'

        try:
            ms_config_resp = httpx.get(nacos_url)
            ms_config_resp.raise_for_status()
        except Exception as e:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'Nacos request failed.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        ms_config = ms_config_resp.json()

        return ms_config

    def btn_generate_idea(self):
        # 最后一步, 当工程特征已经生成, 并且已选择数据字段, 则生成idea
        # 同一份特征工程, 多个数据字段, 生成与数据字段相同数量的 idea
        self.ensure_one()

        # if self.status != 'generated_feature':
        #     return {
        #         'type': 'ir.actions.client',
        #         'tag': 'display_notification',
        #         'params': {
        #             'title': 'ERROR',
        #             'message': 'Feature engineering not completed.',
        #             'type': 'danger',
        #             'sticky': False
        #         }
        #     }

        if not self.datasets_line_ids:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'No data fields selected.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        if not self.feature_engineering:
            return {
                'type': 'ir.actions.client',
                'tag': 'display_notification',
                'params': {
                    'title': 'ERROR',
                    'message': 'No feature engineering selected.',
                    'type': 'danger',
                    'sticky': False
                }
            }

        # 开始生成 idea, 先获取到数据集的 name 和 data_sets_type
        for data_field in self.datasets_line_ids:
            data_fields_name = data_field.datasets_id.datasets_id
            data_sets_type = ''
            region_id = self.region.id
            universe_id = self.universe.id
            for data_field_line in data_field.datasets_id.line_ids:
                data_sets_type = data_field_line.data_sets_type
                break

            # 创建一个alpha.idea记录, 为了防止创建重复的 idea, 当前模型的 name 是 uuid4, 加上 region_id, universe_id, data_fields_name, data_sets_type,这几个字段, 先搜索是否存在相应的记录
            # 如果不存在, 创建, 如果存在, 跳过

            # 根据 data_sets_type 确定 data_type
            data_type = 'MATRIX' if data_sets_type == 'MATRIX' else 'VECTOR'

            # 检查是否已存在相同的 idea 记录（特征工程 + 数据集 都相同）
            existing_idea = self.env['alpha.idea'].search([
                ('region', '=', region_id),
                ('universe', '=', universe_id),
                ('data_type', '=', data_type),
                ('replace_prompt', '=', self.feature_engineering),
                ('needed_data_set_ids.name', '=', data_fields_name)
            ], limit=1)

            if existing_idea:
                continue

            # 创建新的 alpha.idea 记录
            idea_vals = {
                'region': region_id,
                'universe': universe_id,
                'data_type': data_type,
                'delay': self.use_datasets.delay if self.use_datasets else '1',
                'replace_prompt': self.feature_engineering,
                'needed_data_set_ids': [(0, 0, {'name': data_fields_name})],
            }

            # 找到最终的 meta_prompt
            meta_prompt = self.env['alpha.prompt.settings'].search([('prompt_type', '=', 'meta_prompt')], limit=1)

            if meta_prompt:
                idea_vals.update({'meta_prompt': meta_prompt.id})

            new_record = self.env['alpha.idea'].create(idea_vals)

            # 创建成功之后, 顺便点一下
            try:
                new_record.btn_check_and_fetch_data()
                new_record.btn_generate_final_prompt()
            except Exception as e:
                print(f'error: {e}')

        return {
            'type': 'ir.actions.client',
            'tag': 'display_notification',
            'params': {
                'title': 'Success',
                'message': 'Ideas generated successfully.',
                'type': 'success',
                'sticky': False,
            }
        }

    def btn_quick_selection_datasets(self):
        self.ensure_one()

        # 1, 检查是否存在已选择的数据集, 如果存在, 清空self.datasets_line_ids
        if self.datasets_line_ids:
            self.datasets_line_ids.unlink()

        # 2, 查询数据集, 并创建关联行
        datasets = self.env['alpha.datasets'].search([
            ('region', '=', self.region.id),
            ('universe', '=', self.universe.id),
        ])

        # 3, 遍历数据, 插入到 datasets_line_ids
        for dataset in datasets:
            self.datasets_line_ids.create({
                'research_direction_id': self.id,
                'datasets_id': dataset.id,
            })

        return True

    def btn_action_cancel(self):
        self.ensure_one()
        self.write({
            'status': 'cancel'
        })


class AlphaResearchDatasetsLine(models.Model):
    """研究方向数据集关联行"""
    _name = 'alpha.research.datasets.line'
    _description = 'Alpha Research Datasets Line'
    _order = 'id desc'

    research_direction_id = fields.Many2one('alpha.generate.research.direction', string='Research Direction', required=True, ondelete='cascade')

    datasets_id = fields.Many2one('alpha.datasets', string='Dataset', required=True)

    data_field_count = fields.Integer(string='Data Field Count', related='datasets_id.data_field_count', readonly=True)

    datasets_code = fields.Char(string='Dataset Code', related='datasets_id.datasets_id', readonly=True)

    datasets_name = fields.Char(string='Dataset Name', related='datasets_id.name', readonly=True)

    region = fields.Char(string='Region', related='datasets_id.region.name', readonly=True)

    universe = fields.Char(string='Universe', related='datasets_id.universe.name', readonly=True)

    delay = fields.Selection([
        ('1', '1'),
        ('0', '0')
    ], string='Delay', related='datasets_id.delay', readonly=True)