Coverage for langsmith/env/_runtime_env.py: 21%

1"""Environment information.""" 

2 

3import functools 

4import logging 

5import os 

6import platform 

7import subprocess 

8from typing import Dict, List, Optional, Union 

9 

10from langsmith.utils import get_docker_compose_command 

11from langsmith.env._git import exec_git 

12 

13try: 

14 # psutil is an optional dependency 

15 import psutil 

16 

17 _PSUTIL_AVAILABLE = True 

18except ImportError: 

19 _PSUTIL_AVAILABLE = False 

20logger = logging.getLogger(__name__) 

21 

22 

23def get_runtime_and_metrics() -> dict: 

24 """Get the runtime information as well as metrics.""" 

25 return {**get_runtime_environment(), **get_system_metrics()} 

26 

27 

28def get_system_metrics() -> Dict[str, Union[float, dict]]: 

29 """Get CPU and other performance metrics.""" 

30 global _PSUTIL_AVAILABLE 

31 if not _PSUTIL_AVAILABLE: 

32 return {} 

33 try: 

34 process = psutil.Process(os.getpid()) 

35 metrics: Dict[str, Union[float, dict]] = {} 

36 

37 with process.oneshot(): 

38 mem_info = process.memory_info() 

39 metrics["thread_count"] = float(process.num_threads()) 

40 metrics["mem"] = { 

41 "rss": float(mem_info.rss), 

42 } 

43 ctx_switches = process.num_ctx_switches() 

44 cpu_times = process.cpu_times() 

45 metrics["cpu"] = { 

46 "time": { 

47 "sys": cpu_times.system, 

48 "user": cpu_times.user, 

49 }, 

50 "ctx_switches": { 

51 "voluntary": float(ctx_switches.voluntary), 

52 "involuntary": float(ctx_switches.involuntary), 

53 }, 

54 "percent": process.cpu_percent(), 

55 } 

56 return metrics 

57 except Exception as e: 

58 # If psutil is installed but not compatible with the build, 

59 # we'll just cease further attempts to use it. 

60 _PSUTIL_AVAILABLE = False 

61 logger.debug("Failed to get system metrics: %s", e) 

62 return {} 

63 

64 

65@functools.lru_cache(maxsize=1) 

66def get_runtime_environment() -> dict: 

67 """Get information about the environment.""" 

68 # Lazy import to avoid circular imports 

69 from langsmith import __version__ 

70 

71 shas = get_release_shas() 

72 return { 

73 "sdk": "langsmith-py", 

74 "sdk_version": __version__, 

75 "library": "langsmith", 

76 "platform": platform.platform(), 

77 "runtime": "python", 

78 "py_implementation": platform.python_implementation(), 

79 "runtime_version": platform.python_version(), 

80 "langchain_version": get_langchain_environment(), 

81 "langchain_core_version": get_langchain_core_version(), 

82 **shas, 

83 } 

84 

85 

86@functools.lru_cache(maxsize=1) 

87def get_langchain_environment() -> Optional[str]: 

88 try: 

89 import langchain # type: ignore 

90 

91 return langchain.__version__ 

92 except: # noqa 

93 return None 

94 

95 

96@functools.lru_cache(maxsize=1) 

97def get_langchain_core_version() -> Optional[str]: 

98 try: 

99 import langchain_core # type: ignore 

100 

101 return langchain_core.__version__ 

102 except ImportError: 

103 return None 

104 

105 

106@functools.lru_cache(maxsize=1) 

107def get_docker_version() -> Optional[str]: 

108 import subprocess 

109 

110 try: 

111 docker_version = ( 

112 subprocess.check_output(["docker", "--version"]).decode("utf-8").strip() 

113 ) 

114 except FileNotFoundError: 

115 docker_version = "unknown" 

116 except: # noqa 

117 return None 

118 return docker_version 

119 

120 

121@functools.lru_cache(maxsize=1) 

122def get_docker_compose_version() -> Optional[str]: 

123 try: 

124 docker_compose_version = ( 

125 subprocess.check_output(["docker-compose", "--version"]) 

126 .decode("utf-8") 

127 .strip() 

128 ) 

129 except FileNotFoundError: 

130 docker_compose_version = "unknown" 

131 except: # noqa 

132 return None 

133 return docker_compose_version 

134 

135 

136@functools.lru_cache(maxsize=1) 

137def _get_compose_command() -> Optional[List[str]]: 

138 try: 

139 compose_command = get_docker_compose_command() 

140 except ValueError as e: 

141 compose_command = [f"NOT INSTALLED: {e}"] 

142 except: # noqa 

143 return None 

144 return compose_command 

145 

146 

147@functools.lru_cache(maxsize=1) 

148def get_docker_environment() -> dict: 

149 """Get information about the environment.""" 

150 compose_command = _get_compose_command() 

151 return { 

152 "docker_version": get_docker_version(), 

153 "docker_compose_command": ( 

154 " ".join(compose_command) if compose_command is not None else None 

155 ), 

156 "docker_compose_version": get_docker_compose_version(), 

157 } 

158 

159 

160def get_langchain_env_vars() -> dict: 

161 """Retrieve the langchain environment variables.""" 

162 env_vars = {k: v for k, v in os.environ.items() if k.startswith("LANGCHAIN_")} 

163 for key in list(env_vars): 

164 if "key" in key.lower(): 

165 v = env_vars[key] 

166 env_vars[key] = v[:2] + "*" * (len(v) - 4) + v[-2:] 

167 return env_vars 

168 

169 

170@functools.lru_cache(maxsize=1) 

171def get_langchain_env_var_metadata() -> dict: 

172 """Retrieve the langchain environment variables.""" 

173 excluded = { 

174 "LANGCHAIN_API_KEY", 

175 "LANGCHAIN_ENDPOINT", 

176 "LANGCHAIN_TRACING_V2", 

177 "LANGCHAIN_PROJECT", 

178 "LANGCHAIN_SESSION", 

179 "LANGSMITH_RUNS_ENDPOINTS", 

180 } 

181 langchain_metadata = { 

182 k: v 

183 for k, v in os.environ.items() 

184 if (k.startswith("LANGCHAIN_") or k.startswith("LANGSMITH_")) 

185 and k not in excluded 

186 and "key" not in k.lower() 

187 and "secret" not in k.lower() 

188 and "token" not in k.lower() 

189 } 

190 env_revision_id = langchain_metadata.pop("LANGCHAIN_REVISION_ID", None) 

191 if env_revision_id: 

192 langchain_metadata["revision_id"] = env_revision_id 

193 elif default_revision_id := _get_default_revision_id(): 

194 langchain_metadata["revision_id"] = default_revision_id 

195 

196 return langchain_metadata 

197 

198 

199@functools.lru_cache(maxsize=1) 

200def _get_default_revision_id() -> Optional[str]: 

201 """Get the default revision ID based on `git describe`.""" 

202 try: 

203 return exec_git(["describe", "--tags", "--always", "--dirty"]) 

204 except BaseException: 

205 return None 

206 

207 

208@functools.lru_cache(maxsize=1) 

209def get_release_shas() -> Dict[str, str]: 

210 common_release_envs = [ 

211 "VERCEL_GIT_COMMIT_SHA", 

212 "NEXT_PUBLIC_VERCEL_GIT_COMMIT_SHA", 

213 "COMMIT_REF", 

214 "RENDER_GIT_COMMIT", 

215 "CI_COMMIT_SHA", 

216 "CIRCLE_SHA1", 

217 "CF_PAGES_COMMIT_SHA", 

218 "REACT_APP_GIT_SHA", 

219 "SOURCE_VERSION", 

220 "GITHUB_SHA", 

221 "TRAVIS_COMMIT", 

222 "GIT_COMMIT", 

223 "BUILD_VCS_NUMBER", 

224 "bamboo_planRepository_revision", 

225 "Build.SourceVersion", 

226 "BITBUCKET_COMMIT", 

227 "DRONE_COMMIT_SHA", 

228 "SEMAPHORE_GIT_SHA", 

229 "BUILDKITE_COMMIT", 

230 ] 

231 shas = {} 

232 for env in common_release_envs: 

233 env_var = os.environ.get(env) 

234 if env_var is not None: 

235 shas[env] = env_var 

236 return shas