HeurAMS

Appearance

Providers API 文档 ​

概述 ​

Providers 是 HeurAMS 的外部服务提供者模块，抽象外部服务接口并提供多种实现选项。该模块支持音频处理、语言模型和文本转语音等功能。

架构设计 ​

提供者基类 ​

所有提供者都继承自统一的基类，确保一致的接口和行为。

python
class BaseProvider:

核心方法:

  • __init__(config) - 初始化提供者
  • is_available() - 检查提供者是否可用
  • get_capabilities() - 获取提供者能力
  • cleanup() - 清理资源

抽象方法:

  • process(input_data) - 处理输入数据
  • validate_config(config) - 验证配置

子模块 ​

1. Audio - 音频提供者 ​

AudioProvider 类 ​

音频处理提供者基类，支持音频播放和分析。

python
class AudioProvider(BaseProvider):

核心方法:

  • play_audio(file_path) - 播放音频文件
  • record_audio(duration) - 录制音频
  • analyze_audio(file_path) - 分析音频特征
  • convert_format(input_path, output_path, format) - 转换音频格式

支持的音频格式:

  • WAV
  • MP3
  • FLAC
  • OGG

具体实现 ​

SystemAudioProvider 系统默认音频提供者，使用系统音频库。

python
class SystemAudioProvider(AudioProvider):

特性:

  • 使用系统音频 API
  • 低延迟播放
  • 自动格式检测

ExternalAudioProvider 外部音频服务提供者，支持在线音频处理。

python
class ExternalAudioProvider(AudioProvider):

特性:

  • 支持在线音频服务
  • 高质量音频处理
  • 需要网络连接

2. LLM - 语言模型提供者 ​

LLMProvider 类 ​

语言模型提供者基类，支持文本生成和分析。

python
class LLMProvider(BaseProvider):

核心方法:

  • generate_text(prompt, **kwargs) - 生成文本
  • analyze_text(text, task) - 分析文本
  • embed_text(text) - 获取文本嵌入
  • chat(messages) - 对话模式

生成参数:

  • max_tokens - 最大令牌数
  • temperature - 温度参数
  • top_p - 核采样参数
  • stop_sequences - 停止序列

具体实现 ​

OpenAIProvider OpenAI API 提供者。

python
class OpenAIProvider(LLMProvider):

配置:

toml
[llm.openai]
api_key = "your-api-key"
model = "gpt-3.5-turbo"
base_url = "https://api.openai.com/v1"

LocalLLMProvider 本地语言模型提供者。

python
class LocalLLMProvider(LLMProvider):

特性:

  • 离线运行
  • 隐私保护
  • 可自定义模型

3. TTS - 文本转语音提供者 ​

TTSProvider 类 ​

文本转语音提供者基类，支持语音合成。

python
class TTSProvider(BaseProvider):

核心方法:

  • synthesize(text, **kwargs) - 合成语音
  • get_voices() - 获取可用语音
  • set_voice(voice_id) - 设置语音
  • get_supported_languages() - 获取支持的语言

合成参数:

  • voice - 语音标识
  • speed - 语速
  • pitch - 音调
  • volume - 音量

具体实现 ​

SystemTTSProvider 系统 TTS 提供者。

python
class SystemTTSProvider(TTSProvider):

特性:

  • 使用系统 TTS 引擎
  • 支持多种语言
  • 无需额外配置

OnlineTTSProvider 在线 TTS 服务提供者。

python
class OnlineTTSProvider(TTSProvider):

特性:

  • 高质量语音合成
  • 多种语音选项
  • 需要网络连接

提供者注册和管理 ​

ProviderRegistry 类 ​

提供者注册表，管理所有可用的提供者。

python
class ProviderRegistry:

核心方法:

  • register(name, provider_class) - 注册提供者
  • get_provider(name, config) - 获取提供者实例
  • list_providers() - 列出所有提供者
  • get_default_provider(type) - 获取默认提供者

使用示例:

python
from heurams.providers.registry import ProviderRegistry

# 注册提供者
ProviderRegistry.register("audio_system", SystemAudioProvider)
ProviderRegistry.register("llm_openai", OpenAIProvider)

# 获取提供者实例
audio_provider = ProviderRegistry.get_provider("audio_system", audio_config)
llm_provider = ProviderRegistry.get_provider("llm_openai", llm_config)

配置管理 ​

提供者配置 ​

每个提供者都有特定的配置选项：

toml
# 音频提供者配置
[providers.audio.system]
enabled = true
volume = 80
device = "default"

# LLM 提供者配置
[providers.llm.openai]
api_key = "sk-..."
model = "gpt-3.5-turbo"
temperature = 0.7

# TTS 提供者配置
[providers.tts.system]
voice = "zh-CN-XiaoxiaoNeural"
speed = 1.0

配置验证 ​

提供者配置自动验证：

python
# 配置验证示例
config = {
    "api_key": "sk-...",
    "model": "gpt-3.5-turbo"
}

provider = OpenAIProvider(config)
if provider.is_available():
    # 提供者可用
    result = provider.generate_text("Hello")
else:
    # 使用备用提供者
    fallback_provider = ProviderRegistry.get_default_provider("llm")

错误处理 ​

提供者错误 ​

  • 连接错误: 自动重试或切换到备用提供者
  • 配置错误: 提供详细的错误信息和修复建议
  • 资源不足: 清理资源并重试

降级策略 ​

python
def get_fallback_provider(primary_provider, provider_type):
    """获取备用提供者"""
    if primary_provider.is_available():
        return primary_provider

    # 尝试默认提供者
    default_provider = ProviderRegistry.get_default_provider(provider_type)
    if default_provider.is_available():
        return default_provider

    # 返回 None 提供者
    return NoneProvider()

性能优化 ​

连接池 ​

对于需要网络连接的提供者，使用连接池优化性能：

python
class ConnectionPool:
    def __init__(self, max_connections=10):
        self.max_connections = max_connections
        self.connections = []

    def get_connection(self):
        # 获取或创建连接
        pass

    def release_connection(self, connection):
        # 释放连接
        pass

缓存机制 ​

缓存提供者结果，减少重复请求：

python
class ProviderCache:
    def __init__(self, ttl=3600):
        self.cache = {}
        self.ttl = ttl

    def get(self, key):
        # 获取缓存结果
        pass

    def set(self, key, value):
        # 设置缓存
        pass

使用示例 ​

基本使用 ​

python
from heurams.providers.registry import ProviderRegistry

# 获取音频提供者
audio_config = {"volume": 80}
audio_provider = ProviderRegistry.get_provider("audio_system", audio_config)

# 播放音频
if audio_provider.is_available():
    audio_provider.play_audio("sound.wav")

# 获取 LLM 提供者
llm_config = {
    "api_key": "sk-...",
    "model": "gpt-3.5-turbo"
}
llm_provider = ProviderRegistry.get_provider("llm_openai", llm_config)

# 生成文本
response = llm_provider.generate_text(
    "请解释间隔重复算法",
    max_tokens=500,
    temperature=0.7
)

高级使用 ​

python
# 使用多个提供者
providers = {}
for provider_type in ["audio", "llm", "tts"]:
    providers[provider_type] = ProviderRegistry.get_default_provider(provider_type)

# 批量处理
def process_with_fallback(provider, input_data):
    """使用备用策略处理数据"""
    try:
        return provider.process(input_data)
    except ProviderError:
        # 切换到备用提供者
        fallback = get_fallback_provider(provider, provider.type)
        return fallback.process(input_data)

自定义提供者 ​

python
from heurams.providers.base import BaseProvider

class CustomAudioProvider(BaseProvider):
    def __init__(self, config):
        super().__init__(config)
        self.setup_custom_audio()

    def process(self, input_data):
        # 自定义音频处理逻辑
        return self.custom_process(input_data)

    def validate_config(self, config):
        # 验证自定义配置
        required_keys = ["custom_param"]
        for key in required_keys:
            if key not in config:
                raise ConfigError(f"缺少必需的配置项: {key}")

# 注册自定义提供者
ProviderRegistry.register("audio_custom", CustomAudioProvider)

扩展接口 ​

提供者插件 ​

支持通过插件系统扩展提供者：

python
# 插件提供者示例
class PluginAudioProvider(AudioProvider):
    def __init__(self, config, plugin_path):
        super().__init__(config)
        self.load_plugin(plugin_path)

    def load_plugin(self, plugin_path):
        # 加载插件
        pass

提供者组合 ​

组合多个提供者实现复杂功能：

python
class CompositeProvider(BaseProvider):
    def __init__(self, providers):
        self.providers = providers

    def process(self, input_data):
        # 使用多个提供者处理数据
        results = []
        for provider in self.providers:
            if provider.is_available():
                results.append(provider.process(input_data))
        return self.merge_results(results)

兼容性说明 ​

  • Python 版本: 需要 Python 3.8 或更高版本
  • 平台支持: 跨平台支持，具体功能可能因平台而异
  • 网络要求: 部分提供者需要网络连接
  • 依赖项: 根据使用的提供者可能需要额外依赖

© 2025 Wang Zhiyu 保留所有权