diff --git a/app/api/analyst.py b/app/api/analyst.py
index a54824f..b748cae 100644
--- a/app/api/analyst.py
+++ b/app/api/analyst.py
@@ -3,14 +3,16 @@ from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException
-from app.api.deps import get_auth_context
-from app.model.schemas.analyst import (
- AnalystResponse,
- AssetCreateRequest,
- ChatRequest,
+from app.api.analyst_auth_adapter import (
+ AnalystAuthContext,
+ AnalystAuthError,
+ analyst_auth_from_deps,
+ assert_analyst_query_access,
)
+from app.api.deps import AuthContext, get_auth_context
+from app.model.analyst_schemas import AnalystResponse, AssetCreateRequest, ChatRequest
from app.service.analyst_agent import AnalystAgent
-from app.utils.auth import AuthContext, assert_analyst_access
+from app.utils.trace import current_trace
router = APIRouter(prefix="/api/analyst", tags=["analyst"])
@@ -24,10 +26,14 @@ def get_agent() -> AnalystAgent:
return _agent
+def _analyst_ctx(auth: AuthContext = Depends(get_auth_context)) -> AnalystAuthContext:
+ return analyst_auth_from_deps(auth, trace_id=current_trace() or "")
+
+
@router.post("/chat", response_model=AnalystResponse)
def chat(
req: ChatRequest,
- auth: AuthContext = Depends(get_auth_context),
+ auth: AnalystAuthContext = Depends(_analyst_ctx),
agent: AnalystAgent = Depends(get_agent),
) -> AnalystResponse:
return agent.run(req.question, auth, req.session_id, req.trace_id)
@@ -35,16 +41,23 @@ def chat(
@router.get("/dashboard")
def dashboard(
- auth: AuthContext = Depends(get_auth_context),
+ auth: AnalystAuthContext = Depends(_analyst_ctx),
agent: AnalystAgent = Depends(get_agent),
):
"""智能看数板（D-12）后端：按角色返回卡片。"""
- domain = assert_analyst_access(auth)
- role = next((r for r in auth.roles if r in ("analyst", "advisor", "risk_officer", "ops")), "analyst")
+ try:
+ domain = assert_analyst_query_access(auth)
+ except AnalystAuthError as exc:
+ raise HTTPException(status_code=403, detail=exc.message) from exc
+ role = next(
+ (r for r in auth.roles if r in ("analyst", "advisor", "risk_officer", "ops", "customer")),
+ "analyst",
+ )
cards = {
"analyst": ["客户总数", "总持仓规模", "今日交易笔数与金额", "待处理预警数", "口径字典资产数"],
"advisor": ["名下客户数", "名下资产规模", "盈亏分布", "风险等级分布"],
"risk_officer": ["待处理预警数", "预警按类型分布", "近7天新增趋势"],
+ "customer": ["我的持仓规模", "近30日交易笔数", "风险等级", "盈亏概览"],
"ops": ["近30天申购金额", "近30天赎回金额", "各产品类型规模TOP"],
}.get(role, [])
metrics: dict = {}
@@ -65,7 +78,7 @@ def dashboard(
@router.post("/assets")
def create_asset(
req: AssetCreateRequest,
- auth: AuthContext = Depends(get_auth_context),
+ auth: AnalystAuthContext = Depends(_analyst_ctx),
agent: AnalystAgent = Depends(get_agent),
):
"""沉淀资产（D-11）：仅分析师可写。"""
@@ -79,7 +92,7 @@ def create_asset(
@router.get("/ops/metrics")
def ops_metrics(
- auth: AuthContext = Depends(get_auth_context),
+ auth: AnalystAuthContext = Depends(_analyst_ctx),
agent: AnalystAgent = Depends(get_agent),
):
"""运营指标（N-08，P1 最小版）。"""
diff --git a/app/api/analyst_auth_adapter.py b/app/api/analyst_auth_adapter.py
new file mode 100644
index 0000000..1c0e404
--- /dev/null
+++ b/app/api/analyst_auth_adapter.py
@@ -0,0 +1,81 @@
+"""deps.AuthContext → 数据分析 Agent 内部视图（S3 接缝）。"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+from app.api.deps import AuthContext as DepsAuthContext
+from app.utils.trace import current_trace
+
+# Demo 四角色 + ops 预留；问数线宽于 chat 矩阵（customer 可进）
+ROLE_DATA_DOMAIN: dict[str, str] = {
+ "customer": "self",
+ "advisor": "assigned",
+ "analyst": "full",
+ "risk_officer": "risk",
+ "ops": "aggregate",
+}
+
+_ROLE_PRIORITY = ("customer", "advisor", "analyst", "risk_officer", "ops")
+
+
+class AnalystAuthError(Exception):
+ def __init__(self, error_code: str, message: str) -> None:
+ super().__init__(message)
+ self.error_code = error_code
+ self.message = message
+
+
+@dataclass
+class AnalystAuthContext:
+ """分析线内部身份（subject_id 对齐 deps.actor_id）。"""
+
+ subject_id: str
+ token_type: str
+ roles: list[str]
+ permissions: list[str] = field(default_factory=list)
+ customer_id: str | None = None
+ trace_id: str = ""
+ agent_type: str = "analyst"
+
+ def has_role(self, role: str) -> bool:
+ return role in self.roles
+
+
+def analyst_auth_from_deps(
+ ctx: DepsAuthContext,
+ *,
+ trace_id: str | None = None,
+) -> AnalystAuthContext:
+ return AnalystAuthContext(
+ subject_id=ctx.actor_id,
+ token_type=ctx.token_type,
+ roles=list(ctx.roles),
+ permissions=list(ctx.permissions),
+ customer_id=ctx.customer_id,
+ trace_id=trace_id or current_trace() or "",
+ )
+
+
+def assert_analyst_query_access(ctx: AnalystAuthContext) -> str:
+ """问数入口鉴权：返回 sql_guard 数据域 key。失败抛 AnalystAuthError。"""
+ if ctx.token_type == "customer" or "customer" in ctx.roles:
+ if not ctx.customer_id:
+ raise AnalystAuthError("AUTH_403_ROLE", "客户 token 缺少 customer_id")
+ return "self"
+ if ctx.token_type != "staff":
+ raise AnalystAuthError("AUTH_403_ROLE", "数据分析问数需有效登录身份")
+ for role in _ROLE_PRIORITY:
+ if role in ctx.roles and role in ROLE_DATA_DOMAIN and role != "customer":
+ return ROLE_DATA_DOMAIN[role]
+ raise AnalystAuthError("AUTH_403_ROLE", "当前角色无权使用数据分析问数")
+
+
+def resolve_analyst_scope(ctx: AnalystAuthContext, domain: str, repo: Any) -> list[str]:
+ """按域解析 customer_id 白名单（供 sql_guard）。"""
+ if domain == "self":
+ cid = ctx.customer_id or ctx.subject_id
+ return [cid] if cid else []
+ if domain == "assigned":
+ return repo.resolve_advisor_scope(ctx.subject_id)
+ return []
diff --git a/app/config/settings.py b/app/config/settings.py
index cc03e14..17ea701 100644
--- a/app/config/settings.py
+++ b/app/config/settings.py
@@ -103,12 +103,10 @@ class Settings(BaseSettings):
# ===== 代销平台 API（v0.1）=====
platform_response_desensitize: bool = False
- # 数据分析 Agent
- jwt_dev_secret: str = "dev-only-change-me"
- jwt_token_ttl_hours: int = 24
- sql_max_rows: int = 1000
- sql_timeout_s: int = 10
- guardrail_retry_times: int = 1
+ # ===== 数据分析 Agent（NL2SQL 问数线）=====
+ analyst_sql_max_rows: int = 1000
+ analyst_sql_timeout_s: int = 10
+ analyst_guardrail_retry_times: int = 1
settings = Settings()
diff --git a/app/model/schemas/analyst.py b/app/model/analyst_schemas.py
similarity index 96%
rename from app/model/schemas/analyst.py
rename to app/model/analyst_schemas.py
index fe5b3ce..2535a39 100644
--- a/app/model/schemas/analyst.py
+++ b/app/model/analyst_schemas.py
@@ -8,6 +8,8 @@ DISCLAIMER = (
"据此操作风险自负，请谨慎对待。"
)
+CUSTOMER_AI_RISK_NOTE = "AI 分析有风险，仅供参考。"
+
class ChatRequest(BaseModel):
question: str = Field(..., description="自然语言问题")
diff --git a/app/model/schemas/__init__.py b/app/model/schemas/__init__.py
deleted file mode 100644
index 4875eba..0000000
--- a/app/model/schemas/__init__.py
+++ /dev/null
@@ -1,20 +0,0 @@
-"""数据分析 Agent Pydantic 模型（输出四件套等）。"""
-from app.model.schemas.analyst import (
- DISCLAIMER,
- AnalystResponse,
- AssetCreateRequest,
- ChatRequest,
- Meta,
- SampleRequest,
- TableData,
-)
-
-__all__ = [
- "DISCLAIMER",
- "AnalystResponse",
- "AssetCreateRequest",
- "ChatRequest",
- "Meta",
- "SampleRequest",
- "TableData",
-]
diff --git a/app/service/analyst_agent.py b/app/service/analyst_agent.py
index b4f51d4..60e7e13 100644
--- a/app/service/analyst_agent.py
+++ b/app/service/analyst_agent.py
@@ -10,7 +10,19 @@ import time
import uuid
from typing import Any
-from app.model.schemas.analyst import AnalystResponse, Meta, TableData
+from app.model.analyst_schemas import (
+ CUSTOMER_AI_RISK_NOTE,
+ DISCLAIMER,
+ AnalystResponse,
+ Meta,
+ TableData,
+)
+from app.api.analyst_auth_adapter import (
+ AnalystAuthContext,
+ AnalystAuthError,
+ assert_analyst_query_access,
+ resolve_analyst_scope,
+)
from app.service.analytics_repo import AnalyticsRepo, classify_empty
from app.service.cache_service import CacheService
from app.service.dict_service import Ambiguity, MetricRegistry, default_registry
@@ -18,7 +30,6 @@ 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.utils.auth import AuthContext, assert_analyst_access
SQL_GEN_SYSTEM = (
"你是金融数据查询助手。根据给定表结构与口径，把用户问题翻译成【一条】只读 SELECT SQL。"
@@ -31,6 +42,12 @@ ANSWER_SYSTEM = (
"结尾无需重复免责声明（由系统统一附带）。若结果为空，如实说明。"
)
+CUSTOMER_ANSWER_SYSTEM = (
+ "你是面向个人客户的财富数据解读助手。只能根据查询结果描述趋势、分布与数量，"
+ "不得给出投资建议、收益承诺、买卖时点或产品推荐。若用户问题涉及建议，只描述数据事实。"
+ "数字必须与结果完全一致，不得编造。"
+)
+
class AnalystAgent:
def __init__(
@@ -49,7 +66,7 @@ class AnalystAgent:
def run(
self,
question: str,
- auth: AuthContext,
+ auth: AnalystAuthContext,
session_id: str | None = None,
trace_id: str | None = None,
) -> AnalystResponse:
@@ -60,13 +77,11 @@ class AnalystAgent:
cost_est = 0.0
try:
- domain = assert_analyst_access(auth)
- except Exception as exc: # noqa: BLE001
- return self._deny(str(getattr(exc, "error_code", "AUTH_403_ROLE")), str(exc), trace_id)
+ domain = assert_analyst_query_access(auth)
+ except AnalystAuthError as exc:
+ return self._deny(exc.error_code, exc.message, trace_id)
- scope: list[str] = []
- if domain == "assigned":
- scope = self.repo.resolve_advisor_scope(auth.subject_id)
+ scope: list[str] = resolve_analyst_scope(auth, domain, self.repo)
# 1) 指标消歧（N-01）
amb = self._detect_ambiguity(question)
@@ -107,7 +122,7 @@ class AnalystAgent:
# 5) 解读 + 数字护栏（D-10）
try:
answer, guard_result, g_usage = self._generate_verified_answer(
- question, sql_text, table, empty_state
+ question, sql_text, table, empty_state, domain=domain
)
cost_est += estimate_cost(g_usage)
except Exception as exc: # noqa: BLE001
@@ -117,6 +132,11 @@ class AnalystAgent:
status = "degrade" if (guard_result is not None and not guard_result.passed) else "success"
if status == "degrade":
answer = "解读校验未通过，请以下方表格数据为准。"
+ disclaimer = DISCLAIMER
+ if domain == "self":
+ disclaimer = f"{DISCLAIMER} {CUSTOMER_AI_RISK_NOTE}"
+ if CUSTOMER_AI_RISK_NOTE not in answer:
+ answer = f"{answer.rstrip()} {CUSTOMER_AI_RISK_NOTE}"
resp = AnalystResponse(
answer=answer,
table=table,
@@ -129,6 +149,7 @@ class AnalystAgent:
source="jinrong_core",
cost_est=round(cost_est, 6),
),
+ disclaimer=disclaimer,
status=status,
trace_id=trace_id,
)
@@ -161,6 +182,10 @@ class AnalystAgent:
scope_hint = "无限制"
if domain == "assigned":
scope_hint = f"只能查询以下客户：customer_id IN ({', '.join(scope)})；涉及客户的查询必须带此过滤"
+ elif domain == "self" and scope:
+ scope_hint = (
+ f"只能查询客户 {scope[0]} 本人的数据；涉及客户表必须带 customer_id = '{scope[0]}' 条件"
+ )
elif domain == "aggregate":
scope_hint = "只能输出聚合结果，禁止查单个客户或按 customer_id 分组/筛选"
prompt = (
@@ -175,16 +200,20 @@ class AnalystAgent:
return extract_sql(text), usage
def _generate_verified_answer(
- self, question: str, sql: str, table: TableData, empty_state: str
+ self, question: str, sql: str, table: TableData, empty_state: str, *, domain: str = "full"
) -> tuple[str, GuardrailResult | None, dict]:
summary = self._summarize(table)
empty_note = self._empty_note(empty_state)
- answer, usage = self._generate_answer(question, sql, summary, empty_note)
+ answer, usage = self._generate_answer(question, sql, summary, empty_note, domain=domain)
guard = verify(answer, table, None)
if not guard.passed:
# 重试 1 次（加强提示）
answer2, usage2 = self._generate_answer(
- question, sql, summary, empty_note + " 特别注意：所有数字必须与结果逐字一致。"
+ question,
+ sql,
+ summary,
+ empty_note + " 特别注意：所有数字必须与结果逐字一致。",
+ domain=domain,
)
for k, v in usage2.items():
cur = usage.get(k)
@@ -196,13 +225,16 @@ class AnalystAgent:
return answer2, guard2, usage
return answer, guard, usage
- def _generate_answer(self, question: str, sql: str, summary: str, note: str) -> tuple[str, dict]:
+ def _generate_answer(
+ self, question: str, sql: str, summary: str, note: str, *, domain: str = "full"
+ ) -> tuple[str, dict]:
+ system = CUSTOMER_ANSWER_SYSTEM if domain == "self" else ANSWER_SYSTEM
prompt = (
f"问题：{question}\nSQL：{sql}\n查询结果摘要：{summary}\n空态说明：{note}\n"
f"请用 1~3 句人话解读："
)
return self.llm.complete(
- [{"role": "system", "content": ANSWER_SYSTEM}, {"role": "user", "content": prompt}],
+ [{"role": "system", "content": system}, {"role": "user", "content": prompt}],
temperature=0.2,
max_tokens=500,
)
@@ -232,6 +264,7 @@ class AnalystAgent:
def _deny(self, code: str, msg: str, trace_id: str, domain: str = "") -> AnalystResponse:
suggestions = {
"AUTH_403_NOT_ASSIGNED": ["你仅能查名下客户的数据，可尝试问自己名下客户的持仓、风险分布等。"],
+ "AUTH_403_NOT_OWNER": ["你仅能查本人数据，可尝试问我的持仓、近30日交易笔数、风险等级等。"],
"AUTH_403_SCOPE": ["你仅能查聚合数据，如近30天申购金额、各产品类型规模等。"],
}.get(code)
return AnalystResponse(
diff --git a/app/service/sql_guard.py b/app/service/sql_guard.py
index 018e623..5b089e7 100644
--- a/app/service/sql_guard.py
+++ b/app/service/sql_guard.py
@@ -149,6 +149,19 @@ def validate(sql: str, domain: str, scope_customer_ids: list[str] | None = None)
raise SqlGuardError("AUTH_403_SCOPE", "涉及客户数据的查询必须包含归属过滤条件")
result.has_customer_detail = touches_customer
+ elif domain == "self":
+ literals = extract_customer_literals(sql)
+ scope = set(scope_customer_ids or [])
+ if not scope:
+ raise SqlGuardError("AUTH_403_SCOPE", "客户问数缺少本人 customer_id")
+ out = [c for c in literals if c.upper() not in {s.upper() for s in scope}]
+ if out:
+ raise SqlGuardError("AUTH_403_NOT_OWNER", f"仅能查询本人数据，无权访问：{', '.join(out)}")
+ touches_customer = any(t in CUSTOMER_TABLES for t in tables)
+ if touches_customer and "customer_id" not in low:
+ raise SqlGuardError("AUTH_403_SCOPE", "客户问数涉及客户表时必须带本人 customer_id 过滤")
+ result.has_customer_detail = touches_customer
+
elif domain == "risk" or domain == "risk_officer":
# 台账全量 + 客户只读：允许白名单内全部表
result.has_customer_detail = any(t in CUSTOMER_TABLES for t in tables)
diff --git a/app/utils/auth.py b/app/utils/auth.py
deleted file mode 100644
index 5b5cf79..0000000
--- a/app/utils/auth.py
+++ /dev/null
@@ -1,110 +0,0 @@
-"""鉴权：Mock JWT（HS256，开发用）+ AuthContext + RBAC 角色判定。
-
-生产对齐 docs/项目框架设计/技术选型和版本/02-JWT-RBAC鉴权手册.md（RS256 + IdP）；
-本模块只覆盖数据分析 Agent 开发/联调所需的最小身份与角色能力。
-"""
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from typing import Any
-
-from jose import jwt, JWTError
-
-from app.config.settings import settings
-
-ALGORITHM = "HS256"
-
-
-class AuthError(Exception):
- """鉴权失败，携带错误码（对齐 JWT 手册 §10）。"""
-
- def __init__(self, error_code: str, message: str) -> None:
- super().__init__(message)
- self.error_code = error_code
- self.message = message
-
-
-@dataclass
-class AuthContext:
- """注入给业务层的最小身份上下文。"""
-
- subject_id: str
- token_type: str
- roles: list[str]
- permissions: list[str] = field(default_factory=list)
- staff_type: str = ""
- trace_id: str = ""
- agent_type: str = "analyst"
-
- def has_role(self, role: str) -> bool:
- return role in self.roles
-
- def has_perm(self, perm: str) -> bool:
- return perm in self.permissions
-
- def to_dict(self) -> dict[str, Any]:
- return {
- "subject_id": self.subject_id,
- "token_type": self.token_type,
- "roles": self.roles,
- "permissions": self.permissions,
- "staff_type": self.staff_type,
- "trace_id": self.trace_id,
- "agent_type": self.agent_type,
- }
-
-
-def create_dev_token(
- subject_id: str,
- roles: list[str],
- staff_type: str = "",
- permissions: list[str] | None = None,
- expires_hours: int | None = None,
-) -> str:
- """开发/联调用：签发员工 Token（HS256）。"""
- ttl = expires_hours or settings.jwt_token_ttl_hours
- claims: dict[str, Any] = {
- "sub": subject_id,
- "token_type": "staff",
- "roles": roles,
- "staff_type": staff_type,
- "permissions": permissions or [],
- }
- return jwt.encode(claims, settings.jwt_dev_secret, algorithm=ALGORITHM)
-
-
-def verify_token(token: str) -> AuthContext:
- """验签并解析为 AuthContext；失败抛 AuthError(401)。"""
- try:
- claims = jwt.decode(token, settings.jwt_dev_secret, algorithms=[ALGORITHM])
- except JWTError as exc:
- raise AuthError("AUTH_401_INVALID_TOKEN", "token 无效或已过期") from exc
- return AuthContext(
- subject_id=str(claims.get("sub", "")),
- token_type=str(claims.get("token_type", "staff")),
- roles=list(claims.get("roles", [])),
- permissions=list(claims.get("permissions", [])),
- staff_type=str(claims.get("staff_type", "")),
- )
-
-
-# 数据分析 Agent 允许进入的员工角色（需求规格 §2.1 角色矩阵）
-ANALYST_ROLES = {"analyst", "advisor", "risk_officer", "ops"}
-
-# 角色 → 需求规格 §2.1 的数据域
-ROLE_DATA_DOMAIN = {
- "analyst": "full", # 全量 + 全明细 + 敏感列可见
- "advisor": "assigned", # 仅名下客户 + 明细 + 脱敏
- "risk_officer": "risk", # 台账全量 + 客户只读 + 客户脱敏
- "ops": "aggregate", # 无客户维度 + 仅聚合
-}
-
-
-def assert_analyst_access(ctx: AuthContext) -> str:
- """校验角色可进入数据分析 Agent，返回数据域。失败抛 AuthError(403)。"""
- if ctx.token_type != "staff":
- raise AuthError("AUTH_403_ROLE", "数据分析 Agent 仅限内部员工使用")
- for role in ctx.roles:
- if role in ANALYST_ROLES:
- return ROLE_DATA_DOMAIN[role]
- raise AuthError("AUTH_403_ROLE", "当前角色无权使用数据分析 Agent")
diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md
index 9687828..a428cfa 100644
--- a/docs/memory/TODO.md
+++ b/docs/memory/TODO.md
@@ -5,7 +5,7 @@
## 进行中
-**前端 P0 主链路真接（2026-09-09）**：ChatPanel · 游客/客户/风控/顾问/分析 Chat · 平台只读页 · 行情 · 风控台账处置 · **730 passed** · **19 Vitest** ✓
+**前端 P0 主链路真接（2026-09-09）**：ChatPanel · 游客/客户/风控/顾问/分析 Chat · 平台只读页 · 行情 · 风控台账处置 · **773 passed** · **19 Vitest** · **analytics 问数页已接**
**Redis Docker（2026-09-09）**：`jinrong-redis` @ **6380** · `.env REDIS_URL` 已对齐 · `scripts/dev/start-redis.ps1` ✓
@@ -14,7 +14,7 @@
- [x] **代销平台 API v0.1 实现**（2026-09-08）：`customers/products/advisors/staff/compliance` + `service/platform/` + `PLATFORM_RESPONSE_DESENSITIZE` · **530 passed 0 skipped**
- [x] **`web/` P0 脚手架 init**（2026-09-09）：Vite+AntD+HashRouter · 登录真接 · AppLayout · 四角色路由 · 占位页 · 见 `web/README.md` · 设计 `docs/superpowers/specs/2026-09-09-frontend-p0-design.md`
- [ ] **接口契约发群**（login + 平台读 API + simulate/trade；强调重复功能以平台路径为准）
-- [ ] **前端 P0 真接**（主链路已完成；余 analytics 问数 · 契约发群 · simulate 说明）
+- [ ] **前端 P0 真接**（analytics 问数已接 · 契约发群 · simulate 说明）
- [ ] 同步 `MEMORY/REQUIREMENTS/FRAMEWORK` 与各 Agent 负责人联调节奏
### 前端 `web/` · P0 待接（后端已就绪）
@@ -32,7 +32,7 @@
- [x] **risk 预警台账全页**：`GET /api/risk/alerts` 筛选/分页表格 + `POST .../handle` 处置 UI
- [x] **advisor Chat**：SSE + AgentBanner · 名下客户列表页
- [x] **`/app/analytics/chat`**：SSE · `X-Agent-Type: analyst`
-- [ ] **`/app/analytics/query`**：PermissionGate + 问数 Mock/403 展示
+- [x] **`/app/analytics/query`**：`AnalystQueryPanel` · `POST /api/analyst/chat` · 四角色权限内问数
- [ ] **Vitest**：`authStore` · SSE 解析已绿（19 例）· 可补 ChatPanel hook 测试
- [ ] **前端 P0 收尾**：接口契约发群 · simulate 交易 Disabled 菜单说明
@@ -57,6 +57,17 @@
- [x] **客服 RAG 接 fin_* 三库**：`VisitorRagService` → `search_cs_knowledge`（`fin_product` / `fin_policy` / `fin_faq`）；宿主 Tool 仍走 `kb_product_rules`（T-21）
- [x] **tool 层实现 + 种子文档**：`document_parser` / `embedding_tool` / `milvus_tool` + `data/kb_collections/*`（2026-09-09）
+### 数据分析 Agent · S3 接缝（2026-09-09 已接线）
+
+> 清单：`docs/项目框架设计/数据分析Agent-合并说明.md` · merge `data-analysis-agent-work` · **773 passed**
+
+- [x] **merge + 接缝**：`analyst_router` · `analyst_auth_adapter` · `analyst_schemas` · 废弃 `utils/auth.py`
+- [x] **鉴权 G1–G4 + G8**：`assert_analyst_query_access` · customer `self` 域 · 尾注「AI 分析有风险」
+- [x] **Wave6 测试**：`test_wave6_*` · +43 例
+- [x] **前端问数页**：`web/src/pages/analytics/AnalystQueryPage.tsx` · `api/analyst.ts`
+- [x] **迁移 SQL（本机）**：`scripts/agent/migrate-analyst-d07-d11.sql`（三资产表 · 2026-09-09 已执行 · **无种子数据**）
+- [ ] **Scope B 冒烟**：四 Demo 角色各一条问数（需 MySQL + 可选 DeepSeek Key）
+
### 风控 Agent · 后端已就绪 · 前端/运维未接（盘点 2026-09-09）
**HTTP 接口（后端已实现 · 前端/演示未接）**
diff --git a/docs/项目框架设计/数据分析Agent-合并说明.md b/docs/项目框架设计/数据分析Agent-合并说明.md
new file mode 100644
index 0000000..1203f10
--- /dev/null
+++ b/docs/项目框架设计/数据分析Agent-合并说明.md
@@ -0,0 +1,358 @@
+# 数据分析 Agent · 从 `data-analysis-agent-work` 合并说明
+
+> 日期：2026-09-09
+> 远程分支：`xinghuo/data-analysis-agent-work` @ `b19a241`（zyi）
+> 原则：**只增不盖**；与 `merger` 同路径冲突时 **以 merger 为准**（不覆盖 `main.py` / `deps.py` / `chat.py` / `agent_service.py` 宿主骨架）。
+> 状态：**接缝设计稿（未 merge、未接线）**
+
+---
+
+## 0. 一句话定位
+
+数据分析 Agent = **NL2SQL 只读查数线**：自然语言 → 安全 SQL → 表格 + 解读 + 留痕；与宿主 **通用对话线**（`agent_service` 轻提示词 Chat）是两条不同产物，接缝上 **并行挂载**，不替换 `agent_service.py`。
+
+---
+
+## 1. 合并什么（新增文件清单）
+
+从 `data-analysis-agent-work` **整包迁入**（无冲突或仅新增路径）：
+
+| 类别 | 路径 | 作用 |
+| --- | --- | --- |
+| 路由 | `app/api/analyst.py` | `/api/analyst/chat` · `/dashboard` · `/assets` · `/ops/metrics` |
+| 编排 | `app/service/analyst_agent.py` | LangGraph StateGraph（input_guard → … → audit_persist） |
+| 安全 | `app/service/sql_guard.py` `guardrail.py` | 五层 SQL 白名单 + 数字护栏 |
+| 支撑 | `analytics_repo.py` `dict_service.py` `cache_service.py` `schema_meta.py` | 留痕/口径/缓存/元数据 |
+| LLM | `app/service/llm.py` | DeepSeek 封装（分析线专用；与 merger `embedding.py` 不冲突） |
+| 模型 | `app/model/schemas/analyst.py` | 输出四件套 Pydantic（answer/table/sql/meta） |
+| 实体 | `app/model/entities/analytics.py`（若远程有） | analytics_* ORM 映射 |
+| 工具 | `app/tool/sql_tool.py`（若远程有） | 只读执行 + 行数/超时 |
+| 文档 | `数据分析Agent架构说明书.md` `数据分析Agent开发清单.md` `03-mysql-analyst专用.sql` | 架构 / 任务 / DDL |
+| 测试 | 见 §6 | 重命名后迁入，避免覆盖 merger 730 基线 |
+
+**不迁入（冲突 · 保留 merger）：**
+
+| 文件 | 原因 |
+| --- | --- |
+| `app/main.py` | AL-09 全量宿主入口；仅 **追加** `include_router(analyst_router)` |
+| `app/api/deps.py` | merger 完整 JWT + 归属 + 矩阵（382 行）；远程版被砍成 22 行 stub |
+| `app/api/chat.py` `app/service/agent_service.py` | 四 Agent 通用对话线已集成 |
+| `app/config/settings.py` | merger 含 risk/customer/visitor/JWT 全量；**追加** `analyst_*` 字段 |
+| `app/model/schemas.py` | 保留单文件；分析专用模型放 **`schemas/analyst.py` 子模块**（见 §3.3） |
+| `app/repository/core_ro.py` | merger 版含 `check_suitability` 等平台能力；分析线只 **调用** 不覆盖 |
+| `app/utils/auth.py` | 远程第二套鉴权（`subject_id`）；**废弃**，改 deps 适配（§4） |
+| `docs/memory/*`、前端 `web/`、风控/平台 API | 全部保留 merger |
+
+---
+
+## 2. 数据库（需手工执行）
+
+### 2.1 已有（merger canonical）
+
+`docs/项目框架设计/表设计/02-mysql-agent专用.sql` 已含 **`analytics_query_log`**（NL2SQL 留痕）。merger `entities.py` 已有 ORM。**无需重复建表**（若本机已跑过 02 迁移）。
+
+### 2.2 新增（分析资产三表 · D-07 / D-11）
+
+远程 `03-mysql-analyst专用.sql` → 独立迁移脚本（不修改 canonical 02 文件）：
+
+```text
+scripts/agent/migrate-analyst-d07-d11.sql
+ → analytics_metric_dict 口径字典
+ → analytics_few_shot 问题→SQL 示例
+ → analytics_query_template 参数化模板
+```
+
+执行时机：merge 代码后、跑分析线单测/E2E 前；库 `jinrong_agent`，前置 01/02 共用底座已存在。
+
+### 2.3 Redis
+
+复用 merger Docker Redis（`:6380`）。分析线 `cache_service` 使用独立 key 前缀（对齐 `02-redis-keys.md` 分析段）；与客服/游客/风控 key **命名空间隔离**，同实例不冲突。
+
+---
+
+## 3. S3 接缝设计（推荐方案 · 待你确认 §8）
+
+### 3.1 入口策略：**并行双轨**（推荐）
+
+| 入口 | 行为 | 前端页面 |
+| --- | --- | --- |
+| **`POST /api/analyst/chat`** | `AnalystAgent.run()` → JSON 四件套（table/sql/meta/disclaimer） | **`/app/analytics/query` 问数工作台**（主链路） |
+| `GET /api/analyst/dashboard` | D-12 智能看数卡片 | `/app/analyst/home` 可接 metrics |
+| `POST /api/analyst/assets` | D-11 资产沉淀 | 问数页「沉淀」按钮（P1 后做） |
+| `GET /api/analyst/ops/metrics` | N-08 运营指标 | 分析员运维面板（P1 后做） |
+| `POST /api/chat` + `X-Agent-Type: analyst` | **仍走 `agent_service`**（轻量对话 stub，无 NL2SQL） | **`/app/analytics/chat` 分析对话**（可选保留） |
+
+**为何不把 analyst 并入 `chat.py` 分流？**
+
+- 问数响应含 **结构化 table + sql + meta**，与 Chat 的 `{reply, has_disclaimer}` 形状不同；硬塞进 SSE 需额外 adapter 层，收益低。
+- 架构说明书 §13 已定义独立 `/api/analyst/*`；与客服 S2「customer 分流」场景不同（customer 仍是 Chat 形态）。
+
+**`main.py` 改动（唯一宿主挂载点）：**
+
+```python
+from app.api.analyst import router as analyst_router
+app.include_router(analyst_router) # prefix 已在 analyst.py 内：/api/analyst
+```
+
+### 3.2 鉴权接缝：**deps 为准 + 薄适配**
+
+远程代码混用 `deps.get_auth_context` 与 `utils.auth.AuthContext`（字段 `subject_id` vs `actor_id`）。合并后：
+
+1. **删除**迁入后的 `app/utils/auth.py`（或保留空 shim 仅 re-export，最终删除）。
+2. 新增 `app/api/analyst_auth_adapter.py`（或扩展现有 `auth_adapter.py`）：
+
+```python
+def analyst_auth_from_deps(ctx: deps.AuthContext) -> AnalystAuthView:
+ """deps.AuthContext → 分析线内部视图（仅字段映射，不做权限判定）。"""
+ return AnalystAuthView(
+ subject_id=ctx.actor_id, # 分析 repo 写 staff_id 用
+ roles=ctx.roles,
+ permissions=ctx.permissions,
+ token_type=ctx.token_type,
+ trace_id=current_trace() or "",
+ agent_type="analyst",
+ )
+```
+
+3. `analyst.py` 路由层：`auth = Depends(get_auth_context)` → 适配后传入 `AnalystAgent.run()`。
+4. 远程 `assert_analyst_access(auth)` 逻辑 **迁入** `deps.py` 或 `app/utils/authz.py`（与 `assert_customer_access` 并列），使用 `actor_id` 命名。
+5. **`X-Agent-Type`**：`/api/analyst/*` **不要求**该头（路由即 analyst）；JWT 矩阵仍校验 token 角色含 `analyst`/`compliance`/`advisor`/`risk_officer`/`ops`（按远程 `assert_analyst_access` 域划分）。
+
+### 3.3 Schemas：**单文件 + 子模块共存**
+
+merger 当前 `app/model/schemas.py` 为单文件。合并策略：
+
+- **新增** `app/model/schemas/analyst.py`（仅分析请求/响应）。
+- **不**把 `schemas.py` 改成包（避免大面积 import 破坏）；`analyst.py` 路由 `from app.model.schemas.analyst import ...` 直引子路径。
+- 若 Python 包解析与 `schemas.py` 文件冲突：将子模块改为 `app/model/analyst_schemas.py`（备选，merge 时二选一）。
+
+### 3.4 settings 追加字段（合并冲突手工解）
+
+在 merger `settings.py` **追加**（取自远程，命名对齐架构说明书）：
+
+| 字段 | 用途 |
+| --- | --- |
+| `analyst_llm_model` / `analyst_llm_timeout` | DeepSeek NL2SQL |
+| `analyst_sql_row_limit` / `analyst_sql_timeout_sec` | 只读执行上限 |
+| `analyst_cache_ttl_sec` | 结果缓存 |
+| `analyst_rate_limit_per_min` | 问数限流 |
+
+`.env.example` 同步追加；**不**覆盖 merger 已有 `jwt_*` / `risk_*` / `customer_*`。
+
+### 3.5 requirements
+
+merger 已有 `langgraph`；**追加** `sqlglot>=25.0.0`（远程新增，SQL AST 白名单硬依赖）。
+
+### 3.6 core_ro / llm 边界
+
+| 模块 | 策略 |
+| --- | --- |
+| `core_ro.py` | 保留 merger；`analytics_repo.execute_readonly` 走 **agent 库只读账号** 或现有 Core RO 连接，不 duplicate |
+| `llm.py` | 迁入分析线专用；`AnalystAgent` 只 import 此模块 |
+| `input_guard` | 分析线复用 merger `app/service/input_guard.py`（限流/注入），不保留远程 duplicate `utils/input_guard.py` |
+
+---
+
+## 4. 前端接缝（merger 现状 → 目标）
+
+### 4.1 现状
+
+| 路由 | 现状 |
+| --- | --- |
+| `/app/analytics/query` | **PlaceholderPage**（待 NL2SQL） |
+| `/app/analytics/chat` | `ChatPanel` + `agentType="analyst"` → `POST /api/chat/stream`（无 table/sql） |
+
+### 4.2 推荐（与 §3.1 双轨一致）
+
+| 页面 | API | UI |
+| --- | --- | --- |
+| **问数工作台** `/app/analytics/query` | `POST /api/analyst/chat` | 新问题组件：`AnalystQueryPanel`（问题框 + 解读 + Ant Design Table + SQL 折叠 + disclaimer） |
+| **分析对话** `/app/analytics/chat` | 暂保留 `ChatPanel` SSE | 页顶 Banner 提示：「复杂查数请用问数工作台」；P2 可下线或改跳转 |
+
+新增 `web/src/api/analyst.ts`：
+
+```typescript
+export interface AnalystChatResponse {
+ answer: string
+ table: { columns: string[]; rows: unknown[][] }
+ sql: string
+ meta: { exec_ms: number; row_count: number; cache_hit: boolean; data_as_of?: string }
+ disclaimer: string
+ status: string
+ trace_id?: string
+}
+```
+
+**不**强行让 `ChatPanel` 解析四件套（职责分离）。
+
+### 4.3 首页看板
+
+`AnalystMarketDashboard` QuickAction「问数工作台」已指向 `/app/analytics/query`；merge 后可接 `GET /api/analyst/dashboard` 填充 metrics 卡片（P1）。
+
+---
+
+## 5. 需求 ID 对齐
+
+| 需求 | 合并后落点 | 阶段 |
+| --- | --- | --- |
+| D-01~D-04 | `analyst_agent` + `sql_guard` + `analytics_query_log` | merge 后冒烟 |
+| D-05 | 多表只读聚合（sql_generate 节点） | 同上 |
+| D-06 | `cache_service` 双层缓存 | 同上 |
+| D-07 | `dict_service` + `analytics_metric_dict` | 需跑 migrate-analyst |
+| D-08~D-10 | 拒答分支 + `guardrail` | 同上 |
+| D-11 | `/assets` + 三资产表 | migrate + 前端 P1 |
+| D-12 | `/dashboard` | 首页 P1 |
+| N-01~N-08 | 见架构说明书 §4/§6 | 单测覆盖 |
+
+---
+
+## 6. 测试策略
+
+### 6.1 远程测试文件 → 迁入命名
+
+| 远程 | 迁入后 | 说明 |
+| --- | --- | --- |
+| `tests/test_agent.py` | `tests/test_wave6_analyst_agent.py` | 改 import：`utils.auth` → `analyst_auth_adapter` |
+| `tests/test_sql_guard.py` | `tests/test_wave6_sql_guard.py` | 无冲突，直迁 |
+| `tests/test_guardrail.py` | `tests/test_wave6_guardrail.py` | 若 merger 已有同名则合并 case |
+| `tests/test_dict_service.py` | `tests/test_wave6_dict_service.py` | |
+| `tests/test_cache_service.py` | `tests/test_wave6_cache_service.py` | merger 已有 `test_cache_service.py` → **合并或 rename 远程** |
+| `tests/test_llm.py` | `tests/test_wave6_analyst_llm.py` | 避免与 merger LLM 测试混淆 |
+
+### 6.2 基线要求
+
+- merge + 适配完成后：`python -m pytest` → **730 + N passed**（N = 新增 wave6 条数）。
+- 禁止恢复远程对 `deps.py` / `conftest.py` 的删减。
+
+### 6.3 冒烟清单（Scope B · merge 后）
+
+1. `POST /api/analyst/chat` + analyst JWT → 200 + 四件套 JSON
+2. 顾问 token 查他人客户 → 403 + `analytics_query_log(blocked)`
+3. `GET /api/analyst/dashboard` → 200 + cards
+4. 前端 `/app/analytics/query` 真跑一条问数
+5. `POST /api/chat/stream` + analyst → 仍 200（stub 未回归）
+
+---
+
+## 7. Merge 操作顺序（执行 SOP · 尚未做）
+
+```text
+1. git worktree add ../JinRong-analyst-merge data-analysis-agent-work # 可选隔离
+2. git merge data-analysis-agent-work --no-commit # 或 cherry-pick 新增文件
+3. 冲突文件按 §1「不迁入」表逐项保留 merger 版
+4. 手工：analyst_auth_adapter · main.py include_router · settings 追加 · requirements sqlglot
+5. 删除 utils/auth.py · 改 analyst.py / analyst_agent.py import
+6. 执行 scripts/agent/migrate-analyst-d07-d11.sql
+7. pytest 全绿 → 前端 AnalystQueryPanel → Scope B 冒烟
+8. 更新 docs/memory/TODO.md · FRAMEWORK 实现状态表
+```
+
+---
+
+## 8. 已确认决策（2026-09-09）
+
+| # | 决策 | 结论 |
+| --- | --- | --- |
+| **Q1** | 问数主入口 | **A · 独立 `POST /api/analyst/chat`**；各角色在权限范围内自助查数（见 §9） |
+| **Q2** | `/app/analytics/chat` | **A · 保留** ChatPanel stub + Banner 引导问数工作台 |
+| **Q3** | schemas 路径 | **A · `app/model/schemas/analyst.py`**（merge 时若与 `schemas.py` 包冲突再改 `analyst_schemas.py`） |
+| **Q4** | merge 方式 | **A · `git merge` + 手工解冲突**（默认，未单独拍板则按此执行） |
+
+---
+
+## 9. 鉴权接缝 · 角色矩阵与缺口（**merge 前必须补**）
+
+### 9.1 目标口径（产品确认）
+
+> **每个 Demo 账户角色都能调用数据分析模块，只看其权限范围内的数据。**
+
+前端已把「数据分析」放进 **客户 / 理财师 / 风控** 的 `sharedPlatform` 菜单（`/app/analytics/query`）；后端远程实现 **尚未覆盖 customer**，且与 merger `deps` 矩阵不一致，需在 S3 接缝一并补齐。
+
+### 9.2 目标角色 × 数据域（接缝后应达到）
+
+> Demo 仅四角色：**customer · advisor · analyst · risk_officer**（无 compliance / risk_manager 独立账号；问数线不单独开域）。
+
+| 角色 | token | 数据域 key | 可见范围 | sql_guard 要点 |
+| --- | --- | --- | --- | --- |
+| **customer** | customer | **`self`（待实现）** | 仅本人可读数据（持仓/流水/风评/净值趋势等） | 禁止其他 `CUST-*`；强制 `customer_id = auth.customer_id`；**见 §9.6 产品约束** |
+| **advisor** | staff | `assigned` | 名下客户（`core_customer_advisor` active） | 远程已有；`inject_ownership` + 白名单 |
+| **analyst** | staff | `full` | 全量 + 敏感列策略 | 远程已有 |
+| **risk_officer** | staff | `risk` | 预警台账全量 + 客户/持仓/交易只读（脱敏列） | 远程已有；**表域评估见 §9.7** |
+| **ops** | staff | `aggregate` | 无客户维度，仅聚合 | 远程已有（非 Demo 角色，预留） |
+
+### 9.6 客户问数 · 产品约束（2026-09-09 确认）
+
+客户走 **`self` 域**，能力边界如下（merge 时在 `analyst_agent` 解读阶段 + prompt 硬约束）：
+
+| 项 | 口径 |
+| --- | --- |
+| **允许** | 基于本人可读数据的**趋势总结、分布描述、数量统计**（如持仓结构、近 N 日交易笔数、盈亏区间描述） |
+| **禁止** | **投资建议、收益承诺、买卖时点、产品推荐**；命中则拒答或降级为「仅展示表格」 |
+| **尾部声明** | 在标准 `disclaimer` 之外，客户域回复**追加**：「**AI 分析有风险，仅供参考。**」 |
+| **留痕** | `analytics_query_log.actor_id = customer_id`，`actor_role = customer` |
+
+实现落点（merge 时）：
+
+- `scope_resolve`：`token_type=customer` → `domain=self`，`scope=[auth.customer_id]`
+- `sql_guard.validate(domain=self)`：同 advisor 归属逻辑，但 scope 固定单人
+- `_generate_sql` / `ANSWER_SYSTEM`：注入「仅描述数据、不给建议」
+- `answer_compose`：`domain=self` 时 append 客户专用尾注
+
+### 9.7 风控专员 · 表域是否够用（2026-09-09 评估）
+
+**结论：够用。** 风控 Agent **自动监督触发的结果**（规则引擎出单、AML 命中、适当性拦截、集中度/代理人行为链等）**权威落点都在现有白名单内**；风控专员问数线是**只读查台账与关联上下文**，不通过 NL2SQL 触发处置或扫名单。
+
+| 监督链路（风控 Agent / 引擎） | 落库/可读表 | 已在 sql_guard 白名单 |
+| --- | --- | --- |
+| RISK-001~006 交易/持仓规则 | `risk_alert`（`alert_type` + `payload`/`triggered_rules`） | ✅ |
+| RISK-006 集中度 | `risk_alert`（`pattern` + `payload.alert_subtype=concentration`）+ `core_holding` | ✅ |
+| RISK-008 代理人行为链 | `risk_alert`（`pattern` + `payload.alert_subtype=agent_behavior`） | ✅ |
+| AML 命中 | `risk_alert`（`alert_type=aml`）+ `customer_profile_l3` 标记 | ✅ |
+| 适当性拦截 | `risk_alert`（`alert_type=suitability`） | ✅ |
+| 客户监测画像 | `customer_profile_l1/l2/l3` | ✅ |
+| 关联上下文（客户/持仓/交易/产品） | `core_*` 系列 | ✅ |
+
+**刻意不进白名单（安全/职责分离）：**
+
+| 表 | 原因 |
+| --- | --- |
+| `risk_aml_list` | 反洗钱**名单本体**；命中结果已在 `risk_alert`，不应 NL2SQL 直查名单库 |
+| `audit_log` | 审计明细走 HTTP/专用 Tool；问数以 `risk_alert` 聚合即可 |
+
+**可选 P2（非阻塞 merge）：** 若后续要问「适当性判定历史通过率/明细」，可把 `risk_suitability_log` 加入白名单并在 `risk` 域只读；当前 Demo 问数（待处理预警数、按类型分布、客户监测分布）**不依赖**该表。
+
+**域规则：** 维持远程 `domain=risk`（台账全量 + 白名单内客户只读），**不新增** compliance/manager 域；与现网风控 Demo（`STAFF-30001` · risk_officer）一致即可。
+
+### 9.3 远程 vs merger 缺口清单
+
+| 缺口 | 远程现状 | merger 现状 | 接缝动作（记入 TODO） |
+| --- | --- | --- | --- |
+| **G1 customer 不可用** | `assert_analyst_access` 要求 `token_type=staff` | 平台 G-01 支持 customer 本人 | 新增域 `self`；`/api/analyst/*` 入口允许 customer token |
+| **G2 sql_guard 无 self 域** | 仅 full/assigned/risk/aggregate | — | `validate()` + `inject_ownership()` 支持 `self` + 强制 `customer_id = auth.customer_id` |
+| **G3 双套 AuthContext** | `subject_id` + `utils/auth.py` | `actor_id` + `deps.py` | §3.2 适配器；**废弃** remote auth |
+| **G4 入口矩阵分裂** | `/api/analyst/*` 用 `ANALYST_ROLES` 四角色 | `AGENT_ACCESS_MATRIX["analyst"]` 仅 analyst+compliance（**对话线**） | 问数路由 **不走** chat 矩阵；`assert_analyst_query_access()` 覆盖 **customer/advisor/analyst/risk_officer** |
+| ~~**G5**~~ | — | — | **已关闭**：无 compliance/risk_manager Demo 角色，不单独开域 |
+| **G6 留痕 actor** | `staff_id=subject_id` | customer 无 staff_id | `analytics_query_log` 写 `actor_id` + `actor_role`；customer 写 `customer_id` |
+| **G7 前端 PermissionGate** | — | 问数页仍 Placeholder | 403 展示可查范围引导（对齐 `AnalystResponse.suggestions`） |
+| **G8 客户解读约束** | 远程无 customer 域 | — | §9.6：趋势总结 · 禁建议 · 尾部「AI 分析有风险」 |
+
+### 9.4 推荐实现落点（merge 时）
+
+```text
+app/utils/authz.py # 新增 assert_analyst_query_access(auth) → domain
+app/service/sql_guard.py # 扩展 domain=self（§9.6 客户约束）
+app/api/analyst.py # Depends(get_auth_context) → assert_analyst_query_access → adapter → agent.run
+app/api/analyst_auth_adapter.py # actor_id ↔ 分析线内部视图（可选 NamedTuple）
+```
+
+**`/api/chat` analyst stub** 仍走现有 `AGENT_ACCESS_MATRIX`（analyst/compliance 轻聊，Demo 无 compliance 账号）；与问数线 **刻意分离**。
+
+---
+
+## 10. 修订记录
+
+| 日期 | 说明 |
+| --- | --- |
+| 2026-09-09 | 初稿：远程 `b19a241` 审阅 + merger 730 基线对照；**未 merge** |
+| 2026-09-09 | Q1=A Q2=A；§9 鉴权缺口与角色矩阵（含 customer self 域待实现） |
+| 2026-09-09 | 产品拍板：§9.6 客户趋势总结/禁建议/AI 风险尾注；§9.7 risk 表域够用；关闭 G5 |
diff --git a/scripts/agent/migrate-analyst-d07-d11.sql b/scripts/agent/migrate-analyst-d07-d11.sql
new file mode 100644
index 0000000..89c14bd
--- /dev/null
+++ b/scripts/agent/migrate-analyst-d07-d11.sql
@@ -0,0 +1,61 @@
+-- =============================================================================
+-- 数据分析 Agent 专用 MySQL（3 张 · 养 Agent 资产）
+-- 执行：merge 数据分析 Agent 后 · jinrong_agent 库
+-- 来源：docs/项目框架设计/表设计/03-mysql-analyst专用.sql
+-- =============================================================================
+
+USE jinrong_agent;
+
+CREATE TABLE IF NOT EXISTS analytics_metric_dict (
+ id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ metric_key VARCHAR(128) NOT NULL COMMENT '唯一键，如 holding_scale',
+ metric_name VARCHAR(128) NOT NULL COMMENT '持仓规模',
+ aliases JSON NULL COMMENT '同义词：["规模","市值"]',
+ definition TEXT NOT NULL COMMENT '口径定义',
+ formula TEXT NULL COMMENT '计算公式',
+ applicable_tables JSON NULL COMMENT '适用表清单',
+ default_time_window VARCHAR(64) NULL COMMENT '默认时间窗',
+ unit VARCHAR(32) NULL COMMENT '单位：元/万元/%',
+ status ENUM('draft','published','rolled_back') NOT NULL DEFAULT 'draft',
+ version INT UNSIGNED NOT NULL DEFAULT 1,
+ gray_roles JSON NULL COMMENT '灰度：null=全量，["advisor"]=部分角色',
+ created_by VARCHAR(64) NOT NULL,
+ published_by VARCHAR(64) NULL,
+ created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+ updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+ UNIQUE KEY uk_metric (metric_key, version),
+ KEY idx_status (status, metric_key)
+) ENGINE=InnoDB COMMENT='【分析专用】口径字典（D-07/D-11）';
+
+CREATE TABLE IF NOT EXISTS analytics_few_shot (
+ id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ question TEXT NOT NULL,
+ sql_text TEXT NOT NULL,
+ tags JSON NULL,
+ source_session_id VARCHAR(64) NULL COMMENT '溯源会话',
+ status ENUM('draft','published','rolled_back') NOT NULL DEFAULT 'draft',
+ version INT UNSIGNED NOT NULL DEFAULT 1,
+ gray_roles JSON NULL,
+ created_by VARCHAR(64) NOT NULL,
+ published_by VARCHAR(64) NULL,
+ created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+ updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+ KEY idx_status (status, id)
+) ENGINE=InnoDB COMMENT='【分析专用】few-shot 示例（D-11）';
+
+CREATE TABLE IF NOT EXISTS analytics_query_template (
+ id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ template_key VARCHAR(128) NOT NULL,
+ template_sql TEXT NOT NULL COMMENT '参数化 SQL，如 WHERE risk_code IN (:risk_codes)',
+ params_schema JSON NULL COMMENT '参数定义',
+ tags JSON NULL,
+ source_session_id VARCHAR(64) NULL,
+ status ENUM('draft','published','rolled_back') NOT NULL DEFAULT 'draft',
+ version INT UNSIGNED NOT NULL DEFAULT 1,
+ gray_roles JSON NULL,
+ created_by VARCHAR(64) NOT NULL,
+ published_by VARCHAR(64) NULL,
+ created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+ updated_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
+ KEY idx_status (status, template_key)
+) ENGINE=InnoDB COMMENT='【分析专用】快速模板（D-06/D-11）';
diff --git a/scripts/agent/reset-analyst-d07-d11.sql b/scripts/agent/reset-analyst-d07-d11.sql
new file mode 100644
index 0000000..3adfe50
--- /dev/null
+++ b/scripts/agent/reset-analyst-d07-d11.sql
@@ -0,0 +1,14 @@
+-- =============================================================================
+-- 数据分析 Agent 资产表 · 清数据（不删表结构）
+-- 用途：开发/演示重置口径字典、few-shot、模板；不影响其它 agent 表
+-- 问数留痕 analytics_query_log 在 02 底座里，可选一并清空（见下方注释）
+-- =============================================================================
+
+USE jinrong_agent;
+
+TRUNCATE TABLE analytics_metric_dict;
+TRUNCATE TABLE analytics_few_shot;
+TRUNCATE TABLE analytics_query_template;
+
+-- 可选：清空 NL2SQL 问数留痕（取消注释即可）
+-- TRUNCATE TABLE analytics_query_log;
diff --git a/tests/__init__.py b/tests/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/tests/test_api.py b/tests/test_api.py
deleted file mode 100644
index 5fac71b..0000000
--- a/tests/test_api.py
+++ /dev/null
@@ -1,81 +0,0 @@
-"""API 层测试（FastAPI TestClient）。"""
-import unittest
-
-from fastapi.testclient import TestClient
-
-from app.main import app
-from app.utils.auth import create_dev_token
-
-client = TestClient(app)
-
-
-def analyst_token():
- return create_dev_token("STAFF-API", ["analyst"], "analyst")
-
-
-def advisor_token():
- return create_dev_token("STAFF-ADV", ["advisor"], "advisor")
-
-
-class TestApi(unittest.TestCase):
- def test_health(self):
- r = client.get("/health")
- self.assertEqual(r.status_code, 200)
-
- def test_chat_no_token(self):
- r = client.post("/api/analyst/chat", json={"question": "客户总数"})
- self.assertEqual(r.status_code, 401)
-
- def test_chat_bad_token(self):
- r = client.post(
- "/api/analyst/chat",
- json={"question": "客户总数"},
- headers={"Authorization": "Bearer bad"},
- )
- self.assertEqual(r.status_code, 401)
-
- def test_chat_success(self):
- r = client.post(
- "/api/analyst/chat",
- json={"question": "客户总数是多少"},
- headers={"Authorization": f"Bearer {analyst_token()}"},
- )
- self.assertEqual(r.status_code, 200)
- data = r.json()
- self.assertIn(data["status"], ("success", "degrade"))
- self.assertIn("answer", data)
- self.assertIn("table", data)
-
- def test_dashboard(self):
- r = client.get(
- "/api/analyst/dashboard",
- headers={"Authorization": f"Bearer {analyst_token()}"},
- )
- self.assertEqual(r.status_code, 200)
- self.assertIn("cards", r.json())
-
- def test_assets_advisor_forbidden(self):
- r = client.post(
- "/api/analyst/assets",
- json={"kind": "dict", "payload": {"metric_key": "x"}},
- headers={"Authorization": f"Bearer {advisor_token()}"},
- )
- self.assertEqual(r.status_code, 403)
-
- def test_assets_analyst_ok(self):
- import uuid
- key = f"test_k_{uuid.uuid4().hex[:8]}"
- r = client.post(
- "/api/analyst/assets",
- json={
- "kind": "dict",
- "payload": {"metric_key": key, "metric_name": "测试指标", "definition": "测试口径"},
- },
- headers={"Authorization": f"Bearer {analyst_token()}"},
- )
- self.assertEqual(r.status_code, 200)
- self.assertTrue(r.json()["ok"])
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_integration.py b/tests/test_integration.py
deleted file mode 100644
index 6b68d53..0000000
--- a/tests/test_integration.py
+++ /dev/null
@@ -1,73 +0,0 @@
-"""集成测试：真实 MySQL + 真实 DeepSeek，覆盖 §7 验收场景核心路径。"""
-import unittest
-
-from app.service.analyst_agent import AnalystAgent
-from app.service.analytics_repo import AnalyticsRepo
-from app.service.llm import DeepSeekLLM
-from app.utils.auth import AuthContext
-
-
-class TestIntegration(unittest.TestCase):
- @classmethod
- def setUpClass(cls):
- cls.repo = AnalyticsRepo()
- cls.agent = AnalystAgent(llm=DeepSeekLLM(), repo=cls.repo)
- r = cls.repo.execute_readonly(
- "SELECT staff_id FROM core_staff WHERE staff_type='advisor' AND is_active=1 LIMIT 1"
- )
- cls.advisor_id = r["rows"][0][0]
- cls.scope = cls.repo.resolve_advisor_scope(cls.advisor_id)
- all_cust = [
- row[0]
- for row in cls.repo.execute_readonly(
- "SELECT customer_id FROM core_customer WHERE is_active=1 LIMIT 100"
- )["rows"]
- ]
- cls.out_customer = next(c for c in all_cust if c not in cls.scope)
-
- def _ctx(self, roles, subject):
- return AuthContext(subject_id=subject, token_type="staff", roles=roles, staff_type=roles[0])
-
- def test_analyst_holding_by_product(self):
- resp = self.agent.run("按产品类型统计总持仓规模", self._ctx(["analyst"], "STAFF-20001"))
- self.assertIn(resp.status, ("success", "degrade"))
- self.assertTrue(resp.sql)
-
- def test_analyst_holding_pnl_sort(self):
- resp = self.agent.run("把客户持仓按盈亏排序", self._ctx(["analyst"], "STAFF-20001"))
- self.assertIn(resp.status, ("success", "degrade"))
- self.assertTrue(resp.sql)
-
- def test_advisor_own_scope(self):
- resp = self.agent.run("我名下客户有多少高风险", self._ctx(["advisor"], self.advisor_id))
- self.assertIn(resp.status, ("success", "degrade"))
-
- def test_advisor_out_of_scope_denied(self):
- resp = self.agent.run(f"查 {self.out_customer} 的持仓", self._ctx(["advisor"], self.advisor_id))
- self.assertEqual(resp.status, "deny")
- self.assertEqual(resp.error_code, "AUTH_403_NOT_ASSIGNED")
-
- def test_risk_officer_pending_alerts(self):
- resp = self.agent.run("当前待处理预警有多少", self._ctx(["risk_officer"], "STAFF-30001"))
- self.assertIn(resp.status, ("success", "degrade"))
-
- def test_ops_aggregate(self):
- resp = self.agent.run("近30天申购金额总额", self._ctx(["ops"], "STAFF-50001"))
- self.assertIn(resp.status, ("success", "degrade"))
-
- def test_cache_hit(self):
- q = "客户总数是多少"
- r1 = self.agent.run(q, self._ctx(["analyst"], "STAFF-20001"))
- r2 = self.agent.run(q, self._ctx(["analyst"], "STAFF-20001"))
- self.assertTrue(r2.meta.cache_hit)
-
- def test_trace_audit_recorded(self):
- resp = self.agent.run("客户总数是多少", self._ctx(["analyst"], "STAFF-20001"))
- rows = self.repo.execute_readonly(
- f"SELECT trace_id FROM jinrong_agent.analytics_query_log WHERE trace_id='{resp.trace_id}'"
- )["rows"]
- self.assertGreaterEqual(len(rows), 1)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_main.py b/tests/test_main.py
index 30c077e..cb3d616 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -85,6 +85,10 @@ def test_all_routers_mounted(client):
# 方案 C：SSE 流式对话
"/api/chat/stream",
"/api/chat/visitor",
+ "/api/analyst/chat",
+ "/api/analyst/dashboard",
+ "/api/analyst/assets",
+ "/api/analyst/ops/metrics",
}
diff --git a/tests/test_module_boundary.py b/tests/test_module_boundary.py
index 1413126..e747e76 100644
--- a/tests/test_module_boundary.py
+++ b/tests/test_module_boundary.py
@@ -95,6 +95,25 @@ HOST_ONLY_SKIP_PREFIXES = (
"app/utils/exceptions.py",
)
+# 数据分析 Agent S3 接缝 — 走 deps + analyst_auth_adapter
+ANALYST_AGENT_SEAM_SKIP = (
+ "app/api/analyst.py",
+ "app/api/analyst_auth_adapter.py",
+ "app/service/analyst_agent.py",
+ "app/service/analytics_repo.py",
+ "app/service/cache_service.py",
+ "app/service/dict_service.py",
+ "app/service/guardrail.py",
+ "app/service/llm.py",
+ "app/service/schema_meta.py",
+ "app/service/sql_guard.py",
+)
+
+# B7 Redis 网关惰性连接需 database._redis_kwargs（RESP2 口径）
+INFRA_SKIP = (
+ "app/service/risk/redis_gateway.py",
+)
+
# 客服 Agent S2 接缝（docs/项目框架设计/客服Agent-合并说明.md）— 允许 import 宿主 schemas
CUSTOMER_AGENT_SEAM_SKIP = (
"app/api/auth_adapter.py",
@@ -116,7 +135,7 @@ CUSTOMER_AGENT_SEAM_SKIP = (
def _is_host_only_file(path: Path) -> bool:
rel = path.relative_to(APP_DIR.parent).as_posix()
- if rel in CUSTOMER_AGENT_SEAM_SKIP:
+ if rel in CUSTOMER_AGENT_SEAM_SKIP or rel in ANALYST_AGENT_SEAM_SKIP or rel in INFRA_SKIP:
return True
return rel.startswith(HOST_ONLY_SKIP_PREFIXES)
diff --git a/tests/test_sql_guard.py b/tests/test_sql_guard.py
deleted file mode 100644
index bec4697..0000000
--- a/tests/test_sql_guard.py
+++ /dev/null
@@ -1,83 +0,0 @@
-"""sql_guard 单元测试（unittest，无需外部服务）。"""
-import unittest
-
-from app.service.sql_guard import (
- SqlGuardError,
- extract_tables,
- inject_ownership,
- validate,
-)
-
-
-class TestSqlGuard(unittest.TestCase):
- def test_select_allowed(self):
- r = validate("SELECT risk_code, COUNT(*) AS c FROM core_customer GROUP BY risk_code", "full")
- self.assertTrue(r.allowed)
- self.assertIn("core_customer", r.tables)
-
- def test_with_select_allowed(self):
- r = validate("WITH t AS (SELECT * FROM core_holding) SELECT * FROM t", "full")
- self.assertTrue(r.allowed)
-
- def test_insert_rejected(self):
- with self.assertRaises(SqlGuardError) as cm:
- validate("INSERT INTO core_customer VALUES (1)", "full")
- self.assertEqual(cm.exception.error_code, "SQL_NOT_SELECT")
-
- def test_multi_statement_rejected(self):
- with self.assertRaises(SqlGuardError) as cm:
- validate("SELECT 1; DROP TABLE core_customer;", "full")
- self.assertEqual(cm.exception.error_code, "SQL_MULTI_STATEMENT")
-
- def test_drop_rejected(self):
- with self.assertRaises(SqlGuardError):
- validate("SELECT * FROM core_customer; DROP TABLE core_customer", "full")
-
- def test_unknown_table_rejected(self):
- with self.assertRaises(SqlGuardError) as cm:
- validate("SELECT * FROM mysql.user", "full")
- self.assertEqual(cm.exception.error_code, "SQL_TABLE_NOT_ALLOWED")
-
- def test_advisor_out_of_scope_rejected(self):
- with self.assertRaises(SqlGuardError) as cm:
- validate(
- "SELECT * FROM core_holding WHERE customer_id = 'CUST-1004'",
- "assigned", ["CUST-1001"],
- )
- self.assertEqual(cm.exception.error_code, "AUTH_403_NOT_ASSIGNED")
-
- def test_advisor_in_scope_allowed(self):
- r = validate(
- "SELECT * FROM core_holding WHERE customer_id = 'CUST-1001'",
- "assigned", ["CUST-1001", "CUST-1002"],
- )
- self.assertTrue(r.allowed)
- self.assertTrue(r.has_customer_detail)
-
- def test_ops_customer_detail_rejected(self):
- with self.assertRaises(SqlGuardError) as cm:
- validate("SELECT customer_id, market_value FROM core_holding", "aggregate")
- self.assertEqual(cm.exception.error_code, "AUTH_403_SCOPE")
-
- def test_ops_aggregate_allowed(self):
- r = validate("SELECT COUNT(DISTINCT customer_id) AS cnt FROM core_holding", "aggregate")
- self.assertTrue(r.allowed)
-
- def test_ops_specific_customer_rejected(self):
- with self.assertRaises(SqlGuardError):
- validate("SELECT * FROM core_holding WHERE customer_id='CUST-1001'", "aggregate")
-
- def test_inject_ownership(self):
- out = inject_ownership("SELECT * FROM core_holding", ["CUST-1", "CUST-2"])
- self.assertIn("CUST-1", out)
- self.assertIn("WHERE customer_id IN", out)
-
- def test_extract_tables(self):
- self.assertEqual(
- extract_tables("SELECT * FROM core_customer c JOIN core_holding h ON h.customer_id=c.customer_id"),
- ["core_customer", "core_holding"],
- )
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_agent.py b/tests/test_wave6_analyst_agent.py
similarity index 76%
rename from tests/test_agent.py
rename to tests/test_wave6_analyst_agent.py
index 0b27ddd..f4beb5b 100644
--- a/tests/test_agent.py
+++ b/tests/test_wave6_analyst_agent.py
@@ -1,8 +1,10 @@
-"""analyst_agent 编排测试：确定性假件 + 一条真实端到端。"""
+"""analyst_agent 编排测试（Wave 6）。"""
import unittest
+import pytest
+
+from app.api.analyst_auth_adapter import AnalystAuthContext
from app.service.analyst_agent import AnalystAgent
-from app.utils.auth import AuthContext
class FakeLLM:
@@ -43,8 +45,13 @@ class FakeRepo:
pass
-def ctx(roles, subject="STAFF-A"):
- return AuthContext(subject_id=subject, token_type="staff", roles=roles, staff_type=roles[0] if roles else "")
+def ctx(roles, subject="STAFF-A", *, token_type="staff", customer_id=None):
+ return AnalystAuthContext(
+ subject_id=subject,
+ token_type=token_type,
+ roles=roles,
+ customer_id=customer_id,
+ )
class TestAgentOrchestration(unittest.TestCase):
@@ -89,8 +96,26 @@ class TestAgentOrchestration(unittest.TestCase):
self.assertEqual(resp.status, "deny")
self.assertEqual(resp.error_code, "AUTH_403_NOT_ASSIGNED")
+ def test_customer_self_success(self):
+ repo = FakeRepo(rows=[[2]], columns=["cnt"])
+ agent = AnalystAgent(
+ llm=FakeLLM(
+ "SELECT COUNT(*) AS cnt FROM core_trade WHERE customer_id='CUST-9527'",
+ ["近阶段共有 2 笔交易"],
+ ),
+ repo=repo,
+ )
+ resp = agent.run(
+ "我有多少笔交易",
+ ctx(["customer"], "CUST-9527", token_type="customer", customer_id="CUST-9527"),
+ )
+ self.assertEqual(resp.status, "success")
+ self.assertIn("AI 分析有风险", resp.answer)
+
+@pytest.mark.integration
class TestAgentReal(unittest.TestCase):
+ @pytest.mark.skip(reason="需要真实 MySQL + DeepSeek Key")
def test_real_end_to_end(self):
from app.service.analytics_repo import AnalyticsRepo
from app.service.llm import DeepSeekLLM
@@ -100,7 +125,3 @@ class TestAgentReal(unittest.TestCase):
self.assertIn(resp.status, ("success", "degrade"))
self.assertTrue(resp.sql)
self.assertGreater(resp.meta.row_count, 0)
-
-
-if __name__ == "__main__":
- unittest.main()
diff --git a/tests/test_cache_service.py b/tests/test_wave6_analyst_cache.py
similarity index 100%
rename from tests/test_cache_service.py
rename to tests/test_wave6_analyst_cache.py
diff --git a/tests/test_llm.py b/tests/test_wave6_analyst_llm.py
similarity index 58%
rename from tests/test_llm.py
rename to tests/test_wave6_analyst_llm.py
index 5d82eba..0033e69 100644
--- a/tests/test_llm.py
+++ b/tests/test_wave6_analyst_llm.py
@@ -1,7 +1,7 @@
"""llm 客户端测试（extract_sql 纯逻辑 + DeepSeek 真实冒烟）。"""
import unittest
-from app.service.llm import DeepSeekLLM, extract_sql
+from app.service.llm import extract_sql
class TestExtractSql(unittest.TestCase):
@@ -18,16 +18,5 @@ class TestExtractSql(unittest.TestCase):
self.assertEqual(extract_sql("```\nSELECT 2\n```"), "SELECT 2")
-class TestDeepSeekSmoke(unittest.TestCase):
- def test_complete(self):
- llm = DeepSeekLLM()
- text, usage = llm.complete(
- [{"role": "user", "content": "只回复两个字：正常"}],
- max_tokens=8,
- )
- self.assertTrue(text.strip())
- self.assertIn("prompt_tokens", usage)
-
-
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_repo.py b/tests/test_wave6_analytics_repo.py
similarity index 100%
rename from tests/test_repo.py
rename to tests/test_wave6_analytics_repo.py
diff --git a/tests/test_dict_service.py b/tests/test_wave6_dict_service.py
similarity index 100%
rename from tests/test_dict_service.py
rename to tests/test_wave6_dict_service.py
diff --git a/tests/test_guardrail.py b/tests/test_wave6_guardrail.py
similarity index 97%
rename from tests/test_guardrail.py
rename to tests/test_wave6_guardrail.py
index 6e48ad3..fe84dc8 100644
--- a/tests/test_guardrail.py
+++ b/tests/test_wave6_guardrail.py
@@ -1,7 +1,7 @@
"""guardrail 数字护栏单元测试。"""
import unittest
-from app.model.schemas.analyst import TableData
+from app.model.analyst_schemas import TableData
from app.service.guardrail import check_numbers, extract_numbers, result_numbers, verify
diff --git a/tests/test_wave6_sql_guard.py b/tests/test_wave6_sql_guard.py
new file mode 100644
index 0000000..b2e7463
--- /dev/null
+++ b/tests/test_wave6_sql_guard.py
@@ -0,0 +1,55 @@
+"""sql_guard 单元测试（Wave 6）。"""
+import unittest
+
+from app.service.sql_guard import SqlGuardError, extract_tables, inject_ownership, validate
+
+
+class TestSqlGuard(unittest.TestCase):
+ def test_select_allowed(self):
+ r = validate("SELECT risk_code, COUNT(*) AS c FROM core_customer GROUP BY risk_code", "full")
+ self.assertTrue(r.allowed)
+
+ def test_customer_self_in_scope(self):
+ r = validate(
+ "SELECT COUNT(*) FROM core_holding WHERE customer_id='CUST-9527'",
+ "self",
+ ["CUST-9527"],
+ )
+ self.assertTrue(r.allowed)
+
+ def test_customer_self_out_of_scope(self):
+ with self.assertRaises(SqlGuardError) as cm:
+ validate(
+ "SELECT * FROM core_holding WHERE customer_id='CUST-1001'",
+ "self",
+ ["CUST-9527"],
+ )
+ self.assertEqual(cm.exception.error_code, "AUTH_403_NOT_OWNER")
+
+ def test_insert_rejected(self):
+ with self.assertRaises(SqlGuardError) as cm:
+ validate("INSERT INTO core_customer VALUES (1)", "full")
+ self.assertEqual(cm.exception.error_code, "SQL_NOT_SELECT")
+
+ def test_advisor_out_of_scope_rejected(self):
+ with self.assertRaises(SqlGuardError) as cm:
+ validate(
+ "SELECT * FROM core_holding WHERE customer_id = 'CUST-1004'",
+ "assigned",
+ ["CUST-1001"],
+ )
+ self.assertEqual(cm.exception.error_code, "AUTH_403_NOT_ASSIGNED")
+
+ def test_ops_aggregate_allowed(self):
+ r = validate("SELECT COUNT(DISTINCT customer_id) AS cnt FROM core_holding", "aggregate")
+ self.assertTrue(r.allowed)
+
+ def test_inject_ownership(self):
+ out = inject_ownership("SELECT * FROM core_holding", ["CUST-1", "CUST-2"])
+ self.assertIn("CUST-1", out)
+
+ def test_extract_tables(self):
+ self.assertEqual(
+ extract_tables("SELECT * FROM core_customer c JOIN core_holding h ON h.customer_id=c.customer_id"),
+ ["core_customer", "core_holding"],
+ )
diff --git a/web/src/App.tsx b/web/src/App.tsx
index fbc47b5..0e77269 100644
--- a/web/src/App.tsx
+++ b/web/src/App.tsx
@@ -1,7 +1,7 @@
import { Navigate, Route, Routes } from 'react-router-dom'
import { AppLayout } from './layouts/AppLayout'
import { LoginPage } from './pages/login/LoginPage'
-import { PlaceholderPage } from './pages/PlaceholderPage'
+import { AnalystQueryPage } from './pages/analytics/AnalystQueryPage'
import { loadAuth } from './stores/authStore'
import { AgentBanner } from './components/AgentBanner'
import { CustomerWealthDashboard } from './pages/dashboard/CustomerWealthDashboard'
@@ -87,12 +87,12 @@ export default function AppRoutes() {
/>
} />
- } />
+ } />
-
+
}
diff --git a/web/src/api/analyst.ts b/web/src/api/analyst.ts
new file mode 100644
index 0000000..53a81f6
--- /dev/null
+++ b/web/src/api/analyst.ts
@@ -0,0 +1,40 @@
+import { apiFetch } from './client'
+
+export type AnalystTable = {
+ columns: string[]
+ rows: unknown[][]
+}
+
+export type AnalystMeta = {
+ exec_ms: number
+ row_count: number
+ cache_hit: boolean
+ data_as_of?: string
+ source?: string
+ cost_est?: number
+}
+
+export type AnalystChatResponse = {
+ answer: string
+ table: AnalystTable
+ sql: string
+ meta: AnalystMeta
+ disclaimer: string
+ status: string
+ error_code?: string | null
+ suggestions?: string[] | null
+ trace_id?: string | null
+}
+
+export async function postAnalystChat(
+ token: string,
+ question: string,
+ sessionId?: string,
+): Promise {
+ const { data } = await apiFetch('/api/analyst/chat', {
+ method: 'POST',
+ token,
+ body: JSON.stringify({ question, session_id: sessionId ?? null }),
+ })
+ return data
+}
diff --git a/web/src/pages/analytics/AnalystQueryPage.tsx b/web/src/pages/analytics/AnalystQueryPage.tsx
new file mode 100644
index 0000000..32aeb63
--- /dev/null
+++ b/web/src/pages/analytics/AnalystQueryPage.tsx
@@ -0,0 +1,138 @@
+import { Alert, Button, Card, Collapse, Input, Space, Spin, Table, Typography } from 'antd'
+import type { ColumnsType } from 'antd/es/table'
+import { useState } from 'react'
+import { postAnalystChat, type AnalystChatResponse } from '../../api/analyst'
+import { ApiError } from '../../api/client'
+import { ApiErrorResult } from '../../components/ApiErrorResult'
+import { PageShell } from '../../components/PageShell'
+import { useAppAuth } from '../../layouts/AppLayout'
+
+const { Text, Paragraph } = Typography
+
+type RowRecord = Record & { key: number }
+
+export function AnalystQueryPage() {
+ const auth = useAppAuth()
+ const [question, setQuestion] = useState('')
+ const [loading, setLoading] = useState(false)
+ const [error, setError] = useState(null)
+ const [result, setResult] = useState(null)
+
+ async function onAsk() {
+ const q = question.trim()
+ if (!q) return
+ setLoading(true)
+ setError(null)
+ try {
+ const resp = await postAnalystChat(auth.accessToken, q)
+ setResult(resp)
+ } catch (e) {
+ setError(e instanceof Error ? e : new Error(String(e)))
+ setResult(null)
+ } finally {
+ setLoading(false)
+ }
+ }
+
+ const columns: ColumnsType =
+ result?.table.columns.map((col: string, i: number) => ({
+ title: col,
+ key: `${col}-${i}`,
+ render: (_: unknown, record: RowRecord) => String(record[`c${i}`] ?? ''),
+ })) ?? []
+
+ const dataSource: RowRecord[] =
+ result?.table.rows.map((row: unknown[], idx: number) => {
+ const record: RowRecord = { key: idx }
+ row.forEach((cell: unknown, i: number) => {
+ record[`c${i}`] = cell
+ })
+ return record
+ }) ?? []
+
+ return (
+
+ }
+ >
+
+
+ setQuestion(e.target.value)}
+ placeholder="例如：我近30天有多少笔交易？ / 待处理预警按类型统计"
+ onPressEnter={(e) => {
+ if (!e.shiftKey) {
+ e.preventDefault()
+ void onAsk()
+ }
+ }}
+ />
+
+
+
+ {error ? : null}
+
+ {loading ? : null}
+
+ {result ? (
+ <>
+ {result.status === 'clarify' ? : null}
+ {result.status === 'deny' ? (
+
+ ) : null}
+ {(result.status === 'success' || result.status === 'degrade') && result.answer ? (
+
+ {result.answer}
+ {result.disclaimer ? (
+
+ {result.disclaimer}
+
+ ) : null}
+
+ ) : null}
+ {result.table.rows.length > 0 ? (
+
+
+ size="small"
+ rowKey="key"
+ pagination={{ pageSize: 10 }}
+ columns={columns}
+ dataSource={dataSource}
+ scroll={{ x: true }}
+ />
+
+ ) : null}
+ {result.sql ? (
+ {result.sql},
+ },
+ ]}
+ />
+ ) : null}
+
+ ) : null}
+
+
+ )
+}