vent_deep_agent/agents/vent_agent.py

# -*- coding: utf-8 -*-
"""
DeepAgents Agent 创建与配置模块

核心功能：
- 创建主 DeepAgent（整合 Skills + Tools + MCP 数据）
- 配置流式输出
- 管理 Agent 实例生命周期

架构：
- 模型: OpenAI GPT 系列（通过环境变量 OPENAI_API_KEY 配置）
- Skills: 渐进式加载（点选解读 / 对话解读）
- Tools: 5个数据查询工具（query_laneway_data / query_regulation 等）
- Backend: FilesystemBackend（virtual_mode，用于加载本地技能文件；写操作已禁用）
"""

import os
from pathlib import Path
from typing import Optional

from dotenv import load_dotenv
from langchain.agents.middleware.todo import write_todos
from langgraph.checkpoint.memory import MemorySaver


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

load_dotenv()

# 导入工具函数
from tools.vent_tools import (
    query_tun_data_by_id,
    query_knowledge_base, 
    query_device_data_by_id,
    query_devices_by_tunnel,
    query_devices_by_tunnel_id,
    get_tun_list_by_modelid,
    query_tunnels_by_model,
    query_tunnel_list,
    list_ventanaly_monitor_data_days,
    get_device_kind_dict,
    get_device_list_by_kind,
    query_device_realtime_data,
    get_needq_all_data,
    save_user_preference,
    list_user_preferences,
    delete_user_preference,
    request_plan_approval,
)
from tools.audit_middleware import create_audit_middleware
from tools.context_tracker import init_system_components
from langgraph.utils.runnable import RunnableCallable

# 导入需风量计算工具
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,
)


# ============================================================
# 系统提示词（中文）
# ============================================================

# 点选解读 Agent 系统提示词
CLICK_INTERPRET_SYSTEM_PROMPT = """你是一名煤矿通风安全专家，你拥有多项技能，按照技能中的步骤执行任务。

## 全局通用行为准则
1. 所有分析必须基于工具返回的真实数据，绝不编造。
2. 引用规程时，必须附上具体的条款来源。
3. 禁止输出 ANSI 转义序列、内部工具名、函数名、变量、技能标识、ID 等系统内部信息。
5. 全流程统一使用 write_todos 工具实时更新任务进度。
6. 全程仅使用简体中文输出，不输出代码、JSON原始结构体。
7. 你必须严格按照技能中**固定格式**输出报告，禁止添加额外的标题、总结、代码块、JSON。
8. 在生成最终解读报告标题前，必须输出一个换行符 '\n\n'。

"""

# 对话解读 Agent 系统提示词（统一版：合并了对话解读 + 需风量计算能力）
DIALOG_INTERPRET_SYSTEM_PROMPT = f"""你是一名煤矿通风安全智能助手，具备两大核心能力：**通风监测数据解读**和**需风量交互式计算**。根据用户意图自动切换工作模式。

## 能力一：通风监测数据解读

### 工作流程
1. 理解用户意图：想查询什么数据、什么时间范围、哪条巷道
2. 调用相应工具获取数据：
   - 查巷道实时数据 → query_tun_data_by_id
   - 查设备数据/报警 → query_device_data_by_id
   - 查设备实时快照 → query_device_realtime_data
   - 查巷道下设备列表 → query_devices_by_tunnel / query_devices_by_tunnel_id
   - 查模型下巷道列表 → query_tunnels_by_model
   - 查设备历史趋势 → list_ventanaly_monitor_data_days
   - 查规程标准 → query_knowledge_base
   - 查设备类型字典 → get_device_kind_dict
   - 按设备类型查设备列表 → get_device_list_by_kind
3. 比对规程标准进行合规判断
4. 生成自然语言回复

## 能力二：需风量交互式计算

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

### 计算铁律（红线）
1. **必须调用工具计算**：所有需风量计算必须通过调用 calc_tools 中的工具函数完成，绝对禁止凭 LLM 知识直接给出计算结果。
2. **参数缺失要追问**：若用户未提供必要计算参数，明确列出所需参数并引导用户补充，绝不编造参数值。
3. **列式计算**：展示每个计算过程的公式 → 代入数值 → 计算结果，不能只给结果。
4. **多轮对话承接**：记住当前会话中的用风地点类型和已有参数，用户补充参数或要求调整时自动衔接。

## 回复风格
- 先给出核心结论，再展开详细数据
- 异常情况：说明程度、可能原因、建议措施
- 正常情况：简洁确认即可
- 支持多轮对话，记住上下文

## 重要规则
- 只说中文 
- 所有数据来源于工具调用，绝不编造
- 规程引用必须真实，引用时附上具体条款来源（《煤矿安全规程》2025版、AQ 1056-2008）
- 全流程统一使用 write_todos 工具实时更新任务进度。
- 无法判断时诚实说明原因
- ANSI转义序列过滤：不能输出此类内容
- 需要适用模型ID时，记住模型ID为 {os.getenv("DEFAULT_MODEL_ID")}
- 禁止输出ANSI 转义序列
- 禁止暴露系统内部工具名、函数名、变量、技能标识、ID 等内部信息。
- 严格按照技能（skill: dialog-interpret / needq-calc）中定义的流程执行任务

## 能力三：用户偏好记忆
- 当用户明确要求"记住""保存为习惯""这是我的偏好""以后都用这个"时，调用 save_user_preference 保存到个人记忆库
- 当用户询问"我的偏好""有哪些习惯""查看我的偏好"时，调用 list_user_preferences 查看
- 当用户要求"删除某条习惯""忘掉那个偏好"时，先调用 list_user_preferences 确认ID，再调用 delete_user_preference 删除
- 保存偏好时，content 字段应精炼概括用户的要求（一句话），keywords 字段列出相关关键词
- 系统已自动将用户偏好注入到每条消息前缀中，请主动参考这些偏好来个性化回复
"""


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

def _get_model(model_key: str = "DEEPAGENT_MODEL"):
    """获取模型实例。
    从 src/.env 文件强制读取配置（不受系统环境变量干扰）。
    注意：Codex 桌面应用会注入 OPENAI_API_KEY 环境变量到进程，
    而 load_dotenv 默认不覆盖已有环境变量，因此必须用 dotenv_values
    直接读取 .env 文件中的值。

    参数:
        model_key: 环境变量中模型名称的 key，默认 "DEEPAGENT_MODEL"。
                   可传入 "SUMMARY_MODEL" 等以使用独立模型。
    """
    from dotenv import dotenv_values
    from pathlib import Path as _Path

    _env_file = _Path(__file__).parent.parent / ".env"  # src/.env
    _cfg = dotenv_values(str(_env_file))

    # 优先读取指定 key 下的模型，若未配置则 fallback 到 DEEPAGENT_MODEL
    model_str = _cfg.get(model_key, "")
    if not model_str and model_key != "DEEPAGENT_MODEL":
        model_str = _cfg.get("DEEPAGENT_MODEL", "openai:gpt-4o")
    if not model_str:
        model_str = "openai:gpt-4o"

    base_url = _cfg.get("OPENAI_BASE_URL", "")
    api_key = _cfg.get("OPENAI_API_KEY", "")
    timeout_str = _cfg.get("OPENAI_TIMEOUT", "180")

    # 清理可能带入的引号
    api_key = api_key.strip().strip('"').strip("'") if api_key else ""
    base_url = base_url.strip().strip('"').strip("'") if base_url else ""

    # 解析超时时间
    try:
        timeout = float(timeout_str.strip())
    except (ValueError, TypeError):
        timeout = 180.0

    # 确保模型字符串带有 provider 前缀，否则 init_chat_model 可能猜错 provider
    # 例如 "deepseek-v4-pro" 会被错误解析为 ChatDeepSeek（需要 DEEPSEEK_API_KEY）
    if ":" not in model_str:
        model_str = f"openai:{model_str}"

    if base_url and api_key:
        # 使用自定义 Base URL（如阿里百炼 DashScope）
        from langchain.chat_models import init_chat_model
        return init_chat_model(
            model_str,
            openai_api_key=api_key,
            openai_api_base=base_url,
            temperature=0,
            timeout=timeout,
        )
    return model_str


def create_click_interpret_agent():
    """创建「点选数据解读」Agent。

    该 Agent 专门处理单条巷道的深度数据解读，具备：
    - 实时数据查询能力
    - 规程合规判断能力
    - 结构化报告输出能力

    Returns:
        CompiledStateGraph: 编译后的 LangGraph 状态图，支持 .invoke() 和 .stream()
    """
    from deepagents import create_deep_agent  # 延迟导入，避免 PyCharm debug 启动卡死
    from deepagents.backends import FilesystemBackend
    from deepagents import FilesystemPermission

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

    skills = ["skills/click-interpret-tun", "skills/click-interpret-device"]
    print(f"[skills] Agent=click-interpret-agent skills={skills}")

    agent = create_deep_agent(
        model=_get_model(),
        tools=[
            query_tun_data_by_id,
            query_knowledge_base,
            query_device_data_by_id,
            query_tunnels_by_model,
            list_ventanaly_monitor_data_days,
            write_todos
        ],
        skills=skills,
        system_prompt=CLICK_INTERPRET_SYSTEM_PROMPT,
        backend=FilesystemBackend(root_dir=_project_root, virtual_mode=True),
        permissions=[
            FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"),
        ],
        middleware=[],
        name="click-interpret-agent",
        checkpointer=MemorySaver(),
    )
    return agent


def create_dialog_interpret_agent():
    """创建「对话式数据解读」Agent。

    该 Agent 支持多轮自然语言对话，具备：
    - 全量数据查询能力（实时+历史+告警）
    - 上下文记忆（通过 LangGraph checkpoint）
    - 智能意图识别和参数抽取
    - 使用运行时模型配置（支持前端动态切换）

    Returns:
        CompiledStateGraph: 编译后的 LangGraph 状态图
    """
    from deepagents import create_deep_agent  # 延迟导入，避免 PyCharm debug 启动卡死
    from deepagents.backends import FilesystemBackend
    from deepagents import FilesystemPermission
    from api.model_config import get_model_instance

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

    skills = ["skills/"]  # 父目录模式：SkillsMiddleware 自动扫描 skills/ 下所有子目录（含 SKILL.md 的技能目录）
    print(f"[skills] Agent=dialog-interpret-agent skills={skills}")

    agent = create_deep_agent(
        model=get_model_instance(),
        tools=[
            # 数据查询工具
            query_tun_data_by_id,
            query_device_data_by_id,
            query_devices_by_tunnel,
            query_devices_by_tunnel_id,
            query_tunnel_list,
            query_tunnels_by_model,
            list_ventanaly_monitor_data_days,
            query_knowledge_base,
            get_device_kind_dict,
            get_device_list_by_kind,
            query_device_realtime_data,
            # 基础工具
            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,
            # 用户偏好记忆
            save_user_preference,
            list_user_preferences,
            delete_user_preference,
            # 计划审批（Human-in-the-Loop）
            request_plan_approval,
        ],
        skills=skills,
        system_prompt=DIALOG_INTERPRET_SYSTEM_PROMPT,
        backend=FilesystemBackend(root_dir=_project_root, virtual_mode=True),
        permissions=[
            FilesystemPermission(operations=["write"], paths=["/**"], mode="deny"),
        ],
        middleware=[],
        name="dialog-interpret-agent",
        checkpointer=MemorySaver(),
    )

    # ── 初始化上下文用量追踪 ──
    _init_context_tracker_for_dialog()

    return agent


def _init_context_tracker_for_dialog():
    """为对话 agent 初始化系统组件 token 预估（统一版：含对话工具 + 计算工具 + MCP）。"""
    from tools.context_tracker import init_system_components, _tool_to_text

    _all_tools = [
        # 数据查询工具
        query_tun_data_by_id, query_device_data_by_id,
        query_devices_by_tunnel, query_devices_by_tunnel_id,
        query_tunnel_list, query_tunnels_by_model,
        list_ventanaly_monitor_data_days, query_knowledge_base,
        get_device_kind_dict, get_device_list_by_kind,
        query_device_realtime_data, 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,
        # 用户偏好
        save_user_preference, list_user_preferences, delete_user_preference,
        # 计划审批
        request_plan_approval,
    ]
    tool_texts = [_tool_to_text(t) for t in _all_tools]

    # MCP 工具单独归类
    mcp_text = _tool_to_text(get_needq_all_data)

    # 读取技能文件内容（动态扫描 skills/ 下所有技能，不再硬编码列表）
    skill_contents = []
    skills_root = Path(__file__).parent.parent / "skills"
    if skills_root.is_dir():
        for skill_dir in sorted(skills_root.iterdir()):
            if not skill_dir.is_dir() or skill_dir.name.startswith(".") or skill_dir.name == "__pycache__":
                continue
            sf = skill_dir / "SKILL.md"
            if sf.exists():
                skill_contents.append(sf.read_text(encoding="utf-8"))
                print(f"[context_tracker] 已加载技能: {skill_dir.name}")

    init_system_components(
        system_prompt=DIALOG_INTERPRET_SYSTEM_PROMPT,
        tool_defs=tool_texts,
        skill_contents=skill_contents,
        mcp_defs=[mcp_text],
    )


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

# 全局 Agent 实例（懒加载）
_click_agent: Optional[object] = None
_dialog_agent: Optional[object] = None


def invalidate_dialog_cache():
    """失效对话解读 Agent 缓存（模型切换时调用）。

    仅清除 dialog_agent 的缓存，不影响 click_agent（后者继续使用 .env 固定模型）。
    """
    global _dialog_agent
    _dialog_agent = None
    print("[模型切换] dialog-interpret-agent 缓存已失效")


def get_click_agent():
    """获取点选解读 Agent 单例"""
    global _click_agent
    if _click_agent is None:
        _click_agent = create_click_interpret_agent()
    return _click_agent


def get_dialog_agent():
    """获取对话解读 Agent 单例"""
    global _dialog_agent
    if _dialog_agent is None:
        _dialog_agent = create_dialog_interpret_agent()
    return _dialog_agent