Coverage for merco/tools/errors.py: 57%

1"""工具执行错误处理 — 把异常转为 LLM 可消费的结构化 dict。 

2 

3边界：错误分类 + 公共消息脱敏。LLM 错误（APIStatusError 等）由 core.llm.errors 处理。 

4""" 

5from __future__ import annotations 

6 

7import logging 

8from typing import Any 

9 

10logger = logging.getLogger("merco.tools.errors") 

11 

12 

13ERROR_CATEGORIES = { 

14 "param_mismatch": "参数不匹配", 

15 "tool_not_found": "工具不存在", 

16 "timeout": "执行超时", 

17 "permission": "权限不足", 

18 "network": "网络错误", 

19 "resource_not_found": "资源不存在", 

20 "internal": "内部错误", 

21 "unknown": "未知错误", 

22} 

23 

24 

25def classify_error(exc: Exception) -> str: 

26 """将异常映射到错误类别""" 

27 if isinstance(exc, TypeError): 

28 return "param_mismatch" 

29 if isinstance(exc, TimeoutError): 

30 return "timeout" 

31 if isinstance(exc, PermissionError): 

32 return "permission" 

33 if isinstance(exc, FileNotFoundError): 

34 return "resource_not_found" 

35 if isinstance(exc, ConnectionError): 

36 return "network" 

37 if isinstance(exc, OSError): 

38 return "network" 

39 if isinstance(exc, ValueError): 

40 return "param_mismatch" 

41 return "unknown" 

42 

43 

44def tool_error( 

45 exc: Exception, 

46 tool_name: str, 

47 tool_schema: dict | None = None, 

48) -> dict: 

49 """将工具执行异常转为 LLM 可读的结构化错误""" 

50 category = classify_error(exc) 

51 label = ERROR_CATEGORIES.get(category, "未知错误") 

52 result: dict[str, Any] = { 

53 "error": f"[{label}] {tool_name}: {_public_message(exc)}", 

54 "category": category, 

55 "tool": tool_name, 

56 } 

57 if category == "param_mismatch": 

58 hint = _params_hint(tool_schema) 

59 result["suggestion"] = f"参数类型或值不正确。{hint}" 

60 if tool_schema: 

61 result["available_params"] = _param_names(tool_schema) 

62 elif category == "tool_not_found": 

63 result["suggestion"] = "该工具不可用。请使用其他可用工具完成用户请求。" 

64 elif category == "timeout": 

65 result["suggestion"] = "操作超时。可尝试减少数据量、拆分请求，或使用其他工具。" 

66 elif category == "permission": 

67 result["suggestion"] = "权限不足。请检查文件权限，或使用其他路径/工具。" 

68 elif category == "network": 

69 result["suggestion"] = "网络请求失败。可重试，或检查 URL 是否正确。" 

70 elif category == "resource_not_found": 

71 result["suggestion"] = "资源（文件/路径）不存在。请检查路径拼写，或搜索确认位置。" 

72 else: 

73 result["suggestion"] = "执行时发生意外错误。请尝试其他方式完成用户请求。" 

74 logger.warning("工具 %s 未知异常", tool_name, exc_info=True) 

75 return result 

76 

77 

78def empty_response() -> dict: 

79 """空回复错误 — 回调 LLM 让它产出实际内容""" 

80 return { 

81 "error": "[空回复] 你既没有回复用户也没有调用工具。" 

82 "请直接回答用户，或使用工具推进任务。", 

83 "category": "empty_response", 

84 "suggestion": "请直接回复用户，或调用工具获取信息。", 

85 } 

86 

87 

88def _public_message(exc: Exception) -> str: 

89 msg = str(exc) 

90 if len(msg) > 300: 

91 msg = msg[:300] + "..." 

92 return msg 

93 

94 

95def _params_hint(schema: dict | None) -> str: 

96 if not schema: 

97 return "请检查工具调用参数。" 

98 names = _param_names(schema) 

99 required = schema.get("required", []) 

100 if required: 

101 return f"必需: {', '.join(required)}。可用: {', '.join(names)}。" 

102 return f"可用参数: {', '.join(names)}。" 

103 

104 

105def _param_names(schema: dict) -> list[str]: 

106 props = schema.get("properties", {}) 

107 return list(props.keys()) if isinstance(props, dict) else []