.env.example

@@ -64,8 +64,9 @@ LLM_API_EMBED_MODEL=text-embedding-3-small
 # 须与模型输出一致（qwen3-embedding 为 MRL 模型，可任选 32~4096）
 LLM_EMBED_DIMENSIONS=1024
 LLM_TEMPERATURE=0.3
-LLM_MAX_TOKENS=1024
-LLM_TIMEOUT=30
-LLM_MAX_RETRIES=3
+LLM_MAX_TOKENS=2048
+LLM_TIMEOUT=60
+LLM_MAX_RETRIES=2
 LLM_RETRY_BACKOFF_SEC=1
 LLM_FALLBACK_CHAT_MODEL=   # 备用模型：主模型失败自动切换（留空则不启用）
+LLM_API_TRUST_ENV=false

agent/advisor_agent/data_query.py

@@ -7,15 +7,19 @@ from uuid import uuid4
 
 from agent.advisor_agent.auth import ensure_customer_access
 from agent.data_query.agent import DataQueryAgent
+from common.common_const import CUSTOMER_REL_STATUS_SIGNED, CUSTOMER_REL_STATUS_UNSIGNED
 from config import database
 from config.settings import settings
 from nl2sql.contracts import DataQueryRequest, DataQueryResult
+from nl2sql.embedding import EmbeddingError
 from nl2sql.retrieval import retrieve_metadata
 from nl2sql.runtime_config import runtime_config
 from nl2sql.schema import load_authoritative_schema
+from repositories.customer_relation import CustomerRelationRepo
 from service.nl2sql.permission_service import load_query_permission
 from service.nl2sql.query_service import QueryServiceError
 from tool.llm import llm as default_llm
+from utils.exceptions import LLMFailError
 from repositories.fin_holdings import FinHoldingsRepo
 from repositories.fin_product import FinProductRepo
 
@@ -67,7 +71,8 @@ async def execute_advisor_data_query(
     db,
     *,
     advisor_id: int,
-    customer_id: int,
+    customer_id: int | None,
+    scope: str = "customer",
     question: str,
     trace_id: str,
     session_id: str | None = None,
@@ -82,13 +87,27 @@ async def execute_advisor_data_query(
     llm_client=None,
     query_agent=None,
 ) -> dict[str, Any]:
-    """在当前投顾和选中客户范围内执行只读自然语言查询。
+    """在当前投顾或选中客户范围内执行只读自然语言查询。
 
     ``data_scope`` 即使由调用方传入也不会被信任，服务端始终覆盖为当前
-    ``customer_id``，避免投顾借助 NL2SQL 查询其他客户数据。
+    客户关系范围，避免投顾借助 NL2SQL 查询其他客户数据。
     """
-    await ensure_customer_access(db, advisor_id=advisor_id, customer_id=customer_id)
-    if _is_current_holdings_query(question):
+    if scope == "advisor":
+        relations = await CustomerRelationRepo(db).list_by_advisor(advisor_id)
+        customer_ids = [
+            relation.customer_id
+            for relation in relations
+            if relation.status in {CUSTOMER_REL_STATUS_UNSIGNED, CUSTOMER_REL_STATUS_SIGNED}
+        ]
+        if not customer_ids:
+            raise QueryServiceError("当前投顾名下没有可查询客户")
+    else:
+        if customer_id is None:
+            raise QueryServiceError("单客户查询需要明确客户范围")
+        await ensure_customer_access(db, advisor_id=advisor_id, customer_id=customer_id)
+        customer_ids = [customer_id]
+
+    if scope == "customer" and _is_current_holdings_query(question):
         return await _query_current_holdings(
             db,
             customer_id=customer_id,
@@ -107,7 +126,7 @@ async def execute_advisor_data_query(
         trace_id=trace_id,
         session_id=session_id,
         caller_agent="advisor_agent",
-        data_scope={"customer_ids": [customer_id]},
+        data_scope={"customer_ids": customer_ids},
         max_rows=min(max_rows or runtime_config.max_rows, runtime_config.max_rows),
         include_sql=False,
         page=page,
@@ -133,17 +152,19 @@ async def execute_advisor_data_query(
             candidate_tables=table_names,
         )
 
-    result: DataQueryResult = await (query_agent or DataQueryAgent()).query(
-        request,
-        session=db,
-        permission_loader=permission_loader,
-        metadata_retriever=metadata_retriever,
-        schema_loader=schema_loader,
-        llm_client=llm_client,
-        summary_llm=llm_client,
-        masks=permission.get("masks"),
-        redis=redis,
-    )
+    try:
+        result: DataQueryResult = await (query_agent or DataQueryAgent()).query(
+            request,
+            session=db,
+            permission_loader=permission_loader,
+            metadata_retriever=metadata_retriever,
+            schema_loader=schema_loader,
+            llm_client=llm_client,
+            summary_llm=llm_client,
+            masks=permission.get("masks"),
+        )
+    except (EmbeddingError, LLMFailError) as exc:
+        raise QueryServiceError("投顾 Agent 依赖服务不可用，请检查 LLM/Embedding 服务连接") from exc
     payload = asdict(result)
     payload["sql"] = None
     payload["customer_id"] = customer_id

api/routers/advisor_agent.py

@@ -97,6 +97,14 @@ def _not_ready(request: Request):
     return agent_failure(_NOT_READY_CODE, _NOT_READY_MESSAGE, trace_id=_trace_id(request))
 
 
+def _data_query_error_message(exc: QueryServiceError) -> str:
+    """Expose dependency outages without leaking SQL or database details."""
+    message = str(exc)
+    if message.startswith("投顾 Agent 依赖服务不可用"):
+        return message
+    return "客户数据查询失败，请稍后重试"
+
+
 def _advisor_runtime(request: Request):
     app = request.scope.get("app")
     return getattr(getattr(app, "state", None), "advisor_agent_runtime", None)
@@ -251,8 +259,8 @@ async def chat_stream(
 
     payload = None
 
-    # 请求体只传问题时，从问题中解析客户；解析结果仍必须经过投顾关系授权校验。
-    if customer_id is None:
+    # 单客户范围且未传编号时，兼容从问题中解析客户；投顾范围查询不解析客户。
+    if chat_request.scope == "customer" and customer_id is None:
         customer_id, resolve_error = await _resolve_customer_from_query(
             db,
             advisor_id=user.id,
@@ -276,7 +284,19 @@ async def chat_stream(
             payload = None
 
     # 不带客户编号时只提供通用基金问答，不读取客户画像，也不生成个性化草稿。
-    if customer_id is None:
+    if chat_request.scope == "advisor" and inferred_intent != AGENT_INTENT_DATA_QUERY:
+        payload = agent_failure(
+            ERR_CODE_FORBIDDEN_CUSTOMER,
+            "投顾范围查询仅支持客户数据查询",
+            trace_id=trace_id,
+        )
+    elif chat_request.scope == "advisor" and customer_id is not None:
+        payload = agent_failure(
+            ERR_CODE_FORBIDDEN_CUSTOMER,
+            "投顾范围查询不能指定单个客户",
+            trace_id=trace_id,
+        )
+    elif chat_request.scope == "customer" and customer_id is None:
         if inferred_intent in {
             AGENT_INTENT_RECOMMEND,
             AGENT_INTENT_REBALANCE,
@@ -319,10 +339,12 @@ async def chat_stream(
                 headers={"X-Trace-Id": trace_id},
             )
     else:
-        customer_id = chat_request.customer_id
-        relation = await ensure_customer_access(
-            db, advisor_id=user.id, customer_id=int(customer_id)
-        )
+        relation = None
+        if chat_request.scope == "customer":
+            customer_id = chat_request.customer_id
+            relation = await ensure_customer_access(
+                db, advisor_id=user.id, customer_id=int(customer_id)
+            )
         if inferred_intent == AGENT_INTENT_RECOMMEND:
             memories = await _recall_advisor_memories(
                 request,
@@ -382,15 +404,16 @@ async def chat_stream(
                     result = await execute_advisor_data_query(
                         db,
                         advisor_id=user.id,
-                        customer_id=int(customer_id),
+                        customer_id=int(customer_id) if customer_id is not None else None,
+                        scope=chat_request.scope,
                         question=chat_request.query,
                         trace_id=trace_id,
                         llm_client=getattr(_advisor_runtime(request), "llm_client", None),
                     )
-                except QueryServiceError:
+                except QueryServiceError as exc:
                     payload = agent_failure(
                         ERR_CODE_LLM_ERROR,
-                        "客户数据查询失败，请稍后重试",
+                        _data_query_error_message(exc),
                         trace_id=trace_id,
                     )
                 else:
@@ -507,10 +530,10 @@ async def advisor_data_query(
             sort_by=body.sort_by,
             sort_order=body.sort_order,
         )
-    except QueryServiceError:
+    except QueryServiceError as exc:
         return agent_failure(
             ERR_CODE_LLM_ERROR,
-            "客户数据查询失败，请稍后重试",
+            _data_query_error_message(exc),
             trace_id=trace_id,
         )
     result.pop("sql", None)

api/routers/nl2sql.py

@@ -119,14 +119,16 @@ def history_payload(history) -> dict:
 
 async def enrich_query_result(question: str, result, *, llm_client):
     """为已脱敏结果补充摘要和安全的基础图表配置。"""
+    summary = await summarize_result(
+        question,
+        result.columns,
+        result.rows,
+        llm_client=llm_client,
+    )
     return replace(
         result,
-        summary=await summarize_result(
-            question,
-            result.columns,
-            result.rows,
-            llm_client=llm_client,
-        ),
+        summary=summary,
+        answer=summary,
         chart=build_chart_config(result.columns, result.rows),
     )
 
@@ -301,6 +303,11 @@ async def query_data(
                     warnings=[*cached_result.warnings, "cache_hit"],
                 )
             else:
+                parameters = (
+                    {"advisor_id": user.id}
+                    if ":advisor_id" in validated_sql.sql
+                    else None
+                )
                 result = await execute_readonly_sql(
                     db,
                     validated_sql,
@@ -308,8 +315,14 @@ async def query_data(
                     trace_id=trace_id,
                     user_id=user.id,
                     masks=permission.get("masks"),
+                    parameters=parameters,
                 )
         else:
+            parameters = (
+                {"advisor_id": user.id}
+                if ":advisor_id" in validated_sql.sql
+                else None
+            )
             result = await execute_readonly_sql(
                 db,
                 validated_sql,
@@ -317,6 +330,7 @@ async def query_data(
                 trace_id=trace_id,
                 user_id=user.id,
                 masks=permission.get("masks"),
+                parameters=parameters,
             )
         if result.summary is None:
             result = await enrich_query_result(body.question, result, llm_client=llm)

config/database/redis.py

@@ -19,7 +19,6 @@ def client() -> aioredis.Redis:
             socket_connect_timeout=settings.redis.socket_connect_timeout,
             socket_timeout=settings.redis.socket_timeout,
             socket_keepalive=True,          # 长连接保活，防代理/防火墙断连
-            retry_on_timeout=True,          # 读超时自动重试，吸收瞬时抖动
             health_check_interval=settings.redis.health_check_interval,
             retry=Retry(FullJitterBackoff(base=1, cap=10), retries=3),
             max_connections=settings.redis.max_connections,

config/settings.py

@@ -96,6 +96,7 @@ class LLMCfg(BaseSettings):
     api_key: str = ""
     api_chat_model: str = ""
     api_embed_model: str = ""
+    api_trust_env: bool = True  # API 客户端是否读取系统代理环境变量
     embed_dimensions: int  # 向量维度：既作为 embeddings 请求的 dimensions 参数，也是 Milvus 建表/入库校验的维度；须与模型输出一致（MRL 模型如 qwen3-embedding 可任选 32~4096）
     # —— 通用超参 ——
     temperature: float

nl2sql/executor.py

@@ -27,6 +27,7 @@ async def execute_readonly_sql(
     masks: dict[tuple[str, str], str] | None = None,
     max_rows: int | None = None,
     timeout_seconds: float | None = None,
+    parameters: dict[str, object] | None = None,
     user_id: int = 0,
     connection_id: int | None = None,
     kill_query=None,
@@ -49,7 +50,12 @@ async def execute_readonly_sql(
             connection_id=connection_id,
         )
     try:
-        execution = session.execute(text(validated_sql.sql))
+        statement = text(validated_sql.sql)
+        execution = (
+            session.execute(statement, parameters)
+            if parameters
+            else session.execute(statement)
+        )
         result = (
             await asyncio.wait_for(execution, timeout_seconds)
             if timeout_seconds is not None

nl2sql/row_scope.py

@@ -35,7 +35,7 @@ def apply_row_scope(sql: str, permission: dict, data_scope: dict | None) -> str:
             continue
         values = _values(data_scope, scope["type"])
         condition = exp.In(
-            this=exp.column(scope["column"]),
+            this=exp.column(scope["column"], table=table.alias_or_name),
             expressions=[exp.Literal.number(value) for value in values],
         )
         statement = statement.where(condition)

nl2sql/semantics.py

@@ -107,18 +107,26 @@ def resolve_semantics(question: str, *, catalog: dict[str, Any] | None = None) -
         {
             "term": item["term"],
             "field": item["fields"][0],
-            "hint": item.get("metric_hint") or item.get("value_hint", ""),
+            "hint": item["metric_hint"],
         }
-        # metric_hint（指标口径）与 value_hint（字段取值口径）都要进入
-        # SQL 生成上下文；只过滤 metric_hint 会让枚举值提示永远丢失。
         for item in matched
-        if "metric_hint" in item or "value_hint" in item
+        if item.get("metric_hint")
+    ]
+    value_hints = [
+        {
+            "term": item["term"],
+            "field": item["fields"][0],
+            "hint": item["value_hint"],
+        }
+        for item in matched
+        if item.get("value_hint")
     ]
     return {
         "terms": [item["term"] for item in matched],
         "tables": tables,
         "fields": fields,
         "metrics": metrics,
+        "value_hints": value_hints,
         "relationships": list(catalog.get("relationships", [])),
     }
 
@@ -149,6 +157,11 @@ def build_semantic_context(question: str, schema: dict[str, Any]) -> dict[str, A
         for metric in resolved["metrics"]
         if any(metric["field"] == field for _, field in schema_fields if _ in tables)
     ]
+    value_hints = [
+        hint
+        for hint in resolved["value_hints"]
+        if any(hint["field"] == field for _, field in schema_fields if _ in tables)
+    ]
     relationships = [
         relation
         for relation in resolved["relationships"]
@@ -160,5 +173,6 @@ def build_semantic_context(question: str, schema: dict[str, Any]) -> dict[str, A
         "tables": tables,
         "fields": fields,
         "metrics": metrics,
+        "value_hints": value_hints,
         "relationships": relationships,
     }

nl2sql/sql_security.py

@@ -83,7 +83,16 @@ def validate_select_sql(
         if statement.find(exp.Star):
             raise SqlSecurityError("配置字段权限时禁止 SELECT *")
         default_table = next(iter(access_tables), None) if len(access_tables) == 1 else None
+        select_aliases = {
+            expression.alias
+            for expression in statement.expressions
+            if isinstance(expression, exp.Alias) and expression.alias
+        }
         for column in statement.find_all(exp.Column):
+            # ORDER BY may legally reference an alias defined in SELECT. The
+            # alias is already covered by the expressions validated below.
+            if not column.table and column.name in select_aliases and isinstance(column.parent, exp.Ordered):
+                continue
             table_name = aliases.get(column.table, column.table) or default_table
             if table_name is None or column.name not in authorized_columns.get(table_name, set()):
                 raise SqlSecurityError("SQL 访问了未授权字段")

schemas/advisor_agent.py

@@ -34,6 +34,7 @@ class AdvisorRebalanceRunReq(BaseModel):
 
 class AdvisorChatReq(BaseModel):
     # 通用投顾问答不需要客户上下文；个性化推荐时再传入客户编号。
+    scope: Literal["customer", "advisor"] = "customer"
     customer_id: int | None = Field(default=None, gt=0)
     intent: Literal[
         AGENT_INTENT_RECOMMEND,

scripts/check_nl2sql_consistency.py

@@ -15,6 +15,8 @@ from config.database.mysql import get_session_factory
 from config.settings import settings
 from nl2sql.metadata import build_metadata_chunks
 from nl2sql.milvus_collections import NL2SQL_COLLECTION
+from model.base import Base
+from scripts.sync_nl2sql_metadata import collect_orm_tables, merge_orm_metadata_rows
 
 
 async def collect_consistency() -> dict:
@@ -29,7 +31,21 @@ async def collect_consistency() -> dict:
         async with get_session_factory()() as session:
             tables = [dict(row) for row in (await session.execute(table_sql, {"db": settings.mysql.database})).mappings()]
             columns = [dict(row) for row in (await session.execute(column_sql, {"db": settings.mysql.database})).mappings()]
-        expected = len(build_metadata_chunks(tables, columns))
+        # 与同步脚本保持同一口径：ORM 表是唯一权威名单，数据库中存在但
+        # 未被 model/ 定义的表不应影响 NL2SQL 元数据一致性判断。
+        orm_tables = collect_orm_tables()
+        db_tables = {
+            str(row.get("TABLE_NAME") or row.get("table_name") or "").strip()
+            for row in tables
+        }
+        missing_tables = sorted(orm_tables - db_tables)
+        expected_tables, expected_columns = merge_orm_metadata_rows(
+            tables,
+            columns,
+            allowed_tables=orm_tables,
+            missing_table_objects=(Base.metadata.tables[name] for name in missing_tables),
+        )
+        expected = len(build_metadata_chunks(expected_tables, expected_columns))
         rows = await client().query(
             collection_name=NL2SQL_COLLECTION,
             filter="is_valid == true and is_deprecated == false",

scripts/check_nl2sql_e2e.py

@@ -25,6 +25,8 @@ from tool.llm import llm
 async def run_real_query(user_id: int, question: str, *, query_id: str | None = None) -> dict:
     """使用已有员工权限执行一条真实查询，只输出脱敏统计。"""
     from config.database.mysql import get_session_factory
+    from common.common_const import CUSTOMER_REL_STATUS_SIGNED, CUSTOMER_REL_STATUS_UNSIGNED
+    from repositories.customer_relation import CustomerRelationRepo
 
     query_id = query_id or uuid4().hex
     try:
@@ -33,6 +35,22 @@ async def run_real_query(user_id: int, question: str, *, query_id: str | None =
             if not permission.get("can_query"):
                 return {"query_status": "permission_denied", "user_id": user_id}
 
+            data_scope = None
+            if any(
+                scope.get("type") == "customer_ids"
+                for scope in (permission.get("row_scopes") or {}).values()
+            ):
+                relations = await CustomerRelationRepo(db).list_by_advisor(user_id)
+                customer_ids = [
+                    relation.customer_id
+                    for relation in relations
+                    if relation.status in {
+                        CUSTOMER_REL_STATUS_UNSIGNED,
+                        CUSTOMER_REL_STATUS_SIGNED,
+                    }
+                ]
+                data_scope = {"customer_ids": customer_ids}
+
             async def permission_loader(_user_id: int):
                 return permission
 
@@ -51,6 +69,7 @@ async def run_real_query(user_id: int, question: str, *, query_id: str | None =
                     question=question,
                     user_id=user_id,
                     trace_id=f"nl2sql-e2e-{query_id}",
+                    data_scope=data_scope,
                     include_sql=False,
                 ),
                 session=db,

service/nl2sql/query_service.py

@@ -100,12 +100,13 @@ async def query(
             for table, scope in (permission.get("row_scopes") or {}).items():
                 if scope.get("column"):
                     final_columns.setdefault(table, set()).add(scope["column"])
-        return validate_select_sql(
+        validated_option_sql = validate_select_sql(
             option_sql,
             authorized_tables=permission.get("tables", set()),
             authorized_columns=final_columns,
             max_rows=max_rows,
         )
+        return validated_option_sql
     except SqlSecurityError as exc:
         raise QueryServiceError("生成的 SQL 未通过安全校验") from exc
 
@@ -136,6 +137,11 @@ async def execute_query(
         conversation_context=conversation_context,
     )
     try:
+        parameters = (
+            {"advisor_id": request.user_id}
+            if ":advisor_id" in validated.sql
+            else None
+        )
         result = await execute_readonly_sql(
             session,
             validated,
@@ -144,6 +150,7 @@ async def execute_query(
             user_id=request.user_id,
             masks=masks,
             max_rows=request.max_rows,
+            parameters=parameters,
             timeout_seconds=timeout_seconds,
         )
     except QueryExecutionError as exc:

tool/llm.py

@@ -25,6 +25,7 @@ logger = logging.getLogger("tool.llm")
 FALLBACK_REPLY = "抱歉，服务暂时不可用，请稍后再试，或拨打客服热线 400-XXX-XXXX。"
 
 
+
 @dataclass(frozen=True)
 class Backend:
     """一个 OpenAI 兼容端点及其模型；name 为 ollama / api。"""
@@ -34,6 +35,7 @@ class Backend:
     chat_model: str
     embed_model: str
     api_key: str = ""
+    api_trust_env: bool = True
 
     @property
     def is_ollama(self) -> bool:
@@ -50,7 +52,8 @@ class Backend:
         """ollama 走本机端点，必须绕过系统代理：httpx 会读取 Windows 系统代理，
         但不识别其 ProxyOverride 白名单，localhost 请求会被转发到代理并返回 502。
         API 后端保留 trust_env，远端接口可能依赖代理。"""
-        return httpx.AsyncClient(timeout=timeout, trust_env=not self.is_ollama)
+
+        return httpx.AsyncClient(timeout=timeout, trust_env=self.api_trust_env if not self.is_ollama else False)
 
 
 def resolve_backends(cfg: LLMCfg) -> list[Backend]:
@@ -61,6 +64,7 @@ def resolve_backends(cfg: LLMCfg) -> list[Backend]:
             base_url=cfg.ollama_base.rstrip("/") + "/v1",
             chat_model=cfg.ollama_chat_model,
             embed_model=cfg.ollama_embed_model,
+            api_trust_env=False,
         )
         if cfg.ollama_base and cfg.ollama_chat_model
         else None
@@ -72,6 +76,7 @@ def resolve_backends(cfg: LLMCfg) -> list[Backend]:
             chat_model=cfg.api_chat_model,
             embed_model=cfg.api_embed_model,
             api_key=cfg.api_key,
+            api_trust_env=cfg.api_trust_env,
         )
         if cfg.api_key and cfg.api_base and cfg.api_chat_model
         else None
@@ -224,8 +229,8 @@ class LLMClient:
         ) or backend.embed_model
         if not embed_model:
             raise LLMFailError(f"backend={backend.name} 未配置 embed 模型")
-        is_dashscope_compatible = "/compatible-mode/" in backend.base_url.lower()
-        if is_dashscope_compatible:
+        is_dashscope_multimodal = "vision" in embed_model.lower() or "multimodal" in embed_model.lower()
+        if is_dashscope_multimodal:
             base_url = backend.base_url.rstrip("/")
             marker = "/compatible-mode/v1"
             if base_url.lower().endswith(marker):
@@ -242,11 +247,27 @@ class LLMClient:
                 "input": texts,
                 "dimensions": self.cfg.embed_dimensions,
             }
+        response = None
+        max_retries = max(1, int(getattr(self.cfg, "max_retries", 1)))
+        retry_backoff_sec = float(getattr(self.cfg, "retry_backoff_sec", 0))
         async with backend.client(self.cfg.timeout) as client:
-            r = await client.post(url, headers=backend.headers, json=payload)
-            r.raise_for_status()
-        data = r.json()
-        if is_dashscope_compatible:
+            for attempt in range(max_retries):
+                try:
+                    response = await client.post(
+                        url,
+                        headers=backend.headers,
+                        json=payload,
+                    )
+                    response.raise_for_status()
+                    break
+                except Exception:
+                    if attempt == max_retries - 1:
+                        raise
+                    await asyncio.sleep(retry_backoff_sec * (2**attempt))
+        if response is None:  # pragma: no cover - defensive guard
+            raise LLMFailError("Embedding 请求未返回响应")
+        data = response.json()
+        if is_dashscope_multimodal:
             embeddings = data["output"]["embeddings"]
             return [
                 item["embedding"]