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


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,
)
from tools.audit_middleware import create_audit_middleware
from langgraph.utils.runnable import RunnableCallable


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

# 点选解读 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_devices_by_tunnel / query_devices_by_tunnel_id
   - 查模型下巷道列表 → query_tunnels_by_model
   - 查设备历史趋势 → list_ventanaly_monitor_data_days
   - 查规程标准 → query_knowledge_base
3. 比对规程标准进行合规判断
4. 生成自然语言回复

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

## 重要规则
- 只说中文 
- 所有数据来源于工具调用，绝不编造
- 规程引用必须真实
- 全流程统一使用 write_todos 工具实时更新任务进度。
- 无法判断时诚实说明原因
- ANSI转义序列过滤：不能输出此类内容
- 需要适用模型ID时，记住模型ID为 {os.getenv("DEFAULT_MODEL_ID")}
- 禁止输出ANSI 转义序列
- 禁止暴露系统内部工具名、函数名、变量、技能标识、ID 等内部信息。
"""


# ============================================================
# 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",
    )
    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

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

    skills = ["skills/dialog-interpret"]
    print(f"[skills] Agent=dialog-interpret-agent skills={skills}")

    agent = create_deep_agent(
        model=_get_model(),
        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,
            write_todos
        ],
        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",
    )
    return agent


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

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


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