diff --git a/.env.example b/.env.example index f45eced..3373ad8 100644 --- a/.env.example +++ b/.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 diff --git a/agent/advisor_agent/data_query.py b/agent/advisor_agent/data_query.py index 79699e1..d5d389c 100644 --- a/agent/advisor_agent/data_query.py +++ b/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 diff --git a/api/routers/advisor_agent.py b/api/routers/advisor_agent.py index 95b6bde..2c2e792 100644 --- a/api/routers/advisor_agent.py +++ b/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) diff --git a/api/routers/nl2sql.py b/api/routers/nl2sql.py index 000ad61..606ed50 100644 --- a/api/routers/nl2sql.py +++ b/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) diff --git a/config/database/redis.py b/config/database/redis.py index 3b1f4bd..e9a0f43 100644 --- a/config/database/redis.py +++ b/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, diff --git a/config/settings.py b/config/settings.py index 198b5b4..40f9fe0 100644 --- a/config/settings.py +++ b/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 diff --git a/nl2sql/executor.py b/nl2sql/executor.py index e571588..e12c685 100644 --- a/nl2sql/executor.py +++ b/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 diff --git a/nl2sql/row_scope.py b/nl2sql/row_scope.py index fd8ade3..fee7f47 100644 --- a/nl2sql/row_scope.py +++ b/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) diff --git a/nl2sql/semantics.py b/nl2sql/semantics.py index 9ca7193..bcf7b43 100644 --- a/nl2sql/semantics.py +++ b/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, } diff --git a/nl2sql/sql_security.py b/nl2sql/sql_security.py index 9c50390..12f40e5 100644 --- a/nl2sql/sql_security.py +++ b/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 访问了未授权字段") diff --git a/schemas/advisor_agent.py b/schemas/advisor_agent.py index 7744b2d..d9d34d1 100644 --- a/schemas/advisor_agent.py +++ b/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, diff --git a/scripts/check_nl2sql_consistency.py b/scripts/check_nl2sql_consistency.py index 877613d..86f6069 100644 --- a/scripts/check_nl2sql_consistency.py +++ b/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", diff --git a/scripts/check_nl2sql_e2e.py b/scripts/check_nl2sql_e2e.py index 75da4a2..cbcc62e 100644 --- a/scripts/check_nl2sql_e2e.py +++ b/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, diff --git a/service/nl2sql/query_service.py b/service/nl2sql/query_service.py index 6ee2d8a..2693dfe 100644 --- a/service/nl2sql/query_service.py +++ b/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: diff --git a/tool/llm.py b/tool/llm.py index d89cbf6..cd96e17 100644 --- a/tool/llm.py +++ b/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"]