From ddcf53cfa85a04867bebba42b4c2b6425b7a2850 Mon Sep 17 00:00:00 2001 From: Andrew Date: Thu, 10 Sep 2026 15:22:07 +0800 Subject: [PATCH] feat(template): Implement template caching and SQL rendering for enhanced query handling - Introduced `TemplateService` for managing SQL templates, allowing for parameterized queries based on user input. - Added functionality to automatically reload templates upon asset creation in `analyst.py`. - Enhanced `CacheService` to support table generation bumping, ensuring cache invalidation on data changes. - Updated `RiskRepository` and `GatewayRepository` to trigger cache invalidation for relevant operations. - Expanded `analyst_schemas.py` to include new fields for template tracking in response metadata. - Created seed SQL script for populating initial templates and added unit tests for template rendering logic. This update significantly improves the efficiency of query handling by leveraging SQL templates, reducing reliance on LLM for common queries. --- app/api/analyst.py | 5 + app/gateway/gateway_repository.py | 6 + app/model/analyst_schemas.py | 2 + app/repository/risk_repository.py | 32 +++- app/service/analyst_agent.py | 60 +++++- app/service/analyst_cache_invalidate.py | 34 ++++ app/service/analytics_repo.py | 66 ++++++- app/service/cache_service.py | 74 +++++++- app/service/template_service.py | 175 ++++++++++++++++++ docs/memory/TODO.md | 3 + .../agent/seed-analyst-query-templates.sql | 51 +++++ tests/test_wave6_analyst_agent.py | 24 +++ tests/test_wave6_analyst_cache.py | 46 ++++- tests/test_wave6_template_service.py | 80 ++++++++ web/src/api/analyst.ts | 2 + 15 files changed, 632 insertions(+), 28 deletions(-) create mode 100644 app/service/analyst_cache_invalidate.py create mode 100644 app/service/template_service.py create mode 100644 scripts/agent/seed-analyst-query-templates.sql create mode 100644 tests/test_wave6_template_service.py diff --git a/app/api/analyst.py b/app/api/analyst.py index 91b9a8f..b6b5404 100644 --- a/app/api/analyst.py +++ b/app/api/analyst.py @@ -114,6 +114,11 @@ def create_asset( if req.kind not in ("dict", "few_shot", "template"): raise HTTPException(status_code=400, detail="kind 必须是 dict/few_shot/template") asset_id = agent.repo.insert_asset(req.kind, req.payload, auth.subject_id) + if req.kind == "template": + try: + agent.templates.reload() + except Exception: + pass return {"ok": True, "id": asset_id, "kind": req.kind} diff --git a/app/gateway/gateway_repository.py b/app/gateway/gateway_repository.py index 2343025..4017133 100644 --- a/app/gateway/gateway_repository.py +++ b/app/gateway/gateway_repository.py @@ -55,3 +55,9 @@ class GatewayRepository: "at": traded_at, }, ) + try: + from app.service.analyst_cache_invalidate import bump_analyst_cache_for_trade + + bump_analyst_cache_for_trade() + except Exception: + pass diff --git a/app/model/analyst_schemas.py b/app/model/analyst_schemas.py index 2535a39..9d7945d 100644 --- a/app/model/analyst_schemas.py +++ b/app/model/analyst_schemas.py @@ -26,6 +26,8 @@ class Meta(BaseModel): exec_ms: int = 0 row_count: int = 0 cache_hit: bool = False + template_hit: bool = False + template_key: str | None = None data_as_of: str | None = None source: str = "" cost_est: float = 0.0 diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py index 397a108..2cee9db 100644 --- a/app/repository/risk_repository.py +++ b/app/repository/risk_repository.py @@ -17,6 +17,24 @@ from sqlalchemy.engine import Engine from app.config.settings import settings from app.utils.db import get_engine + +def _bump_analyst_cache_for_risk_alert() -> None: + try: + from app.service.analyst_cache_invalidate import bump_analyst_cache_for_risk_alert + + bump_analyst_cache_for_risk_alert() + except Exception: + pass + + +def _bump_analyst_cache_for_l3() -> None: + try: + from app.service.analyst_cache_invalidate import bump_analyst_cache_for_l3 + + bump_analyst_cache_for_l3() + except Exception: + pass + EVENT_ALERT_TYPES = ("large_amount", "freq_trade", "pattern") _AUDIT_SQL = text( @@ -137,6 +155,7 @@ class RiskRepository: ) with self._engine.begin() as conn: conn.execute(sql, self._dump_alert(alert)) + _bump_analyst_cache_for_risk_alert() def append_alert_event( self, @@ -191,6 +210,7 @@ class RiskRepository: "aid": alert_id, }, ) + _bump_analyst_cache_for_risk_alert() def get_alert(self, alert_id: str) -> dict | None: with self._engine.connect() as conn: @@ -342,6 +362,7 @@ class RiskRepository: if res.rowcount != 1: return False conn.execute(_AUDIT_SQL, self._dump_audit(audit_entry)) + _bump_analyst_cache_for_risk_alert() return True def list_pending_alerts_all(self, page_size: int = 1000) -> list[dict]: @@ -389,7 +410,10 @@ class RiskRepository: "aid": alert_id, }, ) - return res.rowcount == 1 + bumped = res.rowcount == 1 + if bumped: + _bump_analyst_cache_for_risk_alert() + return bumped # ---------- 代理人行为链（C6 · RISK-008 数据源） ---------- @@ -650,6 +674,7 @@ class RiskRepository: "computed_at": computed_at, }, ) + _bump_analyst_cache_for_l3() def update_l3( self, @@ -686,7 +711,10 @@ class RiskRepository: params["expected"] = expected_computed_at with self._engine.begin() as conn: res = conn.execute(text(sql), params) - return res.rowcount == 1 + ok = res.rowcount == 1 + if ok: + _bump_analyst_cache_for_l3() + return ok # ---------- risk_aml_list ---------- diff --git a/app/service/analyst_agent.py b/app/service/analyst_agent.py index 60e7e13..5f86291 100644 --- a/app/service/analyst_agent.py +++ b/app/service/analyst_agent.py @@ -30,6 +30,7 @@ from app.service.guardrail import GuardrailResult, verify from app.service.llm import DeepSeekLLM, estimate_cost, extract_sql from app.service.schema_meta import SCHEMA_PROMPT from app.service.sql_guard import SqlGuardError, validate +from app.service.template_service import TemplateService SQL_GEN_SYSTEM = ( "你是金融数据查询助手。根据给定表结构与口径，把用户问题翻译成【一条】只读 SELECT SQL。" @@ -56,11 +57,13 @@ class AnalystAgent: repo: AnalyticsRepo | None = None, cache: CacheService | None = None, registry: MetricRegistry | None = None, + templates: TemplateService | None = None, ) -> None: self.llm = llm or DeepSeekLLM() self.repo = repo or AnalyticsRepo() self.cache = cache or CacheService.auto() self.registry = registry or default_registry() + self.templates = templates if templates is not None else TemplateService(repo=self.repo) # ---------- 主入口 ---------- def run( @@ -88,8 +91,16 @@ class AnalystAgent: if amb is not None: return self._clarify(amb, trace_id) - # 2) 生成 SQL - sql_text, usage = self._generate_sql(question, domain, scope) + # 2) 模板填参（D-06）或 LLM 生成 SQL + template_key: str | None = None + template_hit = False + rendered = self.templates.try_render(question, domain, scope, auth.roles) + if rendered is not None: + sql_text, template_key = rendered + template_hit = True + usage: dict = {} + else: + sql_text, usage = self._generate_sql(question, domain, scope) cost_est += estimate_cost(usage) # 3) 校验（五层） @@ -101,7 +112,7 @@ class AnalystAgent: # 4) 执行（缓存优先） perm_fp = self.cache.permission_fingerprint(auth.subject_id, domain, scope) sql_hash = self.cache.sql_hash(sql_text) - cached = self.cache.get_result(perm_fp, sql_text) + cached = self.cache.get_result(perm_fp, sql_text, vres.tables) cache_hit = cached is not None if cache_hit: exec_result, data_as_of = cached @@ -145,6 +156,8 @@ class AnalystAgent: exec_ms=latency, row_count=len(exec_result["rows"]), cache_hit=cache_hit, + template_hit=template_hit, + template_key=template_key, data_as_of=data_as_of, source="jinrong_core", cost_est=round(cost_est, 6), @@ -155,8 +168,19 @@ class AnalystAgent: ) # 7) 留痕（D-04） - self._persist(question, sql_text, sql_hash, resp, auth, session_id, trace_id, - latency, empty_state, guard_result) + self._persist( + question, + sql_text, + sql_hash, + resp, + auth, + session_id, + trace_id, + latency, + empty_state, + guard_result, + template_key=template_key, + ) return resp # ---------- 各步骤 ---------- @@ -283,13 +307,33 @@ class AnalystAgent: trace_id=trace_id, ) - def _persist(self, question, sql, sql_hash, resp, auth, session_id, trace_id, - latency, empty_state, guard_result) -> None: + def _persist( + self, + question, + sql, + sql_hash, + resp, + auth, + session_id, + trace_id, + latency, + empty_state, + guard_result, + *, + template_key: str | None = None, + ) -> None: + if template_key: + sql_source = "template" + elif resp.meta.cache_hit: + sql_source = "cache" + else: + sql_source = "llm" summary = { "status": resp.status, "empty_state": empty_state, "guardrail": "passed" if (guard_result is None or guard_result.passed) else "degraded", - "source": "llm" if not resp.meta.cache_hit else "cache", + "source": sql_source, + "template_key": template_key, } try: self.repo.log_query( diff --git a/app/service/analyst_cache_invalidate.py b/app/service/analyst_cache_invalidate.py new file mode 100644 index 0000000..f13c46e --- /dev/null +++ b/app/service/analyst_cache_invalidate.py @@ -0,0 +1,34 @@ +"""问数结果缓存主动失效（D-06 写侧钩子入口）。 + +事实表变更时 bump 表世代，fail-open（缓存失效失败不阻断业务写）。 +""" +from __future__ import annotations + +from app.service.cache_service import CacheService + +_svc: CacheService | None = None + + +def _service() -> CacheService: + global _svc + if _svc is None: + _svc = CacheService.auto() + return _svc + + +def bump_analyst_cache(*tables: str) -> None: + if not tables: + return + _service().invalidate_tables(tables) + + +def bump_analyst_cache_for_trade() -> None: + bump_analyst_cache("core_trade") + + +def bump_analyst_cache_for_risk_alert() -> None: + bump_analyst_cache("risk_alert") + + +def bump_analyst_cache_for_l3() -> None: + bump_analyst_cache("customer_profile_l3") diff --git a/app/service/analytics_repo.py b/app/service/analytics_repo.py index 4c12712..bc7abd3 100644 --- a/app/service/analytics_repo.py +++ b/app/service/analytics_repo.py @@ -13,6 +13,17 @@ import pymysql from app.config.settings import settings +def _parse_json_field(value: Any) -> Any: + if value is None: + return None + if isinstance(value, (dict, list)): + return value + try: + return json.loads(value) + except (TypeError, json.JSONDecodeError): + return None + + class AnalyticsRepo: def __init__(self, host=None, port=None, user=None, password=None) -> None: self.host = host or settings.mysql_host @@ -137,6 +148,44 @@ class AnalyticsRepo: finally: conn.close() + # ---------- 模板资产（D-06 / D-11） ---------- + def list_published_templates(self) -> list: + """加载 status=published 的 SQL 模板（模板缓存池）。""" + from app.service.template_service import QueryTemplate + + conn = self._conn(settings.mysql_database) + rows: list[dict] = [] + try: + with conn.cursor() as cur: + cur.execute( + """ + SELECT template_key, template_sql, params_schema, tags, gray_roles + FROM analytics_query_template + WHERE status = 'published' + ORDER BY id DESC + """ + ) + rows = list(cur.fetchall()) + except Exception: + rows = [] + finally: + conn.close() + out: list[QueryTemplate] = [] + for row in rows: + schema = row.get("params_schema") + tags = row.get("tags") + gray = row.get("gray_roles") + out.append( + QueryTemplate( + template_key=row["template_key"], + template_sql=row["template_sql"], + params_schema=_parse_json_field(schema) or {}, + tags=_parse_json_field(tags), + gray_roles=_parse_json_field(gray), + ) + ) + return out + # ---------- 资产沉淀（D-11） ---------- def insert_asset(self, kind: str, payload: dict, created_by: str) -> int: table = {"dict": "analytics_metric_dict", "few_shot": "analytics_few_shot", "template": "analytics_query_template"}[kind] @@ -155,8 +204,21 @@ class AnalyticsRepo: ) else: cur.execute( - f"INSERT INTO {table} (template_key, template_sql, created_by) VALUES (%s,%s,%s)", - (payload.get("template_key"), payload.get("template_sql"), created_by), + f"INSERT INTO {table} (template_key, template_sql, params_schema, tags, created_by, status, published_by) " + f"VALUES (%s,%s,%s,%s,%s,%s,%s)", + ( + payload.get("template_key"), + payload.get("template_sql"), + json.dumps(payload.get("params_schema"), ensure_ascii=False) + if payload.get("params_schema") + else None, + json.dumps(payload.get("tags"), ensure_ascii=False) + if payload.get("tags") + else None, + created_by, + payload.get("status", "draft"), + created_by if payload.get("status") == "published" else None, + ), ) return int(cur.lastrowid or 0) finally: diff --git a/app/service/cache_service.py b/app/service/cache_service.py index 5960eb3..7a9910c 100644 --- a/app/service/cache_service.py +++ b/app/service/cache_service.py @@ -1,6 +1,6 @@ """查询缓存（D-06）：结果缓存 + 模板缓存，Redis 可选（未连上则用内存降级）。 -键构造包含权限指纹，TTL 按表分层：交易类 5min / 台账类 1h / 基础信息类当日。 +键构造包含权限指纹与表世代（写侧 bump 主动失效），TTL 按表分层。 """ from __future__ import annotations @@ -8,7 +8,7 @@ import hashlib import json import time from dataclasses import dataclass -from typing import Any +from typing import Any, Iterable from app.config.settings import settings @@ -21,9 +21,12 @@ TABLES_TTL = { "core_customer": 24 * 60 * 60, "core_product": 24 * 60 * 60, "core_product_nav": 24 * 60 * 60, + "customer_profile_l3": 60 * 60, } DEFAULT_TTL = 60 * 60 +GEN_PREFIX = "cache:analyst:gen:" + class CacheBackend: def get(self, key: str) -> Any | None: @@ -35,12 +38,19 @@ class CacheBackend: def delete(self, key: str) -> None: raise NotImplementedError + def get_counter(self, key: str) -> int | None: + raise NotImplementedError + + def incr_counter(self, key: str) -> int: + raise NotImplementedError + class InMemoryBackend(CacheBackend): """内存缓存（Redis 不可用时的降级）。""" def __init__(self) -> None: self._store: dict[str, tuple[float, Any]] = {} + self._counters: dict[str, int] = {} def get(self, key: str) -> Any | None: item = self._store.get(key) @@ -58,6 +68,14 @@ class InMemoryBackend(CacheBackend): def delete(self, key: str) -> None: self._store.pop(key, None) + def get_counter(self, key: str) -> int | None: + return self._counters.get(key) + + def incr_counter(self, key: str) -> int: + n = self._counters.get(key, 0) + 1 + self._counters[key] = n + return n + class RedisBackend(CacheBackend): def __init__(self, url: str) -> None: @@ -89,6 +107,24 @@ class RedisBackend(CacheBackend): except Exception: pass + def get_counter(self, key: str) -> int | None: + try: + raw = self._r.get(key) + except Exception: + return None + if raw is None: + return None + try: + return int(raw) + except (TypeError, ValueError): + return None + + def incr_counter(self, key: str) -> int: + try: + return int(self._r.incr(key)) + except Exception: + return 1 + @dataclass class CacheService: @@ -113,8 +149,19 @@ class CacheService: raw = f"{subject_id}|{domain}|{','.join(sorted(scope or []))}" return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:12] - def result_key(self, perm_fp: str, sql_hash: str) -> str: - return f"cache:analyst:result:{perm_fp}:{sql_hash}" + def _gen_key(self, table: str) -> str: + return f"{GEN_PREFIX}{table}" + + def table_generation(self, table: str) -> int: + val = self.backend.get_counter(self._gen_key(table)) + return val if val is not None else 0 + + def _generation_suffix(self, tables: list[str]) -> str: + parts = [f"{t}={self.table_generation(t)}" for t in sorted(set(tables))] + return ":".join(parts) + + def result_key(self, perm_fp: str, sql_hash: str, tables: list[str]) -> str: + return f"cache:analyst:result:{perm_fp}:{sql_hash}:{self._generation_suffix(tables)}" def ttl_for(self, tables: list[str]) -> int: for t in tables: @@ -122,11 +169,20 @@ class CacheService: return TABLES_TTL[t] return DEFAULT_TTL - def get_result(self, perm_fp: str, sql: str) -> Any | None: - return self.backend.get(self.result_key(perm_fp, self.sql_hash(sql))) + def get_result(self, perm_fp: str, sql: str, tables: list[str]) -> Any | None: + return self.backend.get(self.result_key(perm_fp, self.sql_hash(sql), tables)) def set_result(self, perm_fp: str, sql: str, value: Any, tables: list[str]) -> None: - self.backend.set(self.result_key(perm_fp, self.sql_hash(sql)), value, self.ttl_for(tables)) + self.backend.set( + self.result_key(perm_fp, self.sql_hash(sql), tables), + value, + self.ttl_for(tables), + ) - def invalidate_by_sql(self, perm_fp: str, sql: str) -> None: - self.backend.delete(self.result_key(perm_fp, self.sql_hash(sql))) + def invalidate_tables(self, tables: Iterable[str]) -> None: + """写侧主动失效：bump 表世代，使依赖该表的结果缓存全部 miss。""" + for table in set(tables): + self.backend.incr_counter(self._gen_key(table)) + + def invalidate_by_sql(self, perm_fp: str, sql: str, tables: list[str]) -> None: + self.backend.delete(self.result_key(perm_fp, self.sql_hash(sql), tables)) diff --git a/app/service/template_service.py b/app/service/template_service.py new file mode 100644 index 0000000..6b9481b --- /dev/null +++ b/app/service/template_service.py @@ -0,0 +1,175 @@ +"""问数 SQL 模板匹配与填参（D-06 模板缓存 / D-11 资产联动）。 + +published 模板由 analytics_query_template 加载；命中后跳过 LLM 生成 SQL。 +""" +from __future__ import annotations + +import json +import re +from dataclasses import dataclass, field +from typing import Any + +from app.service.analytics_repo import AnalyticsRepo + +RECENT_DAYS_PATTERNS = ( + re.compile(r"近\s*(\d+)\s*天"), + re.compile(r"近\s*(\d+)\s*日"), + re.compile(r"最近\s*(\d+)\s*天"), + re.compile(r"(\d+)\s*天内"), +) + + +@dataclass +class QueryTemplate: + template_key: str + template_sql: str + params_schema: dict[str, Any] = field(default_factory=dict) + tags: list[str] | None = None + gray_roles: list[str] | None = None + + +def _parse_json(value: Any) -> Any: + if value is None: + return None + if isinstance(value, (dict, list)): + return value + try: + return json.loads(value) + except (TypeError, json.JSONDecodeError): + return None + + +def _role_allowed(template: QueryTemplate, roles: list[str]) -> bool: + gray = template.gray_roles + if not gray: + return True + return any(r in gray for r in roles) + + +def _question_matches(question: str, template: QueryTemplate) -> bool: + schema = template.params_schema or {} + match_all = schema.get("match_all") or template.tags or [] + if not match_all: + return False + return all(kw in question for kw in match_all) + + +def _extract_recent_days(question: str, default: int) -> int: + for pat in RECENT_DAYS_PATTERNS: + m = pat.search(question) + if m: + return max(1, min(int(m.group(1)), 366)) + if any(x in question for x in ("近一月", "近一个月", "最近一个月", "近30天")): + return 30 + if any(x in question for x in ("近一周", "最近一周", "近7天")): + return 7 + return default + + +def _sanitize_int(value: int, *, lo: int = 1, hi: int = 366) -> int: + return max(lo, min(int(value), hi)) + + +def _sanitize_customer_id(value: str) -> str | None: + v = (value or "").strip() + if re.fullmatch(r"CUST-\d+", v): + return v + return None + + +def _extract_param(name: str, extract: str, question: str, default: Any, scope: list[str]) -> Any: + if extract == "recent_days": + return _extract_recent_days(question, int(default or 30)) + if extract == "scope_customer": + if scope: + return scope[0] + return default + return default + + +def render_template_sql( + template: QueryTemplate, + question: str, + *, + domain: str, + scope: list[str], +) -> str | None: + """填参渲染；参数非法时返回 None。""" + sql = template.template_sql + schema = template.params_schema or {} + for spec in schema.get("params") or []: + placeholder = spec.get("placeholder") or f":{spec.get('name')}" + if placeholder not in sql: + continue + raw = _extract_param( + spec.get("name", ""), + spec.get("extract", ""), + question, + spec.get("default"), + scope, + ) + if spec.get("extract") == "recent_days": + value = str(_sanitize_int(int(raw))) + elif spec.get("extract") == "scope_customer": + cid = _sanitize_customer_id(str(raw or "")) + if not cid: + return None + value = cid + elif isinstance(raw, int): + value = str(_sanitize_int(raw)) + else: + value = str(raw) + if not value: + return None + sql = sql.replace(placeholder, value) + if re.search(r":\w+", sql): + return None + return sql + + +class TemplateService: + def __init__( + self, + repo: AnalyticsRepo | None = None, + templates: list[QueryTemplate] | None = None, + ) -> None: + self.repo = repo or AnalyticsRepo() + self._fixed_templates = templates + self._loaded: list[QueryTemplate] | None = templates + + def _templates(self) -> list[QueryTemplate]: + if self._fixed_templates is not None: + return self._fixed_templates + if self._loaded is None: + self._loaded = self.repo.list_published_templates() + return self._loaded + + def reload(self) -> None: + if self._fixed_templates is None: + self._loaded = None + + def try_render( + self, + question: str, + domain: str, + scope: list[str], + roles: list[str], + ) -> tuple[str, str] | None: + """匹配 published 模板并填参；返回 (sql, template_key) 或 None。""" + q = question.strip() + if not q: + return None + candidates = [ + t + for t in self._templates() + if _role_allowed(t, roles) and _question_matches(q, t) + ] + candidates.sort( + key=lambda t: len((t.params_schema or {}).get("match_all") or t.tags or []), + reverse=True, + ) + for tpl in candidates: + sql = render_template_sql(tpl, q, domain=domain, scope=scope) + if sql: + return sql, tpl.template_key + return None diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md index 215774a..f619d41 100644 --- a/docs/memory/TODO.md +++ b/docs/memory/TODO.md @@ -99,6 +99,9 @@ - [ ] **20 题 live battery 跑分**（见上方「本批次 · 优先收尾」） - [ ] **agent 编排接流水 Ambiguity**：`dict_service` 已返 Ambiguity · 确认 `analyst_agent` 反问链路在 live 20 题里生效（Q 含「流水」） +- [x] **D-06 写侧主动失效**（2026-09-10）：表世代 bump · 模拟交易 `core_trade` · 预警/L3 写侧 · `invalidate_tables` + 单测 +- [x] **D-06 模板缓存**（2026-09-10）：`template_service` 匹配+填参 · 问数编排跳过 LLM · 种子 `seed-analyst-query-templates.sql` +- [ ] **D-06 PII 脱敏后再缓存** — 暂缓（全仓脱敏方案未定） - [ ] **已知挂账（迭代文档）**：Q17 `create` 误杀 · Q7 无城市字段静默改职业 · 待排期 - [ ] **远程分支同步**（可选）：本地 `data-analysis-agent` @ `fd9464d` 落后 `xinghuo/data-analysis-agent` 2 commit diff --git a/scripts/agent/seed-analyst-query-templates.sql b/scripts/agent/seed-analyst-query-templates.sql new file mode 100644 index 0000000..dcddbc7 --- /dev/null +++ b/scripts/agent/seed-analyst-query-templates.sql @@ -0,0 +1,51 @@ +-- 数据分析 Agent · 已发布 SQL 模板种子（D-06 模板缓存演示） +-- 前置：migrate-analyst-d07-d11.sql 已执行 +-- 用法：mysql -u root -p jinrong_agent < scripts/agent/seed-analyst-query-templates.sql + +USE jinrong_agent; + +DELETE FROM analytics_query_template WHERE created_by = 'SEED'; + +INSERT INTO analytics_query_template + (template_key, template_sql, params_schema, tags, status, version, created_by, published_by) +VALUES +( + 'customer_total_count', + 'SELECT COUNT(*) AS customer_count FROM core_customer', + JSON_OBJECT('match_all', JSON_ARRAY('客户', '总数')), + JSON_ARRAY('客户', '总数'), + 'published', + 1, + 'SEED', + 'SEED' +), +( + 'subscribe_amount_recent_days', + 'SELECT COALESCE(SUM(amount), 0) AS subscribe_total FROM core_trade WHERE trade_type = ''subscribe'' AND trade_date >= DATE_SUB(CURDATE(), INTERVAL :days DAY)', + JSON_OBJECT( + 'match_all', JSON_ARRAY('申购', '金额'), + 'params', JSON_ARRAY( + JSON_OBJECT( + 'name', 'days', + 'placeholder', ':days', + 'extract', 'recent_days', + 'default', 30 + ) + ) + ), + JSON_ARRAY('申购', '金额', '近30天'), + 'published', + 1, + 'SEED', + 'SEED' +), +( + 'pending_alert_count', + 'SELECT COUNT(*) AS pending_count FROM jinrong_agent.risk_alert WHERE status = ''pending_review''', + JSON_OBJECT('match_all', JSON_ARRAY('待处理', '预警')), + JSON_ARRAY('待处理', '预警'), + 'published', + 1, + 'SEED', + 'SEED' +); diff --git a/tests/test_wave6_analyst_agent.py b/tests/test_wave6_analyst_agent.py index f4beb5b..a2b1158 100644 --- a/tests/test_wave6_analyst_agent.py +++ b/tests/test_wave6_analyst_agent.py @@ -5,6 +5,7 @@ import pytest from app.api.analyst_auth_adapter import AnalystAuthContext from app.service.analyst_agent import AnalystAgent +from app.service.template_service import QueryTemplate, TemplateService class FakeLLM: @@ -44,6 +45,9 @@ class FakeRepo: def log_audit(self, **kw): pass + def list_published_templates(self): + return [] + def ctx(roles, subject="STAFF-A", *, token_type="staff", customer_id=None): return AnalystAuthContext( @@ -112,6 +116,26 @@ class TestAgentOrchestration(unittest.TestCase): self.assertEqual(resp.status, "success") self.assertIn("AI 分析有风险", resp.answer) + def test_template_hit_skips_llm_sql(self): + tpl = QueryTemplate( + template_key="customer_total_count", + template_sql="SELECT COUNT(*) AS c FROM core_customer", + params_schema={"match_all": ["客户", "总数"]}, + ) + repo = FakeRepo(rows=[[33]], columns=["c"]) + llm = FakeLLM("SELECT 1", ["共 33 个客户"]) + agent = AnalystAgent( + llm=llm, + repo=repo, + templates=TemplateService(templates=[tpl]), + ) + resp = agent.run("客户总数是多少", ctx(["analyst"])) + self.assertEqual(resp.status, "success") + self.assertTrue(resp.meta.template_hit) + self.assertEqual(resp.meta.template_key, "customer_total_count") + self.assertIn("COUNT(*)", resp.sql) + self.assertEqual(llm.calls, 1) + @pytest.mark.integration class TestAgentReal(unittest.TestCase): diff --git a/tests/test_wave6_analyst_cache.py b/tests/test_wave6_analyst_cache.py index b6170da..c32e17f 100644 --- a/tests/test_wave6_analyst_cache.py +++ b/tests/test_wave6_analyst_cache.py @@ -26,16 +26,37 @@ class TestCacheService(unittest.TestCase): def test_roundtrip(self): fp = self.svc.permission_fingerprint("S1", "full") sql = "SELECT COUNT(*) FROM core_customer" - self.assertIsNone(self.svc.get_result(fp, sql)) - self.svc.set_result(fp, sql, {"cnt": 33}, ["core_customer"]) - self.assertEqual(self.svc.get_result(fp, sql), {"cnt": 33}) + tables = ["core_customer"] + self.assertIsNone(self.svc.get_result(fp, sql, tables)) + self.svc.set_result(fp, sql, {"cnt": 33}, tables) + self.assertEqual(self.svc.get_result(fp, sql, tables), {"cnt": 33}) def test_invalidate(self): fp = self.svc.permission_fingerprint("S1", "full") - sql = "SELECT 1" - self.svc.set_result(fp, sql, {"x": 1}, ["core_customer"]) - self.svc.invalidate_by_sql(fp, sql) - self.assertIsNone(self.svc.get_result(fp, sql)) + sql = "SELECT 1 FROM core_customer" + tables = ["core_customer"] + self.svc.set_result(fp, sql, {"x": 1}, tables) + self.svc.invalidate_by_sql(fp, sql, tables) + self.assertIsNone(self.svc.get_result(fp, sql, tables)) + + def test_invalidate_tables_bumps_generation(self): + fp = self.svc.permission_fingerprint("S1", "full") + sql = "SELECT COUNT(*) FROM core_trade" + tables = ["core_trade"] + self.svc.set_result(fp, sql, {"cnt": 1}, tables) + self.assertEqual(self.svc.get_result(fp, sql, tables), {"cnt": 1}) + self.svc.invalidate_tables(["core_trade"]) + self.assertIsNone(self.svc.get_result(fp, sql, tables)) + self.svc.set_result(fp, sql, {"cnt": 2}, tables) + self.assertEqual(self.svc.get_result(fp, sql, tables), {"cnt": 2}) + + def test_invalidate_tables_skips_unrelated(self): + fp = self.svc.permission_fingerprint("S1", "full") + sql = "SELECT COUNT(*) FROM core_customer" + tables = ["core_customer"] + self.svc.set_result(fp, sql, {"cnt": 9}, tables) + self.svc.invalidate_tables(["core_trade"]) + self.assertEqual(self.svc.get_result(fp, sql, tables), {"cnt": 9}) def test_inmemory_expiry(self): b = InMemoryBackend() @@ -44,6 +65,17 @@ class TestCacheService(unittest.TestCase): time.sleep(1.2) self.assertIsNone(b.get("k")) + def test_bump_analyst_cache_helper(self): + from app.service import analyst_cache_invalidate + + analyst_cache_invalidate._svc = self.svc + fp = self.svc.permission_fingerprint("S1", "full") + sql = "SELECT COUNT(*) FROM core_trade" + tables = ["core_trade"] + self.svc.set_result(fp, sql, {"cnt": 1}, tables) + analyst_cache_invalidate.bump_analyst_cache_for_trade() + self.assertIsNone(self.svc.get_result(fp, sql, tables)) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_wave6_template_service.py b/tests/test_wave6_template_service.py new file mode 100644 index 0000000..a240eb6 --- /dev/null +++ b/tests/test_wave6_template_service.py @@ -0,0 +1,80 @@ +"""模板缓存（D-06 第二层）单元测试。""" +import unittest + +from app.service.template_service import QueryTemplate, TemplateService + + +CUSTOMER_COUNT = QueryTemplate( + template_key="customer_total_count", + template_sql="SELECT COUNT(*) AS customer_count FROM core_customer", + params_schema={"match_all": ["客户", "总数"]}, + tags=["客户", "总数"], +) + +SUBSCRIBE_DAYS = QueryTemplate( + template_key="subscribe_amount_recent_days", + template_sql=( + "SELECT COALESCE(SUM(amount), 0) AS subscribe_total FROM core_trade " + "WHERE trade_type = 'subscribe' " + "AND trade_date >= DATE_SUB(CURDATE(), INTERVAL :days DAY)" + ), + params_schema={ + "match_all": ["申购", "金额"], + "params": [{"name": "days", "placeholder": ":days", "extract": "recent_days", "default": 30}], + }, + tags=["申购", "金额"], +) + + +class TestTemplateService(unittest.TestCase): + def setUp(self): + self.svc = TemplateService(templates=[CUSTOMER_COUNT, SUBSCRIBE_DAYS]) + + def test_match_customer_count(self): + hit = self.svc.try_render("平台客户总数是多少", "full", [], ["analyst"]) + self.assertIsNotNone(hit) + sql, key = hit + self.assertEqual(key, "customer_total_count") + self.assertIn("COUNT(*)", sql) + + def test_match_subscribe_with_days(self): + hit = self.svc.try_render("近7天申购金额总额", "full", [], ["analyst"]) + self.assertIsNotNone(hit) + sql, key = hit + self.assertEqual(key, "subscribe_amount_recent_days") + self.assertIn("INTERVAL 7 DAY", sql) + self.assertNotIn(":days", sql) + + def test_default_days_when_unspecified(self): + hit = self.svc.try_render("申购金额汇总", "full", [], ["analyst"]) + self.assertIsNotNone(hit) + sql, _ = hit + self.assertIn("INTERVAL 30 DAY", sql) + + def test_no_match_returns_none(self): + self.assertIsNone(self.svc.try_render("随便问问", "full", [], ["analyst"])) + + def test_self_domain_injects_customer_id(self): + tpl = QueryTemplate( + template_key="self_trade_count", + template_sql=( + "SELECT COUNT(*) AS cnt FROM core_trade " + "WHERE customer_id = ':customer_id'" + ), + params_schema={ + "match_all": ["交易", "多少"], + "params": [ + {"name": "customer_id", "placeholder": ":customer_id", "extract": "scope_customer", "default": ""} + ], + }, + tags=[], + ) + svc = TemplateService(templates=[tpl]) + hit = svc.try_render("我有多少笔交易", "self", ["CUST-9527"], ["customer"]) + self.assertIsNotNone(hit) + sql, _ = hit + self.assertIn("CUST-9527", sql) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/src/api/analyst.ts b/web/src/api/analyst.ts index c7383c2..168d8e5 100644 --- a/web/src/api/analyst.ts +++ b/web/src/api/analyst.ts @@ -9,6 +9,8 @@ export type AnalystMeta = { exec_ms: number row_count: number cache_hit: boolean + template_hit?: boolean + template_key?: string | null data_as_of?: string source?: string cost_est?: number