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