实战案例｜当本体遇上 Agent：让 AI 真正“听懂业务”并“按规矩办事”【上】

tools = [
        # 数据库直接查询
        query_customer, list_orders, query_order,
        # 本体规则推理（根据实际需要修改定制）
        check_shipment_eligibility, check_expedite_eligibility,
        # 本体规则查询等
        get_business_rules,
    ]

    system_prompt = (
        "你是一个本体增强的企业管理助手。你背后接入了基于 OWL 本体的业务推理引擎。"
        "## 工具选择原则\n"
        "- 日常查询（客户、订单列表）→ 直接用数据库查询工具"
        "- 规则判定（能否发货、能否加急等）→ 使用本体推理工具"
        "- 基于语义概念查询（VIP客户、待处理订单等）→ 使用语义概念查询"
        "## 回答原则"
        "- 推理结果要说明判定依据"
        "- 用简洁清晰的中文回答"
    )

    llm = ChatOpenAI(model="gpt-5-mini", temperature=0)
    return create_agent(llm, tools, system_prompt=system_prompt)

class OntologyReasoner:
    """按需将单笔订单数据注入本体，推理后返回业务分类结果。"""

    def reason_order(self, customer_tier, customer_id, order_id, required_qty, allocations):
        onto = self._base_onto
        self._clear_abox() # 清空旧的临时实例数据

        # 1. 注入客户
        cust_cls = self._cls("VIPCustomer") if customer_tier == "VIP"else self._cls("Customer")
        cust_ind = cust_cls("customer_id")

        # 2. 注入库存与质检信息
        alloc_inds = []
        for a in allocations:
            alloc_ind = self._cls("InventoryAllocation")(a["id"])
            alloc_ind.qcPassed = [a["qc_passed"]]
            alloc_inds.append(alloc_ind)

        # 3. 注入订单并建立关系
        order_ind = self._cls("Order")(order_id)
        order_ind.hasCustomer = [cust_ind]
        order_ind.hasAllocation = alloc_inds

        # 4. 调用推理机
        with onto:
            sync_reasoner(infer_property_values=True)

        # 5. 读取推理结果
        return {
            "ready_to_ship": order_ind in self._cls("ReadyToShipOrder").instances(),
            "expedite_eligible": order_ind in self._cls("ExpediteEligibleOrder").instances(),
        }

def classify_product(product):
    categories = []
    if product["is_hazardous"]:
        categories.append("危险品")
    if product["requires_cold_chain"]:
        categories.append("冷链")
    if product["is_export_controlled"]:
        categories.append("出口管控")
    if categories:
        categories.append("特殊处理")
    return categories

# 危险品：含危险成分 → 自动归类
class DangerousGoods(Product):
    equivalent_to = [Product & isHazardous.value(True)]

# 冷链产品：需要冷链运输 → 自动归类
class ColdChainProduct(Product):
    equivalent_to = [Product & requiresColdChain.value(True)]

# 出口管控产品：受出口管制 → 自动归类
class ExportControlledProduct(Product):
    equivalent_to = [Product & isExportControlled.value(True)]

# 特殊处理产品 = ①②③ 的并集
class SpecialHandlingProduct(Product):
    equivalent_to = [Product & (
        isHazardous.value(True)
        | requiresColdChain.value(True)
        | isExportControlled.value(True)
    )]

defreason_product_classification(self, products):
    """把产品属性注入本体，推理机自动完成所有分类。"""
    onto = self._base_onto
    self._clear_abox()

    prod_cls = self._cls("Product")

    # 四个分类类别，全部在本体初始化时由 equivalent_to 声明
    category_classes = {
        "DangerousGoods": self._cls("DangerousGoods"),
        "ColdChainProduct": self._cls("ColdChainProduct"),
        "ExportControlledProduct": self._cls("ExportControlledProduct"),
        "SpecialHandlingProduct": self._cls("SpecialHandlingProduct"),
    }

    # 1. 将数据库中的产品属性注入本体（ABox）
    individuals = {}
    for p in products:
        ind = prod_cls(p["name"])
        individuals[p["name"]] = ind
        ind.isHazardous = [p.get("is_hazardous", False)]
        ind.requiresColdChain = [p.get("requires_cold_chain", False)]
        ind.isExportControlled = [p.get("is_export_controlled", False)]

    # 2. 推理机一次运行 —— 所有分类自动完成
    with onto:
        sync_reasoner(infer_property_values=True)

    # 3. 读取分类结果（不是 if/else，是检查推理机的归类结论）
    .......省略......