diff --git a/agent/advisor_agent/data_query.py b/agent/advisor_agent/data_query.py index b70e5a2..79699e1 100644 --- a/agent/advisor_agent/data_query.py +++ b/agent/advisor_agent/data_query.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import asdict from typing import Any +from uuid import uuid4 from agent.advisor_agent.auth import ensure_customer_access from agent.data_query.agent import DataQueryAgent @@ -15,6 +16,51 @@ from nl2sql.schema import load_authoritative_schema 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 repositories.fin_holdings import FinHoldingsRepo +from repositories.fin_product import FinProductRepo + + +def _is_current_holdings_query(question: str) -> bool: + text = "".join((question or "").split()) + return any(term in text for term in ("当前持仓", "目前持仓", "现有持仓", "持仓明细", "持仓情况")) + + +async def _query_current_holdings(db, *, customer_id: int, trace_id: str) -> dict[str, Any]: + holdings = await FinHoldingsRepo(db).list_by_customer(customer_id, status="持有中") + product_repo = FinProductRepo(db) + rows: list[dict[str, Any]] = [] + total_value = 0 + for holding in holdings: + product = await product_repo.get(holding.product_id) + rows.append( + { + "产品代码": product.product_code if product else None, + "产品名称": product.product_name if product else None, + "风险等级": product.risk_level if product else None, + "持有份额": f"{holding.shares:.4f}", + "成本金额": f"{holding.cost_amount:.2f}", + "当前市值": f"{holding.current_value:.2f}", + "盈亏": f"{holding.profit_loss:.2f}", + "收益率": f"{holding.profit_ratio:.4f}", + "状态": holding.status, + } + ) + total_value += holding.current_value + names = [row["产品名称"] for row in rows if row["产品名称"]] + return { + "query_id": f"holdings-{uuid4().hex}", + "trace_id": trace_id, + "columns": list(rows[0].keys()) if rows else ["产品代码", "产品名称", "风险等级", "持有份额", "成本金额", "当前市值", "盈亏", "收益率", "状态"], + "rows": rows, + "row_count": len(rows), + "truncated": False, + "summary": f"当前持仓共 {len(rows)} 条记录。", + "answer": ( + f"当前持有 {len(rows)} 只基金，总市值约 {total_value:.2f} 元。" + + (f"包括：{'、'.join(names[:6])}。" if names else "") + ), + "sql": None, + } async def execute_advisor_data_query( @@ -42,6 +88,12 @@ async def execute_advisor_data_query( ``customer_id``，避免投顾借助 NL2SQL 查询其他客户数据。 """ await ensure_customer_access(db, advisor_id=advisor_id, customer_id=customer_id) + if _is_current_holdings_query(question): + return await _query_current_holdings( + db, + customer_id=customer_id, + trace_id=trace_id, + ) permission = await load_query_permission(db, advisor_id) if not permission.get("can_query", False): raise QueryServiceError("当前投顾没有 NL2SQL 查询权限") diff --git a/agent/advisor_agent/intent/classifier.py b/agent/advisor_agent/intent/classifier.py new file mode 100644 index 0000000..c8b0993 --- /dev/null +++ b/agent/advisor_agent/intent/classifier.py @@ -0,0 +1,132 @@ +"""投顾聊天意图分类：LLM 优先，规则识别兜底。""" +from __future__ import annotations + +import asyncio +import json +import re +from dataclasses import dataclass + +from common.common_const import ( + AGENT_INTENT_CASUAL_CHAT, + AGENT_INTENT_DATA_QUERY, + AGENT_INTENT_DIALOGUE_SCRIPT, + AGENT_INTENT_FUND_ANALYSIS, + AGENT_INTENT_REBALANCE, + AGENT_INTENT_RECOMMEND, +) +from agent.advisor_agent.intent.recognizer import recognize_advisor_intent + + +VALID_INTENTS = frozenset({ + AGENT_INTENT_RECOMMEND, + AGENT_INTENT_REBALANCE, + AGENT_INTENT_FUND_ANALYSIS, + AGENT_INTENT_DIALOGUE_SCRIPT, + AGENT_INTENT_DATA_QUERY, + AGENT_INTENT_CASUAL_CHAT, +}) + + +@dataclass(frozen=True) +class IntentClassification: + intent: str + confidence: float + source: str + reason: str = "" + + +_CLASSIFIER_PROMPT = """你是基金投顾工作台的意图分类器，只负责分类，不回答用户问题。 +只能从以下分类中选择一个： +- recommend：基金推荐、组合配置或投资方案 +- rebalance：组合偏离、调仓、再平衡、仓位调整 +- fund_analysis：单只基金分析、净值、收益、回撤、波动率、夏普比率 +- dialogue-script：给客户准备沟通话术、解释、安抚、投诉或风险提醒 +- data_query：查询客户持仓、资产、余额、收益、交易、账户明细 +- casual_chat：问候、闲聊、感谢、身份询问或无法归入业务分类的内容 + +只输出 JSON，不要 Markdown，不要额外文字： +{"intent":"分类值","confidence":0到1之间的数字,"reason":"不超过30字的原因"} +""" + +_RECOMMENDATION_TERMS = ("推荐", "组合建议", "配置建议", "买什么", "适合配置", "筛选基金", "投资方案") + + +def _rule_fallback(query: str | None) -> IntentClassification: + intent = recognize_advisor_intent(query) + if intent: + return IntentClassification(intent=intent, confidence=0.72, source="rule", reason="关键词规则匹配") + return IntentClassification(intent=AGENT_INTENT_CASUAL_CHAT, confidence=0.0, source="fallback", reason="无法匹配业务意图") + + +def _parse_model_result(raw: str) -> IntentClassification | None: + text = raw.strip() + fenced = re.search(r"\{.*\}", text, re.DOTALL) + if fenced: + text = fenced.group(0) + try: + payload = json.loads(text) + except (TypeError, json.JSONDecodeError): + return None + intent = payload.get("intent") + if intent not in VALID_INTENTS: + return None + try: + confidence = max(0.0, min(1.0, float(payload.get("confidence", 0.0)))) + except (TypeError, ValueError): + confidence = 0.0 + return IntentClassification( + intent=intent, + confidence=confidence, + source="llm", + reason=str(payload.get("reason") or "模型分类"), + ) + + +async def classify_advisor_intent( + query: str | None, + llm_client=None, + *, + explicit_intent: str | None = None, + timeout: float = 2.0, +) -> IntentClassification: + """分类用户意图；显式意图兼容旧客户端，模型失败时安全回退规则。""" + if explicit_intent in VALID_INTENTS: + return IntentClassification(explicit_intent, 1.0, "explicit", "客户端显式指定") + if not query or not query.strip(): + return IntentClassification("", 0.0, "fallback", "空输入") + if llm_client is not None: + try: + raw = await asyncio.wait_for( + llm_client.chat( + [ + {"role": "system", "content": _CLASSIFIER_PROMPT}, + {"role": "user", "content": query.strip()}, + ], + temperature=0, + max_tokens=120, + ), + timeout=timeout, + ) + parsed = _parse_model_result(raw) + if parsed is not None: + # LLM 偶尔会把“查询持仓/资产”等只读请求误判为推荐； + # 对明确的查询动作以规则结果为准，避免误进入草稿生成分支。 + rule_intent = recognize_advisor_intent(query) + if ( + rule_intent == AGENT_INTENT_DATA_QUERY + and parsed.intent == AGENT_INTENT_RECOMMEND + and not any(term in (query or "") for term in _RECOMMENDATION_TERMS) + ): + return IntentClassification( + intent=rule_intent, + confidence=max(parsed.confidence, 0.9), + source="rule_override", + reason="明确查询类关键词覆盖模型误判", + ) + return parsed + except Exception: + pass + return _rule_fallback(query) + + +__all__ = ["IntentClassification", "classify_advisor_intent"] diff --git a/agent/advisor_agent/intent/recognizer.py b/agent/advisor_agent/intent/recognizer.py index b6d4907..9813958 100644 --- a/agent/advisor_agent/intent/recognizer.py +++ b/agent/advisor_agent/intent/recognizer.py @@ -3,7 +3,13 @@ from __future__ import annotations import re -from common.common_const import AGENT_INTENT_DATA_QUERY +from common.common_const import ( + AGENT_INTENT_DATA_QUERY, + AGENT_INTENT_DIALOGUE_SCRIPT, + AGENT_INTENT_FUND_ANALYSIS, + AGENT_INTENT_REBALANCE, + AGENT_INTENT_RECOMMEND, +) _QUERY_ACTIONS = ( @@ -30,7 +36,14 @@ _DATA_TERMS = ( "客户数据", "账户", ) -_NON_QUERY_INTENTS = ("推荐", "调仓", "再平衡", "话术", "沟通") +_REBALANCE_TERMS = ("调仓", "再平衡", "组合调整", "配置偏离", "偏离目标", "降低仓位", "增加仓位") +_FUND_ANALYSIS_TERMS = ("基金分析", "分析基金", "基金表现", "净值走势", "最大回撤", "夏普比率", "年化波动") +_DIALOGUE_TERMS = ("话术", "沟通", "怎么跟客户说", "如何向客户解释", "安抚客户", "投诉处理") +_RECOMMEND_TERMS = ("推荐", "组合建议", "配置建议", "买什么", "适合配置", "筛选基金", "投资方案") + + +def _contains_any(text: str, terms: tuple[str, ...]) -> bool: + return any(term in text for term in terms) def recognize_advisor_intent(query: str | None, explicit_intent: str | None = None) -> str | None: @@ -42,12 +55,23 @@ def recognize_advisor_intent(query: str | None, explicit_intent: str | None = No if explicit_intent: return explicit_intent text = re.sub(r"\s+", "", query or "") - if not text or any(term in text for term in _NON_QUERY_INTENTS): + if not text: return None + + # 先处理最明确的任务词，避免“查询基金收益”被误判成普通数据查询。 + if _contains_any(text, _REBALANCE_TERMS): + return AGENT_INTENT_REBALANCE + if _contains_any(text, _FUND_ANALYSIS_TERMS): + return AGENT_INTENT_FUND_ANALYSIS + if _contains_any(text, _DIALOGUE_TERMS): + return AGENT_INTENT_DIALOGUE_SCRIPT + has_action = any(term in text for term in _QUERY_ACTIONS) has_data = any(term in text for term in _DATA_TERMS) 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): + return AGENT_INTENT_RECOMMEND return None diff --git a/api/routers/advisor_agent.py b/api/routers/advisor_agent.py index cbb1a12..95b6bde 100644 --- a/api/routers/advisor_agent.py +++ b/api/routers/advisor_agent.py @@ -7,12 +7,13 @@ from typing import Literal from fastapi import APIRouter, BackgroundTasks, Depends, Query, Request from fastapi.responses import StreamingResponse +from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession from agent.advisor_agent.auth import ensure_customer_access from agent.advisor_agent.data_query import execute_advisor_data_query from agent.advisor_agent.intent.fund_analysis import build_fund_analysis -from agent.advisor_agent.intent.recognizer import recognize_advisor_intent +from agent.advisor_agent.intent.classifier import classify_advisor_intent from agent.advisor_agent.intent.talk_script import build_talk_script from agent.advisor_agent.llm import generate_text from agent.advisor_agent.intent.generation_flow import ( @@ -21,7 +22,11 @@ from agent.advisor_agent.intent.generation_flow import ( ) from agent.advisor_agent.protocol import agent_failure, agent_success from common.common_const import ( + AGENT_INTENT_CASUAL_CHAT, AGENT_INTENT_DATA_QUERY, + AGENT_INTENT_DIALOGUE_SCRIPT, + AGENT_INTENT_FUND_ANALYSIS, + AGENT_INTENT_REBALANCE, AGENT_INTENT_RECOMMEND, CUSTOMER_REL_STATUS_SIGNED, ERR_CODE_DRAFT_NOT_FOUND, @@ -34,6 +39,10 @@ from common.common_const import ( SSE_EVENT_TYPE_ERROR, SSE_EVENT_TYPE_META, SSE_EVENT_TYPE_TEXT, + TALK_SCENE_CUSTOMER_COMPLAINT, + TALK_SCENE_MARKET_FLUCTUATION, + TALK_SCENE_PORTFOLIO_DIVERGENCE, + TALK_SCENE_RISK_BLOCK_ORDER, ) from api.deps import audited_advisor from config.deps import get_db @@ -196,9 +205,51 @@ async def chat_stream( db: AsyncSession = Depends(get_db), ): trace_id = _trace_id(request) - chat_request = body + try: + chat_request = ( + body + if isinstance(body, AdvisorChatReq) + else AdvisorChatReq.model_validate(body) + ) + except ValidationError: + payload = agent_failure( + ERR_CODE_FORBIDDEN_CUSTOMER, + "对话请求缺少有效参数", + trace_id=trace_id, + ) + + async def validation_error_events(): + yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_ERROR, **payload}, ensure_ascii=False)}\n\n" + + return StreamingResponse( + validation_error_events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + + runtime = _advisor_runtime(request) + classification = await classify_advisor_intent( + chat_request.query, + getattr(runtime, "llm_client", None), + explicit_intent=chat_request.intent, + ) + inferred_intent = classification.intent customer_id = chat_request.customer_id - inferred_intent = chat_request.intent or _infer_chat_intent(chat_request.query) + if not chat_request.query and not chat_request.intent: + code = ERR_CODE_LLM_ERROR if customer_id is not None else ERR_CODE_FORBIDDEN_CUSTOMER + message = _NOT_READY_MESSAGE if customer_id is not None else "对话请求缺少有效参数" + payload = agent_failure(code, message, trace_id=trace_id) + + async def empty_query_events(): + yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_ERROR, **payload}, ensure_ascii=False)}\n\n" + + return StreamingResponse( + empty_query_events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + + payload = None # 请求体只传问题时，从问题中解析客户；解析结果仍必须经过投顾关系授权校验。 if customer_id is None: @@ -228,9 +279,10 @@ async def chat_stream( if customer_id is None: if inferred_intent in { AGENT_INTENT_RECOMMEND, - "rebalance", - "fund_analysis", - "dialogue-script", + AGENT_INTENT_REBALANCE, + AGENT_INTENT_FUND_ANALYSIS, + AGENT_INTENT_DIALOGUE_SCRIPT, + AGENT_INTENT_DATA_QUERY, }: if payload is None: payload = agent_failure( @@ -239,9 +291,10 @@ async def chat_stream( trace_id=trace_id, ) else: - runtime = _advisor_runtime(request) llm_client = getattr(runtime, "llm_client", None) - if llm_client is None: + if inferred_intent == AGENT_INTENT_CASUAL_CHAT: + answer = "您好，我是投顾助手，请选择客户后使用个性化分析。" + elif llm_client is None: answer = "已收到问题。当前未配置通用投顾模型，请选择客户后使用个性化分析，或联系管理员配置 Agent 服务。" else: answer = await generate_text( @@ -254,7 +307,7 @@ async def chat_stream( async def events(): for event in ( - {"type": SSE_EVENT_TYPE_META, "intent": "general_question"}, + {"type": SSE_EVENT_TYPE_META, "intent": inferred_intent or "general_question"}, {"type": SSE_EVENT_TYPE_TEXT, "content": answer}, {"type": SSE_EVENT_TYPE_DONE}, ): @@ -267,15 +320,10 @@ async def chat_stream( ) else: customer_id = chat_request.customer_id - resolved_intent = recognize_advisor_intent( - chat_request.query, - chat_request.intent, - ) relation = await ensure_customer_access( db, advisor_id=user.id, customer_id=int(customer_id) ) if inferred_intent == AGENT_INTENT_RECOMMEND: - if resolved_intent == AGENT_INTENT_RECOMMEND: memories = await _recall_advisor_memories( request, customer_id=int(customer_id), @@ -286,35 +334,43 @@ async def chat_stream( db, customer_id=int(customer_id) ) if context is not None: - draft = await generate_recommendation_draft( - draft_repo=AdvisorDraftRepo(db), - advisor_id=user.id, - relation_status=relation.status, - trace_id=trace_id, - memories=memories, - llm_client=getattr(runtime, "llm_client", None), - **context, - ) + try: + draft = await generate_recommendation_draft( + draft_repo=AdvisorDraftRepo(db), + advisor_id=user.id, + relation_status=relation.status, + trace_id=trace_id, + memories=memories, + llm_client=getattr(runtime, "llm_client", None), + **context, + ) + except Exception: + logger.warning("advisor recommendation generation failed", exc_info=True) + payload = agent_failure( + ERR_CODE_LLM_ERROR, + "推荐方案生成失败，请稍后重试", + trace_id=trace_id, + ) + else: + async def events(): + for event in ( + { + "type": SSE_EVENT_TYPE_META, + "draft_id": draft.draft_id, + "intent": draft.intent, + "status": draft.status, + }, + {"type": SSE_EVENT_TYPE_TEXT, "content": draft.content}, + {"type": SSE_EVENT_TYPE_DONE, "draft_id": draft.draft_id}, + ): + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" - async def events(): - for event in ( - { - "type": SSE_EVENT_TYPE_META, - "draft_id": draft.draft_id, - "intent": draft.intent, - "status": draft.status, - }, - {"type": SSE_EVENT_TYPE_TEXT, "content": draft.content}, - {"type": SSE_EVENT_TYPE_DONE, "draft_id": draft.draft_id}, - ): - yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" - - return StreamingResponse( - events(), - media_type="text/event-stream", - headers={"X-Trace-Id": trace_id}, - ) - if resolved_intent == AGENT_INTENT_DATA_QUERY: + return StreamingResponse( + events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + if inferred_intent == AGENT_INTENT_DATA_QUERY: if not chat_request.query or not chat_request.query.strip(): payload = agent_failure( ERR_CODE_FORBIDDEN_CUSTOMER, @@ -340,8 +396,9 @@ async def chat_stream( else: async def events(): yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_META, 'intent': AGENT_INTENT_DATA_QUERY, 'query_id': result.get('query_id'), 'trace_id': trace_id}, ensure_ascii=False)}\n\n" - if result.get("summary"): - yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_TEXT, 'content': result['summary']}, ensure_ascii=False)}\n\n" + answer = result.get("answer") or result.get("summary") + if answer: + yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_TEXT, 'content': answer}, ensure_ascii=False)}\n\n" yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_DONE, 'query_id': result.get('query_id')}, ensure_ascii=False)}\n\n" return StreamingResponse( @@ -349,7 +406,73 @@ async def chat_stream( media_type="text/event-stream", headers={"X-Trace-Id": trace_id}, ) - payload = agent_failure(_NOT_READY_CODE, _NOT_READY_MESSAGE, trace_id=trace_id) + if inferred_intent == AGENT_INTENT_FUND_ANALYSIS: + fund_codes = re.findall(r"[A-Za-z]{1,6}\d{3,8}", chat_request.query.upper()) + contexts = await load_fund_analysis_context(db, fund_codes=fund_codes) + if not contexts: + payload = agent_failure( + ERR_CODE_LLM_ERROR, + "未找到可分析的基金数据", + trace_id=trace_id, + ) + else: + result = build_fund_analysis( + contexts[0]["fund"], contexts[0]["performance"] + ) + + async def events(): + for event in ( + {"type": SSE_EVENT_TYPE_META, "intent": AGENT_INTENT_FUND_ANALYSIS}, + {"type": SSE_EVENT_TYPE_TEXT, "content": result["analysis_text"]}, + {"type": SSE_EVENT_TYPE_DONE}, + ): + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + + return StreamingResponse( + events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + if inferred_intent == AGENT_INTENT_DIALOGUE_SCRIPT: + if "市场" in chat_request.query or "波动" in chat_request.query: + scene_type = TALK_SCENE_MARKET_FLUCTUATION + elif "投诉" in chat_request.query: + scene_type = TALK_SCENE_CUSTOMER_COMPLAINT + elif "拦截" in chat_request.query or "风控" in chat_request.query: + scene_type = TALK_SCENE_RISK_BLOCK_ORDER + else: + scene_type = TALK_SCENE_PORTFOLIO_DIVERGENCE + result = build_talk_script(scene_type) + + async def events(): + for event in ( + {"type": SSE_EVENT_TYPE_META, "intent": AGENT_INTENT_DIALOGUE_SCRIPT}, + {"type": SSE_EVENT_TYPE_TEXT, "content": result["content"]}, + {"type": SSE_EVENT_TYPE_DONE}, + ): + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + + return StreamingResponse( + events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + if inferred_intent == AGENT_INTENT_CASUAL_CHAT: + async def events(): + for event in ( + {"type": SSE_EVENT_TYPE_META, "intent": AGENT_INTENT_CASUAL_CHAT}, + {"type": SSE_EVENT_TYPE_TEXT, "content": "您好，我是投顾助手，可以协助您进行基金分析和投资组合管理。"}, + {"type": SSE_EVENT_TYPE_DONE}, + ): + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + + return StreamingResponse( + events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + if payload is None: + payload = agent_failure(_NOT_READY_CODE, _NOT_READY_MESSAGE, trace_id=trace_id) async def events(): yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_ERROR, **payload}, ensure_ascii=False)}\n\n" diff --git a/common/common_const.py b/common/common_const.py index fb64ed3..4161b31 100644 --- a/common/common_const.py +++ b/common/common_const.py @@ -18,6 +18,7 @@ AGENT_INTENT_REBALANCE = "rebalance" AGENT_INTENT_FUND_ANALYSIS = "fund_analysis" AGENT_INTENT_DIALOGUE_SCRIPT = "dialogue-script" AGENT_INTENT_DATA_QUERY = "data_query" +AGENT_INTENT_CASUAL_CHAT = "casual_chat" TALK_SCENE_RISK_BLOCK_ORDER = "risk_block_order" TALK_SCENE_MARKET_FLUCTUATION = "market_fluctuation" diff --git a/config/settings.py b/config/settings.py index adc42b3..198b5b4 100644 --- a/config/settings.py +++ b/config/settings.py @@ -111,7 +111,8 @@ class LLMCfg(BaseSettings): class AdvisorAgentCfg(BaseSettings): """投顾工作台调用 Agent 服务的配置。""" - base_url: str = "" + # Agent 已内置在当前 FastAPI 应用中；有独立部署时可通过 ADVISOR_AGENT_BASE_URL 覆盖。 + base_url: str = "http://127.0.0.1:8000" timeout: float = 1.0 request_timeout: float = 1.0 retry: int = 1 diff --git a/nl2sql/cache.py b/nl2sql/cache.py index f3176d8..325ff2b 100644 --- a/nl2sql/cache.py +++ b/nl2sql/cache.py @@ -106,6 +106,7 @@ def result_to_cache(result: DataQueryResult) -> dict[str, Any]: "row_count": result.row_count, "truncated": result.truncated, "summary": result.summary, + "answer": result.answer, "markdown": result.markdown, "chart": result.chart, "metric_definitions": result.metric_definitions, diff --git a/nl2sql/contracts.py b/nl2sql/contracts.py index 3e5c32a..e941e45 100644 --- a/nl2sql/contracts.py +++ b/nl2sql/contracts.py @@ -44,6 +44,7 @@ class DataQueryResult: row_count: int = 0 truncated: bool = False summary: str | None = None + answer: str | None = None markdown: str | None = None chart: dict[str, Any] | None = None metric_definitions: list[dict[str, Any]] = field(default_factory=list) diff --git a/nl2sql/result.py b/nl2sql/result.py index 195fc41..dbb1fcd 100644 --- a/nl2sql/result.py +++ b/nl2sql/result.py @@ -31,10 +31,31 @@ async def summarize_result( llm_client, max_prompt_rows: int = 20, ) -> str: - """使用脱敏后的受控结果生成摘要，模型失败时返回固定话术。""" + """兼容旧调用方：返回面向用户的自然语言答案。""" + return await render_answer( + question, + columns, + rows, + llm_client=llm_client, + max_prompt_rows=max_prompt_rows, + ) + + +async def render_answer( + question: str, + columns: list[str], + rows: list[dict[str, Any]], + *, + llm_client, + max_prompt_rows: int = 20, +) -> str: + """将受控查询结果渲染为用户回答，不输出原始 JSON 或无关字段。""" fallback = f"查询完成，共返回 {len(rows)} 条记录。" prompt = ( - "请用简洁中文总结查询结果，只基于提供的数据，不要猜测。\n" + "请根据用户问题生成简洁、准确的中文业务回答。\n" + "只使用查询结果中的必要字段回答问题，不要输出 JSON、SQL、Markdown 表格或内部字段名。\n" + "如果有多条记录，优先做合计、数量或关键项概括；只有用户明确需要明细时才列出明细。\n" + "不得补充查询结果中没有的数据，不要解释你的处理过程。\n" f"问题：{question}\n" f"列名：{json.dumps(columns, ensure_ascii=False)}\n" f"结果：{json.dumps(rows[:max_prompt_rows], ensure_ascii=False, default=str)}" @@ -42,7 +63,7 @@ async def summarize_result( try: answer = await llm_client.chat( [ - {"role": "system", "content": "你是数据查询结果摘要助手。"}, + {"role": "system", "content": "你是基金平台的数据回答助手。不要输出 JSON。"}, {"role": "user", "content": prompt}, ], temperature=0, diff --git a/schemas/advisor_agent.py b/schemas/advisor_agent.py index 427189b..7744b2d 100644 --- a/schemas/advisor_agent.py +++ b/schemas/advisor_agent.py @@ -42,7 +42,7 @@ class AdvisorChatReq(BaseModel): AGENT_INTENT_DIALOGUE_SCRIPT, AGENT_INTENT_DATA_QUERY, ] | None = None - query: str = Field(min_length=1, max_length=4000) + query: str | None = Field(default=None, max_length=4000) class AdvisorDataQueryReq(BaseModel): diff --git a/tool/llm.py b/tool/llm.py index 13891a3..d89cbf6 100644 --- a/tool/llm.py +++ b/tool/llm.py @@ -169,21 +169,40 @@ class LLMClient: temperature: float | None, max_tokens: int | None, ) -> str: - payload = { - "model": model, - "messages": messages, - "temperature": self.cfg.temperature if temperature is None else temperature, - "max_tokens": self.cfg.max_tokens if max_tokens is None else max_tokens, - "stream": False, - } + request_max_tokens = self.cfg.max_tokens if max_tokens is None else max_tokens url = f"{backend.base_url}/chat/completions" # 指数退避：retry_backoff_sec → 翻倍 → …，最多 max_retries 次 for attempt in range(self.cfg.max_retries): + payload = { + "model": model, + "messages": messages, + "temperature": self.cfg.temperature if temperature is None else temperature, + "max_tokens": request_max_tokens, + "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() - return r.json()["choices"][0]["message"]["content"] + choice = r.json()["choices"][0] + message = choice["message"] + content = message.get("content") + if isinstance(content, str) and content.strip(): + return content + + # DeepSeek reasoning models may consume the entire small token budget + # in reasoning_content, leaving content empty and finish_reason=length. + if choice.get("finish_reason") == "length" and attempt < self.cfg.max_retries - 1: + request_max_tokens = max(request_max_tokens * 4, 256) + logger.warning( + "chat backend=%s model=%r returned empty content after token limit; " + "retrying with max_tokens=%d", + backend.name, + model, + request_max_tokens, + ) + continue + raise ValueError("LLM 返回空内容") except Exception: if attempt == self.cfg.max_retries - 1: raise