177 lines
6.6 KiB
Python
177 lines
6.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
综合裁决节点
|
|
汇总法务/技术/财务三方意见，消解冲突，输出最终综合评级
|
|
"""
|
|
import json
|
|
from .base_agent import BaseAgent
|
|
|
|
|
|
JUDGE_SYSTEM_PROMPT = """你是一名资深的风险管理委员会主席，负责汇总法务、技术、财务三方专家的研判意见。
|
|
|
|
你的任务是：
|
|
1. 审阅三方专家的评估报告
|
|
2. 识别各方意见的冲突点
|
|
3. 基于优先级原则消解冲突（合规风险 > 技术风险 > 财务风险）
|
|
4. 输出 0-100 的综合风险评分
|
|
5. 给出最终核保建议
|
|
|
|
核保决策标准：
|
|
- 综合风险 ≥ 80分：建议【拒绝承保】
|
|
- 60 ≤ 综合风险 < 80分：建议【附条件承保】（高免赔额/限额）
|
|
- 40 ≤ 综合风险 < 60分：建议【标准承保】（标准费率上浮）
|
|
- 综合风险 < 40分：建议【优先承保】（可享费率优惠）
|
|
|
|
请输出JSON格式：
|
|
{
|
|
"comprehensive_score": 0-100,
|
|
"risk_level": "极高/高/中/低",
|
|
"underwriting_decision": "拒绝承保/附条件承保/标准承保/优先承保",
|
|
"six_dimension_scores": {
|
|
"tech_disruption": 0-100,
|
|
"talent_loss": 0-100,
|
|
"algo_compliance": 0-100,
|
|
"geopolitical": 0-100,
|
|
"rd_capitalization": 0-100,
|
|
"concentration": 0-100
|
|
},
|
|
"conflict_resolution": "...",
|
|
"key_risks": ["..."],
|
|
"underwriting_conditions": ["..."],
|
|
"summary": "..."
|
|
}"""
|
|
|
|
|
|
class JudgeAgent(BaseAgent):
|
|
"""综合裁决节点"""
|
|
|
|
def __init__(self):
|
|
super().__init__(
|
|
name="综合裁决节点",
|
|
system_prompt=JUDGE_SYSTEM_PROMPT,
|
|
role_icon="⚖️",
|
|
)
|
|
|
|
def evaluate(self, company_data: dict, law_result: dict,
|
|
tech_result: dict, finance_result: dict) -> dict:
|
|
"""汇总三方意见，输出综合裁决"""
|
|
prompt = self._build_prompt(company_data, law_result, tech_result, finance_result)
|
|
result = self.infer_json(prompt)
|
|
|
|
if result.get("parse_error"):
|
|
result = self._rule_based_evaluation(company_data, law_result, tech_result, finance_result)
|
|
|
|
result["agent"] = "综合裁决节点"
|
|
result["icon"] = self.role_icon
|
|
return result
|
|
|
|
def _build_prompt(self, company_data: dict, law_result: dict,
|
|
tech_result: dict, finance_result: dict) -> str:
|
|
return f"""请对以下科创企业的三方评估结果进行综合裁决：
|
|
|
|
企业名称：{company_data.get('short_name', '未知')}
|
|
行业：{company_data.get('industry', '未知')}
|
|
|
|
=== 👩⚖️ 法务风控节点评估 ===
|
|
{json.dumps(law_result, ensure_ascii=False, indent=2)}
|
|
|
|
=== 👨🔬 技术风控节点评估 ===
|
|
{json.dumps(tech_result, ensure_ascii=False, indent=2)}
|
|
|
|
=== 👔 财务风控节点评估 ===
|
|
{json.dumps(finance_result, ensure_ascii=False, indent=2)}
|
|
|
|
请消解可能存在的判定冲突，输出综合裁决JSON。"""
|
|
|
|
def _rule_based_evaluation(self, company_data: dict, law_result: dict,
|
|
tech_result: dict, finance_result: dict) -> dict:
|
|
"""规则引擎综合裁决"""
|
|
|
|
# 提取各维度得分
|
|
def safe_score(result: dict, key: str) -> int:
|
|
item = result.get(key, {})
|
|
if isinstance(item, dict):
|
|
return item.get("score", 50)
|
|
return 50
|
|
|
|
# 六维评分
|
|
scores = {
|
|
"tech_disruption": safe_score(tech_result, "tech_disruption_risk"),
|
|
"talent_loss": safe_score(tech_result, "talent_loss_risk"),
|
|
"algo_compliance": safe_score(law_result, "algo_compliance_risk"),
|
|
"geopolitical": safe_score(law_result, "geopolitical_risk"),
|
|
"rd_capitalization": safe_score(finance_result, "rd_capitalization_risk"),
|
|
"concentration": safe_score(finance_result, "concentration_risk"),
|
|
}
|
|
|
|
# 加权综合得分
|
|
weights = {
|
|
"tech_disruption": 0.20,
|
|
"talent_loss": 0.15,
|
|
"algo_compliance": 0.15,
|
|
"geopolitical": 0.20,
|
|
"rd_capitalization": 0.15,
|
|
"concentration": 0.15,
|
|
}
|
|
|
|
comprehensive_score = int(
|
|
sum(scores[k] * weights[k] for k in scores)
|
|
)
|
|
|
|
# 裁决
|
|
if comprehensive_score >= 80:
|
|
decision = "拒绝承保"
|
|
risk_level = "极高"
|
|
elif comprehensive_score >= 60:
|
|
decision = "附条件承保"
|
|
risk_level = "高"
|
|
elif comprehensive_score >= 40:
|
|
decision = "标准承保"
|
|
risk_level = "中"
|
|
else:
|
|
decision = "优先承保"
|
|
risk_level = "低"
|
|
|
|
# 收集关键风险
|
|
key_risks = []
|
|
for dim, score in sorted(scores.items(), key=lambda x: x[1], reverse=True):
|
|
if score >= 60:
|
|
dim_names = {
|
|
"tech_disruption": "技术路线颠覆",
|
|
"talent_loss": "核心人员流失",
|
|
"algo_compliance": "算法/数据合规",
|
|
"geopolitical": "地缘政治/出口管制",
|
|
"rd_capitalization": "研发资本化操纵",
|
|
"concentration": "客户/供应商集中",
|
|
}
|
|
key_risks.append(f"{dim_names.get(dim, dim)}风险({score}分)")
|
|
|
|
# 核保条件
|
|
conditions = []
|
|
if scores["geopolitical"] >= 70:
|
|
conditions.append("要求提供出口管制合规声明及供应链替代方案")
|
|
if scores["rd_capitalization"] >= 60:
|
|
conditions.append("要求额外提供研发资本化会计政策说明及审计意见")
|
|
if scores["concentration"] >= 60:
|
|
conditions.append("要求提供客户分散化计划或前五大客户信用报告")
|
|
if scores["talent_loss"] >= 60:
|
|
conditions.append("要求核心技术人员签署竞业协议且公司有留任激励计划")
|
|
|
|
company_name = company_data.get("short_name", "该企业")
|
|
summary = (
|
|
f"{company_name}综合风险评分{comprehensive_score}分(风险等级:{risk_level})。"
|
|
f"核保建议:【{decision}】。"
|
|
f"主要风险集中在{'、'.join(key_risks[:3]) if key_risks else '无突出风险'}。"
|
|
)
|
|
|
|
return {
|
|
"comprehensive_score": comprehensive_score,
|
|
"risk_level": risk_level,
|
|
"underwriting_decision": decision,
|
|
"six_dimension_scores": scores,
|
|
"conflict_resolution": "基于优先级原则(合规>技术>财务)进行加权裁决",
|
|
"key_risks": key_risks,
|
|
"underwriting_conditions": conditions,
|
|
"summary": summary,
|
|
}
|