diff --git a/api/routers/advisor_agent.py b/api/routers/advisor_agent.py index aae2121..0ad6c48 100644 --- a/api/routers/advisor_agent.py +++ b/api/routers/advisor_agent.py @@ -2,11 +2,11 @@ from __future__ import annotations import json +import re 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 @@ -37,6 +37,7 @@ from config.deps import get_db from config.database import mysql, redis as redis_db from model.sys_user import SysUser from repositories.advisor_draft import AdvisorDraftRepo +from repositories.customer_relation import CustomerRelationRepo from service.advisor_agent.context import ( load_fund_analysis_context, load_customer_risk, @@ -87,6 +88,48 @@ def _advisor_runtime(request: Request): return getattr(getattr(app, "state", None), "advisor_agent_runtime", None) +def _infer_chat_intent(query: str) -> str | None: + """从自然语言问题推断投顾意图；无法确定时保留通用问答。""" + if any(word in query for word in ("调仓", "再平衡", "组合偏离")): + return "rebalance" + if any(word in query for word in ("沟通话术", "怎么和客户说", "解释给客户")): + return "dialogue-script" + if any(word in query for word in ("基金分析", "分析这只基金", "分析产品")): + return "fund_analysis" + if any(word in query for word in ("推荐", "产品建议", "买什么基金", "适合的基金")): + return AGENT_INTENT_RECOMMEND + return None + + +async def _resolve_customer_from_query(db, *, advisor_id: int, query: str) -> tuple[int | None, str | None]: + """解析问题中的客户编号或姓名，并限制在当前投顾客户范围内。""" + number_match = re.search(r"(?:客户|用户)\s*[#编号号:]?\s*(\d+)", query) + relation_repo = CustomerRelationRepo(db) + if number_match: + customer_id = int(number_match.group(1)) + relation = await relation_repo.get_active_relation( + customer_id=customer_id, + advisor_id=advisor_id, + ) + if relation is None: + return None, "问题中的客户不在当前投顾的授权范围内" + return customer_id, None + + rows = await relation_repo.list_customer_rows(advisor_id=advisor_id, limit=100) + matched = { + int(account.id) + for _relation, account, _profile in rows + if account.real_name and account.real_name in query + } + if len(matched) == 1: + return next(iter(matched)), None + if len(matched) > 1: + return None, "问题中的客户姓名无法唯一确定，请补充客户编号" + if "客户" in query or "用户" in query: + return None, "请在问题中补充客户编号或客户姓名" + return None, None + + async def _recall_advisor_memories( request: Request, *, customer_id: int, query: str ) -> list[dict]: @@ -143,29 +186,89 @@ async def _run_rebalance_background( @router.post("/chat/stream") async def chat_stream( request: Request, - body: dict, + body: AdvisorChatReq, user: SysUser = Depends(audited_advisor), db: AsyncSession = Depends(get_db), ): trace_id = _trace_id(request) - try: - chat_request = AdvisorChatReq.model_validate(body) - except ValidationError: - payload = agent_failure( - ERR_CODE_FORBIDDEN_CUSTOMER, - "对话请求缺少有效客户范围或参数", - trace_id=trace_id, + chat_request = body + customer_id = chat_request.customer_id + inferred_intent = chat_request.intent or _infer_chat_intent(chat_request.query) + + # 请求体只传问题时，从问题中解析客户；解析结果仍必须经过投顾关系授权校验。 + if customer_id is None: + customer_id, resolve_error = await _resolve_customer_from_query( + db, + advisor_id=user.id, + query=chat_request.query, ) + if resolve_error: + payload = agent_failure( + ERR_CODE_FORBIDDEN_CUSTOMER, + resolve_error, + trace_id=trace_id, + ) + async def resolve_error_events(): + yield f"data: {json.dumps({'type': SSE_EVENT_TYPE_ERROR, **payload}, ensure_ascii=False)}\n\n" + + return StreamingResponse( + resolve_error_events(), + media_type="text/event-stream", + headers={"X-Trace-Id": trace_id}, + ) + else: + payload = None + + # 不带客户编号时只提供通用基金问答，不读取客户画像，也不生成个性化草稿。 + if customer_id is None: + if inferred_intent in { + AGENT_INTENT_RECOMMEND, + "rebalance", + "fund_analysis", + "dialogue-script", + }: + if payload is None: + payload = agent_failure( + ERR_CODE_FORBIDDEN_CUSTOMER, + "个性化投顾分析需要在问题中明确客户编号或姓名", + trace_id=trace_id, + ) + else: + runtime = _advisor_runtime(request) + llm_client = getattr(runtime, "llm_client", None) + if llm_client is None: + answer = "已收到问题。当前未配置通用投顾模型，请选择客户后使用个性化分析，或联系管理员配置 Agent 服务。" + else: + answer = await generate_text( + llm_client, + system_prompt="你是基金投顾助手，只回答通用基金知识和产品分析问题，不读取或推断任何客户信息，不承诺收益，不代客交易。", + user_prompt=chat_request.query, + fallback=lambda: "当前模型暂时不可用，请稍后重试。", + timeout=5.0, + ) + + async def events(): + for event in ( + {"type": SSE_EVENT_TYPE_META, "intent": "general_question"}, + {"type": SSE_EVENT_TYPE_TEXT, "content": answer}, + {"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}, + ) else: - customer_id = chat_request.customer_id relation = await ensure_customer_access( db, advisor_id=user.id, customer_id=int(customer_id) ) - if chat_request.intent == AGENT_INTENT_RECOMMEND: + if inferred_intent == AGENT_INTENT_RECOMMEND: memories = await _recall_advisor_memories( request, customer_id=int(customer_id), - query=chat_request.query or "", + query=chat_request.query, ) runtime = _advisor_runtime(request) context = await load_recommendation_context( diff --git a/requirements.txt b/requirements.txt index a62e987..991220f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,6 +12,7 @@ pymilvus neo4j python-dotenv pypdf +sqlglot fastapi~=0.141.1 sqlalchemy~=2.0.52 @@ -34,4 +35,4 @@ neo4j~=6.3.0 redis~=8.1.0 pymilvus~=3.0.1 pydantic-settings~=2.15.0 -pyjwt~=2.13.0 \ No newline at end of file +pyjwt~=2.13.0 diff --git a/schemas/advisor_agent.py b/schemas/advisor_agent.py index f48a238..cde13de 100644 --- a/schemas/advisor_agent.py +++ b/schemas/advisor_agent.py @@ -32,14 +32,15 @@ class AdvisorRebalanceRunReq(BaseModel): class AdvisorChatReq(BaseModel): - customer_id: int = Field(gt=0) + # 通用投顾问答不需要客户上下文；个性化推荐时再传入客户编号。 + customer_id: int | None = Field(default=None, gt=0) intent: Literal[ AGENT_INTENT_RECOMMEND, AGENT_INTENT_REBALANCE, AGENT_INTENT_FUND_ANALYSIS, AGENT_INTENT_DIALOGUE_SCRIPT, ] | None = None - query: str | None = Field(default=None, max_length=4000) + query: str = Field(min_length=1, max_length=4000) class AdvisorFundAnalysisReq(BaseModel):