From c5a7a0ee5273fd6d3d81bdd47d65124285811f62 Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Wed, 5 Aug 2026 02:37:41 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#45=20?= =?UTF-8?q?=E2=91=A3=20=E4=BA=91=E7=AB=AF=20API=EF=BC=88Qwen/DeepSeek?= =?UTF-8?q?=EF=BC=89=E6=8E=A5=E5=85=A5=E4=B8=8E=E5=AE=89=E5=85=A8=E7=BD=91?= =?UTF-8?q?=E5=85=B3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/llm-gateway/__init__.py | 3 +- core/llm-gateway/backends.py | 98 ++++++++++++- .../llm-gateway/config/cloudapi.template.yaml | 19 +++ core/llm-gateway/tests/test_cloudapi.py | 131 ++++++++++++++++++ 4 files changed, 249 insertions(+), 2 deletions(-) create mode 100644 core/llm-gateway/config/cloudapi.template.yaml create mode 100644 core/llm-gateway/tests/test_cloudapi.py diff --git a/core/llm-gateway/__init__.py b/core/llm-gateway/__init__.py index 3ddcd08..e76bb6b 100644 --- a/core/llm-gateway/__init__.py +++ b/core/llm-gateway/__init__.py @@ -55,7 +55,7 @@ from .gateway import ( LLMGateway, LocalBackend, ) -from .backends import Local70BBackend +from .backends import CloudApiBackend, Local70BBackend __all__ = [ # dlp @@ -70,5 +70,6 @@ __all__ = [ "InferenceBackend", "LocalBackend", "CloudBackend", # backends "Local70BBackend", + "CloudApiBackend", "GatewayResult", "LLMGateway", ] diff --git a/core/llm-gateway/backends.py b/core/llm-gateway/backends.py index 1715cce..ecccdfd 100644 --- a/core/llm-gateway/backends.py +++ b/core/llm-gateway/backends.py @@ -15,9 +15,10 @@ from __future__ import annotations import json +import os import time import urllib.request -from typing import Optional, Sequence +from typing import Callable, Optional, Sequence from .gateway import InferenceBackend @@ -112,3 +113,98 @@ class Local70BBackend(InferenceBackend): except Exception as exc: # noqa: BLE001 - 健康探测失败仅记录 base["status"] = f"error: {exc}" return base + + +class CloudApiBackend(InferenceBackend): + """云端 API 推理后端（Qwen / DeepSeek 等 OpenAI 兼容）—— issue #45。 + + **安全网关约束（PRD 5.4）**： + - 仅接收 **DLP 放行**的脱敏/通用内容（上游 `LLMGateway` 主编排出站检查 + + cloud 分支输出 DLP 复查）； + - API Key 从**环境变量**读取（`api_key_env`），不硬编码、不落日志； + - 可选 `safety_checker` 出站复查钩子（fail-closed：复查拒绝 → 拦截占位， + 不调用上游）。 + """ + + name = "cloud-api" + + def __init__( + self, + endpoint: str = "", + api_key_env: str = "", + model: str = "deepseek-chat", + timeout_seconds: float = 60.0, + max_tokens: int = 1024, + temperature: float = 0.1, + safety_checker: Optional[Callable[[str], bool]] = None, + ) -> None: + self.endpoint = (endpoint or "").rstrip("/") + self.api_key_env = api_key_env + self.model = model + self.timeout = float(timeout_seconds) + self.max_tokens = int(max_tokens) + self.temperature = float(temperature) + # 出站安全复查：返回 False 即拦截（fail-closed） + self.safety_checker = safety_checker + self._api_key = os.environ.get(api_key_env, "") if api_key_env else "" + + # ------------------------------------------------------------------ + def generate(self, prompt: str, context: Sequence[str]) -> str: + """生成回答。安全网关：safety_checker 拒绝 → 拦截占位，不调用上游。""" + if self.safety_checker is not None and not self.safety_checker(prompt): + return "[云端安全网关拦截] 出站复查未通过，已拦截（数据不出厂）。" + + if not self.endpoint: + return self._dry_run(prompt, context) + + payload = { + "model": self.model, + "messages": [ + {"role": "system", "content": self._system_prompt(context)}, + {"role": "user", "content": prompt}, + ], + "max_tokens": self.max_tokens, + "temperature": self.temperature, + } + body = self._post_json("/v1/chat/completions", payload) + try: + return body["choices"][0]["message"]["content"] + except (KeyError, IndexError, TypeError): + raise RuntimeError( + f"云端 API 响应格式异常: {str(body)[:200]}") + + # ------------------------------------------------------------------ + def _system_prompt(self, context: Sequence[str]) -> str: + refs = "\n".join(f"- {c}" for c in (context or [])) + base = "你是工业 AI 助手。回答须基于给定资料并标注来源。" + return f"{base}\n参考资料：\n{refs}" if refs else base + + def _dry_run(self, prompt: str, context: Sequence[str]) -> str: + head = f"[云端API占位] {prompt[:40]}" + for i, src in enumerate(context[:3], 1): + head += f"\n[来源: {src}]" + if self.safety_checker is not None: + head += "\n[安全网关: 已复查放行]" + return head + + def _post_json(self, path: str, payload: dict) -> dict: + """向后端推理服务发起 JSON POST（Bearer 认证，Key 来自环境变量）。""" + url = self.endpoint + path + data = json.dumps(payload).encode("utf-8") + headers = {"Content-Type": "application/json"} + if self._api_key: + headers["Authorization"] = f"Bearer {self._api_key}" + req = urllib.request.Request(url, data=data, headers=headers) + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + raw = resp.read().decode("utf-8") + return json.loads(raw) if raw else {} + + def health(self) -> dict: + """后端健康信息（含安全网关状态，不含密钥）。""" + return { + "backend": self.name, "model": self.model, + "endpoint": self.endpoint or "(dry-run)", + "api_key_configured": bool(self._api_key), + "safety_checker": self.safety_checker is not None, + "status": "dry-run" if not self.endpoint else "configured", + } diff --git a/core/llm-gateway/config/cloudapi.template.yaml b/core/llm-gateway/config/cloudapi.template.yaml new file mode 100644 index 0000000..d2dfdb7 --- /dev/null +++ b/core/llm-gateway/config/cloudapi.template.yaml @@ -0,0 +1,19 @@ +# -*- coding: utf-8 -*- +# 模板「云端 API 推理后端」配置资产示例：ti-cl4（Template-Ti 一期）。 +# +# 说明（issue #45 / PRD 5.4「④ LLM 网关 + RAG」）： +# - 云端 API（Qwen DashScope / DeepSeek OpenAI 兼容）仅承载**脱敏/通用**内容， +# 敏感数据本地闭环（数据不出厂，出站由 DLP + 安全网关复查）； +# - API Key 从环境变量读取（api_key_env），禁止硬编码/落日志； +# - 换模型/换服务只改本文件，业务代码零改动。 +template: ti-cl4 +version: 1.0.0 + +cloudapi: + provider: deepseek # qwen | deepseek | openai + endpoint: "https://api.deepseek.com/v1" + api_key_env: DEEPSEEK_API_KEY # 从环境变量读取，勿在此填明文 + model: "deepseek-chat" + timeout_seconds: 60 + max_tokens: 1024 + temperature: 0.1 diff --git a/core/llm-gateway/tests/test_cloudapi.py b/core/llm-gateway/tests/test_cloudapi.py new file mode 100644 index 0000000..da3a2f2 --- /dev/null +++ b/core/llm-gateway/tests/test_cloudapi.py @@ -0,0 +1,131 @@ +# -*- coding: utf-8 -*- +"""云端 API 推理后端 + 安全网关测试（issue #45）。 + +覆盖： +1. 参数化与配置资产（provider/endpoint/model/api_key_env）； +2. API Key 从环境变量读取（不硬编码、health 不泄露密钥）； +3. dry-run（未配置 endpoint）占位输出； +4. OpenAI 兼容调用：Bearer 认证头 + mock 响应提取； +5. 安全网关：safety_checker 拒绝 → 拦截占位、不调用上游； +6. 坏响应 → RuntimeError。 +""" +import os +import sys +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from llm_gateway.backends import CloudApiBackend # noqa: E402 + +CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "cloudapi.template.yaml", +) + + +class TestParameterization(unittest.TestCase): + """参数化与配置资产。""" + + def test_config_asset_parses(self): + import yaml + with open(CONFIG, "r", encoding="utf-8") as fh: + cfg = yaml.safe_load(fh)["cloudapi"] + b = CloudApiBackend(**{k: v for k, v in cfg.items() if k != "provider"}) + self.assertEqual(b.endpoint, "https://api.deepseek.com/v1") + self.assertEqual(b.model, "deepseek-chat") + self.assertEqual(b.api_key_env, "DEEPSEEK_API_KEY") + + def test_api_key_from_env(self): + with mock.patch.dict(os.environ, {"MY_KEY": "sk-secret-123"}): + b = CloudApiBackend(endpoint="http://x/v1", api_key_env="MY_KEY") + self.assertEqual(b._api_key, "sk-secret-123") + + def test_health_does_not_leak_key(self): + with mock.patch.dict(os.environ, {"K": "sk-secret-123"}): + b = CloudApiBackend(api_key_env="K") + health = b.health() + self.assertEqual(health["api_key_configured"], True) + self.assertNotIn("sk-secret", str(health)) + + def test_name(self): + self.assertEqual(CloudApiBackend().name, "cloud-api") + + +class TestDryRun(unittest.TestCase): + """未配置 endpoint：占位输出。""" + + def setUp(self): + self.b = CloudApiBackend() + + def test_dry_run_placeholder(self): + out = self.b.generate("海绵钛是什么", ["公开资料"]) + self.assertIn("[云端API占位]", out) + self.assertIn("[来源: 公开资料]", out) + + +class TestOpenAICompat(unittest.TestCase): + """OpenAI 兼容调用（Bearer 认证）。""" + + def setUp(self): + with mock.patch.dict(os.environ, {"DK": "sk-abc"}): + self.b = CloudApiBackend( + endpoint="https://api.deepseek.com/v1", api_key_env="DK") + + def test_generate_with_bearer_auth(self): + fake = {"choices": [{"message": {"content": "海绵钛是钛的一种形态"}}]} + with mock.patch.object(self.b, "_post_json", return_value=fake) as post: + out = self.b.generate("海绵钛是什么", ["公开资料"]) + path, payload = post.call_args[0] + self.assertEqual(path, "/v1/chat/completions") + self.assertEqual(payload["model"], "deepseek-chat") + self.assertEqual(out, "海绵钛是钛的一种形态") + + def test_bearer_header_sent(self): + # 验证 _post_json 实际携带 Authorization: Bearer + opened = [] + def fake_urlopen(req, timeout=None): + opened.append(req) + resp = mock.MagicMock() + resp.read.return_value = b'{"choices":[]}' + cm = mock.MagicMock() + cm.__enter__.return_value = resp + return cm + with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen): + self.b._post_json("/v1/chat/completions", {"model": "x"}) + self.assertEqual(opened[0].get_header("Authorization"), "Bearer sk-abc") + + def test_bad_response_raises(self): + with mock.patch.object(self.b, "_post_json", return_value={}): + with self.assertRaises(RuntimeError): + self.b.generate("x", []) + + +class TestSafetyGateway(unittest.TestCase): + """安全网关：safety_checker 复查（fail-closed）。""" + + def test_checker_reject_blocks(self): + b = CloudApiBackend(endpoint="http://x/v1", + safety_checker=lambda p: "脱敏" in p) + with mock.patch.object(b, "_post_json") as post: + out = b.generate("含敏感内容", []) + post.assert_not_called() # 不调用上游 + self.assertIn("拦截", out) + + def test_checker_allow_passes(self): + b = CloudApiBackend(endpoint="http://x/v1", + safety_checker=lambda p: "脱敏" in p) + fake = {"choices": [{"message": {"content": "ok"}}]} + with mock.patch.object(b, "_post_json", return_value=fake) as post: + out = b.generate("脱敏后的公开问题", []) + post.assert_called_once() + self.assertEqual(out, "ok") + + def test_health_reports_safety(self): + b = CloudApiBackend(safety_checker=lambda p: True) + self.assertTrue(b.health()["safety_checker"]) + + +if __name__ == "__main__": + unittest.main()