diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..0a7faf0 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +*.py text eol=lf diff --git a/agent/advisor_agent/data_query.py b/agent/advisor_agent/data_query.py index d5d389c..e94e7d6 100644 --- a/agent/advisor_agent/data_query.py +++ b/agent/advisor_agent/data_query.py @@ -76,6 +76,7 @@ async def execute_advisor_data_query( question: str, trace_id: str, session_id: str | None = None, + conversation_context: str = "", data_scope: dict[str, Any] | None = None, max_rows: int | None = None, page: int = 1, @@ -162,6 +163,7 @@ async def execute_advisor_data_query( llm_client=llm_client, summary_llm=llm_client, masks=permission.get("masks"), + conversation_context=conversation_context, ) except (EmbeddingError, LLMFailError) as exc: raise QueryServiceError("投顾 Agent 依赖服务不可用，请检查 LLM/Embedding 服务连接") from exc diff --git a/agent/advisor_agent/intent/classifier.py b/agent/advisor_agent/intent/classifier.py index c8b0993..b4a6057 100644 --- a/agent/advisor_agent/intent/classifier.py +++ b/agent/advisor_agent/intent/classifier.py @@ -49,6 +49,7 @@ _CLASSIFIER_PROMPT = """你是基金投顾工作台的意图分类器，只负 """ _RECOMMENDATION_TERMS = ("推荐", "组合建议", "配置建议", "买什么", "适合配置", "筛选基金", "投资方案") +_SCOPE_ERROR_MESSAGE = "投顾范围查询仅支持客户数据查询" def _rule_fallback(query: str | None) -> IntentClassification: @@ -94,6 +95,13 @@ async def classify_advisor_intent( return IntentClassification(explicit_intent, 1.0, "explicit", "客户端显式指定") if not query or not query.strip(): return IntentClassification("", 0.0, "fallback", "空输入") + if re.sub(r"\s+", "", query) == _SCOPE_ERROR_MESSAGE: + return IntentClassification( + AGENT_INTENT_CASUAL_CHAT, + 1.0, + "rule", + "识别为系统提示文本而非业务查询", + ) if llm_client is not None: try: raw = await asyncio.wait_for( @@ -114,7 +122,7 @@ async def classify_advisor_intent( rule_intent = recognize_advisor_intent(query) if ( rule_intent == AGENT_INTENT_DATA_QUERY - and parsed.intent == AGENT_INTENT_RECOMMEND + and parsed.intent != AGENT_INTENT_DATA_QUERY and not any(term in (query or "") for term in _RECOMMENDATION_TERMS) ): return IntentClassification( diff --git a/agent/advisor_agent/intent/recognizer.py b/agent/advisor_agent/intent/recognizer.py index 9813958..99e0317 100644 --- a/agent/advisor_agent/intent/recognizer.py +++ b/agent/advisor_agent/intent/recognizer.py @@ -40,6 +40,8 @@ _REBALANCE_TERMS = ("调仓", "再平衡", "组合调整", "配置偏离", "偏 _FUND_ANALYSIS_TERMS = ("基金分析", "分析基金", "基金表现", "净值走势", "最大回撤", "夏普比率", "年化波动") _DIALOGUE_TERMS = ("话术", "沟通", "怎么跟客户说", "如何向客户解释", "安抚客户", "投诉处理") _RECOMMEND_TERMS = ("推荐", "组合建议", "配置建议", "买什么", "适合配置", "筛选基金", "投资方案") +_CUSTOMER_IDENTITY_TERMS = ("是谁", "姓名", "实名", "基本信息", "联系方式", "手机号") +_SCOPE_ERROR_MESSAGE = "投顾范围查询仅支持客户数据查询" def _contains_any(text: str, terms: tuple[str, ...]) -> bool: @@ -57,6 +59,8 @@ def recognize_advisor_intent(query: str | None, explicit_intent: str | None = No text = re.sub(r"\s+", "", query or "") if not text: return None + if text == _SCOPE_ERROR_MESSAGE: + return None # 先处理最明确的任务词，避免“查询基金收益”被误判成普通数据查询。 if _contains_any(text, _REBALANCE_TERMS): @@ -68,6 +72,13 @@ def recognize_advisor_intent(query: str | None, explicit_intent: str | None = No has_action = any(term in text for term in _QUERY_ACTIONS) has_data = any(term in text for term in _DATA_TERMS) + has_customer_identity = ( + "客户" in text + and _contains_any(text, _CUSTOMER_IDENTITY_TERMS) + and not _contains_any(text, _RECOMMEND_TERMS) + ) + if has_customer_identity: + return AGENT_INTENT_DATA_QUERY if has_data and (has_action or "客户" in text or "近一年" in text or "本月" in text): return AGENT_INTENT_DATA_QUERY if _contains_any(text, _RECOMMEND_TERMS): diff --git a/nl2sql/row_scope.py b/nl2sql/row_scope.py index fee7f47..cd04aaf 100644 --- a/nl2sql/row_scope.py +++ b/nl2sql/row_scope.py @@ -29,7 +29,7 @@ def apply_row_scope(sql: str, permission: dict, data_scope: dict | None) -> str: statement = parse_one(sql, read="mysql") except ParseError as exc: raise RowScopeError("SQL 解析失败") from exc - for table in statement.find_all(exp.Table): + for table in list(statement.find_all(exp.Table)): scope = scopes.get(table.name) if not scope: continue @@ -38,5 +38,6 @@ def apply_row_scope(sql: str, permission: dict, data_scope: dict | None) -> str: this=exp.column(scope["column"], table=table.alias_or_name), expressions=[exp.Literal.number(value) for value in values], ) - statement = statement.where(condition) + owner = table.find_ancestor(exp.Select) or statement + owner.where(condition, copy=False) return statement.sql(dialect="mysql") diff --git a/service/nl2sql/query_service.py b/service/nl2sql/query_service.py index 2693dfe..955f1b7 100644 --- a/service/nl2sql/query_service.py +++ b/service/nl2sql/query_service.py @@ -10,6 +10,7 @@ from nl2sql.executor import QueryExecutionError, execute_readonly_sql from nl2sql.row_scope import RowScopeError, apply_row_scope from nl2sql.result import build_chart_config, summarize_result from nl2sql.query_experience import apply_query_options, build_query_explanation +from nl2sql.query_rewriter import rewrite_query from nl2sql.supervisor import UnsupportedIntent, ensure_query_intent from nl2sql.sql_agent import generate_sql from nl2sql.sql_security import SqlSecurityError, validate_select_sql @@ -19,6 +20,29 @@ class QueryServiceError(RuntimeError): """查询编排失败或当前用户没有查询权限。""" +def _filter_schema_columns(schema: dict[str, Any], permission: dict[str, Any]) -> dict[str, Any]: + """只把当前用户有列权限的字段暴露给 SQL 生成模型。""" + authorized_columns = permission.get("columns") + if authorized_columns is None: + return schema + + filtered = dict(schema) + filtered["columns"] = [ + column + for column in schema.get("columns", []) + if column.get("field_name") + in set(authorized_columns.get(column.get("table_name"), set())) + ] + return filtered + + +def _normalize_advisor_placeholder(sql: str) -> str: + """统一 LLM 可能生成的投顾参数占位符，供 SQLAlchemy 命名绑定。""" + if "advisor_id" not in sql or "?" not in sql: + return sql + return sql.replace("?", ":advisor_id", 1) + + async def query( request: DataQueryRequest, *, @@ -30,8 +54,13 @@ async def query( conversation_context: str = "", ): """执行权限、召回、权威 Schema、生成和安全校验，返回可执行 SQL。""" + effective_question = await rewrite_query( + request.question, + conversation_context, + llm_client=llm_client, + ) try: - ensure_query_intent(request.question) + ensure_query_intent(effective_question) except UnsupportedIntent as exc: raise QueryServiceError(str(exc)) from exc permission = await permission_loader(request.user_id) @@ -40,7 +69,7 @@ async def query( if metadata_retriever is None or schema_loader is None: raise QueryServiceError("NL2SQL 查询依赖未配置") - hits = await metadata_retriever(request.question) + hits = await metadata_retriever(effective_question) candidate_tables = { hit.get("table_name") for hit in hits @@ -48,23 +77,27 @@ async def query( } if not candidate_tables: raise QueryServiceError("未找到有权限的业务表") - schema = await schema_loader(candidate_tables, permission) + schema = _filter_schema_columns( + await schema_loader(candidate_tables, permission), + permission, + ) if not schema.get("tables"): raise QueryServiceError("候选表未通过权威 Schema 校验") few_shot = [] if few_shot_retriever is not None: try: - few_shot = await few_shot_retriever(request.question) + few_shot = await few_shot_retriever(effective_question) except Exception: # noqa: BLE001 Few-shot 故障不阻断主查询 few_shot = [] generated = await generate_sql( - request.question, + effective_question, schema, llm_client=llm_client, few_shot=few_shot, conversation_context=conversation_context, ) + generated = replace(generated, sql=_normalize_advisor_placeholder(generated.sql)) max_rows = request.max_rows or permission.get("max_rows") or 1000 try: validated = validate_select_sql(