当前位置： 首页 > news >正文

山东网站建设开发维护互联网精准营销

news 2026/4/8 2:39:42
山东网站建设开发维护,互联网精准营销,万网域名免费注册,上海高档网站建设让我用一个数据分析项目的例子来展示plan-and-execute框架的应用。这个例子会涉及数据处理、分析和可视化等任务。 from typing import List, Dict, Any from dataclasses import dataclass import json from enum import Enum import logging from datetime import datetime#…

让我用一个数据分析项目的例子来展示plan-and-execute框架的应用。这个例子会涉及数据处理、分析和可视化等任务。

from typing import List, Dict, Any
from dataclasses import dataclass
import json
from enum import Enum
import logging
from datetime import datetime# 任务状态枚举
class TaskStatus(Enum):PENDING = "pending"RUNNING = "running"COMPLETED = "completed"FAILED = "failed"# 任务优先级枚举
class TaskPriority(Enum):LOW = 1MEDIUM = 2HIGH = 3# 任务定义
@dataclass
class Task:id: strname: strdescription: strpriority: TaskPrioritydependencies: List[str]  # 依赖的任务ID列表status: TaskStatusresult: Any = Noneerror: str = None# 工作流执行器
class WorkflowExecutor:def __init__(self):self.tasks = {}self.logger = logging.getLogger(__name__)def add_task(self, task: Task):self.tasks[task.id] = taskdef get_ready_tasks(self) -> List[Task]:"""获取所有依赖已满足的待执行任务"""ready_tasks = []for task in self.tasks.values():if task.status == TaskStatus.PENDING:dependencies_met = all(self.tasks[dep_id].status == TaskStatus.COMPLETEDfor dep_id in task.dependencies)if dependencies_met:ready_tasks.append(task)return sorted(ready_tasks, key=lambda x: x.priority.value, reverse=True)def execute_task(self, task: Task):"""执行单个任务"""task.status = TaskStatus.RUNNINGtry:# 这里实现具体任务的执行逻辑result = self.task_handlers[task.id](task, {dep: self.tasks[dep].result for dep in task.dependencies})task.result = resulttask.status = TaskStatus.COMPLETEDexcept Exception as e:task.status = TaskStatus.FAILEDtask.error = str(e)self.logger.error(f"Task {task.id} failed: {e}")def execute_workflow(self):"""执行整个工作流"""while True:ready_tasks = self.get_ready_tasks()if not ready_tasks:breakfor task in ready_tasks:self.execute_task(task)# 检查是否所有任务都完成all_completed = all(task.status == TaskStatus.COMPLETED for task in self.tasks.values())return all_completed# 数据分析工作流示例
class DataAnalysisWorkflow:def __init__(self, data_path: str, output_path: str):self.data_path = data_pathself.output_path = output_pathself.executor = WorkflowExecutor()def plan_workflow(self):"""规划工作流程"""tasks = [Task(id="load_data",name="加载数据",description="从CSV文件加载数据",priority=TaskPriority.HIGH,dependencies=[],status=TaskStatus.PENDING),Task(id="clean_data",name="数据清洗",description="处理缺失值和异常值",priority=TaskPriority.HIGH,dependencies=["load_data"],status=TaskStatus.PENDING),Task(id="feature_engineering",name="特征工程",description="创建新特征",priority=TaskPriority.MEDIUM,dependencies=["clean_data"],status=TaskStatus.PENDING),Task(id="statistical_analysis",name="统计分析",description="计算基本统计指标",priority=TaskPriority.MEDIUM,dependencies=["clean_data"],status=TaskStatus.PENDING),Task(id="visualization",name="数据可视化",description="生成图表",priority=TaskPriority.MEDIUM,dependencies=["statistical_analysis"],status=TaskStatus.PENDING),Task(id="generate_report",name="生成报告",description="生成分析报告",priority=TaskPriority.LOW,dependencies=["visualization", "feature_engineering"],status=TaskStatus.PENDING)]for task in tasks:self.executor.add_task(task)def register_task_handlers(self):"""注册任务处理函数"""self.executor.task_handlers = {"load_data": self.load_data,"clean_data": self.clean_data,"feature_engineering": self.feature_engineering,"statistical_analysis": self.statistical_analysis,"visualization": self.visualization,"generate_report": self.generate_report}def load_data(self, task: Task, dependencies: Dict):import pandas as pddf = pd.read_csv(self.data_path)return dfdef clean_data(self, task: Task, dependencies: Dict):df = dependencies["load_data"]# 处理缺失值df = df.fillna(df.mean())# 处理异常值# ... 其他清洗逻辑return dfdef feature_engineering(self, task: Task, dependencies: Dict):df = dependencies["clean_data"]# 创建新特征# ... 特征工程逻辑return dfdef statistical_analysis(self, task: Task, dependencies: Dict):df = dependencies["clean_data"]stats = {"basic_stats": df.describe(),"correlations": df.corr(),# ... 其他统计分析}return statsdef visualization(self, task: Task, dependencies: Dict):import matplotlib.pyplot as pltstats = dependencies["statistical_analysis"]figures = []# 生成可视化# ... 可视化逻辑return figuresdef generate_report(self, task: Task, dependencies: Dict):figures = dependencies["visualization"]df_features = dependencies["feature_engineering"]report = {"timestamp": datetime.now().isoformat(),"statistics": str(dependencies["statistical_analysis"]),"features": df_features.columns.tolist(),"figures": [f.to_json() for f in figures]}# 保存报告with open(f"{self.output_path}/report.json", "w") as f:json.dump(report, f, indent=2)return reportdef run(self):"""运行完整的工作流"""self.plan_workflow()self.register_task_handlers()success = self.executor.execute_workflow()if success:final_report = self.executor.tasks["generate_report"].resultprint("工作流执行成功!")return final_reportelse:failed_tasks = [task for task in self.executor.tasks.values()if task.status == TaskStatus.FAILED]print("工作流执行失败。失败的任务:")for task in failed_tasks:print(f"- {task.name}: {task.error}")return None# 使用示例
def main():workflow = DataAnalysisWorkflow(data_path="data/sales_data.csv",output_path="output")result = workflow.run()if result:print("分析报告已生成:", result)else:print("工作流执行失败")if __name__ == "__main__":main()

这个例子展示了：

  1. 工作流框架的核心组件：
  • Task定义
  • 工作流执行器
  • 依赖管理
  • 状态追踪
  • 错误处理
  1. 实现的关键特性：
  • 自动任务规划
  • 依赖关系处理
  • 并行任务执行
  • 结果传递
  • 错误恢复
  1. 可以扩展的方向：
# 1. 添加任务重试机制
class RetryableExecutor(WorkflowExecutor):def execute_task(self, task: Task, max_retries: int = 3):retries = 0while retries < max_retries:try:super().execute_task(task)if task.status == TaskStatus.COMPLETED:breakexcept Exception as e:retries += 1self.logger.warning(f"Retry {retries}/{max_retries} for task {task.id}")# 2. 添加进度监控
class MonitoredWorkflow(DataAnalysisWorkflow):def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.progress_callback = Nonedef set_progress_callback(self, callback):self.progress_callback = callbackdef update_progress(self, task: Task, status: str):if self.progress_callback:self.progress_callback(task, status)# 3. 添加中间结果缓存
class CachedExecutor(WorkflowExecutor):def __init__(self, cache_dir: str):super().__init__()self.cache_dir = cache_dirdef get_cached_result(self, task: Task):cache_path = f"{self.cache_dir}/{task.id}.cache"if os.path.exists(cache_path):return pickle.load(open(cache_path, "rb"))return Nonedef cache_result(self, task: Task):cache_path = f"{self.cache_dir}/{task.id}.cache"pickle.dump(task.result, open(cache_path, "wb"))
  1. 使用建议：
# 1. 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)# 2. 添加性能监控
from time import timeclass PerformanceMonitor:def __init__(self):self.task_times = {}def start_task(self, task_id: str):self.task_times[task_id] = {"start": time()}def end_task(self, task_id: str):self.task_times[task_id]["end"] = time()def get_task_duration(self, task_id: str):times = self.task_times[task_id]return times["end"] - times["start"]# 3. 实现优雅的终止
import signalclass GracefulWorkflow(DataAnalysisWorkflow):def __init__(self, *args, **kwargs):super().__init__(*args, **kwargs)self.should_stop = Falsesignal.signal(signal.SIGINT, self.handle_interrupt)def handle_interrupt(self, signum, frame):print("\nReceived interrupt signal. Cleaning up...")self.should_stop = True

这个框架可以用于很多场景，比如：

  • 数据处理管道
  • ETL工作流
  • 机器学习实验
  • 报告生成系统
  • 自动化测试流程

关键是要根据具体需求调整任务定义和执行逻辑。

http://www.hkea.cn/news/324412/

相关文章：

  • 网站搭建素材短视频培训
  • amazon虚拟机免费做网站百度信息流怎么收费
  • 深圳做网站推广公司聊城seo整站优化报价
  • 深圳专业app网站开发企业网站建设原则是
  • 网站开发师职责柳州网站建设哪里有
  • 自己做的网站怎么改电话网络推广代运营公司
  • 做水果的网站有哪些google高级搜索
  • 怎么用网站做文案百度推广可以自己开户吗
  • 做的好的新闻网站排名优化
  • 购物网站开发功能百度联盟个人怎么接广告
  • 网站如何盈利流量费网站seo搜索引擎的原理是什么
  • 泰安房产价格最新域名年龄对seo的影响
  • 网站打不开怎么回事引流推广平台有哪些
  • 课程网站建设特色成都seo外包
  • 建设厅安全员证书查询网站外链seo推广
  • 邢台手机网站建设服务百度查重软件
  • 网站开发开题报告ppt竞价运营是做什么的
  • 网站代理怎么做的网站推广策划思路
  • 长沙网站seo公司百度权重5的网站能卖多少钱
  • 常德网站开发百度推广登录首页网址
  • 网站建设软件设计推广官网
  • 网站运营阶段站长之家app
  • discuz网站标题百度广告推广价格
  • 广州学校论坛网站建设疫情排行榜最新消息
  • 古董手表网站网络营销的主要方式和技巧
  • 做公司网站要那些资料百度电脑版下载官方
  • 定州网站建设公司企业网站源码
  • 0基础1小时网站建设教程如何给自己的公司建网站
  • 成都网站建设s1emens电商平台怎么加入
  • 六合哪家做网站建设域名注册查询软件