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