在 AI 技术快速普及的今天，借助成熟的 AI API 服务可以让开发者快速为应用注入智能能力，无需从零构建复杂模型。yunwu.ai 作为提供全方位 AI 能力的平台，其 API 接口涵盖自然语言处理、计算机视觉、语音交互等多个领域。本文以程序员视角，详解基于 yunwu.ai 的 API 接入流程、核心功能调用方法和实战优化技巧，通过代码示例展示从配置到落地的完整开发路径，帮助开发者高效集成 AI 能力。

API 接入基础：环境配置与认证机制

接入 yunwu.ai 的 AI 服务首先需要完成环境搭建和认证配置，这是确保 API 调用安全可靠的基础。平台采用 API 密钥认证机制，开发者需要通过官方控制台获取访问凭证，并在请求中正确携带认证信息。

基础配置与认证代码实现：

import requests

import json

import time

from typing import Dict, Optional, Any

import logging

from dotenv import load_dotenv

import os

# 加载环境变量

load_dotenv()

# 配置日志

logging.basicConfig(level=logging.INFO)

logger = logging.getLogger("yunwu_ai_api")

class YunwuAI:

"""yunwu.ai API客户端"""

def __init__(self):

# 从环境变量获取配置

self.api_key = os.getenv("YUNWU_AI_API_KEY")

self.api_secret = os.getenv("YUNWU_AI_API_SECRET")

self.base_url = os.getenv("YUNWU_AI_BASE_URL", "https://api.yunwu.ai/v1")

# 验证配置

if not all([self.api_key, self.api_secret]):

raise ValueError("请配置YUNWU_AI_API_KEY和YUNWU_AI_API_SECRET环境变量")

# 初始化会话

self.session = requests.Session()

self.session.headers.update({

"Content-Type": "application/json",

"User-Agent": "yunwu-ai-python-sdk/1.0.0"

})

# 获取访问令牌

self.access_token = self._get_access_token()

self.token_expiry = time.time() + 3500 # 令牌有效期约1小时

def _get_access_token(self) -> str:

"""获取访问令牌"""

logger.info("获取访问令牌...")

url = f"{self.base_url}/auth/token"

payload = {

"api_key": self.api_key,

"api_secret": self.api_secret

}

try:

response = self.session.post(url, json=payload, timeout=10)

response.raise_for_status()

result = response.json()

if "access_token" not in result:

raise ValueError("获取令牌失败: 响应中缺少access_token")

logger.info("访问令牌获取成功")

return result["access_token"]

except requests.exceptions.RequestException as e:

logger.error(f"令牌获取失败: {str(e)}")

raise

def _ensure_token_valid(self):

"""确保令牌有效"""

if time.time() >= self.token_expiry:

logger.info("令牌即将过期，重新获取...")

self.access_token = self._get_access_token()

self.token_expiry = time.time() + 3500

def _request(self, method: str, endpoint: str,

params: Optional[Dict] = None,

data: Optional[Dict] = None) -> Dict[str, Any]:

"""通用请求方法"""

self._ensure_token_valid()

url = f"{self.base_url}/{endpoint}"

headers = {

"Authorization": f"Bearer {self.access_token}"

}

try:

logger.info(f"发起请求: {method} {url}")

response = self.session.request(

method=method,

url=url,

params=params,

json=data,

headers=headers,

timeout=30

)

# 处理响应

if response.status_code in [200, 201]:

return response.json()

elif response.status_code == 401:

# 令牌过期，重试一次

self.access_token = self._get_access_token()

headers["Authorization"] = f"Bearer {self.access_token}"

response = self.session.request(

method=method,

url=url,

params=params,

json=data,

headers=headers,

timeout=30

)

if response.status_code in [200, 201]:

return response.json()

# 处理错误响应

logger.error(f"API请求失败: {response.status_code} {response.text}")

response.raise_for_status()

return {} # unreachable code

except requests.exceptions.RequestException as e:

logger.error(f"请求发生错误: {str(e)}")

raise

在接入过程中需要注意三个关键事项：一是安全存储凭证，避免将 API 密钥硬编码在代码中，推荐使用环境变量或配置文件管理；二是令牌生命周期管理，实现自动刷新机制避免令牌过期导致的请求失败；三是错误处理机制，针对网络异常、认证失败等情况设计重试逻辑。上述代码实现了完整的客户端基础框架，包括配置加载、令牌管理和通用请求方法，为后续 API 调用提供了稳定的基础支撑。开发者可以根据实际需求扩展该类，添加具体的功能调用方法。

核心功能实战：典型 AI 能力的 API 调用实现

yunwu.ai 提供了丰富的 AI 能力接口，涵盖文本处理、图像识别、语音转换等多个领域。掌握这些核心功能的调用方法，能够快速为应用添加智能特性，提升产品体验。

核心功能调用代码示例：

# 扩展YunwuAI类，添加具体功能调用

class YunwuAI(YunwuAI): # 继承前面定义的基础类

def text_generation(self, prompt: str,

model: str = "yunwu-pro-7b",

max_tokens: int = 512,

temperature: float = 0.7,

top_p: float = 0.9) -> Dict[str, Any]:

"""

文本生成API调用

:param prompt: 提示词

:param model: 模型名称

:param max_tokens: 最大生成 tokens 数

:param temperature: 温度参数，控制随机性

:param top_p: 采样参数

:return: 生成结果

"""

endpoint = "text/generate"

data = {

"model": model,

"prompt": prompt,

"max_tokens": max_tokens,

"temperature": temperature,

"top_p": top_p,

"stream": False

}

return self._request("POST", endpoint, data=data)

def image_classification(self, image_path: str,

model: str = "vision-pro") -> Dict[str, Any]:

"""

图像分类API调用

:param image_path: 本地图像路径

:param model: 模型名称

:return: 分类结果

"""

# 先上传图片获取URL

image_url = self._upload_image(image_path)

endpoint = "vision/classify"

data = {

"model": model,

"image_url": image_url,

"top_k": 5 # 返回 top 5 结果

}

return self._request("POST", endpoint, data=data)

def _upload_image(self, image_path: str) -> str:

"""上传图片获取URL"""

endpoint = "files/upload"

# 读取图片文件

with open(image_path, "rb") as f:

files = {"file": (os.path.basename(image_path), f, "image/jpeg")}

# 构建表单数据

form_data = {

"purpose": "vision"

}

# 使用单独的请求，不使用JSON格式

self._ensure_token_valid()

url = f"{self.base_url}/{endpoint}"

headers = {

"Authorization": f"Bearer {self.access_token}"

}

response = self.session.post(

url=url,

data=form_data,

files=files,

headers=headers,

timeout=60

)

response.raise_for_status()

result = response.json()

if "file_url" not in result:

raise ValueError("图片上传失败: 响应中缺少file_url")

return result["file_url"]

def speech_to_text(self, audio_path: str,

language: str = "zh-CN",

model: str = "audio-pro") -> Dict[str, Any]:

"""

语音转文字API调用

:param audio_path: 音频文件路径

:param language: 语言代码

:param model: 模型名称

:return: 识别结果

"""

# 上传音频文件

audio_url = self._upload_audio(audio_path)

endpoint = "audio/speech-to-text"

data = {

"model": model,

"audio_url": audio_url,

"language": language,

"enable_punctuation": True

}

return self._request("POST", endpoint, data=data)

def _upload_audio(self, audio_path: str) -> str:

"""上传音频文件获取URL"""

endpoint = "files/upload"

with open(audio_path, "rb") as f:

files = {"file": (os.path.basename(audio_path), f, "audio/wav")}

form_data = {"purpose": "audio"}

self._ensure_token_valid()

url = f"{self.base_url}/{endpoint}"

headers = {

"Authorization": f"Bearer {self.access_token}"

}

response = self.session.post(

url=url,

data=form_data,

files=files,

headers=headers,

timeout=60

)

response.raise_for_status()

result = response.json()

if "file_url" not in result:

raise ValueError("音频上传失败: 响应中缺少file_url")

return result["file_url"]

def text_embedding(self, texts: list[str],

model: str = "embedding-v1") -> Dict[str, Any]:

"""

文本嵌入API调用

:param texts: 文本列表

:param model: 模型名称

:return: 嵌入向量结果

"""

endpoint = "text/embeddings"

data = {

"model": model,

"input": texts

}

return self._request("POST", endpoint, data=data)

# 实战应用示例

def ai_api_practical_demo():

# 初始化客户端

ai = YunwuAI()

# 1. 文本生成示例：生成产品描述

print("=== 文本生成示例 ===")

prompt = """请为一款智能恒温杯生成产品描述，突出其特点：

- 智能温控，支持APP调节

- 长效保温24小时

- 食品级不锈钢材质

- 防烫设计，安全可靠

"""

text_result = ai.text_generation(

prompt=prompt,

model="yunwu-pro-7b",

max_tokens=300,

temperature=0.8

)

print("生成结果:", text_result.get("choices", [{}])[0].get("text", ""))

# 2. 图像分类示例：识别产品图片

print("\n=== 图像分类示例 ===")

try:

image_result = ai.image_classification(

image_path="product_image.jpg",

model="vision-pro"

)

print("分类结果:", image_result.get("classifications", []))

except Exception as e:

print(f"图像分类失败: {str(e)}")

# 3. 语音转文字示例：会议记录

print("\n=== 语音转文字示例 ===")

try:

speech_result = ai.speech_to_text(

audio_path="meeting_recording.wav",

language="zh-CN"

)

print("识别文本:", speech_result.get("text", ""))

except Exception as e:

print(f"语音识别失败: {str(e)}")

# 4. 文本嵌入示例：生成语义向量

print("\n=== 文本嵌入示例 ===")

texts = [

"智能恒温杯支持APP远程控制温度",

"这款保温杯采用食品级不锈钢材质",

"高性能笔记本电脑拥有超长续航能力"

]

embedding_result = ai.text_embedding(texts=texts)

print(f"生成嵌入向量数量: {len(embedding_result.get('data', []))}")

# 执行示例

if __name__ == "__main__":

ai_api_practical_demo()

调用 AI API 时需要根据不同功能类型采用相应的调用策略：对于文本类接口，需重点关注提示词工程和参数调优（如 temperature 控制生成随机性）；对于图像和语音接口，需处理文件上传流程并注意格式要求；对于嵌入接口，需合理组织输入文本数量以平衡性能和成本。上述代码实现了文本生成、图像分类、语音转文字和文本嵌入等典型功能的调用方法，并提供了完整的实战示例。在实际应用中，开发者应根据具体场景选择合适的模型，优化参数配置，并做好错误处理和结果验证，确保 AI 功能的稳定可靠运行。

性能优化与最佳实践：提升 API 集成质量

在大规模使用 AI API 时，性能优化和成本控制成为关键考量。通过合理的请求管理、缓存策略和错误处理，能够显著提升 API 调用效率，降低集成成本，同时确保服务稳定性。

优化与最佳实践代码实现：

import functools

import time

from datetime import datetime, timedelta

from cachetools import TTLCache

# 添加缓存和批处理能力的优化扩展

class OptimizedYunwuAI(YunwuAI): # 继承前面的YunwuAI类

def __init__(self, *args, **kwargs):

super().__init__(*args, **kwargs)

# 初始化缓存：键为函数名+参数哈希，值为结果，有效期10分钟

self.cache = TTLCache(maxsize=1000, ttl=600)

# 记录API调用统计

self.call_stats = {

"success": 0,

"failure": 0,

"total_tokens": 0,

"total_time": 0.0

}

def _cache_decorator(self, func):

"""缓存装饰器"""

@functools.wraps(func)

def wrapper(*args, **kwargs):

# 排除不适合缓存的参数

cache_kwargs = {k: v for k, v in kwargs.items()

if k not in ["stream", "max_tokens"]}

# 生成缓存键

cache_key = (func.__name__, args, frozenset(cache_kwargs.items()))

# 检查缓存

if cache_key in self.cache:

logger.info(f"从缓存获取 {func.__name__} 结果")

return self.cache[cache_key]

# 执行函数

start_time = time.time()

try:

result = func(*args, **kwargs)

self.call_stats["success"] += 1

# 记录tokens使用量（如果有）

if "usage" in result and "total_tokens" in result["usage"]:

self.call_stats["total_tokens"] += result["usage"]["total_tokens"]

# 缓存结果

self.cache[cache_key] = result

return result

except Exception as e:

self.call_stats["failure"] += 1

raise

finally:

# 记录耗时

self.call_stats["total_time"] += (time.time() - start_time)

return wrapper

# 为常用方法添加缓存

@_cache_decorator

def text_embedding(self, *args, **kwargs):

return super().text_embedding(*args, **kwargs)

@_cache_decorator

def image_classification(self, *args, **kwargs):

return super().image_classification(*args, **kwargs)

def batch_text_generation(self, prompts: list[str],

batch_size: int = 5,

** kwargs) -> list[Dict[str, Any]]:

"""

批量文本生成，控制并发

:param prompts: 提示词列表

:param batch_size: 每批处理数量

:return: 结果列表

"""

results = []

total = len(prompts)

for i in range(0, total, batch_size):

batch = prompts[i:min(i+batch_size, total)]

logger.info(f"处理批量文本生成: 批次 {i//batch_size + 1}/{(total + batch_size -1)//batch_size}")

# 并发处理批次

batch_results = []

for prompt in batch:

try:

result = self.text_generation(prompt, **kwargs</doubaocanvas>