AI260626/group_fqcd_jr
13
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

优化奶龙字段语义理解

Browse Source
This commit is contained in:
zhangshy
2026-09-11 19:37:26 +08:00
parent 3f7c5ca74f
commit beb67f0a26
4 changed files with 131 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/service/agent/implementations/risk_agent.py
+10 -1
View File
@@ -362,7 +362,8 @@ def _agent_system_prompt(message: str) -> str:
"8. 查询结果包含 summary 时，客户、产品和规则数量必须依据完整 summary，"
"不能因为 items 被截断就回答只覆盖部分记录。\n"
"9. 最终回答使用中文，简洁说明结论、依据和剩余风险，并提醒由风控专员人工复核。\n"
f"{context}\n{filter_context}\n{_truncation_instruction()}"
f"{context}\n{filter_context}\n{_truncation_instruction()}\n"
f"{_field_meaning_instruction()}"
)
@@ -374,6 +375,14 @@ def _truncation_instruction() -> str:
)
def _field_meaning_instruction() -> str:
return (
"11. 工具结果中的 field_meanings 是字段中文含义词典。面向用户回答时必须使用"
"这些业务含义组织内容，不要直接罗列 customer_no、ack_status、behavior_score "
"等数据库字段名；只有用户明确要求查看原始字段时才保留字段名。"
)
def _is_disposition_query(message: str) -> bool:
return any(keyword in message for keyword in (
"误报",
app/service/risk_tools.py
+110 -4
View File
@@ -16,6 +16,111 @@ from app.service.risk_judgement_service import (
)
from app.service.risk_query_service import scope_from_context
FIELD_MEANINGS: dict[str, str] = {
"alert_no": "预警编号",
"customer_id": "客户编号",
"customer_no": "客户编号",
"customer_name": "客户姓名（脱敏）",
"product_code": "产品代码",
"product_name": "产品名称",
"alert_type": "预警类型",
"risk_level": "风险等级",
"rule_codes": "命中规则编号",
"triggered_rules": "命中规则编号",
"evidence_summary": "证据摘要",
"event_status": "事件状态",
"status": "业务状态",
"ack_status": "预警确认状态",
"created_at": "创建时间",
"updated_at": "更新时间",
"due_at": "处理截止时间",
"ack_at": "确认时间",
"escalated_at": "升级时间",
"priority_score": "优先级分数",
"is_escalated": "是否已升级",
"evidence_archived": "证据是否已归档",
"close_reason": "关闭原因",
"disposition_hint": "列表级只读研判建议",
"disposition_assessment": "详情级只读研判草案",
"name": "客户姓名（脱敏）",
"birth_date": "出生日期",
"occupation": "职业",
"mobile_masked": "手机号（脱敏）",
"investor_type": "投资者风险承受等级",
"investment_horizon": "投资期限",
"preferred_asset_class": "偏好资产类别",
"trading_frequency": "交易频率",
"total_asset": "总资产",
"behavior_score": "行为分",
"risk_tags": "风险标签",
"trade_account": "交易账号",
"product_category": "产品类别",
"risk_disclosure_required": "是否要求风险揭示",
"second_confirmation_required": "是否要求二次确认",
"recording_required": "是否要求双录",
"transaction_no": "交易编号",
"order_side": "买卖方向",
"transaction_type": "交易类型",
"amount": "金额",
"channel": "交易渠道",
"trade_status": "交易状态",
"risk_disclosure_signed": "风险揭示签署状态",
"second_confirmation": "二次确认状态",
"confirmed_at": "确认时间",
"executed_at": "成交时间",
"flow_no": "资金流水编号",
"flow_type": "资金流水类型",
"settled_at": "到账时间",
"occurred_at": "发生时间",
"source_type": "资金来源",
"match_status": "资金匹配状态",
"shares": "持有份额",
"cost_amount": "持仓成本",
"current_value": "当前市值",
"profit_loss": "浮动盈亏",
"profit_loss_ratio": "盈亏比例",
"holding_days": "持有天数",
"holding_ratio": "持仓占资产比例",
"login_at": "登录时间",
"login_result": "登录结果",
"ip_region": "登录地区",
"device_id": "登录设备标识",
"is_common_device": "是否常用设备",
"failure_reason": "失败原因",
"notification_no": "通知编号",
"send_status": "发送状态",
"receiver_email": "接收邮箱",
"sent_at": "发送时间",
"handler_id": "处理人编号",
"related_transaction_id": "关联交易编号",
"related_work_order_id": "关联工单编号",
"evidence_truncated": "被截断的证据类型",
"data_truncated": "数据是否被截断",
"truncated": "结果是否被截断",
}
def with_field_meanings(value: dict[str, Any]) -> dict[str, Any]:
"""附加当前结果实际出现字段的中文含义，供模型转换后回答。"""
names: set[str] = set()
def collect(item: object) -> None:
if isinstance(item, dict):
for key, child in item.items():
names.add(str(key))
collect(child)
elif isinstance(item, (list, tuple)):
for child in item:
collect(child)
collect(value)
meanings = {
name: FIELD_MEANINGS[name]
for name in sorted(names)
if name in FIELD_MEANINGS
}
return {**value, "field_meanings": meanings}
async def search_risk_alerts_tool(
arguments: RiskAlertQuery,
@@ -61,7 +166,7 @@ def build_alert_search_result(
hint = item.get("disposition_hint") or {}
if isinstance(hint, dict) and hint.get("verdict"):
disposition_counts[str(hint["verdict"])] += 1
return {
return with_field_meanings({
"total": len(compact_items),
"filters": filters or {},
"summary": {
@@ -76,7 +181,7 @@ def build_alert_search_result(
"complete": True,
},
"items": compact_items,
}
})
def _compact_alert_item(item: dict[str, Any]) -> dict[str, Any]:
@@ -170,7 +275,8 @@ async def get_risk_overview_tool(
) -> dict[str, Any]:
del arguments
async with SessionFactory() as session:
return await RiskRepository(session, scope=scope_from_context(context)).overview()
data = await RiskRepository(session, scope=scope_from_context(context)).overview()
return with_field_meanings(data)
async def get_alert_evidence_tool(
@@ -186,4 +292,4 @@ async def get_alert_evidence_tool(
return None
detail = record.to_dict()
detail["disposition_assessment"] = assess_alert_detail(detail)
return detail
return with_field_meanings(detail)
tests/contract/test_risk_agent_contract.py
+8
View File
@@ -446,3 +446,11 @@ def test_agent_prompt_requires_truncation_disclosure() -> None:
assert "data_truncated=true" in prompt
assert "证据不完整" in prompt
def test_agent_prompt_requires_business_field_names() -> None:
prompt = public_risk_agent._agent_system_prompt("查看当前预警")
assert "field_meanings" in prompt
assert "customer_no" in prompt
assert "ack_status" in prompt
tests/unit/service/test_risk_search_result.py
+3
View File
@@ -25,3 +25,6 @@ def test_search_result_summary_contains_all_customers_and_products() -> None:
assert result["summary"]["complete"] is True
assert result["summary"]["disposition_counts"] == {"可考虑放行": 12}
assert len(result["items"]) == 12
assert result["field_meanings"]["customer_no"] == "客户编号"
assert result["field_meanings"]["risk_level"] == "风险等级"
assert result["field_meanings"]["disposition_hint"] == "列表级只读研判建议"