vent_deep_agent/agents/needq_agent.py

# -*- coding: utf-8 -*-
"""
需风量计算 Agent 创建与配置模块

核心功能：
- 创建需风量计算 DeepAgent（整合 Skills + 计算工具 + MCP 数据）
- 配置流式输出
- 管理 Agent 实例生命周期

架构：
- 模型: 通过 vent_agent._get_model() 获取
- Skills: needq-calc（需风量计算技能）
- Tools: 14个 calc_tools 计算函数 + write_todos + get_needq_all_data
- Backend: FilesystemBackend（virtual_mode）
- 系统提示词仅包含通用行为准则，具体计算规则在各技能的 SKILL.md 中
"""

import os
from pathlib import Path
from typing import Optional

from dotenv import load_dotenv

load_dotenv()

SKILLS_ROOT = str((Path(__file__).parent.parent / "skills").resolve())

# 复用现有项目的模型加载
from agents.vent_agent import _get_model

# 导入 write_todos
from langchain.agents.middleware.todo import write_todos

# 导入全部计算工具
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_air_volume_max,
    calc_chamber_by_equipment,
    calc_chamber_by_wind_speed,
    calc_other_by_wind_speed,
    calc_effective_area,
    calc_total_air_volume,
)

# 导入 MCP 工具
from tools.vent_tools import get_needq_all_data


# ============================================================
# 系统提示词（仅通用行为准则，不写计算规则）
# ============================================================

NEEDQ_CALC_SYSTEM_PROMPT = """你是一名煤矿通风需风量计算专家，帮助用户交互式计算各用风地点的需风量。

## 全局通用行为准则
1. **必须调用工具计算**：所有需风量计算必须通过调用 calc_tools 中的工具函数完成，绝对禁止凭 LLM 知识直接给出计算结果。此条为红线。
2. **参数缺失要追问**：若用户未提供必要计算参数，明确列出所需参数并引导用户补充，绝不编造参数值。
3. **列式计算**：展示每个计算过程的公式 → 代入数值 → 计算结果，不能只给结果。
4. **多轮对话承接**：记住当前会话中的用风地点类型和已有参数，用户补充参数或要求调整时自动衔接。
5. **全程简体中文输出**。
6. **禁止输出 ANSI 转义序列、内部工具名、函数名、技能标识**。
7. **全流程使用 write_todos 工具**实时更新任务进度。
8. 所有数据来源于工具调用结果，绝不编造数据。
9. 引用规程时附上具体条款来源（《煤矿安全规程》2025版、AQ 1056-2008）。
10. 无法判断时诚实说明原因，不臆测。

## 支持的用风地点类型
- 采煤工作面（含备用工作面）
- 掘进工作面
- 机电硐室
- 其他用风地点（主要进回风巷、采区进回风巷、其他通风人行巷道等）
- 多地点汇总
- 通防管控平台数据查询

## 工作模式
严格按照技能（skill: needq-calc）中定义的流程执行任务：
1. 识别用风地点类型
2. 收集/提取计算参数
3. 调用计算工具
4. 格式化输出结果
5. 支持参数调整重算
"""


# ============================================================
# Agent 工厂函数
# ============================================================

def create_needq_calc_agent():
    """创建「需风量计算」Agent。

    该 Agent 专门处理各用风地点的需风量交互式计算，具备：
    - 14个计算工具（覆盖采煤面/掘进面/硐室/其他巷道）
    - MCP 远程数据查询能力（get_needq_all_data）
    - 多轮对话参数承接能力
    - 结构化计算报告输出能力

    Returns:
        CompiledStateGraph: 编译后的 LangGraph 状态图，支持 .invoke() 和 .stream()
    """
    from deepagents import create_deep_agent
    from deepagents.backends import FilesystemBackend
    from deepagents import FilesystemPermission

    _project_root = str(Path(__file__).parent.parent)

    skills = ["skills/needq-calc"]
    print(f"[skills] Agent=needq-calc-agent skills={skills}")

    agent = create_deep_agent(
        model=_get_model(),
        tools=[
            # 基础工具
            write_todos,

            # 辅助工具
            calc_effective_area,
            calc_total_air_volume,

            # 采煤工作面计算
            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_air_volume_max,

            # 硐室计算
            calc_chamber_by_equipment,
            calc_chamber_by_wind_speed,

            # 其他巷道计算
            calc_other_by_wind_speed,

            # MCP 远程数据查询
            get_needq_all_data,
        ],
        skills=skills,
        system_prompt=NEEDQ_CALC_SYSTEM_PROMPT,
        backend=FilesystemBackend(root_dir=_project_root, virtual_mode=True),
        permissions=[
            FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"),
        ],
        middleware=[],
        name="needq-calc-agent",
    )
    return agent


# ============================================================
# 单例缓存
# ============================================================

_needq_calc_agent: Optional[object] = None


def get_needq_calc_agent():
    """获取需风量计算 Agent 单例"""
    global _needq_calc_agent
    if _needq_calc_agent is None:
        _needq_calc_agent = create_needq_calc_agent()
    return _needq_calc_agent