From 058f45115d0734751764d1903b8a0ca4267ac787 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E6=96=8C=E5=BC=BA?= Date: Mon, 14 Sep 2026 21:41:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E4=BF=AE=E5=A4=8D=E6=8A=95=E9=A1=BEagent?= =?UTF-8?q?=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/advisor_agent/data_query.py | 18 ++++++++ agent/advisor_agent/intent/classifier.py | 11 +++++ api/routers/advisor_agent.py | 2 + api/routers/nl2sql.py | 1 + frontend/lib/assistant-api.ts | 5 +-- main.py | 1 + nl2sql/cache.py | 26 ++++++++++++ nl2sql/query_rewriter.py | 7 +++- nl2sql/result.py | 1 + nl2sql/retrieval.py | 13 ++++-- nl2sql/schema.py | 33 ++++++++++++++- nl2sql/sql_agent.py | 2 +- service/client_agent/runtime.py | 2 +- tool/llm.py | 53 ++++++++++++++---------- 14 files changed, 142 insertions(+), 33 deletions(-) diff --git a/agent/advisor_agent/data_query.py b/agent/advisor_agent/data_query.py index f359831..a8c2f26 100644 --- a/agent/advisor_agent/data_query.py +++ b/agent/advisor_agent/data_query.py @@ -13,6 +13,7 @@ from config.settings import settings from nl2sql.contracts import DataQueryRequest, DataQueryResult from nl2sql.embedding import EmbeddingError from nl2sql.retrieval import retrieve_metadata +from nl2sql.cache import cache_get, cache_set, build_question_cache_key from nl2sql.runtime_config import runtime_config from nl2sql.schema import load_authoritative_schema from repositories.customer_relation import CustomerRelationRepo @@ -154,6 +155,7 @@ async def execute_advisor_data_query( pass if scope == "customer" and _is_current_holdings_query(question): + redis = redis or database.redis.client() return await _query_current_holdings( db, customer_id=customer_id, @@ -166,6 +168,21 @@ async def execute_advisor_data_query( milvus = milvus or database.milvus.client() redis = redis or database.redis.client() + cache_key = build_question_cache_key( + question, + permission=permission, + data_scope={"customer_ids": customer_ids}, + page=page, + page_size=page_size, + sort_by=sort_by, + sort_order=sort_order, + ) + cached = await cache_get(redis, cache_key) + if cached is not None: + cached["trace_id"] = trace_id + cached["customer_id"] = customer_id + cached["warnings"] = [*cached.get("warnings", []), "cache_hit"] + return cached llm_client = llm_client or default_llm request = DataQueryRequest( question=question, @@ -222,4 +239,5 @@ async def execute_advisor_data_query( payload["answer"] = f"{existing_answer.rstrip('。')}。客户姓名：{enriched['name_summary']}。" payload["sql"] = None payload["customer_id"] = customer_id + await cache_set(redis, cache_key, payload, ttl=runtime_config.cache_ttl) return payload diff --git a/agent/advisor_agent/intent/classifier.py b/agent/advisor_agent/intent/classifier.py index 8f94778..91dce68 100644 --- a/agent/advisor_agent/intent/classifier.py +++ b/agent/advisor_agent/intent/classifier.py @@ -108,6 +108,17 @@ async def classify_advisor_intent( "rule", "识别为系统提示文本而非业务查询", ) + fast_intent = recognize_advisor_intent(query) + if ( + fast_intent == AGENT_INTENT_DATA_QUERY + and not any(term in query for term in _RECOMMENDATION_TERMS) + ): + return IntentClassification( + fast_intent, + 0.95, + "rule_fast", + "明显查询类问题", + ) if llm_client is not None: try: raw = await asyncio.wait_for( diff --git a/api/routers/advisor_agent.py b/api/routers/advisor_agent.py index bd84d06..e2c1a6c 100644 --- a/api/routers/advisor_agent.py +++ b/api/routers/advisor_agent.py @@ -331,6 +331,7 @@ async def chat_stream( session_id=chat_request.session_id, conversation_context=conversation_context, llm_client=getattr(_advisor_runtime(request), "llm_client", None), + redis=redis_db.client(), ) except QueryServiceError as exc: payload = agent_failure(ERR_CODE_LLM_ERROR, _data_query_error_message(exc), trace_id=trace_id) @@ -565,6 +566,7 @@ async def advisor_data_query( page_size=body.page_size, sort_by=body.sort_by, sort_order=body.sort_order, + redis=redis_db.client(), ) except QueryServiceError as exc: return agent_failure( diff --git a/api/routers/nl2sql.py b/api/routers/nl2sql.py index 606ed50..803ca17 100644 --- a/api/routers/nl2sql.py +++ b/api/routers/nl2sql.py @@ -248,6 +248,7 @@ async def query_data( db, database=settings.mysql.database, candidate_tables=table_names, + redis=redis, ) request_contract = DataQueryRequest( diff --git a/frontend/lib/assistant-api.ts b/frontend/lib/assistant-api.ts index 4ef8ab1..691be21 100644 --- a/frontend/lib/assistant-api.ts +++ b/frontend/lib/assistant-api.ts @@ -72,10 +72,7 @@ export const assistantApi = { sessionHistory: (sessionId: string) => apiFetch(`/advisor-agent/session/${encodeURIComponent(sessionId)}/history`), - /** - * POST /chat/stream —— 响应是 text/event-stream，但后端整包输出， - * 因此读完整条响应后一次性返回聚合结果（非流式渲染）。 - */ + /** POST /chat/stream —— 读取 SSE，并支持调用方实时消费事件。 */ chatOnce( body: { query: string; diff --git a/main.py b/main.py index 738f56c..db40a60 100644 --- a/main.py +++ b/main.py @@ -90,6 +90,7 @@ async def lifespan(app: FastAPI): await app.state.advisor_event_consumer.stop() if app.state.advisor_scheduler is not None: app.state.advisor_scheduler.shutdown() + await llm_client.aclose() await database.dispose() diff --git a/nl2sql/cache.py b/nl2sql/cache.py index 325ff2b..10b76f5 100644 --- a/nl2sql/cache.py +++ b/nl2sql/cache.py @@ -49,6 +49,32 @@ def build_cache_key( return f"nl2sql:{digest}" +def build_question_cache_key( + question: str, + *, + permission: dict[str, Any], + data_scope: dict[str, Any] | None = None, + page: int = 1, + page_size: int = 100, + sort_by: str | None = None, + sort_order: str = "asc", +) -> str: + """生成自然语言查询结果缓存键，覆盖问题、权限和数据范围。""" + payload = { + "question": " ".join((question or "").split()), + "permission": _permission_payload(permission), + "data_scope": data_scope or {}, + "page": page, + "page_size": page_size, + "sort_by": sort_by or "", + "sort_order": sort_order, + } + digest = hashlib.sha256( + json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + return f"nl2sql:question:{digest}" + + async def cache_get(redis, key: str) -> dict[str, Any] | None: """读取缓存，Redis 异常或内容损坏时返回空结果。""" try: diff --git a/nl2sql/query_rewriter.py b/nl2sql/query_rewriter.py index de031f8..f0906bf 100644 --- a/nl2sql/query_rewriter.py +++ b/nl2sql/query_rewriter.py @@ -16,6 +16,11 @@ _SYSTEM_PROMPT = """你是基金平台 NL2SQL 的问题改写器。 不能扩大权限范围，不能臆造查询结果；上下文不足时保留原问题。 """ +_REFERENCE_TERMS = ( + "他", "她", "它", "他们", "她们", "它们", "这个客户", "该客户", "那个客户", + "这个基金", "该基金", "那只基金", "这只基金", "这个结果", "上一轮", "刚才", "前面", +) + def _rewrite_explicit_customer_identity(question: str) -> str | None: normalized = re.sub(r"\s+", "", question) @@ -68,7 +73,7 @@ async def rewrite_query( contextual_identity = _rewrite_customer_name_follow_up(original, context) if contextual_identity: return contextual_identity - if not context or llm_client is None: + if not context or llm_client is None or not any(term in original for term in _REFERENCE_TERMS): return original prompt = ( diff --git a/nl2sql/result.py b/nl2sql/result.py index dbb1fcd..480fe25 100644 --- a/nl2sql/result.py +++ b/nl2sql/result.py @@ -67,6 +67,7 @@ async def render_answer( {"role": "user", "content": prompt}, ], temperature=0, + max_tokens=256, ) except Exception: # noqa: BLE001 摘要失败回退固定文本 return fallback diff --git a/nl2sql/retrieval.py b/nl2sql/retrieval.py index 3d81a1b..76bbfc0 100644 --- a/nl2sql/retrieval.py +++ b/nl2sql/retrieval.py @@ -1,6 +1,7 @@ """NL2SQL 元数据召回与授权过滤。""" from __future__ import annotations +import asyncio from collections.abc import Awaitable, Callable, Iterable from typing import Any @@ -70,15 +71,21 @@ async def retrieve_metadata( "is_valid", "is_deprecated", ] - results: list[dict[str, Any]] = [] - for chunk_type in ("table_meta", "field_meta"): - hits = await milvus_client.search( + async def search_chunk(chunk_type: str): + return await milvus_client.search( collection_name=NL2SQL_COLLECTION, data=[vector], limit=top_k, filter=f'chunk_type == "{chunk_type}" and is_valid == true', output_fields=output_fields, ) + + table_hits, field_hits = await asyncio.gather( + search_chunk("table_meta"), + search_chunk("field_meta"), + ) + results: list[dict[str, Any]] = [] + for hits in (table_hits, field_hits): for hit in _flatten_hits(hits): entity = hit.get("entity") or hit if not entity.get("is_valid", True) or entity.get("is_deprecated", False): diff --git a/nl2sql/schema.py b/nl2sql/schema.py index 772b43f..6c781ef 100644 --- a/nl2sql/schema.py +++ b/nl2sql/schema.py @@ -1,12 +1,23 @@ """从 MySQL information_schema 加载并校验 NL2SQL 权威 Schema。""" from __future__ import annotations +import hashlib +import json +import logging from typing import Any from sqlalchemy import bindparam, text from nl2sql.metadata import normalize_column_row, normalize_table_row +logger = logging.getLogger("nl2sql.schema") +SCHEMA_CACHE_TTL = 60 + + +def _schema_cache_key(database: str, candidates: list[str]) -> str: + payload = json.dumps([database, candidates], ensure_ascii=False, separators=(",", ":")) + return "nl2sql:schema:" + hashlib.sha256(payload.encode()).hexdigest() + TABLES_SQL = text( """ @@ -34,13 +45,25 @@ async def load_authoritative_schema( *, database: str, candidate_tables: set[str] | list[str], + redis=None, + cache_ttl: int = SCHEMA_CACHE_TTL, ) -> dict[str, list[dict[str, Any]]]: """从权威元数据源加载候选表，并剔除不存在的表和孤立字段。""" candidates = {str(name).strip() for name in candidate_tables if str(name).strip()} if not candidates: return {"tables": [], "columns": []} - params = {"database": database, "candidate_tables": sorted(candidates)} + candidate_list = sorted(candidates) + cache_key = _schema_cache_key(database, candidate_list) + if redis is not None: + try: + cached = await redis.get(cache_key) + if cached: + return json.loads(cached) + except Exception: # noqa: BLE001 缓存异常回退权威查询 + logger.warning("Schema 缓存读取失败", exc_info=True) + + params = {"database": database, "candidate_tables": candidate_list} table_result = await session.execute(TABLES_SQL, params) column_result = await session.execute(COLUMNS_SQL, params) @@ -56,4 +79,10 @@ async def load_authoritative_schema( normalized = normalize_column_row(dict(row)) if normalized["table_name"] in valid_table_names and normalized["field_name"]: columns.append(normalized) - return {"tables": tables, "columns": columns} + schema = {"tables": tables, "columns": columns} + if redis is not None: + try: + await redis.set(cache_key, json.dumps(schema, ensure_ascii=False), ex=cache_ttl) + except Exception: # noqa: BLE001 缓存异常不阻断查询 + logger.warning("Schema 缓存写入失败", exc_info=True) + return schema diff --git a/nl2sql/sql_agent.py b/nl2sql/sql_agent.py index e3a544f..aac11d6 100644 --- a/nl2sql/sql_agent.py +++ b/nl2sql/sql_agent.py @@ -74,7 +74,7 @@ async def generate_sql( }, ] try: - output = await llm_client.chat(messages, temperature=0) + output = await llm_client.chat(messages, temperature=0, max_tokens=256) except Exception as exc: # noqa: BLE001 统一收敛模型异常 raise SqlGenerationError("SQL 模型调用失败") from exc sql = _clean_model_output(output) diff --git a/service/client_agent/runtime.py b/service/client_agent/runtime.py index 0d14f54..caa7e09 100644 --- a/service/client_agent/runtime.py +++ b/service/client_agent/runtime.py @@ -304,7 +304,7 @@ def _build_data_query(*, db_session_factory, milvus_client, llm_client, config_g async def schema_loader(table_names: set[str], _permission: dict): return await load_authoritative_schema( - db, database=schema_database, candidate_tables=table_names + db, database=schema_database, candidate_tables=table_names, redis=redis ) request = DataQueryRequest( diff --git a/tool/llm.py b/tool/llm.py index cd96e17..ec811e6 100644 --- a/tool/llm.py +++ b/tool/llm.py @@ -103,6 +103,21 @@ class LLMClient: def __init__(self, cfg: LLMCfg | None = None): self.cfg = cfg or settings.llm self.backends = resolve_backends(self.cfg) + self._clients: dict[str, httpx.AsyncClient] = {} + + def _client_for(self, backend: Backend) -> httpx.AsyncClient: + clients = getattr(self, "_clients", None) + if clients is None: + clients = {} + self._clients = clients + if backend.name not in clients: + clients[backend.name] = backend.client(self.cfg.timeout) + return clients[backend.name] + + async def aclose(self) -> None: + for client in getattr(self, "_clients", {}).values(): + await client.aclose() + getattr(self, "_clients", {}).clear() # ---- 首选后端（健康检查/日志/embed 使用） ----------------------------- @property @@ -186,9 +201,9 @@ class LLMClient: "stream": False, } try: - async with backend.client(self.cfg.timeout) as client: - r = await client.post(url, headers=backend.headers, json=payload) - r.raise_for_status() + client = self._client_for(backend) + r = await client.post(url, headers=backend.headers, json=payload) + r.raise_for_status() choice = r.json()["choices"][0] message = choice["message"] content = message.get("content") @@ -250,20 +265,16 @@ class LLMClient: 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: - 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)) + client = self._client_for(backend) + 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() @@ -286,10 +297,10 @@ class LLMClient: url = backend.base_url.removesuffix("/v1") + "/api/tags" else: url = f"{backend.base_url}/models" - async with backend.client(min(self.cfg.timeout, 10)) as client: - r = await client.get(url, headers=backend.headers) - if r.status_code >= 500: - r.raise_for_status() + client = self._client_for(backend) + r = await client.get(url, headers=backend.headers) + if r.status_code >= 500: + r.raise_for_status() # 全局单例：Agent 统一引入 -- 2.54.0