迭代升级

.env.example

@@ -89,3 +89,16 @@ ANALYST_USERNAME=analyst
 ANALYST_PASSWORD_HASH=
 VIEWER_USERNAME=viewer
 VIEWER_PASSWORD_HASH=
+
+# ---------- 多租户隔离（可选，默认单租户模式） ----------
+# 租户敏感表名集合，逗号分隔；留空则不启用多租户，所有行为与单租户一致
+# 启用后 run_sql 节点会用 sqlglot AST 强制向这些表注入 WHERE tenant_id = X 行级过滤
+# 示例：TENANT_SCOPED_TABLES=orders,order_items,users,addresses
+TENANT_SCOPED_TABLES=
+# 租户字段名，默认 tenant_id，仅当业务表使用其他字段名时需要覆盖
+TENANT_COLUMN=tenant_id
+# 演示账号的租户标识（可选，未配置时回退到 default 单租户）
+# 多租户场景下不同角色可归属不同租户，例如 admin 归属 tenant_a、analyst 归属 tenant_b
+# ADMIN_TENANT_ID=tenant_a
+# ANALYST_TENANT_ID=tenant_b
+# VIEWER_TENANT_ID=default

Dockerfile

@@ -3,11 +3,12 @@
 FROM python:3.14-slim AS builder
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    gcc g++ default-libmysqlclient-dev pkg-config \
-    && rm -rf /var/lib/apt/lists/*
+  gcc g++ default-libmysqlclient-dev pkg-config \
+  && rm -rf /var/lib/apt/lists/*
 
 # 引入 uv 做依赖管理，比 pip 更快且严格按 lockfile 安装
-COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+# 固定版本保证构建可复现，避免 latest 升级破坏 uv sync 行为
+COPY --from=ghcr.io/astral-sh/uv:0.5.4 /uv /usr/local/bin/uv
 
 WORKDIR /app
 
@@ -21,10 +22,10 @@ RUN uv sync --frozen --no-dev
 # 运行时镜像不含编译工具，体积更小，攻击面更窄
 FROM python:3.14-slim
 
-# asyncmy 运行时需要 MySQL 客户端库
+# asyncmy 运行时只需 MySQL 客户端运行时库，不装 dev 头文件以减小镜像体积和攻击面
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    default-libmysqlclient-dev \
-    && rm -rf /var/lib/apt/lists/*
+  libmariadb3 \
+  && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app
 

app/agent/context.py

@@ -34,3 +34,7 @@ class DataAgentContext(TypedDict):
     meta_mysql_repository: MetaMySQLRepository
     # 数仓仓储抽象，具体实现由 dependencies 按配置注入（MySQL / ClickHouse / Postgres）
     dw_repository: DWRepository
+    # 当前请求的租户标识，由 query_service 从 JWT 解析后注入
+    # run_sql 节点据此调用 inject_tenant_filter 在 SQL 执行前强制注入 WHERE tenant_id = X
+    # 未启用多租户（TENANT_SCOPED_TABLES 为空）时为 DEFAULT_TENANT_ID，inject_tenant_filter 原样返回 SQL
+    tenant_id: str

app/agent/graph.py

@@ -146,6 +146,8 @@ async def test():
                 value_es_repository=value_es_repository,
                 meta_mysql_repository=meta_mysql_repository,
                 dw_repository=dw_repository,
+                # 本地调试使用默认租户；生产环境由 query_service 从 JWT 解析注入
+                tenant_id="default",
             )
 
             # stream_mode="custom" 会接收各节点通过 runtime.stream_writer 写出的进度信息

app/agent/nodes/correct_sql.py

@@ -15,6 +15,7 @@
 from app.agent.state import DataAgentState
 from app.core.llm_tracing import get_tracing_config
 from app.core.log import logger
+from app.core.resilience import CircuitOpenError, call_llm_with_resilience
 from app.prompt.prompt_loader import load_prompt
 
 
@@ -53,23 +54,37 @@ async def correct_sql(state: DataAgentState, runtime: Runtime[DataAgentContext])
         output_parser = StrOutputParser()
         chain = prompt | llm | output_parser
 
-        result = await chain.ainvoke(
-            {
-                # 与生成节点保持一致，用 YAML 向模型提供稳定 可读的结构化上下文
-                "table_infos": yaml.dump(
-                    table_infos, allow_unicode=True, sort_keys=False
-                ),
-                "metric_infos": yaml.dump(
-                    metric_infos, allow_unicode=True, sort_keys=False
-                ),
-                "date_info": yaml.dump(date_info, allow_unicode=True, sort_keys=False),
-                "db_info": yaml.dump(db_info, allow_unicode=True, sort_keys=False),
-                "query": query,
-                "sql": sql,
-                "error": error,
-            },
-            config=get_tracing_config(node="correct_sql"),
-        )
+        # LLM 调用统一走熔断 + 重试包装：临时性失败自动重试，连续失败触发熔断时返回降级空 SQL
+        # 返回空 SQL 会让后续 validate_sql 自然失败，配合 correction_count 递增防止无限循环
+        try:
+            result = await call_llm_with_resilience(
+                chain.ainvoke,
+                {
+                    # 与生成节点保持一致，用 YAML 向模型提供稳定 可读的结构化上下文
+                    "table_infos": yaml.dump(
+                        table_infos, allow_unicode=True, sort_keys=False
+                    ),
+                    "metric_infos": yaml.dump(
+                        metric_infos, allow_unicode=True, sort_keys=False
+                    ),
+                    "date_info": yaml.dump(date_info, allow_unicode=True, sort_keys=False),
+                    "db_info": yaml.dump(db_info, allow_unicode=True, sort_keys=False),
+                    "query": query,
+                    "sql": sql,
+                    "error": error,
+                },
+                config=get_tracing_config(node="correct_sql"),
+            )
+        except CircuitOpenError:
+            logger.warning("LLM 熔断中，correct_sql 返回降级空 SQL")
+            writer(
+                {
+                    "type": "error",
+                    "code": "LLM_CIRCUIT_OPEN",
+                    "message": "智能体服务暂时繁忙，请稍后重试",
+                }
+            )
+            return {"sql": "", "correction_count": state.get("correction_count", 0) + 1}
 
         logger.info(f"校正后的SQL：{result}")
         writer({"type": "progress", "step": step, "status": "success"})

app/agent/nodes/filter_metric.py

@@ -15,6 +15,7 @@
 from app.agent.state import DataAgentState, MetricInfoState
 from app.core.llm_tracing import get_tracing_config
 from app.core.log import logger
+from app.core.resilience import CircuitOpenError, call_llm_with_resilience
 from app.prompt.prompt_loader import load_prompt
 
 
@@ -39,15 +40,33 @@ async def filter_metric(state: DataAgentState, runtime: Runtime[DataAgentContext
         # LCEL 管道：填充提示词 -> 调用模型 -> 解析 JSON
         chain = prompt | llm | output_parser
 
-        result = await chain.ainvoke(
-            {
-                "query": query,
-                "metric_infos": yaml.dump(
-                    metric_infos, allow_unicode=True, sort_keys=False
-                ),
-            },
-            config=get_tracing_config(node="filter_metric"),
-        )
+        # LLM 调用统一走熔断 + 重试包装；熔断时跳过过滤直接返回原始候选指标，保留完整口径供下游生成
+        try:
+            result = await call_llm_with_resilience(
+                chain.ainvoke,
+                {
+                    "query": query,
+                    "metric_infos": yaml.dump(
+                        metric_infos, allow_unicode=True, sort_keys=False
+                    ),
+                },
+                config=get_tracing_config(node="filter_metric"),
+            )
+        except CircuitOpenError:
+            logger.warning("LLM 熔断中，filter_metric 跳过过滤，返回原始候选指标")
+            writer(
+                {
+                    "type": "error",
+                    "code": "LLM_CIRCUIT_OPEN",
+                    "message": "智能体服务暂时繁忙，请稍后重试",
+                }
+            )
+            writer({"type": "progress", "step": step, "status": "success"})
+            return {"metric_infos": metric_infos}
+        # LLM 应返回指标名列表，校验后按空过滤处理避免类型错误
+        if not isinstance(result, list):
+            logger.warning(f"filter_metric LLM 返回非 list 类型：{type(result).__name__}")
+            result = []
         # 用模型返回的指标名称过滤原始结构，保留描述 依赖字段 别名等完整上下文
         filtered_metric_infos = [
             metric_info for metric_info in metric_infos if metric_info["name"] in result

app/agent/nodes/filter_table.py

@@ -15,6 +15,7 @@
 from app.agent.state import DataAgentState, TableInfoState
 from app.core.llm_tracing import get_tracing_config
 from app.core.log import logger
+from app.core.resilience import CircuitOpenError, call_llm_with_resilience
 from app.prompt.prompt_loader import load_prompt
 
 
@@ -39,15 +40,33 @@ async def filter_table(state: DataAgentState, runtime: Runtime[DataAgentContext]
         # LCEL 管道：填充提示词 -> 调用模型 -> 解析 JSON
         chain = prompt | llm | output_parser
 
-        result = await chain.ainvoke(
-            {
-                "query": query,
-                "table_infos": yaml.dump(
-                    table_infos, allow_unicode=True, sort_keys=False
-                ),
-            },
-            config=get_tracing_config(node="filter_table"),
-        )
+        # LLM 调用统一走熔断 + 重试包装；熔断时跳过过滤直接返回原始候选表，保留完整上下文供下游生成
+        try:
+            result = await call_llm_with_resilience(
+                chain.ainvoke,
+                {
+                    "query": query,
+                    "table_infos": yaml.dump(
+                        table_infos, allow_unicode=True, sort_keys=False
+                    ),
+                },
+                config=get_tracing_config(node="filter_table"),
+            )
+        except CircuitOpenError:
+            logger.warning("LLM 熔断中，filter_table 跳过过滤，返回原始候选表")
+            writer(
+                {
+                    "type": "error",
+                    "code": "LLM_CIRCUIT_OPEN",
+                    "message": "智能体服务暂时繁忙，请稍后重试",
+                }
+            )
+            writer({"type": "progress", "step": step, "status": "success"})
+            return {"table_infos": table_infos}
+        # LLM 可能返回非预期类型（如 list），校验后按空过滤处理避免 TypeError
+        if not isinstance(result, dict):
+            logger.warning(f"filter_table LLM 返回非 dict 类型：{type(result).__name__}")
+            result = {}
         # 模型只负责选择，程序根据选择结果从原始 TableInfoState 中裁剪，避免模型重写复杂结构出错
         filtered_table_infos: list[TableInfoState] = []
         for table_info in table_infos:

app/agent/nodes/recall_column.py

@@ -15,6 +15,7 @@
 from app.agent.state import DataAgentState
 from app.core.llm_tracing import get_tracing_config
 from app.core.log import logger
+from app.core.resilience import CircuitOpenError, call_llm_with_resilience
 from app.entities.column_info import ColumnInfo
 from app.prompt.prompt_loader import load_prompt
 
@@ -44,10 +45,27 @@ async def recall_column(state: DataAgentState, runtime: Runtime[DataAgentContext
         # LCEL 管道：填充提示词 -> 调用模型 -> 解析 JSON
         chain = prompt | llm | output_parser
 
-        result = await chain.ainvoke(
-            {"query": query},
-            config=get_tracing_config(node="recall_column"),
-        )
+        # LLM 调用统一走熔断 + 重试包装；熔断时跳过 LLM 扩展，仅用原始关键词做向量召回
+        try:
+            result = await call_llm_with_resilience(
+                chain.ainvoke,
+                {"query": query},
+                config=get_tracing_config(node="recall_column"),
+            )
+        except CircuitOpenError:
+            logger.warning("LLM 熔断中，recall_column 跳过扩展词，仅用原始关键词召回")
+            writer(
+                {
+                    "type": "error",
+                    "code": "LLM_CIRCUIT_OPEN",
+                    "message": "智能体服务暂时繁忙，请稍后重试",
+                }
+            )
+            result = []
+        # LLM 应返回扩展词列表，校验后取空列表避免 keywords + result 报 TypeError
+        if not isinstance(result, list):
+            logger.warning(f"recall_column LLM 返回非 list 类型：{type(result).__name__}")
+            result = []
 
         # 原始关键词和 LLM 扩展词一起参与召回；set 去重，避免重复请求同一关键词
         keywords_list = list(set(keywords + result))

app/agent/nodes/recall_metric.py

@@ -15,6 +15,7 @@
 from app.agent.state import DataAgentState
 from app.core.llm_tracing import get_tracing_config
 from app.core.log import logger
+from app.core.resilience import CircuitOpenError, call_llm_with_resilience
 from app.entities.metric_info import MetricInfo
 from app.prompt.prompt_loader import load_prompt
 
@@ -44,10 +45,27 @@ async def recall_metric(state: DataAgentState, runtime: Runtime[DataAgentContext
         # LCEL 管道：填充提示词 -> 调用模型 -> 解析 JSON
         chain = prompt | llm | output_parser
 
-        result = await chain.ainvoke(
-            {"query": query},
-            config=get_tracing_config(node="recall_metric"),
-        )
+        # LLM 调用统一走熔断 + 重试包装；熔断时跳过 LLM 扩展，仅用原始关键词做向量召回
+        try:
+            result = await call_llm_with_resilience(
+                chain.ainvoke,
+                {"query": query},
+                config=get_tracing_config(node="recall_metric"),
+            )
+        except CircuitOpenError:
+            logger.warning("LLM 熔断中，recall_metric 跳过扩展词，仅用原始关键词召回")
+            writer(
+                {
+                    "type": "error",
+                    "code": "LLM_CIRCUIT_OPEN",
+                    "message": "智能体服务暂时繁忙，请稍后重试",
+                }
+            )
+            result = []
+        # LLM 应返回扩展词列表，校验后取空列表避免 keywords + result 报 TypeError
+        if not isinstance(result, list):
+            logger.warning(f"recall_metric LLM 返回非 list 类型：{type(result).__name__}")
+            result = []
 
         # 通用关键词和指标扩展词都参与召回，提升同义指标的命中率
         keywords_list = list(set(keywords + result))

app/agent/nodes/recall_value.py

@@ -13,7 +13,9 @@
 from app.agent.context import DataAgentContext
 from app.agent.llm import llm
 from app.agent.state import DataAgentState
+from app.core.llm_tracing import get_tracing_config
 from app.core.log import logger
+from app.core.resilience import CircuitOpenError, call_llm_with_resilience
 from app.entities.value_info import ValueInfo
 from app.prompt.prompt_loader import load_prompt
 
@@ -43,7 +45,27 @@ async def recall_value(state: DataAgentState, runtime: Runtime[DataAgentContext]
         # LCEL 管道：填充提示词 -> 调用模型 -> 解析 JSON
         chain = prompt | llm | output_parser
 
-        result = await chain.ainvoke({"query": query})
+        # LLM 调用统一走熔断 + 重试包装；熔断时跳过 LLM 扩展，仅用原始关键词做 ES 检索
+        try:
+            result = await call_llm_with_resilience(
+                chain.ainvoke,
+                {"query": query},
+                config=get_tracing_config(node="recall_value"),
+            )
+        except CircuitOpenError:
+            logger.warning("LLM 熔断中，recall_value 跳过扩展词，仅用原始关键词检索")
+            writer(
+                {
+                    "type": "error",
+                    "code": "LLM_CIRCUIT_OPEN",
+                    "message": "智能体服务暂时繁忙，请稍后重试",
+                }
+            )
+            result = []
+        # LLM 应返回扩展词列表，校验后取空列表避免 keywords + result 报 TypeError
+        if not isinstance(result, list):
+            logger.warning(f"recall_value LLM 返回非 list 类型：{type(result).__name__}")
+            result = []
 
         # 通用关键词和字段值扩展词一起检索 ES，尽量提高真实取值召回率
         keywords_list = list(set(keywords + result))

app/agent/nodes/run_sql.py

@@ -12,6 +12,12 @@
 from app.agent.state import DataAgentState
 from app.core.log import logger
 from app.core.metrics import SQL_SAFETY_VIOLATIONS_TOTAL
+from app.core.tenancy import (
+    get_tenant_column,
+    get_tenant_scoped_tables,
+    inject_tenant_filter,
+    resolve_tenant_id,
+)
 
 
 async def run_sql(state: DataAgentState, runtime: Runtime[DataAgentContext]):
@@ -33,6 +39,34 @@ async def run_sql(state: DataAgentState, runtime: Runtime[DataAgentContext]):
         safe_sql = validate_readonly_sql(
             sql, allowed_tables, dialect=dw_repository.dialect
         )
+
+        # 多租户行级过滤：在 SQL 安全校验通过后、执行前，用 sqlglot AST 强制注入
+        # WHERE tenant_id = X 条件，只对配置中声明的租户敏感表生效
+        # 未配置 TENANT_SCOPED_TABLES 时 tenant_scoped_tables 为空集合，原样返回 SQL
+        tenant_id = resolve_tenant_id(runtime.context.get("tenant_id"))
+        tenant_scoped_tables = get_tenant_scoped_tables()
+        if tenant_scoped_tables:
+            try:
+                safe_sql = inject_tenant_filter(
+                    safe_sql,
+                    tenant_id,
+                    tenant_scoped_tables,
+                    tenant_column=get_tenant_column(),
+                    dialect=dw_repository.dialect,
+                )
+            except ValueError as e:
+                # AST 注入失败属于安全错误：拒绝执行，避免无租户过滤的 SQL 落库
+                logger.error(f"租户过滤注入失败：{e}")
+                writer({"type": "progress", "step": step, "status": "error"})
+                writer(
+                    {
+                        "type": "error",
+                        "code": "TENANT_FILTER_FAILED",
+                        "message": "SQL 租户隔离注入失败，请联系管理员",
+                    }
+                )
+                return
+
         logger.info(f"通过安全校验的SQL：{safe_sql}")
 
         # 真实数据库访问统一封装在仓储层，节点只负责从状态取 SQL 并触发执行