Gemma-3-12B-IT WebUI案例展示requests代码安全加固超时重试添加1. 引言从一次线上故障说起上周我们团队的一个内部工具突然“罢工”了。这个工具每天要调用几十个外部API接口平时运行得好好的但那天早上它卡在了一个第三方服务的请求上整整挂了半小时。更糟糕的是由于没有超时设置整个进程都被阻塞导致后续的所有任务都堆积如山。排查后发现问题出在一个简单的requests.get()调用上。对方服务器响应缓慢而我们的代码里既没有设置超时也没有任何重试机制。这让我意识到很多开发者包括曾经的我在使用requests库时都忽略了最基本的安全防护。今天我就以Gemma-3-12B-IT WebUI为例分享如何在实际项目中为requests代码添加安全加固和超时重试机制。这些技巧看似简单但在生产环境中却能避免很多“血泪教训”。2. 为什么你的requests代码需要加固2.1 常见的requests使用误区在开始具体实现之前我们先看看大多数人是怎么用requests的# 常见但危险的写法 import requests def get_user_data(user_id): response requests.get(fhttps://api.example.com/users/{user_id}) return response.json()这段代码至少有3个问题没有超时设置如果API服务器响应慢或挂起你的程序会一直等待没有错误处理网络异常、服务器错误等情况都会导致程序崩溃没有重试机制一次失败就彻底失败没有容错能力2.2 生产环境中的真实风险让我分享几个亲身经历的案例案例一服务雪崩一个微服务调用链中A服务调用B服务B服务调用C服务。C服务响应变慢从200ms变成10秒由于没有超时设置B服务的线程池很快被占满接着A服务也被拖垮。整个调用链像多米诺骨牌一样倒下。案例二资源泄漏长时间挂起的请求会占用连接资源。如果并发量稍大很快就会耗尽系统的文件描述符或端口导致“Too many open files”错误。案例三用户体验灾难前端调用一个API等了30秒还没响应用户以为系统挂了直接关掉页面。实际上只是某个第三方服务响应慢而已。3. 基础加固为requests添加超时和错误处理3.1 最简单的安全写法先从最基本的改进开始import requests import logging logger logging.getLogger(__name__) def safe_get_request(url, paramsNone, timeout10): 安全的GET请求包含超时和基础错误处理 Args: url: 请求地址 params: 查询参数 timeout: 超时时间秒默认10秒 Returns: response对象或None失败时 try: # 关键一定要设置timeout参数 response requests.get(url, paramsparams, timeouttimeout) # 检查HTTP状态码 response.raise_for_status() return response except requests.exceptions.Timeout: logger.error(f请求超时: {url}, 超时时间: {timeout}秒) return None except requests.exceptions.ConnectionError: logger.error(f连接错误: {url}, 可能是网络问题或服务器不可达) return None except requests.exceptions.HTTPError as e: logger.error(fHTTP错误: {url}, 状态码: {e.response.status_code}) return None except Exception as e: logger.error(f未知错误: {url}, 错误信息: {str(e)}) return None # 使用示例 if __name__ __main__: # 基础用法 result safe_get_request( https://api.github.com/users/octocat, timeout5 # 5秒超时 ) if result: print(f请求成功: {result.status_code}) print(f数据: {result.json()}) else: print(请求失败已记录日志)3.2 理解timeout参数的正确用法很多人以为timeout10就是10秒后超时其实不完全正确# timeout参数有两种形式 # 1. 单个数字连接超时和读取超时都用这个值 response requests.get(url, timeout10) # 连接读取总共10秒 # 2. 元组形式(连接超时, 读取超时) response requests.get(url, timeout(3, 10)) # 连接3秒读取10秒 # 实际生产建议使用元组形式 def get_with_detailed_timeout(url): 生产环境推荐分别设置连接超时和读取超时 连接超时connect timeout建立TCP连接的最大等待时间 读取超时read timeout从服务器接收数据的最大等待时间 try: # 连接超时设短些网络问题很快能发现 # 读取超时根据API响应时间调整 response requests.get(url, timeout(3, 30)) return response except requests.exceptions.ConnectTimeout: print(连接超时可能网络不通或服务器端口未开放) except requests.exceptions.ReadTimeout: print(读取超时服务器响应太慢)4. 进阶方案实现智能重试机制4.1 手动实现重试逻辑对于重要的请求一次失败就放弃是不够的。我们需要重试机制import requests import time from typing import Optional, Callable class RetryableRequest: 支持重试的请求类 def __init__(self, max_retries3, base_delay1): 初始化重试配置 Args: max_retries: 最大重试次数 base_delay: 基础延迟时间秒会指数递增 self.max_retries max_retries self.base_delay base_delay def get_with_retry(self, url, **kwargs) - Optional[requests.Response]: 带重试的GET请求 Args: url: 请求URL **kwargs: 传递给requests.get的参数 Returns: 响应对象或None last_exception None for attempt in range(self.max_retries 1): # 1 包括第一次尝试 try: # 设置超时如果kwargs中没有则使用默认值 timeout kwargs.pop(timeout, (3, 10)) response requests.get(url, timeouttimeout, **kwargs) response.raise_for_status() # 请求成功立即返回 return response except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: last_exception e # 如果是最后一次尝试不再等待 if attempt self.max_retries: break # 计算等待时间指数退避 delay self.base_delay * (2 ** attempt) print(f请求失败{delay}秒后重试... (尝试 {attempt 1}/{self.max_retries 1})) time.sleep(delay) except requests.exceptions.HTTPError as e: # HTTP错误如404、500通常不需要重试 print(fHTTP错误: {e.response.status_code}) raise e # 所有重试都失败 print(f请求失败已重试{self.max_retries}次) return None # 使用示例 if __name__ __main__: requester RetryableRequest(max_retries3, base_delay1) # 调用不稳定的API response requester.get_with_retry( https://api.example.com/unstable-endpoint, params{page: 1}, headers{User-Agent: MyApp/1.0} ) if response: print(最终请求成功!) data response.json()4.2 更智能的重试根据状态码决定是否重试不是所有错误都需要重试。比如404资源不存在重试多少次都没用def should_retry(response: requests.Response) - bool: 判断是否需要重试 Args: response: 响应对象 Returns: bool: 是否需要重试 status_code response.status_code # 这些状态码应该重试 retryable_codes { 408, # Request Timeout 429, # Too Many Requests 500, # Internal Server Error 502, # Bad Gateway 503, # Service Unavailable 504, # Gateway Timeout } return status_code in retryable_codes class SmartRetryRequest: 智能重试根据错误类型和状态码决定是否重试 def __init__(self, max_retries3): self.max_retries max_retries def request(self, method, url, **kwargs): 智能重试请求 Args: method: HTTP方法 url: 请求URL **kwargs: requests请求参数 for attempt in range(self.max_retries 1): try: response requests.request(method, url, **kwargs) # 检查是否需要重试 if response.status_code 500 or should_retry(response): if attempt self.max_retries: delay 1 * (2 ** attempt) # 指数退避 print(f服务器错误{delay}秒后重试...) time.sleep(delay) continue # 如果是客户端错误4xx通常不重试 if 400 response.status_code 500: response.raise_for_status() return response except requests.exceptions.Timeout: if attempt self.max_retries: delay 1 * (2 ** attempt) print(f超时{delay}秒后重试...) time.sleep(delay) else: raise return None5. 实战为Gemma-3-12B-IT WebUI添加安全请求层5.1 封装安全的API客户端现在我们把学到的知识应用到Gemma-3-12B-IT WebUI中。假设我们需要从WebUI获取数据import requests import time import logging from typing import Optional, Dict, Any from dataclasses import dataclass from enum import Enum # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class RequestMethod(Enum): HTTP方法枚举 GET GET POST POST PUT PUT DELETE DELETE dataclass class RequestConfig: 请求配置类 timeout: tuple (3, 30) # (连接超时, 读取超时) max_retries: int 3 retry_delay: float 1.0 # 基础延迟 retry_on_status: set None # 需要重试的状态码 def __post_init__(self): if self.retry_on_status is None: self.retry_on_status {500, 502, 503, 504, 408, 429} class SecureAPIClient: 安全的API客户端专为Gemma-3-12B-IT WebUI设计 def __init__(self, base_url: str, config: RequestConfig None): 初始化客户端 Args: base_url: API基础地址如 http://localhost:7860 config: 请求配置 self.base_url base_url.rstrip(/) self.config config or RequestConfig() self.session requests.Session() # 配置Session可以在这里添加默认headers等 self.session.headers.update({ User-Agent: Gemma-WebUI-Client/1.0, Accept: application/json, }) def _should_retry(self, response: Optional[requests.Response] None, exception: Optional[Exception] None) - bool: 判断是否需要重试 Args: response: 响应对象如果有 exception: 异常对象如果有 Returns: bool: 是否需要重试 # 如果是连接超时或读取超时需要重试 if isinstance(exception, (requests.exceptions.ConnectTimeout, requests.exceptions.ReadTimeout, requests.exceptions.ConnectionError)): return True # 如果是HTTP响应检查状态码 if response is not None: return response.status_code in self.config.retry_on_status return False def _make_request(self, method: RequestMethod, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: 执行HTTP请求内部方法 Args: method: HTTP方法 endpoint: API端点 **kwargs: 请求参数 Returns: 解析后的JSON数据或None url f{self.base_url}/{endpoint.lstrip(/)} # 确保使用配置的超时时间 if timeout not in kwargs: kwargs[timeout] self.config.timeout last_exception None for attempt in range(self.config.max_retries 1): try: logger.info(f尝试请求 {url} (尝试 {attempt 1}/{self.config.max_retries 1})) response self.session.request(method.value, url, **kwargs) # 检查是否需要重试 if self._should_retry(responseresponse): if attempt self.config.max_retries: delay self.config.retry_delay * (2 ** attempt) logger.warning(f需要重试等待 {delay:.1f}秒...) time.sleep(delay) continue # 如果不是需要重试的错误检查HTTP状态 response.raise_for_status() # 尝试解析JSON if response.content: return response.json() return {} except requests.exceptions.JSONDecodeError as e: logger.error(fJSON解析失败: {str(e)}) raise except requests.exceptions.HTTPError as e: logger.error(fHTTP错误 {e.response.status_code}: {url}) raise except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: last_exception e logger.warning(f网络错误: {type(e).__name__}) if attempt self.config.max_retries: delay self.config.retry_delay * (2 ** attempt) logger.info(f等待 {delay:.1f}秒后重试...) time.sleep(delay) else: logger.error(f重试{self.config.max_retries}次后仍失败) raise last_exception except Exception as e: logger.error(f未知错误: {type(e).__name__}: {str(e)}) raise return None # 便捷方法 def get(self, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: GET请求 return self._make_request(RequestMethod.GET, endpoint, **kwargs) def post(self, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: POST请求 return self._make_request(RequestMethod.POST, endpoint, **kwargs) def chat(self, message: str, **kwargs) - Optional[Dict[str, Any]]: 与Gemma-3-12B-IT对话 Args: message: 用户消息 **kwargs: 额外参数如temperature, max_tokens等 Returns: 对话响应 payload { message: message, **kwargs } # 这里需要根据实际的WebUI API调整端点 return self.post(/api/chat, jsonpayload) def get_status(self) - Optional[Dict[str, Any]]: 获取WebUI状态 return self.get(/api/status) # 使用示例 if __name__ __main__: # 创建客户端实例 client SecureAPIClient( base_urlhttp://localhost:7860, configRequestConfig( timeout(3, 30), # 连接3秒读取30秒 max_retries3, # 最多重试3次 retry_delay1.0, # 基础延迟1秒 ) ) try: # 1. 检查服务状态 status client.get_status() if status: print(f服务状态: {status}) # 2. 发送对话请求 response client.chat( message用Python写一个快速排序算法, temperature0.7, max_tokens500 ) if response: print(fAI回复: {response.get(content, )}) except Exception as e: print(f请求失败: {e})5.2 异步版本适合高性能场景如果你的应用需要高并发可以考虑异步版本import aiohttp import asyncio from typing import Optional, Dict, Any import logging logger logging.getLogger(__name__) class AsyncSecureAPIClient: 异步安全API客户端 def __init__(self, base_url: str, timeout: int 30, max_retries: int 3): self.base_url base_url self.timeout aiohttp.ClientTimeout(totaltimeout) self.max_retries max_retries self.session: Optional[aiohttp.ClientSession] None async def __aenter__(self): 异步上下文管理器入口 self.session aiohttp.ClientSession(timeoutself.timeout) return self async def __aexit__(self, exc_type, exc_val, exc_tb): 异步上下文管理器出口 if self.session: await self.session.close() async def request_with_retry(self, method: str, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: 带重试的异步请求 Args: method: HTTP方法 endpoint: API端点 **kwargs: 请求参数 Returns: JSON响应数据 if not self.session: raise RuntimeError(Session未初始化请使用async with) url f{self.base_url}/{endpoint.lstrip(/)} last_exception None for attempt in range(self.max_retries 1): try: async with self.session.request(method, url, **kwargs) as response: if response.status 500: # 服务器错误可能需要重试 if attempt self.max_retries: delay 1 * (2 ** attempt) logger.info(f服务器错误等待{delay}秒后重试...) await asyncio.sleep(delay) continue response.raise_for_status() return await response.json() except (aiohttp.ClientConnectorError, aiohttp.ServerTimeoutError, asyncio.TimeoutError) as e: last_exception e if attempt self.max_retries: delay 1 * (2 ** attempt) logger.info(f连接错误等待{delay}秒后重试...) await asyncio.sleep(delay) else: logger.error(f重试{self.max_retries}次后仍失败) raise last_exception return None async def chat(self, message: str, **kwargs) - Optional[Dict[str, Any]]: 异步对话 payload {message: message, **kwargs} return await self.request_with_retry(POST, /api/chat, jsonpayload) # 使用示例 async def main(): async with AsyncSecureAPIClient(http://localhost:7860) as client: try: response await client.chat(你好Gemma!) if response: print(f回复: {response}) except Exception as e: print(f请求失败: {e}) # 运行 if __name__ __main__: asyncio.run(main())6. 监控与告警知道什么时候出了问题6.1 添加请求监控代码加固了还不够我们还需要知道它运行得怎么样import time from datetime import datetime from typing import Dict, Any import statistics class RequestMonitor: 请求监控器 def __init__(self): self.request_history [] self.error_count 0 self.success_count 0 def record_request(self, url: str, success: bool, duration: float, status_code: int None): 记录请求信息 Args: url: 请求URL success: 是否成功 duration: 请求耗时秒 status_code: HTTP状态码 record { timestamp: datetime.now(), url: url, success: success, duration: duration, status_code: status_code } self.request_history.append(record) if success: self.success_count 1 else: self.error_count 1 # 只保留最近1000条记录 if len(self.request_history) 1000: self.request_history.pop(0) def get_stats(self) - Dict[str, Any]: 获取统计信息 Returns: 统计字典 if not self.request_history: return {} successful_requests [r for r in self.request_history if r[success]] failed_requests [r for r in self.request_history if not r[success]] durations [r[duration] for r in self.request_history] return { total_requests: len(self.request_history), success_rate: self.success_count / len(self.request_history) * 100, avg_duration: statistics.mean(durations) if durations else 0, p95_duration: statistics.quantiles(durations, n20)[18] if len(durations) 20 else 0, recent_errors: len(failed_requests[-10:]), # 最近10次中的错误数 most_common_error: self._get_most_common_error() } def _get_most_common_error(self) - str: 获取最常见的错误类型 error_codes {} for record in self.request_history: if not record[success] and record[status_code]: error_codes[record[status_code]] error_codes.get(record[status_code], 0) 1 if error_codes: most_common max(error_codes.items(), keylambda x: x[1]) return fHTTP {most_common[0]} ({most_common[1]}次) return 无错误 def check_health(self) - Dict[str, Any]: 检查服务健康状态 Returns: 健康状态信息 stats self.get_stats() # 简单的健康检查规则 health_status healthy issues [] if stats.get(success_rate, 100) 95: health_status degraded issues.append(f成功率低于95%: {stats[success_rate]:.1f}%) if stats.get(recent_errors, 0) 3: health_status degraded issues.append(f最近错误较多: {stats[recent_errors]}次) if stats.get(avg_duration, 0) 5: # 平均响应超过5秒 health_status degraded issues.append(f平均响应时间过长: {stats[avg_duration]:.1f}秒) return { status: health_status, issues: issues, stats: stats } # 集成到SecureAPIClient中 class MonitoredAPIClient(SecureAPIClient): 带监控的API客户端 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.monitor RequestMonitor() def _make_request(self, method: RequestMethod, endpoint: str, **kwargs): 重写请求方法添加监控 url f{self.base_url}/{endpoint.lstrip(/)} start_time time.time() try: response super()._make_request(method, endpoint, **kwargs) duration time.time() - start_time # 记录成功请求 self.monitor.record_request( urlurl, successTrue, durationduration, status_code200 ) return response except requests.exceptions.HTTPError as e: duration time.time() - start_time status_code e.response.status_code if e.response else None # 记录失败请求 self.monitor.record_request( urlurl, successFalse, durationduration, status_codestatus_code ) raise except Exception as e: duration time.time() - start_time self.monitor.record_request( urlurl, successFalse, durationduration, status_codeNone ) raise def get_health_report(self): 获取健康报告 return self.monitor.check_health()6.2 使用示例监控Gemma WebUI的健康状况def monitor_gemma_webui(): 监控Gemma-3-12B-IT WebUI的健康状况 client MonitoredAPIClient( base_urlhttp://localhost:7860, configRequestConfig(max_retries2) ) print(开始监控Gemma WebUI...) print( * 50) # 模拟多次请求 test_messages [ 你好请介绍一下你自己, 用Python写一个Hello World程序, 什么是机器学习, 今天的天气怎么样, 讲一个笑话 ] for i, message in enumerate(test_messages, 1): print(f\n请求 {i}: {message}) try: response client.chat(message, max_tokens100) if response: print(f✓ 成功: {response.get(content, )[:50]}...) else: print(✗ 无响应) except Exception as e: print(f✗ 失败: {type(e).__name__}) # 每3次请求显示一次健康状态 if i % 3 0: health client.get_health_report() print(f\n当前健康状态: {health[status]}) if health[issues]: print(发现的问题:) for issue in health[issues]: print(f - {issue}) print(f成功率: {health[stats].get(success_rate, 0):.1f}%) print(f平均响应时间: {health[stats].get(avg_duration, 0):.2f}秒) print( * 50) # 最终报告 print(\n * 50) print(监控结束最终报告:) final_health client.get_health_report() for key, value in final_health[stats].items(): print(f{key}: {value}) if final_health[status] healthy: print(✅ 服务状态健康) else: print(⚠️ 服务状态异常请检查) for issue in final_health[issues]: print(f - {issue}) if __name__ __main__: monitor_gemma_webui()7. 总结与最佳实践7.1 核心要点回顾通过今天的分享我们为Gemma-3-12B-IT WebUI的requests代码添加了多层安全防护基础防护为所有请求添加超时设置避免程序无限等待错误处理捕获并妥善处理各种网络异常和HTTP错误智能重试对可重试的错误如超时、5xx错误自动重试使用指数退避策略状态码判断根据HTTP状态码决定是否重试5xx重试4xx通常不重试监控告警记录请求指标及时发现服务异常7.2 生产环境最佳实践根据我的经验以下是在生产环境中使用requests的最佳实践1. 超时设置要合理# 推荐分别设置连接超时和读取超时 timeout(3, 30) # 连接3秒读取30秒 # 根据API特性调整 # - 内部APItimeout(1, 5) # 响应快超时短 # - 外部APItimeout(5, 30) # 响应慢超时长 # - 文件上传timeout(10, 60) # 大文件超时长2. 重试策略要智能# 不要所有错误都重试 retryable_errors { requests.exceptions.Timeout, requests.exceptions.ConnectionError, } # 根据状态码决定 retryable_status_codes {408, 429, 500, 502, 503, 504} # 使用指数退避 delay base_delay * (2 ** attempt) # 1, 2, 4, 8秒...3. 连接池要管理# 使用Session重用连接 session requests.Session() # 配置连接池 adapter requests.adapters.HTTPAdapter( pool_connections10, # 连接池大小 pool_maxsize10, # 最大连接数 max_retries3 # 重试次数 ) session.mount(http://, adapter) session.mount(https://, adapter)4. 监控指标要收集成功率目标99.9%平均响应时间目标1秒P95/P99响应时间错误类型分布重试次数统计7.3 针对Gemma-3-12B-IT WebUI的特殊建议对于大语言模型WebUI还有一些特殊考虑长文本处理如果生成长文本适当增加读取超时流式响应如果支持流式输出考虑使用SSE或WebSocket并发控制LLM推理资源密集控制并发请求数熔断机制连续失败时暂时停止请求避免雪崩# 简单的熔断器实现 class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout60): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failure_count 0 self.last_failure_time None self.state closed # closed, open, half-open def call(self, func, *args, **kwargs): if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half-open else: raise Exception(Circuit breaker is OPEN) try: result func(*args, **kwargs) if self.state half-open: self.state closed self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open raise e7.4 最后的建议不要信任任何外部服务即使是内部API也可能出问题防御性编程假设一切都会失败提前做好准备监控是关键没有监控的代码就像闭着眼睛开车渐进式改进先从添加timeout开始逐步完善重试、熔断等机制测试各种故障场景模拟网络超时、服务不可用、响应缓慢等情况记住代码的健壮性不是一次性能完成的而是在不断遇到问题和解决问题的过程中逐步完善的。今天分享的这些技巧都是我们从实际故障中总结出来的经验。希望它们能帮助你写出更稳定、更可靠的requests代码。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Gemma-3-12B-IT WebUI案例展示：requests代码安全加固+超时重试添加
Gemma-3-12B-IT WebUI案例展示requests代码安全加固超时重试添加1. 引言从一次线上故障说起上周我们团队的一个内部工具突然“罢工”了。这个工具每天要调用几十个外部API接口平时运行得好好的但那天早上它卡在了一个第三方服务的请求上整整挂了半小时。更糟糕的是由于没有超时设置整个进程都被阻塞导致后续的所有任务都堆积如山。排查后发现问题出在一个简单的requests.get()调用上。对方服务器响应缓慢而我们的代码里既没有设置超时也没有任何重试机制。这让我意识到很多开发者包括曾经的我在使用requests库时都忽略了最基本的安全防护。今天我就以Gemma-3-12B-IT WebUI为例分享如何在实际项目中为requests代码添加安全加固和超时重试机制。这些技巧看似简单但在生产环境中却能避免很多“血泪教训”。2. 为什么你的requests代码需要加固2.1 常见的requests使用误区在开始具体实现之前我们先看看大多数人是怎么用requests的# 常见但危险的写法 import requests def get_user_data(user_id): response requests.get(fhttps://api.example.com/users/{user_id}) return response.json()这段代码至少有3个问题没有超时设置如果API服务器响应慢或挂起你的程序会一直等待没有错误处理网络异常、服务器错误等情况都会导致程序崩溃没有重试机制一次失败就彻底失败没有容错能力2.2 生产环境中的真实风险让我分享几个亲身经历的案例案例一服务雪崩一个微服务调用链中A服务调用B服务B服务调用C服务。C服务响应变慢从200ms变成10秒由于没有超时设置B服务的线程池很快被占满接着A服务也被拖垮。整个调用链像多米诺骨牌一样倒下。案例二资源泄漏长时间挂起的请求会占用连接资源。如果并发量稍大很快就会耗尽系统的文件描述符或端口导致“Too many open files”错误。案例三用户体验灾难前端调用一个API等了30秒还没响应用户以为系统挂了直接关掉页面。实际上只是某个第三方服务响应慢而已。3. 基础加固为requests添加超时和错误处理3.1 最简单的安全写法先从最基本的改进开始import requests import logging logger logging.getLogger(__name__) def safe_get_request(url, paramsNone, timeout10): 安全的GET请求包含超时和基础错误处理 Args: url: 请求地址 params: 查询参数 timeout: 超时时间秒默认10秒 Returns: response对象或None失败时 try: # 关键一定要设置timeout参数 response requests.get(url, paramsparams, timeouttimeout) # 检查HTTP状态码 response.raise_for_status() return response except requests.exceptions.Timeout: logger.error(f请求超时: {url}, 超时时间: {timeout}秒) return None except requests.exceptions.ConnectionError: logger.error(f连接错误: {url}, 可能是网络问题或服务器不可达) return None except requests.exceptions.HTTPError as e: logger.error(fHTTP错误: {url}, 状态码: {e.response.status_code}) return None except Exception as e: logger.error(f未知错误: {url}, 错误信息: {str(e)}) return None # 使用示例 if __name__ __main__: # 基础用法 result safe_get_request( https://api.github.com/users/octocat, timeout5 # 5秒超时 ) if result: print(f请求成功: {result.status_code}) print(f数据: {result.json()}) else: print(请求失败已记录日志)3.2 理解timeout参数的正确用法很多人以为timeout10就是10秒后超时其实不完全正确# timeout参数有两种形式 # 1. 单个数字连接超时和读取超时都用这个值 response requests.get(url, timeout10) # 连接读取总共10秒 # 2. 元组形式(连接超时, 读取超时) response requests.get(url, timeout(3, 10)) # 连接3秒读取10秒 # 实际生产建议使用元组形式 def get_with_detailed_timeout(url): 生产环境推荐分别设置连接超时和读取超时 连接超时connect timeout建立TCP连接的最大等待时间 读取超时read timeout从服务器接收数据的最大等待时间 try: # 连接超时设短些网络问题很快能发现 # 读取超时根据API响应时间调整 response requests.get(url, timeout(3, 30)) return response except requests.exceptions.ConnectTimeout: print(连接超时可能网络不通或服务器端口未开放) except requests.exceptions.ReadTimeout: print(读取超时服务器响应太慢)4. 进阶方案实现智能重试机制4.1 手动实现重试逻辑对于重要的请求一次失败就放弃是不够的。我们需要重试机制import requests import time from typing import Optional, Callable class RetryableRequest: 支持重试的请求类 def __init__(self, max_retries3, base_delay1): 初始化重试配置 Args: max_retries: 最大重试次数 base_delay: 基础延迟时间秒会指数递增 self.max_retries max_retries self.base_delay base_delay def get_with_retry(self, url, **kwargs) - Optional[requests.Response]: 带重试的GET请求 Args: url: 请求URL **kwargs: 传递给requests.get的参数 Returns: 响应对象或None last_exception None for attempt in range(self.max_retries 1): # 1 包括第一次尝试 try: # 设置超时如果kwargs中没有则使用默认值 timeout kwargs.pop(timeout, (3, 10)) response requests.get(url, timeouttimeout, **kwargs) response.raise_for_status() # 请求成功立即返回 return response except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: last_exception e # 如果是最后一次尝试不再等待 if attempt self.max_retries: break # 计算等待时间指数退避 delay self.base_delay * (2 ** attempt) print(f请求失败{delay}秒后重试... (尝试 {attempt 1}/{self.max_retries 1})) time.sleep(delay) except requests.exceptions.HTTPError as e: # HTTP错误如404、500通常不需要重试 print(fHTTP错误: {e.response.status_code}) raise e # 所有重试都失败 print(f请求失败已重试{self.max_retries}次) return None # 使用示例 if __name__ __main__: requester RetryableRequest(max_retries3, base_delay1) # 调用不稳定的API response requester.get_with_retry( https://api.example.com/unstable-endpoint, params{page: 1}, headers{User-Agent: MyApp/1.0} ) if response: print(最终请求成功!) data response.json()4.2 更智能的重试根据状态码决定是否重试不是所有错误都需要重试。比如404资源不存在重试多少次都没用def should_retry(response: requests.Response) - bool: 判断是否需要重试 Args: response: 响应对象 Returns: bool: 是否需要重试 status_code response.status_code # 这些状态码应该重试 retryable_codes { 408, # Request Timeout 429, # Too Many Requests 500, # Internal Server Error 502, # Bad Gateway 503, # Service Unavailable 504, # Gateway Timeout } return status_code in retryable_codes class SmartRetryRequest: 智能重试根据错误类型和状态码决定是否重试 def __init__(self, max_retries3): self.max_retries max_retries def request(self, method, url, **kwargs): 智能重试请求 Args: method: HTTP方法 url: 请求URL **kwargs: requests请求参数 for attempt in range(self.max_retries 1): try: response requests.request(method, url, **kwargs) # 检查是否需要重试 if response.status_code 500 or should_retry(response): if attempt self.max_retries: delay 1 * (2 ** attempt) # 指数退避 print(f服务器错误{delay}秒后重试...) time.sleep(delay) continue # 如果是客户端错误4xx通常不重试 if 400 response.status_code 500: response.raise_for_status() return response except requests.exceptions.Timeout: if attempt self.max_retries: delay 1 * (2 ** attempt) print(f超时{delay}秒后重试...) time.sleep(delay) else: raise return None5. 实战为Gemma-3-12B-IT WebUI添加安全请求层5.1 封装安全的API客户端现在我们把学到的知识应用到Gemma-3-12B-IT WebUI中。假设我们需要从WebUI获取数据import requests import time import logging from typing import Optional, Dict, Any from dataclasses import dataclass from enum import Enum # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class RequestMethod(Enum): HTTP方法枚举 GET GET POST POST PUT PUT DELETE DELETE dataclass class RequestConfig: 请求配置类 timeout: tuple (3, 30) # (连接超时, 读取超时) max_retries: int 3 retry_delay: float 1.0 # 基础延迟 retry_on_status: set None # 需要重试的状态码 def __post_init__(self): if self.retry_on_status is None: self.retry_on_status {500, 502, 503, 504, 408, 429} class SecureAPIClient: 安全的API客户端专为Gemma-3-12B-IT WebUI设计 def __init__(self, base_url: str, config: RequestConfig None): 初始化客户端 Args: base_url: API基础地址如 http://localhost:7860 config: 请求配置 self.base_url base_url.rstrip(/) self.config config or RequestConfig() self.session requests.Session() # 配置Session可以在这里添加默认headers等 self.session.headers.update({ User-Agent: Gemma-WebUI-Client/1.0, Accept: application/json, }) def _should_retry(self, response: Optional[requests.Response] None, exception: Optional[Exception] None) - bool: 判断是否需要重试 Args: response: 响应对象如果有 exception: 异常对象如果有 Returns: bool: 是否需要重试 # 如果是连接超时或读取超时需要重试 if isinstance(exception, (requests.exceptions.ConnectTimeout, requests.exceptions.ReadTimeout, requests.exceptions.ConnectionError)): return True # 如果是HTTP响应检查状态码 if response is not None: return response.status_code in self.config.retry_on_status return False def _make_request(self, method: RequestMethod, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: 执行HTTP请求内部方法 Args: method: HTTP方法 endpoint: API端点 **kwargs: 请求参数 Returns: 解析后的JSON数据或None url f{self.base_url}/{endpoint.lstrip(/)} # 确保使用配置的超时时间 if timeout not in kwargs: kwargs[timeout] self.config.timeout last_exception None for attempt in range(self.config.max_retries 1): try: logger.info(f尝试请求 {url} (尝试 {attempt 1}/{self.config.max_retries 1})) response self.session.request(method.value, url, **kwargs) # 检查是否需要重试 if self._should_retry(responseresponse): if attempt self.config.max_retries: delay self.config.retry_delay * (2 ** attempt) logger.warning(f需要重试等待 {delay:.1f}秒...) time.sleep(delay) continue # 如果不是需要重试的错误检查HTTP状态 response.raise_for_status() # 尝试解析JSON if response.content: return response.json() return {} except requests.exceptions.JSONDecodeError as e: logger.error(fJSON解析失败: {str(e)}) raise except requests.exceptions.HTTPError as e: logger.error(fHTTP错误 {e.response.status_code}: {url}) raise except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e: last_exception e logger.warning(f网络错误: {type(e).__name__}) if attempt self.config.max_retries: delay self.config.retry_delay * (2 ** attempt) logger.info(f等待 {delay:.1f}秒后重试...) time.sleep(delay) else: logger.error(f重试{self.config.max_retries}次后仍失败) raise last_exception except Exception as e: logger.error(f未知错误: {type(e).__name__}: {str(e)}) raise return None # 便捷方法 def get(self, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: GET请求 return self._make_request(RequestMethod.GET, endpoint, **kwargs) def post(self, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: POST请求 return self._make_request(RequestMethod.POST, endpoint, **kwargs) def chat(self, message: str, **kwargs) - Optional[Dict[str, Any]]: 与Gemma-3-12B-IT对话 Args: message: 用户消息 **kwargs: 额外参数如temperature, max_tokens等 Returns: 对话响应 payload { message: message, **kwargs } # 这里需要根据实际的WebUI API调整端点 return self.post(/api/chat, jsonpayload) def get_status(self) - Optional[Dict[str, Any]]: 获取WebUI状态 return self.get(/api/status) # 使用示例 if __name__ __main__: # 创建客户端实例 client SecureAPIClient( base_urlhttp://localhost:7860, configRequestConfig( timeout(3, 30), # 连接3秒读取30秒 max_retries3, # 最多重试3次 retry_delay1.0, # 基础延迟1秒 ) ) try: # 1. 检查服务状态 status client.get_status() if status: print(f服务状态: {status}) # 2. 发送对话请求 response client.chat( message用Python写一个快速排序算法, temperature0.7, max_tokens500 ) if response: print(fAI回复: {response.get(content, )}) except Exception as e: print(f请求失败: {e})5.2 异步版本适合高性能场景如果你的应用需要高并发可以考虑异步版本import aiohttp import asyncio from typing import Optional, Dict, Any import logging logger logging.getLogger(__name__) class AsyncSecureAPIClient: 异步安全API客户端 def __init__(self, base_url: str, timeout: int 30, max_retries: int 3): self.base_url base_url self.timeout aiohttp.ClientTimeout(totaltimeout) self.max_retries max_retries self.session: Optional[aiohttp.ClientSession] None async def __aenter__(self): 异步上下文管理器入口 self.session aiohttp.ClientSession(timeoutself.timeout) return self async def __aexit__(self, exc_type, exc_val, exc_tb): 异步上下文管理器出口 if self.session: await self.session.close() async def request_with_retry(self, method: str, endpoint: str, **kwargs) - Optional[Dict[str, Any]]: 带重试的异步请求 Args: method: HTTP方法 endpoint: API端点 **kwargs: 请求参数 Returns: JSON响应数据 if not self.session: raise RuntimeError(Session未初始化请使用async with) url f{self.base_url}/{endpoint.lstrip(/)} last_exception None for attempt in range(self.max_retries 1): try: async with self.session.request(method, url, **kwargs) as response: if response.status 500: # 服务器错误可能需要重试 if attempt self.max_retries: delay 1 * (2 ** attempt) logger.info(f服务器错误等待{delay}秒后重试...) await asyncio.sleep(delay) continue response.raise_for_status() return await response.json() except (aiohttp.ClientConnectorError, aiohttp.ServerTimeoutError, asyncio.TimeoutError) as e: last_exception e if attempt self.max_retries: delay 1 * (2 ** attempt) logger.info(f连接错误等待{delay}秒后重试...) await asyncio.sleep(delay) else: logger.error(f重试{self.max_retries}次后仍失败) raise last_exception return None async def chat(self, message: str, **kwargs) - Optional[Dict[str, Any]]: 异步对话 payload {message: message, **kwargs} return await self.request_with_retry(POST, /api/chat, jsonpayload) # 使用示例 async def main(): async with AsyncSecureAPIClient(http://localhost:7860) as client: try: response await client.chat(你好Gemma!) if response: print(f回复: {response}) except Exception as e: print(f请求失败: {e}) # 运行 if __name__ __main__: asyncio.run(main())6. 监控与告警知道什么时候出了问题6.1 添加请求监控代码加固了还不够我们还需要知道它运行得怎么样import time from datetime import datetime from typing import Dict, Any import statistics class RequestMonitor: 请求监控器 def __init__(self): self.request_history [] self.error_count 0 self.success_count 0 def record_request(self, url: str, success: bool, duration: float, status_code: int None): 记录请求信息 Args: url: 请求URL success: 是否成功 duration: 请求耗时秒 status_code: HTTP状态码 record { timestamp: datetime.now(), url: url, success: success, duration: duration, status_code: status_code } self.request_history.append(record) if success: self.success_count 1 else: self.error_count 1 # 只保留最近1000条记录 if len(self.request_history) 1000: self.request_history.pop(0) def get_stats(self) - Dict[str, Any]: 获取统计信息 Returns: 统计字典 if not self.request_history: return {} successful_requests [r for r in self.request_history if r[success]] failed_requests [r for r in self.request_history if not r[success]] durations [r[duration] for r in self.request_history] return { total_requests: len(self.request_history), success_rate: self.success_count / len(self.request_history) * 100, avg_duration: statistics.mean(durations) if durations else 0, p95_duration: statistics.quantiles(durations, n20)[18] if len(durations) 20 else 0, recent_errors: len(failed_requests[-10:]), # 最近10次中的错误数 most_common_error: self._get_most_common_error() } def _get_most_common_error(self) - str: 获取最常见的错误类型 error_codes {} for record in self.request_history: if not record[success] and record[status_code]: error_codes[record[status_code]] error_codes.get(record[status_code], 0) 1 if error_codes: most_common max(error_codes.items(), keylambda x: x[1]) return fHTTP {most_common[0]} ({most_common[1]}次) return 无错误 def check_health(self) - Dict[str, Any]: 检查服务健康状态 Returns: 健康状态信息 stats self.get_stats() # 简单的健康检查规则 health_status healthy issues [] if stats.get(success_rate, 100) 95: health_status degraded issues.append(f成功率低于95%: {stats[success_rate]:.1f}%) if stats.get(recent_errors, 0) 3: health_status degraded issues.append(f最近错误较多: {stats[recent_errors]}次) if stats.get(avg_duration, 0) 5: # 平均响应超过5秒 health_status degraded issues.append(f平均响应时间过长: {stats[avg_duration]:.1f}秒) return { status: health_status, issues: issues, stats: stats } # 集成到SecureAPIClient中 class MonitoredAPIClient(SecureAPIClient): 带监控的API客户端 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.monitor RequestMonitor() def _make_request(self, method: RequestMethod, endpoint: str, **kwargs): 重写请求方法添加监控 url f{self.base_url}/{endpoint.lstrip(/)} start_time time.time() try: response super()._make_request(method, endpoint, **kwargs) duration time.time() - start_time # 记录成功请求 self.monitor.record_request( urlurl, successTrue, durationduration, status_code200 ) return response except requests.exceptions.HTTPError as e: duration time.time() - start_time status_code e.response.status_code if e.response else None # 记录失败请求 self.monitor.record_request( urlurl, successFalse, durationduration, status_codestatus_code ) raise except Exception as e: duration time.time() - start_time self.monitor.record_request( urlurl, successFalse, durationduration, status_codeNone ) raise def get_health_report(self): 获取健康报告 return self.monitor.check_health()6.2 使用示例监控Gemma WebUI的健康状况def monitor_gemma_webui(): 监控Gemma-3-12B-IT WebUI的健康状况 client MonitoredAPIClient( base_urlhttp://localhost:7860, configRequestConfig(max_retries2) ) print(开始监控Gemma WebUI...) print( * 50) # 模拟多次请求 test_messages [ 你好请介绍一下你自己, 用Python写一个Hello World程序, 什么是机器学习, 今天的天气怎么样, 讲一个笑话 ] for i, message in enumerate(test_messages, 1): print(f\n请求 {i}: {message}) try: response client.chat(message, max_tokens100) if response: print(f✓ 成功: {response.get(content, )[:50]}...) else: print(✗ 无响应) except Exception as e: print(f✗ 失败: {type(e).__name__}) # 每3次请求显示一次健康状态 if i % 3 0: health client.get_health_report() print(f\n当前健康状态: {health[status]}) if health[issues]: print(发现的问题:) for issue in health[issues]: print(f - {issue}) print(f成功率: {health[stats].get(success_rate, 0):.1f}%) print(f平均响应时间: {health[stats].get(avg_duration, 0):.2f}秒) print( * 50) # 最终报告 print(\n * 50) print(监控结束最终报告:) final_health client.get_health_report() for key, value in final_health[stats].items(): print(f{key}: {value}) if final_health[status] healthy: print(✅ 服务状态健康) else: print(⚠️ 服务状态异常请检查) for issue in final_health[issues]: print(f - {issue}) if __name__ __main__: monitor_gemma_webui()7. 总结与最佳实践7.1 核心要点回顾通过今天的分享我们为Gemma-3-12B-IT WebUI的requests代码添加了多层安全防护基础防护为所有请求添加超时设置避免程序无限等待错误处理捕获并妥善处理各种网络异常和HTTP错误智能重试对可重试的错误如超时、5xx错误自动重试使用指数退避策略状态码判断根据HTTP状态码决定是否重试5xx重试4xx通常不重试监控告警记录请求指标及时发现服务异常7.2 生产环境最佳实践根据我的经验以下是在生产环境中使用requests的最佳实践1. 超时设置要合理# 推荐分别设置连接超时和读取超时 timeout(3, 30) # 连接3秒读取30秒 # 根据API特性调整 # - 内部APItimeout(1, 5) # 响应快超时短 # - 外部APItimeout(5, 30) # 响应慢超时长 # - 文件上传timeout(10, 60) # 大文件超时长2. 重试策略要智能# 不要所有错误都重试 retryable_errors { requests.exceptions.Timeout, requests.exceptions.ConnectionError, } # 根据状态码决定 retryable_status_codes {408, 429, 500, 502, 503, 504} # 使用指数退避 delay base_delay * (2 ** attempt) # 1, 2, 4, 8秒...3. 连接池要管理# 使用Session重用连接 session requests.Session() # 配置连接池 adapter requests.adapters.HTTPAdapter( pool_connections10, # 连接池大小 pool_maxsize10, # 最大连接数 max_retries3 # 重试次数 ) session.mount(http://, adapter) session.mount(https://, adapter)4. 监控指标要收集成功率目标99.9%平均响应时间目标1秒P95/P99响应时间错误类型分布重试次数统计7.3 针对Gemma-3-12B-IT WebUI的特殊建议对于大语言模型WebUI还有一些特殊考虑长文本处理如果生成长文本适当增加读取超时流式响应如果支持流式输出考虑使用SSE或WebSocket并发控制LLM推理资源密集控制并发请求数熔断机制连续失败时暂时停止请求避免雪崩# 简单的熔断器实现 class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout60): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failure_count 0 self.last_failure_time None self.state closed # closed, open, half-open def call(self, func, *args, **kwargs): if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half-open else: raise Exception(Circuit breaker is OPEN) try: result func(*args, **kwargs) if self.state half-open: self.state closed self.failure_count 0 return result except Exception as e: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state open raise e7.4 最后的建议不要信任任何外部服务即使是内部API也可能出问题防御性编程假设一切都会失败提前做好准备监控是关键没有监控的代码就像闭着眼睛开车渐进式改进先从添加timeout开始逐步完善重试、熔断等机制测试各种故障场景模拟网络超时、服务不可用、响应缓慢等情况记住代码的健壮性不是一次性能完成的而是在不断遇到问题和解决问题的过程中逐步完善的。今天分享的这些技巧都是我们从实际故障中总结出来的经验。希望它们能帮助你写出更稳定、更可靠的requests代码。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
