# AIGlasses_for_navigation代码实例：Python SDK封装API调用与错误处理范例 ## 1. 引言 如果你正在开发一个像AIGlasses_for_navigation这样的智能可穿戴设备项目，那么你肯定遇到过这样的问题：如何优雅地调用各种AI服务API？如何处理网络波动、服务限流、参数错误等异常情况？如何让代码既健壮又易于维护？ 今天，我就来分享一个实战经验——如何为AIGlasses_for_navigation项目封装一个Python SDK，专门处理阿里云DashScope等AI服务的API调用。这个SDK不仅简化了调用流程，更重要的是，它内置了一套完整的错误处理机制，让你的应用在面对各种异常时都能从容应对。 我会用一个完整的代码实例来展示，从基础封装到高级错误处理，再到实际应用场景。无论你是项目的新手开发者，还是想优化现有代码结构，这篇文章都能给你带来实用的参考价值。 ## 2. 为什么需要封装SDK？ 在AIGlasses_for_navigation项目中，我们需要频繁调用多种AI服务： - 语音识别（ASR）：把用户说的话转成文字 - 语音合成（TTS）：把AI回复转成语音 - 多模态对话：理解图片+语音的复杂指令 - 视觉识别：检测盲道、红绿灯、物品等 如果每次调用都写一遍HTTP请求、错误处理、结果解析，代码会变得冗长且难以维护。更糟糕的是，不同的API可能有不同的错误码、重试策略和超时设置。 ### 2.1 直接调用的痛点 看看下面这个直接调用DashScope API的例子： ```python import requests import json def call_dashscope_directly(api_key, prompt, image_url=None): """直接调用DashScope API（问题示例）""" url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } data = { "model": "qwen-max", "input": { "messages": [ { "role": "user", "content": [ {"text": prompt} ] } ] } } if image_url: data["input"]["messages"][0]["content"].append({ "image": image_url }) try: response = requests.post(url, headers=headers, json=data, timeout=10) response.raise_for_status() result = response.json() if result.get("code") != 200: # 这里开始各种错误判断... if result.get("code") == 400: return {"error": "参数错误"} elif result.get("code") == 401: return {"error": "API Key无效"} elif result.get("code") == 429: return {"error": "请求太频繁"} # ... 更多错误码判断 else: return result.get("output", {}).get("text", "") except requests.exceptions.Timeout: return {"error": "请求超时"} except requests.exceptions.ConnectionError: return {"error": "网络连接失败"} except requests.exceptions.RequestException as e: return {"error": f"请求异常: {str(e)}"} except json.JSONDecodeError: return {"error": "响应解析失败"} except Exception as e: return {"error": f"未知错误: {str(e)}"} ``` 这段代码有什么问题？ 1. **重复代码**：每个API调用都要写一遍类似的try-catch 2. **错误处理分散**：错误码判断散落在各处 3. **难以扩展**：新增API需要复制粘贴大量代码 4. **维护困难**：修改错误处理逻辑需要改很多地方 ### 2.2 SDK封装的好处 封装成SDK后，调用变得非常简单： ```python # 使用封装后的SDK from aiglasses_sdk import AIGlassesClient client = AIGlassesClient(api_key="sk-your-key") # 调用语音识别 text = client.asr.transcribe(audio_file="voice.wav") # 调用多模态对话 response = client.multimodal.chat( prompt="帮我看看这是什么", image_url="https://example.com/image.jpg" ) # 调用语音合成 audio_data = client.tts.synthesize(text="前方有障碍物，请小心") ``` 所有复杂的错误处理、重试逻辑、参数验证都在SDK内部完成，业务代码变得清晰简洁。 ## 3. SDK核心架构设计 一个好的SDK应该具备以下特点： - **易用性**：API设计直观，学习成本低 - **健壮性**：完善的错误处理和重试机制 - **可扩展性**：容易添加新的API接口 - **可维护性**：代码结构清晰，便于调试 ### 3.1 整体架构 ``` AIGlassesSDK/ ├── client.py # 主客户端类 ├── exceptions.py # 自定义异常 ├── retry.py # 重试策略 ├── validators.py # 参数验证 ├── api/ │ ├── base.py # API基类 │ ├── asr.py # 语音识别API │ ├── tts.py # 语音合成API │ ├── multimodal.py # 多模态API │ └── vision.py # 视觉识别API └── utils/ ├── logger.py # 日志工具 ├── cache.py # 缓存工具 └── helpers.py # 辅助函数 ``` ### 3.2 核心类设计 让我们从最基础的自定义异常开始： ```python # exceptions.py class AIGlassesError(Exception): """SDK基础异常类""" pass class APIError(AIGlassesError): """API调用异常""" def __init__(self, message, code=None, request_id=None): super().__init__(message) self.code = code self.request_id = request_id class AuthenticationError(APIError): """认证失败""" pass class RateLimitError(APIError): """频率限制""" pass class InvalidRequestError(APIError): """请求参数错误""" pass class ServiceUnavailableError(APIError): """服务不可用""" pass class TimeoutError(AIGlassesError): """请求超时""" pass class NetworkError(AIGlassesError): """网络错误""" pass ``` 为什么要定义这么多异常类？因为不同的异常需要不同的处理策略： - `AuthenticationError`：需要重新配置API Key - `RateLimitError`：需要等待一段时间再重试 - `InvalidRequestError`：需要检查参数格式 - `ServiceUnavailableError`：需要切换到备用服务 ## 4. 完整SDK实现代码 下面是一个完整的SDK实现，包含了核心的封装逻辑和错误处理机制。 ### 4.1 基础客户端类 ```python # client.py import time import logging from typing import Optional, Dict, Any, Union from dataclasses import dataclass from .exceptions import * from .retry import RetryStrategy from .validators import validate_api_key, validate_request_params from .utils.logger import setup_logger @dataclass class ClientConfig: """客户端配置""" api_key: str base_url: str = "https://dashscope.aliyuncs.com/api/v1" timeout: int = 30 max_retries: int = 3 retry_delay: float = 1.0 enable_logging: bool = True log_level: str = "INFO" class BaseAPIClient: """API客户端基类""" def __init__(self, config: ClientConfig): self.config = config self.logger = setup_logger( name=self.__class__.__name__, level=config.log_level if config.enable_logging else "CRITICAL" ) self.retry_strategy = RetryStrategy( max_retries=config.max_retries, base_delay=config.retry_delay ) def _make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]: """发送HTTP请求（带重试和错误处理）""" # 验证API Key if not validate_api_key(self.config.api_key): raise AuthenticationError("无效的API Key格式") # 准备请求头 headers = { "Authorization": f"Bearer {self.config.api_key}", "Content-Type": "application/json", "User-Agent": f"AIGlasses-SDK/1.0.0" } # 合并自定义headers if "headers" in kwargs: headers.update(kwargs.pop("headers")) # 构建完整URL url = f"{self.config.base_url.rstrip('/')}/{endpoint.lstrip('/')}" # 请求参数 request_kwargs = { "method": method, "url": url, "headers": headers, "timeout": self.config.timeout, **kwargs } # 使用重试策略执行请求 return self.retry_strategy.execute( self._execute_request, request_kwargs ) def _execute_request(self, request_kwargs: Dict[str, Any]) -> Dict[str, Any]: """执行单次HTTP请求""" import requests import json method = request_kwargs.pop("method") url = request_kwargs.pop("url") self.logger.debug(f"发送请求: {method} {url}") try: response = requests.request(method, url, **request_kwargs) response.raise_for_status() # 解析响应 try: result = response.json() except json.JSONDecodeError as e: self.logger.error(f"响应解析失败: {e}, 原始响应: {response.text[:200]}") raise APIError(f"响应解析失败: {str(e)}") # 检查API错误码 if "code" in result and result["code"] != 200: error_msg = result.get("message", "未知错误") error_code = result.get("code") request_id = result.get("request_id") # 根据错误码抛出特定异常 if error_code == 400: raise InvalidRequestError(error_msg, error_code, request_id) elif error_code == 401: raise AuthenticationError(error_msg, error_code, request_id) elif error_code == 429: raise RateLimitError(error_msg, error_code, request_id) elif error_code >= 500: raise ServiceUnavailableError(error_msg, error_code, request_id) else: raise APIError(error_msg, error_code, request_id) return result except requests.exceptions.Timeout as e: self.logger.warning(f"请求超时: {url}") raise TimeoutError(f"请求超时: {str(e)}") except requests.exceptions.ConnectionError as e: self.logger.warning(f"网络连接失败: {url}") raise NetworkError(f"网络连接失败: {str(e)}") except requests.exceptions.RequestException as e: self.logger.error(f"请求异常: {str(e)}") raise APIError(f"请求异常: {str(e)}") ``` ### 4.2 智能重试策略 ```python # retry.py import time import random from typing import Callable, Any, Optional from .exceptions import RateLimitError, ServiceUnavailableError, TimeoutError class RetryStrategy: """智能重试策略""" def __init__( self, max_retries: int = 3, base_delay: float = 1.0, max_delay: float = 60.0, jitter: bool = True ): self.max_retries = max_retries self.base_delay = base_delay self.max_delay = max_delay self.jitter = jitter def execute(self, func: Callable, *args, **kwargs) -> Any: """执行函数，失败时自动重试""" last_exception = None for attempt in range(self.max_retries + 1): try: return func(*args, **kwargs) except (RateLimitError, ServiceUnavailableError, TimeoutError, NetworkError) as e: last_exception = e # 如果是最后一次尝试，直接抛出异常 if attempt == self.max_retries: break # 计算等待时间（指数退避 + 随机抖动） delay = self._calculate_delay(attempt) # 记录重试信息 self._log_retry(attempt, delay, e) # 等待 time.sleep(delay) except Exception as e: # 其他异常不重试，直接抛出 raise e # 所有重试都失败，抛出最后一个异常 raise last_exception def _calculate_delay(self, attempt: int) -> float: """计算重试延迟时间""" # 指数退避：2^attempt * base_delay delay = min(self.base_delay * (2 ** attempt), self.max_delay) # 添加随机抖动，避免多个客户端同时重试 if self.jitter: delay = delay * (0.5 + random.random()) return delay def _log_retry(self, attempt: int, delay: float, exception: Exception): """记录重试日志""" import logging logger = logging.getLogger(__name__) logger.warning( f"第 {attempt + 1} 次重试，等待 {delay:.2f} 秒后重试。" f"异常: {type(exception).__name__}: {str(exception)}" ) ``` ### 4.3 参数验证器 ```python # validators.py import re from typing import Any, Dict, List, Optional from .exceptions import InvalidRequestError def validate_api_key(api_key: str) -> bool: """验证API Key格式""" if not api_key: return False # DashScope API Key格式：sk-开头，32位字符 pattern = r'^sk-[a-zA-Z0-9]{32,}$' return bool(re.match(pattern, api_key)) def validate_request_params(params: Dict[str, Any], required: List[str] = None) -> None: """验证请求参数""" if required: missing = [field for field in required if field not in params] if missing: raise InvalidRequestError(f"缺少必要参数: {', '.join(missing)}") # 验证特定参数格式 if "model" in params: valid_models = ["qwen-max", "qwen-plus", "qwen-turbo", "qwen-vl-max"] if params["model"] not in valid_models: raise InvalidRequestError(f"无效的模型: {params['model']}") if "temperature" in params: temp = params["temperature"] if not isinstance(temp, (int, float)) or temp < 0 or temp > 2: raise InvalidRequestError("temperature必须在0到2之间") if "max_tokens" in params: tokens = params["max_tokens"] if not isinstance(tokens, int) or tokens < 1 or tokens > 2000: raise InvalidRequestError("max_tokens必须在1到2000之间") def validate_audio_file(file_path: str) -> bool: """验证音频文件""" import os from pathlib import Path # 检查文件是否存在 if not os.path.exists(file_path): return False # 检查文件大小（不超过10MB） file_size = os.path.getsize(file_path) if file_size > 10 * 1024 * 1024: # 10MB return False # 检查文件格式 valid_extensions = ['.wav', '.mp3', '.m4a', '.flac'] ext = Path(file_path).suffix.lower() return ext in valid_extensions def validate_image_url(url: str) -> bool: """验证图片URL""" import re # 简单的URL格式验证 pattern = r'^https?://.+\.(jpg|jpeg|png|gif|bmp|webp)(\?.*)?$' return bool(re.match(pattern, url, re.IGNORECASE)) ``` ### 4.4 语音识别API封装 ```python # api/asr.py import base64 from typing import Optional, Dict, Any from ..client import BaseAPIClient from ..validators import validate_audio_file class ASRClient(BaseAPIClient): """语音识别客户端""" def transcribe( self, audio_file: str, language: str = "zh-CN", sample_rate: int = 16000, format: str = "wav" ) -> str: """ 语音转文字 Args: audio_file: 音频文件路径 language: 语言代码，默认中文 sample_rate: 采样率 format: 音频格式 Returns: 识别出的文本 Raises: InvalidRequestError: 参数错误 APIError: API调用失败 """ # 验证音频文件 if not validate_audio_file(audio_file): raise InvalidRequestError(f"无效的音频文件: {audio_file}") # 读取并编码音频文件 with open(audio_file, "rb") as f: audio_data = f.read() audio_base64 = base64.b64encode(audio_data).decode("utf-8") # 构建请求数据 request_data = { "model": "paraformer-realtime-v2", "input": { "audio": f"data:audio/{format};base64,{audio_base64}" }, "parameters": { "language": language, "sample_rate": sample_rate } } # 发送请求 response = self._make_request( method="POST", endpoint="services/asr/transcription", json=request_data ) # 提取识别结果 if "output" in response and "text" in response["output"]: return response["output"]["text"] else: return "" def transcribe_stream(self, audio_stream, **kwargs) -> str: """流式语音识别（适用于实时音频）""" # 这里可以实现流式识别逻辑 # 为了简化示例，我们先返回空字符串 return "" ``` ### 4.5 多模态对话API封装 ```python # api/multimodal.py from typing import Optional, Dict, Any, List from ..client import BaseAPIClient from ..validators import validate_image_url class MultimodalClient(BaseAPIClient): """多模态对话客户端""" def chat( self, prompt: str, image_url: Optional[str] = None, model: str = "qwen-vl-max", temperature: float = 0.7, max_tokens: int = 1000 ) -> str: """ 多模态对话（文本+图片） Args: prompt: 用户输入的文本 image_url: 图片URL（可选） model: 使用的模型 temperature: 生成温度 max_tokens: 最大生成长度 Returns: AI回复文本 """ # 构建消息内容 content = [{"text": prompt}] if image_url: # 验证图片URL if not validate_image_url(image_url): raise InvalidRequestError(f"无效的图片URL: {image_url}") content.append({"image": image_url}) # 构建请求数据 request_data = { "model": model, "input": { "messages": [ { "role": "user", "content": content } ] }, "parameters": { "temperature": temperature, "max_tokens": max_tokens } } # 发送请求 response = self._make_request( method="POST", endpoint="services/aigc/multimodal-generation/generation", json=request_data ) # 提取回复 if "output" in response and "choices" in response["output"]: choices = response["output"]["choices"] if choices and "message" in choices[0]: return choices[0]["message"]["content"] return "" def chat_with_history( self, messages: List[Dict[str, Any]], model: str = "qwen-vl-max", **kwargs ) -> str: """ 带历史记录的多轮对话 Args: messages: 对话历史，格式：[{"role": "user", "content": "..."}, ...] model: 使用的模型 Returns: AI回复文本 """ # 验证消息格式 for msg in messages: if "role" not in msg or "content" not in msg: raise InvalidRequestError("消息格式错误，必须包含role和content字段") request_data = { "model": model, "input": {"messages": messages}, **kwargs } response = self._make_request( method="POST", endpoint="services/aigc/multimodal-generation/generation", json=request_data ) # 提取回复 if "output" in response and "choices" in response["output"]: choices = response["output"]["choices"] if choices and "message" in choices[0]: return choices[0]["message"]["content"] return "" ``` ### 4.6 主客户端类 ```python # aiglasses_sdk/__init__.py from .client import ClientConfig from .api.asr import ASRClient from .api.multimodal import MultimodalClient from .api.tts import TTSClient from .api.vision import VisionClient class AIGlassesClient: """AIGlasses SDK主客户端""" def __init__(self, api_key: str, **kwargs): """ 初始化客户端 Args: api_key: DashScope API Key **kwargs: 其他配置参数 """ config = ClientConfig(api_key=api_key, **kwargs) # 初始化各个API客户端 self.asr = ASRClient(config) self.multimodal = MultimodalClient(config) self.tts = TTSClient(config) self.vision = VisionClient(config) # 记录初始化日志 self._log_init(config) def _log_init(self, config: ClientConfig): """记录初始化日志""" import logging logger = logging.getLogger(__name__) logger.info("AIGlasses SDK初始化成功") logger.debug(f"API端点: {config.base_url}") logger.debug(f"超时设置: {config.timeout}秒") logger.debug(f"最大重试次数: {config.max_retries}") # 安全地记录API Key（只显示前8位） masked_key = config.api_key[:8] + "..." if config.api_key else "未设置" logger.info(f"API Key: {masked_key}") ``` ## 5. 在AIGlasses项目中的实际应用 现在，让我们看看如何在AIGlasses_for_navigation项目中使用这个SDK。 ### 5.1 初始化配置 ```python # config.py import os from typing import Optional from aiglasses_sdk import AIGlassesClient class AIGlassesConfig: """AIGlasses项目配置管理""" def __init__(self): self.api_key = self._load_api_key() self.sdk_client = None def _load_api_key(self) -> Optional[str]: """从配置文件加载API Key""" # 尝试从环境变量读取 api_key = os.getenv("DASHSCOPE_API_KEY") # 如果环境变量没有，尝试从配置文件读取 if not api_key: config_file = os.path.expanduser("~/.aiglasses/config.json") if os.path.exists(config_file): import json with open(config_file, "r") as f: config = json.load(f) api_key = config.get("api_key") return api_key def get_client(self) -> AIGlassesClient: """获取SDK客户端实例（单例模式）""" if self.sdk_client is None: if not self.api_key: raise ValueError("未找到API Key，请先配置") self.sdk_client = AIGlassesClient( api_key=self.api_key, timeout=30, # 超时时间30秒 max_retries=3, # 最大重试3次 enable_logging=True, log_level="INFO" ) # 测试连接 self._test_connection() return self.sdk_client def _test_connection(self): """测试API连接""" try: # 发送一个简单的请求测试连接 client = self.sdk_client # 这里可以添加一个简单的测试请求 print("API连接测试成功") except Exception as e: print(f"API连接测试失败: {e}") self.sdk_client = None raise ``` ### 5.2 导航系统核心逻辑 ```python # navigation_system.py import asyncio from typing import Dict, Any, Optional from dataclasses import dataclass from enum import Enum from config import AIGlassesConfig class NavigationState(Enum): """导航状态""" IDLE = "idle" # 空闲 BLINDWAY_NAV = "blindway_nav" # 盲道导航 CROSSING = "crossing" # 过马路 FINDING = "finding" # 寻找物品 CONVERSATION = "conversation" # 对话中 @dataclass class NavigationCommand: """导航指令""" action: str direction: Optional[str] = None distance: Optional[float] = None confidence: float = 0.0 class NavigationSystem: """导航系统核心类""" def __init__(self): self.config = AIGlassesConfig() self.client = self.config.get_client() self.state = NavigationState.IDLE self.current_target = None async def process_voice_command(self, audio_file: str) -> Optional[NavigationCommand]: """ 处理语音指令 Args: audio_file: 录音文件路径 Returns: 导航指令或None """ try: # 1. 语音识别 text = self.client.asr.transcribe(audio_file) if not text: print("未识别到语音") return None print(f"识别结果: {text}") # 2. 根据状态处理指令 if self.state == NavigationState.IDLE: return await self._handle_idle_command(text) elif self.state == NavigationState.BLINDWAY_NAV: return await self._handle_navigation_command(text) elif self.state == NavigationState.CROSSING: return await self._handle_crossing_command(text) elif self.state == NavigationState.FINDING: return await self._handle_finding_command(text) elif self.state == NavigationState.CONVERSATION: return await self._handle_conversation(text) except Exception as e: print(f"处理语音指令失败: {e}") # 这里可以添加错误恢复逻辑 return None async def _handle_idle_command(self, text: str) -> Optional[NavigationCommand]: """处理空闲状态下的指令""" text_lower = text.lower() # 开始盲道导航 if "开始导航" in text_lower or "盲道导航" in text_lower: self.state = NavigationState.BLINDWAY_NAV return NavigationCommand( action="start_navigation", confidence=0.9 ) # 开始过马路 elif "过马路" in text_lower or "帮我过马路" in text_lower: self.state = NavigationState.CROSSING return NavigationCommand( action="start_crossing", confidence=0.9 ) # 寻找物品 elif "帮我找" in text_lower or "找一下" in text_lower: self.state = NavigationState.FINDING # 提取物品名称 item = self._extract_item_name(text) self.current_target = item return NavigationCommand( action="find_item", target=item, confidence=0.8 ) # 普通对话 else: self.state = NavigationState.CONVERSATION response = await self._get_ai_response(text) # 这里可以播放语音回复 print(f"AI回复: {response}") self.state = NavigationState.IDLE return None async def _get_ai_response(self, text: str) -> str: """获取AI回复""" try: return self.client.multimodal.chat( prompt=text, temperature=0.7, max_tokens=500 ) except Exception as e: print(f"获取AI回复失败: {e}") return "抱歉，我现在无法回答这个问题。" def _extract_item_name(self, text: str) -> str: """从文本中提取物品名称""" # 简单的关键词提取 import re patterns = [ r"帮我找一下(.+?)(?:$|，|。)", r"找一下(.+?)(?:$|，|。)", r"帮我找(.+?)(?:$|，|。)" ] for pattern in patterns: match = re.search(pattern, text) if match: return match.group(1).strip() return "物品" ``` ### 5.3 错误处理最佳实践 ```python # error_handler.py import logging from typing import Callable, Any from functools import wraps from aiglasses_sdk.exceptions import * class ErrorHandler: """统一的错误处理器""" def __init__(self): self.logger = logging.getLogger(__name__) def handle_api_error(self, func: Callable) -> Callable: """ API错误处理装饰器 用法: @error_handler.handle_api_error def my_function(): # 调用API的代码 pass """ @wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except AuthenticationError as e: self.logger.error(f"认证失败: {e}") # 通知用户重新配置API Key self._notify_user("API Key无效，请重新配置") return None except RateLimitError as e: self.logger.warning(f"频率限制: {e}") # 等待后重试 import time time.sleep(5) try: return func(*args, **kwargs) except Exception: self._notify_user("服务繁忙，请稍后再试") return None except InvalidRequestError as e: self.logger.error(f"请求参数错误: {e}") # 记录错误参数，便于调试 self._log_request_details(args, kwargs) return None except (ServiceUnavailableError, TimeoutError, NetworkError) as e: self.logger.error(f"服务不可用: {e}") # 切换到离线模式 self._switch_to_offline_mode() return self._get_offline_response(func.__name__) except APIError as e: self.logger.error(f"API错误: {e}") return None except Exception as e: self.logger.exception(f"未知错误: {e}") return None return wrapper def _notify_user(self, message: str): """通知用户（语音或显示）""" # 这里可以实现语音播报或屏幕显示 print(f"[通知] {message}") def _log_request_details(self, args, kwargs): """记录请求详情（脱敏后）""" # 脱敏处理，不记录敏感信息 safe_args = [] for arg in args: if isinstance(arg, str) and "sk-" in arg: safe_args.append("***MASKED***") else: safe_args.append(str(arg)[:100]) # 只记录前100字符 self.logger.debug(f"请求参数 - args: {safe_args}, kwargs: {kwargs}") def _switch_to_offline_mode(self): """切换到离线模式""" self.logger.info("切换到离线模式") # 这里可以实现离线功能，如本地语音提示 def _get_offline_response(self, func_name: str) -> Any: """获取离线响应""" offline_responses = { "transcribe": "离线模式：语音识别不可用", "chat": "离线模式：AI对话不可用", "synthesize": "离线模式：语音合成不可用" } return offline_responses.get(func_name, "离线模式：功能不可用") # 使用示例 error_handler = ErrorHandler() @error_handler.handle_api_error def process_navigation_command(audio_file: str): """处理导航指令（带错误处理）""" system = NavigationSystem() return system.process_voice_command(audio_file) ``` ## 6. 测试用例 一个好的SDK必须有完善的测试。下面是一些关键测试用例： ```python # test_sdk.py import pytest import tempfile import os from unittest.mock import Mock, patch from aiglasses_sdk import AIGlassesClient, ClientConfig from aiglasses_sdk.exceptions import * class TestAIGlassesSDK: """SDK测试类""" def setup_method(self): """测试前准备""" # 使用模拟的API Key进行测试 self.config = ClientConfig( api_key="sk-test123456789012345678901234567890", base_url="https://mock.dashscope.aliyuncs.com/api/v1", timeout=5, max_retries=2 ) def test_api_key_validation(self): """测试API Key验证""" from aiglasses_sdk.validators import validate_api_key # 有效的API Key assert validate_api_key("sk-12345678901234567890123456789012") == True # 无效的API Key assert validate_api_key("") == False assert validate_api_key("sk-123") == False assert validate_api_key("invalid-key") == False def test_client_initialization(self): """测试客户端初始化""" client = AIGlassesClient( api_key="sk-test123456789012345678901234567890" ) assert client.asr is not None assert client.multimodal is not None assert client.tts is not None assert client.vision is not None @patch('requests.request') def test_asr_success(self, mock_request): """测试语音识别成功""" # 模拟成功的响应 mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { "output": { "text": "你好，世界" }, "code": 200 } mock_request.return_value = mock_response client = AIGlassesClient( api_key="sk-test123456789012345678901234567890" ) # 创建临时音频文件 with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f: f.write(b"fake audio data") audio_file = f.name try: text = client.asr.transcribe(audio_file) assert text == "你好，世界" finally: os.unlink(audio_file) @patch('requests.request') def test_asr_authentication_error(self, mock_request): """测试认证错误""" # 模拟认证错误 mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = { "code": 401, "message": "Invalid API Key" } mock_request.return_value = mock_response client = AIGlassesClient( api_key="sk-invalid-key" ) with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f: f.write(b"fake audio data") audio_file = f.name try: with pytest.raises(AuthenticationError): client.asr.transcribe(audio_file) finally: os.unlink(audio_file) @patch('requests.request') def test_retry_on_timeout(self, mock_request): """测试超时重试""" # 模拟第一次超时，第二次成功 mock_request.side_effect = [ TimeoutError("Request timeout"), Mock(status_code=200, json=lambda: { "output": {"text": "重试成功"}, "code": 200 }) ] client = AIGlassesClient( api_key="sk-test123456789012345678901234567890", max_retries=3 ) with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f: f.write(b"fake audio data") audio_file = f.name try: text = client.asr.transcribe(audio_file) assert text == "重试成功" assert mock_request.call_count == 2 finally: os.unlink(audio_file) def test_navigation_system(self): """测试导航系统""" # 这里可以添加导航系统的单元测试 system = NavigationSystem() # 测试状态转换 assert system.state == NavigationState.IDLE # 测试指令解析 command = system._handle_idle_command("开始导航") assert command is not None assert command.action == "start_navigation" def test_error_handler(self): """测试错误处理器""" handler = ErrorHandler() # 测试装饰器 @handler.handle_api_error def failing_function(): raise AuthenticationError("Invalid API Key") result = failing_function() assert result is None if __name__ == "__main__": pytest.main([__file__, "-v"]) ``` ## 7. 总结 通过这个完整的SDK封装实例，我们可以看到如何将复杂的API调用和错误处理逻辑封装成简洁易用的接口。总结一下关键点： ### 7.1 核心价值 1. **代码简化**：业务代码从复杂的HTTP请求和错误处理中解放出来，专注于核心逻辑 2. **统一错误处理**：所有API调用都有统一的错误处理机制，便于维护和调试 3. **智能重试**：针对不同的错误类型（如网络超时、服务限流）采用不同的重试策略 4. **易于测试**：良好的架构设计使得单元测试和集成测试更容易编写 5. **可扩展性**：新增API接口只需要继承基类，实现特定方法即可 ### 7.2 在AIGlasses项目中的应用 在AIGlasses_for_navigation这样的智能导航设备中，稳定可靠的API调用至关重要： - **实时性要求高**：导航指令需要快速响应，不能因为网络波动而卡顿 - **容错性要求强**：在户外环境中，网络条件可能不稳定，需要有完善的错误恢复机制 - **用户体验重要**：错误处理要友好，不能因为API调用失败就让整个系统崩溃 ### 7.3 最佳实践建议 1. **日志记录要详细**：记录请求参数、响应时间、错误信息，便于问题排查 2. **监控关键指标**：监控API调用成功率、响应时间、错误率等指标 3. **实现降级策略**：当云端服务不可用时，要有本地降级方案 4. **定期更新SDK**：随着API服务的更新，SDK也需要相应更新 5. **文档要完善**：提供清晰的API文档和使用示例 这个SDK封装方案不仅适用于AIGlasses_for_navigation项目，也可以作为其他AI应用项目的参考。关键是要根据实际需求调整错误处理策略、重试逻辑和监控指标。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。