大模型应用：RETE高效规则匹配+混元大模型：智能决策系统中的核心模式匹配技术.90

from typing import List, Dict, Callable, Tuple
from dataclasses import dataclass

# ===================== 1. 定义基础数据结构 =====================
# 事实（Fact）：业务数据，比如用户、订单、交易
@dataclass
class Fact:
    """事实类：表示业务中的一条数据（如用户、订单）"""
    type: str  # 事实类型（如"user", "order", "transaction"）
    attrs: Dict[str, any]  # 事实属性（如{"age":25, "id":1001}）

# 规则（Rule）：由条件和动作组成
@dataclass
class Rule:
    """规则类：条件满足时执行对应动作"""
    name: str  # 规则名称
    conditions: List[Callable[[Fact | Tuple[Fact, Fact]], bool]]  # 条件列表
    action: Callable[[], None]  # 满足条件后执行的动作

# ===================== 2. 定义RETE核心节点 =====================
class AlphaNode:
    """Alpha节点：单事实单条件过滤（如age>18、amount>1000）"""
    def __init__(self, condition: Callable[[Fact], bool], is_left: bool = False):
        self.condition = condition  # 过滤条件
        self.is_left = is_left  # 是否为左输入（用于区分多个Alpha节点）
        self.outputs: List["BetaNode | TerminalNode"] = []  # 输出节点
        self.matched_facts: List[Fact] = []  # Alpha内存：缓存满足条件的事实

    def process(self, fact: Fact) -> None:
        """处理事实：满足条件则缓存并传递给下一级节点"""
        if self.condition(fact):
            self.matched_facts.append(fact)
            for node in self.outputs:
                node.process_alpha(fact, is_left=self.is_left)  # 传递给下一级，带上is_left标识

    def remove_fact(self, fact: Fact) -> None:
        """事实删除时：清理缓存并通知下游"""
        if fact in self.matched_facts:
            self.matched_facts.remove(fact)
            for node in self.outputs:
                node.remove_alpha(fact)

class BetaNode:
    """Beta节点：多事实关联匹配（如user.id == order.user_id）"""
    def __init__(self, join_condition: Callable[[Fact, Fact], bool]):
        self.join_condition = join_condition  # 关联条件
        self.outputs: List["TerminalNode"] = []  # 输出节点
        # Beta内存：缓存部分匹配的事实对（left_fact, right_fact）
        self.matched_pairs: List[Tuple[Fact, Fact]] = []
        # 左右输入缓存（分别存储两个类型的事实）
        self.left_facts: List[Fact] = []
        self.right_facts: List[Fact] = []

    def process_alpha(self, fact: Fact, is_left: bool = False) -> None:
        """处理来自Alpha节点的事实：关联匹配并传递"""
        # 1. 先缓存当前事实到左/右输入
        if is_left:
            self.left_facts.append(fact)
            # 用新左事实匹配所有右事实
            for right_fact in self.right_facts:
                if self.join_condition(fact, right_fact):
                    pair = (fact, right_fact)
                    self.matched_pairs.append(pair)
                    for node in self.outputs:
                        node.process_beta(pair)
        else:
            self.right_facts.append(fact)
            # 用新右事实匹配所有左事实
            for left_fact in self.left_facts:
                if self.join_condition(left_fact, fact):
                    pair = (left_fact, fact)
                    self.matched_pairs.append(pair)
                    for node in self.outputs:
                        node.process_beta(pair)

    def remove_alpha(self, fact: Fact, is_left: bool = False) -> None:
        """清理事实并删除关联对"""
        if is_left and fact in self.left_facts:
            self.left_facts.remove(fact)
        elif not is_left and fact in self.right_facts:
            self.right_facts.remove(fact)
        # 清理关联对
        self.matched_pairs = [p for p in self.matched_pairs if fact not in p]

class TerminalNode:
    """终端节点：规则匹配完成，触发动作"""
    def __init__(self, rule: Rule):
        self.rule = rule  # 关联的规则

    def process_beta(self, fact_pair: Tuple[Fact, Fact]) -> None:
        """Beta节点匹配完成，执行规则动作"""
        print(f"✅ 规则[{self.rule.name}]匹配成功！")
        print(f"   匹配的事实对：{fact_pair[0]} | {fact_pair[1]}")
        self.rule.action()

# ===================== 3. 构建RETE网络并测试 =====================
def main():
    # -------------------- 步骤1：定义规则和动作 --------------------
    # 规则1：用户年龄>18 且 订单金额>1000 → 触发高价值用户提醒
    def user_condition(fact: Fact) -> bool:
        """Alpha条件：用户年龄>18"""
        return fact.type == "user" and fact.attrs.get("age", 0) > 18

    def order_condition(fact: Fact) -> bool:
        """Alpha条件：订单金额>1000"""
        return fact.type == "order" and fact.attrs.get("amount", 0) > 1000

    def join_condition(user_fact: Fact, order_fact: Fact) -> bool:
        """Beta关联条件：用户ID=订单用户ID"""
        return user_fact.attrs.get("id") == order_fact.attrs.get("user_id")

    def alert_action():
        """规则动作：高价值用户提醒"""
        print("   🚨 执行动作：发送高价值用户营销短信！\n")

    # 定义规则
    high_value_rule = Rule(
        name="高价值用户规则",
        conditions=[user_condition, order_condition, join_condition],
        action=alert_action
    )

    # -------------------- 步骤2：构建RETE网络 --------------------
    # 1. 创建Alpha节点
    alpha_user = AlphaNode(user_condition, is_left=True)
    alpha_order = AlphaNode(order_condition, is_left=False)

    # 2. 创建Beta节点（关联用户和订单）
    beta_join = BetaNode(join_condition)
    alpha_user.outputs.append(beta_join)  # 用户Alpha节点→Beta节点（左输入）
    alpha_order.outputs.append(beta_join)  # 订单Alpha节点→Beta节点（右输入）

    # 3. 创建终端节点（关联规则）
    terminal = TerminalNode(high_value_rule)
    beta_join.outputs.append(terminal)

    # -------------------- 步骤3：注入事实并测试匹配 --------------------
    # 事实1：用户（ID=1001，年龄25）
    user_fact = Fact(type="user", attrs={"id": 1001, "age": 25})
    # 事实2：订单（用户ID=1001，金额1500）
    order_fact = Fact(type="order", attrs={"user_id": 1001, "amount": 1500})
    # 事实3：无效订单（用户ID=1001，金额500）
    invalid_order = Fact(type="order", attrs={"user_id": 1001, "amount": 500})

    print("=== 注入有效用户事实 ===")
    alpha_user.process(user_fact)  # 注入用户事实（Alpha层通过）

    print("=== 注入无效订单事实 ===")
    alpha_order.process(invalid_order)  # 注入无效订单（Alpha层过滤，无触发）

    print("=== 注入有效订单事实 ===")
    alpha_order.process(order_fact)  # 注入有效订单（Alpha+Beta匹配，触发规则）

if __name__ == "__main__":
    main()

import yaml
from openai import OpenAI
import json

# ====================== 1. 加载规则配置（规则引擎核心）======================
def load_rules(rule_path="rules.yaml"):
    """加载规则配置文件，模拟规则引擎的规则存储"""
    try:
        with open(rule_path, 'r', encoding='utf-8') as f:
            rules = yaml.safe_load(f)
        return rules
    except FileNotFoundError:
        # 默认规则配置
        return {
            "user_rules": {
                "普通用户": {
                    "daily_transfer_limit": 50000,
                    "description": "单日转账限额5万元"
                },
                "VIP用户": {
                    "daily_transfer_limit": 200000,
                    "description": "单日转账限额20万元"
                }
            },
            "risk_list": [
                "6228480402564890123",
                "6225880212345678"
            ]
        }

# ====================== 2. 大模型建议模块 ======================
def init_llm():
    """初始化混元大模型客户端"""
    client = OpenAI(
        api_key='sk-bWlJPKjBrSF*************VXvZ5NP8Ze',
        base_url="https://api.hunyuan.cloud.tencent.com/v1",
    )
    return client

def llm_analysis(user_request, user_info, rules, client):
    """
    大模型分析用户请求，生成建议
    :param user_request: 用户请求（如"转账20万给6228480402564890123"）
    :param user_info: 用户信息（如{"level": "普通用户", "daily_transfer_amount": 0}）
    :param rules: 规则配置
    :param client: 混元大模型客户端
    :return: 大模型分析建议
    """
    # 构造大模型输入提示词（Prompt Engineering）
    prompt = f"""
你是金融转账的智能分析助手，需要分析用户请求并给出建议。
规则：{json.dumps(rules['user_rules'][user_info['level']], ensure_ascii=False)}
用户请求：{user_request}
用户信息：{json.dumps(user_info, ensure_ascii=False)}
请你：
1. 提取关键信息：转账金额、收款账户；
2. 结合规则给出建议（仅回答"建议通过"/"建议拒绝"/"建议人工复核"）。
请直接给出建议，不要过多解释。
"""
    
    try:
        # 调用混元大模型生成建议
        completion = client.chat.completions.create(
            model="hunyuan-lite",
            messages=[
                {'role': 'system', 'content': '你是一个金融转账智能分析助手，负责分析用户转账请求并给出安全建议。'},
                {'role': 'user', 'content': prompt}
            ],
            temperature=0.3
        )
        result = completion.choices[0].message.content
        
        # 提取建议
        if "建议拒绝" in result:
            suggestion = "建议拒绝"
        elif "建议人工复核" in result:
            suggestion = "建议人工复核"
        else:
            suggestion = "建议通过"
        
        # 提取关键信息（正则匹配）
        import re
        amount_match = re.search(r'转账?\s*(\d+(?:,\d+)*)\s*元', user_request)
        account_match = re.search(r'\b\d{16,19}\b', user_request)
        
        amount = int(amount_match.group(1).replace(',', '')) if amount_match else None
        account = account_match.group(0) if account_match else None
        
        return {
            "key_info": {"amount": amount, "account": account},
            "suggestion": suggestion,
            "raw_response": result
        }
    except Exception as e:
        print(f"大模型调用失败: {e}")
        # 降级处理：用简单规则
        import re
        amount_match = re.search(r'(\d+)\s*元', user_request)
        account_match = re.search(r'\b\d{16,19}\b', user_request)
        amount = int(amount_match.group(1)) if amount_match else None
        account = account_match.group(0) if account_match else None
        return {
            "key_info": {"amount": amount, "account": account},
            "suggestion": "建议通过",
            "raw_response": "大模型调用失败，使用默认建议"
        }

# ====================== 3. 规则引擎校验模块 ======================
def rule_engine_check(llm_result, user_info, rules):
    """
    规则引擎执行校验（审批）
    :param llm_result: 大模型分析结果
    :param user_info: 用户信息
    :param rules: 规则配置
    :return: 校验结果、原因
    """
    key_info = llm_result["key_info"]
    user_level = user_info["level"]
    daily_limit = rules["user_rules"][user_level]["daily_transfer_limit"]
    risk_list = rules["risk_list"]
    
    # 1. 金额阈值校验
    if key_info["amount"] > daily_limit:
        return False, f"用户等级为{user_level}，单日转账限额{daily_limit}元，本次申请{key_info['amount']}元，超出阈值"
    # 2. 风险账户校验
    if key_info["account"] in risk_list:
        return False, f"收款账户{key_info['account']}命中反诈黑名单，拒绝转账"
    # 3. 权限校验（简化：默认有转账权限）
    # 4. 流程校验（简化：默认符合流程）
    return True, "所有规则校验通过，允许执行转账"

# ====================== 4. 主执行流程 ======================
def main():
    # 1. 加载规则
    rules = load_rules()
    # 2. 初始化混元大模型
    client = init_llm()
    # 3. 模拟用户请求和用户信息
    user_request = "我要转账200000元给6228480402564890123"  # 20万，黑名单账户
    user_info = {
        "level": "普通用户",
        "daily_transfer_amount": 0  # 今日已转账金额
    }
    # 4. 大模型分析（思考与建议）
    llm_result = llm_analysis(user_request, user_info, rules, client)
    print("===== 大模型分析结果 =====")
    print(f"提取的关键信息：{llm_result['key_info']}")
    print(f"大模型建议：{llm_result['suggestion']}")
    if 'raw_response' in llm_result:
        print(f"大模型原始响应：{llm_result['raw_response']}")
    
    # 5. 规则引擎校验（审批与执行）
    check_result, check_reason = rule_engine_check(llm_result, user_info, rules)
    print("\n===== 规则引擎校验结果 =====")
    print(f"校验通过：{check_result}")
    print(f"原因：{check_reason}")
    
    # 6. 执行动作
    if check_result:
        print("\n执行动作：完成转账")
    else:
        print("\n执行动作：拒绝转账")

if __name__ == "__main__":
    main()

# 规则配置文件
user_rules:
  普通用户:
    daily_transfer_limit: 50000  # 单日转账限额5万
    risk_check: True             # 开启风险校验
  VIP用户:
    daily_transfer_limit: 200000 # 单日转账限额20万
    risk_check: True

risk_list:
  - "6228480402564890123"  # 反诈黑名单账户