diff --git a/app/agent/data_analysis.py b/app/agent/data_analysis.py index 8bd2829..156a90e 100644 --- a/app/agent/data_analysis.py +++ b/app/agent/data_analysis.py @@ -1,17 +1,17 @@ from pydantic import Field -from app.agent.browser import BrowserAgent +from app.agent.toolcall import ToolCallAgent from app.config import config -from app.prompt.browser import NEXT_STEP_PROMPT as BROWSER_NEXT_STEP_PROMPT 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.chart_prepare import ( + VisualizationPrepare, +) -class DataAnalysis(BrowserAgent): +class DataAnalysis(ToolCallAgent): """ A data analysis agent that uses planning to solve various data analysis tasks. @@ -34,34 +34,8 @@ class DataAnalysis(BrowserAgent): available_tools: ToolCollection = Field( default_factory=lambda: ToolCollection( NormalPythonExecute(), - DataAnalysisPythonExecute(), + VisualizationPrepare(), ChartVisualization(), - BrowserUseTool(), Terminate(), ) ) - - async def think(self) -> bool: - """Process current state and decide next actions with appropriate context.""" - # Store original prompt - original_prompt = self.next_step_prompt - - # Only check recent messages (last 3) for browser activity - recent_messages = self.memory.messages[-3:] if self.memory.messages else [] - browser_in_use = any( - "browser_use" in msg.content.lower() - for msg in recent_messages - if hasattr(msg, "content") and isinstance(msg.content, str) - ) - - if browser_in_use: - # Override with browser-specific prompt temporarily to get browser context - self.next_step_prompt = BROWSER_NEXT_STEP_PROMPT - - # Call parent's think method - result = await super().think() - - # Restore original prompt - self.next_step_prompt = original_prompt - - return result diff --git a/app/tool/chart_visualization/__init__.py b/app/tool/chart_visualization/__init__.py index 6de1c65..9dcd09e 100644 --- a/app/tool/chart_visualization/__init__.py +++ b/app/tool/chart_visualization/__init__.py @@ -1,5 +1,5 @@ from app.tool.chart_visualization.chart_visualization import ChartVisualization -from app.tool.chart_visualization.data_analysis_python import DataAnalysisPythonExecute +from app.tool.chart_visualization.chart_prepare import VisualizationPrepare from app.tool.chart_visualization.normal_python_execute import NormalPythonExecute -__all__ = ["ChartVisualization", "DataAnalysisPythonExecute", "NormalPythonExecute"] +__all__ = ["ChartVisualization", "VisualizationPrepare", "NormalPythonExecute"] diff --git a/app/tool/chart_visualization/chart_prepare.py b/app/tool/chart_visualization/chart_prepare.py new file mode 100644 index 0000000..d0f7e68 --- /dev/null +++ b/app/tool/chart_visualization/chart_prepare.py @@ -0,0 +1,43 @@ +from app.tool.chart_visualization.normal_python_execute import NormalPythonExecute + + +class VisualizationPrepare(NormalPythonExecute): + """A tool for Chart Generation Preparation""" + + name: str = "visualization_preparation" + description: str = ( + "Using Python code to Generates structured visualization datasets with metadata. Outputs: 1) Cleaned CSV data files 2) JSON info with csv path and visualization description." + ) + parameters: dict = { + "type": "object", + "properties": { + "code": { + "type": "string", + "description": """Python code template EXCLUSIVELY for visualization prepare. Must Contains: +1. Data loading logic (handle dataframe/dict/file/url/json/web crawler) +2. Csv Data and chart description generate +2.1 Csv data (The data you want to visulazation, cleaning / transform from origin data, saved in .csv) +2.2 Chart description of csv data (The chart title or description should be concise and clear. Examples: 'Product sales distribution', 'Monthly revenue trend'.) +3. Save information in json file.( format: {"csvFilePath": string, "chartTitle": string}[] encoding='utf-8') +3. Json file saving with path print: print(json_path) +# Note +You can generate one or multiple csv data with different visualization needs. +""", + }, + }, + "required": ["code"], + } + + async def execute(self, code: str, timeout=5): + """ + Executes the provided Python code with a timeout. + + Args: + code (str): The Python code to execute. + analysis_content (str): The analysis content of current task. + timeout (int): Execution timeout in seconds. + + Returns: + Dict: Contains 'output' with execution output or error message and 'success' status. + """ + return await super().execute(code, timeout) diff --git a/app/tool/chart_visualization/chart_visualization.py b/app/tool/chart_visualization/chart_visualization.py index 8905c05..9791d89 100644 --- a/app/tool/chart_visualization/chart_visualization.py +++ b/app/tool/chart_visualization/chart_visualization.py @@ -1,8 +1,6 @@ -import subprocess import json -import base64 +import asyncio import pandas as pd -import aiofiles import os from typing import Any, Hashable from pydantic import Field, model_validator @@ -10,23 +8,20 @@ from pydantic import Field, model_validator from app.llm import LLM from app.tool.base import BaseTool from app.logger import logger +from app.config import config class ChartVisualization(BaseTool): - name: str = "generate_data_visualization" - description: str = """Visualize a statistical chart using csv data and chart description. The tool accepts local csv data file path and description of the chart, and output a chart in png or html. + name: str = "data_visualization_with_insight" + description: str = """Visualize statistical chart with JSON info from visualization_preparation tool. Outputs: 1) Charts (png/html) 2) Charts Insights (.md). Note: Each tool call generates only one single chart. """ parameters: dict = { "type": "object", "properties": { - "csv_path": { + "json_path": { "type": "string", - "description": """file path of csv data with ".csv" in the end""", - }, - "chart_description": { - "type": "string", - "description": "The chart title or description should be concise and clear. Examples: 'Product sales distribution', 'Monthly revenue trend'.", + "description": """file path of json info with ".json" in the end""", }, "output_type": { "description": "Rendering format (html=interactive)", @@ -35,7 +30,7 @@ Note: Each tool call generates only one single chart. "enum": ["png", "html"], }, }, - "required": ["code", "chart_description"], + "required": ["code"], } llm: LLM = Field(default_factory=LLM, description="Language model instance") @@ -46,40 +41,74 @@ Note: Each tool call generates only one single chart. self.llm = LLM(config_name=self.name.lower()) return self - async def execute( - self, csv_path: str, chart_description: str, output_type: str - ) -> str: - logger.info( - f"📈 Chart Generation with data and description: {chart_description} with {csv_path} " - ) + def success_output_template(self, result: list[dict[str, str]]) -> str: + content = "" + for item in result: + content += f"""## {item["title"]} +Chart saved in: {item["savedPath"]}""" + if len(item["insightsText"]) > 0: + insight_content = "" + for index, text in enumerate(item["insightsText"]): + insight_content += f"{index}. {text}\n" + content += f"""\n### Insights of Chart\n{insight_content}""" + else: + content += "\n" + return f"Chart Generated Successful! Detail is below:\n{content}" + + async def execute(self, json_path: str, output_type: str) -> str: + logger.info(f"📈 Chart Generation with json path: {json_path} ") try: - df = pd.read_csv(csv_path) - df = df.astype(object) - df = df.where(pd.notnull(df), None) - data_dict_list = df.to_json(orient="records", force_ascii=False) - result = await self.invoke_vmind( - data_dict_list, chart_description, output_type - ) - if "error" in result: + with open(json_path, "r", encoding="utf-8") as file: + json_info = json.load(file) + data_list = [] + for item in json_info: + df = pd.read_csv(item["csvFilePath"]) + df = df.astype(object) + df = df.where(pd.notnull(df), None) + data_dict_list = df.to_json(orient="records", force_ascii=False) + + data_list.append( + { + "file_name": os.path.basename(item["csvFilePath"]).replace( + ".csv", "" + ), + "dict_data": data_dict_list, + "chart_description": item["chartTitle"], + } + ) + tasks = [ + self.invoke_vmind( + item["dict_data"], + item["chart_description"], + item["file_name"], + output_type, + ) + for item in data_list + ] + + results = await asyncio.gather(*tasks) + error_list = [] + success_list = [] + for index, result in enumerate(results): + csv_path = json_info[index]["csvFilePath"] + if "error" in result: + error_list.append(f"Error in {csv_path}: {result["error"]}") + else: + success_list.append( + { + **result, + "title": json_info[index]["chart_description"], + } + ) + if len(error_list) > 0: return { - "observation": f"Error: {result["error"]}", + "observation": f"# Error chart generated{'\n'.join(error_list)}\nCharts saved successful are below: \n{self.success_output_template(success_list)}", "success": False, } - chart_file_path = csv_path.replace(".csv", f".{output_type}") - while os.path.exists(chart_file_path): - chart_file_path = chart_file_path.replace( - f".{output_type}", f"_new.{output_type}" - ) - if output_type == "png": - byte_data = base64.b64decode(result["res"]) - async with aiofiles.open(chart_file_path, "wb") as file: - await file.write(byte_data) else: - async with aiofiles.open( - chart_file_path, "w", encoding="utf-8" - ) as file: - await file.write(result["res"]) - return {"observation": f"chart successfully saved to {chart_file_path}"} + return { + "observation": f"All charts saved successful!\n{self.success_output_template(success_list)}" + } except Exception as e: return { "observation": f"Error: {e}", @@ -90,6 +119,7 @@ Note: Each tool call generates only one single chart. self, dict_data: list[dict[Hashable, Any]], chart_description: str, + file_name: str, output_type: str, ): llm_config = { @@ -102,16 +132,27 @@ Note: Each tool call generates only one single chart. "user_prompt": chart_description, "dataset": dict_data, "output_type": output_type, + "file_name": file_name, + "directory": str(config.workspace_root), } - process = subprocess.run( - ["npx", "ts-node", "src/chartVisualize.ts"], - input=json.dumps(vmind_params), - capture_output=True, - text=True, - encoding="utf-8", + print(vmind_params) + # build async sub process + process = await asyncio.create_subprocess_exec( + "npx", + "ts-node", + "src/chartVisualize.ts", + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, cwd=os.path.dirname(__file__), ) - if process.returncode == 0: - return json.loads(process.stdout) - else: - return {"error": f"Node.js Error: {process.stderr}"} + + input_json = json.dumps(vmind_params).encode("utf-8") + try: + stdout, stderr = await process.communicate(input_json) + if process.returncode == 0: + return json.loads(stdout) + else: + return {"error": f"Node.js Error: {stderr}"} + except Exception as e: + return {"error": f"Subprocess Error: {str(e)}"} diff --git a/app/tool/chart_visualization/data_analysis_python.py b/app/tool/chart_visualization/data_analysis_python.py deleted file mode 100644 index e92aed0..0000000 --- a/app/tool/chart_visualization/data_analysis_python.py +++ /dev/null @@ -1,42 +0,0 @@ -from app.tool.chart_visualization.normal_python_execute import NormalPythonExecute - - -class DataAnalysisPythonExecute(NormalPythonExecute): - """A tool for executing Python code in data analysis task with timeout and safety restrictions.""" - - name: str = "data_analysis_python_execute" - description: str = ( - "Executes Python code string in data analysis task, save data table in csv file. Note: Only print outputs are visible, function return values are not captured. Use print statements to see results." - ) - parameters: dict = { - "type": "object", - "properties": { - "code": { - "type": "string", - "description": """Python code template EXCLUSIVELY for data analysis. Must Contains: -1. Data loading logic (handle dataframe/dict/file/url/json/web crawler) -2. Data analysis (cleaning/transformation) -3. CSV saving with path print: print(csv_path) -""", - }, - "analysis_content": { - "type": "string", - "description": "Your analysis of current task, ensure your analysis is concise, clear, and easy to understand.", - }, - }, - "required": ["code"], - } - - async def execute(self, code: str, analysis_content: str, timeout=5): - """ - Executes the provided Python code with a timeout. - - Args: - code (str): The Python code to execute. - analysis_content (str): The analysis content of current task. - timeout (int): Execution timeout in seconds. - - Returns: - Dict: Contains 'output' with execution output or error message and 'success' status. - """ - return await super().execute(code, timeout) diff --git a/app/tool/chart_visualization/normal_python_execute.py b/app/tool/chart_visualization/normal_python_execute.py index 5102eb3..f49b622 100644 --- a/app/tool/chart_visualization/normal_python_execute.py +++ b/app/tool/chart_visualization/normal_python_execute.py @@ -12,9 +12,10 @@ class NormalPythonExecute(PythonExecute): name: str = "common_python_execute" description: str = ( - """Executes Python code strings. Note: + """Executes Python code strings to do data analysis. Note: 1. Only outputs from print() are visible; function return values are not captured. Use print() statements to display results -2. Applicable to scenarios **excluding data analysis and chart generation**""" +2. Do data analysis (cleaning / transform) saved in *.csv +3. Generate a data analysis report in *.md""" ) parameters: dict = { "type": "object", diff --git a/app/tool/chart_visualization/src/chartVisualize.ts b/app/tool/chart_visualization/src/chartVisualize.ts index b566153..1819a37 100644 --- a/app/tool/chart_visualization/src/chartVisualize.ts +++ b/app/tool/chart_visualization/src/chartVisualize.ts @@ -1,10 +1,28 @@ import Canvas from "canvas"; import path from "path"; +import fs from "fs"; import { readFileSync } from "fs"; -import VMind from "@visactor/vmind"; +import VMind, { ChartType } from "@visactor/vmind"; import VChart from "@visactor/vchart"; import { isString } from "@visactor/vutils"; +declare enum AlgorithmType { + OverallTrending = "overallTrend", + AbnormalTrend = "abnormalTrend", + PearsonCorrelation = "pearsonCorrelation", + SpearmanCorrelation = "spearmanCorrelation", + ExtremeValue = "extremeValue", + MajorityValue = "majorityValue", + StatisticsAbnormal = "statisticsAbnormal", + StatisticsBase = "statisticsBase", + DbscanOutlier = "dbscanOutlier", + LOFOutlier = "lofOutlier", + TurningPoint = "turningPoint", + PageHinkley = "pageHinkley", + DifferenceOutlier = "differenceOutlier", + Volatility = "volatility", +} + const getBase64 = async (spec: any, width?: number, height?: number) => { spec.animation = false; width && (spec.width = width); @@ -77,6 +95,22 @@ async function getHtmlVChart(spec: any, width: number, height: number) { `; } +function getSavedPathName( + directory: string, + fileName: string, + outputType: "html" | "png" | "json" +) { + let newFileName = fileName; + while ( + fs.existsSync( + path.join(directory, "visualization", `${newFileName}.${outputType}`) + ) + ) { + newFileName += "_new"; + } + return path.join(directory, "visualization", `${newFileName}.${outputType}`); +} + async function generateChart() { const inputData = JSON.parse(readFileSync(process.stdin.fd, "utf-8")); try { @@ -87,6 +121,8 @@ async function generateChart() { output_type: outputType = "png", width, height, + file_name: fileName, + directory, } = inputData; const { base_url: baseUrl, model, api_key: apiKey } = llm_config; const vmind = new VMind({ @@ -98,7 +134,7 @@ async function generateChart() { }, }); const jsonDataset = isString(dataset) ? JSON.parse(dataset) : dataset; - const { spec, error } = await vmind.generateChart( + const { spec, error, chartType } = await vmind.generateChart( userPrompt, undefined, jsonDataset, @@ -107,6 +143,43 @@ async function generateChart() { theme: "light", } ); + spec.title = { + text: userPrompt, + }; + const insights = []; + if ( + chartType && + [ + ChartType.BarChart, + ChartType.LineChart, + ChartType.AreaChart, + ChartType.ScatterPlot, + ChartType.DualAxisChart, + ].includes(chartType) + ) { + const { insights: vmindInsights } = await vmind.getInsights(spec, { + maxNum: 6, + algorithms: [ + AlgorithmType.OverallTrending, + AlgorithmType.AbnormalTrend, + AlgorithmType.PearsonCorrelation, + AlgorithmType.SpearmanCorrelation, + AlgorithmType.StatisticsAbnormal, + AlgorithmType.LOFOutlier, + AlgorithmType.DbscanOutlier, + AlgorithmType.MajorityValue, + AlgorithmType.PageHinkley, + AlgorithmType.TurningPoint, + AlgorithmType.StatisticsBase, + AlgorithmType.Volatility, + ], + usePolish: false, + }); + insights.push(...vmindInsights); + } + const insightsText = insights.map( + (insight) => insight.textContent?.plainText + ); if (error || !spec) { console.log( JSON.stringify({ @@ -115,15 +188,23 @@ async function generateChart() { ); return; } - if (outputType === "png") { - console.log( - JSON.stringify({ res: await getBase64(spec, width, height) }) - ); - } else { - console.log( - JSON.stringify({ res: await getHtmlVChart(spec, width, height) }) - ); + spec.insights = insights; + if (!fs.existsSync(path.join(directory, "visualization"))) { + fs.mkdirSync(path.join(directory, "visualization")); } + fs.writeFileSync( + getSavedPathName(directory, fileName, "json"), + JSON.stringify(spec, null, 2) + ); + const savedPath = getSavedPathName(directory, fileName, outputType); + if (outputType === "png") { + const base64 = await getBase64(spec, width, height); + fs.writeFileSync(savedPath, base64); + } else { + const html = await getHtmlVChart(spec, width, height); + fs.writeFileSync(savedPath, html, "utf-8"); + } + console.log(JSON.stringify({ savedPath, insightsText })); } catch (error) { console.log(JSON.stringify({ error })); } diff --git a/app/tool/chart_visualization/test/tool_test.py b/app/tool/chart_visualization/test/tool_test.py new file mode 100644 index 0000000..3101e8b --- /dev/null +++ b/app/tool/chart_visualization/test/tool_test.py @@ -0,0 +1,32 @@ +import asyncio +from app.tool.chart_visualization import ChartVisualization + + +async def mock_request(delay, value): + print("!!!!") + await asyncio.sleep(delay) # 模拟异步IO操作（如网络请求） + return value + + +async def main(): + # 创建多个异步任务 + tasks = [ + mock_request(1, "结果1"), + mock_request(2, "结果2"), + mock_request(3, "结果3"), + ] + + # 并发执行所有任务，等待全部完成 + results = await asyncio.gather(*tasks) + print(results) # 输出: ['结果1', '结果2', '结果3'] + + +async def test_chart(): + chartTool = ChartVisualization() + print(await chartTool.execute("./data/visualization_info.json", "html")) + + +if __name__ == "__main__": + asyncio.run(test_chart()) + # 运行主协程 + # asyncio.run(main())