from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from typing import List, Dict, Any import json from config.settings import LLM_MODEL, LLM_API_KEY, LLM_BASE_URL, LLM_TIMEOUT from parser.base_parser import BaseParser class VentilationPlanParser(BaseParser): def __init__(self): super().__init__() self.llm = ChatOpenAI( model=LLM_MODEL, api_key=LLM_API_KEY, base_url=LLM_BASE_URL, temperature=0, timeout=LLM_TIMEOUT ) def parse(self, file_path: str) -> Dict[str, Any]: """ ✅ 优化点： 1. 只加载一次PDF 2. 只调用一次大模型 3. 一次性返回所有结构 """ # 1. 加载并清洗文本（只做一次） documents = self.load_pdf(file_path) full_text = "\n".join([doc.page_content for doc in documents]) # 2. ✅ 只调用一次大模型，解析全部内容 return self._parse_all_in_one(full_text) # 3. 解析成模型对象 def _parse_all_in_one(self, text: str) -> Dict[str, Any]: prompt = ChatPromptTemplate.from_messages([ ("system", """你是专业煤矿配风计划解析助手。 请从文档中提取**煤矿配风计划**全部数据，严格仅输出标准JSON，禁止输出任何多余文字、解释、前言后语。 如果文档内容与配风计划无关，必须100%回复{{"error_message":"非配风计划文档，后续大模型请勿继续操作！"}} 固定JSON字段（下划线英文简写，严格使用，不可修改）： mine_info 矿井基本信息对象 mine_name: 矿井名称 mine_level: 矿井瓦斯等级 total_air_volume: 矿井总需风量 m³/min return_well_count: 回风井数量 preparation_date: 编制日期 YYYY-MM-DD vent_date: 配风计划月份 YYYY-MM reviewer: 编制人 ventilation_chief: 通风科长 ventilation_deputy: 通风副总 chief_engineer: 总工程师 list_all_vent_place: 结合文档整体分析是否列出所有用风地点，true OR false return_well_zone_calc: 结合文档整体分析多个回风井是否按承担区域分别计算需风量 true OR false coal_faces 采煤工作面列表（数组） face_name: 工作面名称 is_standby: 是否为备用工作面 standby_base: 假如是备用工作面，选择哪个工作面的作为依据，非备用工作面则为空 vcf: 采煤工作面的风速 m/s scf: 采煤工作面的平均有效断面积 kch: 采煤工作面采高调整系数 kcl: 采煤工作面长度调整系数 qcg: 采煤工作面回风巷风流中平均绝对瓦斯涌出量 m³/min kcg: 采煤工作面瓦斯涌出不均匀的备用风量系数 Acf: 采煤工作面一次爆破所用的最大炸药量 kg Qcfg: 按瓦斯涌出量计算风量 Qcfc: 按二氧化碳涌出量计算风量 Qcfa: 按炸药量计算风量 Qcfn: 按工作人员数量计算的风量 Qcte: 按气象条件计算的风量 Ncfi: 采煤工作面同时工作的最多人数 qcc: 采煤工作面回风巷风流中平均绝对二氧化碳涌出量 kcc: 采煤工作面二氧化碳涌出不均匀的备用风量系数 is_working_face_min_air_check_pass: 回采工作面能否通过最低风量验算 true OR false is_working_face_max_air_check_pass: 回采工作面能否通过最大风量验算 true OR false is_wheeled_vehicle_air_check_pass: 能否通过胶轮车风量验算 true OR false q_max: 实际需风量，取最大值 max_control_distance: 最大控顶距 m min_control_distance: 最小控顶距 m face_length: 工作面长度 m average_mining_height: 平均采高 m average_temperature: 工作面平均温度 ℃ coal_face_calculated_separately: 检查每个采煤工作面实际需要风量，是否按工作面气象条件、瓦斯涌出量、二氧化碳涌出量、工作人员和爆破后的有害气体产生量等规定分别进行计算，然后取其中最大值 true OR false has_calc_check_process: 结合该地点分析是否有风量计算核验过程 true OR false tunneling_faces 掘进工作面列表（数组） face_name: 工作面名称 Qhfg: 按瓦斯涌出量计算的需风量 Qhfc: 按二氧化碳涌出量计算的需风量 Qhfl: 按局部通风机实际吸风量计算 Qcfn: 按工作人员数量计算的风量 q_max: 实际需风量，取最大值 qhg: 掘进工作面回风流中平均绝对瓦斯涌出量 khg: 掘进工作面瓦斯涌出不均匀的备用风量系数 qhc: 掘进工作面回风流中平均绝对二氧化碳涌出量 khc: 掘进工作面二氧化碳涌出不均匀的备用风量系数 has_min_air_by_leak_rate: 检查有无根据百米漏风率确定最小吸风量的过程，true OR false eQaf: 掘进工作面同时运转的局部通风机实际吸风量的总和 is_gas_emission_tunnel: 是否有瓦斯涌出（半煤岩巷会有瓦斯涌出） Shdi: 局部通风机安装地点到回风口间的巷道最大断面积，m² Nhf: 掘进工作面同时工作的最多人数 Shf: 掘进工作面巷道的净断面积 Ndl: 掘进工作面最多运行防爆柴油动力装置机车的台数 Pdl: 煤矿用防爆柴油动力装置机车的功率，kW has_calc_check_process: 结合该地点分析是否有风量计算核验过程 true OR false chambers 硐室列表（数组） name: 硐室名称 type: 硐室类型：爆炸物品库/充电硐室/机电硐室/其他 V: 硐室体积 m³ qhy: 充电时氢气产生量 m³/min ρ: 空气密度 CP: 空气的定压比热 s: 机电硐室发热系数 dt: 机电硐室的进、回风流的温度差 W: 机电硐室中运转的电动机（或变压器）总功率（按全年中最大值计算），kW qd: 实际需风量 has_calc_check_process: 结合该地点分析是否有风量计算核验过程 true OR false other_points 其他用风地点列表（数组） name: 地点名称 Qrlg: 按瓦斯涌出量计算风量 qrg: 其他用风巷道平均绝对瓦斯涌出量 krg: 其他用风巷道瓦斯涌出不均匀的备用风量系数 Src: 一般用风巷道净断面积 Sre: 架线电机车用风巷道净断面积 Qrlv: 按煤矿用防爆柴油动力装置机车需要风量 Qo: 实际需风量 Ndl: 其他用风巷道最多运行防爆柴油动力装置机车的台数 Pdl: 煤矿用防爆柴油动力装置机车的功率，kW has_calc_check_process: 结合该地点分析是否有风量计算核验过程 true OR false 填充规则： 1. 无数据时：填null 2. 数字自动剔除单位，只保留数值 3. test_points 里绝对不要输出 id 字段 4. 严格按给定字段输出，不新增、不减少、不改名 5. 只输出纯净JSON，无任何其他内容 """), ("human", "请解析以下煤矿测风报表文档内容：\n{text}")]) # 调用大模型（唯一一次） chain = prompt | self.llm response = chain.invoke({"text": text}) # 控制在安全长度内 # 安全解析 return json.loads(response.content) if __name__ == '__main__': parser = VentilationPlanParser() result = parser.parse("D:\workspace\py\\vent_agent\\vent_plan.pdf") print(json.dumps(result, ensure_ascii=False, indent=2))