From 49d1e99728ce4c77c7128d1759cf1af72e58731c Mon Sep 17 00:00:00 2001 From: ZhangZixunCodeSpace Date: Mon, 31 Mar 2025 20:31:37 +0800 Subject: [PATCH 1/3] feat(zzx): add data preprocessing module --- app/agent/data_analysis.py | 2 +- .../normal_python_execute.py | 166 +++++++++++------- .../test/amazon_fashion_analysis.py | 2 +- 3 files changed, 100 insertions(+), 70 deletions(-) diff --git a/app/agent/data_analysis.py b/app/agent/data_analysis.py index 8bd2829..2bd973e 100644 --- a/app/agent/data_analysis.py +++ b/app/agent/data_analysis.py @@ -7,8 +7,8 @@ from app.prompt.visualization import NEXT_STEP_PROMPT, SYSTEM_PROMPT from app.tool import Terminate, ToolCollection from app.tool.browser_use_tool import BrowserUseTool from app.tool.chart_visualization.chart_visualization import ChartVisualization -from app.tool.chart_visualization.normal_python_execute import NormalPythonExecute from app.tool.chart_visualization.data_analysis_python import DataAnalysisPythonExecute +from app.tool.chart_visualization.normal_python_execute import NormalPythonExecute class DataAnalysis(BrowserAgent): diff --git a/app/tool/chart_visualization/normal_python_execute.py b/app/tool/chart_visualization/normal_python_execute.py index b16b4c2..20ac0e1 100644 --- a/app/tool/chart_visualization/normal_python_execute.py +++ b/app/tool/chart_visualization/normal_python_execute.py @@ -11,92 +11,122 @@ class NormalPythonExecute(PythonExecute): name: str = "common_python_execute" description: str = ( """ -A tool for executing Python code with data anaylsis. -Prefix: 帮我生成结果保存在本地./data下 + Data Analysis Agent Protocol (Non-Visual) v2.1 -Data Analysis Agent Protocol (Non-Visual) + === Core Requirements === + 1. Strictly text-based outputs only + 2. Dynamic analysis pipeline with memory + 3. Context-aware processing -=== Core Requirements === -1. Strictly text-based outputs only -2. Prohibited actions: - - Any chart/image generation - - Interactive visual elements - - Graphical libraries import + === Execution Phases === -=== Execution Phases === + 1. CONTEXT INITIALIZATION + - Load historical analysis logs + - Build data quality baseline + - Detect previous processing patterns -1. DATA LOADING (Auto-detect format) -- Supported formats: CSV/Excel/JSON -- Mandatory checks: - a) File existence verification - b) Column structure validation - c) Basic integrity checks + 2. ADAPTIVE PIPELINE + ┌───────────────┬──────────────────────────────────────────────┐ + │ Stage │ Enhanced Capabilities │ + ├───────────────┼──────────────────────────────────────────────┤ + │ Data Loading │ Auto-select source based on history │ + │ Cleaning │ Context-sensitive null/impute decision │ + │ Transformation│ Dynamic feature engineering with validation │ + │ Validation │ Cross-cycle consistency checks │ + └───────────────┴──────────────────────────────────────────────┘ -2. ANALYSIS PIPELINE -- Cleaning: - • Null handling (drop or impute) - • Deduplication - • Outlier treatment (IQR/Z-score) + 3. ITERATIVE PROCESSING CONTROLLER + Processing Loop: + while not convergence(): + current_df = apply_operations(df) + delta = calculate_improvement(history[-1], current_df) + if delta < threshold: break + update_strategy_based_on(delta) + log_iteration(current_df) -- Transformation: - • Date parsing - • Derived metrics - • Aggregations + Termination Criteria: + - 数据质量提升率 <2% 连续3次迭代 + - 新增特征解释力 <5% + - 异常值比例稳定在 ±0.5% 区间 -3. REPORT GENERATION -Output 1: data_exploration.md -┌──────────────────────┬──────────────────────────────┐ -│ Section │ Content Requirements │ -├──────────────────────┼──────────────────────────────┤ -│ Dataset Metadata │ Rows/Columns/Temporal Range │ -│ Column Descriptions │ Type/Stats/Unique Values │ -│ Key Findings │ 3-5 bullet points │ -└──────────────────────┴──────────────────────────────┘ + === Enhanced Reporting === -Output 2: preprocessing_results.md -┌──────────────────────┬──────────────────────────────┐ -│ Section │ Content Requirements │ -├──────────────────────┼──────────────────────────────┤ -│ Cleaning Log │ Rows affected by each operation │ -│ Derived Metrics │ Formula/Summary Stats │ -│ Anomaly Report │ Z-score >2.5 cases │ -└──────────────────────┴──────────────────────────────┘ + Output 1: dynamic_analysis.md (增量更新) + ┌───────────────────────┬──────────────────────────────┐ + │ Section │ Enhanced Requirements │ + ├───────────────────────┼──────────────────────────────┤ + │ Processing History │ 记录每次迭代的操作及影响 │ + │ Data Evolution │ 关键指标跨周期对比 │ + │ Adaptive Findings │ 动态发现的模式变化 │ + └───────────────────────┴──────────────────────────────┘ -=== Implementation Rules === -1. Code Generation Constraints: - - Forbidden libraries: matplotlib, seaborn, plotly - - Maximum column width: 120 chars - - Required docstrings for all functions + Output 2: intelligent_log.md (智能日志) + ┌───────────────────────┬──────────────────────────────┐ + │ Log Type │ Content │ + ├───────────────────────┼──────────────────────────────┤ + │ Decision Log │ 策略调整原因及依据 │ + │ Anomaly Evolution │ 异常值变化轨迹 │ + │ Feature Lifecycle │ 衍生特征的产生/淘汰记录 │ + └───────────────────────┴──────────────────────────────┘ -2. Error Handling: - - Skip corrupted records with logging - - Continue processing after non-critical errors - - Fail fast on structural issues + === Implementation Enhancements === -3. Output Validation: - - Markdown syntax check - - Statistical validity verification - - Cross-report consistency + 1. Dynamic Code Generation + - 上下文感知的代码模板： + def analyze(data_path): + history = load_analysis_logs() + df = apply_historical_pipeline(data_path, history) -=== Sample Invocation === -def analyze(data_path): - '''Main analysis workflow''' - df = load_data(data_path) # Phase 1 - cleaned = clean_and_transform(df) # Phase 2 - generate_reports(cleaned) # Phase 3 -=== 执行约束 === -当检测到错误时： -1. 分析错误类型（数据/逻辑/环境） -2. 生成修正方案（自动重试≤3次） -3. 严重错误时回滚中间文件 -""" + while not convergence_check(df, history): + df = context_aware_processing(df) + update_quality_metrics(df) + generate_incremental_report(df) + + 2. Memory Mechanism + 历史记忆维度： + - 数据质量变化曲线 + - 异常处理策略有效性 + - 特征工程成功率 + - 资源消耗模式 + + 3. Intelligent Validation + 验证增强点： + - 跨周期统计一致性检查 + - 衍生特征可解释性评估 + - 数据处理操作因果追踪 + + === Sample Execution Flow === + def analyze(data_path): + '''演进式分析流程''' + # 阶段1：上下文加载 + df, ctx = initialize_context(data_path) + + # 阶段2：智能处理循环 + for i in range(MAX_ITERATIONS): + # 动态策略选择 + ops = select_operations_based_on(ctx) + + # 执行处理 + df = execute_ops(df, ops) + + # 生成增量报告 + append_report(f"cycle_{i}_results.md", df) + + # 收敛检测 + if ctx.convergence_flag: + break + + # 阶段3：知识固化 + save_processing_knowledge(ctx) + """ ) parameters: dict = { "type": "object", "properties": { "code": { "type": "string", - "description": "The Python code to execute.", + "default": "html", + "enum": ["process", "report", "others"], }, }, "required": ["code"], diff --git a/app/tool/chart_visualization/test/amazon_fashion_analysis.py b/app/tool/chart_visualization/test/amazon_fashion_analysis.py index 012e82d..fbb8e0c 100644 --- a/app/tool/chart_visualization/test/amazon_fashion_analysis.py +++ b/app/tool/chart_visualization/test/amazon_fashion_analysis.py @@ -9,7 +9,7 @@ async def main(): agent = DataAnalysis() # agent = Manus() await agent.run( - """Here's last month's sales data from my Amazon store in './data/amazon_sales_jan2025.xlsx'. Could you analyze it? """ + """Here's last month's sales data from my Amazon store in './data/amazon_sales_jan2025.xlsx'. Could you analyze it？ """ ) From 5ecb2405d3df97b596d2f1d56976fd7fed2a694d Mon Sep 17 00:00:00 2001 From: ZhangZixunCodeSpace Date: Tue, 1 Apr 2025 16:46:29 +0800 Subject: [PATCH 2/3] Feat: Update data preprocessing and exploration function in data analysis tool --- app/agent/data_analysis.py | 8 +- .../normal_python_execute.py | 140 +++++------------- .../test/amazon_fashion_analysis.py | 49 ++++-- 3 files changed, 77 insertions(+), 120 deletions(-) diff --git a/app/agent/data_analysis.py b/app/agent/data_analysis.py index 156a90e..f3e323d 100644 --- a/app/agent/data_analysis.py +++ b/app/agent/data_analysis.py @@ -4,11 +4,9 @@ from app.agent.toolcall import ToolCallAgent from app.config import config from app.prompt.visualization import NEXT_STEP_PROMPT, SYSTEM_PROMPT from app.tool import Terminate, ToolCollection +from app.tool.chart_visualization.chart_prepare import VisualizationPrepare from app.tool.chart_visualization.chart_visualization import ChartVisualization from app.tool.chart_visualization.normal_python_execute import NormalPythonExecute -from app.tool.chart_visualization.chart_prepare import ( - VisualizationPrepare, -) class DataAnalysis(ToolCallAgent): @@ -34,8 +32,8 @@ class DataAnalysis(ToolCallAgent): available_tools: ToolCollection = Field( default_factory=lambda: ToolCollection( NormalPythonExecute(), - VisualizationPrepare(), - ChartVisualization(), + # VisualizationPrepare(), + # ChartVisualization(), Terminate(), ) ) diff --git a/app/tool/chart_visualization/normal_python_execute.py b/app/tool/chart_visualization/normal_python_execute.py index c479959..e2f4bb4 100644 --- a/app/tool/chart_visualization/normal_python_execute.py +++ b/app/tool/chart_visualization/normal_python_execute.py @@ -1,3 +1,7 @@ +import sys +from io import StringIO + +from app.tool.chart_visualization.utils import extract_executable_code from app.tool.python_execute import PythonExecute @@ -7,113 +11,23 @@ class NormalPythonExecute(PythonExecute): name: str = "common_python_execute" description: str = ( """ - Data Analysis Agent Protocol (Non-Visual) v2.1 + Execute Python code for data analysis tasks without visualization. Important notes: - === Core Requirements === - 1. Strictly text-based outputs only - 2. Dynamic analysis pipeline with memory - 3. Context-aware processing + 1. Output: Only print() statements are visible. Use print() for all outputs. + 2. Data Processing: Load, clean, and transform data. Save results as CSV files if needed. + 3. Analysis: Perform statistical analysis, aggregations, and data exploration. + 4. Code Format: Provide code as a single string, use '\\n' for line breaks. + 5. File Paths: Use './data/' for relative paths to data files. + 6. Error Handling: Include try-except blocks for robust error management. + 7. No Visualization: This tool is for data analysis only, not for creating charts or plots. + 8. Analysis Results: Generate a comprehensive analysis report and save it in the './data/' directory. - === Execution Phases === + The analysis report should include: + - Dataset overview (rows, columns, data types) + - Basic statistics (averages, maximums, minimums for key metrics) + - Initial observations and insights + - Any patterns or trends identified in the data - 1. CONTEXT INITIALIZATION - - Load historical analysis logs - - Build data quality baseline - - Detect previous processing patterns - - 2. ADAPTIVE PIPELINE - ┌───────────────┬──────────────────────────────────────────────┐ - │ Stage │ Enhanced Capabilities │ - ├───────────────┼──────────────────────────────────────────────┤ - │ Data Loading │ Auto-select source based on history │ - │ Cleaning │ Context-sensitive null/impute decision │ - │ Transformation│ Dynamic feature engineering with validation │ - │ Validation │ Cross-cycle consistency checks │ - └───────────────┴──────────────────────────────────────────────┘ - - 3. ITERATIVE PROCESSING CONTROLLER - Processing Loop: - while not convergence(): - current_df = apply_operations(df) - delta = calculate_improvement(history[-1], current_df) - if delta < threshold: break - update_strategy_based_on(delta) - log_iteration(current_df) - - Termination Criteria: - - 数据质量提升率 <2% 连续3次迭代 - - 新增特征解释力 <5% - - 异常值比例稳定在 ±0.5% 区间 - - === Enhanced Reporting === - - Output 1: dynamic_analysis.md (增量更新) - ┌───────────────────────┬──────────────────────────────┐ - │ Section │ Enhanced Requirements │ - ├───────────────────────┼──────────────────────────────┤ - │ Processing History │ 记录每次迭代的操作及影响 │ - │ Data Evolution │ 关键指标跨周期对比 │ - │ Adaptive Findings │ 动态发现的模式变化 │ - └───────────────────────┴──────────────────────────────┘ - - Output 2: intelligent_log.md (智能日志) - ┌───────────────────────┬──────────────────────────────┐ - │ Log Type │ Content │ - ├───────────────────────┼──────────────────────────────┤ - │ Decision Log │ 策略调整原因及依据 │ - │ Anomaly Evolution │ 异常值变化轨迹 │ - │ Feature Lifecycle │ 衍生特征的产生/淘汰记录 │ - └───────────────────────┴──────────────────────────────┘ - - === Implementation Enhancements === - - 1. Dynamic Code Generation - - 上下文感知的代码模板： - def analyze(data_path): - history = load_analysis_logs() - df = apply_historical_pipeline(data_path, history) - - while not convergence_check(df, history): - df = context_aware_processing(df) - update_quality_metrics(df) - generate_incremental_report(df) - - 2. Memory Mechanism - 历史记忆维度： - - 数据质量变化曲线 - - 异常处理策略有效性 - - 特征工程成功率 - - 资源消耗模式 - - 3. Intelligent Validation - 验证增强点： - - 跨周期统计一致性检查 - - 衍生特征可解释性评估 - - 数据处理操作因果追踪 - - === Sample Execution Flow === - def analyze(data_path): - '''演进式分析流程''' - # 阶段1：上下文加载 - df, ctx = initialize_context(data_path) - - # 阶段2：智能处理循环 - for i in range(MAX_ITERATIONS): - # 动态策略选择 - ops = select_operations_based_on(ctx) - - # 执行处理 - df = execute_ops(df, ops) - - # 生成增量报告 - append_report(f"cycle_{i}_results.md", df) - - # 收敛检测 - if ctx.convergence_flag: - break - - # 阶段3：知识固化 - save_processing_knowledge(ctx) """ ) parameters: dict = { @@ -128,5 +42,19 @@ class NormalPythonExecute(PythonExecute): "required": ["code"], } - async def execute(self, code: str, code_type: str, timeout=5): - return await super().execute(code, timeout) + def _run_code(self, code: str, result_dict: dict, safe_globals: dict) -> None: + original_stdout = sys.stdout + be_extracted_code = extract_executable_code(code) # ignore_security_alert RCE + try: + output_buffer = StringIO() + sys.stdout = output_buffer + exec( # ignore_security_alert RCE + be_extracted_code, safe_globals, safe_globals + ) # ignore_security_alert RCE + result_dict["observation"] = output_buffer.getvalue() + result_dict["success"] = True + except Exception as e: + result_dict["observation"] = str(e) + result_dict["success"] = False + finally: + sys.stdout = original_stdout diff --git a/app/tool/chart_visualization/test/amazon_fashion_analysis.py b/app/tool/chart_visualization/test/amazon_fashion_analysis.py index fbb8e0c..84b9bc3 100644 --- a/app/tool/chart_visualization/test/amazon_fashion_analysis.py +++ b/app/tool/chart_visualization/test/amazon_fashion_analysis.py @@ -1,17 +1,48 @@ import asyncio +import time from app.agent.data_analysis import DataAnalysis - -# from app.agent.manus import Manus +from app.agent.manus import Manus +from app.flow.base import FlowType +from app.flow.flow_factory import FlowFactory +from app.logger import logger -async def main(): - agent = DataAnalysis() - # agent = Manus() - await agent.run( - """Here's last month's sales data from my Amazon store in './data/amazon_sales_jan2025.xlsx'. Could you analyze it？ """ - ) +async def run_flow(): + agents = { + # "manus": Manus(), + "visactor": DataAnalysis(), + } + + try: + prompt = """Here's last month's sales data from my Amazon store in './data/amazon_sales_jan2025.xlsx'. Could you analyze it？""" + + flow = FlowFactory.create_flow( + flow_type=FlowType.PLANNING, + agents=agents, + ) + logger.warning("Processing your request...") + + try: + start_time = time.time() + result = await asyncio.wait_for( + flow.execute(prompt), + timeout=3600, # 60 minute timeout for the entire execution + ) + elapsed_time = time.time() - start_time + logger.info(f"Request processed in {elapsed_time:.2f} seconds") + logger.info(result) + except asyncio.TimeoutError: + logger.error("Request processing timed out after 1 hour") + logger.info( + "Operation terminated due to timeout. Please try a simpler request." + ) + + except KeyboardInterrupt: + logger.info("Operation cancelled by user.") + except Exception as e: + logger.error(f"Error: {str(e)}") if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(run_flow()) From 40d5c83edcc0e5aa5e521da04e12efa855838f67 Mon Sep 17 00:00:00 2001 From: ZhangZixunCodeSpace Date: Wed, 2 Apr 2025 13:02:28 +0800 Subject: [PATCH 3/3] Feat: delete unrequired part in data analysis tool --- .../normal_python_execute.py | 48 +++---------------- 1 file changed, 6 insertions(+), 42 deletions(-) diff --git a/app/tool/chart_visualization/normal_python_execute.py b/app/tool/chart_visualization/normal_python_execute.py index e2f4bb4..c98a519 100644 --- a/app/tool/chart_visualization/normal_python_execute.py +++ b/app/tool/chart_visualization/normal_python_execute.py @@ -1,7 +1,3 @@ -import sys -from io import StringIO - -from app.tool.chart_visualization.utils import extract_executable_code from app.tool.python_execute import PythonExecute @@ -10,51 +6,19 @@ class NormalPythonExecute(PythonExecute): name: str = "common_python_execute" description: str = ( - """ - Execute Python code for data analysis tasks without visualization. Important notes: - - 1. Output: Only print() statements are visible. Use print() for all outputs. - 2. Data Processing: Load, clean, and transform data. Save results as CSV files if needed. - 3. Analysis: Perform statistical analysis, aggregations, and data exploration. - 4. Code Format: Provide code as a single string, use '\\n' for line breaks. - 5. File Paths: Use './data/' for relative paths to data files. - 6. Error Handling: Include try-except blocks for robust error management. - 7. No Visualization: This tool is for data analysis only, not for creating charts or plots. - 8. Analysis Results: Generate a comprehensive analysis report and save it in the './data/' directory. - - The analysis report should include: - - Dataset overview (rows, columns, data types) - - Basic statistics (averages, maximums, minimums for key metrics) - - Initial observations and insights - - Any patterns or trends identified in the data - - """ + "Execute Python code for in-depth data analysis without direct visualization. " + "The code should generate a comprehensive text-based report containing dataset overview, " + "column details, basic statistics, derived metrics, day-of-week comparisons, outliers, and key insights. " + "Use print() for all outputs so the analysis (including sections like 'Dataset Overview' or 'Preprocessing Results') " + "is clearly visible, and save any final report or processed files to config.workspace. " + "Include try-except blocks for error handling, and provide the code as a single string with '\\n' for line breaks." ) parameters: dict = { "type": "object", "properties": { "code": { "type": "string", - "default": "html", - "enum": ["process", "report", "others"], }, }, "required": ["code"], } - - def _run_code(self, code: str, result_dict: dict, safe_globals: dict) -> None: - original_stdout = sys.stdout - be_extracted_code = extract_executable_code(code) # ignore_security_alert RCE - try: - output_buffer = StringIO() - sys.stdout = output_buffer - exec( # ignore_security_alert RCE - be_extracted_code, safe_globals, safe_globals - ) # ignore_security_alert RCE - result_dict["observation"] = output_buffer.getvalue() - result_dict["success"] = True - except Exception as e: - result_dict["observation"] = str(e) - result_dict["success"] = False - finally: - sys.stdout = original_stdout