419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""多厂商 LLM：供应商表从 etc/llm.yaml 加载（密钥仍走环境变量）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import json
|
||
import os
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import yaml
|
||
|
||
_DEFAULT_CONFIG_CANDIDATES = (
|
||
Path(__file__).resolve().parent / "etc" / "llm.yaml",
|
||
Path("/app/etc/llm.yaml"),
|
||
)
|
||
|
||
|
||
def _config_path() -> Path:
|
||
for key in ("AI_CONFIG_PATH", "LLM_CONFIG_PATH"):
|
||
raw = (os.getenv(key) or "").strip()
|
||
if raw:
|
||
return Path(raw)
|
||
for p in _DEFAULT_CONFIG_CANDIDATES:
|
||
if p.is_file():
|
||
return p
|
||
return _DEFAULT_CONFIG_CANDIDATES[0]
|
||
|
||
|
||
@lru_cache(maxsize=1)
|
||
def load_ai_config() -> dict[str, Any]:
|
||
path = _config_path()
|
||
if not path.is_file():
|
||
raise FileNotFoundError(
|
||
f"AI 配置文件不存在: {path}（可用 AI_CONFIG_PATH / LLM_CONFIG_PATH 指定）"
|
||
)
|
||
with path.open("r", encoding="utf-8") as f:
|
||
data = yaml.safe_load(f) or {}
|
||
if not isinstance(data, dict):
|
||
raise ValueError(f"AI 配置格式错误（须为 mapping）: {path}")
|
||
providers = data.get("providers") or {}
|
||
if not isinstance(providers, dict) or not providers:
|
||
raise ValueError(f"AI 配置缺少 providers: {path}")
|
||
return data
|
||
|
||
|
||
def reload_ai_config() -> dict[str, Any]:
|
||
"""供热加载 / 测试清空缓存。"""
|
||
load_ai_config.cache_clear()
|
||
return load_ai_config()
|
||
|
||
|
||
def get_providers() -> dict[str, dict[str, Any]]:
|
||
raw = load_ai_config().get("providers") or {}
|
||
out: dict[str, dict[str, Any]] = {}
|
||
for pid, meta in raw.items():
|
||
if not isinstance(meta, dict):
|
||
continue
|
||
out[str(pid).strip().lower()] = dict(meta)
|
||
return out
|
||
|
||
|
||
# 兼容旧代码：PROVIDERS 为属性式访问，始终读当前配置
|
||
class _ProvidersProxy(dict):
|
||
def _sync(self) -> None:
|
||
self.clear()
|
||
self.update(get_providers())
|
||
|
||
def __contains__(self, key: object) -> bool: # type: ignore[override]
|
||
self._sync()
|
||
return dict.__contains__(self, key)
|
||
|
||
def __getitem__(self, key: str) -> dict[str, Any]:
|
||
self._sync()
|
||
return dict.__getitem__(self, key)
|
||
|
||
def get(self, key: str, default: Any = None) -> Any: # type: ignore[override]
|
||
self._sync()
|
||
return dict.get(self, key, default)
|
||
|
||
def items(self): # type: ignore[override]
|
||
self._sync()
|
||
return dict.items(self)
|
||
|
||
def keys(self): # type: ignore[override]
|
||
self._sync()
|
||
return dict.keys(self)
|
||
|
||
def values(self): # type: ignore[override]
|
||
self._sync()
|
||
return dict.values(self)
|
||
|
||
|
||
PROVIDERS: dict[str, dict[str, Any]] = _ProvidersProxy() # type: ignore[assignment]
|
||
|
||
|
||
def _default_provider_id() -> str:
|
||
cfg = load_ai_config()
|
||
return (
|
||
(os.getenv("LLM_PROVIDER") or "").strip().lower()
|
||
or str(cfg.get("default_provider") or "deepseek").strip().lower()
|
||
or "deepseek"
|
||
)
|
||
|
||
|
||
def _aliases() -> dict[str, str]:
|
||
raw = load_ai_config().get("aliases") or {}
|
||
return {str(k).strip().lower(): str(v).strip().lower() for k, v in raw.items()}
|
||
|
||
|
||
def _vision_cfg() -> dict[str, Any]:
|
||
v = load_ai_config().get("vision") or {}
|
||
return v if isinstance(v, dict) else {}
|
||
|
||
|
||
def _normalize_pid(provider_id: str | None, *, aliases: dict[str, str] | None = None) -> str:
|
||
pid = (provider_id or "").strip().lower()
|
||
amap = aliases if aliases is not None else _aliases()
|
||
if pid in amap:
|
||
pid = amap[pid]
|
||
return pid
|
||
|
||
|
||
def list_providers() -> list[dict[str, Any]]:
|
||
out: list[dict[str, Any]] = []
|
||
for pid, meta in get_providers().items():
|
||
key_env = meta.get("api_key_env") or ""
|
||
configured = True
|
||
if key_env:
|
||
configured = bool(os.getenv(str(key_env), "").strip())
|
||
out.append(
|
||
{
|
||
"id": pid,
|
||
"label": meta.get("label") or pid,
|
||
"configured": configured,
|
||
"default_model": meta.get("default_model") or "",
|
||
"models": meta.get("models") or [],
|
||
"supports_vision": bool(meta.get("supports_vision")),
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def resolve_provider(provider_id: str | None, model: str | None = None) -> tuple[str, dict[str, Any], str]:
|
||
"""返回 (provider_id, meta, model_name)。"""
|
||
providers = get_providers()
|
||
pid = _normalize_pid(provider_id) or _default_provider_id()
|
||
if pid not in providers:
|
||
fallback = _default_provider_id()
|
||
pid = fallback if fallback in providers else next(iter(providers))
|
||
meta = dict(providers[pid])
|
||
# 单供应商可被环境变量覆盖 base_url（如 DASHSCOPE_BASE_URL）
|
||
env_base_key = f"{pid.upper()}_BASE_URL"
|
||
env_base = (os.getenv(env_base_key) or "").strip()
|
||
if env_base:
|
||
meta["base_url"] = env_base
|
||
elif pid == "dashscope":
|
||
# 兼容旧变量名
|
||
legacy = (os.getenv("DASHSCOPE_BASE_URL") or "").strip()
|
||
if legacy:
|
||
meta["base_url"] = legacy
|
||
vision = _vision_cfg()
|
||
if pid == (str(vision.get("provider") or "dashscope").strip().lower()):
|
||
if (os.getenv("VISION_MODEL") or "").strip() and not (model or "").strip():
|
||
meta["default_model"] = os.getenv("VISION_MODEL", "").strip()
|
||
model_name = (model or "").strip() or str(meta.get("default_model") or "")
|
||
return pid, meta, model_name
|
||
|
||
|
||
def resolve_vision_provider(
|
||
provider_id: str | None = None,
|
||
model: str | None = None,
|
||
) -> tuple[str, dict[str, Any], str]:
|
||
"""优先配置 vision / 环境变量 VISION_*；禁止把纯文本模型名传给视觉通道。"""
|
||
providers = get_providers()
|
||
vision = _vision_cfg()
|
||
vision_aliases = {
|
||
**_aliases(),
|
||
**{
|
||
str(k).strip().lower(): str(v).strip().lower()
|
||
for k, v in (vision.get("aliases") or {}).items()
|
||
},
|
||
}
|
||
env_pid = (os.getenv("VISION_PROVIDER") or "").strip().lower()
|
||
cfg_pid = str(vision.get("provider") or "").strip().lower()
|
||
vision_pid = _normalize_pid(env_pid or provider_id or cfg_pid, aliases=vision_aliases)
|
||
|
||
env_model = (os.getenv("VISION_MODEL") or "").strip()
|
||
cfg_model = str(vision.get("model") or "").strip()
|
||
passed = (model or "").strip()
|
||
hints = [str(h).lower() for h in (load_ai_config().get("vision_model_hints") or [])]
|
||
if not hints:
|
||
hints = ["vl", "vision", "qwen3", "qwen-vl", "gpt-4o", "gemini"]
|
||
vision_like = bool(
|
||
passed
|
||
and any(k in passed.lower() for k in hints)
|
||
and "deepseek" not in passed.lower()
|
||
and passed.lower() != "chat"
|
||
)
|
||
vision_model = env_model or cfg_model or (passed if vision_like else "") or ""
|
||
|
||
def _has_key(pid: str) -> bool:
|
||
meta = providers.get(pid) or {}
|
||
env = meta.get("api_key_env") or ""
|
||
return bool(env) and bool(os.getenv(str(env), "").strip())
|
||
|
||
# 文本供应商无视觉时，回退到配置的视觉供应商（若已配密钥）
|
||
if vision_pid in providers and not providers[vision_pid].get("supports_vision"):
|
||
fallback = _normalize_pid(cfg_pid or "dashscope", aliases=vision_aliases)
|
||
if fallback in providers and providers[fallback].get("supports_vision") and _has_key(fallback):
|
||
vision_pid = fallback
|
||
else:
|
||
vision_pid = ""
|
||
|
||
if vision_pid and vision_pid in providers and providers[vision_pid].get("supports_vision"):
|
||
return resolve_provider(vision_pid, vision_model or None)
|
||
|
||
pid, meta, _ = resolve_provider(provider_id, None)
|
||
if meta.get("supports_vision"):
|
||
return resolve_provider(pid, vision_model or None)
|
||
|
||
fallback = _normalize_pid(cfg_pid or "dashscope", aliases=vision_aliases)
|
||
if fallback in providers and providers[fallback].get("supports_vision") and _has_key(fallback):
|
||
return resolve_provider(fallback, vision_model or None)
|
||
return pid, meta, vision_model or str(meta.get("default_model") or "")
|
||
|
||
|
||
def _api_key(meta: dict[str, Any]) -> str:
|
||
env = meta.get("api_key_env") or ""
|
||
if not env:
|
||
return ""
|
||
return os.getenv(str(env), "").strip()
|
||
|
||
|
||
def enhance_blueprint_with_llm(
|
||
draft: dict[str, Any],
|
||
prompt: str,
|
||
excel_summary: str,
|
||
provider_id: str | None = None,
|
||
model: str | None = None,
|
||
) -> tuple[dict[str, Any], list[str]]:
|
||
pid, meta, model_name = resolve_provider(provider_id, model)
|
||
if pid == "heuristic":
|
||
return draft, ["使用本地启发式蓝图（未调用大模型）"]
|
||
|
||
api_key = _api_key(meta)
|
||
if not api_key:
|
||
env = meta.get("api_key_env") or ""
|
||
return draft, [f"未配置 {env}，回退启发式蓝图"]
|
||
|
||
base = (meta.get("base_url") or "").rstrip("/")
|
||
system = (
|
||
"你是低代码 CMS 蓝图助手。只能输出 JSON 对象，字段必须兼容现有 draft 结构。"
|
||
"field.type 仅允许: string,text,int,bigint,decimal,boolean,date,datetime,enum,json,file_ref。"
|
||
"所有 name/slug/table/path 必须是 snake_case（小写字母数字下划线），禁止驼峰与连字符。"
|
||
"可优化 meta.name/description、field.label、pages.title；不要发明新 type，不要输出 SQL。"
|
||
"优先在给定 draft 上微调，保留 entities/fields/apis/pages 结构完整。"
|
||
"若 draft.meta.ui_preset 或页面 layout.preset 为 screenshot_faithful，必须原样保留，"
|
||
"禁止改成通用后台壳；用户未声明修改的标题/分区/筛选/图表/功能按键文案不得擅自更换。"
|
||
"若 layout.action_labels 已给出按钮原文，必须保留；actions 顺序也尽量保留。"
|
||
"页面形态须跟随用户需求与截图还原，禁止所有应用统一成同一种列表模板；"
|
||
"禁止把任意业务改写成固定行业文案（如铁路沉降），除非用户或截图本身如此。"
|
||
)
|
||
user = {
|
||
"user_prompt": prompt,
|
||
"excel_summary": excel_summary,
|
||
"draft": draft,
|
||
"instruction": "返回完整 draft JSON（不要 markdown）",
|
||
}
|
||
body: dict[str, Any] = {
|
||
"model": model_name,
|
||
"temperature": 0.2,
|
||
"messages": [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": json.dumps(user, ensure_ascii=False)},
|
||
],
|
||
}
|
||
if meta.get("json_mode"):
|
||
body["response_format"] = {"type": "json_object"}
|
||
|
||
try:
|
||
with httpx.Client(timeout=90.0) as client:
|
||
resp = client.post(
|
||
f"{base}/chat/completions",
|
||
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
|
||
json=body,
|
||
)
|
||
resp.raise_for_status()
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
text = content.strip()
|
||
if text.startswith("```"):
|
||
text = text.strip("`")
|
||
if text.startswith("json"):
|
||
text = text[4:].lstrip()
|
||
enhanced = json.loads(text)
|
||
if "version" in enhanced and "entities" in enhanced:
|
||
enhanced.setdefault("storage", draft.get("storage"))
|
||
enhanced.setdefault("security", draft.get("security"))
|
||
enhanced.setdefault("apis", draft.get("apis"))
|
||
return enhanced, [f"已使用 {meta.get('label', pid)}/{model_name} 润色蓝图"]
|
||
if "draft" in enhanced and isinstance(enhanced["draft"], dict):
|
||
d = enhanced["draft"]
|
||
d.setdefault("storage", draft.get("storage"))
|
||
d.setdefault("security", draft.get("security"))
|
||
return d, [f"已使用 {meta.get('label', pid)}/{model_name} 润色蓝图"]
|
||
return draft, ["LLM 返回结构无效，回退启发式"]
|
||
except Exception as e: # noqa: BLE001
|
||
return draft, [f"{meta.get('label', pid)} 调用失败，回退启发式: {e}"]
|
||
|
||
|
||
_VISION_PROMPT = """这些是目标业务系统的界面截图。请用中文输出「可直接写入领域说明」的结构化要点，务必逐字抄录可见文案，不要臆造。
|
||
|
||
多图规则（重要）：
|
||
- 若多张图是**同一页面**的不同状态（展开/收起某一区、滚动到不同里程、筛选前后、空表/有数据），视为**一个界面**，合并描述，不要当成多个互斥布局。
|
||
- 以「信息最完整」的那张为主结构（例如展开后能看见主图+监督条+表）；收起态只补充「可折叠/可滚动」交互，不要因此删掉展开态才有的分区。
|
||
- 若文案冲突，优先采用更清晰、更完整的一帧；并在备注里写「同页多状态」。
|
||
- 仅当明显是不同页面（不同顶栏系统名/不同主导航页）时，才分页面描述。
|
||
|
||
按下列小标题组织（若没有的写「未见」）：
|
||
### 顶栏
|
||
- 左侧平台抬头、中间系统名、右侧链接原文
|
||
### 次级导航
|
||
- 页签原文（顿号分隔）
|
||
### 筛选条
|
||
- 工程/业务上下文原文
|
||
- 统计文案（个数类，原文照抄）
|
||
- 单选选项原文（全部 / …）
|
||
- 下拉标签原文
|
||
- 右侧提示原文（若有）
|
||
### 主图区
|
||
- 左侧竖排标签原文
|
||
- 图例系列名原文
|
||
- 图表类型与坐标大致含义
|
||
### 状态条/副图
|
||
- 左侧竖排标签原文
|
||
- 形态：单色格 / 双段条 / 其它（据实描述）
|
||
- 条上是否显示类目文字
|
||
- 是否可折叠/收起（若另一张图为收起态请注明）
|
||
### 表格区
|
||
- 左侧竖排标签、表头横标题、列名原文
|
||
### 操作按键
|
||
- 格式：操作：……、……
|
||
### 多图关系（若有多张）
|
||
- 一句话说明：同页展开/收起，或不同页
|
||
|
||
用户补充需求：{prompt}
|
||
"""
|
||
|
||
|
||
def understand_images(
|
||
prompt: str,
|
||
images: list[tuple[str, bytes, str]],
|
||
provider_id: str | None = None,
|
||
model: str | None = None,
|
||
) -> tuple[str, list[str]]:
|
||
"""返回补充描述 + warnings。无视觉能力时回退 VISION_PROVIDER / 配置中的视觉供应商。"""
|
||
warnings: list[str] = []
|
||
if not images:
|
||
return "", warnings
|
||
|
||
names = [n for n, _, _ in images]
|
||
hint = (
|
||
f"用户上传了界面截图: {', '.join(names)}。"
|
||
"默认按截图原样还原（screenshot_faithful）：顶栏、页签、筛选、工具栏功能按键原文、图表分区、表格列均与原图一致；"
|
||
"仅用户文字明确要求修改的部分可调整。"
|
||
)
|
||
|
||
pid, meta, model_name = resolve_vision_provider(provider_id, model)
|
||
if not meta.get("supports_vision"):
|
||
warnings.append(
|
||
f"当前视觉通道 {pid or '无'} 不支持看图；请在 etc/llm.yaml 配置 vision.provider 并设置对应 API Key"
|
||
)
|
||
return hint, warnings
|
||
|
||
api_key = _api_key(meta)
|
||
if not api_key:
|
||
env = meta.get("api_key_env") or ""
|
||
warnings.append(f"未配置 {env}，截图仅作占位提示")
|
||
return hint, warnings
|
||
|
||
base = (meta.get("base_url") or "").rstrip("/")
|
||
content: list[dict[str, Any]] = [
|
||
{"type": "text", "text": _VISION_PROMPT.format(prompt=prompt or "（无额外文字）")}
|
||
]
|
||
for name, raw, mime in images:
|
||
b64 = base64.b64encode(raw).decode("ascii")
|
||
content.append(
|
||
{
|
||
"type": "image_url",
|
||
"image_url": {"url": f"data:{mime or 'image/png'};base64,{b64}"},
|
||
}
|
||
)
|
||
_ = name
|
||
|
||
body = {
|
||
"model": model_name,
|
||
"temperature": 0.1,
|
||
"messages": [{"role": "user", "content": content}],
|
||
}
|
||
try:
|
||
with httpx.Client(timeout=120.0) as client:
|
||
resp = client.post(
|
||
f"{base}/chat/completions",
|
||
headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
|
||
json=body,
|
||
)
|
||
resp.raise_for_status()
|
||
text = (resp.json()["choices"][0]["message"]["content"] or "").strip()
|
||
if not text:
|
||
return hint, warnings + ["视觉模型返回空内容"]
|
||
return f"{hint}\n\n【截图理解】\n{text}", warnings + [
|
||
f"已用 {meta.get('label', pid)}/{model_name} 理解截图"
|
||
]
|
||
except Exception as e: # noqa: BLE001
|
||
warnings.append(f"看图失败: {e}")
|
||
return hint, warnings
|