第2天：核心API客户端开发
📋 第2天概述
在第1天完成环境搭建的基础上，第2天将重点开发核心API客户端，建立与阿里云三大AI服务（文字生成、图片生成、语音合成）的稳定连接和基础功能调用。
🎯 第2天目标
主要任务：实现完整的阿里云API客户端，包含错误处理、重试机制和响应解析
核心需求：构建可复用的API调用模块，支持三大AI服务的稳定调用
🏗️ 架构设计
客户端类设计
class AliyunClient:"""阿里云AI服务统一客户端"""def __init__(self, config):self.config = configself.session = requests.Session()# 文字生成相关方法def generate_text(self, prompt, **kwargs)def _call_text_api(self, payload)# 图片生成相关方法def generate_image(self, prompt, **kwargs)def _call_image_api(self, payload)# 语音合成相关方法def synthesize_speech(self, text, **kwargs)def _generate_nls_token(self)def _call_speech_api(self, payload)# 通用工具方法def _get_headers(self, service_type)def _handle_response(self, response)def _retry_request(self, func, max_retries=3)
🔧 具体实施步骤
步骤1：创建核心客户端类
创建 src/core/aliyun_client.py：
import requests
import json
import time
import base64
import hmac
import hashlib
import urllib.parse
from datetime import datetime
from typing import Dict, Any, Optionalclass AliyunClient:"""阿里云百炼大模型API客户端用于封装与阿里云API的交互"""def __init__(self, config):self.config = configself.session = requests.Session()# 设置请求超时和重试策略self.session.mount('http://', requests.adapters.HTTPAdapter(max_retries=3))self.session.mount('https://', requests.adapters.HTTPAdapter(max_retries=3))def _get_headers(self, service_type: str) -> Dict[str, str]:"""获取不同服务的请求头"""base_headers = {"Content-Type": "application/json"}if service_type == "text":base_headers["Authorization"] = f"Bearer {self.config.ALIYUN_API_KEY}"elif service_type == "image":base_headers["Authorization"] = f"Bearer {self.config.ALIYUN_API_KEY}"elif service_type == "speech":# 语音服务需要动态Tokentoken = self._generate_nls_token()base_headers["X-NLS-Token"] = tokenreturn base_headersdef _generate_nls_token(self) -> str:"""生成智能语音交互API的访问令牌"""if not self.config.ALIYUN_ACCESS_KEY_ID or not self.config.ALIYUN_ACCESS_KEY_SECRET:raise ValueError("缺少AccessKey配置，无法生成Token")# 阿里云智能语音交互Token生成APItoken_url = f"{self.config.TOKEN_SERVICE_URL}/pop/2018-05-18/tokens"# 构建请求参数params = {"Action": "CreateToken"}# 添加阿里云POP协议要求的公共参数timestamp = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")params.update({"Format": "JSON","Version": "2019-02-28","AccessKeyId": self.config.ALIYUN_ACCESS_KEY_ID,"SignatureMethod": "HMAC-SHA1","SignatureVersion": "1.0","SignatureNonce": str(int(datetime.utcnow().timestamp() * 1000)),"Timestamp": timestamp})# 对参数进行排序sorted_params = sorted(params.items())# 构建规范化请求字符串canonicalized_query_string = '&'.join([f"{urllib.parse.quote(k)}={urllib.parse.quote(v)}" for k, v in sorted_params])# 构建签名字符串string_to_sign = "POST&%2F&" + urllib.parse.quote(canonicalized_query_string)# 计算签名key = self.config.ALIYUN_ACCESS_KEY_SECRET + "&"signature = base64.b64encode(hmac.new(key.encode('utf-8'), string_to_sign.encode('utf-8'), hashlib.sha1).digest()).decode('utf-8')# 添加签名到参数params["Signature"] = signature# 构建完整URLquery_string = '&'.join([f"{k}={urllib.parse.quote(v)}" for k, v in params.items()])full_url = f"{token_url}?{query_string}"try:# 发送请求获取Tokenresponse = self.session.post(full_url, timeout=10)if response.status_code == 200:result = response.json()if "Token" in result and "Id" in result["Token"]:return result["Token"]["Id"]else:raise ValueError(f"Token响应格式错误: {result}")else:raise ValueError(f"Token API请求失败: {response.status_code} - {response.text}")except Exception as e:raise Exception(f"Token生成失败: {e}")def _retry_request(self, func, max_retries: int = 3, delay: float = 1.0) -> Any:"""重试机制实现"""for attempt in range(max_retries):try:return func()except Exception as e:if attempt == max_retries - 1:raise etime.sleep(delay * (2 ** attempt)) # 指数退避def _handle_response(self, response: requests.Response) -> Dict[str, Any]:"""统一处理API响应"""if response.status_code != 200:raise Exception(f"API请求失败: {response.status_code} - {response.text}")try:return response.json()except json.JSONDecodeError:raise Exception(f"响应解析失败: {response.text}")
步骤2：实现文字生成功能
def generate_text(self, prompt: str, model: str = "qwen-max", temperature: float = 0.7, max_tokens: int = 1000, **kwargs) -> Dict[str, Any]:"""调用文本生成APIArgs:prompt: 提示词model: 模型名称temperature: 生成温度max_tokens: 最大生成长度**kwargs: 其他参数Returns:dict: API响应结果"""def _call_api():headers = self._get_headers("text")# 兼容模式对话API使用OpenAI兼容格式payload = {"model": model,"messages": [{"role": "user","content": prompt}],"temperature": temperature,"max_tokens": max_tokens}# 添加额外参数payload.update(kwargs)response = self.session.post(self.config.TEXT_GENERATION_URL,headers=headers,json=payload,timeout=30)return self._handle_response(response)return self._retry_request(_call_api)
步骤3：实现图片生成功能
def generate_image(self, prompt: str, model: str = "wan2.5-t2i-preview", size: str = "1024x1024", style: str = "cartoon", n: int = 1, **kwargs) -> Dict[str, Any]:"""调用图像生成APIArgs:prompt: 提示词model: 模型名称size: 图片尺寸style: 图片风格n: 生成数量**kwargs: 其他参数Returns:dict: API响应结果"""def _call_api():headers = self._get_headers("image")payload = {"model": model,"input": {"prompt": prompt},"parameters": {"size": size,"style": style,"n": n}}# 添加额外参数payload["parameters"].update(kwargs)response = self.session.post(self.config.IMAGE_GENERATION_URL,headers=headers,json=payload,timeout=60 # 图片生成需要更长时间)return self._handle_response(response)return self._retry_request(_call_api)
步骤4：实现语音合成功能
def synthesize_speech(self, text: str, voice: str = "xiaoyun", format: str = "wav", sample_rate: int = 16000, volume: int = 50, **kwargs) -> Dict[str, Any]:"""调用语音合成APIArgs:text: 要合成的文本voice: 语音类型format: 音频格式sample_rate: 采样率volume: 音量**kwargs: 其他参数Returns:dict: API响应结果"""def _call_api():headers = self._get_headers("speech")payload = {"appkey": self.config.SPEECH_APP_KEY,"text": text,"voice": voice,"format": format,"sample_rate": sample_rate,"volume": volume}# 添加额外参数payload.update(kwargs)response = self.session.post(self.config.SPEECH_SYNTHESIS_URL,headers=headers,json=payload,timeout=30)return self._handle_response(response)return self._retry_request(_call_api)
步骤5：创建响应解析工具
def parse_text_response(self, response: Dict[str, Any]) -> str:"""解析文本生成响应"""# 尝试从不同的可能路径提取文本generated_text = ""# 路径1: result.choices[0].message.content (OpenAI格式)if 'choices' in response and len(response['choices']) > 0:choice = response['choices'][0]if 'message' in choice and 'content' in choice['message']:generated_text = choice['message']['content']# 路径2: result.output.text (阿里云格式)elif 'output' in response and 'text' in response['output']:generated_text = response['output']['text']# 路径3: result.content (简化格式)elif 'content' in response:generated_text = response['content']# 路径4: result.text (直接格式)elif 'text' in response:generated_text = response['text']if not generated_text:raise ValueError("无法从响应中提取文本内容")return generated_textdef parse_image_response(self, response: Dict[str, Any]) -> str:"""解析图片生成响应"""if 'output' in response and 'task_id' in response['output']:return response['output']['task_id']elif 'data' in response and len(response['data']) > 0:return response['data'][0].get('url', '')else:raise ValueError("无法从响应中提取图片信息")def parse_speech_response(self, response: Dict[str, Any]) -> bytes:"""解析语音合成响应"""if 'audio' in response:# 返回base64编码的音频数据return base64.b64decode(response['audio'])else:raise ValueError("无法从响应中提取音频数据")
🧪 测试用例
创建 tests/test_aliyun_client.py：
import unittest
from src.core.aliyun_client import AliyunClient
from src.core.config import Configclass TestAliyunClient(unittest.TestCase):def setUp(self):self.config = Config()self.client = AliyunClient(self.config)def test_text_generation(self):"""测试文字生成功能"""try:response = self.client.generate_text("写一个简短的测试故事")self.assertIsInstance(response, dict)text = self.client.parse_text_response(response)self.assertIsInstance(text, str)self.assertTrue(len(text) > 0)print(f"✅ 文字生成测试通过，生成文本长度: {len(text)}")except Exception as e:print(f"⚠️ 文字生成测试跳过（API可能不可用）: {e}")def test_image_generation(self):"""测试图片生成功能"""try:response = self.client.generate_image("一只可爱的小猫")self.assertIsInstance(response, dict)task_id = self.client.parse_image_response(response)self.assertIsInstance(task_id, str)print(f"✅ 图片生成测试通过，任务ID: {task_id}")except Exception as e:print(f"⚠️ 图片生成测试跳过（API可能不可用）: {e}")def test_speech_synthesis(self):"""测试语音合成功能"""try:response = self.client.synthesize_speech("这是一个测试语音")self.assertIsInstance(response, dict)audio_data = self.client.parse_speech_response(response)self.assertIsInstance(audio_data, bytes)print(f"✅ 语音合成测试通过，音频数据长度: {len(audio_data)}")except Exception as e:print(f"⚠️ 语音合成测试跳过（API可能不可用）: {e}")if __name__ == "__main__":unittest.main()
📊 第2天成果
完成内容
- ✅ 核心API客户端类实现
- ✅ 三大AI服务功能封装
- ✅ 错误处理和重试机制
- ✅ 响应解析工具
- ✅ 完整的测试用例
技术要点
- 统一的请求头管理
- 智能Token生成机制
- 指数退避重试策略
- 多格式响应解析
- 完整的异常处理
🔄 明日计划
第3天将基于今天的API客户端，实现文字生成服务的具体业务逻辑和用户界面。
文档创建时间：2024年开发计划第2天
预计耗时：6-8小时
关键文件：src/core/aliyun_client.py, tests/test_aliyun_client.py