一、系统总体设计

1.1  设计哲学

建筑规范校验系统的本质是条件判断引擎，而非生成式 AI 应用。这一定位决定了整个系统的技术选型逻辑：

• 规则引擎负责判断对错（确定性、可追溯）

• 大模型负责解释原因（自然语言表达）

• 工作流负责串联流程（自动化、可监控）

系统分五层，每层职责单一，层间通过标准化接口通信：

▸ 架构总览

┌─────────────────────────────────────────┐

│   第五层：工作流自动化 (n8n)              │

├─────────────────────────────────────────┤

│   第四层：AI 解释层 (Dify / LLM)         │

├─────────────────────────────────────────┤

│   第三层：规则引擎 (Python DSL + YAML)   │

├─────────────────────────────────────────┤

│   第二层：参数抽取层 (IFC Parser)         │

├─────────────────────────────────────────┤

│   第一层：数据输入层 (Revit/CAD/手动录入) │

└─────────────────────────────────────────┘

1.2  技术栈选型

1.3  核心数据流

▸ 数据流

BIM文件 (IFC)

     │

     ▼

参数抽取器 ──────► JSON 参数包

                       │

                       ▼

               规则引擎加载器

                       │

                       ▼

               逐条规则判断

                       │

                       ▼

        校验结果列表 [ {rule_id, status, value, threshold} ]

                       │

                       ▼

               AI 解释层

                       │

                       ▼

          PDF/Word 报告  +  推送通知

二、第一层：数据输入层

2.1  BIM 模型接入（主路径）

Revit 导出 IFC 是最优先的输入方式。需要在 Revit 中配置导出参数，确保关键元素被正确标记。以下为 Revit Python Shell / Dynamo 导出脚本：

▸ Python (Revit API)

# revit_exporter.py

import clr

clr.AddReference(‘RevitAPI’)

from Autodesk.Revit.DB import IFCExportOptions, IFCVersion, Transaction

def export_to_ifc(doc, output_path):

    ifc_options = IFCExportOptions()

    ifc_options.FileVersion = IFCVersion.IFC2x3CV2

    ifc_options.ExportBaseQuantities = True       # 导出基础工程量

    ifc_options.WallAndColumnSplitting = True     # 墙柱按楼层拆分

    ifc_options.SpaceBoundaryLevel = 1            # 空间边界（防火分区）

    ifc_options.ExportInternalRevitPropertySets = True

    ifc_options.ExportUserDefinedPsets = True

    ifc_options.ExportUserDefinedPsetsFileName = “fire_safety_psets.txt”

    transaction = Transaction(doc, “Export IFC”)

    transaction.Start()

    try:

        doc.Export(output_path, “model.ifc”, ifc_options)

        transaction.Commit()

        print(f”IFC 导出成功: {output_path}/model.ifc”)

    except Exception as e:

        transaction.RollBack()

        raise RuntimeError(f”导出失败: {str(e)}”)

2.2  IFC 文件上传接口（FastAPI）

▸ Python (FastAPI)

# api/upload.py

from fastapi import APIRouter, UploadFile, File, BackgroundTasks, HTTPException

from pathlib import Path

import uuid, aiofiles

router = APIRouter(prefix=”/api/v1″, tags=[“upload”])

UPLOAD_DIR = Path(“uploads”); UPLOAD_DIR.mkdir(exist_ok=True)

@router.post(“/projects/{project_id}/upload”)

async def upload_ifc(project_id: str, background_tasks: BackgroundTasks,

                     file: UploadFile = File(…)):

    if not file.filename.endswith((‘.ifc’, ‘.ifczip’)):

        raise HTTPException(400, “仅支持 .ifc 或 .ifczip 格式”)

    content = await file.read()

    if len(content) > 500 * 1024 * 1024:

        raise HTTPException(413, “文件超过 500MB 限制”)

    file_id = str(uuid.uuid4())

    async with aiofiles.open(UPLOAD_DIR / f”{file_id}.ifc”, ‘wb’) as f:

        await f.write(content)

    background_tasks.add_task(trigger_parse_pipeline,

                              project_id, file_id, str(UPLOAD_DIR / f”{file_id}.ifc”))

    return {“file_id”: file_id, “status”: “processing”}

async def trigger_parse_pipeline(project_id, file_id, file_path):

    from services.parser import IFCParser

    from services.rule_engine import RuleEngine

    parser = IFCParser(file_path)

    params  = parser.extract_all()

    engine  = RuleEngine()

    results = engine.run(params)

    await save_results(project_id, file_id, results)

    await update_status(file_id, “completed”)

2.3  手动参数录入（备用路径）

当项目没有 BIM 模型时，提供结构化 Pydantic 模型进行参数录入，同样能进入后续规则引擎流程：

▸ Python (Pydantic)

# api/manual_input.py

from pydantic import BaseModel, Field, validator

from typing import Optional, List

from enum import Enum

class BuildingType(str, Enum):

    RESIDENTIAL_HIGH = “高层住宅”

    OFFICE           = “办公建筑”

    COMMERCIAL       = “商业建筑”

    HOSPITAL         = “医疗建筑”

class FireZone(BaseModel):

    zone_id:       str

    area:          float = Field(…, gt=0,  description=”防火分区面积（㎡）”)

    floor:         int   = Field(…,        description=”所在楼层”)

    has_sprinkler: bool  = Field(False,      description=”是否设置自动喷水灭火系统”)

class StairInfo(BaseModel):

    stair_id:     str

    net_width:    float      = Field(…, gt=0, description=”楼梯净宽（m）”)

    stair_type:   str        = Field(…,       description=”封闭/防烟/敞开”)

    serves_floors: List[int] = Field(…,       description=”服务楼层列表”)

class BuildingParams(BaseModel):

    project_name:             str

    building_type:            BuildingType

    total_area:               float = Field(…, gt=0)

    building_height:          float = Field(…, gt=0)

    floor_count:              int   = Field(…, gt=0)

    basement_count:           int   = Field(0,  ge=0)

    fire_zones:               List[FireZone]   = Field(…, min_items=1)

    stairs:                   List[StairInfo]  = Field(…, min_items=1)

    exit_door_count:          int   = Field(…, gt=0)

    max_evacuation_distance:  float = Field(…, gt=0)

    min_sunshine_hours:       Optional[float]  = None

三、第二层：参数抽取层

3.1  IFCParser 核心类

IFCParser 是整个系统的数据网关，负责将非结构化的 IFC 模型转换为规则引擎所需的标准化参数包 BuildingParameters。所有下游模块均消费这份 JSON 格式的参数包。

▸ Python

# services/parser.py

import ifcopenshell

import ifcopenshell.geom

from dataclasses import dataclass, asdict

from typing import Dict, List, Optional

@dataclass

class FireZoneData:

    zone_id: str;  area: float;  floor: int

    floor_name: str;  has_sprinkler: bool

@dataclass

class StairData:

    stair_id: str;  net_width: float;  stair_type: str

    bottom_floor: int;  top_floor: int

@dataclass

class BuildingParameters:

    project_name: str;         ifc_source: str

    building_type: str;        building_height: float

    floor_count: int;          basement_count: int

    total_area: float;         footprint_area: float

    fire_zones: List[FireZoneData]

    max_fire_zone_area: float

    stairs: List[StairData]

    stair_count: int;          min_stair_width: float

    exit_door_count: int;      max_evacuation_distance: float

    min_sunshine_hours: Optional[float]

    has_elevator: bool;        has_refuge_floor: bool

3.2  建筑高度提取逻辑

建筑高度是规则触发的核心参数，采用两阶段获取策略：优先从 IFC 属性集读取精确值，失败时从最高楼层标高加估算层高推算。

▸ Python

def _get_building_height(self) -> float:

    buildings = self.model.by_type(“IfcBuilding”)

    if not buildings:

        return 0.0

    building = buildings[0]

    # 策略1：从属性集读取

    for rel in building.IsDefinedBy:

        if rel.is_a(“IfcRelDefinesByProperties”):

            pset = rel.RelatingPropertyDefinition

            if hasattr(pset, ‘Name’) and ‘Height’ in (pset.Name or ”):

                for prop in pset.HasProperties:

                    if ‘Height’ in prop.Name:

                        return float(prop.NominalValue.wrappedValue)

    # 策略2：从楼层标高推算

    storeys = self.model.by_type(“IfcBuildingStorey”)

    elevations = [s.Elevation for s in storeys if s.Elevation is not None]

    positive = [e for e in elevations if e >= 0]

    if positive:

        return max(positive) + 3.6   # 估算顶层层高

    return 0.0

3.3  防火分区提取逻辑

防火分区提取采用双策略：优先解析 IfcZone 中明确标记的防火分区；若 BIM 建模不规范（未建防火分区），则退而按楼层聚合 IfcSpace 面积作为近似值，并在报告中标记为【需人工确认】。

▸ Python

def _extract_fire_zones(self) -> List[FireZoneData]:

    zones = []

    # 策略1：解析明确标记的 IfcZone

    for zone in self.model.by_type(“IfcZone”):

        if zone.Name and (‘防火’ in zone.Name or ‘Fire’ in zone.Name):

            zones.append(FireZoneData(

                zone_id=zone.GlobalId,

                area=self._get_zone_area(zone),

                floor=self._get_zone_floor(zone),

                floor_name=zone.Name,

                has_sprinkler=self._zone_has_sprinkler(zone),

            ))

    # 策略2：按楼层聚合 IfcSpace（模型未建防火分区时使用）

    if not zones:

        zones = self._aggregate_spaces_by_floor()

    return zones

def _aggregate_spaces_by_floor(self) -> List[FireZoneData]:

    floor_areas: Dict[int, float] = {}

    for space in self.model.by_type(“IfcSpace”):

        floor_num = self._get_element_floor(space)

        area = self._get_space_area(space)

        floor_areas[floor_num] = floor_areas.get(floor_num, 0) + area

    return [

        FireZoneData(zone_id=f”floor_{n}”, area=a, floor=n,

                     floor_name=f”第{n}层（聚合）”, has_sprinkler=False)

        for n, a in sorted(floor_areas.items()) if n >= 0

    ]

四、第三层：规则引擎（核心）

4.1  规则 DSL 结构设计

每条规则由以下字段构成，全部以 YAML 格式存储，支持 Git 版本管理：

4.2  规则 YAML 示例（防火规范）

▸ YAML (规则定义)

# rules/fire_safety.yaml

metadata:

  standard: “GB 50016-2014”

  version:  “2018”

rules:

  – id: FIRE_001

    name:         “高层建筑安全出口数量”

    standard_ref: “GB50016-2014 第5.5.25条”

    category:     “疏散设施”

    severity:     “严重”

    apply_when:

      – { field: building_height, operator: “>”,  value: 27   }

      – { field: building_type,   operator: “in”, value: [“高层住宅”,”住宅建筑”] }

    check:

      field:    exit_door_count

      operator: “>=”

      value:    2

    fail_message: >

      建筑高度为{building_height}m，超过27m，属于高层住宅，

      要求安全出口不少于2个，当前仅有{exit_door_count}个。

    suggestion: “增加安全出口，或核查是否符合可设置1个安全出口的例外条款”

  – id: FIRE_002

    name:         “防火分区面积（高层公共建筑）”

    standard_ref: “GB50016-2014 表5.3.1”

    category:     “防火分区”

    severity:     “严重”

    apply_when:

      – { field: building_height, operator: “>”,  value: 50 }

      – { field: building_type,   operator: “in”, value: [“办公建筑”,”商业建筑”] }

    check:

      type:    “foreach”

      iterate: “fire_zones”

      condition:

        – { field: has_sprinkler, operator: “==”, value: false,

            check_field: area, operator: “<=”, check_value: 1000 }

      condition_alt:

        – { field: has_sprinkler, operator: “==”, value: true,

            check_field: area, operator: “<=”, check_value: 2000 }

    fail_message: >

      防火分区 {zone_id}（第{floor}层）面积 {area}㎡ 超过允许值 {threshold}㎡。

  – id: FIRE_003

    name:         “疏散楼梯净宽度（高层公共建筑）”

    standard_ref: “GB50016-2014 第5.5.18条”

    severity:     “严重”

    apply_when:

      – { field: building_height, operator: “>”, value: 24 }

    check:

      type: “foreach”;  iterate: “stairs”

      check_field: net_width;  operator: “>=”;  check_value: 1.2

  – id: FIRE_005

    name:         “超高层建筑避难层”

    standard_ref: “GB50016-2014 第5.5.23条”

    severity:     “严重”

    apply_when:

      – { field: building_height, operator: “>”,      value: 100 }

      – { field: building_type,   operator: “not_in”, value: [“高层住宅”,”住宅建筑”] }

    check:

      field: has_refuge_floor;  operator: “==”;  value: true

    fail_message: >

      建筑高度 {building_height}m 超过100m，必须设置避难层，当前未检测到。

4.3  规则引擎执行器

规则引擎的核心是 RuleEngine 类和 ConditionEvaluator 类。前者负责加载规则、遍历执行；后者负责单条表达式求值，支持多种比较运算符和嵌套字段访问。

▸ Python (规则引擎核心)

# services/rule_engine.py

import yaml

from pathlib import Path

from dataclasses import dataclass

from typing import List, Dict, Any, Optional

@dataclass

class RuleResult:

    rule_id: str;    rule_name: str;    standard_ref: str

    category: str;   severity: str;     status: str

    message: str;    detail: Dict[str, Any]

    suggestion: Optional[str] = None

@dataclass

class EngineReport:

    project_name: str;   total_rules: int

    passed: int;         failed: int

    warnings: int;       not_applicable: int

    results: List[RuleResult]

    @property

    def pass_rate(self) -> float:

        checked = self.passed + self.failed

        return self.passed / checked if checked > 0 else 0.0

    @property

    def critical_fails(self) -> List[RuleResult]:

        return [r for r in self.results

                if r.status == “fail” and r.severity == “严重”]

class ConditionEvaluator:

    OPERATORS = {

        “>”:      lambda a,b: a > b,   “>=”: lambda a,b: a >= b,

        “<“:      lambda a,b: a < b,   “<=”: lambda a,b: a <= b,

        “==”:     lambda a,b: a == b,  “!=”: lambda a,b: a != b,

        “in”:     lambda a,b: a in b,  “not_in”: lambda a,b: a not in b,

    }

    def evaluate_conditions(self, conditions, params) -> bool:

        return all(

            self.OPERATORS[c[“operator”]](

                self._get(params, c[“field”]), c[“value”])

            for c in conditions

        )

    def _get(self, params, field):

        v = params

        for k in field.split(‘.’):

            if isinstance(v, dict): v = v.get(k)

            elif isinstance(v, list):

                try: v = v[int(k)]

                except: return None

            else: return None

        return v

class RuleEngine:

    def __init__(self, rules_dir=”rules”):

        self.rules = []

        self.evaluator = ConditionEvaluator()

        for f in sorted(Path(rules_dir).glob(“**/*.yaml”)):

            pack = yaml.safe_load(open(f, encoding=’utf-8′))

            self.rules.extend(pack.get(‘rules’, []))

    def run(self, params) -> EngineReport:

        if not isinstance(params, dict):

            from dataclasses import asdict

            params = asdict(params)

        results = [self._evaluate_rule(r, params) for r in self.rules]

        return EngineReport(

            project_name=params.get(‘project_name’,”),

            total_rules=len(results),

            passed=sum(1 for r in results if r.status==”pass”),

            failed=sum(1 for r in results if r.status==”fail”),

            warnings=sum(1 for r in results if r.status==”warning”),

            not_applicable=sum(1 for r in results if r.status==”not_applicable”),

            results=results

        )

    def _evaluate_rule(self, rule, params) -> RuleResult:

        apply_when = rule.get(‘apply_when’, [])

        if apply_when and not self.evaluator.evaluate_conditions(apply_when, params):

            return RuleResult(rule[‘id’],rule[‘name’],rule.get(‘standard_ref’,”),

                              rule.get(‘category’,”),rule.get(‘severity’,’一般’),

                              “not_applicable”,”条件不适用”,{})

        check = rule.get(‘check’,{})

        if check.get(‘type’) == ‘foreach’:

            return self._eval_foreach(rule, check, params)

        return self._eval_single(rule, check, params)

    def _eval_single(self, rule, check, params) -> RuleResult:

        field  = check[‘field’];   op = check[‘operator’];   threshold = check[‘value’]

        actual = self.evaluator._get(params, field)

        passed = self.evaluator.OPERATORS[op](actual, threshold) if actual is not None else False

        msg = f”✓ {field} = {actual}” if passed else               rule.get(‘fail_message’,’不满足要求’).format(**{**params, ‘threshold’:threshold})

        return RuleResult(rule[‘id’],rule[‘name’],rule.get(‘standard_ref’,”),

                          rule.get(‘category’,”),rule.get(‘severity’,’一般’),

                          “pass” if passed else “fail”, msg,

                          {“actual”:actual,”threshold”:threshold},

                          rule.get(‘suggestion’) if not passed else None)

    def _eval_foreach(self, rule, check, params) -> RuleResult:

        items = params.get(check[‘iterate’], [])

        failed = []

        for item in items:

            merged = {**params, **item}

            actual = item.get(check.get(‘check_field’))

            op     = check.get(‘operator’)

            thresh = check.get(‘check_value’)

            if actual is not None and not self.evaluator.OPERATORS[op](actual, thresh):

                failed.append({‘item’:item,’actual’:actual,’threshold’:thresh})

        if not failed:

            return RuleResult(rule[‘id’],rule[‘name’],rule.get(‘standard_ref’,”),

                              rule.get(‘category’,”),rule.get(‘severity’,’一般’),

                              “pass”,f”✓ 全部 {len(items)} 项满足要求”,

                              {“checked”:len(items)})

        msgs = [rule.get(‘fail_message’,”).format(**{**params,**fi[‘item’],

                                                      ‘threshold’:fi[‘threshold’]})

                for fi in failed]

        return RuleResult(rule[‘id’],rule[‘name’],rule.get(‘standard_ref’,”),

                          rule.get(‘category’,”),rule.get(‘severity’,’一般’),

                          “fail”,”

“.join(msgs),

                          {“failed_count”:len(failed),”items”:failed},

                          rule.get(‘suggestion’))

五、第四层：AI 解释层

5.1  设计原则

AI 解释层遵循「不参与计算，只参与表达」原则。规则引擎输出的是机器格式的判断结果（pass/fail + 数值），AI 的作用是将这些结果转换为符合建筑工程行业习惯的专业语言，并提供有建设性的整改建议。

• AI 温度（temperature）设置为 0.3，保证专业性和一致性

• 系统提示词锁定角色为「资深建筑规范审查工程师」

• AI 不得改变规则引擎的判断结论，只能补充解释

• 支持流式输出，提升大报告的响应体验

5.2  AIExplainer 类实现

▸ Python (AI 解释层)

# services/ai_explainer.py

import httpx

from typing import List

class AIExplainer:

    def __init__(self, api_url: str, api_key: str, model=”gpt-4″):

        self.api_url = api_url

        self.api_key = api_key

        self.model   = model

    SYSTEM_PROMPT = “””你是一位资深建筑规范审查工程师，具有20年以上审查经验。

将建筑规范自动校验结果转化为专业审查报告，要求：

1. 语言专业，符合建筑工程行业表达习惯

2. 对不合规项：明确引用规范条文，说明违规内容，给出可操作的整改建议

3. 整改建议按严重程度排序

4. 温度设为0.3，避免创造性发挥，确保报告可重复”””

    async def generate_report(self, engine_report, params: dict) -> str:

        failed = [r for r in engine_report.results if r.status == “fail”]

        prompt = self._build_prompt(engine_report, params, failed)

        return await self._call_llm(prompt)

    def _build_prompt(self, report, params, failed_rules) -> str:

        info = f”””项目：{params.get(‘project_name’)}

建筑类型：{params.get(‘building_type’)} | 高度：{params.get(‘building_height’)}m

总面积：{params.get(‘total_area’)}㎡ | 层数：{params.get(‘floor_count’)}F

校验摘要：共{report.total_rules}条规则，合规{report.passed}条，

不合规{report.failed}条，合规率{report.pass_rate:.1%}

不合规项：

“”” + “

“.join(

            f”[{r.rule_id}] {r.rule_name}（{r.standard_ref}）

问题：{r.message}”

            for r in failed_rules

        )

        return info + “

请输出：①总体评价 ②逐项分析 ③整改优先级 ④总结”

    async def _call_llm(self, user_prompt: str) -> str:

        async with httpx.AsyncClient(timeout=60.0) as client:

            resp = await client.post(

                f”{self.api_url}/v1/chat/completions”,

                headers={“Authorization”: f”Bearer {self.api_key}”},

                json={

                    “model”: self.model,

                    “temperature”: 0.3,

                    “max_tokens”: 2000,

                    “messages”: [

                        {“role”: “system”, “content”: self.SYSTEM_PROMPT},

                        {“role”: “user”,   “content”: user_prompt},

                    ]

                }

            )

        return resp.json()[“choices”][0][“message”][“content”]

六、数据库设计

6.1  核心表结构（PostgreSQL）

▸ SQL

— schema.sql

— 项目表

CREATE TABLE projects (

    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),

    name        VARCHAR(200) NOT NULL,

    building_type VARCHAR(50),

    region      VARCHAR(50)  DEFAULT ‘national’,

    created_at  TIMESTAMP    DEFAULT NOW()

);

— 上传文件表

CREATE TABLE uploaded_files (

    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),

    project_id  UUID REFERENCES projects(id),

    filename    VARCHAR(300),

    file_path   TEXT,

    status      VARCHAR(20)  DEFAULT ‘pending’,

    error_msg   TEXT,

    uploaded_at TIMESTAMP    DEFAULT NOW()

);

— 建筑参数快照表

CREATE TABLE building_params (

    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),

    project_id  UUID REFERENCES projects(id),

    file_id     UUID REFERENCES uploaded_files(id),

    params_json JSONB  NOT NULL,

    building_height   DECIMAL(8,2),

    total_area        DECIMAL(12,2),

    floor_count       INTEGER,

    extracted_at      TIMESTAMP DEFAULT NOW()

);

CREATE INDEX idx_params_project ON building_params(project_id);

— 校验结果明细表

CREATE TABLE check_results (

    id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),

    project_id UUID REFERENCES projects(id),

    params_id  UUID REFERENCES building_params(id),

    rule_id    VARCHAR(50)  NOT NULL,

    rule_name  VARCHAR(200),

    category   VARCHAR(50),

    severity   VARCHAR(20),

    status     VARCHAR(20),   — pass/fail/warning/not_applicable

    message    TEXT,

    detail     JSONB,

    suggestion TEXT,

    checked_at TIMESTAMP DEFAULT NOW()

);

CREATE INDEX idx_results_project ON check_results(project_id);

CREATE INDEX idx_results_status  ON check_results(status);

CREATE INDEX idx_results_rule    ON check_results(rule_id);

— AI 生成报告表

CREATE TABLE ai_reports (

    id               UUID PRIMARY KEY DEFAULT gen_random_uuid(),

    project_id       UUID REFERENCES projects(id),

    report_text      TEXT,

    prompt_tokens    INTEGER,

    completion_tokens INTEGER,

    generated_at     TIMESTAMP DEFAULT NOW()

);

七、测试策略

7.1  规则引擎单元测试

每条规则都需要有覆盖「应通过」「应失败」「不适用」三种场景的测试用例，形成规则测试矩阵。以下是 pytest 测试示例：

▸ Python (pytest)

# tests/test_rule_engine.py

import pytest

from services.rule_engine import RuleEngine

@pytest.fixture

def engine():

    return RuleEngine(rules_dir=”rules”)

@pytest.fixture

def high_rise_office():

    return {

        “project_name”:   “测试高层办公楼”,

        “building_type”:  “办公建筑”,

        “building_height”: 54.0,

        “floor_count”:    15,

        “total_area”:     18000,

        “fire_zones”: [

            {“zone_id”:”FZ01″,”area”:980, “floor”:1,”has_sprinkler”:True},

            {“zone_id”:”FZ02″,”area”:1050,”floor”:2,”has_sprinkler”:True},

            {“zone_id”:”FZ03″,”area”:2600,”floor”:3,”has_sprinkler”:False}, # 超标

        ],

        “stairs”: [

            {“stair_id”:”ST01″,”net_width”:1.3,”stair_type”:”防烟楼梯间”,”bottom_floor”:1,”top_floor”:15},

            {“stair_id”:”ST02″,”net_width”:1.0,”stair_type”:”防烟楼梯间”,”bottom_floor”:1,”top_floor”:15}, # 不足

        ],

        “stair_count”:2, “exit_door_count”:2,

        “max_evacuation_distance”:28, “has_refuge_floor”:False,

        “has_elevator”:True, “min_sunshine_hours”:None,

        “basement_count”:2, “footprint_area”:1200

    }

class TestFireSafetyRules:

    def test_exit_count_pass(self, engine, high_rise_office):

        report = engine.run(high_rise_office)

        r = next(x for x in report.results if x.rule_id == “FIRE_001”)

        assert r.status == “pass”, r.message

    def test_stair_width_fail(self, engine, high_rise_office):

        report = engine.run(high_rise_office)

        r = next((x for x in report.results if x.rule_id == “FIRE_003”), None)

        if r: assert r.status == “fail” and “ST02” in r.message

    def test_fire_zone_area_fail(self, engine, high_rise_office):

        report = engine.run(high_rise_office)

        r = next((x for x in report.results if x.rule_id == “FIRE_002”), None)

        if r: assert r.status == “fail”

    def test_no_refuge_not_required_at_54m(self, engine, high_rise_office):

        report = engine.run(high_rise_office)

        r = next((x for x in report.results if x.rule_id == “FIRE_005”), None)

        if r: assert r.status == “not_applicable”  # 54m < 100m，不适用

    def test_refuge_required_at_110m(self, engine, high_rise_office):

        params = {**high_rise_office, “building_height”:110, “has_refuge_floor”:False}

        report = engine.run(params)

        r = next((x for x in report.results if x.rule_id == “FIRE_005”), None)

        if r: assert r.status == “fail”

    def test_pass_rate_correct(self, engine, high_rise_office):

        report = engine.run(high_rise_office)

        checked = report.passed + report.failed

        if checked > 0:

            assert abs(report.pass_rate – report.passed/checked) < 0.001

    def test_empty_stairs_returns_warning(self, engine, high_rise_office):

        params = {**high_rise_office, “stairs”:[], “stair_count”:0}

        report = engine.run(params)

        # 不应抛异常，返回 warning 或 not_applicable

        assert report is not None

7.2  集成测试：全流程验证

▸ Python (集成测试)

# tests/test_integration.py

import asyncio, json, os

from services.parser import IFCParser

from services.rule_engine import RuleEngine

def test_full_pipeline_with_sample_ifc():

    sample = “tests/fixtures/sample_high_rise.ifc”

    if not os.path.exists(sample):

        pytest.skip(“测试 IFC 文件不存在”)

    parser = IFCParser(sample)

    params = parser.extract_all()

    # 验证参数完整性

    assert params.building_height > 0

    assert params.floor_count > 0

    assert len(params.fire_zones) > 0

    engine = RuleEngine()

    report = engine.run(params)

    # 验证报告完整性

    assert report.total_rules > 0

    assert report.passed + report.failed + report.warnings + report.not_applicable == report.total_rules

    print(f”\n合规率：{report.pass_rate:.1%}，严重不合规：{len(report.critical_fails)}项”)

八、部署配置

8.1  Docker Compose 生产配置

▸ YAML (Docker Compose)

# docker-compose.yml

version: ‘3.9’

services:

  api:

    build: .

    ports: [“8000:8000”]

    environment:

      DATABASE_URL: postgresql://postgres:password@db:5432/building_check

      REDIS_URL:    redis://redis:6379/0

      LLM_API_KEY:  ${LLM_API_KEY}

      RULES_DIR:    /app/rules

    volumes:

      – ./rules:/app/rules:ro

      – uploads:/app/uploads

    depends_on:

      db:    { condition: service_healthy }

      redis: { condition: service_healthy }

  worker:

    build: .

    command: celery -A worker.celery_app worker –loglevel=info -c 4

    environment:

      DATABASE_URL: postgresql://postgres:password@db:5432/building_check

      REDIS_URL:    redis://redis:6379/0

    depends_on: [db, redis]

    volumes:

      – ./rules:/app/rules:ro

      – uploads:/app/uploads

  db:

    image: postgres:15-alpine

    environment:

      POSTGRES_DB:       building_check

      POSTGRES_USER:     postgres

      POSTGRES_PASSWORD: password

    volumes:

      – postgres_data:/var/lib/postgresql/data

      – ./schema.sql:/docker-entrypoint-initdb.d/schema.sql

    healthcheck:

      test: [“CMD-SHELL”,”pg_isready -U postgres”]

      interval: 10s; retries: 5

  redis:

    image: redis:7-alpine

    healthcheck:

      test: [“CMD”,”redis-cli”,”ping”]

    volumes: [redis_data:/data]

  nginx:

    image: nginx:alpine

    ports: [“80:80″,”443:443”]

    volumes:

      – ./nginx.conf:/etc/nginx/conf.d/default.conf:ro

      – ./frontend/dist:/usr/share/nginx/html:ro

    depends_on: [api]

volumes:

  postgres_data:; redis_data:; uploads:

8.2  核心依赖清单

▸ requirements.txt

# requirements.txt

fastapi==0.104.1

uvicorn[standard]==0.24.0

ifcopenshell==0.7.0

pyyaml==6.0.1

pydantic==2.5.0

sqlalchemy==2.0.23

asyncpg==0.29.0

redis==5.0.1

celery==5.3.4

httpx==0.25.2

aiofiles==23.2.1

python-multipart==0.0.6

pytest==7.4.3

pytest-asyncio==0.21.1

九、API 接口设计

9.1  RESTful 接口一览

9.2  校验接口请求/响应示例

▸ JSON (API 示例)

# 请求

POST /api/v1/check/manual

Content-Type: application/json

{

  “params”: {

    “project_name”:  “示例办公楼”,

    “building_type”: “办公建筑”,

    “building_height”: 54.0,

    “floor_count”:   15,

    “total_area”:    18000,

    “fire_zones”: [

      {“zone_id”:”FZ01″,”area”:980,”floor”:1,”has_sprinkler”:true}

    ],

    “stairs”: [

      {“stair_id”:”ST01″,”net_width”:1.3,”stair_type”:”防烟楼梯间”,”bottom_floor”:1,”top_floor”:15}

    ],

    “stair_count”:2, “exit_door_count”:2,

    “max_evacuation_distance”:28, “has_refuge_floor”:false,

    “has_elevator”:true, “min_sunshine_hours”:null,

    “basement_count”:1, “footprint_area”:1200

  },

  “region”: “national”

}

# 响应

{

  “project_name”: “示例办公楼”,

  “summary”: {

    “total”: 12, “passed”: 9, “failed”: 2, “warnings”: 1,

    “pass_rate”: “81.8%”

  },

  “critical_issues”: [

    {

      “rule_id”: “FIRE_002”,

      “name”: “防火分区面积（高层公共建筑）”,

      “message”: “防火分区FZ03面积2600㎡超过允许值1000㎡”,

      “suggestion”: “将防火分区FZ03拆分为两个独立防火分区，或增设自动喷水灭火系统（面积可扩至2000㎡）”

    }

  ]

}

十、扩展路线图

第一阶段 MVP（1-2个月）

核心目标：10条高频规则跑通全流程。验收标准：上传 IFC → 5分钟内输出报告，覆盖防火分区、安全出口、楼梯宽度三大类。

• 实现 IFC 上传 + 参数抽取流水线

• 完成 10 条核心规则的 YAML 编写与测试

• FastAPI 后端 + 简单 React 前端展示结果

• 对接一个 LLM API 生成基础文字报告

第二阶段 规模化（3-6个月）

• 扩展至 50-100 条规则，覆盖住宅/办公/商业三类建筑

• 接入地方规范差异包（北京/上海/广东）

• 规则可视化编辑器（让规范专家无需编程即可维护规则）

• Celery 异步任务队列，支持大文件（>200MB）并发处理

• 校验报告导出为 PDF / Word 格式

第三阶段 产品化（6-12个月）

• Revit 插件：设计过程中实时校验，即时提示不合规项

• 规范知识图谱：条文之间的引用关系自动解析

• 多租户 SaaS：按项目数量或规则包订阅计费

• 与 BIM 360/Bentium iTwin 等平台深度集成

• 历史对比：同一项目多版本校验结果趋势分析

结语

这套系统的核心价值在于把规范的专业判断力从人脑中解放出来，变成可复用、可版本化、可审计的数字资产。

规则引擎保证判断的确定性与可追溯性；AI 层保证表达的专业性与可读性；工作流保证整个过程的自动化与规模化。三者分工明确，互不越位——这是构建这类系统最重要的架构原则。

从一个只覆盖 10 条规则的 MVP 出发，到最终构建覆盖数百条规范条文的专业平台，每一步都要保持规则引擎的核心地位不动摇。AI 是锦上添花，规则是系统的灵魂。