# -*- coding: utf-8 -*- """ 配风计划审查 - Agent 工厂与编排器模块 核心功能： - 创建3个独立的审查 DeepAgent 实例（形式审查 / 数据一致性 / 计算核验） - 创建汇总 Agent 实例 - 编排器：提取 PDF → 并行运行3个Agent → 汇总结果 → SSE 流式输出 架构： - 复用 vent_agent.py 的 _get_model() 和 FilesystemBackend 模式 - 每个审查 Agent 有独立的 system_prompt + skills + tools - 编排器 stream_review() 使用 asyncio.Queue 实现并行 + 流式进度反馈 """ import asyncio import json import os import time import traceback import uuid from pathlib import Path from typing import AsyncGenerator, Optional from dotenv import load_dotenv load_dotenv() # 复用现有工具的模型加载函数 from agents.vent_agent import _get_model as _get_vent_model # 导入 PDF 工具 from tools.pdf_tools import extract_pdf_text, save_upload_file, calculate_file_hash # 导入计算工具 from tools.calc_tools import ( calc_face_by_gas, calc_face_by_workers, calc_face_by_wind_speed, calc_face_air_volume_max, calc_tunnel_by_gas, calc_tunnel_by_explosives, calc_tunnel_by_workers, calc_tunnel_by_wind_speed, calc_tunnel_by_vehicle, calc_tunnel_air_volume_max, calc_chamber_by_equipment, calc_chamber_by_wind_speed, calc_other_by_wind_speed, calc_effective_area, calc_total_air_volume, calc_gas_emission_from_wind, ) # 工具中文名统一映射 from tools.tool_names_cn import TOOL_NAME_CN as _tool_names # 导入 MCP 配风计划查询工具 from tools.vent_plan_tools import ( query_mining_plan, query_face_procedure, query_gas_report, query_wind_report, ) # 导入报告生成工具 from tools.report_utils import save_review_report, extract_mine_name # 导入系统时间工具 from tools.time_tools import get_current_time # 导入数据库存储 from db.chat_store import save_message, create_session, update_session_title # ============================================================ # 中文名称映射（所有 yield 消息只输出中文，不出现英文代号） # ============================================================ # 子智能体内部名称 → 中文显示名称 AGENT_NAME_CN = { "form-reviewer": "基本形式审查", "data-checker": "数据一致性审查", "calc-verifier": "计算核验审查", } # 工具函数名 → 中文描述（统一维护在 tools/tool_names_cn.py） TOOL_NAME_CN = _tool_names def _cn_agent(name: str) -> str: """将子智能体内部名转为中文显示名，未收录则返回原名。""" return AGENT_NAME_CN.get(name, name) def _cn_tool(name: str) -> str: """将工具函数名转为中文描述，未收录则返回原名。""" return TOOL_NAME_CN.get(name, name) from langchain.agents.middleware.todo import write_todos from langgraph.checkpoint.memory import MemorySaver SKILLS_ROOT = str((Path(__file__).parent.parent / "skills").resolve()) PROJECT_ROOT = str(Path(__file__).parent.parent) # ============================================================ # 模型获取（复用现有项目的 _get_model） # ============================================================ def _get_model(model_key: str = "DEEPAGENT_MODEL"): """获取模型实例，复用现有配置。 参数: model_key: 环境变量中的模型 key，默认 "DEEPAGENT_MODEL"。 summary agent 使用 "SUMMARY_MODEL" 以获取更快/更便宜的模型。 """ return _get_vent_model(model_key=model_key) def _compress_result(text: str, max_chars: int = 8000) -> str: """轻度压缩子智能体输出文本，保留核心结构化数据的同时降低 token 数。 策略： 1. 若文本未超过 max_chars，仅清理冗余空行后返回 2. 超过时按优先级保留：表格 → 标题行 → 列表项 → 正文段落 3. 被截断的正文段落用 [...] 标记 """ if not text: return text # 统一换行符 text = text.replace("\r\n", "\n").replace("\r", "\n") def _normalize_newlines(s: str) -> str: """将连续 3 个以上换行压缩为 2 个""" import re return re.sub(r"\n{3,}", "\n\n", s) # 先做空行归一化 text = _normalize_newlines(text).strip() if len(text) <= max_chars: return text # ── 超过 max_chars，按优先级保留 ── lines = text.split("\n") result_lines: list = [] chars_used = 0 truncation_threshold = max_chars # 达到此阈值后停止添加 # 第一遍：保留表格行和标题行（高优先级） kept_indices: set = set() for i, line in enumerate(lines): stripped = line.strip() # 表格分隔行或表格数据行 if stripped.startswith("|---") or "|---" in stripped or stripped.startswith("|"): kept_indices.add(i) # Markdown 标题 elif stripped.startswith("#") and not stripped.startswith("```"): kept_indices.add(i) # 列表项 elif stripped.startswith("- ") or stripped.startswith("* ") or stripped.startswith("1. "): kept_indices.add(i) # 第二遍：按行拼接，优先保留标记行 for i, line in enumerate(lines): if i in kept_indices: result_lines.append(line) chars_used += len(line) + 1 elif chars_used < truncation_threshold: result_lines.append(line) chars_used += len(line) + 1 compressed = "\n".join(result_lines) compressed = _normalize_newlines(compressed).strip() # 若仍然超限，硬截断并标记 if len(compressed) > max_chars: # 尽量在换行处截断 cut_pos = compressed.rfind("\n", 0, max_chars - 80) if cut_pos == -1: cut_pos = max_chars - 80 compressed = compressed[:cut_pos] + "\n\n...[已截断，完整内容见子智能体原始输出]" if len(compressed) < len(text): compressed += "\n\n...[已截断，完整内容见子智能体原始输出]" return compressed # ============================================================ # 系统提示词 # ============================================================ # 子智能体1: 基本形式审查 FORM_REVIEW_SYSTEM_PROMPT = """你是一名煤矿配风计划形式审查专家。你的任务是对配风计划PDF文件进行基本形式审查。 ## 可用工具 - **get_current_time**: 获取当前系统日期时间（中国标准时间）。在版本审查和编制时间审查时必须先调用此工具获取当前时间，然后与PDF中提取的日期进行对比，判断时效性。 - **write_todos**: 更新审查任务进度。 ## 全局行为准则 1. 所有分析必须基于 PDF 文档中实际提取到的内容，绝不编造。 2. 未提取到的信息标注"文档中未找到"而非臆测。 3. 全程使用简体中文输出。 4. 禁止输出 ANSI 转义序列、内部工具名、函数名。 5. 全流程使用 write_todos 工具实时更新任务进度。 ## 审查内容 严格按照技能（skill）中定义的5个审查维度执行： 1. 版本审查 - 检查配风计划月份是否为当月最新版本 2. 签字审查 - 仅检查签字栏是否有姓名记录，仅作提示，不做手写体/打印体区分，不纳入违规判定 3. 编制时间审查 - 检查编制时间是否在计划月份的上一个月 4. 计算过程完整性 - 检查所有用风地点是否均有完整计算和验算过程 5. 语病逻辑审查 - 检查前后文描述不一致的地方（不需要太严格） ## 输出要求 严格按照技能中定义的格式输出审查结果，不要添加额外标题或代码块。 """ # 子智能体2: 数据一致性审查 DATA_CONSISTENCY_SYSTEM_PROMPT = """你是一名煤矿配风计划数据一致性审查专家。你的任务是对配风计划中数据的内部一致性和外部一致性进行审查。 ## 可用 MCP 工具 你可以调用以下 MCP 工具查询外部数据源进行交叉校验： 1. **query_mining_plan(mine_name, plan_month)** - 查询煤矿月度采掘计划，获取在采在掘工作面清单 - 用于「用风地点完整性」检查：对比采掘计划与配风计划中的工作面是否一致 2. **query_face_procedure(mine_name)** - 查询工作面作业规程，获取最大控顶距、最小控顶距、平均采高等参数 - 用于「工作面参数一致性」检查：对比作业规程与配风计划中的参数是否一致 3. **query_gas_report(mine_name, year)** - 查询瓦斯等级鉴定报告，获取各工作面瓦斯/CO2 涌出量数据 - **不强制要求本年度**：优先查询当前年度，若无数据则回退查询上一年度（year - 1） - 用于「瓦斯与二氧化碳数据一致性」检查：校验涌出量数据是否一致 - **必须在审查结论中标注数据来源的报告年度** 4. **query_wind_report(mine_name, report_month)** - 查询测风报表，获取各测点温度、风速、风量、瓦斯浓度、CO2浓度数据 - 用于「风速与温度匹配」检查：核验温度系数选择是否合理 - **同时用于「瓦斯与CO2数据一致性」交叉验证**：找到各工作面的回风顺槽测点，调用 calc_gas_emission_from_wind 反算涌出量（实测风量 × 浓度 / 100），与配风计划值对比 ## 全局行为准则 1. 所有分析必须基于 PDF 文档内容 + MCP 工具返回的真实数据，绝不编造。 2. 从 PDF 中提取煤矿名称、计划月份等信息，用于构造 MCP 查询参数。 3. **优先调用 MCP 工具**：每个审查维度开始时必须先调用对应的 MCP 工具获取数据，只有工具明确返回错误时才标注查询失败。 4. **瓦斯与CO2数据一致性双重验证**： - 来源A：query_gas_report（瓦斯鉴定报告） - 来源B：query_wind_report → 提取回风顺槽测点 → calc_gas_emission_from_wind 反算 - 两方数据交叉对比，综合判定 5. 全程使用简体中文输出。 6. 禁止输出 ANSI 转义序列、内部工具名、函数名。 7. 全流程使用 write_todos 工具实时更新任务进度。 ## 审查内容 严格按照技能（skill）中定义的4个审查维度执行： 1. 用风地点完整性 - 调用 query_mining_plan 获取采掘计划，对比配风计划 2. 瓦斯与二氧化碳数据一致性 - 调用 query_gas_report 获取鉴定报告（优先本年度，无数据则查上一年度），逐项对比，必须标注报告年度 3. 工作面参数一致性 - 调用 query_face_procedure 获取作业规程，逐参数对比 4. 风速与温度匹配 - 调用 query_wind_report 获取测风数据，核验温度系数 ## 输出要求 严格按照技能中定义的格式输出审查结果，每个审查维度标注数据来源。 """ # 子智能体3: 计算核验 CALC_VERIFY_SYSTEM_PROMPT = """你是一名煤矿通风需风量计算核验专家。你的任务是对配风计划中各用风地点的需风量计算过程进行逐项核验。 ## 全局行为准则 1. 必须调用计算工具（calc_face_air_volume_max、calc_tunnel_air_volume_max 等）获取准确计算结果，不能仅凭 LLM 知识计算。 2. 每个计算过程都要展示公式、代入数值、计算结果（列式计算），不能只给结果。 3. 若 PDF 中缺少某参数，标注"文档中无此数据，无法计算"，绝不编造。 4. 全程使用简体中文输出。 5. 禁止输出 ANSI 转义序列、内部工具名、函数名。 6. 全流程使用 write_todos 工具实时更新任务进度。 ## 核验流程 1. 从 PDF 文本中提取每个用风地点的参数（名称、类型、瓦斯涌出量、人数、风速、断面积等） 2. 根据用风地点类型，调用对应的计算工具 3. 将配风计划中的需风量值与工具计算值进行对比 4. 按偏差判定：≤5% 一致，5%~10% 符合，>10% 不符 ## 输出要求 严格按照技能中定义的表格格式输出核验结果，每个地点一张表，最后汇总。 """ # 汇总 Agent（仅负责元信息生成：概况 + 违规汇总 + 修改建议） # 子智能体的详细审查内容（第2/3/4节）由系统直接嵌入报告，不经过 LLM，确保 100% 不丢失。 SUMMARY_SYSTEM_PROMPT = """你是一名煤矿配风计划审查主管。你的任务是根据子智能体的审查关键发现，生成最终审查报告的**元信息部分**。 ## ⚠️ 你的职责范围（重要） 你**只负责**以下3个部分： 1. **审查概况**：从子智能体发现中提炼整体结论 2. **违规项汇总**：从子智能体发现中提取并归类所有违规项 3. **修改建议**：针对违规项给出具体可操作的建议 ## ⚠️ 你不需要做的 - **不要**重复输出子智能体的详细审查内容（形式审查详情、数据对比表、计算核验表等）——这些由系统直接嵌入报告，你只需从关键发现中提炼结论。 - **不要**写"参见上文""详细内容见附件"等托词——你的输出是独立的元信息。 ## 汇总规则 1. 从子智能体的关键发现中提取：煤矿名称、计划月份、核心问题 2. 违规项分为"红线问题"（硬性违规）和"一般问题"（需整改） 3. 每个违规项必须对应至少一条修改建议 4. 全程使用简体中文输出 5. 禁止输出 ANSI 转义序列、内部工具名、函数名 ## 输出格式（严格遵循） 按以下顺序输出，从 ## 标题开始，不要加代码块标记： ## 一、审查概况 | 项目 | 内容 | |------|------| | 煤矿名称 | {从子智能体发现中提取} | | 计划月份 | {从子智能体发现中提取} | | 审查结论 | {通过/不通过/需整改}（一句话说明理由） | ## 五、违规项汇总 ### 红线问题 （逐条列出，含问题描述 + 严重程度 + 涉及的具体用风地点或数据项） ### 一般问题 （逐条列出，含问题描述 + 严重程度 + 涉及的具体用风地点或数据项） ## 六、修改建议 （逐条列出，每条对应上述一个违规项，给出具体、可操作的修改方案） """ # ============================================================ # Agent 创建函数 # ============================================================ # 全局 Agent 实例缓存 _form_review_agent: Optional[object] = None _data_consistency_agent: Optional[object] = None _calc_verify_agent: Optional[object] = None _summary_agent: Optional[object] = None def create_form_review_agent(): """创建「基本形式审查」Agent。 审查配风计划的基本形式合规性：版本、签字、编制时间、计算过程、语病逻辑。 使用运行时模型配置（支持前端动态切换）。 """ from deepagents import create_deep_agent from deepagents.backends import FilesystemBackend from deepagents import FilesystemPermission from api.model_config import get_model_instance skills = ["skills/vent-plan-review-form"] print(f"[skills] Agent=form-review-agent skills={skills}") agent = create_deep_agent( model=get_model_instance(), tools=[write_todos, get_current_time], skills=skills, system_prompt=FORM_REVIEW_SYSTEM_PROMPT, backend=FilesystemBackend(root_dir=PROJECT_ROOT, virtual_mode=True), permissions=[ FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"), ], middleware=[], name="form-review-agent", checkpointer=MemorySaver(), ) return agent def create_data_consistency_agent(): """创建「数据一致性审查」Agent。 审查配风计划数据的内部和外部一致性。 使用运行时模型配置（支持前端动态切换）。 """ from deepagents import create_deep_agent from deepagents.backends import FilesystemBackend from deepagents import FilesystemPermission from api.model_config import get_model_instance skills = ["skills/vent-plan-review-data"] print(f"[skills] Agent=data-consistency-agent skills={skills}") agent = create_deep_agent( model=get_model_instance(), tools=[ write_todos, get_current_time, query_mining_plan, query_face_procedure, query_gas_report, query_wind_report, calc_gas_emission_from_wind, ], skills=skills, system_prompt=DATA_CONSISTENCY_SYSTEM_PROMPT, backend=FilesystemBackend(root_dir=PROJECT_ROOT, virtual_mode=True), permissions=[ FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"), ], middleware=[], name="data-consistency-agent", checkpointer=MemorySaver(), ) return agent def create_calc_verification_agent(): """创建「计算核验」Agent。 逐地点核验需风量计算过程，使用本地计算工具。 使用运行时模型配置（支持前端动态切换）。 """ from deepagents import create_deep_agent from deepagents.backends import FilesystemBackend from deepagents import FilesystemPermission from api.model_config import get_model_instance skills = ["skills/vent-plan-review-calc"] print(f"[skills] Agent=calc-verification-agent skills={skills}") agent = create_deep_agent( model=get_model_instance(), tools=[ write_todos, calc_face_by_gas, calc_face_by_workers, calc_face_by_wind_speed, calc_face_air_volume_max, calc_tunnel_by_gas, calc_tunnel_by_explosives, calc_tunnel_by_workers, calc_tunnel_by_wind_speed, calc_tunnel_by_vehicle, calc_tunnel_air_volume_max, calc_chamber_by_equipment, calc_chamber_by_wind_speed, calc_other_by_wind_speed, calc_effective_area, calc_total_air_volume, ], skills=skills, system_prompt=CALC_VERIFY_SYSTEM_PROMPT, backend=FilesystemBackend(root_dir=PROJECT_ROOT, virtual_mode=True), permissions=[ FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"), ], middleware=[], name="calc-verification-agent", checkpointer=MemorySaver(), ) return agent def create_summary_agent(): """创建「汇总审查」Agent。 汇总3个子智能体的审查结果，生成最终审查报告。 使用运行时模型配置（支持前端动态切换）。 """ from deepagents import create_deep_agent from deepagents.backends import FilesystemBackend from deepagents import FilesystemPermission from api.model_config import get_model_instance skills = ["skills/vent-plan-review-summary"] print(f"[skills] Agent=summary-agent skills={skills}") agent = create_deep_agent( model=get_model_instance(), tools=[write_todos], skills=skills, system_prompt=SUMMARY_SYSTEM_PROMPT, backend=FilesystemBackend(root_dir=PROJECT_ROOT, virtual_mode=True), permissions=[ FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"), ], middleware=[], name="summary-agent", checkpointer=MemorySaver(), ) return agent def get_form_review_agent(): """获取形式审查 Agent 单例""" global _form_review_agent if _form_review_agent is None: _form_review_agent = create_form_review_agent() return _form_review_agent def get_data_consistency_agent(): """获取数据一致性审查 Agent 单例""" global _data_consistency_agent if _data_consistency_agent is None: _data_consistency_agent = create_data_consistency_agent() return _data_consistency_agent def get_calc_verification_agent(): """获取计算核验 Agent 单例""" global _calc_verify_agent if _calc_verify_agent is None: _calc_verify_agent = create_calc_verification_agent() return _calc_verify_agent def get_summary_agent(): """获取汇总 Agent 单例""" global _summary_agent if _summary_agent is None: _summary_agent = create_summary_agent() return _summary_agent def invalidate_cache(): """失效所有审查 Agent 缓存（模型切换时调用）。""" global _form_review_agent, _data_consistency_agent, _calc_verify_agent, _summary_agent _form_review_agent = None _data_consistency_agent = None _calc_verify_agent = None _summary_agent = None print("[模型切换] 审查 Agent 缓存已全部失效") # ============================================================ # SSE 事件辅助函数 # ============================================================ def _sse_event(event_type: str, **kwargs) -> str: """生成 SSE 格式的事件字符串。 Args: event_type: 事件类型 (progress / agent_start / agent_done / agent_error / token / done / error) **kwargs: 事件的其他字段 Returns: SSE 格式字符串: "data: {json}\n\n" """ data = {"type": event_type, **kwargs} return f"data: {json.dumps(data, ensure_ascii=False)}\n\n" def _extract_agent_response(result) -> str: """从 Agent 调用结果中提取文本响应。 Args: result: agent.ainvoke() 的返回值 Returns: str: 提取的文本内容 """ if result is None: return "" # 尝试从 messages 中提取最后一条 AI 消息 if isinstance(result, dict): messages = result.get("messages", []) if messages: # 从后往前找最后一条 AI 消息 for msg in reversed(messages): if hasattr(msg, "content") and hasattr(msg, "type") and msg.type == "ai": content = msg.content if isinstance(content, str): return content elif isinstance(content, list): return "".join( block.get("text", "") if isinstance(block, dict) else str(block) for block in content ) elif isinstance(msg, dict) and msg.get("type") == "ai": return msg.get("content", "") return str(result) # ============================================================ # 核心编排器 - stream_review() # ============================================================ async def stream_review( file, message: Optional[str] = None, session_id: Optional[str] = None, force_refresh: bool = False, ) -> AsyncGenerator[str, None]: """配风计划审查 SSE 流式编排器。 流程： 1. 保存上传的 PDF 文件并计算哈希 2. 检查缓存 → 提取文本（或从缓存读取） 3. 创建3个审查 Agent（或获取缓存单例） 4. 使用 asyncio.Queue 并行运行3个 Agent，收集进度事件 5. 流式 yield SSE 事件到前端 6. 汇总3个审查结果，生成最终报告 Args: file: FastAPI UploadFile 对象 message: 用户附加消息（可选） session_id: 会话ID（可选） force_refresh: 是否强制重新提取（忽略缓存），默认 False Yields: SSE 格式字符串 """ # 生成会话ID if session_id is None: session_id = str(uuid.uuid4()) # 确保会话存在于数据库，并设置标题 user_msg = message or "请对配风计划进行审查" create_session(session_id, title=f"配风计划审查: {file.filename or '未知文件'}") save_message(session_id, "user", user_msg) pdf_path = None try: overall_start = time.time() # ============================================================ # 阶段1: 保存文件并提取 PDF 文本 # ============================================================ yield _sse_event("progress", message="正在保存上传的 PDF 文件...", agent="system") pdf_path = await save_upload_file(file) # 计算文件哈希，用于缓存 key loop = asyncio.get_event_loop() file_hash = await loop.run_in_executor(None, calculate_file_hash, pdf_path) yield _sse_event("progress", message=f"PDF 文件已保存 ({os.path.basename(pdf_path)})，文件标识: {file_hash[:12]}...，正在提取文本内容...", agent="system", hash_id=file_hash) # 检查缓存状态（用于前端提示） cache_path = Path(PROJECT_ROOT) / "data" / "pdf_cache" / f"{file_hash}.txt" cache_hit = cache_path.exists() and not force_refresh if cache_hit: yield _sse_event("progress", message=f"📦 缓存命中！该 PDF 已解析过，直接从缓存加载...", agent="system", hash_id=file_hash, cache_hit=True) # 创建进度队列：OCR 线程通过 call_soon_threadsafe 写入，主循环轮询读取 progress_queue: asyncio.Queue = asyncio.Queue() def on_ocr_progress(page_num: int, total_pages: int): """OCR 页码回调（线程中执行 → 桥接到异步）""" try: loop.call_soon_threadsafe( progress_queue.put_nowait, {"page": page_num, "total": total_pages} ) except Exception: pass # 在 executor 中运行 PDF 提取（不阻塞事件循环），同时轮询进度队列 yield SSE extract_task = loop.run_in_executor( None, lambda: extract_pdf_text( pdf_path, force_refresh=force_refresh, progress_callback=on_ocr_progress, ) ) # 轮询：每 0.3 秒检查提取是否完成 + 是否有新页码事件 while not extract_task.done(): try: while True: evt = progress_queue.get_nowait() if evt is not None: yield _sse_event( "progress", message=f"文字识别正在处理第 {evt['page']}/{evt['total']} 页...", agent="system", hash_id=file_hash, ocr_progress={"page": evt["page"], "total": evt["total"]}, ) except asyncio.QueueEmpty: pass await asyncio.sleep(0.3) pdf_text = await extract_task if pdf_text.startswith("[错误]"): yield _sse_event("error", message=pdf_text, hash_id=file_hash) yield _sse_event("done", session_id=session_id) return yield _sse_event("progress", message=f"PDF 文本提取完成，共 {len(pdf_text)} 字符。正在初始化审查智能体...", agent="system", hash_id=file_hash, cache_hit=cache_hit) # 从 PDF 文本中提取煤矿名称，用于报告文件名 mine_name = await loop.run_in_executor(None, extract_mine_name, pdf_text) if mine_name: print(f"[审查] 识别煤矿名称: {mine_name}") # ============================================================ # 阶段2: 创建/获取3个审查 Agent # ============================================================ # 在 executor 中运行 Agent 创建（create_deep_agent 编译状态图可能耗时） agent1 = await loop.run_in_executor(None, get_form_review_agent) agent2 = await loop.run_in_executor(None, get_data_consistency_agent) agent3 = await loop.run_in_executor(None, get_calc_verification_agent) yield _sse_event("progress", message="3个审查智能体已就绪，正在并行执行审查...", agent="system") # ============================================================ # 阶段3: 并行执行3个 Agent # ============================================================ # 构建各 Agent 的输入消息 user_message = message or "请对配风计划进行审查" input_msg = { "messages": [ {"role": "user", "content": f"用户指令：{user_message}\n\n以下是配风计划PDF文件提取的文本内容，请按照技能中的步骤执行审查任务：\n\n{pdf_text}"} ] } # 使用 asyncio.Queue 收集进度事件 queue: asyncio.Queue = asyncio.Queue() async def run_agent(agent_name: str, agent, agent_input: dict): """在后台流式运行单个 Agent，将所有中间事件推送到队列。 事件类型: - agent_start: 开始审查 - agent_thinking: LLM 正在推理 - agent_executing: 正在执行工具（含工具名列表 + 中文名） - agent_todos: write_todos 更新了任务列表 - agent_tool_result: 工具执行完成（含中文名） - agent_done: 审查完成（含完整回复文本 + 耗时） - agent_error: 审查出错 """ agent_cn = _cn_agent(agent_name) t0 = time.time() try: await queue.put({ "type": "agent_start", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」开始审查...", }) full_response = "" async for chunk in agent.astream( agent_input, stream_mode=["updates", "messages", "custom"], version="v2", ): # ── updates 模式：步骤级事件 ── if chunk["type"] == "updates": for node_name in chunk["data"]: if node_name == "model": await queue.put({ "type": "agent_thinking", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」正在分析...", }) elif node_name == "tools": tools_data = chunk["data"].get(node_name, {}) tool_names = [] for msg in tools_data.get("messages", []): name = getattr(msg, "name", None) or (msg.get("name") if isinstance(msg, dict) else None) if name: tool_names.append(name) if tool_names: descs = [_cn_tool(t) for t in tool_names] await queue.put({ "type": "agent_executing", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」正在{'、'.join(descs)}...", "tools": tool_names, "cn_tools": descs, }) # ── messages 模式：token / tool_result / todo 事件 ── elif chunk["type"] == "messages": token_data = chunk["data"] if isinstance(token_data, (list, tuple)) and len(token_data) >= 1: msg_obj = token_data[0] else: msg_obj = token_data # 工具结果 is_tool_msg = hasattr(msg_obj, "type") and msg_obj.type == "tool" if is_tool_msg: tool_name = getattr(msg_obj, "name", "unknown") cn_tool = _cn_tool(tool_name) if tool_name == "write_todos": tool_content = _extract_text_content(msg_obj) or "" todo_list = _parse_todo_json(tool_content) if todo_list: await queue.put({ "type": "agent_todos", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」已更新任务进度", "todos": todo_list, }) else: await queue.put({ "type": "agent_tool_result", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」{cn_tool} 完成", "tool": tool_name, "cn_tool": cn_tool, }) else: # AI 生成的文本 token content = _extract_text_content(msg_obj) if content and isinstance(content, str): full_response += content duration_ms = int((time.time() - t0) * 1000) await queue.put({ "type": "agent_done", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」审查完成（{duration_ms}ms）", "content": full_response, "duration_ms": duration_ms, }) except Exception as e: traceback.print_exc() await queue.put({ "type": "agent_error", "agent": agent_name, "cn_agent": agent_cn, "message": f"「{agent_cn}」审查出错: {str(e)}", "error": str(e), }) # 并行启动3个 Agent tasks = [ asyncio.create_task(run_agent("form-reviewer", agent1, input_msg)), asyncio.create_task(run_agent("data-checker", agent2, input_msg)), asyncio.create_task(run_agent("calc-verifier", agent3, input_msg)), ] # 从队列读取事件并 yield SSE completed = 0 results = {} while completed < 3: event = await queue.get() if event["type"] == "agent_start": yield _sse_event("agent_start", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"]) elif event["type"] == "agent_thinking": yield _sse_event("agent_thinking", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"]) elif event["type"] == "agent_executing": yield _sse_event("agent_executing", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"], tools=event.get("tools", []), cn_tools=event.get("cn_tools", event.get("tools", []))) elif event["type"] == "agent_todos": yield _sse_event("agent_todos", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"], todos=event.get("todos", [])) elif event["type"] == "agent_tool_result": yield _sse_event("agent_tool_result", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"], tool=event.get("tool", ""), cn_tool=event.get("cn_tool", event.get("tool", ""))) elif event["type"] == "agent_done": completed += 1 results[event["agent"]] = event["content"] preview = event["content"][:300] + "..." if len(event["content"]) > 300 else event["content"] yield _sse_event("agent_done", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"], preview=preview, duration_ms=event.get("duration_ms", 0), progress=f"{completed}/3") elif event["type"] == "agent_error": completed += 1 results[event["agent"]] = f"[审查出错] {event.get('error', '未知错误')}" yield _sse_event("agent_error", agent=event["agent"], cn_agent=event.get("cn_agent", event["agent"]), message=event["message"], progress=f"{completed}/3") # 等待所有 task 完成 await asyncio.gather(*tasks, return_exceptions=True) # ============================================================ # 阶段4: 汇总结果（新架构：子智能体输出直接嵌入 + LLM 仅生成元信息） # # 旧架构问题：把 3 个子智能体全量输出（可达 50k+ chars）传给 flash 模型 # 要求"全量重现"，模型偶尔摆烂只输出一句托词 → 子智能体结果全部丢失。 # # 新架构：子智能体原始输出直接嵌入报告第 2/3/4 节（100% 不丢失）； # LLM 仅接收关键摘要（~6k chars），负责生成第 1 节（概况）、 # 第 5 节（违规汇总）、第 6 节（修改建议）。 # ============================================================ yield _sse_event("progress", message="3个子智能体均已审查完成，正在汇总审查结果...", agent="system") # ── 获取子智能体原始完整输出 ── raw_form = results.get("form-reviewer", "无结果") raw_data = results.get("data-checker", "无结果") raw_calc = results.get("calc-verifier", "无结果") print(f"[审查] 子智能体结果长度: form={len(raw_form)}, " f"data={len(raw_data)}, calc={len(raw_calc)}") # ── Step 1: 提取关键摘要（供 LLM 综合判断，每份最多 2000 字符）── key_form = _extract_key_findings(raw_form) key_data = _extract_key_findings(raw_data) key_calc = _extract_key_findings(raw_calc) total_key_chars = len(key_form) + len(key_data) + len(key_calc) print(f"[审查] 关键摘要总长度: {total_key_chars} 字符 " f"(原始: {len(raw_form) + len(raw_data) + len(raw_calc)} 字符)") yield _sse_event("progress", message="正在生成审查概况与违规汇总...", agent="system") # ── Step 2: 调用 summary LLM 生成元信息（概况 + 违规汇总 + 修改建议）── # 只传关键摘要，不传全量子智能体输出，大幅降低上下文大小， # 从根本上避免 flash 模型因上下文过大而"摆烂"。 meta_prompt = f"""请根据以下3个子智能体的审查关键发现，生成最终审查报告的元信息部分。 ## 审查关键发现 ### 基本形式审查（关键发现）： {key_form} ### 数据一致性审查（关键发现）： {key_data} ### 计算核验审查（关键发现）： {key_calc} ## 任务 请严格按照系统提示词（system prompt）中定义的格式，输出以下3个部分 （只输出这3个部分，不要输出代码块标记，不要输出子智能体的详细审查内容）： ## 一、审查概况 （表格形式：煤矿名称、计划月份、审查结论 + 一句话理由） ## 五、违规项汇总 （红线问题 + 一般问题，逐条列出，含涉及的具体用风地点） ## 六、修改建议 （逐条列出，每条对应一个违规项，给出具体可操作的修改方案） """ summary_agent = await loop.run_in_executor(None, get_summary_agent) meta_input = { "messages": [{"role": "user", "content": meta_prompt}] } # 流式获取 LLM 元信息输出（带心跳，防止长时间等待无反馈） meta_text = "" try: import asyncio as _asyncio_mod astream_iter = summary_agent.astream( meta_input, stream_mode=["updates", "messages"], version="v2", ).__aiter__() first_token_received = False heartbeat_start = time.time() while True: try: if first_token_received: chunk = await astream_iter.__anext__() else: chunk = await _asyncio_mod.wait_for( astream_iter.__anext__(), timeout=5.0 ) if not first_token_received: first_token_received = True elapsed = int(time.time() - heartbeat_start) if elapsed >= 5: print(f"[审查] Summary LLM 首 token 延迟: {elapsed}s") except _asyncio_mod.TimeoutError: elapsed = int(time.time() - heartbeat_start) yield _sse_event("progress", message=f"模型正在处理中（已等待 {elapsed} 秒）...", agent="system") continue except StopAsyncIteration: break if chunk["type"] == "messages": token_data = chunk["data"] if isinstance(token_data, (list, tuple)) and len(token_data) >= 1: msg_obj = token_data[0] else: msg_obj = token_data is_tool_msg = hasattr(msg_obj, "type") and msg_obj.type == "tool" if not is_tool_msg: content = _extract_text_content(msg_obj) if content and isinstance(content, str): meta_text += content except Exception as e: print(f"[审查] Summary LLM 调用出错: {e}") meta_text = "" # ── Step 3: 校验 LLM 输出质量 ── lazy_detected = _is_lazy_summary(meta_text) if lazy_detected: print(f"[审查] ⚠️ 检测到 LLM 懒输出（长度={len(meta_text)}），将使用兜底模板") else: print(f"[审查] Summary LLM 元信息生成完成，长度={len(meta_text)}") # ── Step 4: 解析 LLM 输出中的三个部分 ── sec1 = "" # 审查概况 sec5 = "" # 违规项汇总 sec6 = "" # 修改建议 if meta_text and not lazy_detected: # 清理前导语：LLM 可能输出"好的，以下是..."等前缀 import re as _re anchor_match = _re.search(r"## 一、审查概况", meta_text) if anchor_match and anchor_match.start() > 0: skipped = meta_text[:anchor_match.start()].strip() if skipped: print(f"[审查] 已跳过 LLM 前导语 ({len(skipped)} 字符): {skipped[:80]}...") meta_text = meta_text[anchor_match.start():] parts_1 = meta_text.split("## 五、违规项汇总", 1) sec1 = parts_1[0].strip() if len(parts_1) > 1: parts_2 = parts_1[1].split("## 六、修改建议", 1) sec5 = ("## 五、违规项汇总\n" + parts_2[0]).strip() if len(parts_2) > 1: sec6 = ("## 六、修改建议\n" + parts_2[1]).strip() # ── Step 5: 流式组装最终报告并 yield SSE token 事件 ── mine_title = mine_name or "" report_title = ( f"# {mine_title}配风计划审查报告\n\n" if mine_title else "# 配风计划审查报告\n\n" ) yield _sse_event("token", content=report_title) # 收集各部分用于最终组装 full_report（DB 存储 + Word 导出） report_parts = [report_title] # 第1节：审查概况（LLM 生成 or 兜底） if sec1: yield _sse_event("token", content=sec1 + "\n\n") report_parts.append(sec1 + "\n\n") else: fbk1 = ( "## 一、审查概况\n\n" f"| 项目 | 内容 |\n|------|------|\n" f"| 煤矿名称 | {mine_title or '（待补充）'} |\n" "| 审查结论 | 详见以下各节审查详情 |\n\n" ) yield _sse_event("token", content=fbk1) report_parts.append(fbk1) # 第2节：形式审查结果（子智能体原始输出直接嵌入，分块传输避免前端卡顿） yield _sse_event("token", content="## 二、形式审查结果\n\n") report_parts.append("## 二、形式审查结果\n\n") for chunk_event in _chunk_text_for_sse(raw_form): yield chunk_event report_parts.append(raw_form) report_parts.append("\n\n") # 第3节：数据一致性审查结果（直接嵌入） yield _sse_event("token", content="## 三、数据一致性审查结果\n\n") report_parts.append("## 三、数据一致性审查结果\n\n") for chunk_event in _chunk_text_for_sse(raw_data): yield chunk_event report_parts.append(raw_data) report_parts.append("\n\n") # 第4节：计算核验结果（直接嵌入，通常是最大的一块内容） yield _sse_event("token", content="## 四、计算核验结果\n\n") report_parts.append("## 四、计算核验结果\n\n") for chunk_event in _chunk_text_for_sse(raw_calc): yield chunk_event report_parts.append(raw_calc) report_parts.append("\n\n") # 第5节：违规项汇总（LLM 生成 or 兜底） if sec5: yield _sse_event("token", content=sec5 + "\n\n") report_parts.append(sec5 + "\n\n") else: fbk5 = ( "## 五、违规项汇总\n\n" "> ⚠ 。" "请参见以上各节中标注的不符/缺失/违规/红线等标记，逐一核实。\n\n" "（请根据以上各节中标注的具体问题进行人工汇总）\n\n" ) yield _sse_event("token", content=fbk5) report_parts.append(fbk5) # 第6节：修改建议（LLM 生成 or 兜底） if sec6: yield _sse_event("token", content=sec6) report_parts.append(sec6) else: fbk6 = ( "## 六、修改建议\n\n" "> ⚠️ 自动生成失败。" "请参见以上各节中各子智能体给出的具体审查意见和修改建议。\n" ) yield _sse_event("token", content=fbk6) report_parts.append(fbk6) # ── 组装最终完整报告（用于 DB 存储和 Word 导出）── full_report = "".join(report_parts) print(f"[审查] 最终报告总长度: {len(full_report)} 字符 " f"(子智能体内容: {len(raw_form) + len(raw_data) + len(raw_calc)} 字符, " f"元信息: {len(meta_text)} 字符, " f"LLM状态: {'懒输出-已兜底' if lazy_detected else '正常'})") # ============================================================ # 阶段5: 生成 Word 文档 + 下载链接 # ============================================================ yield _sse_event("progress", message="正在生成 Word 审查报告...", agent="system") download_url = None if full_report.strip(): # ── 保存审查报告到数据库（历史查询）── try: save_message(session_id, "assistant", full_report) # 更新会话标题为煤矿名称（更易识别） if mine_name: update_session_title(session_id, f"配风计划审查: {mine_name}") print(f"[审查] 报告已保存到会话 {session_id}") except Exception as e: print(f"[审查] 保存报告到数据库失败: {e}") try: # 在 executor 中运行 docx 转换（pypandoc 是同步的） # 使用煤矿名称作为文件名标题 _, download_url = await loop.run_in_executor( None, lambda: save_review_report(full_report, title=mine_name), ) print(f"[报告] Word 文档已生成，下载URL: {download_url}") yield _sse_event("progress", message=f"Word 报告已生成，点击下载", agent="system", download_url=download_url) except Exception as e: import traceback print(f"[报告] Word 生成失败: {e}") traceback.print_exc() yield _sse_event("progress", message=f"Word 报告生成失败（{e}），但审查文本已完整生成", agent="system") # ============================================================ # 阶段6: 完成 # ============================================================ yield _sse_event("done", session_id=session_id, download_url=download_url, duration_ms=int((time.time() - overall_start) * 1000)) # 打印完整报告到控制台 print("\n" + "=" * 60) print("配风计划审查完成") print("=" * 60) print(full_report) print("=" * 60) except Exception as e: traceback.print_exc() yield _sse_event("error", message=f"审查过程出错: {str(e)}") yield _sse_event("done", session_id=session_id) finally: # 清理临时文件 if pdf_path and os.path.exists(pdf_path): try: os.unlink(pdf_path) except OSError: pass def _extract_text_content(msg_obj) -> str | None: """从 LangChain 消息对象中提取文本内容。""" if isinstance(msg_obj, dict): return msg_obj.get("content") elif hasattr(msg_obj, "content"): raw = getattr(msg_obj, "content", None) if isinstance(raw, str): return raw elif isinstance(raw, list): parts = [] for block in raw: if isinstance(block, dict) and block.get("type") == "text": parts.append(block.get("text", "")) elif hasattr(block, "type") and getattr(block, "type", "") == "text": parts.append(getattr(block, "text", "")) return "".join(parts) if parts else None return None def _parse_todo_json(text: str) -> list | None: """从 write_todos 输出文本中提取 todo 列表。 write_todos 输出格式: "Updated todo list to [{'content': '...', 'status': '...'}, ...]" 返回 JSON-serializable list of dicts，失败返回 None。 """ import re import ast match = re.search(r"\[.*\]", text, re.DOTALL) if not match: return None try: python_list = ast.literal_eval(match.group()) if isinstance(python_list, list): return python_list except (ValueError, SyntaxError): pass return None # ============================================================ # 汇总输出质量保障（防止 LLM 懒输出/内容丢失） # ============================================================ def _is_lazy_summary(text: str) -> bool: """检测 LLM 汇总输出是否为懒输出（内容过短或仅含托词）。 懒输出的典型特征： - 总长度极短（< 100 字符）且无结构化内容 - 包含"完整呈现/全量输出/无省略"等托词但实际内容极少 注意：阈值不能太高，因为 LLM 可能输出短但有效的元信息（100-300 字符的概况+违规+建议）。 Returns: True 若检测到懒输出 """ if not text or not text.strip(): return True stripped = text.strip() # 极短（< 80 字符）：几乎不可能是有效报告 if len(stripped) < 80: return True # 匹配典型托词模式：LLM 输出了"完整呈现"之类的敷衍话 import re lazy_patterns = [ r"以上为本次.*完整.*(?:报告|审查)", r"所有.*子智能体.*已.*(?:完整|全量).*呈现", r"全量.*逐条.*逐表.*完整", r"以上.*审查.*完成[。.]?$", r"完整.*最终报告.*无.*(?:省略|摘要|删减)", r"详细内容.*见.*(?:附件|原始输出|子智能体)", ] for pattern in lazy_patterns: if re.search(pattern, stripped): # 匹配了托词模式：如果文本 < 500 字符，大概率是纯敷衍 if len(stripped) < 500: return True # 即使 > 500 字符，如果托词占据了 > 30% 的内容，也很可疑 # （正常输出的结尾语不会这么长） match = re.search(pattern, stripped) if match and len(match.group()) > len(stripped) * 0.3: return True # 检查内容密度：如果文本中实质性内容行极少（< 3 行非标题/分隔符内容）， # 且总长度 < 200 字符，判定为懒输出 lines = [l.strip() for l in stripped.split("\n")] content_lines = [ l for l in lines if l and not l.startswith("#") and not l.startswith("---") and not l.startswith("|") # 表格分隔行不算实质性内容 and l not in ("---", "|---|---", "") ] if len(content_lines) < 3 and len(stripped) < 200: return True return False def _extract_key_findings(text: str, max_chars: int = 2000) -> str: """从子智能体完整输出中提取关键发现摘要。 策略：取前 max_chars 字符（子智能体通常在开头给出审查结论）， 并额外提取后续包含审查关键词的重要行。 Args: text: 子智能体的完整输出文本 max_chars: 基础摘要的最大字符数 Returns: 压缩后的关键发现摘要（远小于原文，但保留核心结论） """ if not text: return "（无内容）" # 清理换行 text = text.replace("\r\n", "\n").replace("\r", "\n") if len(text) <= max_chars: return text head = text[:max_chars] # 额外提取包含关键词的重要行（从 max_chars 之后的部分） keywords = [ "问题", "不符", "缺失", "错误", "偏差", "违规", "红线", "结论", "判定", "审查意见", "不一致", ] important_lines: list = [] for line in text[max_chars:].split("\n"): stripped = line.strip() if any(kw in stripped for kw in keywords) and len(stripped) > 8: important_lines.append(stripped) if len(important_lines) >= 15: # 最多收集 15 行 break if important_lines: head += "\n\n...[后续关键发现]...\n" + "\n".join(important_lines) return head def _assemble_report_fallback( raw_form: str, raw_data: str, raw_calc: str, mine_name: str = "", ) -> str: """LLM 懒输出时的兜底方案：用模板组装完整报告。 子智能体原始输出直接嵌入第 2/3/4 节，元信息用简单模板填充。 确保即使 LLM 完全失败，审查报告的实质内容也 100% 不丢失。 Args: raw_form: 形式审查子智能体完整输出 raw_data: 数据一致性审查子智能体完整输出 raw_calc: 计算核验子智能体完整输出 mine_name: 煤矿名称（可选） Returns: 完整的 Markdown 审查报告 """ mine_str = mine_name or "（待补充）" title = f"# {mine_str}配风计划审查报告" if mine_name else "# 配风计划审查报告" report = f"""{title} ## 一、审查概况 | 项目 | 内容 | |------|------| | 煤矿名称 | {mine_str} | | 审查结论 | 请参见以下各节中3个子智能体的逐项审查详情 | ## 二、形式审查结果 {raw_form} ## 三、数据一致性审查结果 {raw_data} ## 四、计算核验结果 {raw_calc} ## 五、违规项汇总 > ⚠️ LLM 自动汇总生成失败，以下为系统自动兜底。请参见以上各节中标注的"不符""缺失""违规""红线"等标记，逐一核实。 （请根据以上各节中标注的具体问题进行人工汇总） ## 六、修改建议 > ⚠️ LLM 自动生成失败。请参见以上各节中各子智能体给出的具体审查意见和修改建议。 """ return report def _chunk_text_for_sse(text: str, chunk_size: int = 2000): """将大文本分块，逐块 yield SSE token 事件。 用于子智能体原始输出直接嵌入时避免单次 SSE 事件过大导致前端卡顿。 Args: text: 要分块的文本 chunk_size: 每块最大字符数 Yields: SSE 事件字符串 """ for i in range(0, len(text), chunk_size): yield _sse_event("token", content=text[i:i + chunk_size])