diff --git a/agent/advisor_agent/data_query.py b/agent/advisor_agent/data_query.py index e94e7d6..f359831 100644 --- a/agent/advisor_agent/data_query.py +++ b/agent/advisor_agent/data_query.py @@ -24,12 +24,43 @@ from repositories.fin_holdings import FinHoldingsRepo from repositories.fin_product import FinProductRepo +def enrich_customer_rows( + rows: list[dict[str, Any]], + columns: list[str], + name_by_id: dict[int, str], +) -> dict[str, Any]: + """为查询结果补充客户姓名，并生成可直接展示的姓名摘要。""" + id_keys = ("customer_id", "客户ID", "客户编号", "客户_id") + enriched = [dict(row) for row in rows] + names: list[str] = [] + for row in enriched: + customer_id = next((row.get(key) for key in id_keys if row.get(key) is not None), None) + try: + customer_id = int(customer_id) + except (TypeError, ValueError): + continue + name = name_by_id.get(customer_id) + if name: + row["客户姓名"] = name + names.append(f"客户{customer_id}（{name}）") + output_columns = list(columns) + if any("客户姓名" in row for row in enriched) and "客户姓名" not in output_columns: + insert_at = next( + (index + 1 for index, column in enumerate(output_columns) if column in id_keys), + len(output_columns), + ) + output_columns.insert(insert_at, "客户姓名") + return {"rows": enriched, "columns": output_columns, "name_summary": "、".join(dict.fromkeys(names))} + + 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]: +async def _query_current_holdings( + db, *, customer_id: int, trace_id: str, customer_name: str | None = None +) -> dict[str, Any]: holdings = await FinHoldingsRepo(db).list_by_customer(customer_id, status="持有中") product_repo = FinProductRepo(db) rows: list[dict[str, Any]] = [] @@ -60,7 +91,8 @@ async def _query_current_holdings(db, *, customer_id: int, trace_id: str) -> dic "truncated": False, "summary": f"当前持仓共 {len(rows)} 条记录。", "answer": ( - f"当前持有 {len(rows)} 只基金，总市值约 {total_value:.2f} 元。" + (f"客户{customer_id}（{customer_name}）" if customer_name else f"客户{customer_id}") + + f"当前持有 {len(rows)} 只基金，总市值约 {total_value:.2f} 元。" + (f"包括：{'、'.join(names[:6])}。" if names else "") ), "sql": None, @@ -108,11 +140,25 @@ async def execute_advisor_data_query( await ensure_customer_access(db, advisor_id=advisor_id, customer_id=customer_id) customer_ids = [customer_id] + name_by_id: dict[int, str] = {} + try: + relation_rows = await CustomerRelationRepo(db).list_customer_rows( + advisor_id=advisor_id, limit=max(len(customer_ids), 100) + ) + name_by_id = { + int(account.id): account.real_name + for _relation, account, _profile in relation_rows + if account.real_name + } + except Exception: # noqa: BLE001 姓名增强失败不阻断数据查询 + pass + if scope == "customer" and _is_current_holdings_query(question): return await _query_current_holdings( db, customer_id=customer_id, trace_id=trace_id, + customer_name=name_by_id.get(customer_id), ) permission = await load_query_permission(db, advisor_id) if not permission.get("can_query", False): @@ -168,6 +214,12 @@ async def execute_advisor_data_query( except (EmbeddingError, LLMFailError) as exc: raise QueryServiceError("投顾 Agent 依赖服务不可用，请检查 LLM/Embedding 服务连接") from exc payload = asdict(result) + enriched = enrich_customer_rows(payload.get("rows", []), payload.get("columns", []), name_by_id) + payload["rows"] = enriched["rows"] + payload["columns"] = enriched["columns"] + if enriched["name_summary"]: + existing_answer = payload.get("answer") or payload.get("summary") or "" + payload["answer"] = f"{existing_answer.rstrip('。')}。客户姓名：{enriched['name_summary']}。" payload["sql"] = None payload["customer_id"] = customer_id return payload diff --git a/agent/advisor_agent/intent/generation_flow.py b/agent/advisor_agent/intent/generation_flow.py index d8e82d6..7d2eb74 100644 --- a/agent/advisor_agent/intent/generation_flow.py +++ b/agent/advisor_agent/intent/generation_flow.py @@ -71,6 +71,7 @@ async def generate_recommendation_draft( memories: list[dict] | None = None, llm_client=None, llm_timeout: float = 5.0, + customer_context: dict | None = None, ): draft_data = build_recommendation_draft( customer_id=customer_id, @@ -87,7 +88,10 @@ async def generate_recommendation_draft( user_prompt=( "请根据以下候选基金和客户记忆生成简洁推荐说明：" + json.dumps( - {"candidates": candidates, "memories": memories or []}, + { + "candidates": candidates, + "customer_context": customer_context or {}, + }, ensure_ascii=False, ) ), diff --git a/api/routers/advisor_agent.py b/api/routers/advisor_agent.py index 04b105c..bd84d06 100644 --- a/api/routers/advisor_agent.py +++ b/api/routers/advisor_agent.py @@ -56,7 +56,9 @@ from service.advisor_agent.context import ( load_rebalance_context, load_recommendation_context, ) +from service.advisor_agent.customer_context import load_customer_context from service.event_publisher import publish_event +from nl2sql.session_context import SessionContextStore, build_conversation_context from schemas.advisor_agent import ( AdvisorDraftOperateReq, AdvisorDraftSaveReq, @@ -110,6 +112,16 @@ def _advisor_runtime(request: Request): return getattr(getattr(app, "state", None), "advisor_agent_runtime", None) +@router.get("/session/{session_id}/history") +async def advisor_session_history( + session_id: str, + user: SysUser = Depends(audited_advisor), +): + """读取当前投顾自己的短期 Agent 会话记录。""" + messages = await SessionContextStore(redis_db.client()).load(user.id, session_id) + return agent_success(messages, trace_id=new_request_id()) + + def _infer_chat_intent(query: str) -> str | None: """从自然语言问题推断投顾意图；无法确定时保留通用问答。""" if any(word in query for word in ("调仓", "再平衡", "组合偏离")): @@ -242,7 +254,8 @@ async def chat_stream( explicit_intent=None, ) inferred_intent = classification.intent - customer_id = None + scope = chat_request.scope + customer_id = chat_request.customer_id if not chat_request.query: code = ERR_CODE_FORBIDDEN_CUSTOMER message = "对话请求缺少有效参数" @@ -259,60 +272,30 @@ async def chat_stream( payload = None - # 单客户范围且未传编号时，兼容从问题中解析客户；投顾范围查询不解析客户。 - if inferred_intent in { - AGENT_INTENT_RECOMMEND, - AGENT_INTENT_REBALANCE, - AGENT_INTENT_FUND_ANALYSIS, - AGENT_INTENT_DIALOGUE_SCRIPT, - AGENT_INTENT_DATA_QUERY, - }: - customer_id, resolve_error = await _resolve_customer_from_query( + # 客户 ID 是可选上下文；未选择客户时尝试从自然语言识别姓名，失败不阻断 Agent 判断。 + if customer_id is not None: + await ensure_customer_access(db, advisor_id=user.id, customer_id=int(customer_id)) + else: + resolved_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}) + if resolved_customer_id is not None: + customer_id = resolved_customer_id + scope = "customer" - # 不带客户编号时只提供通用基金问答，不读取客户画像，也不生成个性化草稿。 - if False: - payload = agent_failure( - ERR_CODE_FORBIDDEN_CUSTOMER, - "投顾范围查询仅支持客户数据查询", - trace_id=trace_id, - ) - elif False: - payload = agent_failure( - ERR_CODE_FORBIDDEN_CUSTOMER, - "投顾范围查询不能指定单个客户", - trace_id=trace_id, - ) - elif customer_id is None: - if inferred_intent in { - AGENT_INTENT_RECOMMEND, - AGENT_INTENT_REBALANCE, - AGENT_INTENT_FUND_ANALYSIS, - AGENT_INTENT_DIALOGUE_SCRIPT, - AGENT_INTENT_DATA_QUERY, - }: - if payload is None: - payload = agent_failure( - ERR_CODE_FORBIDDEN_CUSTOMER, - "个性化投顾分析需要在问题中明确客户编号或姓名", - trace_id=trace_id, - ) + if customer_id is None: + if inferred_intent == AGENT_INTENT_DATA_QUERY: + scope = "advisor" else: llm_client = getattr(runtime, "llm_client", None) if inferred_intent == AGENT_INTENT_CASUAL_CHAT: - answer = "您好，我是投顾助手，请选择客户后使用个性化分析。" + answer = "您好，我是投顾助手，可以协助您查询名下客户数据、分析基金和生成投顾辅助方案。" elif llm_client is None: - answer = "已收到问题。当前未配置通用投顾模型，请选择客户后使用个性化分析，或联系管理员配置 Agent 服务。" + answer = "已收到问题。当前未配置通用投顾模型，但您可以直接查询名下客户数据。" else: answer = await generate_text( llm_client, - system_prompt="你是基金投顾助手，只回答通用基金知识和产品分析问题，不读取或推断任何客户信息，不承诺收益，不代客交易。", + system_prompt="你是基金投顾助手。投顾未指定单个客户时，你可以回答通用问题或说明需要的客户范围；不得越权读取客户数据，不承诺收益，不代客交易。", user_prompt=chat_request.query, fallback=lambda: "当前模型暂时不可用，请稍后重试。", timeout=5.0, @@ -331,7 +314,42 @@ async def chat_stream( media_type="text/event-stream", headers={"X-Trace-Id": trace_id}, ) - else: + if inferred_intent == AGENT_INTENT_DATA_QUERY: + effective_scope = "customer" if customer_id is not None else "advisor" + try: + context_store = SessionContextStore(redis_db.client()) + conversation_context = build_conversation_context( + await context_store.load(user.id, chat_request.session_id) + ) + result = await execute_advisor_data_query( + db, + advisor_id=user.id, + customer_id=int(customer_id) if customer_id is not None else None, + scope=effective_scope, + question=chat_request.query, + trace_id=trace_id, + session_id=chat_request.session_id, + conversation_context=conversation_context, + llm_client=getattr(_advisor_runtime(request), "llm_client", None), + ) + except QueryServiceError as exc: + payload = agent_failure(ERR_CODE_LLM_ERROR, _data_query_error_message(exc), trace_id=trace_id) + else: + await context_store.append( + user.id, chat_request.session_id, chat_request.query, + result.get("answer") or result.get("summary") or "查询完成", + ) + + 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" + 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(events(), media_type="text/event-stream", headers={"X-Trace-Id": trace_id}) + + if customer_id is not None: relation = await ensure_customer_access( db, advisor_id=user.id, customer_id=int(customer_id) ) @@ -345,6 +363,9 @@ async def chat_stream( context = await load_recommendation_context( db, customer_id=int(customer_id) ) + customer_context = await load_customer_context( + db, customer_id=int(customer_id), memories=memories + ) if context is not None: try: draft = await generate_recommendation_draft( @@ -354,6 +375,7 @@ async def chat_stream( trace_id=trace_id, memories=memories, llm_client=getattr(runtime, "llm_client", None), + customer_context=customer_context, **context, ) except Exception: @@ -377,43 +399,6 @@ async def chat_stream( ): 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_DATA_QUERY: - if not chat_request.query or not chat_request.query.strip(): - payload = agent_failure( - ERR_CODE_FORBIDDEN_CUSTOMER, - "查询问题不能为空", - trace_id=trace_id, - ) - else: - try: - result = await execute_advisor_data_query( - db, - advisor_id=user.id, - customer_id=int(customer_id) if customer_id is not None else None, - scope="customer", - question=chat_request.query, - trace_id=trace_id, - llm_client=getattr(_advisor_runtime(request), "llm_client", None), - ) - except QueryServiceError as exc: - payload = agent_failure( - ERR_CODE_LLM_ERROR, - _data_query_error_message(exc), - trace_id=trace_id, - ) - 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" - 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( events(), media_type="text/event-stream", @@ -432,6 +417,35 @@ async def chat_stream( result = build_fund_analysis( contexts[0]["fund"], contexts[0]["performance"] ) + runtime = _advisor_runtime(request) + if getattr(runtime, "llm_client", None) is not None: + customer_context = await load_customer_context( + db, + customer_id=int(customer_id), + memories=await _recall_advisor_memories( + request, + customer_id=int(customer_id), + query=chat_request.query, + ), + ) + result["analysis_text"] = await generate_text( + runtime.llm_client, + system_prompt=( + "你是合规的基金投顾分析助手。只能根据基金数据和客户上下文回答，" + "不得承诺收益，不得代客交易；如果信息不足要明确说明。" + ), + user_prompt=json.dumps( + { + "question": chat_request.query, + "fund": contexts[0]["fund"], + "performance": contexts[0]["performance"], + "customer_context": customer_context, + }, + ensure_ascii=False, + ), + fallback=lambda: result["analysis_text"], + timeout=5.0, + ) async def events(): for event in ( @@ -456,6 +470,33 @@ async def chat_stream( else: scene_type = TALK_SCENE_PORTFOLIO_DIVERGENCE result = build_talk_script(scene_type) + runtime = _advisor_runtime(request) + if getattr(runtime, "llm_client", None) is not None: + memories = await _recall_advisor_memories( + request, + customer_id=int(customer_id), + query=chat_request.query, + ) + customer_context = await load_customer_context( + db, customer_id=int(customer_id), memories=memories + ) + result["content"] = await generate_text( + runtime.llm_client, + system_prompt=( + "你是华夏基金合规投顾助手。请生成简洁、克制、尊重客户的沟通参考话术，" + "结合客户上下文但不要暴露内部字段，不承诺收益，不代客交易。" + ), + user_prompt=json.dumps( + { + "scene": scene_type, + "request": chat_request.query, + "customer_context": customer_context, + }, + ensure_ascii=False, + ), + fallback=lambda: result["content"], + timeout=5.0, + ) async def events(): for event in ( diff --git a/frontend/.env.example b/frontend/.env.example deleted file mode 100644 index 44711ac..0000000 --- a/frontend/.env.example +++ /dev/null @@ -1,2 +0,0 @@ -NEXT_PUBLIC_API_BASE_URL=/backend -BACKEND_API_ORIGIN=http://127.0.0.1:8000 diff --git a/nl2sql/query_rewriter.py b/nl2sql/query_rewriter.py new file mode 100644 index 0000000..de031f8 --- /dev/null +++ b/nl2sql/query_rewriter.py @@ -0,0 +1,95 @@ +"""NL2SQL 查询问题改写与会话上下文补全。""" +from __future__ import annotations + +import logging +import re + +from tool.llm import llm + + +logger = logging.getLogger("nl2sql.query_rewriter") + +_SYSTEM_PROMPT = """你是基金平台 NL2SQL 的问题改写器。 +将当前问题改写为不依赖上下文、可直接交给数据库查询系统理解的完整中文问题。 +只输出改写后的问题，不要输出解释、JSON、SQL 或 Markdown。 +只补全指代和省略，不改变用户的查询目标、客户范围、时间范围或排序条件。 +不能扩大权限范围，不能臆造查询结果；上下文不足时保留原问题。 +""" + + +def _rewrite_explicit_customer_identity(question: str) -> str | None: + normalized = re.sub(r"\s+", "", question) + match = re.fullmatch( + r"(?:查询|请问|告诉我)?客户(?:编号|号|#)?(\d+)(是谁|姓名是什么|姓名|基本信息|联系方式|手机号是什么|手机号)", + normalized, + ) + if not match: + return None + return f"查询客户编号{match.group(1)}的姓名" + + +def _rewrite_customer_name_follow_up(question: str, context: str) -> str | None: + """从最近的客户身份问答中恢复客户编号，保证追问可独立查询。""" + normalized_question = re.sub(r"\s+", "", question) + if not any(term in normalized_question for term in ("持仓", "资产", "基金", "产品", "收益", "盈亏", "余额", "交易")): + return None + + lines = context.splitlines() + for index, line in enumerate(lines): + customer_match = re.search(r"客户(?:编号|号|#)?(\d+)", line) + if not customer_match or not any(term in line for term in ("是谁", "姓名")): + continue + for answer_line in lines[index + 1 : index + 4]: + if not answer_line.startswith("assistant:"): + continue + answer = answer_line.split(":", 1)[1].strip() + name_match = re.search(r"(?:姓名|是)\s*[:：]?\s*([\u4e00-\u9fff]{2,6})", answer) + candidates = [name_match.group(1)] if name_match else re.findall(r"[\u4e00-\u9fff]{2,6}", answer) + for name in reversed(candidates): + if name in normalized_question: + return f"查询客户编号{customer_match.group(1)}的持仓和资产信息" + return None + + +async def rewrite_query( + question: str, + conversation_context: str = "", + *, + llm_client=llm, +) -> str: + """使用有限会话上下文补全问题；改写失败时安全返回原问题。""" + original = (question or "").strip() + context = (conversation_context or "").strip() + if not original: + return original + explicit_identity = _rewrite_explicit_customer_identity(original) + if explicit_identity: + return explicit_identity + contextual_identity = _rewrite_customer_name_follow_up(original, context) + if contextual_identity: + return contextual_identity + if not context or llm_client is None: + return original + + prompt = ( + f"会话上下文：\n{context[:4000]}\n\n" + f"当前问题：\n{original[:2000]}\n\n" + "改写后的独立问题：" + ) + try: + rewritten = await llm_client.chat( + [ + {"role": "system", "content": _SYSTEM_PROMPT}, + {"role": "user", "content": prompt}, + ], + temperature=0, + max_tokens=512, + ) + except Exception: # noqa: BLE001 改写失败不阻断原查询 + logger.warning("NL2SQL 查询问题改写失败，继续使用原问题", exc_info=True) + return original + value = str(rewritten or "").strip() + return value or original + + +__all__ = ["rewrite_query"] diff --git a/repositories/fin_transaction.py b/repositories/fin_transaction.py index 488f0c8..3212c24 100644 --- a/repositories/fin_transaction.py +++ b/repositories/fin_transaction.py @@ -36,3 +36,13 @@ class FinTransactionRepo(BaseRepository): FinTransaction.customer_id == customer_id ) ) + + async def list_recent(self, customer_id: int, *, limit: int = 20) -> list[FinTransaction]: + """按客户读取最近成交记录，供投顾上下文摘要使用。""" + stmt = ( + select(FinTransaction) + .where(FinTransaction.customer_id == customer_id) + .order_by(FinTransaction.create_time.desc(), FinTransaction.id.desc()) + .limit(max(1, min(limit, 100))) + ) + return list((await self.db.scalars(stmt)).all()) diff --git a/schemas/advisor_agent.py b/schemas/advisor_agent.py index b41f488..f426a6d 100644 --- a/schemas/advisor_agent.py +++ b/schemas/advisor_agent.py @@ -22,6 +22,9 @@ class AdvisorRebalanceRunReq(BaseModel): class AdvisorChatReq(BaseModel): query: str = Field(min_length=1, max_length=4000) + session_id: str | None = Field(default=None, max_length=64) + scope: Literal["customer", "advisor"] = "advisor" + customer_id: int | None = Field(default=None, gt=0) class AdvisorDataQueryReq(BaseModel): diff --git a/scripts/seed_advisor_nl2sql_permissions.py b/scripts/seed_advisor_nl2sql_permissions.py new file mode 100644 index 0000000..8a263c9 --- /dev/null +++ b/scripts/seed_advisor_nl2sql_permissions.py @@ -0,0 +1,147 @@ +"""为投顾角色初始化 NL2SQL 最小只读权限。""" +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from sqlalchemy import text + +from config import database +from config.database.mysql import get_session_factory + + +ROLE = { + "role_code": "advisor", + "role_name": "投顾", + "employee_role": "投顾", + "can_query": 1, + "max_rows": 1000, + "daily_quota": 100, + "status": "active", +} + +TABLE_PERMISSIONS = [ + ("fin_holdings", "customer_ids", "customer_id"), + ("fin_transaction", "customer_ids", "customer_id"), + ("fin_customer_profile", "customer_ids", "customer_id"), + ("customer_relation", "customer_ids", "customer_id"), + ("sys_user", "customer_ids", "id"), + ("fin_product", "none", None), + ("fund_performance", "none", None), +] + +COLUMNS = { + "fin_holdings": ( + "customer_id", "product_id", "shares", "cost_amount", "current_value", + "profit_loss", "profit_ratio", "status", "update_time", + ), + "fin_transaction": ( + "customer_id", "product_id", "transaction_type", "amount", "shares", + "nav", "fee", "status", "create_time", + ), + "fin_customer_profile": ( + "customer_id", "risk_level", "risk_score", "investment_experience", + "annual_income_range", "total_assets", "customer_level", "update_time", + ), + "customer_relation": ( + "customer_id", "advisor_id", "assign_time", "signed_time", "status", + ), + "sys_user": ( + "id", "real_name", "phone", "customer_level", "status", + ), + "fin_product": ( + "id", "product_code", "product_name", "product_type", "risk_level", + "expected_return", "nav", "nav_date", "fee_rate", "term_days", + "fund_manager", "status", + ), + "fund_performance": ( + "product_id", "period", "return_rate", "annual_volatility", + "max_drawdown", "sharpe", "calc_date", + ), +} + + +async def main() -> None: + async with get_session_factory()() as db: + await db.execute( + text( + """ + INSERT INTO nl2sql_query_role + (role_code, role_name, employee_role, can_query, max_rows, daily_quota, status) + VALUES + (:role_code, :role_name, :employee_role, :can_query, :max_rows, :daily_quota, :status) + ON DUPLICATE KEY UPDATE + role_name = VALUES(role_name), + can_query = VALUES(can_query), + max_rows = VALUES(max_rows), + daily_quota = VALUES(daily_quota), + status = VALUES(status) + """ + ), + ROLE, + ) + role_id = int( + await db.scalar( + text("SELECT id FROM nl2sql_query_role WHERE employee_role = :employee_role"), + {"employee_role": ROLE["employee_role"]}, + ) + ) + + await db.execute( + text("DELETE FROM nl2sql_role_table_permission WHERE role_id = :role_id"), + {"role_id": role_id}, + ) + await db.execute( + text("DELETE FROM nl2sql_role_column_permission WHERE role_id = :role_id"), + {"role_id": role_id}, + ) + + for table_name, scope_type, scope_column in TABLE_PERMISSIONS: + await db.execute( + text( + """ + INSERT INTO nl2sql_role_table_permission + (role_id, table_name, permission, row_scope_type, row_scope_column, status) + VALUES + (:role_id, :table_name, 'SELECT', :row_scope_type, :row_scope_column, 'active') + """ + ), + { + "role_id": role_id, + "table_name": table_name, + "row_scope_type": scope_type, + "row_scope_column": scope_column, + }, + ) + + for table_name, columns in COLUMNS.items(): + for column_name in columns: + access_mode = "mask" if table_name == "sys_user" and column_name == "phone" else "allow" + await db.execute( + text( + """ + INSERT INTO nl2sql_role_column_permission + (role_id, table_name, column_name, access_mode, mask_type, status) + VALUES + (:role_id, :table_name, :column_name, :access_mode, :mask_type, 'active') + """ + ), + { + "role_id": role_id, + "table_name": table_name, + "column_name": column_name, + "access_mode": access_mode, + "mask_type": "partial" if access_mode == "mask" else None, + }, + ) + + await db.commit() + print({"role_id": role_id, "tables": len(TABLE_PERMISSIONS), "columns": sum(map(len, COLUMNS.values()))}) + await database.mysql.dispose() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/service/advisor/customers.py b/service/advisor/customers.py index cbda198..a81328d 100644 --- a/service/advisor/customers.py +++ b/service/advisor/customers.py @@ -25,7 +25,7 @@ from repositories.risk_assessment import CustomerProfileRepo from repositories.sys_user import SysUserRepo from schemas.advisor import RelationReq from service.advisor.audit_writer import write_audit -from service.advisor.masking import mask_name, mask_phone +from service.advisor.masking import mask_phone from service.advisor.permissions import ensure_customer_owned, require_owned_relation from utils.exceptions import ForbiddenError, NotFoundError from utils.pagination import normalize_pagination, pagination_result @@ -72,7 +72,7 @@ async def list_customers( items.append( { "customer_id": rel.customer_id, - "real_name": mask_name(account.real_name), + "real_name": account.real_name or "", "phone": mask_phone(account.phone), "risk_level": profile.risk_level if profile else None, "customer_level": account.customer_level, @@ -107,7 +107,7 @@ async def get_customer( return { "customer_id": customer_id, - "real_name": mask_name(account.real_name), + "real_name": account.real_name or "", "phone": phone, "risk_level": profile.risk_level if profile else None, "risk_score": profile.risk_score if profile else None, diff --git a/service/advisor_agent/customer_context.py b/service/advisor_agent/customer_context.py new file mode 100644 index 0000000..0f48dba --- /dev/null +++ b/service/advisor_agent/customer_context.py @@ -0,0 +1,115 @@ +"""为投顾 Agent 组装经过裁剪的客户综合上下文。""" +from __future__ import annotations + +import logging +from decimal import Decimal + +from repositories.fin_customer_profile import FinCustomerProfileRepo +from repositories.fin_holdings import FinHoldingsRepo +from repositories.fin_transaction import FinTransactionRepo + +MAX_HOLDINGS = 50 +MAX_TRANSACTIONS = 20 +MAX_MEMORIES = 10 +logger = logging.getLogger("advisor.customer_context") + + +def _number(value): + if isinstance(value, Decimal): + return float(value) + return value + + +def _profile_payload(profile) -> dict: + if profile is None: + return {} + return { + "risk_level": getattr(profile, "risk_level", None), + "risk_score": getattr(profile, "risk_score", None), + "investment_experience": getattr(profile, "investment_experience", None), + "annual_income_range": getattr(profile, "annual_income_range", None), + "total_assets": _number(getattr(profile, "total_assets", None)), + "asset_allocation": getattr(profile, "asset_allocation", None), + "product_preference": getattr(profile, "product_preference", None), + "customer_level": getattr(profile, "customer_level", None), + } + + +def _holding_payload(holding, product) -> dict: + return { + "product_id": holding.product_id, + "product_code": getattr(product, "product_code", None), + "product_name": getattr(product, "product_name", None), + "shares": _number(holding.shares), + "cost_amount": _number(holding.cost_amount), + "current_value": _number(holding.current_value), + "profit_loss": _number(holding.profit_loss), + "profit_ratio": _number(holding.profit_ratio), + "status": holding.status, + } + + +def _transaction_payload(transaction) -> dict: + return { + "product_id": transaction.product_id, + "transaction_type": transaction.transaction_type, + "amount": _number(transaction.amount), + "shares": _number(transaction.shares), + "nav": _number(transaction.nav), + "status": transaction.status, + "create_time": transaction.create_time.isoformat() + if hasattr(transaction.create_time, "isoformat") + else str(transaction.create_time), + } + + +def _memory_payload(memory) -> dict: + return { + "tag": memory.get("tag"), + "content": str(memory.get("content") or "")[:500], + "info_type": memory.get("info_type"), + "memory_type": memory.get("memory_type"), + } + + +async def load_customer_context( + db, + *, + customer_id: int, + memories: list[dict] | None = None, + profile_repo_cls=FinCustomerProfileRepo, + holdings_repo_cls=FinHoldingsRepo, + transaction_repo_cls=FinTransactionRepo, +) -> dict: + """读取并裁剪客户画像、持仓、交易和长期记忆。""" + try: + profile = await profile_repo_cls(db).get_by_customer_id(customer_id) + except Exception: # noqa: BLE001 个性化上下文故障不阻断主流程 + logger.warning("customer profile context unavailable", exc_info=True) + profile = None + try: + holding_rows = await holdings_repo_cls(db).list_with_products( + customer_id, include_closed=False + ) + except Exception: # noqa: BLE001 个性化上下文故障不阻断主流程 + logger.warning("customer holdings context unavailable", exc_info=True) + holding_rows = [] + try: + transactions = await transaction_repo_cls(db).list_recent( + customer_id, limit=MAX_TRANSACTIONS + ) + except Exception: # noqa: BLE001 个性化上下文故障不阻断主流程 + logger.warning("customer transaction context unavailable", exc_info=True) + transactions = [] + return { + "customer_id": customer_id, + "profile": _profile_payload(profile), + "holdings": [ + _holding_payload(holding, product) + for holding, product in holding_rows[:MAX_HOLDINGS] + ], + "transactions": [ + _transaction_payload(item) for item in transactions[:MAX_TRANSACTIONS] + ], + "memories": [_memory_payload(item) for item in (memories or [])[:MAX_MEMORIES]], + }