foundationagents--openmanus
zjyhhhher
229 行
7.5 KiB
Python
229 行
7.5 KiB
Python
import sys
|
|
import asyncio
|
|
import json
|
|
import os
|
|
print(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
|
|
|
|
from typing import Any, Hashable
|
|
|
|
import pandas as pd
|
|
from pydantic import Field, model_validator
|
|
|
|
from app.config import config
|
|
from app.llm import LLM
|
|
from app.logger import logger
|
|
from app.tool.base import BaseTool
|
|
|
|
|
|
class AddInsights(BaseTool):
|
|
name: str = "add_insights"
|
|
description: str = (
|
|
"Enhances charts by adding insights markers and annotations "
|
|
"using JSON data generated by the insights_selection tool. "
|
|
"This creates the final annotated visualization output."
|
|
)
|
|
|
|
parameters: dict = {
|
|
"type": "object",
|
|
"properties": {
|
|
"json_path": {
|
|
"type": "string",
|
|
"description": """Path to the JSON file generated by insights_selection tool.
|
|
Contains chart insights data in format:
|
|
{
|
|
"chartPath": string,
|
|
"insights_id": number[]
|
|
}""",
|
|
},
|
|
"output_type": {
|
|
"type": "string",
|
|
"description": "Visualization output format selection",
|
|
"default": "html",
|
|
"enum": [
|
|
"png", # Static image format
|
|
"html" # Interactive web format (recommended)
|
|
],
|
|
},
|
|
},
|
|
"required": ["json_path"],
|
|
}
|
|
llm: LLM = Field(default_factory=LLM, description="Language model instance")
|
|
|
|
@model_validator(mode="after")
|
|
def initialize_llm(self):
|
|
"""Initialize llm with default settings if not provided."""
|
|
if self.llm is None or not isinstance(self.llm, LLM):
|
|
self.llm = LLM(config_name=self.name.lower())
|
|
return self
|
|
|
|
def load_chart_with_css(self, chart_path):
|
|
# 读取 HTML 文件
|
|
with open(chart_path, 'r', encoding='utf-8') as f:
|
|
html_content = f.read()
|
|
html_content = html_content.replace('`', "'")
|
|
|
|
# 在 <head> 里插入 CSS
|
|
css = """
|
|
<style>
|
|
body, html {
|
|
margin: 0;
|
|
padding: 0;
|
|
height: 100%;
|
|
overflow: hidden;
|
|
}
|
|
#chart-container {
|
|
width: 100%;
|
|
height: 100%;
|
|
}
|
|
</style>
|
|
"""
|
|
|
|
# 如果原文件没有 <head>，直接插入到最前面
|
|
if "<head>" in html_content:
|
|
html_content = html_content.replace("<head>", "<head>" + css)
|
|
else:
|
|
html_content = css + html_content
|
|
|
|
with open(chart_path, 'w', encoding='utf-8') as f:
|
|
f.write(html_content)
|
|
|
|
def get_file_path(
|
|
self,
|
|
json_info: list[dict[str, str]],
|
|
path_str: str,
|
|
directory: str = None,
|
|
) -> list[str]:
|
|
res = []
|
|
for item in json_info:
|
|
if os.path.exists(item[path_str]):
|
|
res.append(item[path_str])
|
|
elif os.path.exists(
|
|
os.path.join(f"{directory or config.workspace_root}", item[path_str])
|
|
):
|
|
res.append(
|
|
os.path.join(
|
|
f"{directory or config.workspace_root}", item[path_str]
|
|
)
|
|
)
|
|
else:
|
|
raise Exception(f"No such file or directory: {item[path_str]}")
|
|
return res
|
|
|
|
async def add_insights(
|
|
self, json_info: list[dict[str, str]], output_type: str
|
|
) -> str:
|
|
data_list = []
|
|
chart_file_path = self.get_file_path(
|
|
json_info, "chartPath", os.path.join(config.workspace_root, "visualization")
|
|
)
|
|
for index, item in enumerate(json_info):
|
|
if "insights_id" in item:
|
|
data_list.append(
|
|
{
|
|
"file_name": os.path.basename(chart_file_path[index]).replace(
|
|
f".{output_type}", ""
|
|
),
|
|
"insights_id": item["insights_id"],
|
|
}
|
|
)
|
|
tasks = [
|
|
self.invoke_vmind(
|
|
insights_id=item["insights_id"],
|
|
file_name=item["file_name"],
|
|
output_type=output_type,
|
|
task_type="insight",
|
|
)
|
|
for item in data_list
|
|
]
|
|
results = await asyncio.gather(*tasks)
|
|
error_list = []
|
|
success_list = []
|
|
for index, result in enumerate(results):
|
|
chart_path = chart_file_path[index]
|
|
if "error" in result and "chart_path" not in result:
|
|
error_list.append(f"Error in {chart_path}: {result['error']}")
|
|
else:
|
|
success_list.append(chart_path)
|
|
self.load_chart_with_css(chart_path)
|
|
|
|
success_template = (
|
|
f"# Charts Update with Insights\n{','.join(success_list)}"
|
|
if len(success_list) > 0
|
|
else ""
|
|
)
|
|
if len(error_list) > 0:
|
|
return {
|
|
"observation": f"# Error in chart insights:{'\n'.join(error_list)}\n{success_template}",
|
|
"success": False,
|
|
}
|
|
else:
|
|
return {"observation": f"{success_template}"}
|
|
|
|
async def execute(
|
|
self,
|
|
json_path: str,
|
|
output_type: str | None = "html",
|
|
tool_type: str | None = "visualization",
|
|
language: str | None = "en",
|
|
) -> str:
|
|
try:
|
|
logger.info(f"📈 data_visualization with {json_path} in: {tool_type} ")
|
|
with open(json_path, "r", encoding="utf-8") as file:
|
|
json_info = json.load(file)
|
|
return await self.add_insights(json_info, output_type)
|
|
except Exception as e:
|
|
return {
|
|
"observation": f"Error: {e}",
|
|
"success": False,
|
|
}
|
|
|
|
async def invoke_vmind(
|
|
self,
|
|
file_name: str,
|
|
output_type: str,
|
|
task_type: str,
|
|
insights_id: list[str] = None,
|
|
dict_data: list[dict[Hashable, Any]] = None,
|
|
chart_description: str = None,
|
|
language: str = "en",
|
|
):
|
|
llm_config = {
|
|
"base_url": self.llm.base_url,
|
|
"model": self.llm.model,
|
|
"api_key": self.llm.api_key,
|
|
}
|
|
vmind_params = {
|
|
"llm_config": llm_config,
|
|
"user_prompt": chart_description,
|
|
"dataset": dict_data,
|
|
"file_name": file_name,
|
|
"output_type": output_type,
|
|
"insights_id": insights_id,
|
|
"task_type": task_type,
|
|
"directory": str(config.workspace_root),
|
|
"language": language,
|
|
}
|
|
|
|
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__),
|
|
)
|
|
input_json = json.dumps(vmind_params, ensure_ascii=False).encode("utf-8")
|
|
try:
|
|
stdout, stderr = await process.communicate(input_json)
|
|
stdout_str = stdout.decode("utf-8")
|
|
stderr_str = stderr.decode("utf-8")
|
|
if process.returncode == 0:
|
|
return json.loads(stdout_str)
|
|
else:
|
|
return {"error": f"Node.js Error: {stderr_str}"}
|
|
except Exception as e:
|
|
return {"error": f"Subprocess Error: {str(e)}"}
|
|
|