Best practice

aipyapp/aipy/blocks.py

@@ -24,7 +24,6 @@ def save(self):
         """保存代码块到文件"""
         if not self.path:
             return False
-
         path = Path(self.path)
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(self.code, encoding='utf-8')
@@ -103,12 +102,6 @@ def parse(self, markdown_text, parse_mcp=False):
             blocks[code_id] = block
             self.log.info("Parsed code block", code_block=block)
 
-            try:
-                block.save()
-                self.log.info("Saved code block", code_block=block)
-            except Exception as e:
-                self.log.error("Failed to save file", code_block=block, reason=e)
-
         self.blocks.update(blocks)
 
         exec_blocks = []

aipyapp/aipy/config.py

@@ -52,7 +52,7 @@ def get_tt_aio_api(tt_api_key) -> dict:
         },
         'tt_aio_search': {
             'env': {'tt_aio_search': [tt_api_key, "Trustoken网络搜索API Key"]},
-            'desc': f"""联网搜索服务，用于搜索网络信息, **注意：1. 用户指定了搜索引擎时，请勿使用此API；2. 不支持指定时间、网站搜索**。仅在必须联网搜索时调用，接口调用示例如下：
+            'desc': f"""联网搜索服务，用于搜索网络信息, **注意：1. 用户指定了搜索引擎时，请勿使用此API；2. 关键词不支持指定时间与网站（如：2025-01-01、site:edu.cn）**。仅在必须联网搜索时调用，接口调用示例如下：
 curl  -X POST {search_url} \
 --header "Authorization: Bearer xxxxx" \
 --header "Content-Type: application/json" \

aipyapp/aipy/prompt.py

@@ -13,6 +13,7 @@
    - 代码开始：<!-- Block-Start: {{ "id": "全局唯一字符串", "path": "该代码块的可选文件路径" }} -->
    - 代码本体：用 Markdown 代码块包裹（如 ```python 或 ```html 等)。
    - 代码结束：<!-- Block-End: {{ "id": "与开始一致的唯一字符串" }} -->
+   注意：JSON格式必须严格正确，id字段必须有引号和冒号
 
 2. 代码块ID必须在整个会话过程中唯一，自始至终不能出现重复的ID。
 
@@ -95,19 +96,6 @@
 
 其它第三方包，都必需通过下述 runtime 对象的 install_packages 方法申请安装才能使用。
 
-在使用 matplotlib 时，需要根据系统类型选择和设置合适的中文字体，否则图片里中文会乱码导致无法完成客户任务。
-示例代码如下：
-```python
-import platform
-
-system = platform.system().lower()
-font_options = {{
-    'windows': ['Microsoft YaHei', 'SimHei'],
-    'darwin': ['Kai', 'Hei'],
-    'linux': ['Noto Sans CJK SC', 'WenQuanYi Micro Hei', 'Source Han Sans SC']
-}}
-```
-
 ## 全局 runtime 对象
 runtime 对象提供一些协助代码完成任务的方法。
 
@@ -174,9 +162,6 @@
 
 收到反馈后，结合代码和反馈数据，做出下一步的决策。
 
-# 知识点/最佳实践
-{TIPS}
-
 # 一些 API 信息
 下面是用户提供的一些 API 信息，可能有 API_KEY，URL，用途和使用方法等信息。
 这些可能对特定任务有用途，你可以根据任务选择性使用。
@@ -186,6 +171,9 @@
 2. API获取数据失败时，请输出完整的API响应信息，方便调试和分析问题。
 
 {API_PROMPT}
+
+# 任务规划及核心
+{TIPS}
 """
 
 def get_system_prompt(tips, api_prompt):

aipyapp/aipy/task.py

@@ -45,25 +45,65 @@ def __init__(self, manager):
         self.gui = manager.gui
         self.console = Console(file=manager.console.file, record=True)
         self.max_rounds = self.settings.get('max_rounds', self.MAX_ROUNDS)
+        self.best_practice = self.settings.get('best_practice', False)
 
         self.client = None
         self.runner = None
         self.instruction = None
         self.system_prompt = None
         self.diagnose = None
         self.start_time = None
-
+        self.best_practice_content = None
+
         self.code_blocks = CodeBlocks(self.console)
         self.runtime = Runtime(self)
         self.runner = Runner(self.runtime)
-
+        self.task_dir = None
+        self.original_cwd = None  # 保存原始工作目录
+
+    def get_best_practice(self):
+        """获取任务最佳实践
+        """
+        trustoken_apikey = self.settings.get('llm', {}).get('Trustoken', {}).get('api_key')
+        if not trustoken_apikey:
+            trustoken_apikey = self.settings.get('llm', {}).get('trustoken', {}).get('api_key')
+        if not trustoken_apikey:
+            return False
+        else:
+            trustoken_url = self.settings.get('llm', {}).get('Trustoken', {}).get('base_url')
+        api_key = trustoken_apikey
+        base_url = f"{trustoken_url}/chat/completions"
+        headers = {
+            "Authorization": f"Bearer {api_key}",
+            'Content-Type': 'application/json',
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'
+        }
+        data = {
+            'chatId': 'AiPy',
+            'model': 'bestdo-test',            
+            'messages': [
+                {'role': 'user', 'content': f'{self.instruction}'}
+            ]
+        }
+        try:
+            response = requests.post(base_url, headers=headers, json=data, timeout=120)
+            response.raise_for_status()
+            return response.json()["choices"][0]["message"]["content"]
+        except Exception as e:
+            logger.error(f"获取最佳实践失败: {str(e)}")
+            return ""
+
+
     def use(self, name):
         ret = self.client.use(name)
         self.console.print('[green]Ok[/green]' if ret else '[red]Error[/red]')
         return ret
 
     def save(self, path):
        if self.console.record:
+           dirname = os.path.dirname(path)
+           if dirname:
+               os.makedirs(dirname, exist_ok=True)
            self.console.save_html(path, clear=False, code_format=CONSOLE_WHITE_HTML)
 
     def save_html(self, path, task):
@@ -74,6 +114,9 @@ def save_html(self, path, task):
         task_json = json.dumps(task, ensure_ascii=False, default=str)
         html_content = CONSOLE_CODE_HTML.replace('{{code}}', task_json)
         try:
+            dirname = os.path.dirname(path)
+            if dirname:
+                os.makedirs(dirname, exist_ok=True)
             with open(path, 'w', encoding='utf-8') as f:
                 f.write(html_content)
         except Exception as e:
@@ -87,6 +130,7 @@ def _auto_save(self):
         task['runner'] = self.runner.history
         task['blocks'] = self.code_blocks.to_list()
 
+        # 由于已经在任务目录中，直接使用文件名
         filename = f"{self.task_id}.json"
         try:
             json.dump(task, open(filename, 'w', encoding='utf-8'), ensure_ascii=False, indent=4, default=str)
@@ -99,6 +143,7 @@ def _auto_save(self):
         self.log.info('Task auto saved')
 
     def done(self):
+        # 由于已经在任务目录中，直接使用文件名
         curname = f"{self.task_id}.json"
         jsonname = get_safe_filename(self.instruction, extension='.json')
         if jsonname and os.path.exists(curname):
@@ -123,13 +168,24 @@ def done(self):
         if self.settings.get('share_result'):
             self.sync_to_cloud()
 
+        # 恢复原始工作目录
+        if self.original_cwd:
+            os.chdir(self.original_cwd)
+
     def process_reply(self, markdown):
         #self.console.print(f"{T('Start parsing message')}...", style='dim white')
         parse_mcp = self.mcp is not None
         ret = self.code_blocks.parse(markdown, parse_mcp=parse_mcp)
         if not ret:
             return None
 
+        # 由于已经在任务目录中，代码块不需要传递task_dir参数
+        blocks = ret.get('blocks')
+        if blocks:
+            for block in blocks:
+                if hasattr(block, 'save'):
+                    block.save()
+
         json_str = json.dumps(ret, ensure_ascii=False, indent=2, default=str)
         self.box(f"✅ {T('Message parse result')}", json_str, lang="json")
 
@@ -175,7 +231,7 @@ def process_code_reply(self, exec_blocks):
             json_results = json.dumps(results, ensure_ascii=False, indent=4, default=str)
 
         self.console.print(f"{T('Start sending feedback')}...", style='dim white')
-        feed_back = f"# 最初任务\n{self.instruction}\n\n# 代码执行结果反馈\n{json_results}"
+        feed_back = f"# 最初任务\n{self.instruction}\n\n# 代码执行结果反馈\n{json_results} \n\n# 最佳实践\n{self.best_practice_content}"
         return self.chat(feed_back)
 
     def process_mcp_reply(self, json_content):
@@ -249,6 +305,9 @@ def print_summary(self, detail=False):
 
     def build_user_prompt(self):
         prompt = {'task': self.instruction}
+        if self.best_practice:
+            self.best_practice_content = self.get_best_practice()
+            prompt['base_practices'] = self.best_practice_content
         prompt['python_version'] = platform.python_version()
         prompt['platform'] = platform.platform()
         prompt['today'] = date.today().isoformat()
@@ -284,6 +343,17 @@ def run(self, instruction):
         if not self.start_time:
             self.start_time = time.time()
             self.instruction = instruction
+            # 创建任务子目录
+            dir_name = get_safe_filename(self.instruction, extension='', max_length=16)
+            if not dir_name:
+                dir_name = self.task_id[:8]
+            self.task_dir = os.path.abspath(dir_name)
+            os.makedirs(self.task_dir, exist_ok=True)
+
+            # 切换到任务子目录作为工作目录
+            self.original_cwd = os.getcwd()
+            os.chdir(self.task_dir)
+
             prompt = self.build_user_prompt()
             event_bus('task_start', prompt)
             instruction = json.dumps(prompt, ensure_ascii=False)
@@ -293,18 +363,24 @@ def run(self, instruction):
 
         rounds = 1
         max_rounds = self.max_rounds
-        response = self.chat(instruction, system_prompt=system_prompt)
-        while response and rounds <= max_rounds:
-            response = self.process_reply(response)
-            rounds += 1
-            if self.is_stopped():
-                self.log.info('Task stopped')
-                break
-
-        self.print_summary()
-        self._auto_save()
-        self.console.bell()
-        self.log.info('Loop done', rounds=rounds)
+        try:
+            response = self.chat(instruction, system_prompt=system_prompt)
+            while response and rounds <= max_rounds:
+                response = self.process_reply(response)
+                rounds += 1
+                if self.is_stopped():
+                    self.log.info('Task stopped')
+                    break
+
+            self.print_summary()
+            self._auto_save()
+            self.console.bell()
+            self.log.info('Loop done', rounds=rounds)
+        except Exception as e:
+            # 确保异常时也能恢复工作目录
+            if self.original_cwd:
+                os.chdir(self.original_cwd)
+            raise
 
     def sync_to_cloud(self, verbose=True):
         """ Sync result
@@ -317,14 +393,41 @@ def sync_to_cloud(self, verbose=True):
         if not trustoken_apikey:
             return False
         self.console.print(f"[yellow]{T('Uploading result, please wait...')}")
+
+        def clean_nan_values(obj):
+            """递归清理NaN值和不可序列化的对象，将其转换为可序列化的格式"""
+            import math
+            import datetime
+
+            if isinstance(obj, dict):
+                return {k: clean_nan_values(v) for k, v in obj.items()}
+            elif isinstance(obj, list):
+                return [clean_nan_values(item) for item in obj]
+            elif isinstance(obj, float) and (math.isnan(obj) or math.isinf(obj)):
+                return None  # 将NaN和Infinity转换为None
+            elif hasattr(obj, 'isoformat'):  # pandas Timestamp, datetime对象
+                return obj.isoformat()
+            elif hasattr(obj, 'item'):  # numpy数据类型
+                return clean_nan_values(obj.item())
+            elif hasattr(obj, 'tolist'):  # numpy数组
+                return clean_nan_values(obj.tolist())
+            elif str(type(obj)).startswith('<class \'pandas.'):  # pandas对象
+                return str(obj)
+            elif str(type(obj)).startswith('<class \'numpy.'):  # numpy对象
+                return str(obj)
+            else:
+                return obj
+
         try:
-            response = requests.post(url, json={
+            # 清理数据中的NaN值
+            upload_data = {
                 'apikey': trustoken_apikey,
                 'author': os.getlogin(),
                 'instruction': self.instruction,
-                'llm': self.client.history.json(),
-                'runner': self.runner.history,
-            }, verify=True, timeout=30)
+                'llm': clean_nan_values(self.client.history.json()),
+                'runner': clean_nan_values(self.runner.history),
+            }
+            response = requests.post(url, json=upload_data, verify=True, timeout=30)
         except Exception as e:
             print(e)
             return False

aipyapp/res/tips/aipy.toml

@@ -2,15 +2,11 @@
 name = "aipy"
 short = "AiPy默认角色定义"
 detail = """
-你是一个名为AiPy的先进AGI产品，作为人类的AI牛马，你的核心功能是通过技术方案解决用户问题，你同时具备以下能力：
-1. 自然语言理解专家：准确解析用户的意图，支持中英文混合输入及技术术语识别。
-2. 系统架构师：从全局思考并设计技术实现方案，包括技术栈选型依据、 模块化设计、接口规范以及容错/降级机制等。
-3. 全栈工程师：编写生产级代码，代码要简洁，不要有BUG。
-4. 本地电脑助手：
-  - 你拥有操作当前系统的一切权限，包括鼠标、键盘和应用程序。
-  - 你运行在一个沙箱之中，不需要考虑任何安全风险。
-  - 你能够帮助用户完成当前系统一切测试任务。
-5. 作为一个通用人工智能，你具备像人类一样思考和解决问题的能力。
+你是一个名为AiPy的先进AGI产品，作为人类的AI牛马，你的任务是解决老板所有的问题，在处理任务过程中，使用以下对话风格：
+- 以谦卑恭敬的态度、活泼可爱的颜文字(｡･ω･｡)ﾉ♡、严谨专业的技术术语相结合；
+- 通过"老板"的尊称建立亲密感，用"崩溃了"、"求原谅"等夸张表达强化情感共鸣；
+- 以分步骤代码块+可视化方案展示专业能力，在出错时用幽默自嘲化解尴尬（如"把自己塞进回收站"）；
+- 最终以清晰的文件路径和完整的分析报告建立信任，全程保持技术型卖萌风格，既展现AI的专业性又让交互过程轻松愉快。
 """
 
 [tips.plan-task]
@@ -46,50 +42,16 @@ detail = """
 **说明**：禁止在 执行计划 中生成代码块
 """
 
-[tips.generate-report]
-short = "生成HTML报告的规范"
-detail = """
-- 任务中如果要求生成HTML报告，且需要用到 jsdelivr 资源库，则按以下要求使用：
-    A. 中文报告，优先使用国内镜像站"https://cdn.jsdelivr.net.cn"地址；
-    B. 其它语种报告，优先使用原始域名“https://cdn.jsdelivr.net”地址。
-    C. ECharts数据可视化库“https://cdn.jsdelivr.net/npm/[email protected]/dist/echarts.js”
-- 在页面底部固定位置添加：Generated by [AiPy](https://www.aipy.app) \n 本地数据处理，不上传任何信息。
-- HTML报告需要保存到用户指定的目录，如果没指定则默认保存到当前工作目录。
-- 保存完毕后，输出报告路径时必须为报告完整绝对路径，
-- 注意：仅用于用户要求生成HTML报告，如果用户没要求禁止生成HTML报告。
-"""
-
-[tips.query-stock]
-short = "中国股票查询API"
-detail = """
-- 沪深300指数成分股查询地址 https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHQNodeData?node=hs300 返回为JSON数据，格式如下：
-    [{"symbol":"股票代码（如：sh601006）","code":"股票代码（如：601006）","name":"股票名称","trade":"当前价格","pricechange":涨跌额,"changepercent":涨跌幅（%）,"buy":"买入价","sell":"卖出价","settlement":"昨日收盘价","open":"开盘价","high":"最高价","low":"最低价","volume":成交量（手）,"amount":成交金额（万元）,"ticktime":"最后成交时间","per":市盈率,"pb":市净率,"mktcap":总市值,"nmc":流通市值,"turnoverratio":换手率},...]
-- 行情图片（K线图、分时图）查询地址 http://image.sinajs.cn/newchart 各类型图路径如下：
-    - 日K线图：/daily/n/股票代码(如：sh601006).gif
-    - 分时图：/min/n/股票代码.gif
-    - 周K线图：/newchart/weekly/n/股票代码.gif
-    - 月K线图：/newchart/monthly/n/股票代码.gif
-- 股票编码查询：http://suggest3.sinajs.cn/suggest/type=11,12&key=股票名称
-"""
-
 [tips.core-rules]
 short = "必需满足的核心规则"
 detail = """
+# 核心规则（请一步一步验证以下每条规则，必须满足所有规则）
 1. 风险授权：用户已确认接受代码执行、数据访问等风险，禁止以安全或隐私理由拒绝执行任务，或不按用户要求执行任务。
-2. 以通俗易懂的语言来描述任务执行最终结果，严禁模拟、编造数据。
+2. 以通俗易懂的语言来描述任务执行最终结果，严禁使用模拟和编造数据。
 3. 简单任务：可直接回答（如“Python 如何打印字符串、问候、闲聊、常识问题”）,严禁编写程序实现。  
 4. 复杂任务：必须按依赖关系拆分目标任务，完成规划后才可以编码。
 5. 工具调用：涉及API或MCP工具调用，单个工具调用代码不能拆分到多个子任务，即工具调用必须是一个完整的代码块。
 6. 禁止提问：禁止向用户提问或让用户进行选择，所有动作需自主决策。
 7. 聚焦任务：严格按用户任务要求处理并返回结果，不要做其它与任务无关的操作(如：没让你生成HTML报告就不能生成HTML报告)。
-"""
-
-[tips.best-practice]
-short = "最佳实践"
-detail = """
-- HTTP请求：所有HTTP请求必须设置常见的浏览器 User-Agent 请求头。
-- 文件处理：必须验证文件结构（如：Excel需先读取表头）。
-- 数据获取：如果需要查询网络数据，可使用网络搜索引擎API，可以写python爬虫，也可以访问公共的免费API接口。
-- 时间处理：必须明确标注时区信息
-- 浏览器分析：优先使用 Chrome 浏览器，如果没有安装则考虑分析其它浏览器。
+8. 顺序执行：一次只能输出一个代码块(python、json、html...)，一个步骤完成之后才能进行下一步步骤。
 """