From 7a3da4eea208bf76263dfd49bc51c8867003024e Mon Sep 17 00:00:00 2001 From: zichun <26684461+reporkey@users.noreply.github.com> Date: Wed, 22 Jul 2026 10:15:23 +0800 Subject: [PATCH] feat: NL2SQL table-allowlist + tenant injection (§9) + Langfuse export (§1) + chat UI (question/form_collect/token) --- src/main/java/com/xly/config/TracingChatModelListener.java | 122 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------- src/main/java/com/xly/tool/QueryTool.java | 150 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------- src/main/resources/application-saaslocal.yml | 8 ++++++++ src/main/resources/templates/chat.html | 68 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------- 4 files changed, 323 insertions(+), 25 deletions(-) diff --git a/src/main/java/com/xly/config/TracingChatModelListener.java b/src/main/java/com/xly/config/TracingChatModelListener.java index 436fcc0..5c6c869 100644 --- a/src/main/java/com/xly/config/TracingChatModelListener.java +++ b/src/main/java/com/xly/config/TracingChatModelListener.java @@ -1,5 +1,6 @@ package com.xly.config; +import com.fasterxml.jackson.databind.ObjectMapper; import dev.langchain4j.model.chat.listener.ChatModelErrorContext; import dev.langchain4j.model.chat.listener.ChatModelListener; import dev.langchain4j.model.chat.listener.ChatModelRequestContext; @@ -7,44 +8,147 @@ import dev.langchain4j.model.chat.listener.ChatModelResponseContext; import dev.langchain4j.model.output.TokenUsage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.Base64; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + /** - * LLM 调用观测（轻量 tracing）—— 每次模型调用记录耗时 / token 用量 / 错误到 `com.xly.trace.llm` 日志。 + * LLM 调用观测（tracing）—— 每次模型调用记录耗时 / token 用量 / 错误。 * - *

这是 **Langfuse 的接入点**：Langfuse 本体是需自托管的观测服务（需要实例 + 密钥），此处的 - * onResponse/onError 就是把 span 转发给 Langfuse 的挂钩位；在没有实例的环境下先落到日志，保证 - * 「LLM 可观测」这一能力有实现、可随时对接 Langfuse。业务审计另见 `ai_audit_log`（与 LLM tracing 分离）。 + *

默认落 {@code com.xly.trace.llm} 日志（保证「LLM 可观测」这一能力始终有实现）。当配置 + * {@code langfuse.enabled=true} 且给了 host + 公私钥时，额外把一条 generation span **转发到自托管 Langfuse** + * 的 ingestion API（不引第三方依赖，直接用 JDK HttpClient，best-effort、异步、失败不影响主流程）。 + * 自托管方式见 {@code docker-compose.langfuse.yml}。业务审计另见 {@code ai_audit_log}（与 LLM tracing 分离）。 */ @Component public class TracingChatModelListener implements ChatModelListener { private static final Logger log = LoggerFactory.getLogger("com.xly.trace.llm"); + private final ObjectMapper mapper; + private final HttpClient http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(5)).build(); + + @Value("${langfuse.enabled:false}") + private boolean langfuseEnabled; + @Value("${langfuse.host:http://localhost:3000}") + private String langfuseHost; + @Value("${langfuse.public-key:}") + private String publicKey; + @Value("${langfuse.secret-key:}") + private String secretKey; + @Value("${langchain4j.ollama.chat-model-name:unknown}") + private String modelName; + + public TracingChatModelListener(ObjectMapper mapper) { + this.mapper = mapper; + } + @Override public void onRequest(ChatModelRequestContext ctx) { ctx.attributes().put("t0", System.nanoTime()); + ctx.attributes().put("startTs", Instant.now().toString()); } @Override public void onResponse(ChatModelResponseContext ctx) { long ms = elapsedMs(ctx.attributes().get("t0")); - String tok = "?"; + Integer in = null; + Integer out = null; try { TokenUsage u = ctx.chatResponse() == null ? null : ctx.chatResponse().tokenUsage(); if (u != null) { - tok = u.inputTokenCount() + "/" + u.outputTokenCount(); + in = u.inputTokenCount(); + out = u.outputTokenCount(); } } catch (Exception ignore) { } - log.info("LLM ok {}ms tokens(in/out)={}", ms, tok); - // Langfuse 接入点：此处可 forward 一个 span（model, prompt, completion, latency, tokens）。 + log.info("LLM ok {}ms tokens(in/out)={}/{}", ms, in, out); + exportToLangfuse(String.valueOf(ctx.attributes().get("startTs")), in, out, null); } @Override public void onError(ChatModelErrorContext ctx) { Throwable e = ctx.error(); - log.warn("LLM error: {}", e == null ? "?" : e.getMessage()); + String msg = e == null ? "?" : e.getMessage(); + log.warn("LLM error: {}", msg); + exportToLangfuse(String.valueOf(ctx.attributes().get("startTs")), null, null, msg); + } + + /** 把一条 generation span 转发到 Langfuse（best-effort，异步，失败仅告警）。未启用则直接返回。 */ + private void exportToLangfuse(String startTs, Integer in, Integer out, String error) { + if (!langfuseEnabled || publicKey == null || publicKey.isBlank() || secretKey == null || secretKey.isBlank()) { + return; + } + try { + String traceId = UUID.randomUUID().toString(); + String now = Instant.now().toString(); + String start = (startTs == null || "null".equals(startTs)) ? now : startTs; + + Map usage = new LinkedHashMap<>(); + usage.put("input", in); + usage.put("output", out); + usage.put("unit", "TOKENS"); + + Map genBody = new LinkedHashMap<>(); + genBody.put("id", UUID.randomUUID().toString()); + genBody.put("traceId", traceId); + genBody.put("name", "xlyAi-agent"); + genBody.put("model", modelName); + genBody.put("startTime", start); + genBody.put("endTime", now); + genBody.put("usage", usage); + if (error != null) { + genBody.put("level", "ERROR"); + genBody.put("statusMessage", error); + } + + Map traceBody = new LinkedHashMap<>(); + traceBody.put("id", traceId); + traceBody.put("name", "xlyAi-agent"); + traceBody.put("timestamp", start); + + List> batch = List.of( + event("trace-create", now, traceBody), + event("generation-create", now, genBody)); + String body = mapper.writeValueAsString(Map.of("batch", batch)); + + String auth = "Basic " + Base64.getEncoder().encodeToString( + (publicKey + ":" + secretKey).getBytes(StandardCharsets.UTF_8)); + HttpRequest req = HttpRequest.newBuilder(URI.create(langfuseHost + "/api/public/ingestion")) + .header("Content-Type", "application/json") + .header("Authorization", auth) + .timeout(Duration.ofSeconds(5)) + .POST(HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8)) + .build(); + http.sendAsync(req, HttpResponse.BodyHandlers.discarding()) + .exceptionally(t -> { + log.debug("Langfuse export failed: {}", t.getMessage()); + return null; + }); + } catch (Exception e) { + log.debug("Langfuse export error: {}", e.getMessage()); + } + } + + private Map event(String type, String ts, Map body) { + Map ev = new LinkedHashMap<>(); + ev.put("id", UUID.randomUUID().toString()); + ev.put("type", type); + ev.put("timestamp", ts); + ev.put("body", body); + return ev; } private long elapsedMs(Object t0) { diff --git a/src/main/java/com/xly/tool/QueryTool.java b/src/main/java/com/xly/tool/QueryTool.java index 00d475a..7f06ed9 100644 --- a/src/main/java/com/xly/tool/QueryTool.java +++ b/src/main/java/com/xly/tool/QueryTool.java @@ -5,13 +5,23 @@ import com.xly.service.AuditService; import dev.langchain4j.agent.tool.P; import dev.langchain4j.agent.tool.Tool; import dev.langchain4j.model.ollama.OllamaChatModel; +import net.sf.jsqlparser.expression.StringValue; +import net.sf.jsqlparser.expression.operators.conditional.AndExpression; +import net.sf.jsqlparser.expression.operators.relational.EqualsTo; import net.sf.jsqlparser.parser.CCJSqlParserUtil; +import net.sf.jsqlparser.schema.Column; +import net.sf.jsqlparser.schema.Table; import net.sf.jsqlparser.statement.Statement; +import net.sf.jsqlparser.statement.select.PlainSelect; import net.sf.jsqlparser.statement.select.Select; +import net.sf.jsqlparser.util.TablesNamesFinder; import org.springframework.jdbc.core.JdbcTemplate; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; /** * Query 工具：**只读 SQL 兜底** —— 回答没有现成表单/记录能直接答的临时统计/分析问题 @@ -60,7 +70,7 @@ public class QueryTool { audit.log(null, null, "query", "REJECTED", sql, false, reject); return "无法安全执行该查询（" + reject + "）。可以换个更具体的问法。"; } - String limited = forceLimit(sql); + String limited = forceLimit(applyTenant(sql)); try { List> rows = jdbc.queryForList(limited); audit.log(null, null, "query", "ok", limited, true, "rows=" + rows.size() + (attempt > 0 ? " (retry " + attempt + ")" : "")); @@ -85,11 +95,17 @@ public class QueryTool { 你是 MySQL 专家。根据【问题】生成 **一条** MySQL SELECT 查询来回答它。 数据库 = xlyweberp_saas。可用的表和字段（列名=中文名）： %s - 规则：只用 SELECT（严禁任何写操作 / 文件操作）；需要时 JOIN；务必带合适的 LIMIT（<=100）； + 规则：只用 SELECT（严禁任何写操作 / 文件操作）；**只能查询上面列出的业务表/视图**，不得访问其它表； + 需要时 JOIN；务必带合适的 LIMIT（<=100）；**含 sBrandsId 列的表务必加 `sBrandsId='%s'` 过滤**（本企业数据）； **列别名一律用英文**（如 cnt、total、name），ORDER BY 用英文列名或序号，**绝不要用中文做别名**； 表名、列名一律用上面给定的英文名。**只输出 SQL 本身**，不要解释、不要 markdown 代码围栏。 问题：%s%s - """.formatted(hint, question, repair); + """.formatted(hint, brandHint(), question, repair); + } + + private String brandHint() { + String b = identity == null ? null : identity.brandsId(); + return (b == null || b.isBlank()) ? "本企业" : b; } private String rootMsg(Throwable e) { @@ -145,7 +161,15 @@ public class QueryTool { return s.trim(); } - /** 单条 SELECT + 挡危险构造。返回 null=通过，否则=拒绝原因。 */ + // AI 可用业务表/视图白名单（静态元数据，跨请求缓存）；含 sBrandsId 列的表缓存。 + private static volatile Set ALLOWED_TABLES; + private static final ConcurrentHashMap BRAND_COL = new ConcurrentHashMap<>(); + + /** + * 单条 SELECT + 挡危险构造 + **表白名单**（架构 §9 的“视图白名单”）。 + * 白名单 = AI 可用的业务表/视图（viw_* + 表单数据源 + 字段字典表），把凭证/权限/暂存等敏感表挡在外面 + * （如 gdslogininfo、sysjurisdiction、ai_op_queue），防 NL2SQL 击穿权限。返回 null=通过，否则=拒绝原因。 + */ private String validate(String sql) { if (sql == null || sql.isBlank()) { return "未生成SQL"; @@ -158,15 +182,127 @@ public class QueryTool { return "含禁止构造: " + b; } } + Statement stmt; + try { + stmt = CCJSqlParserUtil.parse(sql); + } catch (Exception e) { + return "SQL 解析失败"; + } + if (!(stmt instanceof Select)) { + return "只允许 SELECT"; + } + Set allowed = allowedTables(); + if (!allowed.isEmpty()) { + List tables; + try { + tables = new TablesNamesFinder().getTableList(stmt); + } catch (Exception e) { + return "无法解析查询涉及的表"; + } + for (String t : tables) { + if (!allowed.contains(normTable(t))) { + return "涉及不允许访问的表(" + normTable(t) + ")"; + } + } + } + return null; + } + + /** AI 可用表/视图集合（小写）：所有 viw_* 视图 + 表单数据源 + 字段字典里出现的基础表。静态缓存。 */ + private Set allowedTables() { + Set c = ALLOWED_TABLES; + if (c != null) { + return c; + } + Set s = new HashSet<>(); + try { + for (Map r : jdbc.queryForList( + "SELECT LOWER(TABLE_NAME) t FROM information_schema.VIEWS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'viw\\_%'")) { + s.add(String.valueOf(r.get("t"))); + } + for (Map r : jdbc.queryForList( + "SELECT DISTINCT LOWER(sDataSource) t FROM viw_ai_useful_forms WHERE IFNULL(sDataSource,'')<>''")) { + s.add(String.valueOf(r.get("t"))); + } + for (Map r : jdbc.queryForList( + "SELECT DISTINCT LOWER(sTable) t FROM viw_kg_field_dict WHERE IFNULL(sTable,'')<>''")) { + s.add(String.valueOf(r.get("t"))); + } + } catch (Exception ignore) { + // 元数据不可用时返回空集 → 不启用白名单（保持可用），但仍有 SELECT-only + 禁止构造兜底。 + } + ALLOWED_TABLES = s; + return s; + } + + /** 规整表名：去反引号、去 schema 前缀、转小写。 */ + private static String normTable(String t) { + if (t == null) { + return ""; + } + String x = t.replace("`", "").trim(); + int dot = x.lastIndexOf('.'); + if (dot >= 0) { + x = x.substring(dot + 1); + } + return x.toLowerCase(); + } + + /** + * 租户注入（架构 §9）：单表（无 JOIN）且该表含 sBrandsId 列、且身份带品牌时，追加 + * {@code AND sBrandsId=''}，把结果限定在本企业。视图(viw_*)通常已按品牌预筛，跳过。 + * 任何异常都退回原 SQL（不因注入失败而阻断，白名单已是主要边界）。 + */ + private String applyTenant(String sql) { + String brand = identity == null ? null : identity.brandsId(); + if (brand == null || brand.isBlank()) { + return sql; + } try { Statement stmt = CCJSqlParserUtil.parse(sql); if (!(stmt instanceof Select)) { - return "只允许 SELECT"; + return sql; + } + Select select = (Select) stmt; + if (!(select.getSelectBody() instanceof PlainSelect)) { + return sql; } + PlainSelect ps = (PlainSelect) select.getSelectBody(); + if (ps.getJoins() != null && !ps.getJoins().isEmpty()) { + return sql; // 多表：交给白名单，不做注入 + } + if (!(ps.getFromItem() instanceof Table)) { + return sql; + } + String table = normTable(((Table) ps.getFromItem()).getName()); + if (table.startsWith("viw")) { + return sql; // 视图预筛，不注入 + } + if (!hasBrandCol(table)) { + return sql; + } + EqualsTo eq = new EqualsTo(); + eq.setLeftExpression(new Column("sBrandsId")); + eq.setRightExpression(new StringValue(brand)); + ps.setWhere(ps.getWhere() == null ? eq : new AndExpression(ps.getWhere(), eq)); + return select.toString(); } catch (Exception e) { - return "SQL 解析失败"; + return sql; } - return null; + } + + /** 该基础表是否有 sBrandsId 列（缓存）。 */ + private boolean hasBrandCol(String table) { + return BRAND_COL.computeIfAbsent(table, t -> { + try { + Integer n = jdbc.queryForObject( + "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() " + + "AND TABLE_NAME=? AND COLUMN_NAME='sBrandsId'", Integer.class, t); + return n != null && n > 0; + } catch (Exception e) { + return false; + } + }); } private String forceLimit(String sql) { diff --git a/src/main/resources/application-saaslocal.yml b/src/main/resources/application-saaslocal.yml index a6da75f..79c2741 100644 --- a/src/main/resources/application-saaslocal.yml +++ b/src/main/resources/application-saaslocal.yml @@ -39,3 +39,11 @@ erp: subsidiary: "1111111111" username: admin password: "666666" + +# LLM 可观测（Langfuse，架构 §1）。默认关闭；自托管起来后填 key 开启： +# docker compose -f docker-compose.langfuse.yml up -d → http://localhost:3000 拿 public/secret key。 +langfuse: + enabled: false + host: http://localhost:3000 + public-key: "" + secret-key: "" diff --git a/src/main/resources/templates/chat.html b/src/main/resources/templates/chat.html index c5c4e4d..ad2b540 100644 --- a/src/main/resources/templates/chat.html +++ b/src/main/resources/templates/chat.html @@ -675,7 +675,13 @@ body: JSON.stringify({ text: message, userid: userid, - conversationId: conversationId + conversationId: conversationId, + // 透传身份 + ERP 登录 token（后端据此按用户真实权限收紧；token 只用于转发给 ERP，不进 prompt） + authorization: authorization, + username: username, + brandsid: brandsid, + subsidiaryid: subsidiaryid, + usertype: usertype }) }); if (!response.ok) throw new Error("HTTP " + response.status); @@ -713,6 +719,8 @@ renderProposalCard(evt.opId, evt.summary); } else if (evt.type === "form_collect") { renderFormCollect(evt.entity, evt.fields || []); + } else if (evt.type === "question") { + renderQuestion(evt.question, evt.options || []); } else if (evt.type === "error") { if (aiMsgId === null) { hideTypingIndicator(); aiMsgId = addMessage('', 'ai'); } aiText += (aiText ? "\n\n" : "") + "⚠️ " + evt.content; @@ -773,7 +781,7 @@ card.find('.op-actions button').prop('disabled', true); card.find('.op-result').text('处理中…'); try { - const res = await fetch(CONFIG.backendUrl + '/api/agent/op/' + opId + '/confirm', { method:'POST' }); + const res = await fetch(CONFIG.backendUrl + '/api/agent/op/' + opId + '/confirm', { method:'POST', headers:{ 'Authorization': authorization } }); const data = await res.json(); if (data.status === 'executed') { card.find('.op-actions').remove(); @@ -790,19 +798,34 @@ async function cancelOp(opId) { const card = $('#op-' + opId); - try { await fetch(CONFIG.backendUrl + '/api/agent/op/' + opId + '/cancel', { method:'POST' }); } catch (e) {} + try { await fetch(CONFIG.backendUrl + '/api/agent/op/' + opId + '/cancel', { method:'POST', headers:{ 'Authorization': authorization } }); } catch (e) {} card.find('.op-actions').remove(); card.find('.op-result').text('已取消'); } - // ====================== FormCollect：对话内动态表单 ====================== + // ====================== FormCollect：对话内动态表单（字段来自 ERP 表单元数据） ====================== + // fields = [{name,label,control,required,default,options?}]（新版对象）或 ["中文名",...]（旧版字符串，兼容） function renderFormCollect(entity, fields) { hideTypingIndicator(); const fid = 'fc-' + Date.now() + '-' + Math.random().toString(36).slice(2, 6); - const inputs = (fields || []).map(f => - `

` + - `
` - ).join(''); + const norm = (fields || []).map(f => (typeof f === 'string') + ? { name: f, label: f, required: false } + : f); + const inputs = norm.map(f => { + const label = escapeHtml(f.label || f.name); + const req = f.required ? ' *' : ''; + const defv = f.default != null ? escapeHtml(String(f.default)) : ''; + let control; + if (Array.isArray(f.options) && f.options.length > 0) { + const opts = [''] + .concat(f.options.map(o => `${escapeHtml(String(o))}`)) + .join(''); + control = ``; + } else { + control = ``; + } + return `
${control}
`; + }).join(''); const html = `
@@ -824,7 +847,34 @@ }); if (parts.length === 0) { alert('请至少填写一个字段'); return; } $(this).prop('disabled', true).text('已提交'); - $('#messageInput').val('为「' + entity + '」新增，请据此生成新增提议：' + parts.join('，')); + $('#messageInput').val('为「' + entity + '」新增，字段如下：' + parts.join('，') + '。请据此用 proposeCreate 生成待确认的新增。'); + sendMessage(); + }); + scrollToBottom(); + } + + // ====================== AskUser：带可点选项的澄清问题 ====================== + function renderQuestion(question, options) { + hideTypingIndicator(); + const qid = 'q-' + Date.now() + '-' + Math.random().toString(36).slice(2, 6); + const chips = (options || []).map(o => + `` + ).join(''); + const html = ` +
+
+
+
❓ ${escapeHtml(question)}
+
${chips}
+
+
+
`; + $('#chatMessages').append(html); + $(`#${qid} .q-opt`).on('click', function () { + const val = $(this).data('val'); + $(`#${qid} .q-opt`).prop('disabled', true).css('opacity', '0.6'); + $(this).css({background:'#667eea', color:'#fff'}); + $('#messageInput').val(String(val)); sendMessage(); }); scrollToBottom(); -- libgit2 0.22.2