vent_agent_2.0/extensions/main_fan/tools.py

"""历史固定服务的兼容工具，当前 Agent 默认注册 exploration_tools 中的自主工具。"""
from typing import Any
from langchain.tools import ToolRuntime

from extensions import define_tool
from extensions.main_fan.schemas import MainFanResult


@define_tool(group="主通风机", label="检查主通风机材料目录")
def inspect_main_fan_materials(materials: list[dict[str, Any]], mine_code: str = "",
                               mine_name: str = "", fan_id: str = "") -> dict:
    """仅检查材料目录。实际初步筛查使用 analyze_main_fan，不能以目录检查代替业务研判。"""
    supplied = bool(materials)
    return MainFanResult(
        status="needs_materials",
        mine_code=mine_code, mine_name=mine_name,
        check_object=f"矿井主通风机 {fan_id}".strip(),
        result_title="主通风机材料检查",
        result=("已接收材料目录，需要通过 analyze_main_fan 提取、核验与初步筛查。仅目录检查无法形成正式研判结论。"
                if supplied else "尚未提供主通风机资料，无法形成正式研判结论。"),
        check_process="仅检查材料目录是否为空，未核验文件内容或运行参数。",
        missing_materials=[] if supplied else ["主通风机相关报告或运行资料"],
    ).model_dump(mode="json")


@define_tool(group="主通风机", label="生成主通风机与备用风机一致性报告")
async def analyze_main_fan(data: dict[str, Any] | str, runtime: ToolRuntime) -> dict:
    """执行主通风机与备用风机一致性七阶段流程，返回不可改写的结构化原件。

    简短请求的data只需原目标文字或{"target":"众源煤业1号主通风机"}；服务端解析身份、网关取数与识别策略。
    完整data仍支持MainFanInput：mine_code/mine_name，fan_id/fan_name，devices/points/reports，
    materials 为当前会话 /workspace/ 文件路径；use_gateway=true 从已配置网关取数。
    命名关联只产生候选，confirmed_bindings 仅录入用户实际提供的人工核验，禁止虚构。
    不要猜测身份或向用户索要已配置的编码；未唯一匹配时返回待补充的业务对象信息。
    completed 表示筛查过程完成，不表示工程合格。
    """
    from extensions.main_fan.service import analyze
    from extensions.main_fan.targets import resolve_request, TargetResolutionError
    identity = runtime.config.get("configurable", {})
    if not identity.get("workspace_user") or not identity.get("workspace_session"):
        raise ValueError("业务工具缺少服务端会话身份")
    # 队列任务以服务端校验的输入为准，模型不能添加虚构测点或人工确认。
    trusted = identity.get("business_task_input") if identity.get("business_agent") == "main_fan" else None
    try:
        resolved = resolve_request(trusted if trusted is not None else data)
    except TargetResolutionError as exc:
        if trusted is not None:
            raise
        return MainFanResult(status="needs_materials", check_object="矿井主通风机",
            result_title="请明确研判对象", result=str(exc), check_process="目标身份解析未完成，尚未获取资料或进行研判。",
            missing_materials=[str(exc)]).model_dump(mode="json")
    result = await analyze(resolved,
                           identity["workspace_user"], identity["workspace_session"], job_id=identity.get("business_task_id"),
                           document_audit_dir=identity.get("document_audit_dir"))
    if (result.get("analysis") or {}).get("document_audits") or result.get('workflow_id'):
        # 大报告无需让编排模型重写全部数字和证据，结果验证器从数据库恢复原件。
        result = {**result, "analysis": None, "assessment":None, "clues": []}
        result["platform_payload"] = None
        # 只交给编排模型一份接收凭据，最终正文从数据库原件恢复。
        result["result"] = "整体研判报告已生成，最终正文由服务端恢复。"
        result["missing_materials"] = []
    return result


def _component(workflow_id, runtime, name):
    from db.main_fan_store import workflow
    from extensions.main_fan.pipeline import component
    identity=runtime.config.get('configurable',{})
    if not identity.get('workspace_user') or not identity.get('workspace_session'):
        raise ValueError('阶段工具缺少服务端会话身份')
    return component(workflow(workflow_id,identity['workspace_user'],identity['workspace_session']),name)


@define_tool(group="主通风机阶段",label="读取已解析的风机证据文档")
def parse_fan_evidence_documents(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取本会话已执行流程的文档解析结果；新任务先调用analyze_main_fan实际取数识别。"""
    return _component(workflow_id,runtime,'parse_fan_evidence_documents')


@define_tool(group="主通风机阶段",label="读取风机运行时序与切换记录")
def query_fan_operating_data(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取已接入的原始运行时序、完整风机状态和切换记录；不生成默认数据。"""
    return _component(workflow_id,runtime,'query_fan_operating_data')


@define_tool(group="主通风机阶段",label="读取双机身份和参数标准化结果")
def normalize_and_validate_fan_data(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取同会话流程的双机身份、单位和基础参数比较结果。"""
    return _component(workflow_id,runtime,'normalize_and_validate_fan_data')


@define_tool(group="主通风机阶段",label="读取两台风机性能数据差异")
def analyze_fan_performance_curve(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取受控插值及曲线差异指标，区分同角度同频率可比与仅同频数据比较。"""
    return _component(workflow_id,runtime,'analyze_fan_performance_curve')


@define_tool(group="主通风机阶段",label="读取切换前后稳定工况分析")
def analyze_fan_switch_event(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取实际切换事件、前后稳定窗口、变化指标和恢复时间。"""
    return _component(workflow_id,runtime,'analyze_fan_switch_event')


@define_tool(group="主通风机阶段",label="读取适用规则和阈值版本")
def match_fan_rules_and_knowledge(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取本次已冻结的规则、阈值版本与各判定项；不以模型记忆改变阈值。"""
    return _component(workflow_id,runtime,'match_fan_rules_and_knowledge')


@define_tool(group="主通风机阶段",label="读取主备风机综合一致性结果")
def assess_fan_equivalent_capability(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取固定规则形成的综合结果，未评估项目不能自动视为通过。"""
    return _component(workflow_id,runtime,'assess_fan_equivalent_capability')


@define_tool(group="主通风机阶段",label="读取结论与原始证据对应关系")
def trace_fan_evidence_and_review(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取结果、参数、公式和原文件页码的证据索引及版本信息。"""
    return _component(workflow_id,runtime,'trace_fan_evidence_and_review')


@define_tool(group="主通风机阶段",label="读取整体主备风机一致性报告")
def render_fan_assessment_report(workflow_id: str, runtime: ToolRuntime) -> dict:
    """读取已生成的整体报告，正文仅包括实际参与比较的项目。"""
    return _component(workflow_id,runtime,'render_fan_assessment_report')


@define_tool(group="主通风机",label="从指定阶段重跑主备风机研判")
async def resume_main_fan_assessment(workflow_id: str, start_stage: int, runtime: ToolRuntime) -> dict:
    """重跑1～7中的指定阶段及下游阶段。原文件未变化时复用本次实际识别的结构化事实，不再次OCR。"""
    from extensions.main_fan.service import resume_assessment
    identity=runtime.config.get('configurable',{})
    if not identity.get('workspace_user') or not identity.get('workspace_session'):
        raise ValueError('阶段工具缺少服务端会话身份')
    result=await resume_assessment(workflow_id,start_stage,identity['workspace_user'],identity['workspace_session'],job_id=identity.get('business_task_id'))
    return {**result,'result':'整体报告已按指定阶段重新生成，正文由服务端恢复。','analysis':None,'assessment':None,'clues':[],'platform_payload':None}