Coverage for agentos/llm/pangu_provider.py: 55%

1""" 

2Huawei Pangu Provider — 华为盘古大模型，纯 httpx 实现。 

3支持 Pangu 系列模型：pangu-4, pangu-3.1, pangu-code 等。 

4 

5环境变量: PANGU_API_KEY, PANGU_BASE_URL 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11 

12from agentos.llm.openai_provider import OpenAIProvider 

13 

14__all__ = ["PanguProvider"] 

15 

16PANGU_DEFAULT_BASE = "https://pangu-api.huaweicloud.com/v1" 

17PANGU_DEFAULT_MODEL = "pangu-4" 

18 

19 

20class PanguProvider(OpenAIProvider): 

21 """华为盘古大模型 Provider — 基于 OpenAI 兼容协议。 

22 

23 支持模型: pangu-4, pangu-3.1, pangu-code, pangu-vision 等。 

24 

25 环境变量: 

26 PANGU_API_KEY: 华为云 API Key 

27 PANGU_BASE_URL: API 端点，默认 https://pangu-api.huaweicloud.com/v1 

28 """ 

29 

30 def __init__( 

31 self, 

32 model: str = PANGU_DEFAULT_MODEL, 

33 api_key: str = "", 

34 base_url: str = "", 

35 timeout: float = 120.0, 

36 ): 

37 import os 

38 

39 resolved_base = base_url or os.getenv( 

40 "PANGU_BASE_URL", PANGU_DEFAULT_BASE 

41 ) 

42 resolved_key = api_key or os.getenv("PANGU_API_KEY", "") 

43 

44 super().__init__( 

45 model=model, 

46 api_key=resolved_key, 

47 base_url=resolved_base, 

48 ) 

49 self._timeout = timeout 

50 

51 @property 

52 def provider_name(self) -> str: 

53 return "pangu" 

54 

55 def chat(self, *args: Any, **kwargs: Any): 

56 kwargs.setdefault("timeout", self._timeout) 

57 return super().chat(*args, **kwargs) 

58 

59 async def achat(self, *args: Any, **kwargs: Any): 

60 kwargs.setdefault("timeout", self._timeout) 

61 return await super().achat(*args, **kwargs)