88-大模型应用：Drools+混元大模型：企业级智能决策的“规则+底线”双引擎.88

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
import uvicorn

# --------------------------
# 1. 数据模型定义（模拟Drools的Fact和Rule）
# --------------------------
# 规则条件模型（模拟DRL的LHS：Left Hand Side 条件部分）
class RuleCondition(BaseModel):
    field: str          # 匹配的字段名（如amount、has_legal_sign）
    operator: str       # 运算符（>、<、==、!=、>=、<=）
    value: Any          # 匹配值

# 规则动作模型（模拟DRL的RHS：Right Hand Side 动作部分）
class RuleAction(BaseModel):
    action_type: str    # 动作类型（set_field：设置字段，log：日志输出）
    field: Optional[str] = None  # 要设置的字段名
    value: Any          # 要设置的值/日志内容

# 规则模型（模拟Drools的Rule定义）
class Rule(BaseModel):
    rule_id: str        # 规则唯一ID（如CONTRACT_001）
    salience: int = 0   # 优先级（数值越高越先执行，模拟Drools的salience）
    conditions: List[RuleCondition]  # 规则条件（多个条件为AND关系）
    actions: List[RuleAction]        # 规则动作（按顺序执行）

# 事实模型（模拟Drools的Fact，以合同审核为例）
class ContractFact(BaseModel):
    contract_id: str
    amount: float               # 合同金额
    has_legal_sign: bool        # 是否有法人签字
    has_stamp: bool             # 是否盖公章
    audit_result: str = ""      # 审核结果（初始为空）
    violation_reason: str = ""  # 违规原因（初始为空）

# --------------------------
# 2. 规则引擎核心类（模拟Drools的推理引擎+Agenda+Working Memory）
# --------------------------
class DroolsSimulator:
    def __init__(self):
        # 规则库（模拟Rule Base）：存储所有规则
        self.rule_base: List[Rule] = []
        # 工作内存（模拟Working Memory）：存储待匹配的Fact
        self.working_memory: List[Any] = []

    def add_rule(self, rule: Rule):
        """添加规则到规则库"""
        self.rule_base.append(rule)
        # 按优先级排序（模拟Agenda的冲突解决策略）
        self.rule_base.sort(key=lambda x: x.salience, reverse=True)

    def insert_fact(self, fact: Any):
        """插入Fact到工作内存"""
        self.working_memory.append(fact)

    def _match_condition(self, fact: Any, condition: RuleCondition) -> bool:
        """匹配单个规则条件（核心：实现运算符逻辑）"""
        # 获取Fact的字段值
        fact_value = getattr(fact, condition.field, None)
        if fact_value is None:
            return False
        
        # 执行运算符匹配
        if condition.operator == ">":
            return fact_value > condition.value
        elif condition.operator == "<":
            return fact_value < condition.value
        elif condition.operator == "==":
            return fact_value == condition.value
        elif condition.operator == "!=":
            return fact_value != condition.value
        elif condition.operator == ">=":
            return fact_value >= condition.value
        elif condition.operator == "<=":
            return fact_value <= condition.value
        else:
            raise ValueError(f"不支持的运算符：{condition.operator}")

    def _match_rule(self, fact: Any, rule: Rule) -> bool:
        """匹配单个规则的所有条件（AND关系）"""
        for condition in rule.conditions:
            if not self._match_condition(fact, condition):
                return False
        return True

    def _execute_action(self, fact: Any, action: RuleAction):
        """执行规则动作"""
        if action.action_type == "set_field":
            # 设置Fact字段值（模拟Drools的修改Fact）
            setattr(fact, action.field, action.value)
        elif action.action_type == "log":
            # 日志输出（模拟Drools的控制台输出）
            print(f"[规则执行日志] {action.value}")
        else:
            raise ValueError(f"不支持的动作类型：{action.action_type}")

    def fire_all_rules(self) -> List[Any]:
        """触发所有匹配的规则（模拟Drools的fireAllRules）"""
        results = []
        for fact in self.working_memory:
            # 遍历规则库（已按优先级排序）
            for rule in self.rule_base:
                # 匹配规则
                if self._match_rule(fact, rule):
                    # 执行规则动作
                    for action in rule.actions:
                        self._execute_action(fact, action)
            results.append(fact)
        # 清空工作内存（模拟会话结束）
        self.working_memory.clear()
        return results

# --------------------------
# 3. FastAPI接口搭建（模拟Drools的REST API）
# --------------------------
app = FastAPI(title="Drools规则引擎模拟器", version="1.0")
# 初始化规则引擎实例
drools_engine = DroolsSimulator()

# --------------------------
# 接口1：添加规则到规则库
# --------------------------
@app.post("/drools/rules/add", summary="添加规则到规则库")
def add_rule(rule: Rule):
    try:
        drools_engine.add_rule(rule)
        return {"code": 200, "msg": f"规则{rule.rule_id}添加成功", "data": rule}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"添加规则失败：{str(e)}")

# --------------------------
# 接口2：合同审核规则校验（核心接口）
# --------------------------
@app.post("/drools/contract/check", summary="合同审核规则校验")
def contract_check(fact: ContractFact):
    try:
        # 插入Fact到工作内存
        drools_engine.insert_fact(fact)
        # 触发规则执行
        result = drools_engine.fire_all_rules()[0]
        return {
            "code": 200,
            "msg": "规则校验完成",
            "data": {
                "contract_id": result.contract_id,
                "audit_result": result.audit_result,
                "violation_reason": result.violation_reason,
                "original_data": {
                    "amount": result.amount,
                    "has_legal_sign": result.has_legal_sign,
                    "has_stamp": result.has_stamp
                }
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"规则校验失败：{str(e)}")

# --------------------------
# 接口3：获取规则库所有规则
# --------------------------
@app.get("/drools/rules/list", summary="获取规则库所有规则")
def list_rules():
    return {
        "code": 200,
        "msg": "获取规则列表成功",
        "data": [rule.dict() for rule in drools_engine.rule_base]
    }

# --------------------------
# 启动服务
# --------------------------
if __name__ == "__main__":
    # 启动UVicorn服务，访问地址：http://127.0.0.1:8000/docs 可查看接口文档
    uvicorn.run(app, host="0.0.0.0", port=8000)

{
    "rule_id": "CONTRACT_001",
    "salience": 10,
    "conditions": [
        {"field": "amount", "operator": ">", "value": 1000000},
        {"field": "has_legal_sign", "operator": "==", "value": False}
    ],
    "actions": [
        {"action_type": "set_field", "field": "audit_result", "value": "驳回"},
        {"action_type": "set_field", "field": "violation_reason", "value": "合同金额超100万但无法人签字，违反CONTRACT_001规则"},
        {"action_type": "log", "value": "触发规则CONTRACT_001：合同驳回"}
    ]
}

{
    "rule_id": "CONTRACT_002",
    "salience": 5,
    "conditions": [
        {"field": "has_legal_sign", "operator": "==", "value": True},
        {"field": "has_stamp", "operator": "==", "value": False}
    ],
    "actions": [
        {"action_type": "set_field", "field": "audit_result", "value": "通过（需补章）"},
        {"action_type": "set_field", "field": "violation_reason", "value": "合同有法人签字但无公章，需补充公章"},
        {"action_type": "log", "value": "触发规则CONTRACT_002：合同需补章"}
    ]
}

{
    "rule_id": "CONTRACT_003",
    "salience": 1,
    "conditions": [
        {"field": "amount", "operator": "<=", "value": 1000000},
        {"field": "has_legal_sign", "operator": "==", "value": True},
        {"field": "has_stamp", "operator": "==", "value": True}
    ],
    "actions": [
        {"action_type": "set_field", "field": "audit_result", "value": "通过"},
        {"action_type": "set_field", "field": "violation_reason", ""},
        {"action_type": "log", "value": "触发规则CONTRACT_003：合同审核通过"}
    ]
}

import os
import json
import requests
from openai import OpenAI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# --------------------------
# 1. 配置项
# --------------------------
# 模拟Drools REST API地址（实际替换为真实Drools服务地址）
DROOLS_API_URL = "http://localhost:8000/drools/contract/check"
# 腾讯混元大模型配置
HUNYUAN_API_KEY = os.getenv("HUNYUAN_API_KEY", "sk-WlJPK********************vZ5NP8Ze")
client = OpenAI(
    api_key=HUNYUAN_API_KEY,
    base_url="https://api.hunyuan.cloud.tencent.com/v1"
)

# --------------------------
# 2. Drools规则校验函数（模拟+真实接口兼容）
# --------------------------
def drools_contract_check(contract_data):
    """
    调用Drools做硬规则校验（兜底）
    :param contract_data: 合同结构化数据
    :return: 规则校验结果
    """
    try:
        # 调用真实Drools REST API
        response = requests.post(
            DROOLS_API_URL,
            json=contract_data,
            headers={"Content-Type": "application/json"}
        )
        response.raise_for_status()
        return response.json()
    except Exception as e:
        # 本地模拟规则校验（方便测试）
        print(f"Drools接口未部署，使用本地模拟：{e}")
        # 核心规则：金额>100万且无法人签字 → 违规
        if contract_data["amount"] > 1000000 and not contract_data["has_legal_sign"]:
            return {
                "pass": False,
                "violation_reason": "合同金额超100万但无法人签字，违反合规规则CONTRACT_001",
                "rule_id": "CONTRACT_001"
            }
        else:
            return {"pass": True, "violation_reason": "", "rule_id": ""}

# --------------------------
# 3. 大模型柔性处理函数
# --------------------------
def llm_contract_optimize(contract_text, rule_result):
    """
    调用大模型做柔性处理（体验优化）
    :param contract_text: 合同原始文本
    :param rule_result: 规则校验结果
    :return: 大模型生成的建议
    """
    # 规则违规：仅生成解释+整改建议
    if not rule_result["pass"]:
        prompt = f"""
        请以通俗易懂的语言解释以下合同违规原因，并给出具体整改建议：
        违规原因：{rule_result["violation_reason"]}
        要求：整改建议可落地，语言简洁。
        """
    # 规则合规：优化文本+提示潜在风险
    else:
        prompt = f"""
        请优化以下合同文本，要求：
        1. 表述更专业、无歧义，不改变核心权责；
        2. 提示潜在的非合规风险（非违规，但需注意）；
        3. 建议简洁明了，分点列出。
        合同文本：{contract_text}
        """

    # 调用腾讯混元大模型
    try:
        completion = client.chat.completions.create(
            model="hunyuan-lite",
            messages=[
                {"role": "system", "content": "你是专业的合同审核顾问，擅长合规解释和文本优化。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3  # 降低随机性，保证结果稳定
        )
        return completion.choices[0].message.content
    except Exception as e:
        return f"大模型处理失败：{str(e)}"

# --------------------------
# 4. 主流程：Drools+大模型融合
# --------------------------
def contract_review_main(contract_data, contract_text):
    """合同审核主流程：先规则，后大模型"""
    # 第一步：Drools硬规则校验
    print("===== 1. Drools规则引擎校验 =====")
    rule_result = drools_contract_check(contract_data)
    print(f"规则结果：{rule_result}\n")

    # 第二步：大模型柔性处理
    print("===== 2. 大模型柔性处理 =====")
    llm_result = llm_contract_optimize(contract_text, rule_result)
    print(f"大模型建议：{llm_result}\n")

    # 第三步：整合最终结果
    final_result = {
        "rule_check": rule_result,
        "llm_suggestion": llm_result,
        "final_conclusion": "审核不通过" if not rule_result["pass"] else "审核通过，建议参考优化意见"
    }

    print("===== 3. 最终审核结果 =====")
    print(final_result)
    return final_result

# --------------------------
# 5. 测试用例
# --------------------------
if __name__ == "__main__":
    # 测试用例1：违规合同（金额150万，无法人签字）
    test_contract_data_1 = {
        "contract_id": "CON20260213001",
        "amount": 1500000,
        "has_legal_sign": False,
        "parties": ["甲方A", "乙方B"]
    }
    test_contract_text_1 = """
    甲方A与乙方B签订购销合同，合同金额150万元，甲方应在2026年3月1日前支付全款，乙方应在付款后7日内交货。
    本合同一式两份，双方各执一份，签字后生效。
    """
    contract_review_main(test_contract_data_1, test_contract_text_1)

    # 测试用例2：合规合同（金额50万，有法人签字）
    print("\n" + "-"*60 + "\n")
    test_contract_data_2 = {
        "contract_id": "CON20260213002",
        "amount": 500000,
        "has_legal_sign": True,
        "parties": ["甲方C", "乙方D"]
    }
    test_contract_text_2 = """
    甲方C与乙方D签订服务合同，合同金额50万元，甲方应在服务完成后10日内支付费用，乙方应提供6个月的售后保障。
    本合同无明确的违约责任条款，双方口头约定如有争议协商解决。
    """
    contract_review_main(test_contract_data_2, test_contract_text_2)