From 169492a6dbddf42859c6a7f83b062da1283ed753 Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Wed, 5 Aug 2026 01:56:59 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=8C=E6=88=90=20issue=20#44=20?= =?UTF-8?q?=E2=91=A3=20=E6=9C=AC=E5=9C=B0=2070B=20=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E6=8E=A5=E5=85=A5=E4=B8=8E=E6=8E=A8=E7=90=86=E5=B0=81=E8=A3=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/llm-gateway/__init__.py | 5 + core/llm-gateway/backends.py | 114 +++++++++++++++ .../llm-gateway/config/local70b.template.yaml | 18 +++ core/llm-gateway/tests/test_local70b.py | 136 ++++++++++++++++++ 4 files changed, 273 insertions(+) create mode 100644 core/llm-gateway/backends.py create mode 100644 core/llm-gateway/config/local70b.template.yaml create mode 100644 core/llm-gateway/tests/test_local70b.py diff --git a/core/llm-gateway/__init__.py b/core/llm-gateway/__init__.py index a04f83e..3ddcd08 100644 --- a/core/llm-gateway/__init__.py +++ b/core/llm-gateway/__init__.py @@ -17,6 +17,8 @@ 分解，配套评测报告脚本 evaluate_hallucination.py）。 - gateway 混合网关主编排（EPIC #6 主体）：路由 → 生成 → 溯源校验 → DLP 出站防线，端到端闭环。 +- backends 本地 70B 推理后端实现（Issue #44 完成交付）：OpenAI 兼容 + vLLM/TGI 接入、参数化、dry-run 兼容，数据不出厂。 测试：`python -m unittest discover -s tests -v`（在 core/llm-gateway 目录下执行）。 """ @@ -53,6 +55,7 @@ from .gateway import ( LLMGateway, LocalBackend, ) +from .backends import Local70BBackend __all__ = [ # dlp @@ -65,5 +68,7 @@ __all__ = [ "GuardVerdict", "HallucinationGuard", # gateway "InferenceBackend", "LocalBackend", "CloudBackend", + # backends + "Local70BBackend", "GatewayResult", "LLMGateway", ] diff --git a/core/llm-gateway/backends.py b/core/llm-gateway/backends.py new file mode 100644 index 0000000..1715cce --- /dev/null +++ b/core/llm-gateway/backends.py @@ -0,0 +1,114 @@ +# -*- coding: utf-8 -*- +"""推理后端实现 —— 本地 70B 模型接入与推理封装（issue #44）。 + +在 `gateway.InferenceBackend` 抽象之上交付**真实可用的本地后端**： +- OpenAI 兼容接口（vLLM / TGI 等本地推理服务，`/v1/chat/completions`）， + 仅用标准库 urllib，无第三方依赖； +- 参数化：endpoint / model / timeout / max_tokens / temperature / context 引用注入； +- **数据不出厂**（PRD 5.4）：敏感/核心内容走本地后端，云端仅接收脱敏内容； +- 未配置 endpoint 时进入 dry-run 占位模式（保持与旧 LocalBackend 一致的 + 可测试行为，供端到端演示与联调）。 + +业务代码只依赖 `gateway.InferenceBackend.generate(prompt, context)`， +切换后端 = 换实现（见 `LLMGateway(local=...)`）。 +""" +from __future__ import annotations + +import json +import time +import urllib.request +from typing import Optional, Sequence + +from .gateway import InferenceBackend + + +class Local70BBackend(InferenceBackend): + """本地 70B 推理后端（OpenAI 兼容 vLLM/TGI，参数化）。""" + + name = "local-70b" + + def __init__( + self, + endpoint: str = "", + model: str = "iaop-local-70b", + timeout_seconds: float = 60.0, + max_tokens: int = 1024, + temperature: float = 0.1, + echo_context: bool = True, + ) -> None: + self.endpoint = (endpoint or "").rstrip("/") + self.model = model + self.timeout = float(timeout_seconds) + self.max_tokens = int(max_tokens) + self.temperature = float(temperature) + self.echo_context = echo_context + + # ------------------------------------------------------------------ + def generate(self, prompt: str, context: Sequence[str]) -> str: + """根据 prompt 与 RAG 上下文生成回答。 + + - 未配置 endpoint：dry-run 占位（回显 prompt 前 40 字符 + 来源引用）； + - 已配置：调用本地 OpenAI 兼容服务（/v1/chat/completions）。 + """ + 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"本地推理服务响应格式异常: {str(body)[:200]}") + + # ------------------------------------------------------------------ + def _system_prompt(self, context: Sequence[str]) -> str: + """把 RAG 引用注入 system 提示（引用溯源，PRD 5.4）。""" + 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"[本地70B占位] {prompt[:40]}" + if self.echo_context: + for i, src in enumerate(context[:3], 1): + head += f"\n[来源: {src}]" + return head + + def _post_json(self, path: str, payload: dict) -> dict: + """向后端推理服务发起 JSON POST（标准库 urllib）。""" + url = self.endpoint + path + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, data=data, + headers={"Content-Type": "application/json"}) + 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: + """后端健康信息（本地推理服务可探测 /health）。""" + base = { + "backend": self.name, "model": self.model, + "endpoint": self.endpoint or "(dry-run)", + } + if not self.endpoint: + base["status"] = "dry-run" + return base + try: + started = time.monotonic() + with urllib.request.urlopen( + self.endpoint + "/health", timeout=self.timeout) as resp: + base["status"] = "ok" if resp.status == 200 else f"http-{resp.status}" + base["latency_ms"] = round((time.monotonic() - started) * 1000, 2) + except Exception as exc: # noqa: BLE001 - 健康探测失败仅记录 + base["status"] = f"error: {exc}" + return base diff --git a/core/llm-gateway/config/local70b.template.yaml b/core/llm-gateway/config/local70b.template.yaml new file mode 100644 index 0000000..53139a8 --- /dev/null +++ b/core/llm-gateway/config/local70b.template.yaml @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +# 模板「本地 70B 推理后端」配置资产示例：ti-cl4（Template-Ti 一期）。 +# +# 说明（issue #44 / PRD 5.4「④ LLM 网关 + RAG」）： +# - 本地 70B（vLLM/TGI 等 OpenAI 兼容服务）承载敏感/核心内容，数据不出厂； +# - endpoint 为空时进入 dry-run 占位模式（联调/演示）； +# - 切换/新增本地推理服务只改本文件，业务代码零改动。 +template: ti-cl4 +version: 1.0.0 + +local70b: + # 本地推理服务地址（OpenAI 兼容；空 = dry-run） + endpoint: "http://10.20.0.30:8000/v1" + model: "iaop-local-70b" + timeout_seconds: 60 + max_tokens: 1024 + temperature: 0.1 + echo_context: true # 是否在输出回显 RAG 来源引用（溯源） diff --git a/core/llm-gateway/tests/test_local70b.py b/core/llm-gateway/tests/test_local70b.py new file mode 100644 index 0000000..04d15c5 --- /dev/null +++ b/core/llm-gateway/tests/test_local70b.py @@ -0,0 +1,136 @@ +# -*- coding: utf-8 -*- +"""本地 70B 推理后端测试（issue #44）。 + +覆盖： +1. 参数化：endpoint / model / timeout / max_tokens / temperature / echo_context； +2. dry-run（未配置 endpoint）：占位输出 + 来源回显，与旧 LocalBackend 兼容； +3. OpenAI 兼容调用：mock /v1/chat/completions 响应 → 提取 answer； +4. 响应格式异常 → RuntimeError； +5. 与 LLMGateway 组合：本地后端承载敏感内容（数据不出厂）。 +""" +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 Local70BBackend # noqa: E402 +from llm_gateway.gateway import LLMGateway # noqa: E402 +from llm_gateway.prompts import PromptRegistry # noqa: E402 + +PROMPTS_CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "prompts.template.yaml", +) + + +def make_gateway(**kwargs) -> LLMGateway: + """构建带提示词版本库的网关（默认本地 70B 后端）。""" + return LLMGateway( + local=Local70BBackend(), + prompts=PromptRegistry.from_template_config(PROMPTS_CONFIG), + **kwargs, + ) + +CONFIG = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "local70b.template.yaml", +) + + +def load_config(): + import yaml + with open(CONFIG, "r", encoding="utf-8") as fh: + return yaml.safe_load(fh) or {} + + +class TestParameterization(unittest.TestCase): + """参数化提取与配置资产。""" + + def test_defaults(self): + b = Local70BBackend() + self.assertEqual(b.endpoint, "") + self.assertEqual(b.model, "iaop-local-70b") + self.assertEqual(b.max_tokens, 1024) + self.assertEqual(b.temperature, 0.1) + + def test_config_asset_parses(self): + cfg = load_config()["local70b"] + b = Local70BBackend(**{k: v for k, v in cfg.items()}) + self.assertEqual(b.endpoint, "http://10.20.0.30:8000/v1") + self.assertEqual(b.model, "iaop-local-70b") + + def test_name(self): + self.assertEqual(Local70BBackend().name, "local-70b") + + +class TestDryRun(unittest.TestCase): + """未配置 endpoint：占位 + 来源回显（兼容旧行为）。""" + + def setUp(self): + self.b = Local70BBackend() # endpoint 默认空 + + def test_dry_run_with_context(self): + out = self.b.generate("请解释炉温报警", ["SOP-CL-001", "工艺规范"]) + self.assertIn("[本地70B占位]", out) + self.assertIn("[来源: SOP-CL-001]", out) + + def test_dry_run_no_echo(self): + b = Local70BBackend(echo_context=False) + out = b.generate("hi", ["s1"]) + self.assertNotIn("[来源", out) + + def test_health_dry_run(self): + health = self.b.health() + self.assertEqual(health["status"], "dry-run") + + +class TestOpenAICompat(unittest.TestCase): + """OpenAI 兼容 /v1/chat/completions 调用。""" + + def setUp(self): + self.b = Local70BBackend(endpoint="http://local:8000/v1") + + def test_generate_extracts_answer(self): + fake = {"choices": [{"message": {"content": "炉温偏高，建议降氯气流量"}}]} + with mock.patch.object(self.b, "_post_json", return_value=fake) as post: + out = self.b.generate("炉温异常", ["SOP-CL-001"]) + post.assert_called_once() + path, payload = post.call_args[0] + self.assertEqual(path, "/v1/chat/completions") + self.assertEqual(payload["model"], "iaop-local-70b") + # system 提示注入 RAG 引用（溯源） + self.assertIn("SOP-CL-001", payload["messages"][0]["content"]) + self.assertEqual(out, "炉温偏高，建议降氯气流量") + + def test_bad_response_raises(self): + with mock.patch.object(self.b, "_post_json", return_value={"choices": []}): + with self.assertRaises(RuntimeError): + self.b.generate("x", []) + + def test_health_ok(self): + with mock.patch("urllib.request.urlopen") as urlopen: + resp = mock.MagicMock() + resp.status = 200 + urlopen.return_value.__enter__ = mock.MagicMock(return_value=resp) + urlopen.return_value.__exit__ = mock.MagicMock(return_value=False) + health = self.b.health() + self.assertEqual(health["status"], "ok") + self.assertEqual(health["backend"], "local-70b") + + +class TestGatewayIntegration(unittest.TestCase): + """与 LLMGateway 组合：本地后端承载敏感内容（数据不出厂）。""" + + def test_gateway_with_local70b(self): + gw = make_gateway() + result = gw.ask("炉温是多少", rag_context=["工艺规范"]) + self.assertIn("本地70B", result.answer) + # 敏感内容路由本地（CLF 工艺参数 → local） + self.assertEqual(result.route.target, "local") + + +if __name__ == "__main__": + unittest.main()