LQAgentPlatform/utils_test/Completeness_Enhanced_Test/test_classification_optimization.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
三级分类优化效果测试

测试优化后的分类器对"验收内容"等内容的分类效果
"""

import asyncio
import sys
from pathlib import Path

project_root = Path(__file__).parent.parent.parent

from core.construction_review.component.reviewers.utils.llm_content_classifier_v2 import (
    ContentClassifierClient,
    SectionContent,
    CategoryStandardLoader,
    SecondCategoryStandardLoader
)


async def test_classification_optimization():
    """测试分类优化效果"""

    print("=" * 60)
    print("三级分类优化效果测试")
    print("=" * 60)

    # 加载标准分类
    csv_path = project_root / "core/construction_review/component/doc_worker/config/StandardCategoryTable.csv"
    loader = CategoryStandardLoader(csv_path)

    second_csv_path = project_root / "core/construction_review/component/doc_worker/config/construction_plan_standards.csv"
    second_loader = SecondCategoryStandardLoader(second_csv_path) if second_csv_path.exists() else None

    # 创建分类客户端
    classifier = ContentClassifierClient(
        model="qwen3.5-122b-a10b",
        semaphore=asyncio.Semaphore(5),
        embedding_client=None,
        second_category_loader=second_loader
    )

    # 测试用例1: 验收内容
    print("\n" + "-" * 60)
    print("测试用例1: 验收内容分类")
    print("-" * 60)

    test_content_1 = """<1> 三、验收内容
<2> 针对项目认定的关键工序进行验收，按照相关规范进行验收，其检验方法：检
<3> 查质量证明文件、观察、尺量、测量、砼强度检测等。验收内容及合格标准如下：
<4> 1、材料、机具检查验收对所有材料和机械进行进场登记验收，清退不合格材料和机械。
<5> 2、钢筋、预应力钢绞线、锚夹具、波纹管、压浆料、起吊钢绳、吊具等材料应具有出厂质量证明书和试验报告单。
<6> 3、进场时除检查其外观和标志外，按不同的品种、等级、牌号、规格及生产厂家分批抽取试样进行性能检验。
<7> 4、检验试验方法应符合现行国家标准的规定。
<8> 5、施工工艺验收包括模板安装工艺验收、混凝土浇筑工艺验收等。
<9> 6、机械设备验收包括塔式起重机验收、混凝土泵车验收等。
<10> 7、临时支撑结构验收包括脚手架验收、满堂支架验收等。"""

    lines = test_content_1.strip().split('\n')
    standards = loader.get_standards_by_second_code("Content")  # 验收内容

    section = SectionContent(
        section_key="test_acceptance",
        section_name="验收内容",
        lines=lines,
        numbered_content=test_content_1,
        category_standards=standards,
        line_number_map=list(range(1, len(lines) + 1))
    )

    print(f"二级分类: 验收内容 (Content)")
    print(f"三级分类标准数量: {len(standards)}")
    print(f"测试内容行数: {len(lines)}")

    if standards:
        print(f"\n三级分类标准列表:")
        for std in standards[:5]:
            print(f"  - {std.third_name} ({std.third_code})")

    # 调用分类器
    result = await classifier.classify_content(section)

    print(f"\n分类结果:")
    print(f"  成功: {result.error is None}")
    print(f"  耗时: {result.latency:.2f}s")
    print(f"  分类数量: {len(result.classified_contents)}")
    print(f"  分类率: {result.coverage_rate:.1f}%")

    if result.classified_contents:
        print(f"\n识别到的三级分类:")
        for content in result.classified_contents:
            print(f"  - {content.third_category_name} ({content.third_category_code})")
            print(f"    行号: {content.start_line} - {content.end_line}")

    # 检查是否误分类为"非标准项"
    no_standard_count = sum(1 for c in result.classified_contents if c.third_category_code == "no_standard")
    standard_count = len(result.classified_contents) - no_standard_count

    print(f"\n分类统计:")
    print(f"  标准分类: {standard_count}")
    print(f"  非标准项: {no_standard_count}")

    if no_standard_count > standard_count:
        print(f"  [警告] 非标准项分类过多!")
    else:
        print(f"  [OK] 标准分类占多数")

    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(test_classification_optimization())