添加阿里绿网云盾 API

konabot/common/apis/ali_content_safety.py

@@ -0,0 +1,90 @@
+import asyncio
+import json
+
+from alibabacloud_green20220302.client import Client as AlibabaGreenClient
+from alibabacloud_green20220302.models import TextModerationPlusRequest
+from alibabacloud_tea_openapi.models import Config as AlibabaTeaConfig
+from loguru import logger
+from pydantic import BaseModel
+
+import nonebot
+
+
+class AlibabaGreenPluginConfig(BaseModel):
+    module_aligreen_enable: bool = False
+    module_aligreen_access_key_id: str = ""
+    module_aligreen_access_key_secret: str = ""
+    module_aligreen_region_id: str = "cn-shenzhen"
+    module_aligreen_endpoint: str = "green-cip.cn-shenzhen.aliyuncs.com"
+    module_aligreen_service: str = "llm_query_moderation"
+
+
+class AlibabaGreen:
+    _client: AlibabaGreenClient | None = None
+    _config: AlibabaGreenPluginConfig | None = None
+
+    @staticmethod
+    def get_client() -> AlibabaGreenClient:
+        assert AlibabaGreen._client is not None
+        return AlibabaGreen._client
+
+    @staticmethod
+    def get_config() -> AlibabaGreenPluginConfig:
+        assert AlibabaGreen._config is not None
+        return AlibabaGreen._config
+
+    @staticmethod
+    def init():
+        config = nonebot.get_plugin_config(AlibabaGreenPluginConfig)
+        AlibabaGreen._config = config
+        if not config.module_aligreen_enable:
+            logger.info("该环境未启用阿里内容审查，跳过初始化")
+            return
+        AlibabaGreen._client = AlibabaGreenClient(AlibabaTeaConfig(
+            access_key_id=config.module_aligreen_access_key_id,
+            access_key_secret=config.module_aligreen_access_key_secret,
+            connect_timeout=10000,
+            read_timeout=3000,
+            region_id=config.module_aligreen_region_id,
+            endpoint=config.module_aligreen_endpoint,
+        ))
+
+    @staticmethod
+    def _detect_sync(content: str) -> bool:
+        if not AlibabaGreen.get_config().module_aligreen_enable:
+            logger.debug("该环境未启用阿里内容审查，直接跳过")
+            return True
+
+        client = AlibabaGreen.get_client()
+        try:
+            response = client.text_moderation_plus(TextModerationPlusRequest(
+                service=AlibabaGreen.get_config().module_aligreen_service,
+                service_parameters=json.dumps({
+                    "content": content,
+                }),
+            ))
+            if response.status_code == 200:
+                result = response.body
+                logger.debug(f"检测违规内容 API 调用成功：{result}")
+                risk_level: str = result.data.risk_level or "none"
+                if risk_level == "high":
+                    return False
+                return True
+            logger.error(f"检测违规内容 API 调用失败：{response}")
+            return True
+        except Exception as e:
+            logger.error("检测违规内容 API 调用失败")
+            logger.exception(e)
+            return True
+
+    @staticmethod
+    async def detect(content: str) -> bool:
+        return await asyncio.to_thread(AlibabaGreen._detect_sync, content)
+
+
+driver = nonebot.get_driver()
+
+@driver.on_startup
+async def _():
+    AlibabaGreen.init()
+

konabot/common/llm/__init__.py

@@ -59,6 +59,9 @@ def get_llm(llm_model: str | None = None):
     if llm_model is None:
         llm_model = llm_config.default_llm
     if llm_model not in llm_config.llms:
-        raise NotImplementedError("LLM 未配置，该功能无法使用")
+        if llm_config.default_llm in llm_config.llms:
+            logger.warning(f"[LLM] 需求的 LLM 不存在，回退到默认模型 REQUIRED={llm_model}")
+            return llm_config.llms[llm_config.default_llm]
+        raise NotImplementedError("[LLM] LLM 未配置，该功能无法使用")
     return llm_config.llms[llm_model]