#!/usr/bin/env python3 """ 调度系统行为规范集成模块 为模块2调度系统添加行为规范检查功能 """ import json import os import sys import logging from datetime import datetime from pathlib import Path # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/root/data/disk/system/logs/behavior_rules_scheduler.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class SchedulerBehaviorRulesIntegration: """调度系统行为规范集成类""" def __init__(self): self.rules_base = "/root/data/disk/system/behavior_rules" self.scheduler_base = "/root/data/disk/system" self.load_rules() def load_rules(self): """加载行为规则""" try: config_path = os.path.join(self.rules_base, "config", "behavior_rules.json") with open(config_path, 'r', encoding='utf-8') as f: self.rules = json.load(f) logger.info(f"成功加载行为规则配置") # 加载集成配置 integration_config = os.path.join(self.rules_base, "integration_config.json") if os.path.exists(integration_config): with open(integration_config, 'r', encoding='utf-8') as f: self.integration_config = json.load(f) logger.info(f"成功加载集成配置") else: self.integration_config = {} except Exception as e: logger.error(f"加载行为规则失败: {e}") self.rules = {} self.integration_config = {} def check_task_compliance(self, task_info, agent_id): """ 检查任务是否符合行为规范 Args: task_info: 任务信息字典 agent_id: Agent ID Returns: tuple: (是否通过, 检查结果, 建议) """ logger.info(f"检查任务 {task_info.get('task_id')} 对 {agent_id} Agent 的行为规范符合性") compliance_results = { "task_id": task_info.get("task_id", "unknown"), "agent_id": agent_id, "check_time": datetime.now().isoformat(), "checks": [], "overall_passed": True } # 1. 检查Agent专用规则 agent_key = f"{agent_id}_Agent" if agent_key in self.rules.get("agent_specific_rules", {}): agent_rules = self.rules["agent_specific_rules"][agent_key] # 检查任务类型是否在Agent职责范围内 task_type = task_info.get("task_type", "") agent_responsibilities = agent_rules.get("key_responsibilities", []) responsibility_check = { "name": "职责范围检查", "passed": self.check_task_in_responsibility(task_type, agent_responsibilities), "details": f"任务类型: {task_type}, Agent职责: {', '.join(agent_responsibilities)}" } compliance_results["checks"].append(responsibility_check) if not responsibility_check["passed"]: compliance_results["overall_passed"] = False logger.warning(f"任务 {task_info.get('task_id')} 不在 {agent_id} Agent 职责范围内") # 2. 检查通用行为规则 if self.rules.get("general_rules"): # 工作伦理检查 work_ethics_check = { "name": "工作伦理检查", "passed": True, # 默认通过，实际应根据任务内容检查 "details": "检查任务是否符合工作伦理原则" } compliance_results["checks"].append(work_ethics_check) # 技能使用检查 skill_check = { "name": "技能使用检查", "passed": self.check_skill_usage(task_info), "details": "检查任务使用的技能是否符合规范" } compliance_results["checks"].append(skill_check) if not skill_check["passed"]: compliance_results["overall_passed"] = False # 3. 生成检查报告 compliance_results["summary"] = self.generate_compliance_summary(compliance_results["checks"]) return compliance_results["overall_passed"], compliance_results def check_task_in_responsibility(self, task_type, agent_responsibilities): """检查任务类型是否在Agent职责范围内""" # 这里可以根据实际任务类型和职责进行匹配检查 # 目前简化处理，假设所有任务都在职责范围内 return True def check_skill_usage(self, task_info): """检查技能使用是否符合规范""" # 检查任务是否指定了必要的技能 # 检查技能使用是否符合安全规范 # 目前简化处理，假设技能使用符合规范 return True def generate_compliance_summary(self, checks): """生成符合性总结""" total = len(checks) passed = sum(1 for check in checks if check["passed"]) pass_rate = (passed / total * 100) if total > 0 else 0 summary = { "total_checks": total, "passed_checks": passed, "pass_rate": pass_rate, "status": "PASS" if pass_rate >= 90 else "WARNING" if pass_rate >= 80 else "FAIL" } return summary def remind_agent_follow_rules(self, agent_id, task_info): """提醒Agent按规范执行任务""" logger.info(f"提醒 {agent_id} Agent 按行为规范执行任务 {task_info.get('task_id')}") reminder = { "type": "behavior_rule_reminder", "agent_id": agent_id, "task_id": task_info.get("task_id"), "reminder_time": datetime.now().isoformat(), "message": f"请按 {agent_id} Agent 行为规范执行任务", "rules_to_follow": self.get_agent_rules_to_follow(agent_id, task_info) } # 保存提醒记录 self.save_reminder_record(reminder) return reminder def get_agent_rules_to_follow(self, agent_id, task_info): """获取Agent需要遵守的规则""" rules_to_follow = [] # 添加通用规则 if self.rules.get("general_rules"): rules_to_follow.append("通用行为原则") rules_to_follow.append("技能使用通用规则") rules_to_follow.append("文件操作通用标准") # 添加专用规则 agent_key = f"{agent_id}_Agent" if agent_key in self.rules.get("agent_specific_rules", {}): rules_to_follow.append(f"{agent_id} Agent专用行为规范") return rules_to_follow def save_reminder_record(self, reminder): """保存提醒记录""" try: reminders_dir = os.path.join(self.scheduler_base, "behavior_rule_reminders") os.makedirs(reminders_dir, exist_ok=True) file_path = os.path.join(reminders_dir, f"{reminder['task_id']}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json") with open(file_path, 'w', encoding='utf-8') as f: json.dump(reminder, f, ensure_ascii=False, indent=2) logger.info(f"保存提醒记录到: {file_path}") except Exception as e: logger.error(f"保存提醒记录失败: {e}") def monitor_task_execution(self, task_info, agent_id): """监控任务执行过程""" logger.info(f"开始监控 {agent_id} Agent 执行任务 {task_info.get('task_id')}") monitoring_info = { "task_id": task_info.get("task_id"), "agent_id": agent_id, "start_time": datetime.now().isoformat(), "monitoring_items": [ "技能使用规范符合性", "文件操作规范符合性", "数据更新规范符合性", "通信协作规范符合性" ], "status": "monitoring" } # 保存监控记录 self.save_monitoring_record(monitoring_info) return monitoring_info def save_monitoring_record(self, monitoring_info): """保存监控记录""" try: monitoring_dir = os.path.join(self.scheduler_base, "behavior_rule_monitoring") os.makedirs(monitoring_dir, exist_ok=True) file_path = os.path.join(monitoring_dir, f"{monitoring_info['task_id']}.json") with open(file_path, 'w', encoding='utf-8') as f: json.dump(monitoring_info, f, ensure_ascii=False, indent=2) except Exception as e: logger.error(f"保存监控记录失败: {e}") def generate_compliance_report(self, task_info, agent_id, execution_result): """生成规范符合性报告""" logger.info(f"为任务 {task_info.get('task_id')} 生成行为规范符合性报告") report = { "report_id": f"behavior_compliance_{task_info.get('task_id')}_{datetime.now().strftime('%Y%m%d_%H%M%S')}", "task_id": task_info.get("task_id"), "agent_id": agent_id, "report_time": datetime.now().isoformat(), "execution_result": execution_result, "compliance_evaluation": self.evaluate_compliance(task_info, agent_id, execution_result), "improvement_suggestions": self.generate_improvement_suggestions(task_info, agent_id, execution_result) } # 保存报告 self.save_compliance_report(report) return report def evaluate_compliance(self, task_info, agent_id, execution_result): """评估规范符合性""" evaluation = { "overall_score": 85, # 示例分数，实际应根据执行结果计算 "category_scores": { "work_ethics": 90, "skill_usage": 85, "file_operations": 80, "data_updates": 85, "communication": 90 }, "strengths": [ "工作伦理遵守良好", "沟通协作及时有效" ], "weaknesses": [ "文件操作规范性有待提高", "数据更新流程可以进一步优化" ] } return evaluation def generate_improvement_suggestions(self, task_info, agent_id, execution_result): """生成改进建议""" suggestions = [ { "area": "文件操作", "suggestion": "建议统一使用绝对路径，避免相对路径带来的问题", "priority": "medium" }, { "area": "数据更新", "suggestion": "建议在数据更新前创建备份，确保数据安全", "priority": "high" }, { "area": "技能使用", "suggestion": "建议优化技能调用参数，提高效率", "priority": "low" } ] return suggestions def save_compliance_report(self, report): """保存符合性报告""" try: reports_dir = os.path.join(self.scheduler_base, "behavior_compliance_reports") os.makedirs(reports_dir, exist_ok=True) file_path = os.path.join(reports_dir, f"{report['report_id']}.json") with open(file_path, 'w', encoding='utf-8') as f: json.dump(report, f, ensure_ascii=False, indent=2) logger.info(f"保存符合性报告到: {file_path}") except Exception as e: logger.error(f"保存符合性报告失败: {e}") def main(): """主函数 - 测试集成模块""" print("=== 调度系统行为规范集成模块测试 ===") # 创建集成实例 integration = SchedulerBehaviorRulesIntegration() # 测试数据 test_task = { "task_id": "test_video_001", "task_type": "video_generation", "description": "生成产品介绍短视频", "priority": "high" } test_agent = "Producer" print(f"\n1. 检查任务符合性...") passed, compliance_results = integration.check_task_compliance(test_task, test_agent) print(f" 通过: {passed}") print(f" 检查结果: {json.dumps(compliance_results['summary'], indent=2, ensure_ascii=False)}") print(f"\n2. 发送规范提醒...") reminder = integration.remind_agent_follow_rules(test_agent, test_task) print(f" 提醒内容: {reminder['message']}") print(f"\n3. 开始任务监控...") monitoring_info = integration.monitor_task_execution(test_task, test_agent) print(f" 监控项目: {', '.join(monitoring_info['monitoring_items'])}") print(f"\n4. 生成符合性报告...") execution_result = { "status": "completed", "execution_time": "00:05:30", "quality_score": 8.5, "issues_found": ["文件路径使用了相对路径"] } report = integration.generate_compliance_report(test_task, test_agent, execution_result) print(f" 报告ID: {report['report_id']}") print(f" 总体评分: {report['compliance_evaluation']['overall_score']}") print(f"\n=== 测试完成 ===") if __name__ == "__main__": main()