import logging
from typing import Union, Generator, Optional, List
from dify_plugin.provider_kits.llm import LargeLanguageModel # 基底クラス
from dify_plugin.provider_kits.llm import LLMResult, LLMResultChunk, LLMUsage # 結果クラス
from dify_plugin.provider_kits.llm import PromptMessage, PromptMessageTool # メッセージクラス
from dify_plugin.errors.provider_error import InvokeError, InvokeAuthorizationError # エラークラス
logger = logging.getLogger(__name__)
class MyProviderLargeLanguageModel(LargeLanguageModel):
def _invoke(self, model: str, credentials: dict, prompt_messages: List[PromptMessage],
model_parameters: dict, tools: Optional[List[PromptMessageTool]] = None,
stop: Optional[List[str]] = None, stream: bool = True,
user: Optional[str] = None) -> Union[LLMResult, Generator[LLMResultChunk, None, None]]:
"""
モデル API を呼び出すためのコアメソッド。
パラメータ:
model: 呼び出すモデル識別子
credentials: 認証情報
prompt_messages: 送信するメッセージのリスト
model_parameters: temperature、max_tokens などのパラメータ
tools: 関数呼び出し用のオプションのツール定義
stop: オプションの停止シーケンスのリスト
stream: レスポンスをストリーミングするか（True）、完全なレスポンスを返すか（False）
user: API トラッキング用のオプションのユーザー識別子
戻り値:
stream=True の場合: LLMResultChunk オブジェクトを生成するジェネレータ
stream=False の場合: 完全な LLMResult オブジェクト
"""
# API リクエストパラメータの準備
api_params = self._prepare_api_params(
credentials, model_parameters, prompt_messages, tools, stop
)
try:
# ストリーミング設定に基づいて適切なヘルパーメソッドを呼び出す
if stream:
return self._invoke_stream(model, api_params, user)
else:
return self._invoke_sync(model, api_params, user)
except Exception as e:
# エラーを処理してマッピング
self._handle_api_error(e)
def _invoke_stream(self, model: str, api_params: dict, user: Optional[str]) -> Generator[LLMResultChunk, None, None]:
"""ストリーミング API 呼び出し用のヘルパーメソッド"""
# ストリーミング呼び出しの実装詳細
pass
def _invoke_sync(self, model: str, api_params: dict, user: Optional[str]) -> LLMResult:
"""同期 API 呼び出し用のヘルパーメソッド"""
# 同期呼び出しの実装詳細
pass
def validate_credentials(self, model: str, credentials: dict) -> None:
"""
この特定のモデルに対して認証情報が機能するかを検証します。
ユーザーが認証情報を追加または変更しようとするときに呼び出されます。
"""
# 認証情報検証の実装
pass
def get_num_tokens(self, model: str, credentials: dict,
prompt_messages: List[PromptMessage],
tools: Optional[List[PromptMessageTool]] = None) -> int:
"""
指定された入力のトークン数を推定します。
オプションですが、正確なコスト見積もりのために推奨されます。
"""
# トークンカウントの実装
pass
@property
def _invoke_error_mapping(self) -> dict[type[InvokeError], list[type[Exception]]]:
"""
ベンダー固有の例外から Dify 標準の例外へのマッピングを定義します。
これにより、異なるプロバイダー間でエラー処理を標準化できます。
"""
return {
InvokeAuthorizationError: [
# ベンダー固有の認証エラーをここにリスト
],
# その他のエラーマッピング
}
