From 6b67bf6f93377cab8906c11e777c7252672bf033 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Mon, 27 Oct 2025 23:41:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=9E=81=E5=A4=A7=E7=9A=84=E6=8F=90?= =?UTF-8?q?=E5=8D=87=20SQL=20=E5=B7=A5=E5=85=B7=E8=B0=83=E7=94=A8=E7=9A=84?= =?UTF-8?q?=E7=A8=B3=E5=AE=9A=E6=80=A7=EF=BC=8C=E7=A7=BB=E9=99=A4=20LIMINT?= =?UTF-8?q?=20=E5=B9=B6=E5=B0=86=E9=BB=98=E8=AE=A4=E8=B6=85=E6=97=B6?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E4=BF=AE=E6=94=B9=E4=B8=BA=2060s=EF=BC=8C?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0MySQL=E8=BF=9E=E6=8E=A5=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E5=99=A8=E5=92=8C=E5=AE=89=E5=85=A8=E6=A3=80=E6=9F=A5=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=EF=BC=8C=E4=BC=98=E5=8C=96=E6=9F=A5=E8=AF=A2=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E5=92=8C=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86=E9=80=BB?= =?UTF-8?q?=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 重构 MySQLConnectionManager，在使用前验证游标，向使用者提供辅助函数，并使用基于线程池的超时机制替代 SIGALRM，使查询可运行长达 600 秒，同时安全地失效不良连接（src/agents/common/toolkits/mysql/connection.py） - 扩大安全防护中的允许超时窗口，移除未使用的限制验证器，并更新查询模型默认值（src/agents/common/toolkits/mysql/security.py，src/agents/common/toolkits/mysql/tools.py） - 使 mysql_query 依赖 execute_query_with_timeout，移除隐式 LIMIT 注入，确保失败时连接保持健康，扩展结果格式化和故障排除提示；记录 list-table 调用，并在避免按表 COUNT 扫描的同时，丰富 mysql_describe_table 的列注释（src/agents/common/toolkits/mysql/tools.py） - 将高级代理指南重命名为 agents-config，更新 VitePress 导航，并在文档中增加 LangGraph/MCP/MySQL 配置指导（docs/.vitepress/config.mts，docs/advanced/agents-config.md） - 调整路线图错误列表，将 LangGraph 工具问题标记为已解决，并跟踪 LightRAG 查看器的缺失功能（docs/changelog/roadmap.md） - 添加 CLAUDE.md，引导贡献者参阅 AGENTS.md 获取更多信息 --- CLAUDE.md | 1 + docs/.vitepress/config.mts | 2 +- docs/advanced/{agents.md => agents-config.md} | 2 +- docs/changelog/roadmap.md | 5 +- .../common/toolkits/mysql/connection.py | 131 ++++++++----- src/agents/common/toolkits/mysql/security.py | 7 +- src/agents/common/toolkits/mysql/tools.py | 173 ++++++++++-------- 7 files changed, 190 insertions(+), 131 deletions(-) create mode 100644 CLAUDE.md rename docs/advanced/{agents.md => agents-config.md} (98%) diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000..a865c741 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +Read AGENTS.md \ No newline at end of file diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index 49caf5cf..18879225 100644 --- a/docs/.vitepress/config.mts +++ b/docs/.vitepress/config.mts @@ -35,7 +35,7 @@ export default defineConfig({ items: [ { text: '配置系统详解', link: '/advanced/configuration' }, { text: '文档解析', link: '/advanced/document-processing' }, - { text: '智能体', link: '/advanced/agents' }, + { text: '智能体', link: '/advanced/agents-config' }, { text: '品牌自定义', link: '/advanced/branding' }, { text: '其他配置', link: '/advanced/misc' } ] diff --git a/docs/advanced/agents.md b/docs/advanced/agents-config.md similarity index 98% rename from docs/advanced/agents.md rename to docs/advanced/agents-config.md index ee82231d..a75edbc1 100644 --- a/docs/advanced/agents.md +++ b/docs/advanced/agents-config.md @@ -101,7 +101,7 @@ MYSQL_CHARSET=utf8mb4 - ✅ **只读操作**: 仅允许 SELECT、SHOW、DESCRIBE、EXPLAIN 操作 - ✅ **SQL 注入防护**: 严格的表名参数验证 -- ✅ **超时控制**: 默认 10 秒，最大 60 秒 +- ✅ **超时控制**: 默认 60 秒，最大 600 秒 - ✅ **结果限制**: 默认 10000 字符，100 行，最大 1000 行 ### 可视化图表-MCP-Server diff --git a/docs/changelog/roadmap.md b/docs/changelog/roadmap.md index a7edeff4..b8b31fb9 100644 --- a/docs/changelog/roadmap.md +++ b/docs/changelog/roadmap.md @@ -7,8 +7,8 @@ ## Bugs -- [ ] v1 版本的 LangGraph 的工具渲染有问题 - [ ] upload 接口会阻塞主进程 +- [ ] LightRAG 知识库查看不了解析后的文本 ## Next @@ -39,4 +39,5 @@ - [x] 修改现有的智能体Demo，并尽量将默认助手的特性兼容到 LangGraph 的 [`create_agent`](https://docs.langchain.com/oss/python/langchain/agents) 中 - [x] 基于 create_agent 创建 SQL Viewer 智能体 - [x] 优化 MCP 逻辑，支持 common + special 创建方式 -- [x] 修复本地知识库的 metadata 和 向量数据库中不一致的情况。 \ No newline at end of file +- [x] 修复本地知识库的 metadata 和 向量数据库中不一致的情况。 +- [x] v1 版本的 LangGraph 的工具渲染有问题 \ No newline at end of file diff --git a/src/agents/common/toolkits/mysql/connection.py b/src/agents/common/toolkits/mysql/connection.py index 8268ec14..495f4b88 100644 --- a/src/agents/common/toolkits/mysql/connection.py +++ b/src/agents/common/toolkits/mysql/connection.py @@ -1,4 +1,4 @@ -import signal +import concurrent.futures import threading import time from contextlib import contextmanager @@ -93,48 +93,65 @@ class MySQLConnectionManager: pass return False + def _invalidate_connection(self, connection: pymysql.Connection | None = None): + """关闭并清理失效的连接""" + try: + if connection: + connection.close() + except Exception: + pass + finally: + self.connection = None + @contextmanager def get_cursor(self): """获取数据库游标的上下文管理器""" max_retries = 2 + cursor = None + connection = None + last_error: Exception | None = None + + # 优先确保成功获取游标再交给调用方执行查询 for attempt in range(max_retries): try: connection = self._get_connection() cursor = connection.cursor() - - try: - yield cursor - connection.commit() - break # 成功，退出重试循环 - - except Exception as e: - connection.rollback() - - # 如果是连接错误，尝试重新连接 - if "MySQL" in str(e) or "connection" in str(e).lower(): - if attempt < max_retries - 1: - logger.warning(f"Connection error, retrying (attempt {attempt + 1}): {e}") - # 强制重新连接 - if self.connection: - try: - self.connection.close() - except Exception as _: - pass - self.connection = None - time.sleep(1) - continue - - raise e # 其他错误直接抛出 - - finally: - if cursor: - cursor.close() - + break except Exception as e: + last_error = e + logger.warning(f"Failed to acquire cursor (attempt {attempt + 1}): {e}") + self._invalidate_connection(connection) + cursor = None + connection = None if attempt == max_retries - 1: - raise e # 最后一次尝试失败，抛出异常 + raise e time.sleep(1) + if cursor is None or connection is None: + raise last_error or ConnectionError("Unable to acquire MySQL cursor") + + try: + yield cursor + connection.commit() + except Exception as e: + try: + connection.rollback() + except Exception: + pass + + # 标记连接失效，等待下一次获取时重建 + if "MySQL" in str(e) or "connection" in str(e).lower(): + logger.warning(f"MySQL connection error encountered, invalidating connection: {e}") + self._invalidate_connection(connection) + + raise + finally: + if cursor: + try: + cursor.close() + except Exception: + pass + def close(self): """关闭数据库连接""" if self.connection: @@ -142,6 +159,19 @@ class MySQLConnectionManager: self.connection = None logger.info("MySQL connection closed") + def get_connection(self) -> pymysql.Connection: + """对外暴露的连接获取方法""" + return self._get_connection() + + def invalidate_connection(self): + """手动标记连接失效""" + self._invalidate_connection(self.connection) + + @property + def database_name(self) -> str: + """返回当前配置的数据库名称""" + return self.config["database"] + class QueryTimeoutError(Exception): """查询超时异常""" @@ -156,25 +186,30 @@ class QueryResultTooLargeError(Exception): def execute_query_with_timeout(connection: pymysql.Connection, sql: str, params: tuple = None, timeout: int = 10): - """带超时的查询执行""" + """使用线程池实现超时控制，避免信号导致的生成器问题""" - def timeout_handler(_signum, _frame): - raise QueryTimeoutError(f"Query timeout after {timeout} seconds") - - # 设置信号处理 - old_handler = signal.signal(signal.SIGALRM, timeout_handler) - signal.alarm(timeout) - - try: + def query_worker(): + """查询工作函数，在单独线程中执行""" cursor = connection.cursor(DictCursor) - cursor.execute(sql, params or ()) - result = cursor.fetchall() - cursor.close() - return result - finally: - # 恢复原始信号处理 - signal.alarm(0) - signal.signal(signal.SIGALRM, old_handler) + try: + if params is None: + cursor.execute(sql) + else: + cursor.execute(sql, params) + result = cursor.fetchall() + return result + finally: + cursor.close() + + # 使用线程池执行查询，设置超时 + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(query_worker) + try: + return future.result(timeout=timeout) + except concurrent.futures.TimeoutError: + # 尝试取消任务 + future.cancel() + raise QueryTimeoutError(f"Query timeout after {timeout} seconds") def limit_result_size(result: list, max_chars: int = 10000) -> list: diff --git a/src/agents/common/toolkits/mysql/security.py b/src/agents/common/toolkits/mysql/security.py index dcb91556..a87f1c29 100644 --- a/src/agents/common/toolkits/mysql/security.py +++ b/src/agents/common/toolkits/mysql/security.py @@ -76,12 +76,7 @@ class MySQLSecurityChecker: # 检查表名只包含字母、数字、下划线 return bool(re.match(r"^[a-zA-Z_][a-zA-Z0-9_]*$", table_name)) - @classmethod - def validate_limit(cls, limit: int) -> bool: - """验证limit参数""" - return isinstance(limit, int) and 0 < limit <= 1000 - @classmethod def validate_timeout(cls, timeout: int) -> bool: """验证timeout参数""" - return isinstance(timeout, int) and 1 <= timeout <= 60 + return isinstance(timeout, int) and 1 <= timeout <= 600 diff --git a/src/agents/common/toolkits/mysql/tools.py b/src/agents/common/toolkits/mysql/tools.py index d4bb401a..3f12cac9 100644 --- a/src/agents/common/toolkits/mysql/tools.py +++ b/src/agents/common/toolkits/mysql/tools.py @@ -5,7 +5,12 @@ from pydantic import BaseModel, Field from src.utils import logger -from .connection import MySQLConnectionManager, limit_result_size +from .connection import ( + MySQLConnectionManager, + QueryTimeoutError, + execute_query_with_timeout, + limit_result_size, +) from .exceptions import MySQLConnectionError from .security import MySQLSecurityChecker @@ -56,6 +61,7 @@ def mysql_list_tables() -> str: with conn_manager.get_cursor() as cursor: # 获取表名 cursor.execute("SHOW TABLES") + logger.debug("Executed `SHOW TABLES` query") tables = cursor.fetchall() if not tables: @@ -68,17 +74,18 @@ def mysql_list_tables() -> str: table_names.append(table_name) # 获取每个表的行数信息 - table_info = [] - for table_name in table_names: - try: - cursor.execute(f"SELECT COUNT(*) as count FROM `{table_name}`") - count_result = cursor.fetchone() - row_count = count_result["count"] - table_info.append(f"- {table_name} (约 {row_count} 行)") - except Exception: - table_info.append(f"- {table_name} (无法获取行数)") + # table_info = [] + # for table_name in table_names: + # try: + # cursor.execute(f"SELECT COUNT(*) as count FROM `{table_name}`") + # logger.debug(f"Executed `SELECT COUNT(*) FROM {table_name}` query") + # count_result = cursor.fetchone() + # row_count = count_result["count"] + # table_info.append(f"- {table_name} (约 {row_count} 行)") + # except Exception: + # table_info.append(f"- {table_name} (无法获取行数)") - result = "数据库中的表:\n" + "\n".join(table_info) + result = "数据库中的表:\n" + "\n".join(table_names) logger.info(f"Retrieved {len(table_names)} tables from database") return result @@ -116,9 +123,28 @@ def mysql_describe_table(table_name: Annotated[str, "要查询结构的表名"]) if not columns: return f"表 {table_name} 不存在或没有字段" + # 获取字段备注信息 + column_comments: dict[str, str] = {} + try: + cursor.execute( + """ + SELECT COLUMN_NAME, COLUMN_COMMENT + FROM information_schema.COLUMNS + WHERE TABLE_NAME = %s AND TABLE_SCHEMA = %s + """, + (table_name, conn_manager.database_name), + ) + comment_rows = cursor.fetchall() + for row in comment_rows: + column_name = row.get("COLUMN_NAME") + if column_name: + column_comments[column_name] = row.get("COLUMN_COMMENT") or "" + except Exception as e: + logger.warning(f"Failed to fetch column comments for table {table_name}: {e}") + # 格式化输出 result = f"表 `{table_name}` 的结构:\n\n" - result += "字段名\t\t类型\t\tNULL\t键\t默认值\t\t额外\n" + result += "字段名\t\t类型\t\tNULL\t键\t默认值\t\t额外\t备注\n" result += "-" * 80 + "\n" for col in columns: @@ -128,9 +154,13 @@ def mysql_describe_table(table_name: Annotated[str, "要查询结构的表名"]) key_str = col["Key"] or "" default_str = col.get("Default") or "" extra_str = col.get("Extra") or "" + comment_str = column_comments.get(field, "") # 格式化输出 - result += f"{field:<16}\t{type_str:<16}\t{null_str:<8}\t{key_str:<4}\t{default_str:<16}\t{extra_str}\n" + result += ( + f"{field:<16}\t{type_str:<16}\t{null_str:<8}\t{key_str:<4}\t" + f"{default_str:<16}\t{extra_str:<16}\t{comment_str}\n" + ) # 获取索引信息 try: @@ -164,15 +194,13 @@ class QueryModel(BaseModel): """执行SQL查询的参数模型""" sql: str = Field(description="要执行的SQL查询语句（只能是SELECT语句）", example="SELECT * FROM users WHERE id = 1") - limit: int | None = Field(default=100, description="限制返回的最大行数，默认100，最大1000", ge=1, le=1000) - timeout: int | None = Field(default=10, description="查询超时时间（秒），默认10秒，最大60秒", ge=1, le=60) + timeout: int | None = Field(default=60, description="查询超时时间（秒），默认60秒，最大600秒", ge=1, le=600) @tool(name_or_callable="执行 SQL 查询", args_schema=QueryModel) def mysql_query( sql: Annotated[str, "要执行的SQL查询语句（只能是SELECT语句）"], - limit: Annotated[int | None, "限制返回的最大行数，默认100，最大1000"] = 100, - timeout: Annotated[int | None, "查询超时时间（秒），默认10秒，最大60秒"] = 10, + timeout: Annotated[int | None, "查询超时时间（秒），默认60秒，最大600秒"] = 60, ) -> str: """执行只读的SQL查询语句 @@ -181,7 +209,6 @@ def mysql_query( 参数: - sql: SQL查询语句 - - limit: 返回结果的最大行数（防止结果过大） - timeout: 查询超时时间（防止长时间运行的查询） """ try: @@ -189,88 +216,88 @@ def mysql_query( if not MySQLSecurityChecker.validate_sql(sql): return "SQL语句包含不安全的操作或可能的注入攻击，请检查SQL语句" - # 验证参数 - if not MySQLSecurityChecker.validate_limit(limit): - return "limit参数必须在1-1000之间" - if not MySQLSecurityChecker.validate_timeout(timeout): - return "timeout参数必须在1-60之间" - - # 如果SQL中没有LIMIT，添加LIMIT子句 - if "LIMIT" not in sql.upper() and "limit" not in sql: - sql_upper = sql.strip().upper() - # 确保是SELECT语句才添加LIMIT - if sql_upper.startswith("SELECT"): - sql = f"{sql} LIMIT {limit}" + return "timeout参数必须在1-600之间" conn_manager = get_connection_manager() + connection = conn_manager.get_connection() - with conn_manager.get_cursor() as cursor: - # 执行查询 - cursor.execute(sql) - result = cursor.fetchall() + effective_timeout = timeout or 60 + try: + result = execute_query_with_timeout(connection, sql, timeout=effective_timeout) + except QueryTimeoutError as timeout_error: + logger.error(f"MySQL query timed out after {effective_timeout} seconds: {timeout_error}") + raise + except Exception: + conn_manager.invalidate_connection() + raise - if not result: - return "查询执行成功，但没有返回任何结果" + if not result: + return "查询执行成功，但没有返回任何结果" - # 限制结果大小 - limited_result = limit_result_size(result, max_chars=10000) + # 限制结果大小 + limited_result = limit_result_size(result, max_chars=10000) - # 检查结果是否被截断 - if len(limited_result) < len(result): - warning = f"\n\n⚠️ 警告: 查询结果过大，只显示了前 {len(limited_result)} 行（共 {len(result)} 行）。\n" - warning += "建议使用更精确的查询条件或使用LIMIT子句来减少返回的数据量。" - else: - warning = "" + # 检查结果是否被截断 + if len(limited_result) < len(result): + warning = f"\n\n⚠️ 警告: 查询结果过大，只显示了前 {len(limited_result)} 行（共 {len(result)} 行）。\n" + warning += "建议使用更精确的查询条件或使用LIMIT子句来减少返回的数据量。" + else: + warning = "" - # 格式化输出 - if limited_result: - # 获取列名 - columns = list(limited_result[0].keys()) + # 格式化输出 + if limited_result: + # 获取列名 + columns = list(limited_result[0].keys()) - # 计算每列的最大宽度 - col_widths = {} - for col in columns: - col_widths[col] = max(len(str(col)), max(len(str(row.get(col, ""))) for row in limited_result)) - col_widths[col] = min(col_widths[col], 50) # 限制最大宽度 + # 计算每列的最大宽度 + col_widths = {} + for col in columns: + col_widths[col] = max(len(str(col)), max(len(str(row.get(col, ""))) for row in limited_result)) + col_widths[col] = min(col_widths[col], 50) # 限制最大宽度 - # 构建表头 - header = "| " + " | ".join(f"{col:<{col_widths[col]}}" for col in columns) + " |" - separator = "|" + "|".join("-" * (col_widths[col] + 2) for col in columns) + "|" + # 构建表头 + header = "| " + " | ".join(f"{col:<{col_widths[col]}}" for col in columns) + " |" + separator = "|" + "|".join("-" * (col_widths[col] + 2) for col in columns) + "|" - # 构建数据行 - rows = [] - for row in limited_result: - row_str = "| " + " | ".join(f"{str(row.get(col, '')):<{col_widths[col]}}" for col in columns) + " |" - rows.append(row_str) + # 构建数据行 + rows = [] + for row in limited_result: + row_str = "| " + " | ".join(f"{str(row.get(col, '')):<{col_widths[col]}}" for col in columns) + " |" + rows.append(row_str) - result_str = f"查询结果（共 {len(limited_result)} 行）:\n\n" - result_str += header + "\n" + separator + "\n" - result_str += "\n".join(rows[:50]) # 最多显示50行 + result_str = f"查询结果（共 {len(limited_result)} 行）:\n\n" + result_str += header + "\n" + separator + "\n" + result_str += "\n".join(rows[:50]) # 最多显示50行 - if len(rows) > 50: - result_str += f"\n\n... 还有 {len(rows) - 50} 行未显示 ..." + if len(rows) > 50: + result_str += f"\n\n... 还有 {len(rows) - 50} 行未显示 ..." - result_str += warning + result_str += warning - logger.info(f"Query executed successfully, returned {len(limited_result)} rows") - return result_str - else: - return "查询执行成功，但没有返回任何结果" + logger.info(f"Query executed successfully, returned {len(limited_result)} rows") + return result_str + + return "查询执行成功，但返回数据为空" except Exception as e: - error_msg = f"SQL查询执行失败: {str(e)}" + error_msg = f"SQL查询执行失败: {str(e)}\n\n{sql}" # 提供更有用的错误信息 if "timeout" in str(e).lower(): error_msg += "\n\n💡 建议：查询超时了，请尝试以下方法：\n" error_msg += "1. 减少查询的数据量（使用WHERE条件过滤）\n" error_msg += "2. 使用LIMIT子句限制返回行数\n" - error_msg += "3. 增加timeout参数值（最大60秒）" + error_msg += "3. 增加timeout参数值（最大600秒）" elif "table" in str(e).lower() and "doesn't exist" in str(e).lower(): error_msg += "\n\n💡 建议：表不存在，请使用 mysql_list_tables 查看可用的表名" elif "column" in str(e).lower() and "doesn't exist" in str(e).lower(): error_msg += "\n\n💡 建议：列不存在，请使用 mysql_describe_table 查看表结构" + elif "not enough arguments for format string" in str(e).lower(): + error_msg += ( + "\n\n💡 建议：SQL 中的百分号 (%) 被当作参数占位符使用。" + " 如需匹配包含百分号的文本，请将百分号写成双百分号 (%%) 或使用参数化查询。" + ) logger.error(error_msg) return error_msg