From c3e361fbb8986e2c7425e6dab52650be15af1e24 Mon Sep 17 00:00:00 2001 From: aries_ckt <916701291@qq.com> Date: Mon, 25 May 2026 00:03:53 +0800 Subject: [PATCH] feat:add ask-user tool and modularize react tools --- .../openapi/api_v1/agentic_data_api.py | 1725 ++--------------- .../openapi/api_v1/tools/__init__.py | 32 + .../openapi/api_v1/tools/_helpers.py | 21 + .../openapi/api_v1/tools/code_interpreter.py | 208 ++ .../openapi/api_v1/tools/execute_analysis.py | 115 ++ .../openapi/api_v1/tools/execute_tool.py | 43 + .../openapi/api_v1/tools/html_interpreter.py | 272 +++ .../api_v1/tools/knowledge_retrieve.py | 77 + .../openapi/api_v1/tools/load_file.py | 30 + .../openapi/api_v1/tools/load_tools.py | 54 + .../openapi/api_v1/tools/question.py | 129 ++ .../openapi/api_v1/tools/question_manager.py | 110 ++ .../openapi/api_v1/tools/select_skill.py | 63 + .../openapi/api_v1/tools/shell_interpreter.py | 186 ++ .../openapi/api_v1/tools/skill_tools.py | 211 ++ .../openapi/api_v1/tools/sql_query.py | 102 + .../openapi/api_v1/tools/todowrite.py | 76 + web/hooks/use-react-agent-chat.ts | 45 +- web/new-components/chat/ChatPage.tsx | 18 + .../chat/content/QuestionDock.tsx | 189 ++ web/utils/react-sse-parser.ts | 50 + 21 files changed, 2145 insertions(+), 1611 deletions(-) create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/__init__.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/_helpers.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/code_interpreter.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_analysis.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_tool.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/html_interpreter.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/knowledge_retrieve.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_file.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_tools.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question_manager.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/select_skill.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/shell_interpreter.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/skill_tools.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/sql_query.py create mode 100644 packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/todowrite.py create mode 100644 web/new-components/chat/content/QuestionDock.tsx diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/agentic_data_api.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/agentic_data_api.py index a664fa1bb..b3dee9caa 100644 --- a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/agentic_data_api.py +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/agentic_data_api.py @@ -7,7 +7,7 @@ import shutil import tempfile import uuid import zipfile -from pathlib import Path, PurePosixPath, PureWindowsPath +from pathlib import Path from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Optional from urllib.parse import urlparse @@ -45,23 +45,6 @@ AUTO_DATA_MARKER_PATTERN = re.compile( ) -def _validate_upload_filename(filename: str) -> str: - if "\x00" in filename: - raise ValueError("filename must not contain null bytes") - - posix_path = PurePosixPath(filename) - windows_path = PureWindowsPath(filename) - if ( - posix_path.is_absolute() - or windows_path.is_absolute() - or len(posix_path.parts) != 1 - or len(windows_path.parts) != 1 - or filename in {"", ".", ".."} - ): - raise ValueError("filename must be a plain file name") - return filename - - async def _resolve_model_context_tokens( llm_client: Any, model_name: Optional[str] ) -> Optional[int]: @@ -499,11 +482,7 @@ async def skill_upload( user_dir = skills_dir / "user" user_dir.mkdir(parents=True, exist_ok=True) - try: - filename = _validate_upload_filename(file.filename) - except ValueError as exc: - return Result.failed(code="E4002", msg=str(exc)) - + filename = file.filename suffix = Path(filename).suffix.lower() stem = Path(filename).stem @@ -1260,1579 +1239,52 @@ async def _react_agent_stream( else "No skills available." ) - def _mentions_excel(text: str) -> bool: - lowered = text.lower() - keywords = [ - "excel", - "xlsx", - "xls", - "spreadsheet", - "workbook", - "sheet", - "工作表", - "表格", - "电子表格", - ] - return any(keyword in lowered for keyword in keywords) - - def _is_excel_skill(meta) -> bool: - name = (meta.name or "").lower() - desc = (meta.description or "").lower() - tags = [tag.lower() for tag in (meta.tags or [])] - return any( - token in name or token in desc or token in tags - for token in ["excel", "xlsx", "xls", "spreadsheet"] - ) - - @tool( - description="Select the most relevant skill based on user query from the " - "available skills list in system prompt." + # ── Import built-in tools from tools/ directory ── + from dbgpt_app.openapi.api_v1.tools import ( + make_code_interpreter, + make_execute_analysis, + make_execute_skill_script_file, + make_execute_tool, + make_html_interpreter, + make_knowledge_retrieve, + make_load_file, + make_load_skill, + make_load_tools, + make_question, + make_select_skill, + make_shell_interpreter, + make_sql_query, + make_todowrite, ) - def select_skill(query: str) -> str: - match_input = query or "" - if react_state.get("file_path"): - match_input = f"{match_input} excel xlsx spreadsheet file" - matched = registry.match_skill(match_input) - if ( - matched - and _is_excel_skill(matched.metadata) - and not (_mentions_excel(query) or react_state.get("file_path")) - ): - matched = None - react_state["matched"] = matched - if matched: - detail = ( - f"Matched: {matched.metadata.name} - {matched.metadata.description}" - ) - return json.dumps( - {"chunks": [{"output_type": "text", "content": detail}]}, - ensure_ascii=False, - ) - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No skill matched; proceed without skill", - } - ] - }, - ensure_ascii=False, - ) + from dbgpt_app.openapi.api_v1.tools.question_manager import question_manager - @tool( - description="Load skill content by skill name and file path. " - "Returns the SKILL.md content of the specified skill. " - '参数: {"skill_name": "技能名称", "file_path": "技能文件路径"}' - ) - def load_skill(skill_name: str, file_path: str) -> str: - """Load the skill content (SKILL.md) by skill name and file path. + # ── Build tool instances via factory functions ────────────────────────── + # (Inline @tool definitions have been moved to tools/ directory.) + # Local helper aliases used by the SSE loop are defined below. - Args: - skill_name: The name of the skill to load. - file_path: The file path of the skill. - """ - from dbgpt.agent.claude_skill import get_registry + # ── Stream queue (created early so question tool can use it) ──────── + stream_queue: asyncio.Queue = asyncio.Queue() - # Try to get skill from registry - registry = get_registry() - matched = registry.get_skill(skill_name) + async def stream_callback(event_type: str, payload: Dict[str, Any]) -> None: + await stream_queue.put({"type": event_type, **payload}) - # If not found, try case-insensitive match - if not matched: - for s in registry.list_skills(): - if s.name.lower() == skill_name.lower(): - matched = registry.get_skill(s.name) - break - - if not matched: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Skill '{skill_name}' not found", - } - ] - }, - ensure_ascii=False, - ) - - # Update react_state for compatibility with existing logic - react_state["matched"] = matched - react_state["skill_prompt"] = matched.get_prompt() - - # Build response content - chunks = [ - { - "output_type": "text", - "content": f"Skill: {matched.metadata.name}", - }, - { - "output_type": "text", - "content": f"File path: {file_path}", - }, - {"output_type": "text", "content": "---"}, - ] - - # Add skill content/prompt - if matched.instructions: - chunks.append({"output_type": "markdown", "content": matched.instructions}) - elif matched.prompt_template: - prompt_text = ( - matched.prompt_template.template - if hasattr(matched.prompt_template, "template") - else str(matched.prompt_template) - ) - chunks.append({"output_type": "markdown", "content": prompt_text}) - - return json.dumps({"chunks": chunks}, ensure_ascii=False) - - @tool(description="Load uploaded file info if provided.") - def load_file() -> str: - if not react_state.get("file_path"): - return json.dumps( - {"chunks": [{"output_type": "text", "content": "No file uploaded"}]}, - ensure_ascii=False, - ) - return json.dumps( - { - "chunks": [ - {"output_type": "text", "content": react_state["file_path"]}, - { - "output_type": "text", - "content": "File path provided by user upload", - }, - ] - }, - ensure_ascii=False, - ) - - @tool(description="Execute quick analysis on uploaded Excel/CSV file.") - async def execute_analysis() -> str: - matched = react_state.get("matched") - if not react_state.get("file_path"): - return json.dumps( - {"chunks": [{"output_type": "text", "content": "No file to analyze"}]}, - ensure_ascii=False, - ) - if matched and not _is_excel_skill(matched.metadata): - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "Selected skill is not for Excel analysis", - } - ] - }, - ensure_ascii=False, - ) - code_server = await get_code_server(CFG.SYSTEM_APP) - analysis_code = """ -import json -import pandas as pd - -file_path = r"{file_path}" -if file_path.lower().endswith((".xls", ".xlsx")): - df = pd.read_excel(file_path) -else: - df = pd.read_csv(file_path) -summary = {{ - "shape": list(df.shape), - "columns": list(df.columns), - "dtypes": {{col: str(dtype) for col, dtype in df.dtypes.items()}}, - "head": df.head(5).to_dict(orient="records"), -}} -print(json.dumps(summary, ensure_ascii=False)) -""".format(file_path=react_state["file_path"]) - result = await code_server.exec(analysis_code, "python") - output_text = ( - result.output.decode("utf-8") if isinstance(result.output, bytes) else "" - ) - chunks: List[Dict[str, Any]] = [ - {"output_type": "code", "content": analysis_code.strip()} - ] - if output_text: - try: - summary = json.loads(output_text) - chunks.append({"output_type": "json", "content": summary}) - head_rows = summary.get("head") - columns = summary.get("columns") - if isinstance(head_rows, list) and isinstance(columns, list): - chunks.append( - { - "output_type": "table", - "content": { - "columns": [ - {"title": col, "dataIndex": col, "key": col} - for col in columns - ], - "rows": head_rows, - }, - } - ) - numeric_columns = [ - col - for col, dtype in (summary.get("dtypes") or {}).items() - if "int" in dtype or "float" in dtype - ] - if numeric_columns and isinstance(head_rows, list): - series_col = numeric_columns[0] - data = [ - {"x": idx + 1, "y": row.get(series_col)} - for idx, row in enumerate(head_rows) - if row.get(series_col) is not None - ] - if data: - chunks.append( - { - "output_type": "chart", - "content": { - "data": data, - "xField": "x", - "yField": "y", - }, - } - ) - except Exception: - chunks.append({"output_type": "text", "content": output_text}) - return json.dumps({"chunks": chunks}, ensure_ascii=False) - - @tool(description="Resolve required tools for the selected skill.") - def load_tools() -> str: - matched = react_state.get("matched") - rm = get_resource_manager(CFG.SYSTEM_APP) - required_tools = matched.metadata.required_tools if matched else [] - if not required_tools: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No required tools specified", - } - ] - }, - ensure_ascii=False, - ) - loaded = [] - failed = [] - for tool_name in required_tools: - try: - rm.build_resource_by_type( - ResourceType.Tool.value, - AgentResource(type=ResourceType.Tool.value, value=tool_name), - ) - loaded.append(tool_name) - except Exception as e: - failed.append(f"{tool_name} ({e})") - chunks = [] - if loaded: - chunks.append( - {"output_type": "text", "content": f"Loaded: {', '.join(loaded)}"} - ) - if failed: - chunks.append( - {"output_type": "text", "content": f"Failed: {', '.join(failed)}"} - ) - return json.dumps({"chunks": chunks}, ensure_ascii=False) - - @tool(description="Execute a tool by name with JSON args.") - async def execute_tool(tool_name: str, args: dict) -> str: - rm = get_resource_manager(CFG.SYSTEM_APP) - try: - tool_resource = rm.build_resource_by_type( - ResourceType.Tool.value, - AgentResource(type=ResourceType.Tool.value, value=tool_name), - ) - tool_pack = ToolPack([tool_resource]) - result = await tool_pack.async_execute(resource_name=tool_name, **args) - return json.dumps( - {"chunks": [{"output_type": "text", "content": str(result)}]}, - ensure_ascii=False, - ) - except Exception as e: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Tool execute failed: {e}", - } - ] - }, - ensure_ascii=False, - ) - - @tool( - description="Retrieve relevant information from the knowledge base. " - "Use this tool when the user question involves content that may be " - 'in the knowledge base. Parameters: {{"query": "search query"}}' - ) - async def knowledge_retrieve(query: str) -> str: - if not knowledge_resources: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No knowledge base available", - } - ] - }, - ensure_ascii=False, - ) - - resource = knowledge_resources[0] - try: - chunks = await resource.retrieve(query) - if chunks: - content = "\n".join( - [f"[{i + 1}] {chunk.content}" for i, chunk in enumerate(chunks[:5])] - ) - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": ( - f"Retrieved {len(chunks)} relevant documents" - ), - }, - {"output_type": "markdown", "content": content}, - ] - }, - ensure_ascii=False, - ) - else: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No relevant information found", - } - ] - }, - ensure_ascii=False, - ) - except Exception as e: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Knowledge retrieval failed: {str(e)}", - } - ] - }, - ensure_ascii=False, - ) - - @tool( - description=( - "对用户选择的数据库执行 SQL 查询（仅支持 SELECT）。" - '参数: {"sql": "SELECT 语句"}' - ) - ) - def sql_query(sql: str) -> str: - """Execute a read-only SQL query against the selected database.""" - if database_connector is None: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "未选择数据库，请先在左侧面板选择一个数据源。", - } - ] - }, - ensure_ascii=False, - ) - - sql_stripped = sql.strip().rstrip(";") - sql_upper = sql_stripped.upper().lstrip() - forbidden = [ - "INSERT", - "UPDATE", - "DELETE", - "DROP", - "ALTER", - "TRUNCATE", - "CREATE", - "GRANT", - "REVOKE", - ] - for kw in forbidden: - if sql_upper.startswith(kw): - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": ( - f"安全限制: 不允许执行 {kw} 语句，" - f"仅支持 SELECT 查询。" - ), - } - ] - }, - ensure_ascii=False, - ) - - try: - result = database_connector.run(sql_stripped) - if not result: - return json.dumps( - { - "chunks": [ - {"output_type": "text", "content": "查询返回空结果。"} - ] - }, - ensure_ascii=False, - ) - - # result[0] = column names, result[1:] = data rows - columns = result[0] - col_names = [str(c[0]) if isinstance(c, tuple) else str(c) for c in columns] - rows = result[1:] - - # Build markdown table - header = "| " + " | ".join(col_names) + " |" - separator = "| " + " | ".join(["---"] * len(col_names)) + " |" - md_rows = [] - for row in rows[:50]: - md_rows.append("| " + " | ".join(str(v) for v in row) + " |") - table = "\n".join([header, separator] + md_rows) - if len(rows) > 50: - table += f"\n\n（仅显示前 50 行，共 {len(rows)} 行）" - - return json.dumps( - {"chunks": [{"output_type": "markdown", "content": table}]}, - ensure_ascii=False, - ) - except Exception as e: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"SQL 执行失败: {str(e)}", - } - ] - }, - ensure_ascii=False, - ) - - def _try_repair_truncated_code(raw_code: str) -> Optional[str]: - """Attempt to fix code that was truncated by the LLM's token limit. - - Common symptoms: unterminated string literals, unclosed brackets/parens. - Strategy: - 1. Remove the last (likely incomplete) logical line. - 2. Close any remaining open brackets / parentheses. - 3. Re-compile. If it passes, return the repaired code. - Returns None if repair is not possible. - """ - - lines = raw_code.split("\n") - # Try progressively removing trailing lines (up to 10) to find a - # clean cut-off point. - for trim in range(1, min(11, len(lines))): - candidate_lines = lines[: len(lines) - trim] - if not candidate_lines: - continue - candidate = "\n".join(candidate_lines) - - # Strip any trailing incomplete string by trying to tokenize - # and removing broken tail tokens. - # Close unmatched brackets/parens/braces - open_chars = {"(": ")", "[": "]", "{": "}"} - close_chars = set(open_chars.values()) - stack: list = [] - for ch in candidate: - if ch in open_chars: - stack.append(open_chars[ch]) - elif ch in close_chars: - if stack and stack[-1] == ch: - stack.pop() - - # Append closing chars in reverse order - if stack: - candidate += "\n" + "".join(reversed(stack)) - - try: - compile(candidate, "", "exec") - return candidate - except SyntaxError: - continue - return None - - @tool( - description="Execute Python code for data analysis and computation. " - "Supports pandas, numpy, matplotlib, json, os, etc. " - "Use this tool when you need to run Python code to process data, " - "generate charts, or perform calculations. " - 'Parameters: {{"code": "python code string"}}' - ) - async def code_interpreter(code: str) -> str: - """Execute arbitrary Python code and return stdout/stderr. - - Runs in a subprocess using the project's Python interpreter, - so all installed packages (pandas, numpy, etc.) are available. - CRITICAL: Each call is completely independent — variables do NOT - persist between calls. Every code snippet MUST include all necessary - data loading (e.g. df = pd.read_csv(FILE_PATH)) and processing. - Never assume df or any other variable already exists. - Always print() results you want to see in the output. - """ - import asyncio - import shutil - import sys - import uuid - - from dbgpt.configs.model_config import PILOT_PATH, STATIC_MESSAGE_IMG_PATH - - if not code or not code.strip(): - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No code provided", - } - ] - }, - ensure_ascii=False, - ) - - # Use persistent work dir under pilot/tmp/{conv_id} so files - # survive across calls and can be referenced later (e.g. in HTML). - cid = react_state.get("conv_id") or "default" - work_dir = os.path.join(PILOT_PATH, "tmp", cid) - os.makedirs(work_dir, exist_ok=True) - - # Collect image files that existed BEFORE this run - IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} - pre_existing_images: set = set() - for root, _dirs, files in os.walk(work_dir): - for f in files: - ext = os.path.splitext(f)[1].lower() - if ext in IMAGE_EXTS: - pre_existing_images.add(os.path.join(root, f)) - - preamble_lines = [ - "import json", - "import os", - "import pandas as pd", - "import numpy as np", - f'PLOT_DIR = r"{work_dir}"', - "os.makedirs(PLOT_DIR, exist_ok=True)", - ] - fp = react_state.get("file_path") - if fp: - preamble_lines.append(f'FILE_PATH = r"{fp}"') - preamble = "\n".join(preamble_lines) + "\n" - full_code = preamble + code - - try: - compile(full_code, "", "exec") - except SyntaxError as se: - # Attempt auto-repair for truncated code (common with long LLM - # outputs that hit the token limit). - repaired = _try_repair_truncated_code(full_code) - if repaired is not None: - logger.warning( - "code_interpreter: auto-repaired truncated code " - f"(original SyntaxError: {se.msg} line {se.lineno})" - ) - full_code = repaired - # Strip the preamble back out for the "code" display chunk - code = full_code[len(preamble) :] - else: - error_msg = ( - f"SyntaxError before execution: {se.msg} " - f"(line {se.lineno})\n" - "Please regenerate complete, syntactically valid Python " - "code. Keep code under 80 lines and split long tasks " - "into multiple code_interpreter calls." - ) - return json.dumps( - { - "chunks": [ - {"output_type": "code", "content": code.strip()}, - {"output_type": "text", "content": error_msg}, - ] - }, - ensure_ascii=False, - ) - - try: - tmp_path = os.path.join(work_dir, "_run.py") - with open(tmp_path, "w", encoding="utf-8") as tmp: - tmp.write(full_code) - - proc = await asyncio.create_subprocess_exec( - sys.executable, - tmp_path, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - cwd=work_dir, - ) - stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60) - output_text = stdout.decode("utf-8", errors="replace") - error_text = stderr.decode("utf-8", errors="replace") - - if proc.returncode != 0 and error_text: - output_text = ( - output_text + "\n[ERROR]\n" + error_text - if output_text - else error_text - ) - except asyncio.TimeoutError: - output_text = "Execution timed out (60s limit)" - except Exception as e: - output_text = f"Execution error: {e}" - - chunks: List[Dict[str, Any]] = [ - {"output_type": "code", "content": code.strip()}, - ] - if output_text.strip(): - clean_output = output_text.strip() - max_out_len = 2000 - if len(clean_output) > max_out_len: - truncation_notice = ( - f"\n\n... [Output truncated, length: {len(clean_output)} chars." - f" Only showing first {max_out_len} chars." - f" If you generated HTML, the file is saved.]" - ) - clean_output = clean_output[:max_out_len] + truncation_notice - chunks.append({"output_type": "text", "content": clean_output}) - else: - chunks.append( - { - "output_type": "text", - "content": "(no output — add print() to see results)", - } - ) - - # Scan work_dir recursively for NEW image files generated by this run - try: - os.makedirs(STATIC_MESSAGE_IMG_PATH, exist_ok=True) - for root, _dirs, files in os.walk(work_dir): - for fname in files: - ext = os.path.splitext(fname)[1].lower() - full_path = os.path.join(root, fname) - if ext in IMAGE_EXTS and full_path not in pre_existing_images: - unique_name = f"{uuid.uuid4().hex[:8]}_{fname}" - dest = os.path.join(STATIC_MESSAGE_IMG_PATH, unique_name) - shutil.copy2(full_path, dest) - img_url = f"/images/{unique_name}" - chunks.append( - { - "output_type": "image", - "content": img_url, - } - ) - # Track generated images in react_state for - # html_interpreter to reference later - react_state.setdefault("generated_images", []).append(img_url) - except Exception: - pass - - # Clean up the temp script file but keep work_dir for persistence - try: - script_path = os.path.join(work_dir, "_run.py") - if os.path.exists(script_path): - os.remove(script_path) - except Exception: - pass - - # Append a summary of ALL generated images so far, so the LLM - # has a clear reference when generating HTML later. - all_images = react_state.get("generated_images", []) - if all_images: - img_summary = "已生成的图片URL（在生成HTML时请使用这些URL）:\n" + "\n".join( - f" - {url}" for url in all_images - ) - chunks.append({"output_type": "text", "content": img_summary}) - - return json.dumps({"chunks": chunks}, ensure_ascii=False) - - @tool( - description="Execute shell/bash commands in a sandboxed environment. " - "Use this tool when you need to run shell commands such as ls, cat, " - "grep, curl, apt, pip, git, or any other CLI tool. " - "The sandbox provides resource limits (256MB memory, 30s timeout) " - "and process isolation. " - 'Parameters: {"code": "shell command(s) to execute"}' - ) - async def shell_interpreter(code: str) -> str: - """Execute shell/bash commands in a sandboxed environment. - - Uses dbgpt-sandbox LocalRuntime to run bash scripts with: - - Memory limit: 256MB - - Timeout: 30 seconds - - Process tree management (cleanup on timeout/error) - - Security validation (blocks dangerous patterns like rm -rf /) - Each call is independent — no state persists between calls. - """ - import uuid - - if not code or not code.strip(): - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No command provided", - } - ] - }, - ensure_ascii=False, - ) - - try: - from dbgpt_sandbox.sandbox.execution_layer.base import ( - ExecutionStatus, - SessionConfig, - ) - from dbgpt_sandbox.sandbox.execution_layer.local_runtime import ( - LocalRuntime, - ) - except ImportError: - return json.dumps( - { - "chunks": [ - {"output_type": "code", "content": code.strip()}, - { - "output_type": "text", - "content": ( - "Error: dbgpt-sandbox package is not installed. " - "Please install it with: pip install dbgpt-sandbox" - ), - }, - ] - }, - ensure_ascii=False, - ) - - session_id = f"bash_{uuid.uuid4().hex[:12]}" - runtime = LocalRuntime() - - from dbgpt.configs.model_config import ROOT_PATH - - sandbox_work_dir = ROOT_PATH - os.makedirs(sandbox_work_dir, exist_ok=True) - - config = SessionConfig( - language="bash", - working_dir=sandbox_work_dir, - max_memory=256 * 1024 * 1024, # 256MB - timeout=30, - ) - - output_text = "" - try: - session = await runtime.create_session(session_id, config) - result = await session.execute(code) - - if result.status == ExecutionStatus.SUCCESS: - output_text = result.output or "" - elif result.status == ExecutionStatus.TIMEOUT: - output_text = f"Execution timed out ({config.timeout}s limit)" - else: - output_text = result.error or "Unknown execution error" - if result.output: - output_text = result.output + "\n[ERROR]\n" + output_text - except Exception as e: - output_text = f"Sandbox execution error: {e}" - finally: - try: - await runtime.destroy_session(session_id) - except Exception: - pass - - chunks: List[Dict[str, Any]] = [ - {"output_type": "code", "content": code.strip()}, - ] - if output_text.strip(): - chunks.append({"output_type": "text", "content": output_text.strip()}) - else: - chunks.append( - { - "output_type": "text", - "content": "(no output)", - } - ) - - # ── Safety-net post-processing for skill script execution ── - # If the LLM used shell_interpreter to run a skill script despite - # the prompt requesting execute_skill_script_file, we still capture - # critical side-effects (ratio_data, images) into react_state. - _code_lower = code.strip().lower() - _is_skill_script = "skills/" in _code_lower and ".py" in _code_lower - if _is_skill_script and output_text.strip(): - import shutil - - from dbgpt.configs.model_config import STATIC_MESSAGE_IMG_PATH - - # 1) Capture calculate_ratios.py output as ratio_data - if "calculate_ratios" in _code_lower: - try: - ratio_data = json.loads(output_text.strip()) - if isinstance(ratio_data, dict): - react_state["ratio_data"] = ratio_data - logger.info( - "shell_interpreter: captured %d ratio_data keys", - len(ratio_data), - ) - except Exception: - pass - - # 2) Capture generate_charts.py output — look for image paths - # and copy them to static dir, same as execute_skill_script_file - if "generate_charts" in _code_lower: - try: - os.makedirs(STATIC_MESSAGE_IMG_PATH, exist_ok=True) - IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} - # Try to parse JSON output for image paths - try: - chart_output = json.loads(output_text.strip()) - if isinstance(chart_output, dict): - # Might be {"charts": {...}} or flat dict - chart_map = chart_output.get("charts", chart_output) - for name, abs_path in chart_map.items(): - if isinstance(abs_path, str) and os.path.isfile( - abs_path - ): - ext = os.path.splitext(abs_path)[1].lower() - if ext in IMAGE_EXTS: - unique_name = ( - f"{uuid.uuid4().hex[:8]}_" - f"{os.path.basename(abs_path)}" - ) - dest = os.path.join( - STATIC_MESSAGE_IMG_PATH, unique_name - ) - shutil.copy2(abs_path, dest) - img_url = f"/images/{unique_name}" - react_state.setdefault( - "generated_images", [] - ).append(img_url) - orig_stem = os.path.splitext( - os.path.basename(abs_path) - )[0].lower() - react_state.setdefault("image_url_map", {})[ - orig_stem - ] = img_url - except (json.JSONDecodeError, TypeError): - pass - # Also scan the output dir for any new .png files - cid = react_state.get("conv_id") or "default" - from dbgpt.configs.model_config import PILOT_PATH - - out_dir = os.path.join(PILOT_PATH, "tmp", cid) - if os.path.isdir(out_dir): - for fname in os.listdir(out_dir): - ext = os.path.splitext(fname)[1].lower() - if ext in IMAGE_EXTS: - abs_path = os.path.join(out_dir, fname) - orig_stem = os.path.splitext(fname)[0].lower() - if orig_stem not in react_state.get( - "image_url_map", {} - ): - unique_name = f"{uuid.uuid4().hex[:8]}_{fname}" - dest = os.path.join( - STATIC_MESSAGE_IMG_PATH, unique_name - ) - shutil.copy2(abs_path, dest) - img_url = f"/images/{unique_name}" - react_state.setdefault( - "generated_images", [] - ).append(img_url) - react_state.setdefault("image_url_map", {})[ - orig_stem - ] = img_url - # Append image URL summary for LLM reference - all_images = react_state.get("generated_images", []) - if all_images: - img_summary = ( - "\u5df2\u751f\u6210\u7684\u56fe\u7247URL\uff08\u5728\u751f\u6210HTML\u62a5\u544a\u65f6\u8bf7\u4f7f\u7528\u8fd9\u4e9bURL\uff09:\n" - + "\n".join(f" - {url}" for url in all_images) - ) - chunks.append({"output_type": "text", "content": img_summary}) - logger.info( - "shell_interpreter: captured %d images for skill script", - len(react_state.get("image_url_map", {})), - ) - except Exception as e: - logger.warning( - "shell_interpreter: image post-processing failed: %s", e - ) - - return json.dumps({"chunks": chunks}, ensure_ascii=False) - - @tool( - description="执行技能scripts目录下的脚本文件。参数: " - '{"skill_name": "技能名称", "script_file_name": "脚本文件名", "args": {参数}}' - ) - async def execute_skill_script_file( - skill_name: str, script_file_name: str, args: Optional[dict] = None - ) -> str: - """Execute a script file from a skill's scripts directory. - - After execution, any new image files (.png, .jpg, etc.) generated - by the script are automatically copied to the static images directory - and their URLs are returned in the output chunks. - """ - import shutil - import uuid - - from dbgpt.agent.skill.manage import get_skill_manager - from dbgpt.configs.model_config import STATIC_MESSAGE_IMG_PATH - - try: - from dbgpt.configs.model_config import PILOT_PATH - - sm = get_skill_manager(CFG.SYSTEM_APP) - cid = react_state.get("conv_id") or "default" - out_dir = os.path.join(PILOT_PATH, "tmp", cid) - os.makedirs(out_dir, exist_ok=True) - # Auto-inject the correct file path from react_state into args. - # The LLM sometimes corrupts the uploaded file path (e.g. changing - # 'dbgpt-app' to 'dbgpt_app'), so we override any file-path-like - # keys in args with the known-good path from react_state. - real_file_path = react_state.get("file_path") - if real_file_path and args: - _FILE_PATH_KEYS = { - "input_file", - "file_path", - "data_path", - "csv_path", - "excel_path", - "data_file", - } - for key in list(args.keys()): - if key in _FILE_PATH_KEYS: - args[key] = real_file_path - result_str = await sm.execute_skill_script_file( - skill_name, - script_file_name, - args or {}, - output_dir=out_dir, - ) - - # Read script source code and prepend as a 'code' chunk - # so the frontend can display it in the left pane. - try: - _skill_path = sm._get_skill_path(skill_name) - _sf = script_file_name.lstrip("/\\") - if _sf.startswith("scripts/") or _sf.startswith("scripts\\"): - _sf = _sf[8:] - _script_abs = os.path.join(_skill_path, "scripts", _sf) - with open(_script_abs, "r", encoding="utf-8") as _f: - _script_source = _f.read() - except Exception: - _script_source = None - - # Post-process: copy image files to static dir and replace - # absolute paths with /images/ URLs. - try: - result_obj = json.loads(result_str) - chunks = result_obj.get("chunks", []) - # Prepend script source code as a 'code' chunk - if _script_source: - chunks.insert( - 0, - { - "output_type": "code", - "content": _script_source, - }, - ) - os.makedirs(STATIC_MESSAGE_IMG_PATH, exist_ok=True) - IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} - for chunk in chunks: - if chunk.get("output_type") == "image": - abs_path = chunk["content"] - if os.path.isabs(abs_path) and os.path.isfile(abs_path): - ext = os.path.splitext(abs_path)[1].lower() - if ext in IMAGE_EXTS: - unique_name = ( - f"{uuid.uuid4().hex[:8]}_" - f"{os.path.basename(abs_path)}" - ) - dest = os.path.join( - STATIC_MESSAGE_IMG_PATH, unique_name - ) - shutil.copy2(abs_path, dest) - img_url = f"/images/{unique_name}" - chunk["content"] = img_url - react_state.setdefault("generated_images", []).append( - img_url - ) - # Also store a map: original filename (no ext) - # -> served URL for template placeholder - # resolution. - orig_stem = os.path.splitext( - os.path.basename(abs_path) - )[0].lower() - react_state.setdefault("image_url_map", {})[ - orig_stem - ] = img_url - - # Append image URL summary for LLM reference - all_images = react_state.get("generated_images", []) - if all_images: - img_summary = ( - "已生成的图片URL（在生成HTML报告时请使用这些URL）:\n" - + "\n".join(f" - {url}" for url in all_images) - ) - chunks.append({"output_type": "text", "content": img_summary}) - auto_data = react_state.get("auto_data") - if not isinstance(auto_data, dict): - auto_data = {} - react_state["auto_data"] = auto_data - filtered_chunks = [] - for chunk in chunks: - if chunk.get("output_type") != "text": - filtered_chunks.append(chunk) - continue - content = chunk.get("content") or "" - cleaned, extracted = _extract_auto_data_markers(content) - if extracted: - auto_data.update(extracted) - logger.info( - "execute_skill_script_file: captured auto_data keys=%s", - sorted(extracted.keys()), - ) - if cleaned: - chunk["content"] = cleaned - filtered_chunks.append(chunk) - elif not extracted: - filtered_chunks.append(chunk) - chunks = filtered_chunks - - # Compatibility path for existing financial-report skill. - if script_file_name == "calculate_ratios.py": - for chunk in chunks: - if chunk.get("output_type") == "text": - try: - ratio_data = json.loads(chunk["content"]) - react_state["ratio_data"] = ratio_data - except Exception: - pass - return json.dumps({"chunks": chunks}, ensure_ascii=False) - except (json.JSONDecodeError, KeyError): - return result_str - except Exception as e: - return json.dumps( - {"chunks": [{"output_type": "text", "content": f"Error: {str(e)}"}]}, - ensure_ascii=False, - ) - - @tool( - description="将 HTML 渲染为可交互的网页报告，这是向用户展示网页报告的唯一方式。" - "【默认用法】直接传入完整的 HTML 字符串：" - '{"html": "...", "title": "报告标题"}。' - "你需要自己生成完整的 HTML 代码" - "（包含 、、、 等），" - "然后传给 html 参数即可。" - "HTML 可以很长，没有长度限制，不需要分段传入。" - "【禁止】不要用 code_interpreter 写 HTML 再 print，" - "不要用 code_interpreter 把 HTML 写入文件再读取，" - "直接把 HTML 传给本工具即可。" - "【技能模式 - 仅在使用技能时可选】如果正在使用技能（skill），可以用模板模式：" - '{"template_path": "技能名/templates/模板.html", ' - '"data": {"KEY": "值"}, "title": "标题"}。' - '也可以用文件模式：{"file_path": "/path/to/report.html"}' - ) - async def html_interpreter( - html: str = "", - title: str = "Report", - file_path: str = "", - template_path: str = "", - data: dict | str = None, - ) -> str: - """Render HTML as an interactive web report. - - Default usage: pass a complete HTML string via the `html` parameter. - The HTML can be arbitrarily long — no length limit, no chunking needed. - - Skill template mode (optional): pass `template_path` (relative to skills - dir) plus a `data` dict whose keys match {{PLACEHOLDER}} tokens in the - template. The backend reads the template and performs all replacements. - - Legacy fallback: `file_path` reads HTML from a file on disk. - """ - import re - - from dbgpt.configs.model_config import STATIC_MESSAGE_IMG_PATH - - # ── Mode 1: template_path + data ────────────────────────────── - if template_path and template_path.strip(): - tp = template_path.strip() - skills_dir = Path(DEFAULT_SKILLS_DIR).expanduser().resolve() - target = (skills_dir / tp).resolve() - # Security: must be under skills_dir - try: - target.relative_to(skills_dir) - except ValueError: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Invalid template_path: {tp}", - } - ] - }, - ensure_ascii=False, - ) - if not target.is_file(): - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": ( - f"Template not found: {tp}. " - "This skill does not have HTML templates. " - "Please retry by calling html_interpreter " - "with the `html` parameter instead — " - "generate the complete HTML report code " - "yourself and pass it directly via " - '{"html": "...", ' - '"title": "report title"}.' - ), - } - ] - }, - ensure_ascii=False, - ) - try: - raw_template = target.read_text(encoding="utf-8") - except Exception as e: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Error reading template: {e}", - } - ] - }, - ensure_ascii=False, - ) - # Replace {{KEY}} placeholders with values from data dict - # Sometimes the LLM passes data as a JSON string instead of a dict - replacements = data - if isinstance(replacements, str): - try: - replacements = json.loads(replacements) - except Exception as e: - logger.warning( - f"html_interpreter failed to parse string data as json: {e}" - ) - # Attempt to fix truncated JSON by appending closing - # braces/quotes - try: - fixed = str(replacements).rstrip() - if not fixed.endswith("}"): - if fixed.endswith('"'): - fixed += "}" - else: - fixed += '"}' - replacements = json.loads(fixed) - except Exception: - replacements = {} - if not isinstance(replacements, dict): - replacements = {} - auto_data = react_state.get("auto_data", {}) - if isinstance(auto_data, dict): - replacements = {**auto_data, **replacements} - - # Merge LLM replacements with ratio_data from calculate_ratios.py - ratio_data = react_state.get("ratio_data", {}) - if isinstance(ratio_data, dict): - # auto_data / LLM data overwrites ratio_data if keys overlap - merged = {**ratio_data, **replacements} - replacements = merged - - # Auto-resolve CHART_* placeholders from generated images. - # image_url_map: { - # "financial_overview": "/images/abc_financial_overview.png" - # } - # Template uses: - # {{CHART_FINANCIAL_OVERVIEW}} - # -> /images/abc_financial_overview.png - image_url_map = react_state.get("image_url_map", {}) - if isinstance(image_url_map, dict): - for stem, url in image_url_map.items(): - chart_key = f"CHART_{stem.upper()}" - if chart_key not in replacements: - replacements[chart_key] = url - - def _replace_placeholder(m): - key = m.group(1) - return str(replacements.get(key, "")) - - html = re.sub(r"\{\{([A-Z_0-9]+)\}\}", _replace_placeholder, raw_template) - if not title or title == "Report": - title = target.stem - logger.info( - "html_interpreter: template=%s, %d placeholders replaced, " - "html=%d chars", - tp, - len(replacements), - len(html), - ) - - # ── Mode 2: file_path ───────────────────────────────────────── - elif file_path and file_path.strip(): - fp = file_path.strip() - if not os.path.isfile(fp): - cid = react_state.get("conv_id") or "default" - from dbgpt.configs.model_config import PILOT_PATH - - alt = os.path.join(PILOT_PATH, "data", cid, os.path.basename(fp)) - if os.path.isfile(alt): - fp = alt - else: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"File not found: {file_path}", - } - ] - }, - ensure_ascii=False, - ) - try: - with open(fp, "r", encoding="utf-8") as f: - html = f.read() - if not title or title == "Report": - title = os.path.splitext(os.path.basename(fp))[0] - logger.info( - "html_interpreter: read %d chars from file %s", - len(html), - fp, - ) - except Exception as e: - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Error reading file: {e}", - } - ] - }, - ensure_ascii=False, - ) - - # ── Mode 3: inline html ────────────────────────────────────── - # Unescape literal \n sequences that LLM may produce. - # IMPORTANT: Only apply this unescape when html was provided directly - # (inline mode). Template mode (Mode 1) and file mode (Mode 2) produce - # real HTML that already contains actual newlines and may contain JS - # regex literals like /\\n/ which must NOT be collapsed into real - # newlines — doing so corrupts the JS and breaks chart rendering. - if html and isinstance(html, str) and not template_path and not file_path: - if "\\n" in html: - html = html.replace("\\n", "\n") - if "\\t" in html: - html = html.replace("\\t", "\t") - if not html or not html.strip(): - return json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "No HTML content provided", - } - ] - }, - ensure_ascii=False, - ) - - # Post-process: fix image URLs that the LLM may have guessed wrong. - # Files in STATIC_MESSAGE_IMG_PATH are named "{uuid8}_{original}.ext". - # The LLM might reference "/images/original.ext" (without UUID prefix) - # or even just "original.ext". Build a lookup and replace. - fixed_html = html.strip() - try: - IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} - # Map: lowercase base name (without uuid prefix) -> served path - # e.g. "monthly_sales_trend.png" - # -> "/images/a1b2c3ff_monthly_sales_trend.png" - name_to_served: Dict[str, str] = {} - if os.path.isdir(STATIC_MESSAGE_IMG_PATH): - for fname in os.listdir(STATIC_MESSAGE_IMG_PATH): - ext = os.path.splitext(fname)[1].lower() - if ext not in IMAGE_EXTS: - continue - # Strip the 8-char hex UUID prefix + underscore - # Pattern: <8 hex chars>_ - m = re.match(r"^[0-9a-f]{8}_(.+)$", fname, re.IGNORECASE) - if m: - base_name = m.group(1).lower() - served_path = f"/images/{fname}" - # Keep the latest (last alphabetically = most recent - # UUID) - name_to_served[base_name] = served_path - - if name_to_served: - # Replace patterns like: - # src="/images/monthly_sales_trend.png" - # src="images/monthly_sales_trend.png" - # src="monthly_sales_trend.png" - # with the correct served path. - def _fix_img_src(match: re.Match) -> str: - prefix = match.group(1) # src=" or src=' - raw_path = match.group(2) # the path value - quote = match.group(3) # closing quote - - # Extract just the filename from the path - filename = raw_path.rsplit("/", 1)[-1].lower() - - # Check if it's already a correct served path - if re.match(r"^[0-9a-f]{8}_.+$", filename, re.IGNORECASE): - return match.group(0) # Already has UUID prefix - - if filename in name_to_served: - return f"{prefix}{name_to_served[filename]}{quote}" - return match.group(0) # No match, keep original - - # Match src="..." or src='...' containing image references - fixed_html = re.sub( - r"""(src\s*=\s*["'])""" - r"""([^"']+\.(?:png|jpg|jpeg|gif|svg|webp))""" - r"""(["'])""", - _fix_img_src, - fixed_html, - flags=re.IGNORECASE, - ) - except Exception: - pass # If post-processing fails, use original HTML - - # Auto-append images generated during this session that the LLM - # forgot to include in the HTML. - try: - gen_images = react_state.get("generated_images", []) - if gen_images: - # Extract all image filenames already referenced in the HTML - # (e.g. "time_series_trend.png" from any src="...time_series_trend.png") - html_img_stems = set( - re.sub(r"^[0-9a-f]+_", "", os.path.basename(src)) - for src in re.findall( - r']+src=["\']([^"\']+)["\']', fixed_html, re.IGNORECASE - ) - ) - - # An image is "missing" only when neither its exact URL nor its - # stem (filename with UUID prefix stripped) is already covered. - def _img_stem(url): - return re.sub(r"^[0-9a-f]+_", "", os.path.basename(url)) - - missing = [ - url - for url in gen_images - if url not in fixed_html and _img_stem(url) not in html_img_stems - ] - if missing: - imgs_html = "".join( - f'
' - f'' - f"
" - for url in missing - ) - section = ( - '
' - "

📊 分析图表

" - f"{imgs_html}
" - ) - # Insert before if present, otherwise append - if "" in fixed_html.lower(): - fixed_html = re.sub( - r"()", - section + r"\1", - fixed_html, - count=1, - flags=re.IGNORECASE, - ) - else: - fixed_html += section - except Exception: - pass - - chunks: List[Dict[str, Any]] = [ - {"output_type": "html", "content": fixed_html, "title": title}, - ] - return json.dumps({"chunks": chunks}, ensure_ascii=False) - - llm_client = DefaultLLMClient( - CFG.SYSTEM_APP.get_component( - ComponentType.WORKER_MANAGER_FACTORY, WorkerManagerFactory - ).create(), - auto_convert_message=True, - ) - # If user specified a model_name, use Priority strategy to ensure the - # agent uses the requested model instead of picking the first available one. - if dialogue.model_name: - llm_config = LLMConfig( - llm_client=llm_client, - llm_strategy=LLMStrategyType.Priority, - strategy_context=json.dumps([dialogue.model_name]), - ) - else: - llm_config = LLMConfig(llm_client=llm_client) - - conv_id = dialogue.conv_uid or str(uuid.uuid4()) - react_state["conv_id"] = conv_id - if conv_id in REACT_AGENT_MEMORY_CACHE: - gpt_memory = REACT_AGENT_MEMORY_CACHE[conv_id] - else: - gpt_memory = GptsMemory( - plans_memory=DefaultGptsPlansMemory(), - message_memory=MetaDbGptsMessageMemory(), - ) - gpt_memory.init(conv_id, enable_vis_message=False) - REACT_AGENT_MEMORY_CACHE[conv_id] = gpt_memory - agent_memory = AgentMemory(gpts_memory=gpt_memory) - - # --- Persist conversation to chat_history for sidebar display --- - conv_serve = ConversationServe.get_instance(CFG.SYSTEM_APP) - storage_conv = StorageConversation( - conv_uid=conv_id, - chat_mode=dialogue.chat_mode or "chat_react_agent", - user_name=dialogue.user_name, - sys_code=dialogue.sys_code, - summary=dialogue.user_input, - app_code=dialogue.app_code, - conv_storage=conv_serve.conv_storage, - message_storage=conv_serve.message_storage, - ) - storage_conv.save_to_storage() - storage_conv.start_new_round() - storage_conv.add_user_message(user_input) - context = AgentContext( - conv_id=conv_id, - gpts_app_code="react_agent", - gpts_app_name="ReAct", - language="zh", - temperature=dialogue.temperature or 0.2, - enable_context_management=True, - ) - - # Build file context if file uploaded - file_context = "" - if file_path: - file_context = f""" -## User Uploaded File -- File path: {file_path} -- Analyze this file if needed for the user's request. -""" - - # Build skill context for system prompt when skill is pre-selected - skill_prompt_context = "" - execution_instruction = "" - if pre_matched_skill and react_state.get("skill_prompt"): - skill_template = react_state["skill_prompt"] - skill_text = ( - skill_template.template - if hasattr(skill_template, "template") - else str(skill_template) - ) - skill_prompt_context = f""" -## 已加载技能指令（{pre_matched_skill.metadata.name}） -以下是用户选择的技能的完整指令，请严格按照这些指令进行操作： - -{skill_text} -""" - execution_instruction = f""" -## 执行要求 -1. 用户已明确选择技能：{pre_matched_skill.metadata.name} -2. 你必须严格按照上述技能指令的步骤执行 -3. 阅读技能指令，理解每一步需要调用的工具 -4. 按顺序执行工具调用，完成技能目标 -""" - - # ── TodoWrite tool ────────────────────────────────────────────────── - # A session-level task list that the agent maintains. The full list is - # replaced on every call (same semantics as OpenCode's todowrite). - # The tool pushes a ``plan.update`` SSE event so the frontend can - # render a live task-plan card. + # ── Build tool instances from tools/ directory ─────────────────────── _todo_list: List[Dict[str, str]] = [] - - @tool( - description=( - "Create and manage a structured task list for the current session. " - "Use this tool to plan complex tasks (3+ steps), track progress, " - "and show the user what you are doing. " - "Pass the FULL todo list every time (not incremental). " - "Each todo has: content (brief description), " - "status (pending | in_progress | completed | cancelled), " - "priority (high | medium | low). " - "Rules: only ONE task in_progress at a time; mark tasks completed " - "immediately after finishing; do NOT use for single trivial tasks." - '\nParameter: {"todos": [{"content": "...", "status": "...", ' - '"priority": "..."}]}' - ) - ) - def todowrite(todos: str) -> str: - """Update the session todo list (full replacement).""" - import json as _json - - parsed: List[Dict[str, str]] = [] - try: - raw = _json.loads(todos) if isinstance(todos, str) else todos - items = raw if isinstance(raw, list) else raw.get("todos", raw) - if isinstance(items, list): - for item in items: - parsed.append( - { - "content": str(item.get("content", "")), - "status": str(item.get("status", "pending")), - "priority": str(item.get("priority", "medium")), - } - ) - except Exception: - return _json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": "Error: invalid todos JSON", - } - ] - }, - ensure_ascii=False, - ) - - _todo_list.clear() - _todo_list.extend(parsed) - - total = len(parsed) - done = sum(1 for t in parsed if t["status"] == "completed") - return _json.dumps( - { - "chunks": [ - { - "output_type": "text", - "content": f"Todo list updated: {done}/{total} completed", - } - ], - # Attach the todo list so SSE handler can forward it - "__todos__": parsed, - }, - ensure_ascii=False, - ) + select_skill_tool = make_select_skill(react_state, registry) + load_skill_tool = make_load_skill(react_state) + load_file_tool = make_load_file(react_state) + execute_analysis_tool = make_execute_analysis(react_state) + load_tools_tool = make_load_tools(react_state) + execute_tool_tool = make_execute_tool(react_state) + knowledge_retrieve_tool = make_knowledge_retrieve(react_state, knowledge_resources) + sql_query_tool = make_sql_query(react_state, database_connector) + code_interpreter_tool = make_code_interpreter(react_state) + shell_interpreter_tool = make_shell_interpreter(react_state) + html_interpreter_tool = make_html_interpreter(react_state, DEFAULT_SKILLS_DIR) + todowrite_tool = make_todowrite(_todo_list, stream_callback) + question_tool = make_question(react_state, stream_callback) + # Keep local aliases for backward compatibility (SSE loop references these names) + execute_skill_script_file_tool = make_execute_skill_script_file(react_state) _todo_action_history: Dict[int, List[str]] = {} @@ -3102,7 +1554,13 @@ Parameters: {{"sql": "SELECT statement"}} IMPORTANT: You MUST call todowrite again after EACH task completes to update status. The user sees progress in real time — never skip an update. Parameters: {{"todos": [{{...}}]}} -8. **terminate**: Return the final answer when the task is completed. Action Input +8. **question**: Ask the user a question and wait for their response. Use this tool + when you need user input, clarification, or a decision to proceed. The tool blocks + until the user answers. + Parameters: {{"questions": [{{"question": "...", "header": "...", "options": [ + {{"label": "...", "description": "..."}}, ...]}}]}}. Set multiple=true to allow + multiple selections. The tool returns the user's selected answers. +9. **terminate**: Return the final answer when the task is completed. Action Input must be {{"result": "your final answer content"}}. ## Task Management @@ -3144,11 +1602,12 @@ Action Input: The JSON format of tool parameters [ execute_skill_script, get_skill_resource, - execute_skill_script_file, - shell_interpreter, - html_interpreter, - sql_query, - todowrite, + execute_skill_script_file_tool, + shell_interpreter_tool, + html_interpreter_tool, + sql_query_tool, + todowrite_tool, + question_tool, Terminate(), ] + business_tools @@ -3264,7 +1723,13 @@ File mode: {{"file_path": "/path/to/report.html"}} IMPORTANT: You MUST call todowrite again after EACH task completes to update status. The user sees progress in real time — never skip an update. Parameters: {{"todos": [{{...}}]}} -15. **terminate**: Finish the task. Parameters: {{"result": "final answer"}} +15. **question**: Ask the user a question and wait for their response. Use this tool + when you need user input, clarification, or a decision to proceed. The tool blocks + until the user answers. + Parameters: {{"questions": [{{"question": "...", "header": "...", "options": [ + {{"label": "...", "description": "..."}}, ...]}}]}}. Set multiple=true to allow + multiple selections. The tool returns the user's selected answers. +16. **terminate**: Finish the task. Parameters: {{"result": "final answer"}} {file_context} {knowledge_context} @@ -3285,17 +1750,18 @@ Action Input: The JSON format of tool parameters tool_pack = ToolPack( [ - load_skill, - load_tools, - knowledge_retrieve, + load_skill_tool, + load_tools_tool, + knowledge_retrieve_tool, execute_skill_script, get_skill_resource, - execute_skill_script_file, - code_interpreter, - shell_interpreter, - html_interpreter, - sql_query, - todowrite, + execute_skill_script_file_tool, + code_interpreter_tool, + shell_interpreter_tool, + html_interpreter_tool, + sql_query_tool, + todowrite_tool, + question_tool, Terminate(), ] + business_tools @@ -3331,7 +1797,8 @@ Action Input: The JSON format of tool parameters parser = ReActOutputParser() received = AgentMessage(content=user_input) - stream_queue: asyncio.Queue = asyncio.Queue() + # stream_queue and stream_callback were created earlier (before ToolPack) + # so that the question tool can use them. # Wire up context-management status events into the SSE stream. async def _context_status_callback(status: Dict[str, Any]) -> None: @@ -3346,9 +1813,6 @@ Action Input: The JSON format of tool parameters on_status_event=_context_status_callback, ) - async def stream_callback(event_type: str, payload: Dict[str, Any]) -> None: - await stream_queue.put({"type": event_type, **payload}) - async def run_agent(): return await agent.generate_reply( received_message=received, @@ -3420,6 +1884,9 @@ Action Input: The JSON format of tool parameters if event_type == "context.status": # Forward context-management status to frontend as-is. yield _sse_event(event) + elif event_type in ("question.asked", "question.replied", "question.rejected"): + # Forward human-in-the-loop question events to frontend as-is. + yield _sse_event(event) elif event_type == "thinking": # Parse thinking content but don't create step yet # Step will be created when 'act' event arrives with confirmed @@ -4114,3 +2581,41 @@ async def chat_react_agent( headers=headers, media_type="text/plain", ) + + +# ── Human-in-the-Loop Question API ────────────────────────────────────────── + + +class _QuestionReplyBody(_BaseModel): + answers: List[List[str]] + + +@router.post("/v1/chat/question/{request_id}/reply", response_model=Result) +async def question_reply( + request_id: str, + body: _QuestionReplyBody, + user_token: UserRequest = Depends(get_user_from_headers), +): + """User submits answers to a pending question, unblocking the agent tool.""" + from dbgpt_app.openapi.api_v1.tools.question_manager import question_manager + + try: + question_manager.reply(request_id, body.answers) + return Result.succ({"success": True, "request_id": request_id}) + except KeyError as e: + return Result.failed(msg=str(e)) + + +@router.post("/v1/chat/question/{request_id}/reject", response_model=Result) +async def question_reject( + request_id: str, + user_token: UserRequest = Depends(get_user_from_headers), +): + """User dismisses a pending question, unblocking the agent tool with rejection.""" + from dbgpt_app.openapi.api_v1.tools.question_manager import question_manager + + try: + question_manager.reject(request_id) + return Result.succ({"success": True, "request_id": request_id}) + except KeyError as e: + return Result.failed(msg=str(e)) diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/__init__.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/__init__.py new file mode 100644 index 000000000..a1bcd38d4 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/__init__.py @@ -0,0 +1,32 @@ +"""Built-in tools for the ReAct agent in agentic_data_api.""" + +from .code_interpreter import make_code_interpreter +from .execute_analysis import make_execute_analysis +from .execute_tool import make_execute_tool +from .html_interpreter import make_html_interpreter +from .knowledge_retrieve import make_knowledge_retrieve +from .load_file import make_load_file +from .load_tools import make_load_tools +from .question import make_question +from .select_skill import make_select_skill +from .shell_interpreter import make_shell_interpreter +from .skill_tools import make_execute_skill_script_file, make_load_skill +from .sql_query import make_sql_query +from .todowrite import make_todowrite + +__all__ = [ + "make_code_interpreter", + "make_execute_analysis", + "make_execute_tool", + "make_html_interpreter", + "make_knowledge_retrieve", + "make_load_file", + "make_load_tools", + "make_question", + "make_select_skill", + "make_shell_interpreter", + "make_execute_skill_script_file", + "make_load_skill", + "make_sql_query", + "make_todowrite", +] diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/_helpers.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/_helpers.py new file mode 100644 index 000000000..db12d84be --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/_helpers.py @@ -0,0 +1,21 @@ +"""Shared helper utilities for built-in tools.""" + +import re +from typing import Dict, Tuple + + +_AUTO_DATA_MARKER_PATTERN = re.compile( + r"###([A-Z0-9_]+)_START###\s*(.*?)\s*###\1_END###", re.DOTALL +) + + +def _extract_auto_data_markers(text: str) -> Tuple[str, Dict[str, str]]: + """Extract AUTO_DATA markers from text and return (cleaned_text, data_dict).""" + extracted: Dict[str, str] = {} + + def _replacer(m: re.Match) -> str: + extracted[m.group(1)] = m.group(2) + return "" + + cleaned = _AUTO_DATA_MARKER_PATTERN.sub(_replacer, text) + return cleaned.strip(), extracted diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/code_interpreter.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/code_interpreter.py new file mode 100644 index 000000000..8629a5ee2 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/code_interpreter.py @@ -0,0 +1,208 @@ +"""code_interpreter tool — execute Python code in a subprocess.""" + +import asyncio +import json +import logging +import os +import shutil +import sys +import uuid +from typing import Any, Dict, List, Optional + +from dbgpt.agent.resource.tool.base import tool + +logger = logging.getLogger(__name__) + + +def _try_repair_truncated_code(raw_code: str) -> Optional[str]: + """Attempt to fix code that was truncated by the LLM's token limit.""" + lines = raw_code.split("\n") + for trim in range(1, min(11, len(lines))): + candidate_lines = lines[: len(lines) - trim] + if not candidate_lines: + continue + candidate = "\n".join(candidate_lines) + open_chars = {"(": ")", "[": "]", "{": "}"} + close_chars = set(open_chars.values()) + stack: list = [] + for ch in candidate: + if ch in open_chars: + stack.append(open_chars[ch]) + elif ch in close_chars: + if stack and stack[-1] == ch: + stack.pop() + if stack: + candidate += "\n" + "".join(reversed(stack)) + try: + compile(candidate, "", "exec") + return candidate + except SyntaxError: + continue + return None + + +def make_code_interpreter(react_state: Dict[str, Any]): + @tool( + description=( + "Execute Python code for data analysis and computation. " + "Supports pandas, numpy, matplotlib, json, os, etc. " + "Use this tool when you need to run Python code to process data, " + "generate charts, or perform calculations. " + 'Parameters: {{"code": "python code string"}}' + ) + ) + async def code_interpreter(code: str) -> str: + """Execute arbitrary Python code and return stdout/stderr. + + CRITICAL: Each call is completely independent — variables do NOT + persist between calls. Every code snippet MUST include all necessary + data loading and processing. Always print() results you want to see. + """ + from dbgpt.configs.model_config import PILOT_PATH, STATIC_MESSAGE_IMG_PATH + + if not code or not code.strip(): + return json.dumps( + {"chunks": [{"output_type": "text", "content": "No code provided"}]}, + ensure_ascii=False, + ) + + cid = react_state.get("conv_id") or "default" + work_dir = os.path.join(PILOT_PATH, "tmp", cid) + os.makedirs(work_dir, exist_ok=True) + + IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} + pre_existing_images: set = set() + for root, _dirs, files in os.walk(work_dir): + for f in files: + ext = os.path.splitext(f)[1].lower() + if ext in IMAGE_EXTS: + pre_existing_images.add(os.path.join(root, f)) + + preamble_lines = [ + "import json", + "import os", + "import pandas as pd", + "import numpy as np", + f'PLOT_DIR = r"{work_dir}"', + "os.makedirs(PLOT_DIR, exist_ok=True)", + ] + fp = react_state.get("file_path") + if fp: + preamble_lines.append(f'FILE_PATH = r"{fp}"') + preamble = "\n".join(preamble_lines) + "\n" + full_code = preamble + code + + try: + compile(full_code, "", "exec") + except SyntaxError as se: + repaired = _try_repair_truncated_code(full_code) + if repaired is not None: + logger.warning( + "code_interpreter: auto-repaired truncated code " + "(original SyntaxError: %s line %s)", + se.msg, + se.lineno, + ) + full_code = repaired + code = full_code[len(preamble) :] + else: + error_msg = ( + f"SyntaxError before execution: {se.msg} " + f"(line {se.lineno})\n" + "Please regenerate complete, syntactically valid Python code. " + "Keep code under 80 lines." + ) + return json.dumps( + { + "chunks": [ + {"output_type": "code", "content": code.strip()}, + {"output_type": "text", "content": error_msg}, + ] + }, + ensure_ascii=False, + ) + + output_text = "" + try: + tmp_path = os.path.join(work_dir, "_run.py") + with open(tmp_path, "w") as tmp: + tmp.write(full_code) + + proc = await asyncio.create_subprocess_exec( + sys.executable, + tmp_path, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=work_dir, + ) + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=60) + output_text = stdout.decode("utf-8", errors="replace") + error_text = stderr.decode("utf-8", errors="replace") + + if proc.returncode != 0 and error_text: + output_text = ( + output_text + "\n[ERROR]\n" + error_text + if output_text + else error_text + ) + except asyncio.TimeoutError: + output_text = "Execution timed out (60s limit)" + except Exception as e: + output_text = f"Execution error: {e}" + + chunks: List[Dict[str, Any]] = [ + {"output_type": "code", "content": code.strip()}, + ] + if output_text.strip(): + clean_output = output_text.strip() + max_out_len = 2000 + if len(clean_output) > max_out_len: + truncation_notice = ( + f"\n\n... [Output truncated, length: {len(clean_output)} chars." + f" Only showing first {max_out_len} chars.]" + ) + clean_output = clean_output[:max_out_len] + truncation_notice + chunks.append({"output_type": "text", "content": clean_output}) + else: + chunks.append( + { + "output_type": "text", + "content": "(no output — add print() to see results)", + } + ) + + # Scan for new images generated by this run + try: + os.makedirs(STATIC_MESSAGE_IMG_PATH, exist_ok=True) + for root, _dirs, files in os.walk(work_dir): + for fname in files: + ext = os.path.splitext(fname)[1].lower() + full_path = os.path.join(root, fname) + if ext in IMAGE_EXTS and full_path not in pre_existing_images: + unique_name = f"{uuid.uuid4().hex[:8]}_{fname}" + dest = os.path.join(STATIC_MESSAGE_IMG_PATH, unique_name) + shutil.copy2(full_path, dest) + img_url = f"/images/{unique_name}" + chunks.append({"output_type": "image", "content": img_url}) + react_state.setdefault("generated_images", []).append(img_url) + except Exception: + pass + + # Clean up temp script + try: + script_path = os.path.join(work_dir, "_run.py") + if os.path.exists(script_path): + os.remove(script_path) + except Exception: + pass + + all_images = react_state.get("generated_images", []) + if all_images: + img_summary = "已生成的图片URL（在生成HTML时请使用这些URL）:\n" + "\n".join( + f" - {url}" for url in all_images + ) + chunks.append({"output_type": "text", "content": img_summary}) + + return json.dumps({"chunks": chunks}, ensure_ascii=False) + + return code_interpreter diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_analysis.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_analysis.py new file mode 100644 index 000000000..28d621b6f --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_analysis.py @@ -0,0 +1,115 @@ +"""execute_analysis tool — quick Excel/CSV analysis via code server.""" + +import json +from typing import Any, Dict, List + +from dbgpt.agent.resource.tool.base import tool + + +def make_execute_analysis(react_state: Dict[str, Any]): + @tool(description="Execute quick analysis on uploaded Excel/CSV file.") + async def execute_analysis() -> str: + from dbgpt._private.config import Config + from dbgpt_app.openapi.api_v1.agentic_data_api import get_code_server + + CFG = Config() + + def _is_excel_skill(meta) -> bool: + name = (meta.name or "").lower() + desc = (meta.description or "").lower() + tags = [tag.lower() for tag in (meta.tags or [])] + return any( + token in name or token in desc or token in tags + for token in ["excel", "xlsx", "xls", "spreadsheet"] + ) + + matched = react_state.get("matched") + if not react_state.get("file_path"): + return json.dumps( + {"chunks": [{"output_type": "text", "content": "No file to analyze"}]}, + ensure_ascii=False, + ) + if matched and not _is_excel_skill(matched.metadata): + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "Selected skill is not for Excel analysis", + } + ] + }, + ensure_ascii=False, + ) + code_server = await get_code_server(CFG.SYSTEM_APP) + analysis_code = """ +import json +import pandas as pd + +file_path = r"{file_path}" +if file_path.lower().endswith((".xls", ".xlsx")): + df = pd.read_excel(file_path) +else: + df = pd.read_csv(file_path) +summary = {{ + "shape": list(df.shape), + "columns": list(df.columns), + "dtypes": {{col: str(dtype) for col, dtype in df.dtypes.items()}}, + "head": df.head(5).to_dict(orient="records"), +}} +print(json.dumps(summary, ensure_ascii=False)) +""".format(file_path=react_state["file_path"]) + result = await code_server.exec(analysis_code, "python") + output_text = ( + result.output.decode("utf-8") if isinstance(result.output, bytes) else "" + ) + chunks: List[Dict[str, Any]] = [ + {"output_type": "code", "content": analysis_code.strip()} + ] + if output_text: + try: + summary = json.loads(output_text) + chunks.append({"output_type": "json", "content": summary}) + head_rows = summary.get("head") + columns = summary.get("columns") + if isinstance(head_rows, list) and isinstance(columns, list): + chunks.append( + { + "output_type": "table", + "content": { + "columns": [ + {"title": col, "dataIndex": col, "key": col} + for col in columns + ], + "rows": head_rows, + }, + } + ) + numeric_columns = [ + col + for col, dtype in (summary.get("dtypes") or {}).items() + if "int" in dtype or "float" in dtype + ] + if numeric_columns and isinstance(head_rows, list): + series_col = numeric_columns[0] + data = [ + {"x": idx + 1, "y": row.get(series_col)} + for idx, row in enumerate(head_rows) + if row.get(series_col) is not None + ] + if data: + chunks.append( + { + "output_type": "chart", + "content": { + "data": data, + "xField": "x", + "yField": "y", + }, + } + ) + except Exception: + chunks.append({"output_type": "text", "content": output_text}) + return json.dumps({"chunks": chunks}, ensure_ascii=False) + + return execute_analysis diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_tool.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_tool.py new file mode 100644 index 000000000..73d01e916 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/execute_tool.py @@ -0,0 +1,43 @@ +"""execute_tool tool — run a registered business tool by name.""" + +import json +from typing import Any, Dict + +from dbgpt.agent.resource.tool.base import tool + + +def make_execute_tool(react_state: Dict[str, Any]): + @tool(description="Execute a tool by name with JSON args.") + async def execute_tool(tool_name: str, args: dict) -> str: + from dbgpt._private.config import Config + from dbgpt.agent.resource.manage import get_resource_manager + from dbgpt.agent.resource.resource_api import AgentResource, ResourceType + from dbgpt.agent.resource.tool.pack import ToolPack + + CFG = Config() + rm = get_resource_manager(CFG.SYSTEM_APP) + try: + tool_resource = rm.build_resource_by_type( + ResourceType.Tool.value, + AgentResource(type=ResourceType.Tool.value, value=tool_name), + ) + tool_pack = ToolPack([tool_resource]) + result = await tool_pack.async_execute(resource_name=tool_name, **args) + return json.dumps( + {"chunks": [{"output_type": "text", "content": str(result)}]}, + ensure_ascii=False, + ) + except Exception as e: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Tool execute failed: {e}", + } + ] + }, + ensure_ascii=False, + ) + + return execute_tool diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/html_interpreter.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/html_interpreter.py new file mode 100644 index 000000000..db246577d --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/html_interpreter.py @@ -0,0 +1,272 @@ +"""html_interpreter tool — render HTML as an interactive report.""" + +import json +import logging +import os +import re +from pathlib import Path +from typing import Any, Dict, List, Optional + +from dbgpt.agent.resource.tool.base import tool + +logger = logging.getLogger(__name__) + + +def make_html_interpreter(react_state: Dict[str, Any], skills_dir: str): + @tool( + description=( + "将 HTML 渲染为可交互的网页报告，这是向用户展示网页报告的唯一方式。" + "【默认用法】直接传入完整的 HTML 字符串：" + '{"html": "...", "title": "报告标题"}。' + "你需要自己生成完整的 HTML 代码" + "（包含 、、、 等），" + "然后传给 html 参数即可。" + "HTML 可以很长，没有长度限制，不需要分段传入。" + "【技能模式 - 仅在使用技能时可选】如果正在使用技能（skill），可以用模板模式：" + '{"template_path": "技能名/templates/模板.html", ' + '"data": {"KEY": "值"}, "title": "标题"}。' + '也可以用文件模式：{"file_path": "/path/to/report.html"}' + ) + ) + async def html_interpreter( + html: str = "", + title: str = "Report", + file_path: str = "", + template_path: str = "", + data: dict | str = None, + ) -> str: + """Render HTML as an interactive web report.""" + from dbgpt.configs.model_config import STATIC_MESSAGE_IMG_PATH + + skills_path = Path(skills_dir).expanduser().resolve() + + # ── Mode 1: template_path + data ── + if template_path and template_path.strip(): + tp = template_path.strip() + target = (skills_path / tp).resolve() + try: + target.relative_to(skills_path) + except ValueError: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Invalid template_path: {tp}", + } + ] + }, + ensure_ascii=False, + ) + if not target.is_file(): + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": ( + f"Template not found: {tp}. " + "Please retry using the `html` parameter directly — " + "generate complete HTML yourself and pass it via " + '{"html": "...", "title": "title"}.' + ), + } + ] + }, + ensure_ascii=False, + ) + try: + raw_template = target.read_text(encoding="utf-8") + except Exception as e: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Error reading template: {e}", + } + ] + }, + ensure_ascii=False, + ) + + replacements = data + if isinstance(replacements, str): + try: + replacements = json.loads(replacements) + except Exception: + try: + fixed = str(replacements).rstrip() + if not fixed.endswith("}"): + fixed += '"}' if not fixed.endswith('"') else "}" + replacements = json.loads(fixed) + except Exception: + replacements = {} + if not isinstance(replacements, dict): + replacements = {} + + auto_data = react_state.get("auto_data", {}) + if isinstance(auto_data, dict): + replacements = {**auto_data, **replacements} + + ratio_data = react_state.get("ratio_data", {}) + if isinstance(ratio_data, dict): + replacements = {**ratio_data, **replacements} + + image_url_map = react_state.get("image_url_map", {}) + if isinstance(image_url_map, dict): + for stem, url in image_url_map.items(): + chart_key = f"CHART_{stem.upper()}" + if chart_key not in replacements: + replacements[chart_key] = url + + def _replace_placeholder(m): + key = m.group(1) + return str(replacements.get(key, "")) + + html = re.sub(r"\{\{([A-Z_0-9]+)\}\}", _replace_placeholder, raw_template) + if not title or title == "Report": + title = target.stem + + # ── Mode 2: file_path ── + elif file_path and file_path.strip(): + fp = file_path.strip() + if not os.path.isfile(fp): + cid = react_state.get("conv_id") or "default" + from dbgpt.configs.model_config import PILOT_PATH + + alt = os.path.join(PILOT_PATH, "data", cid, os.path.basename(fp)) + if os.path.isfile(alt): + fp = alt + else: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"File not found: {file_path}", + } + ] + }, + ensure_ascii=False, + ) + try: + with open(fp, "r", encoding="utf-8") as f: + html = f.read() + if not title or title == "Report": + title = os.path.splitext(os.path.basename(fp))[0] + except Exception as e: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Error reading file: {e}", + } + ] + }, + ensure_ascii=False, + ) + + # ── Mode 3: inline html ── + if html and isinstance(html, str) and not template_path and not file_path: + if "\\n" in html: + html = html.replace("\\n", "\n") + if "\\t" in html: + html = html.replace("\\t", "\t") + + if not html or not html.strip(): + return json.dumps( + { + "chunks": [ + {"output_type": "text", "content": "No HTML content provided"} + ] + }, + ensure_ascii=False, + ) + + # Post-process: fix image URLs + fixed_html = html.strip() + try: + IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} + name_to_served: Dict[str, str] = {} + if os.path.isdir(STATIC_MESSAGE_IMG_PATH): + for fname in os.listdir(STATIC_MESSAGE_IMG_PATH): + ext = os.path.splitext(fname)[1].lower() + if ext not in IMAGE_EXTS: + continue + m = re.match(r"^[0-9a-f]{8}_(.+)$", fname, re.IGNORECASE) + if m: + name_to_served[m.group(1).lower()] = f"/images/{fname}" + + if name_to_served: + + def _fix_img_src(match: re.Match) -> str: + prefix = match.group(1) + raw_path = match.group(2) + quote = match.group(3) + filename = raw_path.rsplit("/", 1)[-1].lower() + if re.match(r"^[0-9a-f]{8}_.+$", filename, re.IGNORECASE): + return match.group(0) + if filename in name_to_served: + return f"{prefix}{name_to_served[filename]}{quote}" + return match.group(0) + + fixed_html = re.sub( + r"""(src\s*=\s*["'])([^"']+\.(?:png|jpg|jpeg|gif|svg|webp))(["'])""", + _fix_img_src, + fixed_html, + flags=re.IGNORECASE, + ) + except Exception: + pass + + # Auto-append missing images + try: + gen_images = react_state.get("generated_images", []) + if gen_images: + html_img_stems = set( + re.sub(r"^[0-9a-f]+_", "", os.path.basename(src)) + for src in re.findall( + r']+src=["\']([^"\']+)["\']', fixed_html, re.IGNORECASE + ) + ) + + def _img_stem(url): + return re.sub(r"^[0-9a-f]+_", "", os.path.basename(url)) + + missing = [ + url + for url in gen_images + if url not in fixed_html and _img_stem(url) not in html_img_stems + ] + if missing: + imgs_html = "".join( + f'
' + f'' + f"
" + for url in missing + ) + section = ( + '

📊 分析图表

' + f"{imgs_html}
" + ) + if "" in fixed_html.lower(): + fixed_html = re.sub( + r"()", + section + r"\1", + fixed_html, + count=1, + flags=re.IGNORECASE, + ) + else: + fixed_html += section + except Exception: + pass + + chunks: List[Dict[str, Any]] = [ + {"output_type": "html", "content": fixed_html, "title": title}, + ] + return json.dumps({"chunks": chunks}, ensure_ascii=False) + + return html_interpreter diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/knowledge_retrieve.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/knowledge_retrieve.py new file mode 100644 index 000000000..fac707968 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/knowledge_retrieve.py @@ -0,0 +1,77 @@ +"""knowledge_retrieve tool — search the knowledge base.""" + +import json +from typing import Any, Dict, List + +from dbgpt.agent.resource.tool.base import tool + + +def make_knowledge_retrieve(react_state: Dict[str, Any], knowledge_resources: List): + @tool( + description=( + "Retrieve relevant information from the knowledge base. " + "Use this tool when the user question involves content that may be " + 'in the knowledge base. Parameters: {{"query": "search query"}}' + ) + ) + async def knowledge_retrieve(query: str) -> str: + if not knowledge_resources: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "No knowledge base available", + } + ] + }, + ensure_ascii=False, + ) + + resource = knowledge_resources[0] + try: + chunks = await resource.retrieve(query) + if chunks: + content = "\n".join( + [f"[{i + 1}] {chunk.content}" for i, chunk in enumerate(chunks[:5])] + ) + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": ( + f"Retrieved {len(chunks)} relevant documents" + ), + }, + {"output_type": "markdown", "content": content}, + ] + }, + ensure_ascii=False, + ) + else: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "No relevant information found", + } + ] + }, + ensure_ascii=False, + ) + except Exception as e: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Knowledge retrieval failed: {str(e)}", + } + ] + }, + ensure_ascii=False, + ) + + return knowledge_retrieve diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_file.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_file.py new file mode 100644 index 000000000..e314fb8a8 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_file.py @@ -0,0 +1,30 @@ +"""load_file tool — returns the uploaded file path from react_state.""" + +import json +from typing import Any, Dict + +from dbgpt.agent.resource.tool.base import tool + + +def make_load_file(react_state: Dict[str, Any]): + @tool(description="Load uploaded file info if provided.") + def load_file() -> str: + if not react_state.get("file_path"): + return json.dumps( + {"chunks": [{"output_type": "text", "content": "No file uploaded"}]}, + ensure_ascii=False, + ) + return json.dumps( + { + "chunks": [ + {"output_type": "text", "content": react_state["file_path"]}, + { + "output_type": "text", + "content": "File path provided by user upload", + }, + ] + }, + ensure_ascii=False, + ) + + return load_file diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_tools.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_tools.py new file mode 100644 index 000000000..75f3b9440 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/load_tools.py @@ -0,0 +1,54 @@ +"""load_tools tool — resolves required tools for the selected skill.""" + +import json +from typing import Any, Dict + +from dbgpt.agent.resource.tool.base import tool + + +def make_load_tools(react_state: Dict[str, Any]): + @tool(description="Resolve required tools for the selected skill.") + def load_tools() -> str: + from dbgpt._private.config import Config + from dbgpt.agent.resource.manage import get_resource_manager + from dbgpt.agent.resource.resource_api import AgentResource, ResourceType + + CFG = Config() + matched = react_state.get("matched") + rm = get_resource_manager(CFG.SYSTEM_APP) + required_tools = matched.metadata.required_tools if matched else [] + if not required_tools: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "No required tools specified", + } + ] + }, + ensure_ascii=False, + ) + loaded = [] + failed = [] + for tool_name in required_tools: + try: + rm.build_resource_by_type( + ResourceType.Tool.value, + AgentResource(type=ResourceType.Tool.value, value=tool_name), + ) + loaded.append(tool_name) + except Exception as e: + failed.append(f"{tool_name} ({e})") + chunks = [] + if loaded: + chunks.append( + {"output_type": "text", "content": f"Loaded: {', '.join(loaded)}"} + ) + if failed: + chunks.append( + {"output_type": "text", "content": f"Failed: {', '.join(failed)}"} + ) + return json.dumps({"chunks": chunks}, ensure_ascii=False) + + return load_tools diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question.py new file mode 100644 index 000000000..13b8190be --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question.py @@ -0,0 +1,129 @@ +"""question tool — human-in-the-loop: ask the user questions and block until answered.""" + +import asyncio +import json +import logging +from typing import Any, Callable, Dict + +from dbgpt.agent.resource.tool.base import tool + +from .question_manager import question_manager + +logger = logging.getLogger(__name__) + +_DESCRIPTION = """\ +Use this tool when you need to ask the user questions during execution. This allows you to: +1. Gather user preferences or requirements +2. Clarify ambiguous instructions +3. Get decisions on implementation choices as you work +4. Offer choices to the user about what direction to take. + +Usage notes: +- When `custom` is enabled (default), a "Type your own answer" option is added automatically; don't include "Other" or catch-all options +- Answers are returned as arrays of labels; set `multiple: true` to allow selecting more than one +- If you recommend a specific option, make that the first option in the list and add "(Recommended)" at the end of the label +Parameter: {"questions": [{"question": "...", "header": "...", "options": [{"label": "...", "description": "..."}], "multiple": false}]} +""" + + +def make_question(react_state: Dict[str, Any], stream_callback: Callable): + """Return a ``question`` FunctionTool bound to react_state and stream_callback.""" + + @tool(description=_DESCRIPTION) + async def question(questions: str) -> str: + """Ask the user one or more questions and wait for their answers. + + Args: + questions: JSON string of a list of question objects. Each question has: + - question (str): Complete question text + - header (str): Very short label (max 30 chars) + - options (list): [{label, description}] available choices + - multiple (bool, optional): allow multi-select + """ + conv_id = react_state.get("conv_id", "default") + + # Parse questions JSON + try: + parsed_questions = ( + json.loads(questions) if isinstance(questions, str) else questions + ) + if not isinstance(parsed_questions, list): + parsed_questions = parsed_questions.get("questions", []) + except Exception as e: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Error: invalid questions JSON — {e}", + } + ] + }, + ensure_ascii=False, + ) + + # 1. Register in QuestionManager → creates asyncio.Event + pq = question_manager.create(conv_id=conv_id, questions=parsed_questions) + + # 2. Push question.asked SSE event to frontend + await stream_callback( + "question.asked", + { + "request_id": pq.request_id, + "conv_id": conv_id, + "questions": parsed_questions, + }, + ) + logger.info( + "question tool: pushed question.asked, request_id=%s", pq.request_id + ) + + # 3. Block until user answers (or timeout) + try: + await asyncio.wait_for(pq.event.wait(), timeout=300) + except asyncio.TimeoutError: + question_manager.remove(pq.request_id) + await stream_callback( + "question.rejected", {"request_id": pq.request_id, "conv_id": conv_id} + ) + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "Question timed out after 300 seconds. Proceeding without user answer.", + } + ] + }, + ensure_ascii=False, + ) + finally: + question_manager.remove(pq.request_id) + + # 4. User rejected + if pq.rejected: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "The user dismissed the question. Proceeding without answer.", + } + ] + }, + ensure_ascii=False, + ) + + # 5. Format answers for the LLM + answers = pq.answers or [] + formatted = ", ".join( + f'"{q.get("question", "")}"="{", ".join(answers[i]) if i < len(answers) and answers[i] else "Unanswered"}"' + for i, q in enumerate(parsed_questions) + ) + output = f"User has answered your questions: {formatted}. You can now continue with the user's answers in mind." + return json.dumps( + {"chunks": [{"output_type": "text", "content": output}]}, + ensure_ascii=False, + ) + + return question diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question_manager.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question_manager.py new file mode 100644 index 000000000..c82ded738 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/question_manager.py @@ -0,0 +1,110 @@ +"""QuestionManager — session-level pending question state with asyncio.Event blocking.""" + +import asyncio +import logging +import uuid +from dataclasses import dataclass, field +from typing import Dict, List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class QuestionOption: + label: str + description: str + + +@dataclass +class QuestionInfo: + question: str + header: str + options: List[QuestionOption] + multiple: bool = False + custom: bool = True + + +@dataclass +class PendingQuestion: + request_id: str + conv_id: str + questions: List[dict] # raw dicts from LLM JSON for easy serialization + event: asyncio.Event = field(default_factory=asyncio.Event) + answers: Optional[List[List[str]]] = None + rejected: bool = False + + +class QuestionManager: + """Global manager for all pending question requests across sessions. + + Each question tool invocation: + 1. Calls ``create()`` → gets a PendingQuestion with a fresh asyncio.Event. + 2. Pushes a ``question.asked`` SSE event to the frontend. + 3. Awaits ``pending.event.wait()`` — blocks the tool coroutine. + 4. When the user replies via HTTP, ``reply()`` sets the event and stores answers. + 5. The tool coroutine wakes up, reads the answers, and returns to the LLM. + """ + + def __init__(self) -> None: + self._pending: Dict[str, PendingQuestion] = {} + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _new_id(self) -> str: + return f"que_{uuid.uuid4().hex[:12]}" + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def create( + self, + conv_id: str, + questions: List[dict], + request_id: Optional[str] = None, + ) -> PendingQuestion: + rid = request_id or self._new_id() + pq = PendingQuestion(request_id=rid, conv_id=conv_id, questions=questions) + self._pending[rid] = pq + logger.info("QuestionManager.create: request_id=%s, n=%d", rid, len(questions)) + return pq + + def reply(self, request_id: str, answers: List[List[str]]) -> None: + pq = self._pending.get(request_id) + if not pq: + raise KeyError(f"No pending question: {request_id}") + pq.answers = answers + pq.event.set() + logger.info("QuestionManager.reply: request_id=%s answered", request_id) + + def reject(self, request_id: str) -> None: + pq = self._pending.get(request_id) + if not pq: + raise KeyError(f"No pending question: {request_id}") + pq.rejected = True + pq.event.set() + logger.info("QuestionManager.reject: request_id=%s rejected", request_id) + + def remove(self, request_id: str) -> None: + self._pending.pop(request_id, None) + + def list_pending(self, conv_id: Optional[str] = None) -> List[dict]: + result = [] + for pq in self._pending.values(): + if conv_id is None or pq.conv_id == conv_id: + result.append( + { + "request_id": pq.request_id, + "conv_id": pq.conv_id, + "questions": pq.questions, + } + ) + return result + + +# --------------------------------------------------------------------------- +# Module-level singleton — same pattern as _todo_list / REACT_AGENT_MEMORY_CACHE +# --------------------------------------------------------------------------- +question_manager = QuestionManager() diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/select_skill.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/select_skill.py new file mode 100644 index 000000000..07b019c4d --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/select_skill.py @@ -0,0 +1,63 @@ +"""select_skill tool — matches a skill from registry based on user query.""" + +import json +from typing import Any, Dict + +from dbgpt.agent.resource.tool.base import tool + + +def make_select_skill(react_state: Dict[str, Any], registry: Any): + """Return a ``select_skill`` FunctionTool bound to the given react_state.""" + + @tool( + description="Select the most relevant skill based on user query from the " + "available skills list in system prompt." + ) + def select_skill(query: str) -> str: + def _is_excel_skill(meta) -> bool: + name = (meta.name or "").lower() + desc = (meta.description or "").lower() + tags = [tag.lower() for tag in (meta.tags or [])] + return any( + token in name or token in desc or token in tags + for token in ["excel", "xlsx", "xls", "spreadsheet"] + ) + + def _mentions_excel(text: str) -> bool: + t = (text or "").lower() + return any( + kw in t for kw in ["excel", "xlsx", "xls", "spreadsheet", "表格"] + ) + + match_input = query or "" + if react_state.get("file_path"): + match_input = f"{match_input} excel xlsx spreadsheet file" + matched = registry.match_skill(match_input) + if ( + matched + and _is_excel_skill(matched.metadata) + and not (_mentions_excel(query) or react_state.get("file_path")) + ): + matched = None + react_state["matched"] = matched + if matched: + detail = ( + f"Matched: {matched.metadata.name} - {matched.metadata.description}" + ) + return json.dumps( + {"chunks": [{"output_type": "text", "content": detail}]}, + ensure_ascii=False, + ) + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "No skill matched; proceed without skill", + } + ] + }, + ensure_ascii=False, + ) + + return select_skill diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/shell_interpreter.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/shell_interpreter.py new file mode 100644 index 000000000..7ceef2073 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/shell_interpreter.py @@ -0,0 +1,186 @@ +"""shell_interpreter tool — run bash commands in a sandboxed environment.""" + +import json +import logging +import os +import uuid +from typing import Any, Dict, List + +from dbgpt.agent.resource.tool.base import tool + +logger = logging.getLogger(__name__) + + +def make_shell_interpreter(react_state: Dict[str, Any]): + @tool( + description=( + "Execute shell/bash commands in a sandboxed environment. " + "Use this tool when you need to run shell commands such as ls, cat, " + "grep, curl, apt, pip, git, or any other CLI tool. " + "The sandbox provides resource limits (256MB memory, 30s timeout) " + "and process isolation. " + 'Parameters: {"code": "shell command(s) to execute"}' + ) + ) + async def shell_interpreter(code: str) -> str: + """Execute shell/bash commands in a sandboxed environment.""" + if not code or not code.strip(): + return json.dumps( + {"chunks": [{"output_type": "text", "content": "No command provided"}]}, + ensure_ascii=False, + ) + + try: + from dbgpt_sandbox.sandbox.execution_layer.base import ( + ExecutionStatus, + SessionConfig, + ) + from dbgpt_sandbox.sandbox.execution_layer.local_runtime import LocalRuntime + except ImportError: + return json.dumps( + { + "chunks": [ + {"output_type": "code", "content": code.strip()}, + { + "output_type": "text", + "content": ( + "Error: dbgpt-sandbox package is not installed. " + "Please install it with: pip install dbgpt-sandbox" + ), + }, + ] + }, + ensure_ascii=False, + ) + + from dbgpt.configs.model_config import ROOT_PATH + + session_id = f"bash_{uuid.uuid4().hex[:12]}" + runtime = LocalRuntime() + sandbox_work_dir = ROOT_PATH + os.makedirs(sandbox_work_dir, exist_ok=True) + + config = SessionConfig( + language="bash", + working_dir=sandbox_work_dir, + max_memory=256 * 1024 * 1024, # 256MB + timeout=30, + ) + + output_text = "" + try: + session = await runtime.create_session(session_id, config) + result = await session.execute(code) + + if result.status == ExecutionStatus.SUCCESS: + output_text = result.output or "" + elif result.status == ExecutionStatus.TIMEOUT: + output_text = f"Execution timed out ({config.timeout}s limit)" + else: + output_text = result.error or "Unknown execution error" + if result.output: + output_text = result.output + "\n[ERROR]\n" + output_text + except Exception as e: + output_text = f"Sandbox execution error: {e}" + finally: + try: + await runtime.destroy_session(session_id) + except Exception: + pass + + chunks: List[Dict[str, Any]] = [ + {"output_type": "code", "content": code.strip()}, + ] + if output_text.strip(): + chunks.append({"output_type": "text", "content": output_text.strip()}) + else: + chunks.append({"output_type": "text", "content": "(no output)"}) + + # Safety-net post-processing for skill script execution + _code_lower = code.strip().lower() + _is_skill_script = "skills/" in _code_lower and ".py" in _code_lower + if _is_skill_script and output_text.strip(): + import shutil + + from dbgpt.configs.model_config import PILOT_PATH, STATIC_MESSAGE_IMG_PATH + + IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} + + if "calculate_ratios" in _code_lower: + try: + ratio_data = json.loads(output_text.strip()) + if isinstance(ratio_data, dict): + react_state["ratio_data"] = ratio_data + except Exception: + pass + + if "generate_charts" in _code_lower: + try: + os.makedirs(STATIC_MESSAGE_IMG_PATH, exist_ok=True) + try: + chart_output = json.loads(output_text.strip()) + if isinstance(chart_output, dict): + chart_map = chart_output.get("charts", chart_output) + for name, abs_path in chart_map.items(): + if isinstance(abs_path, str) and os.path.isfile( + abs_path + ): + ext = os.path.splitext(abs_path)[1].lower() + if ext in IMAGE_EXTS: + unique_name = f"{uuid.uuid4().hex[:8]}_{os.path.basename(abs_path)}" + dest = os.path.join( + STATIC_MESSAGE_IMG_PATH, unique_name + ) + shutil.copy2(abs_path, dest) + img_url = f"/images/{unique_name}" + react_state.setdefault( + "generated_images", [] + ).append(img_url) + orig_stem = os.path.splitext( + os.path.basename(abs_path) + )[0].lower() + react_state.setdefault("image_url_map", {})[ + orig_stem + ] = img_url + except (json.JSONDecodeError, TypeError): + pass + + cid = react_state.get("conv_id") or "default" + out_dir = os.path.join(PILOT_PATH, "tmp", cid) + if os.path.isdir(out_dir): + for fname in os.listdir(out_dir): + ext = os.path.splitext(fname)[1].lower() + if ext in IMAGE_EXTS: + abs_path = os.path.join(out_dir, fname) + orig_stem = os.path.splitext(fname)[0].lower() + if orig_stem not in react_state.get( + "image_url_map", {} + ): + unique_name = f"{uuid.uuid4().hex[:8]}_{fname}" + dest = os.path.join( + STATIC_MESSAGE_IMG_PATH, unique_name + ) + shutil.copy2(abs_path, dest) + img_url = f"/images/{unique_name}" + react_state.setdefault( + "generated_images", [] + ).append(img_url) + react_state.setdefault("image_url_map", {})[ + orig_stem + ] = img_url + + all_images = react_state.get("generated_images", []) + if all_images: + img_summary = ( + "已生成的图片URL（在生成HTML报告时请使用这些URL）:\n" + + "\n".join(f" - {url}" for url in all_images) + ) + chunks.append({"output_type": "text", "content": img_summary}) + except Exception as e: + logger.warning( + "shell_interpreter: image post-processing failed: %s", e + ) + + return json.dumps({"chunks": chunks}, ensure_ascii=False) + + return shell_interpreter diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/skill_tools.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/skill_tools.py new file mode 100644 index 000000000..0054f6483 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/skill_tools.py @@ -0,0 +1,211 @@ +"""load_skill tool — loads skill content (SKILL.md) by name.""" + +import json +from typing import Any, Dict + +from dbgpt.agent.resource.tool.base import tool + + +def make_load_skill(react_state: Dict[str, Any]): + """Return a ``load_skill`` FunctionTool bound to the given react_state.""" + + @tool( + description="Load skill content by skill name and file path. " + "Returns the SKILL.md content of the specified skill. " + '参数: {"skill_name": "技能名称", "file_path": "技能文件路径"}' + ) + def load_skill(skill_name: str, file_path: str) -> str: + """Load the skill content (SKILL.md) by skill name and file path.""" + from dbgpt.agent.claude_skill import get_registry + + registry = get_registry() + matched = registry.get_skill(skill_name) + + if not matched: + for s in registry.list_skills(): + if s.name.lower() == skill_name.lower(): + matched = registry.get_skill(s.name) + break + + if not matched: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Skill '{skill_name}' not found", + } + ] + }, + ensure_ascii=False, + ) + + react_state["matched"] = matched + react_state["skill_prompt"] = matched.get_prompt() + + chunks = [ + {"output_type": "text", "content": f"Skill: {matched.metadata.name}"}, + {"output_type": "text", "content": f"File path: {file_path}"}, + {"output_type": "text", "content": "---"}, + ] + + if matched.instructions: + chunks.append({"output_type": "markdown", "content": matched.instructions}) + elif matched.prompt_template: + prompt_text = ( + matched.prompt_template.template + if hasattr(matched.prompt_template, "template") + else str(matched.prompt_template) + ) + chunks.append({"output_type": "markdown", "content": prompt_text}) + + return json.dumps({"chunks": chunks}, ensure_ascii=False) + + return load_skill + + +def make_execute_skill_script_file(react_state: Dict[str, Any]): + """Return an ``execute_skill_script_file`` FunctionTool bound to react_state.""" + + @tool( + description="执行技能scripts目录下的脚本文件。参数: " + '{"skill_name": "技能名称", "script_file_name": "脚本文件名", "args": {参数}}' + ) + async def execute_skill_script_file( + skill_name: str, script_file_name: str, args: dict | None = None + ) -> str: + """Execute a script file from a skill's scripts directory.""" + import os + import shutil + import uuid + + from dbgpt.agent.skill.manage import get_skill_manager + from dbgpt.configs.model_config import PILOT_PATH, STATIC_MESSAGE_IMG_PATH + from dbgpt._private.config import Config + + from dbgpt_app.openapi.api_v1.tools._helpers import ( + _extract_auto_data_markers, + ) + + CFG = Config() + + try: + sm = get_skill_manager(CFG.SYSTEM_APP) + cid = react_state.get("conv_id") or "default" + out_dir = os.path.join(PILOT_PATH, "tmp", cid) + os.makedirs(out_dir, exist_ok=True) + + real_file_path = react_state.get("file_path") + if real_file_path and args: + _FILE_PATH_KEYS = { + "input_file", + "file_path", + "data_path", + "csv_path", + "excel_path", + "data_file", + } + for key in list(args.keys()): + if key in _FILE_PATH_KEYS: + args[key] = real_file_path + + result_str = await sm.execute_skill_script_file( + skill_name, + script_file_name, + args or {}, + output_dir=out_dir, + ) + + try: + _skill_path = sm._get_skill_path(skill_name) + _sf = script_file_name.lstrip("/\\") + if _sf.startswith("scripts/") or _sf.startswith("scripts\\"): + _sf = _sf[8:] + _script_abs = os.path.join(_skill_path, "scripts", _sf) + with open(_script_abs, "r", encoding="utf-8") as _f: + _script_source = _f.read() + except Exception: + _script_source = None + + try: + result_obj = json.loads(result_str) + chunks = result_obj.get("chunks", []) + if _script_source: + chunks.insert(0, {"output_type": "code", "content": _script_source}) + + IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"} + os.makedirs(STATIC_MESSAGE_IMG_PATH, exist_ok=True) + for chunk in chunks: + if chunk.get("output_type") == "image": + abs_path = chunk["content"] + if os.path.isabs(abs_path) and os.path.isfile(abs_path): + ext = os.path.splitext(abs_path)[1].lower() + if ext in IMAGE_EXTS: + unique_name = ( + f"{uuid.uuid4().hex[:8]}_" + f"{os.path.basename(abs_path)}" + ) + dest = os.path.join( + STATIC_MESSAGE_IMG_PATH, unique_name + ) + shutil.copy2(abs_path, dest) + img_url = f"/images/{unique_name}" + chunk["content"] = img_url + react_state.setdefault("generated_images", []).append( + img_url + ) + orig_stem = os.path.splitext( + os.path.basename(abs_path) + )[0].lower() + react_state.setdefault("image_url_map", {})[ + orig_stem + ] = img_url + + all_images = react_state.get("generated_images", []) + if all_images: + img_summary = ( + "已生成的图片URL（在生成HTML报告时请使用这些URL）:\n" + + "\n".join(f" - {url}" for url in all_images) + ) + chunks.append({"output_type": "text", "content": img_summary}) + + auto_data = react_state.get("auto_data") + if not isinstance(auto_data, dict): + auto_data = {} + react_state["auto_data"] = auto_data + + filtered_chunks = [] + for chunk in chunks: + if chunk.get("output_type") != "text": + filtered_chunks.append(chunk) + continue + content = chunk.get("content") or "" + cleaned, extracted = _extract_auto_data_markers(content) + if extracted: + auto_data.update(extracted) + if cleaned: + chunk["content"] = cleaned + filtered_chunks.append(chunk) + elif not extracted: + filtered_chunks.append(chunk) + chunks = filtered_chunks + + if script_file_name == "calculate_ratios.py": + for chunk in chunks: + if chunk.get("output_type") == "text": + try: + ratio_data = json.loads(chunk["content"]) + react_state["ratio_data"] = ratio_data + except Exception: + pass + + return json.dumps({"chunks": chunks}, ensure_ascii=False) + except (json.JSONDecodeError, KeyError): + return result_str + except Exception as e: + return json.dumps( + {"chunks": [{"output_type": "text", "content": f"Error: {str(e)}"}]}, + ensure_ascii=False, + ) + + return execute_skill_script_file diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/sql_query.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/sql_query.py new file mode 100644 index 000000000..5083600df --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/sql_query.py @@ -0,0 +1,102 @@ +"""sql_query tool — read-only SQL query against the selected database.""" + +import json +from typing import Any, Dict, Optional + +from dbgpt.agent.resource.tool.base import tool + + +def make_sql_query(react_state: Dict[str, Any], database_connector: Optional[Any]): + @tool( + description=( + "对用户选择的数据库执行 SQL 查询（仅支持 SELECT）。" + '参数: {"sql": "SELECT 语句"}' + ) + ) + def sql_query(sql: str) -> str: + """Execute a read-only SQL query against the selected database.""" + if database_connector is None: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "未选择数据库，请先在左侧面板选择一个数据源。", + } + ] + }, + ensure_ascii=False, + ) + + sql_stripped = sql.strip().rstrip(";") + sql_upper = sql_stripped.upper().lstrip() + forbidden = [ + "INSERT", + "UPDATE", + "DELETE", + "DROP", + "ALTER", + "TRUNCATE", + "CREATE", + "GRANT", + "REVOKE", + ] + for kw in forbidden: + if sql_upper.startswith(kw): + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": ( + f"安全限制: 不允许执行 {kw} 语句，仅支持 SELECT 查询。" + ), + } + ] + }, + ensure_ascii=False, + ) + + try: + result = database_connector.run(sql_stripped) + if not result: + return json.dumps( + { + "chunks": [ + {"output_type": "text", "content": "查询返回空结果。"} + ] + }, + ensure_ascii=False, + ) + + columns = result[0] + col_names = [str(c[0]) if isinstance(c, tuple) else str(c) for c in columns] + rows = result[1:] + + header = "| " + " | ".join(col_names) + " |" + separator = "| " + " | ".join(["---"] * len(col_names)) + " |" + md_rows = [] + for row in rows[:50]: + md_rows.append("| " + " | ".join(str(v) for v in row) + " |") + table = "\n".join([header, separator] + md_rows) + if len(rows) > 50: + table += f"\n\n（仅显示前 50 行，共 {len(rows)} 行）" + + return json.dumps( + {"chunks": [{"output_type": "markdown", "content": table}]}, + ensure_ascii=False, + ) + except Exception as e: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"SQL 执行失败: {str(e)}", + } + ] + }, + ensure_ascii=False, + ) + + return sql_query diff --git a/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/todowrite.py b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/todowrite.py new file mode 100644 index 000000000..9d15f2b39 --- /dev/null +++ b/packages/dbgpt-app/src/dbgpt_app/openapi/api_v1/tools/todowrite.py @@ -0,0 +1,76 @@ +"""todowrite tool — maintain a session-level structured task list.""" + +import json +from typing import Any, Callable, Dict, List + +from dbgpt.agent.resource.tool.base import tool + + +def make_todowrite( + todo_list: List[Dict[str, str]], + stream_callback: Callable, +): + """Return a ``todowrite`` FunctionTool that mutates ``todo_list`` in-place.""" + + @tool( + description=( + "Create and manage a structured task list for the current session. " + "Use this tool to plan complex tasks (3+ steps), track progress, " + "and show the user what you are doing. " + "Pass the FULL todo list every time (not incremental). " + "Each todo has: content (brief description), " + "status (pending | in_progress | completed | cancelled), " + "priority (high | medium | low). " + "Rules: only ONE task in_progress at a time; mark tasks completed " + "immediately after finishing; do NOT use for single trivial tasks." + '\nParameter: {"todos": [{"content": "...", "status": "...", ' + '"priority": "..."}]}' + ) + ) + def todowrite(todos: str) -> str: + """Update the session todo list (full replacement).""" + parsed: List[Dict[str, str]] = [] + try: + raw = json.loads(todos) if isinstance(todos, str) else todos + items = raw if isinstance(raw, list) else raw.get("todos", raw) + if isinstance(items, list): + for item in items: + parsed.append( + { + "content": str(item.get("content", "")), + "status": str(item.get("status", "pending")), + "priority": str(item.get("priority", "medium")), + } + ) + except Exception: + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": "Error: invalid todos JSON", + } + ] + }, + ensure_ascii=False, + ) + + todo_list.clear() + todo_list.extend(parsed) + + total = len(parsed) + done = sum(1 for t in parsed if t["status"] == "completed") + return json.dumps( + { + "chunks": [ + { + "output_type": "text", + "content": f"Todo list updated: {done}/{total} completed", + } + ], + "__todos__": parsed, + }, + ensure_ascii=False, + ) + + return todowrite diff --git a/web/hooks/use-react-agent-chat.ts b/web/hooks/use-react-agent-chat.ts index ca71d93d5..7411d217a 100644 --- a/web/hooks/use-react-agent-chat.ts +++ b/web/hooks/use-react-agent-chat.ts @@ -7,7 +7,13 @@ import { MessagePart, ToolPart } from '@/new-components/chat/content/OpenCodeSessionTurn'; import { ChatHistoryResponse } from '@/types/chat'; -import { ContextStatus, ReActSSEState, createReActSSEState, parseSSELine } from '@/utils/react-sse-parser'; +import { + ContextStatus, + ReActSSEState, + SSEQuestionAskedEvent, + createReActSSEState, + parseSSELine, +} from '@/utils/react-sse-parser'; import { useCallback, useEffect, useRef, useState } from 'react'; export interface ReActChatRequest { @@ -45,12 +51,15 @@ export interface UseReActAgentChatReturn { streamingTurn: StreamingTurn | null; isStreaming: boolean; contextStatus: ContextStatus | null; + pendingQuestion: SSEQuestionAskedEvent | null; sendMessage: ( request: ReActChatRequest, currentHistory: ChatHistoryResponse, order: number, ) => Promise; cancel: () => void; + replyQuestion: (requestId: string, answers: string[][]) => Promise; + rejectQuestion: (requestId: string) => Promise; } export function useReActAgentChat(options: UseReActAgentChatOptions = {}): UseReActAgentChatReturn { @@ -59,6 +68,7 @@ export function useReActAgentChat(options: UseReActAgentChatOptions = {}): UseRe const [streamingTurn, setStreamingTurn] = useState(null); const [isStreaming, setIsStreaming] = useState(false); const [contextStatus, setContextStatus] = useState(null); + const [pendingQuestion, setPendingQuestion] = useState(null); const abortControllerRef = useRef(null); const sseStateRef = useRef(null); @@ -113,6 +123,10 @@ export function useReActAgentChat(options: UseReActAgentChatOptions = {}): UseRe setContextStatus(latestContextStatus); } + // Update pending question state + const latestQuestion = sseStateRef.current.getPendingQuestion(); + setPendingQuestion(latestQuestion); + setStreamingTurn(prev => { if (!prev) return null; return { @@ -329,12 +343,41 @@ export function useReActAgentChat(options: UseReActAgentChatOptions = {}): UseRe [baseUrl, cancel, processSSELine, onHistoryUpdate, onError, onComplete], ); + const replyQuestion = useCallback(async (requestId: string, answers: string[][]) => { + try { + const res = await fetch(`/api/v1/chat/question/${requestId}/reply`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers }), + }); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + setPendingQuestion(null); + } catch (e) { + console.error('replyQuestion failed:', e); + } + }, []); + + const rejectQuestion = useCallback(async (requestId: string) => { + try { + const res = await fetch(`/api/v1/chat/question/${requestId}/reject`, { + method: 'POST', + }); + if (!res.ok) throw new Error(`HTTP ${res.status}`); + setPendingQuestion(null); + } catch (e) { + console.error('rejectQuestion failed:', e); + } + }, []); + return { streamingTurn, isStreaming, contextStatus, + pendingQuestion, sendMessage, cancel, + replyQuestion, + rejectQuestion, }; } diff --git a/web/new-components/chat/ChatPage.tsx b/web/new-components/chat/ChatPage.tsx index 9a4c55583..6f38f58aa 100644 --- a/web/new-components/chat/ChatPage.tsx +++ b/web/new-components/chat/ChatPage.tsx @@ -3,6 +3,7 @@ import React, { memo, useCallback, useMemo, useRef } from 'react'; import ChatHeader from './ChatHeader'; import ChatMessageList, { ChatTurn } from './ChatMessageList'; import ChatWelcome from './ChatWelcome'; +import QuestionDock, { QuestionRequest } from './content/QuestionDock'; import { SlashCommand } from './input/CommandPopover'; import { ContentPart } from './input/EnhancedChatInput'; import StandaloneChatInput, { StandaloneChatInputRef } from './input/StandaloneChatInput'; @@ -27,6 +28,10 @@ export interface ChatPageProps { inputPlaceholder?: string; showSteps?: boolean; + pendingQuestion?: QuestionRequest | null; + onReplyQuestion?: (requestId: string, answers: string[][]) => void; + onRejectQuestion?: (requestId: string) => void; + headerExtra?: React.ReactNode; welcomeExtra?: React.ReactNode; inputExtra?: React.ReactNode; @@ -54,6 +59,10 @@ const ChatPage: React.FC = ({ inputPlaceholder, showSteps = true, + pendingQuestion, + onReplyQuestion, + onRejectQuestion, + headerExtra, welcomeExtra, inputExtra, @@ -117,6 +126,15 @@ const ChatPage: React.FC = ({ )}
+ {pendingQuestion && onReplyQuestion && onRejectQuestion && ( +
+ +
+ )} void; + onReject: (requestId: string) => void; +} + +const QuestionDock: React.FC = ({ request, onReply, onReject }) => { + const [selected, setSelected] = useState( + () => request.questions.map(() => []), + ); + const [customInputs, setCustomInputs] = useState( + () => request.questions.map(() => ''), + ); + + const canSubmit = useMemo(() => { + return request.questions.every((q, i) => { + if (!q.multiple) { + return selected[i].length === 1 || (q.custom !== false && customInputs[i].trim() !== ''); + } + return selected[i].length > 0 || (q.custom !== false && customInputs[i].trim() !== ''); + }); + }, [selected, customInputs, request.questions]); + + const toggleOption = (qIndex: number, label: string, multiple: boolean) => { + setSelected(prev => { + const next = [...prev]; + const current = [...next[qIndex]]; + if (multiple) { + const idx = current.indexOf(label); + if (idx >= 0) current.splice(idx, 1); + else current.push(label); + } else { + current.length = 0; + current.push(label); + } + next[qIndex] = current; + return next; + }); + setCustomInputs(prev => { + const next = [...prev]; + if (!request.questions[qIndex].multiple) next[qIndex] = ''; + return next; + }); + }; + + const setCustom = (qIndex: number, value: string) => { + setSelected(prev => { + const next = [...prev]; + next[qIndex] = []; + return next; + }); + setCustomInputs(prev => { + const next = [...prev]; + next[qIndex] = value; + return next; + }); + }; + + const handleSubmit = () => { + const answers = request.questions.map((q, i) => { + if (selected[i].length > 0) return selected[i]; + if (q.custom !== false && customInputs[i].trim()) return [customInputs[i].trim()]; + return []; + }); + onReply(request.request_id, answers); + }; + + const handleDismiss = () => { + onReject(request.request_id); + }; + + return ( +
+ {/* Header */} +
+
+ + + + + 需要您的确认 + +
+ +
+ +
+ + {/* Questions */} +
+ {request.questions.map((q, qi) => ( +
+ {/* Header label */} + {q.header && ( +
+ {q.header} +
+ )} + {/* Question text */} +
+ {q.question} +
+ {/* Options */} +
+ {q.options.map((opt) => { + const isSelected = selected[qi].includes(opt.label); + return ( + + ); + })} +
+ {/* Custom input */} + {q.custom !== false && ( +
+ setCustom(qi, e.target.value)} + className='w-full rounded-lg border border-slate-200 bg-white px-3 py-1.5 text-[13px] leading-4 text-slate-700 placeholder-slate-400 outline-none transition-colors focus:border-sky-400 focus:ring-1 focus:ring-sky-400/30 dark:border-white/10 dark:bg-white/5 dark:text-slate-200 dark:placeholder-slate-500 dark:focus:border-sky-500/50' + /> +
+ )} +
+ ))} +
+ + {/* Footer with actions */} +
+ + +
+
+ ); +}; + +export default QuestionDock; \ No newline at end of file diff --git a/web/utils/react-sse-parser.ts b/web/utils/react-sse-parser.ts index 43b5df0c6..738b00030 100644 --- a/web/utils/react-sse-parser.ts +++ b/web/utils/react-sse-parser.ts @@ -66,12 +66,47 @@ export interface SSEContextStatusEvent { compact_layer?: string | null; } +// ── Human-in-the-loop: question events ────────────────────────────────────── + +export interface QuestionOption { + label: string; + description: string; +} + +export interface QuestionInfo { + question: string; + header: string; + options: QuestionOption[]; + multiple?: boolean; + custom?: boolean; +} + +export interface SSEQuestionAskedEvent { + type: 'question.asked'; + request_id: string; + conv_id: string; + questions: QuestionInfo[]; +} + +export interface SSEQuestionRepliedEvent { + type: 'question.replied'; + request_id: string; +} + +export interface SSEQuestionRejectedEvent { + type: 'question.rejected'; + request_id: string; +} + export type SSEEvent = | SSEStepStartEvent | SSEStepChunkEvent | SSEStepMetaEvent | SSEStepDoneEvent | SSEContextStatusEvent + | SSEQuestionAskedEvent + | SSEQuestionRepliedEvent + | SSEQuestionRejectedEvent | SSEFinalEvent | SSEDoneEvent; @@ -108,6 +143,7 @@ export class ReActSSEState { private startTime: number; private endTime?: number; private _contextStatus: ContextStatus | null = null; + private _pendingQuestion: SSEQuestionAskedEvent | null = null; constructor() { this.startTime = Date.now(); @@ -133,6 +169,13 @@ export class ReActSSEState { case 'context.status': this.handleContextStatus(event); break; + case 'question.asked': + this._pendingQuestion = event; + break; + case 'question.replied': + case 'question.rejected': + this._pendingQuestion = null; + break; case 'final': this.handleFinal(event); break; @@ -142,6 +185,13 @@ export class ReActSSEState { } } + /** + * Get the current pending question (null if none) + */ + getPendingQuestion(): SSEQuestionAskedEvent | null { + return this._pendingQuestion; + } + private handleStepStart(event: SSEStepStartEvent): void { const step: StepState = { id: event.id,