Commit 4aed2229 authored by Data Governance Dev's avatar Data Governance Dev

feat(web3): AI 生成正则 + Python re 测试 + 规则弹窗保存按钮

LLM 接入(P9 plan 阶段)
- 后端:
  - backend/core/llm.py 从 web/core/llm.py 拷贝,default_config_path() 改读 web3/backend/configs/llm.yaml
  - backend/core/ai_regex.py:强制走 LLM(删规则库),LLM 返回用 re.compile 校验,编译失败回弹修复,最多 5 次重试
  - backend/routers/ai.py:注册 POST /api/ai/regex
  - backend/configs/llm.example.yaml:多 provider 示例模板(真实 llm.yaml 已 gitignore)
  - backend/app.py:注册 ai_router
  - backend/requirements.txt:+anthropic +openai +pyyaml
- 前端:
  - src/api/ai.js:genRegex(desc)
  - src/components/RuleModal.vue:删前端 mock 改调后端

测试改走 Python re(避免 JS RegExp 方言差异)
- backend/core/ai_regex.py:+test_regex(regex, value) → {ok, match, error, groups}
- backend/routers/ai.py:+POST /api/regex/test
- src/api/ai.js:+testRegex(regex, value)
- src/components/RuleModal.vue:testRegex() 改调后端,捕获组带回前端展示

规则弹窗保存按钮
- src/components/RuleModal.vue:emit 'save' + 取消/保存两按钮 + onSave() + hasValidRules
  (必须 ≥1 条规则 + desc/regex 都非空才允许保存)
- watch 监听 [open, field.key],每次打开从父组件重新同步
- src/components/TaskModal.vue:onRuleSave 回写 field.rules(徽章数字刷新)
- TaskModal onSave() payload 加 fields: [{key, en, cn, rules, show_default}](P4 落库用)

端到端验证
- 身份证/URL/UUID/产品型号 多 desc 都通过 LLM 生成 + Python re 校验
- 7 例 /api/regex/test 覆盖:真身份证、短串、捕获组日期、分隔符错、空 value、语法错、空 regex

踩坑
- 规则库'数字'关键词过宽易误命中,按用户意见删整段强制走 LLM
- curl 中文 JSON body 在 Git Bash 转义破坏,用 --data @file 或 printf
- python web3/backend/start.py 报 ModuleNotFoundError,要 python -m web3.backend.start
- Vue3 const 不 hoist,hasValidRules 必须放在 ruleList 声明之后
parent 130b8398
......@@ -50,6 +50,7 @@ app.add_middleware(
)
from web3.backend.routers.ai import router as ai_router # noqa: E402
from web3.backend.routers.connection_presets import router as connection_presets_router # noqa: E402
from web3.backend.routers.db import router as db_router # noqa: E402
from web3.backend.routers.task_groups import router as task_groups_router # noqa: E402
......@@ -58,7 +59,8 @@ app.include_router(task_groups_router, prefix="/api")
app.include_router(tasks_router, prefix="/api")
app.include_router(db_router, prefix="/api")
app.include_router(connection_presets_router, prefix="/api")
logger.info("已注册路由:/api/task-groups, /api/tasks, /api/connect/*, /api/connection-presets")
app.include_router(ai_router, prefix="/api")
logger.info("已注册路由:/api/task-groups, /api/tasks, /api/connect/*, /api/connection-presets, /api/ai/*")
@app.get("/api/health")
......
# ============================================================
# LLM 配置文件示例 — web3 版本(2026-08-20)
# ============================================================
# 用法:复制本文件为 llm.yaml,按需填入真实 key。
# 真实 llm.yaml 已在 .gitignore,不会上传到仓库。
# ============================================================
# ── 方式 1:MiniMax(推荐本项目用户使用) ──
provider: minimax
api_key: sk-请填入你的_MiniMax_API_Key
base_url: https://api.minimaxi.com/anthropic
model: MiniMax-Text-01
# ── 方式 2:Anthropic Claude ──
# provider: anthropic
# api_key: sk-ant-xxxxx
# model: claude-sonnet-5
# (base_url 留空,用 Anthropic 官方)
# ── 方式 3:OpenAI ──
# provider: openai
# api_key: sk-xxxxx
# model: gpt-4o
# base_url: https://api.openai.com/v1
# ── 方式 4:自部署 / 第三方代理(OpenAI 兼容) ──
# provider: openai
# api_key: 自定义
# base_url: https://your-proxy.com/v1
# model: your-model-name
# 生成参数(通用)
max_tokens: 1024
temperature: 0.2
timeout: 30
\ No newline at end of file
"""AI 生成正则服务(包装 web3.backend.core.llm)
强制走 LLM,不做规则库兜底(2026-08-20 用户决策):
- 每次请求都过 LLM,让模型能见到真实描述(持续学习用户表述)
- 拿到正则后用 re.compile 在 Python 里真实校验
- 编译失败则把错误反馈回 LLM 让它修复,最多 5 次
降级策略:
- LLM 不可用(无 key / 网络挂)→ 直接返回空串 + 提示文案,不抛异常
- 5 次重试全部失败 → 返回空串 + 失败提示,由前端展示
"""
from __future__ import annotations
import re
from web3.backend.core.llm import LLMUnavailable, get_llm_client
from web3.backend._logging import get_logger
logger = get_logger("backend.ai_regex")
# ── LLM Prompt ────────────────────────────────────────────
# 系统提示词必须明确:
# - 你是 Python re 正则专家
# - 只输出一行正则字符串(不要 markdown、解释、前后缀)
# - 严格匹配("必须是 18 位" → 长度恰好 18,不要写 * 或 +)
# - 失败时(带前次错误信息)只输出修复后的正则,不要任何多余内容
_SYSTEM_PROMPT = """你是 Python 正则表达式专家。"""
_USER_TEMPLATE_INITIAL = """请把以下规则描述转换成一个 Python 正则表达式(re 语法)。
规则描述:「{desc}」
要求:
1. 只输出一行正则字符串,不要任何解释、不要 Markdown 代码块、不要前后缀
2. 严格匹配("必须是 N 位" → 长度恰好 N,不要写成 {{N,}} 或 *)
3. 中国大陆场景默认值:手机 11 位、邮编 6 位、身份证 18 位末位可为 X、日期 yyyy-MM-dd
4. 用 ^ 与 $ 锚定首尾(除非描述里明确说"包含")
5. 中文用 \\u4e00-\\u9fff 这种 Unicode 区间,不要写汉字本身
只输出一行正则字符串。"""
_USER_TEMPLATE_FIX = """上一次生成的正则编译失败。
规则描述:「{desc}」
上次生成的正则:`{bad_regex}`
Python 报的错误:`{error_msg}`
请输出修复后的正则字符串。
要求:
1. 只输出一行正则字符串,不要任何解释、不要 Markdown 代码块、不要前后缀
2. 修掉上面的语法错误(注意字符类、转义、括号配对)
3. 严格匹配("必须是 N 位" → 长度恰好 N)
4. 用 ^ 与 $ 锚定首尾
只输出一行修复后的正则字符串。"""
# ── 从 LLM 返回中提取正则 ─────────────────────────────────
# LLM 偶尔会包裹一层 ```python ... ``` 或加引号,这里做容错提取。
_REGEX_FENCE = re.compile(r"^```(?:python|regex|re)?\s*\n?(.*?)\n?```$", re.DOTALL)
def _extract_regex(text: str) -> str:
"""从 LLM 返回中提取「那一行正则」。
容忍:
- markdown 代码块包裹
- 开头/结尾的引号
- 前后的"答案:xxx"说明文字(取第一个像正则的行)
"""
if not text:
return ""
text = text.strip()
# 1) 代码块包裹
m = _REGEX_FENCE.match(text)
if m:
return m.group(1).strip()
# 2) 首尾引号
if (text.startswith('"') and text.endswith('"')) or \
(text.startswith("'") and text.endswith("'")):
return text[1:-1].strip()
# 3) 多行:取第一非空行
for line in text.splitlines():
line = line.strip()
if line:
return line
return ""
def _validate_regex(regex: str) -> tuple[bool, str]:
"""用 re.compile 校验;返回 (ok, error_msg)"""
if not regex:
return False, "空字符串"
try:
re.compile(regex)
return True, ""
except re.error as e:
return False, f"{type(e).__name__}: {e}"
def test_regex(regex: str, value: str) -> dict:
"""用 Python re 校验 value 是否匹配 regex。
返回 dict:
- ok: bool 编译是否成功
- match: bool 值是否匹配(regex/re.search 语义)
- error: str 编译/匹配错误信息(成功时空串)
- groups: list 捕获组内容(无捕获组时空 list)
"""
if not regex:
return {"ok": False, "match": False, "error": "正则为空", "groups": []}
try:
compiled = re.compile(regex)
except re.error as e:
return {"ok": False, "match": False, "error": f"编译失败: {type(e).__name__}: {e}", "groups": []}
if value is None or value == "":
return {"ok": True, "match": False, "error": "测试值为空", "groups": []}
m = compiled.search(value)
if not m:
return {"ok": True, "match": False, "error": "", "groups": []}
# 把捕获组也带回去(方便 UI 展示分组提取)
try:
groups = [g if g is not None else "" for g in m.groups()]
except Exception:
groups = []
return {"ok": True, "match": True, "error": "", "groups": groups}
# ── 主入口 ─────────────────────────────────────────────
def gen_regex(desc: str) -> tuple[str, str]:
"""强制调 LLM 生成 Python 正则。
返回 (regex, note):
- regex 非空 = 成功;空 = 失败
- note 给前端弹窗用的提示文案
流程:LLM 调用 + re.compile 校验 → 失败最多重试 5 次 → 失败空串
"""
desc = (desc or "").strip()
if not desc:
return "", "规则描述为空"
client = get_llm_client()
if not client.available:
return "", "LLM 未配置 API Key,请联系管理员配置或手动填写"
bad_regex = ""
error_msg = ""
last_regex = ""
MAX_TRIES = 5
for attempt in range(1, MAX_TRIES + 1):
try:
if attempt == 1:
prompt = _USER_TEMPLATE_INITIAL.format(desc=desc)
else:
prompt = _USER_TEMPLATE_FIX.format(
desc=desc, bad_regex=bad_regex, error_msg=error_msg
)
text = client.complete(prompt, system=_SYSTEM_PROMPT)
last_regex = _extract_regex(text)
ok, err = _validate_regex(last_regex)
if ok:
logger.info(f"AI 正则:LLM 第 {attempt} 次成功 desc={desc!r} → {last_regex!r}")
return last_regex, f"已用 LLM 生成(第 {attempt} 次)"
# 校验失败 → 把错误反馈给下一轮
bad_regex = last_regex
error_msg = err
logger.warning(
f"AI 正则:LLM 第 {attempt}/{MAX_TRIES} 次生成的正则编译失败 "
f"desc={desc!r} regex={last_regex!r} err={err}"
)
except LLMUnavailable as e:
logger.warning(f"AI 正则:LLM 第 {attempt} 次调用失败 desc={desc!r} err={e}")
return "", f"LLM 调用失败:{e}"
return "", f"LLM 重试 {MAX_TRIES} 次仍未生成合法正则(最后错误:{error_msg}),请补充描述或手动填写"
\ No newline at end of file
"""LLM 客户端封装(支持 MiniMax / Anthropic / OpenAI)
直接拷贝自 web/core/llm.py,2026-08-20 接入 web3 用作「AI 生成正则」。
改动点:default_config_path() 改成读 web3/backend/configs/llm.yaml,其他不动,
保持 web3 子项目自包含(不依赖 web.*)。
所有方法都设计为可降级 — 如果 LLM 不可用(无 API Key / 超时 / 解析失败),返回 None 或 fallback 值,
不阻断主流程。
支持的 Provider:
- minimax: MiniMax(M2.7 / MiniMax-Text-01 等)— 通过 Anthropic 兼容接口
- anthropic: Claude(官方 Anthropic SDK)
- openai: GPT(OpenAI 兼容协议)
配置优先级(从高到低):
1. 环境变量(最高):LLM_PROVIDER / LLM_API_KEY / LLM_BASE_URL / LLM_MODEL
2. 配置文件(中间):web3/backend/configs/llm.yaml
3. 内置默认值(最低)
启动时 LLM 客户端初始化失败会自动降级,主流程仍可继续。
"""
from __future__ import annotations
import json
import logging
import os
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional
logger = logging.getLogger(__name__)
# ── Provider 预设 ────────────────────────────────────────
# 各 Provider 的默认 base_url 与模型,避免每次都填
PROVIDER_PRESETS: dict[str, dict] = {
"minimax": {
"base_url": "https://api.minimaxi.com/anthropic",
"model": "MiniMax-Text-01",
},
"anthropic": {
"base_url": "", # 不填,使用官方默认
"model": "claude-sonnet-5",
},
"openai": {
"base_url": "", # 不填,使用官方默认
"model": "gpt-4o",
},
}
# ── 配置文件路径 ─────────────────────────────────────────
def default_config_path() -> Path:
"""默认 LLM 配置文件路径:web3/backend/configs/llm.yaml
可通过环境变量 LLM_CONFIG_FILE 覆盖:
export LLM_CONFIG_FILE=/path/to/my-llm.yaml
"""
env_path = os.environ.get("LLM_CONFIG_FILE")
if env_path:
return Path(env_path)
# web3/backend/core/llm.py → 往上两级是 web3/backend/,再拼 configs/llm.yaml
return Path(__file__).resolve().parent.parent / "configs" / "llm.yaml"
# ── 配置 ────────────────────────────────────────────────
@dataclass
class LLMConfig:
provider: str = "minimax" # minimax / anthropic / openai
api_key: Optional[str] = None
base_url: Optional[str] = None # 留空则按 provider 预设
model: Optional[str] = None # 留空则按 provider 预设
max_tokens: int = 1024
temperature: float = 0.2
timeout: int = 30
@property
def effective_model(self) -> str:
return self.model or PROVIDER_PRESETS.get(self.provider, {}).get("model", "claude-sonnet-5")
@property
def effective_base_url(self) -> str:
return self.base_url or PROVIDER_PRESETS.get(self.provider, {}).get("base_url", "")
@property
def available(self) -> bool:
return bool(self.api_key)
@classmethod
def load(cls, config_path: Optional[Path] = None) -> "LLMConfig":
"""三段式加载:默认值 < 配置文件 < 环境变量(环境变量最高优先)
优先级说明:
- 文件配置:方便项目级持久化(推荐把 provider / model / base_url 写这里)
- 环境变量:方便临时切换 / 容器化部署 / CI
"""
path = config_path or default_config_path()
cfg = cls() # 默认值
# 1. 文件
file_cfg = _read_yaml_config(path)
if file_cfg:
for k, v in file_cfg.items():
if v is not None and hasattr(cfg, k):
setattr(cfg, k, v)
# 2. 先把 provider 应用(再决定从哪个专用 env 拿 key)
env_provider = os.environ.get("LLM_PROVIDER")
if env_provider:
cfg.provider = env_provider
# 3. 环境变量(最高)
env_overrides: dict[str, Any] = {
"api_key": os.environ.get("LLM_API_KEY"),
"base_url": os.environ.get("LLM_BASE_URL"),
"model": os.environ.get("LLM_MODEL"),
}
if env_overrides["api_key"] is None:
# 兼容各 provider 的专用 env var(以更新后的 provider 为准)
if cfg.provider == "anthropic":
env_overrides["api_key"] = os.environ.get("ANTHROPIC_API_KEY")
elif cfg.provider == "openai":
env_overrides["api_key"] = os.environ.get("OPENAI_API_KEY")
elif cfg.provider == "minimax":
env_overrides["api_key"] = os.environ.get("MINIMAX_API_KEY")
env_max_tokens = os.environ.get("LLM_MAX_TOKENS")
if env_max_tokens:
try:
cfg.max_tokens = int(env_max_tokens)
except (TypeError, ValueError):
pass
for k, v in env_overrides.items():
if v is not None and v != "":
setattr(cfg, k, v)
return cfg
def _read_yaml_config(path: Path) -> Optional[dict]:
"""读取 YAML 配置文件"""
if not path.exists():
return None
try:
import yaml
with open(path, encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
return data if isinstance(data, dict) else None
except ImportError:
logger.warning("未安装 PyYAML,跳过文件配置加载")
return None
except Exception as e:
logger.warning(f"读取 LLM 配置文件失败 {path}: {e}")
return None
# ── 客户端基类 ────────────────────────────────────────────
class LLMUnavailable(Exception):
"""LLM 不可用(无 Key / 网络/解析失败)"""
class LLMClient:
"""统一 LLM 客户端。失败抛 LLMUnavailable,由调用方降级。"""
def __init__(self, cfg: LLMConfig | None = None, config_path: Optional[Path] = None):
self.cfg = cfg or LLMConfig.load(config_path)
self._provider = None
if self.cfg.available:
try:
self._provider = self._init_provider()
logger.info(
f"LLM 客户端就绪: provider={self.cfg.provider}, "
f"model={self.cfg.effective_model}, "
f"base_url={self.cfg.effective_base_url or '默认'}"
)
except Exception as e:
logger.warning(f"LLM 初始化失败: {e}")
else:
logger.info(
f"LLM 未配置 API Key(provider={self.cfg.provider}),"
f"如需启用请在 web3/backend/configs/llm.yaml 设置 api_key 或配置环境变量"
)
def _init_provider(self):
"""根据 provider 创建对应 SDK 客户端"""
provider = self.cfg.provider
api_key = self.cfg.api_key
base_url = self.cfg.effective_base_url or None
if provider in ("anthropic", "minimax"):
# MiniMax 用 Anthropic SDK + 自定义 base_url
import anthropic
kwargs: dict[str, Any] = {
"api_key": api_key,
"timeout": self.cfg.timeout,
}
if base_url:
kwargs["base_url"] = base_url
return anthropic.Anthropic(**kwargs)
elif provider == "openai":
import openai
kwargs = {"api_key": api_key, "timeout": self.cfg.timeout}
if base_url:
kwargs["base_url"] = base_url
return openai.OpenAI(**kwargs)
raise ValueError(
f"不支持的 LLM provider: {provider!r}(目前支持 minimax / anthropic / openai)"
)
@property
def available(self) -> bool:
return self._provider is not None
# ── 核心调用 ──
def complete(self, prompt: str, system: str = "", json_mode: bool = False) -> str:
"""调用 LLM,返回纯文本。失败抛 LLMUnavailable(携带可读的错误信息)。"""
if not self.available:
raise LLMUnavailable("LLM 客户端未配置 API Key")
try:
if self.cfg.provider in ("anthropic", "minimax"):
kwargs: dict[str, Any] = {
"model": self.cfg.model,
"max_tokens": self.cfg.max_tokens,
"temperature": self.cfg.temperature,
"messages": [{"role": "user", "content": prompt}],
}
if system:
kwargs["system"] = system
msg = self._provider.messages.create(**kwargs)
return msg.content[0].text
# openai
kwargs = {
"model": self.cfg.model,
"max_tokens": self.cfg.max_tokens,
"temperature": self.cfg.temperature,
"messages": [{"role": "system", "content": system},
{"role": "user", "content": prompt}],
}
if json_mode:
kwargs["response_format"] = {"type": "json_object"}
resp = self._provider.chat.completions.create(**kwargs)
return resp.choices[0].message.content
# 未支持的 provider——按理 _init_provider 已拦过,这里再防一次
raise LLMUnavailable(
f"complete() 未实现 provider={self.cfg.provider!r} 的调用分支"
)
except LLMUnavailable:
raise
except Exception as e:
detail = _format_api_error(e, self.cfg.provider)
logger.warning(
f"LLM 调用失败 (provider={self.cfg.provider}, "
f"model={self.cfg.effective_model}): {detail}"
)
raise LLMUnavailable(detail)
# ── JSON 模式(带解析兜底) ──
def complete_json(self, prompt: str, system: str = "") -> dict | list | None:
"""调用 LLM 并要求返回 JSON。解析失败返回 None。"""
json_system = (system + "\n\n" if system else "") + (
"严格返回合法 JSON,不要 Markdown 代码块包裹,不要任何额外文字。"
)
text = self.complete(prompt, system=json_system, json_mode=True)
return _safe_parse_json(text)
# ── API 错误信息提取 ──────────────────────────────────────
def _format_api_error(exc: Exception, provider: str) -> str:
"""从 SDK 异常中提取可读的错误信息,供前端实时日志展示。
优先处理 Anthropic/OpenAI SDK 的结构化异常,
兜底处理网络/超时等通用异常。
"""
exc_type = type(exc).__name__
exc_msg = str(exc)
# ── Anthropic SDK 异常(MiniMax 兼容接口也走这里) ──
try:
from anthropic import (
APIStatusError,
RateLimitError,
AuthenticationError,
PermissionDeniedError,
NotFoundError,
APIConnectionError,
APITimeoutError,
)
if isinstance(exc, RateLimitError):
return (
f"API Error: 请求被限流 (429) · {_extract_body_message(exc) or exc_msg}"
)
if isinstance(exc, AuthenticationError):
return (
f"API Error: 认证失败 (401) · 请检查 API Key 是否正确或已过期"
)
if isinstance(exc, PermissionDeniedError):
return (
f"API Error: 权限不足 (403) · {_extract_body_message(exc) or exc_msg}"
)
if isinstance(exc, NotFoundError):
return (
f"API Error: 资源不存在 (404) · {_extract_body_message(exc) or exc_msg}"
)
if isinstance(exc, APIStatusError):
status = getattr(exc, "status_code", "?")
body_msg = _extract_body_message(exc)
detail = body_msg or exc_msg
# 对常见状态码给出中文提示
hint = {
429: "请升级 Token Plan 套餐或稍后重试",
500: "服务端内部错误,请稍后重试",
502: "网关错误,服务可能暂时不可用",
503: "服务暂不可用,请稍后重试",
}.get(status, "")
if hint:
return f"API Error: 请求被拒绝 ({status}) · {detail} ({hint})"
return f"API Error: 请求失败 ({status}) · {detail}"
if isinstance(exc, APIConnectionError):
return f"API Error: 网络连接失败 · {exc_msg}"
if isinstance(exc, APITimeoutError):
return f"API Error: 请求超时 · {exc_msg}"
except ImportError:
pass
# ── OpenAI SDK 异常 ──
try:
from openai import (
APIStatusError as OAIStatusError,
RateLimitError as OAIRateLimitError,
AuthenticationError as OAIAuthError,
APIConnectionError as OAIConnectionError,
APITimeoutError as OAITimeoutError,
)
if isinstance(exc, OAIRateLimitError):
return (
f"API Error: 请求被限流 (429) · {exc_msg}"
)
if isinstance(exc, OAIAuthError):
return (
f"API Error: 认证失败 (401) · 请检查 API Key 是否正确或已过期"
)
if isinstance(exc, OAIStatusError):
status = getattr(exc, "status_code", "?")
return f"API Error: 请求失败 ({status}) · {exc_msg}"
if isinstance(exc, OAIConnectionError):
return f"API Error: 网络连接失败 · {exc_msg}"
if isinstance(exc, OAITimeoutError):
return f"API Error: 请求超时 · {exc_msg}"
except ImportError:
pass
# ── 通用网络/超时异常 ──
import builtins
if isinstance(exc, builtins.ConnectionError):
return f"API Error: 网络连接失败 · {exc_msg}"
if isinstance(exc, builtins.TimeoutError):
return f"API Error: 请求超时 · {exc_msg}"
# ── 兜底 ──
return f"API Error: {exc_type} · {exc_msg}"
def _extract_body_message(exc: Exception) -> str | None:
"""尝试从 Anthropic APIStatusError 的 body 中提取错误详情。"""
body = getattr(exc, "body", None)
if not body or not isinstance(body, dict):
return None
# body 结构: {"error": {"message": "..."}}
error = body.get("error")
if isinstance(error, dict):
msg = error.get("message")
if msg:
return str(msg)
return None
def _safe_parse_json(text: str) -> Any:
"""从 LLM 返回中提取 JSON(容忍代码块包裹、尾部多余文字)"""
if not text:
return None
# 去掉 Markdown 代码块
text = re.sub(r"^```(?:json)?\s*", "", text.strip())
text = re.sub(r"\s*```$", "", text)
# 找第一个 { 或 [
start = -1
for i, ch in enumerate(text):
if ch in "[{":
start = i
break
if start == -1:
return None
candidate = text[start:].strip()
# 尝试解析,失败则尝试逐步截断尾部字符
try:
return json.loads(candidate)
except json.JSONDecodeError:
pass
# 容错:去掉尾部的非 JSON 字符
for end in range(len(candidate), 0, -1):
try:
return json.loads(candidate[:end])
except json.JSONDecodeError:
continue
return None
# ── 模块级单例 ────────────────────────────────────────────
_client: LLMClient | None = None
def get_llm_client() -> LLMClient:
"""获取全局 LLM 客户端单例(懒初始化)"""
global _client
if _client is None:
_client = LLMClient()
return _client
def reset_llm_client() -> None:
"""重置单例(配置变更时调用)"""
global _client
_client = None
\ No newline at end of file
......@@ -5,4 +5,10 @@ pydantic>=2.5
# 数据库驱动(web3/backend/core/db_adapter.py 需要;dmPython 装不上不影响启动,只影响连达梦)
pymysql>=1.1
oracledb>=1.4
dmPython>=2.5
\ No newline at end of file
dmPython>=2.5
# LLM(web3/backend/core/ai_regex.py → core/llm.py 用)
# anthropic 走 Anthropic SDK(MiniMax 也走这个,兼容接口);openai 走 OpenAI SDK
# pyyaml 读 web3/backend/configs/llm.yaml
anthropic>=0.30
openai>=1.30
pyyaml>=6.0
\ No newline at end of file
"""AI 相关 API(web3)
端点:
POST /api/ai/regex body: {desc: str} → {ok, regex, note}
POST /api/regex/test body: {regex, value} → {ok, match, error, groups}
「测试」端点走 Python re(不依赖前端 JS RegExp),保证 LLM 生成的 Python 正则
能被同一种语义验证,避免 JS/Python 正则方言差异导致误判。
"""
from __future__ import annotations
import time
from fastapi import APIRouter
from pydantic import BaseModel, Field
from web3.backend.core.ai_regex import gen_regex, test_regex
from web3.backend._logging import get_logger
router = APIRouter(prefix="", tags=["ai"])
logger = get_logger("backend.routers.ai")
# ── /ai/regex ─────────────────────────────────────────────
class GenRegexRequest(BaseModel):
desc: str = Field(..., min_length=1, description="规则自然语言描述")
class GenRegexResponse(BaseModel):
ok: bool
regex: str = ""
note: str = ""
@router.post("/ai/regex", response_model=GenRegexResponse, summary="AI 生成 Python 正则")
async def ai_gen_regex(req: GenRegexRequest):
"""前端「规则设置」弹窗的「AI 生成正则」按钮调用。
返回 ok=true 表示 regex 非空;ok=false 表示降级(无 key / 描述不清 / 5 次重试全败)。
"""
logger.info("─" * 60)
logger.info(f"POST /api/ai/regex desc={req.desc!r}")
started = time.monotonic()
try:
regex, note = gen_regex(req.desc)
except Exception as e:
elapsed_ms = (time.monotonic() - started) * 1000
logger.exception(f"❌ AI 正则服务抛出异常(耗时 {elapsed_ms:.0f}ms)")
logger.info("─" * 60)
return GenRegexResponse(ok=False, regex="", note=f"AI 正则服务异常: {type(e).__name__}: {e}")
elapsed_ms = (time.monotonic() - started) * 1000
if regex:
logger.info(f"✅ AI 正则生成成功(耗时 {elapsed_ms:.0f}ms)regex={regex!r}")
else:
logger.warning(f"❌ AI 正则生成失败(耗时 {elapsed_ms:.0f}ms)note={note!r}")
logger.info("─" * 60)
return GenRegexResponse(ok=bool(regex), regex=regex, note=note)
# ── /regex/test ───────────────────────────────────────────
class TestRegexRequest(BaseModel):
regex: str = Field(..., min_length=1, description="要测试的正则表达式")
value: str = Field("", description="测试值")
class TestRegexResponse(BaseModel):
ok: bool # 正则编译是否成功
match: bool # value 是否匹配
error: str = "" # 编译/匹配错误信息
groups: list[str] = [] # 捕获组内容(无捕获组或未匹配时空)
@router.post("/regex/test", response_model=TestRegexResponse, summary="用 Python re 校验 value 是否匹配 regex")
async def ai_test_regex(req: TestRegexRequest):
"""前端「规则设置」弹窗的「测试」按钮调用。
用 Python re 校验(不是 JS RegExp),保证和后端检测引擎语义一致。
"""
result = test_regex(req.regex, req.value)
return TestRegexResponse(**result)
\ No newline at end of file
/**
* AI 相关 API(2026-08-20 接入)
*
* - genRegex(desc) → POST /api/ai/regex 后端走 LLM(强制,无规则库)+ re.compile 校验
* - testRegex(...) → POST /api/regex/test 后端用 Python re 校验(不是 JS RegExp)
*/
import { http } from './client'
/**
* AI 生成正则表达式
* @param {string} desc 规则说明(自然语言)
* @returns { ok: boolean, regex: string, note: string }
* - ok=true:regex 非空,可直接填到「正则表达式」输入框
* - ok=false:regex 空字符串;note 是失败原因(前端展示)
*/
export function genRegex(desc) {
return http.post('/ai/regex', { desc })
}
/**
* 用 Python re 校验 value 是否匹配 regex
* @param {string} regex
* @param {string} value
* @returns { ok: boolean, match: boolean, error: string, groups: string[] }
* - ok false = 正则编译失败;error 是 Python 报错(不是 JS 报错)
* - ok true + match true/false = 是否命中
* - groups = 捕获组内容(无捕获组或未匹配时 [])
*/
export function testRegex(regex, value) {
return http.post('/regex/test', { regex, value })
}
\ No newline at end of file
......@@ -54,7 +54,8 @@
<button class="add-rule-btn" @click="addRule">+ 添加规则</button>
</div>
<div class="modal-footer">
<button class="btn" @click="$emit('update:open', false)">关闭</button>
<button class="btn" @click="$emit('update:open', false)">取消</button>
<button class="btn btn-primary" :disabled="!hasValidRules" @click="onSave">保存</button>
</div>
</div>
</div>
......@@ -64,12 +65,13 @@
<script setup>
import { computed, reactive, ref, watch } from 'vue'
import { ElMessage } from 'element-plus'
import { genRegex as apiGenRegex, testRegex as apiTestRegex } from '@/api/ai'
const props = defineProps({
open: { type: Boolean, default: false },
field: { type: Object, default: null },
})
defineEmits(['update:open'])
const emit = defineEmits(['update:open', 'save'])
const title = computed(() => {
if (!props.field) return '规则设置'
......@@ -79,10 +81,16 @@ const title = computed(() => {
const ruleList = ref([])
const aiLoading = ref({})
watch(() => [props.open, props.field], ([open, f]) => {
if (open && f) {
// 深拷贝字段上的规则,避免直接改父组件
ruleList.value = (f.rules || []).map((r) => ({ ...r, testVal: '', testResult: null }))
// 「至少有一条规则且 desc+regex 都填了」才能保存(避免保存空规则)
// 必须在 ruleList 之后定义(const 引用前要先声明)
const hasValidRules = computed(() =>
ruleList.value.length > 0 && ruleList.value.every((r) => r.desc.trim() && r.regex.trim())
)
watch(() => [props.open, props.field?.key], ([open]) => {
if (open && props.field) {
// 每次打开都从 field.rules 重新同步一次(保证父组件保存后再开看到的是最新)
ruleList.value = (props.field.rules || []).map((r) => ({ ...r, testVal: '', testResult: null }))
aiLoading.value = {}
}
}, { immediate: true })
......@@ -100,41 +108,29 @@ function removeRule(idx) {
ruleList.value.splice(idx, 1)
}
// AI 模拟:根据规则说明返回正则
function aiGenerateRegex(desc) {
if (/身份证|id_card/i.test(desc)) return { regex: '^[1-9]\\d{5}(18|19|20)\\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\\d|3[01])\\d{3}[\\dXx]$', note: '已识别为18位身份证号' }
if (/手机|电话/.test(desc)) return { regex: '^1[3-9]\\d{9}$', note: '已识别为11位手机号' }
if (/邮箱|email|邮件/i.test(desc)) return { regex: '^[\\w.-]+@[\\w.-]+\\.[a-zA-Z]{2,}$', note: '已识别为邮箱格式' }
if (/银行卡|卡号/.test(desc)) return { regex: '^\\d{16,19}$', note: '已识别为银行卡号' }
if (/日期|yyyy/i.test(desc)) return { regex: '^\\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\\d|3[01])$', note: '已识别为日期格式' }
if (/邮编/.test(desc)) return { regex: '^[1-9]\\d{5}$', note: '已识别为6位邮编' }
if (/不能为空|非空|不能空/.test(desc)) return { regex: '^.+$', note: '已识别为非空校验' }
if (/正整数/.test(desc)) return { regex: '^[1-9]\\d*$', note: '已识别为正整数' }
if (/纯数字|数字/.test(desc)) return { regex: '^\\d+$', note: '已识别为纯数字' }
const m = desc.match(/(\d+)\s*[-到至]\s*(\d+)\s*个/)
if (m) return { regex: `^.{${m[1]},${m[2]}}$`, note: `已识别为长度${m[1]}-${m[2]}位` }
const n = desc.match(/(\d+)\s*位/)
if (n) return { regex: `^.{${n[1]}}$`, note: `已识别为固定${n[1]}位` }
return { regex: '', note: '未能识别,请补充描述或手动填写正则' }
}
function genRegex(idx) {
// AI 生成正则:调后端 POST /api/ai/regex
// 后端流程:规则库命中 → LLM 调用 + re.compile 校验(最多 5 次重试)→ 返回
async function genRegex(idx) {
const rule = ruleList.value[idx]
if (!rule.desc.trim()) {
ElMessage.warning('请先填写规则说明')
return
}
aiLoading.value[idx] = true
setTimeout(() => {
const r = aiGenerateRegex(rule.desc)
aiLoading.value[idx] = false
if (r.regex) {
try {
const r = await apiGenRegex(rule.desc)
if (r.ok && r.regex) {
rule.regex = r.regex
ElMessage.success(r.note)
ElMessage.success(r.note || '已生成正则')
} else {
ElMessage.warning(r.note)
// 后端降级(无 key / 重试 5 次失败 / 描述不清):弹 warning,把 note 给用户看
ElMessage.warning(r.note || '生成失败,请手动填写或补充描述')
}
}, 500)
} catch (e) {
ElMessage.error(`AI 生成失败:${e.message}`)
} finally {
aiLoading.value[idx] = false
}
}
function copyRegex(rule) {
......@@ -146,12 +142,29 @@ function copyRegex(rule) {
function testRegex(idx) {
const rule = ruleList.value[idx]
if (!rule.regex) { rule.testResult = { text: '无正则', cls: 'fail' }; return }
if (!rule.testVal) { rule.testResult = { text: '请输入', cls: 'fail' }; return }
try {
const ok = new RegExp(rule.regex).test(rule.testVal)
rule.testResult = { text: ok ? '✓ 通过' : '✗ 不匹配', cls: ok ? 'pass' : 'fail' }
} catch (e) {
rule.testResult = { text: '语法错误', cls: 'fail' }
}
apiTestRegex(rule.regex, rule.testVal).then((r) => {
if (!r.ok) {
rule.testResult = { text: r.error || '语法错误', cls: 'fail' }
} else if (!r.match) {
rule.testResult = { text: rule.testVal ? '✗ 不匹配' : '请输入', cls: 'fail' }
} else {
const g = r.groups && r.groups.length
? `(捕获:${r.groups.join(', ')})`
: ''
rule.testResult = { text: `✓ 通过${g}`, cls: 'pass' }
}
}).catch((e) => {
rule.testResult = { text: `测试失败:${e.message}`, cls: 'fail' }
})
}
// 保存:把 UI 状态(testVal/testResult)剥掉,只留真实数据 emit 给父组件
function onSave() {
const rules = ruleList.value.map((r) => ({
desc: r.desc.trim(),
regex: r.regex.trim(),
}))
emit('save', { fieldKey: props.field?.key, rules })
emit('update:open', false)
}
</script>
\ No newline at end of file
......@@ -229,7 +229,7 @@
</Teleport>
<!-- 规则弹窗 -->
<RuleModal v-model:open="ruleModalOpen" :field="activeField" />
<RuleModal v-model:open="ruleModalOpen" :field="activeField" @save="onRuleSave" />
</template>
<script setup>
......@@ -530,6 +530,13 @@ function openRule(f) {
ruleModalOpen.value = true
}
// 规则保存回写:RuleModal.emit('save', {fieldKey, rules})
// 把 rules 写回对应字段的 field.rules,让「规则设置 N」徽章实时刷新
function onRuleSave({ fieldKey, rules }) {
const f = fieldList.value.find((x) => x.key === fieldKey)
if (f) f.rules = rules
}
function onSave() {
if (!form.name) {
ElMessage.warning('请填写任务名称')
......@@ -537,7 +544,7 @@ function onSave() {
return
}
// 后端 P4 才接 fields/rules,P3 只送头部 + conn_json
// (弹窗里改的字段/规则暂存 form,下次保存会带上;当前阶段不写库)
// 字段+规则先挂 payload 上,等后端接好就发;目前后端忽略
const connJson = JSON.stringify({
connName: form.connName,
host: form.host,
......@@ -558,6 +565,13 @@ function onSave() {
source_table: form.table || null,
description: form.desc,
conn_json: connJson,
fields: fieldList.value.map((f) => ({ // P4 落库用
key: f.key,
en: f.en,
cn: f.cn,
rules: f.rules || [],
show_default: f.showDefault,
})),
}
emit('saved', payload)
}
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment