1. 集成概述

Dify 支持 40+ 主流大语言模型提供商，通过统一的模型运行时层（Model Runtime）实现了对不同 LLM 服务的抽象和封装。这种设计使得开发者可以轻松切换不同的模型提供商，而无需修改应用代码。

核心价值

• 统一接口: 提供一致的 API 调用方式，屏蔽底层差异
• 灵活切换: 支持动态配置和切换模型提供商
• 多模型支持: 同时支持多个提供商和模型
• 类型完整: 支持 LLM、Embedding、Rerank、语音等多种模型类型

2. 支持的服务/产品

2.1 主流商业模型

服务名称	API 版本	状态	说明
OpenAI	Latest	✅ 支持	GPT-3.5, GPT-4, GPT-4o 等
Anthropic	Claude 3.x	✅ 支持	Claude 3 Opus, Sonnet, Haiku
Google	Gemini API	✅ 支持	Gemini Pro, Gemini Ultra
Azure OpenAI	Latest	✅ 支持	Azure 托管的 OpenAI 模型
Cohere	v1	✅ 支持	Command, Embed 系列
Mistral AI	Latest	✅ 支持	Mistral Large, Medium, Small

2.2 云服务商托管模型

服务名称	API 版本	状态	说明
AWS Bedrock	Latest	✅ 支持	Claude, Titan 等模型
Google Vertex AI	Latest	✅ 支持	PaLM, Gemini 等模型
Huawei Pangu	Latest	✅ 支持	华为盘古大模型
Volcengine MAAS	Latest	✅ 支持	火山引擎模型服务

2.3 中国区模型

服务名称	API 版本	状态	说明
智谱 AI (ZhipuAI)	Latest	✅ 支持	GLM-4, ChatGLM 系列
百川 (Baichuan)	Latest	✅ 支持	Baichuan2 系列
讯飞星火 (Spark)	v3.5	✅ 支持	Spark Max, Pro, Lite
文心一言 (Wenxin)	Latest	✅ 支持	ERNIE 系列
通义千问 (Tongyi)	Latest	✅ 支持	Qwen 系列
月之暗面 (Moonshot)	Latest	✅ 支持	Moonshot v1
腾讯混元 (Hunyuan)	Latest	✅ 支持	Hunyuan 系列
MiniMax	Latest	✅ 支持	MiniMax 系列
阶跃星辰 (StepFun)	Latest	✅ 支持	Step 系列
零一万物 (Yi)	Latest	✅ 支持	Yi 系列
智海 (Zhinao)	Latest	✅ 支持	360 智海模型

2.4 开源模型平台

服务名称	API 版本	状态	说明
Ollama	Latest	✅ 支持	本地运行开源模型
LocalAI	Latest	✅ 支持	OpenAI 兼容的本地 API
Xinference	Latest	✅ 支持	开源模型推理服务
OpenLLM	Latest	✅ 支持	BentoML 的 LLM 服务
HuggingFace Hub	Latest	✅ 支持	HuggingFace 推理 API

2.5 推理服务

服务名称	API 版本	状态	说明
Together AI	Latest	✅ 支持	快速推理服务
Replicate	Latest	✅ 支持	模型托管和推理
Groq	Latest	✅ 支持	超快速推理硬件
Fireworks AI	Latest	✅ 支持	高性能推理服务
OpenRouter	Latest	✅ 支持	模型路由服务
SiliconFlow	Latest	✅ 支持	硅基流动推理服务
Upstage	Latest	✅ 支持	专业 AI 推理
PerfXCloud	Latest	✅ 支持	性能优化推理

2.6 其他专用服务

服务名称	API 版本	状态	说明
Jina AI	Latest	✅ 支持	Embedding 和 Rerank
Nomic	Latest	✅ 支持	Embedding 模型
Voyage AI	Latest	✅ 支持	Embedding 模型
Mixedbread	Latest	✅ 支持	Embedding 和 Rerank
DeepSeek	Latest	✅ 支持	DeepSeek Coder 系列
NVIDIA NIM	Latest	✅ 支持	NVIDIA 推理微服务
Triton Inference	Latest	✅ 支持	NVIDIA Triton 服务器

2.7 兼容接口

服务名称	状态	说明
OpenAI API Compatible	✅ 支持	兼容 OpenAI API 格式的任意服务

3. 集成方式

3.1 架构设计

┌─────────────────────────────────────────────────────────┐
│                   Application Layer                      │
├─────────────────────────────────────────────────────────┤
│                   Model Manager                          │
│  ┌──────────────────────────────────────────────────┐  │
│  │          Provider Manager                         │  │
│  │  - 管理模型提供商实例                             │  │
│  │  - 凭证验证和管理                                 │  │
│  │  - 模型配置加载                                   │  │
│  └──────────────────────────────────────────────────┘  │
├─────────────────────────────────────────────────────────┤
│                Model Runtime Layer                       │
│  ┌──────────────────────────────────────────────────┐  │
│  │       Model Provider Factory                      │  │
│  │  - 动态创建提供商实例                             │  │
│  │  - 提供商注册和发现                               │  │
│  └──────────────────────────────────────────────────┘  │
│  ┌──────────────────────────────────────────────────┐  │
│  │          Base Model Classes                       │  │
│  │  - LargeLanguageModel                            │  │
│  │  - TextEmbeddingModel                            │  │
│  │  - RerankModel                                   │  │
│  │  - Speech2TextModel                              │  │
│  │  - TTSModel                                      │  │
│  │  - ModerationModel                               │  │
│  └──────────────────────────────────────────────────┘  │
├─────────────────────────────────────────────────────────┤
│              Provider Implementations                    │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐      │
│  │ OpenAI  │ │Anthropic│ │ Google  │ │  Ollama │ ...  │
│  │Provider │ │Provider │ │Provider │ │Provider │      │
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘      │
└─────────────────────────────────────────────────────────┘

3.2 核心接口

模型提供商基类

class AIModelProvider:
    """Base class for model provider"""
    
    def validate_provider_credentials(self, credentials: dict) -> None:
        """Validate provider credentials"""
        raise NotImplementedError
    
    def get_provider_schema(self) -> ProviderConfig:
        """Get provider configuration schema"""
        raise NotImplementedError

大语言模型基类

class LargeLanguageModel(AIModel):
    """Base class for large language models"""
    
    def invoke(
        self,
        model: str,
        credentials: dict,
        prompt_messages: list[PromptMessage],
        model_parameters: dict,
        tools: list[PromptMessageTool] | None = None,
        stop: list[str] | None = None,
        stream: bool = True,
        user: str | None = None
    ) -> LLMResult | Generator[LLMResultChunk, None, None]:
        """
        Invoke large language model
        
        :param model: model name
        :param credentials: model credentials
        :param prompt_messages: prompt messages
        :param model_parameters: model parameters
        :param tools: tools for tool calling
        :param stop: stop sequences
        :param stream: whether to stream response
        :param user: unique user id
        :return: full response or stream response chunk generator
        """
        raise NotImplementedError
    
    def get_num_tokens(
        self,
        model: str,
        credentials: dict,
        messages: list[PromptMessage]
    ) -> int:
        """
        Get number of tokens for given messages
        
        :param model: model name
        :param credentials: model credentials
        :param messages: messages
        :return: number of tokens
        """
        raise NotImplementedError

3.3 配置方式

环境变量配置

# OpenAI 配置
OPENAI_API_KEY=sk-xxx
OPENAI_API_BASE=https://api.openai.com/v1

# Anthropic 配置
ANTHROPIC_API_KEY=sk-ant-xxx
ANTHROPIC_API_URL=https://api.anthropic.com

# Azure OpenAI 配置
AZURE_OPENAI_API_KEY=xxx
AZURE_OPENAI_API_BASE=https://xxx.openai.azure.com
AZURE_OPENAI_API_VERSION=2023-05-15

# 默认模型配置
DEFAULT_LLM_MODEL_PROVIDER=openai
DEFAULT_LLM_MODEL_NAME=gpt-3.5-turbo

数据库配置

模型凭证存储在数据库中：

# models/provider.py
class Provider(Base):
    """Provider model"""
    
    provider_name: str  # 提供商名称
    provider_type: str  # 提供商类型 (system/custom)
    encrypted_config: str  # 加密的配置
    is_valid: bool  # 是否有效
    quota_type: str  # 配额类型
    quota_limit: int  # 配额限制

4. 代码实现

4.1 核心代码路径

api/core/model_runtime/
├── model_providers/           # 模型提供商实现
│   ├── __base/               # 基类定义
│   │   ├── ai_model.py
│   │   ├── large_language_model.py
│   │   ├── text_embedding_model.py
│   │   ├── rerank_model.py
│   │   └── ...
│   ├── openai/               # OpenAI 实现
│   ├── anthropic/            # Anthropic 实现
│   ├── google/               # Google 实现
│   ├── ollama/               # Ollama 实现
│   └── .../                  # 其他提供商
├── entities/                 # 实体定义
│   ├── model_entities.py     # 模型实体
│   ├── llm_entities.py       # LLM 实体
│   ├── message_entities.py   # 消息实体
│   └── provider_entities.py  # 提供商实体
├── errors/                   # 错误定义
│   ├── invoke.py            # 调用错误
│   └── validate.py          # 验证错误
├── callbacks/                # 回调处理
│   ├── base_callback.py
│   └── logging_callback.py
└── model_provider_factory.py # 提供商工厂

api/core/model_manager.py     # 模型管理器
api/core/provider_manager.py  # 提供商管理器

4.2 提供商位置配置

# api/core/model_runtime/model_providers/_position.yaml
- openai
- deepseek
- anthropic
- azure_openai
- google
- vertex_ai
- nvidia
- nvidia_nim
- cohere
- upstage
- bedrock
- togetherai
- openrouter
- ollama
- mistralai
- groq
- replicate
- huggingface_hub
- xinference
- triton_inference_server
- zhipuai
- baichuan
- spark
- minimax
- tongyi
- wenxin
- moonshot
- tencent
- jina
- chatglm
- yi
- openllm
- localai
- volcengine_maas
- openai_api_compatible
- hunyuan
- siliconflow
- perfxcloud
- zhinao
- fireworks
- mixedbread
- nomic
- voyage

4.3 模型工厂实现

# api/core/model_runtime/model_providers/model_provider_factory.py
class ModelProviderFactory:
    """Model provider factory"""
    
    @staticmethod
    def get_provider_instance(provider: str):
        """
        Get provider instance by provider name
        
        :param provider: provider name
        :return: provider instance
        """
        # 动态导入提供商模块
        provider_module = importlib.import_module(
            f'core.model_runtime.model_providers.{provider}'
        )
        
        # 获取提供商类
        provider_class = getattr(provider_module, f'{provider.capitalize()}Provider')
        
        # 创建实例
        return provider_class()

5. 使用示例

5.1 调用 LLM

from core.model_manager import ModelManager
from core.model_runtime.entities.message_entities import (
    PromptMessageTool,
    SystemPromptMessage,
    UserPromptMessage,
)

# 创建模型管理器
model_manager = ModelManager()

# 准备消息
messages = [
    SystemPromptMessage(content="You are a helpful assistant."),
    UserPromptMessage(content="What is the capital of France?"),
]

# 调用模型
response = model_manager.llm.invoke(
    tenant_id="tenant_id",
    provider="openai",
    model="gpt-3.5-turbo",
    credentials={
        "openai_api_key": "sk-xxx",
    },
    prompt_messages=messages,
    model_parameters={
        "temperature": 0.7,
        "max_tokens": 1000,
    },
    stream=False,
)

print(response.message.content)

5.2 流式调用

# 流式调用
response_stream = model_manager.llm.invoke(
    tenant_id="tenant_id",
    provider="openai",
    model="gpt-3.5-turbo",
    credentials={"openai_api_key": "sk-xxx"},
    prompt_messages=messages,
    model_parameters={"temperature": 0.7},
    stream=True,
)

# 处理流式响应
for chunk in response_stream:
    if chunk.delta.message and chunk.delta.message.content:
        print(chunk.delta.message.content, end="", flush=True)

5.3 工具调用 (Function Calling)

from core.model_runtime.entities.message_entities import PromptMessageTool

# 定义工具
tools = [
    PromptMessageTool(
        name="get_weather",
        description="Get the weather for a location",
        parameters={
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The city and state, e.g. San Francisco, CA",
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                },
            },
            "required": ["location"],
        },
    )
]

# 调用模型
response = model_manager.llm.invoke(
    tenant_id="tenant_id",
    provider="openai",
    model="gpt-3.5-turbo",
    credentials={"openai_api_key": "sk-xxx"},
    prompt_messages=messages,
    model_parameters={"temperature": 0.7},
    tools=tools,
    stream=False,
)

# 处理工具调用
if response.message.tool_calls:
    for tool_call in response.message.tool_calls:
        print(f"Tool: {tool_call.function.name}")
        print(f"Arguments: {tool_call.function.arguments}")

5.4 计算 Token 数量

# 计算 token 数量
num_tokens = model_manager.llm.get_num_tokens(
    provider="openai",
    model="gpt-3.5-turbo",
    credentials={"openai_api_key": "sk-xxx"},
    messages=messages,
)

print(f"Token count: {num_tokens}")

5.5 使用 Embedding 模型

# 获取文本嵌入
embeddings = model_manager.text_embedding.invoke(
    tenant_id="tenant_id",
    provider="openai",
    model="text-embedding-ada-002",
    credentials={"openai_api_key": "sk-xxx"},
    texts=["Hello world", "How are you?"],
)

print(f"Embeddings: {embeddings}")

6. 错误处理

6.1 常见错误类型

from core.model_runtime.errors.invoke import (
    InvokeAuthorizationError,      # 认证错误
    InvokeBadRequestError,          # 请求错误
    InvokeConnectionError,          # 连接错误
    InvokeRateLimitError,           # 速率限制错误
    InvokeServerUnavailableError,   # 服务不可用错误
)

6.2 错误处理示例

from core.model_runtime.errors.invoke import (
    InvokeAuthorizationError,
    InvokeRateLimitError,
)

try:
    response = model_manager.llm.invoke(
        tenant_id="tenant_id",
        provider="openai",
        model="gpt-3.5-turbo",
        credentials={"openai_api_key": "sk-xxx"},
        prompt_messages=messages,
        model_parameters={"temperature": 0.7},
        stream=False,
    )
except InvokeAuthorizationError as e:
    # 认证失败，检查 API 密钥
    print(f"Authorization failed: {e}")
except InvokeRateLimitError as e:
    # 速率限制，需要等待或降低请求频率
    print(f"Rate limit exceeded: {e}")
except Exception as e:
    # 其他错误
    print(f"Error: {e}")

6.3 重试策略

import time
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
from core.model_runtime.errors.invoke import InvokeRateLimitError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type(InvokeRateLimitError),
)
def invoke_with_retry():
    return model_manager.llm.invoke(
        tenant_id="tenant_id",
        provider="openai",
        model="gpt-3.5-turbo",
        credentials={"openai_api_key": "sk-xxx"},
        prompt_messages=messages,
        model_parameters={"temperature": 0.7},
        stream=False,
    )

7. 性能优化

7.1 连接池配置

# OpenAI HTTP 客户端配置
import httpx

http_client = httpx.Client(
    limits=httpx.Limits(
        max_keepalive_connections=20,
        max_connections=100,
    ),
    timeout=httpx.Timeout(60.0),
)

7.2 缓存策略

# 缓存模型响应
from extensions.ext_redis import redis_client

cache_key = f"llm:response:{hash(prompt)}"
cached_response = redis_client.get(cache_key)

if cached_response:
    return cached_response

response = model_manager.llm.invoke(...)
redis_client.setex(cache_key, 3600, response)  # 缓存 1 小时

7.3 批量处理

# 批量获取 embeddings
texts = ["text1", "text2", "text3", ...]
batch_size = 100

embeddings = []
for i in range(0, len(texts), batch_size):
    batch = texts[i:i + batch_size]
    batch_embeddings = model_manager.text_embedding.invoke(
        tenant_id="tenant_id",
        provider="openai",
        model="text-embedding-ada-002",
        credentials={"openai_api_key": "sk-xxx"},
        texts=batch,
    )
    embeddings.extend(batch_embeddings)

8. 监控与日志

8.1 日志记录

import logging

logger = logging.getLogger(__name__)

# 记录模型调用
logger.info(f"Invoking model: {provider}/{model}")
logger.debug(f"Parameters: {model_parameters}")

# 记录响应
logger.info(f"Response received: {response.usage}")

8.2 性能指标

import time

start_time = time.time()
response = model_manager.llm.invoke(...)
end_time = time.time()

# 记录性能指标
latency = end_time - start_time
logger.info(f"Model latency: {latency:.2f}s")
logger.info(f"Tokens used: {response.usage.total_tokens}")

8.3 OpenTelemetry 追踪

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span("llm_invoke") as span:
    span.set_attribute("provider", provider)
    span.set_attribute("model", model)
    
    response = model_manager.llm.invoke(...)
    
    span.set_attribute("tokens", response.usage.total_tokens)

9. 测试

9.1 单元测试

import pytest
from unittest.mock import Mock, patch

@patch('openai.ChatCompletion.create')
def test_openai_invoke(mock_create):
    # Mock OpenAI 响应
    mock_create.return_value = {
        'choices': [{
            'message': {
                'content': 'Paris'
            }
        }],
        'usage': {
            'total_tokens': 10
        }
    }
    
    # 测试调用
    response = model_manager.llm.invoke(
        tenant_id="test",
        provider="openai",
        model="gpt-3.5-turbo",
        credentials={"openai_api_key": "test-key"},
        prompt_messages=[
            UserPromptMessage(content="What is the capital of France?")
        ],
        model_parameters={"temperature": 0.7},
        stream=False,
    )
    
    assert response.message.content == "Paris"

9.2 集成测试

@pytest.mark.integration
def test_real_openai_invoke():
    """Test with real OpenAI API"""
    response = model_manager.llm.invoke(
        tenant_id="test",
        provider="openai",
        model="gpt-3.5-turbo",
        credentials={"openai_api_key": os.getenv("OPENAI_API_KEY")},
        prompt_messages=[
            UserPromptMessage(content="Say hello")
        ],
        model_parameters={"temperature": 0.7},
        stream=False,
    )
    
    assert response.message.content
    assert response.usage.total_tokens > 0

10. 扩展新服务

10.1 创建提供商目录结构

mkdir -p api/core/model_runtime/model_providers/my_provider
cd api/core/model_runtime/model_providers/my_provider

10.2 实现提供商类

# my_provider_provider.py
from core.model_runtime.model_providers.__base.model_provider import ModelProvider

class MyProviderProvider(ModelProvider):
    """My provider implementation"""
    
    def validate_provider_credentials(self, credentials: dict) -> None:
        """Validate provider credentials"""
        # 验证 API 密钥等凭证
        if not credentials.get('api_key'):
            raise CredentialsValidateFailedError('api_key is required')

10.3 实现模型类

# llm/my_model.py
from core.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel

class MyModelLargeLanguageModel(LargeLanguageModel):
    """My model LLM implementation"""
    
    def _invoke(self, model: str, credentials: dict,
                prompt_messages: list, model_parameters: dict,
                tools: list | None = None, stop: list | None = None,
                stream: bool = True, user: str | None = None) -> LLMResult:
        """Invoke the model"""
        # 实现模型调用逻辑
        pass
    
    def get_num_tokens(self, model: str, credentials: dict,
                      messages: list) -> int:
        """Get number of tokens"""
        # 实现 token 计算
        pass

10.4 创建配置文件

# my_provider.yaml
provider: my_provider
label:
  en_US: My Provider
  zh_Hans: 我的提供商
description:
  en_US: My awesome AI provider
  zh_Hans: 我的 AI 提供商
icon_small:
  en_US: icon_s_en.svg
icon_large:
  en_US: icon_l_en.svg
supported_model_types:
  - llm
  - text-embedding
configurate_methods:
  - predefined-model
provider_credential_schema:
  credential_form_schemas:
    - variable: api_key
      label:
        en_US: API Key
      type: secret-input
      required: true

10.5 添加到位置配置

# _position.yaml
- openai
- anthropic
# ... other providers
- my_provider  # 添加新提供商

10.6 测试提供商

# tests/unit_tests/core/model_runtime/model_providers/my_provider/test_provider.py
import pytest
from core.model_runtime.model_providers.my_provider import MyProviderProvider

def test_validate_credentials():
    provider = MyProviderProvider()
    
    # 测试有效凭证
    provider.validate_provider_credentials({
        'api_key': 'valid-key'
    })
    
    # 测试无效凭证
    with pytest.raises(CredentialsValidateFailedError):
        provider.validate_provider_credentials({})

11. 最佳实践

11.1 凭证管理

• 使用环境变量存储敏感信息
• 加密存储数据库中的凭证
• 定期轮换 API 密钥
• 使用最小权限原则

11.2 错误处理

• 捕获并处理所有可能的异常
• 实现指数退避重试策略
• 记录详细的错误信息
• 提供友好的错误消息给用户

11.3 性能优化

• 使用连接池复用连接
• 实现请求缓存机制
• 批量处理相似请求
• 监控和优化延迟

11.4 安全性

• 验证和清理用户输入
• 实施速率限制
• 使用 HTTPS 加密通信
• 定期更新依赖库

12. 常见问题

Q1: 如何切换模型提供商？

在应用配置中修改 provider 和 model 参数即可，无需修改代码。

Q2: 如何处理 API 配额限制？

实现速率限制和队列机制，使用指数退避重试策略。

Q3: 如何优化响应延迟？

使用流式响应、实现缓存、选择地理位置更近的服务。

Q4: 如何实现模型降级？

配置多个备用模型，当主模型失败时自动切换到备用模型。