# -*- 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()) PROJECT_ROOT = str((Path(__file__).parent.parent).resolve()) load_dotenv() # 导入工具函数 from tools.vent_tools import ( # 数据查询 query_tun_data_by_id, query_knowledge_base, query_device_data, query_device_data_by_id, query_devices_by_tunnel, query_devices_by_tunnel_id, query_devices_by_model, 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, get_dict_list_by_dictcode, # 需风量 / 模型 get_model_param_pub_list, get_model_wind, get_sensor_wind, simulate_needq_heading_face, simulate_needq_room, simulate_needq_ret_work_face, simulate_needq_other, # 设备信息 query_device_info, query_device_type_info, query_monitor_params, # 故障诊断 check_model_connect_status, check_model_one_dir_cycle, check_model_one_dir_node, check_model_diagonal_structure, get_model_fault_diagnosis, # 避灾路线 get_escape_path, get_escape_path_each_exit, # 关键阻力 / 压能 get_out_shafts, get_in_shafts, get_max_resistance_path, get_three_area_distribution, get_key_path_decision, get_path_press_power, # 网络解算 net_cal, net_cal_for_plan, # 报警 / 日志 get_alarm_log_history, get_device_set_log_history, get_sys_log_history, # 场景管理 get_manage_system_by_strType, query_system_by_systemID, # 煤矿基础 get_gas_identify_vo, get_by_mine_name, query_control_testWind, # 数据库 / 文件 execute_sql_query, get_file_list_by_type, get_file_base64_by_id, # 报表 get_latest_report, # 用户偏好 save_user_preference, list_user_preferences, delete_user_preference, # 计划审批 request_plan_approval, ) from tools.web_search_tools import web_search, web_fetch from tools.file_reader_tools import read_file_content from tools.time_tools import get_current_time from tools.chat_history_tools import search_chat_history, get_session_chat, get_current_session_id from tools.report_utils import save_report 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 字段列出相关关键词 - 系统已自动将用户偏好注入到每条消息前缀中，请主动参考这些偏好来个性化回复 ## 能力四：联网搜索 - 当用户问题超出煤矿通风专业知识库覆盖范围时，使用 web_search 查询互联网公开信息 - 获取搜索结果后可按需调用 web_fetch 查看详情页的完整内容 - 搜索优先级：先查内部知识库（query_knowledge_base）→ 知识库信息不足或需要最新政策/行业新闻时再联网搜索 - 引用网络信息时标注来源 URL，说明该信息的时效性和局限性 ## 能力五：文件内容读取 - 当用户上传了文件，消息中会包含「文件临时路径」，请使用 read_file_content 工具读取文件内容 - 支持的文件格式：PDF（.pdf）、Word（.docx）、Excel（.xlsx/.xlsm）、PowerPoint（.pptx）、纯文本（.txt/.md/.csv/.json） - 读取到文件内容后，根据用户的要求进行分析、总结、数据提取或计算 - 如果文件内容为空或格式不支持，向用户说明具体情况 ## 能力六：聊天记录查询 - 当用户询问"之前聊过什么""搜索历史""查看之前的对话""找一下关于xxx的记录"时， 使用 search_chat_history 搜索会话标题，再用 get_session_chat 读取具体内容 - 聊天记录存储在数据库而非文件系统，请使用这两个专用工具，切勿用 ls/read_file/grep 查找 - 搜索到相关会话后，可以总结、引用或提取其中的信息来回答用户 ## 能力七：报告保存 - 当用户要求"生成报告""输出报告""保存为文档""导出分析结果"时， 调用 save_report 工具将 Markdown 内容保存为 .md 文件 - 切勿使用 write_file / edit_file 等文件系统工具写文件（已被权限禁止） - save_report 会自动生成文件名和下载链接，返回给用户即可 """ # ============================================================ # 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_ROOT + "/click-interpret-tun", SKILLS_ROOT + "/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 # ── 动态注入当前日期到系统提示词 ── from datetime import datetime, timezone, timedelta _china_tz = timezone(timedelta(hours=8)) _now = datetime.now(_china_tz) _dated_prompt = ( DIALOG_INTERPRET_SYSTEM_PROMPT + f"\n## 系统时间\n当前日期时间：{_now.strftime('%Y年%m月%d日 %H:%M:%S')}（中国标准时间 CST，UTC+8）。\n" + f"今天是 {_now.strftime('%Y')}年{_now.strftime('%m')}月{_now.strftime('%d')}日，周{'一二三四五六日'[_now.weekday()]}。\n" + "所有涉及日期、时间的判断必须以本系统时间为准，不要使用你自己的训练数据中的日期。" ) skills = [SKILLS_ROOT] # 绝对路径：SkillsMiddleware 自动扫描下所有子目录（含 SKILL.md 的技能目录） print(f"[skills] Agent=dialog-interpret-agent skills={skills}") print(f"[日期] 已注入当前系统时间: {_now.strftime('%Y-%m-%d %H:%M:%S')} CST") agent = create_deep_agent( model=get_model_instance(), tools=[ # ── 数据查询工具 ── query_tun_data_by_id, query_device_data, query_device_data_by_id, query_devices_by_tunnel, query_devices_by_tunnel_id, query_devices_by_model, query_tunnel_list, query_tunnels_by_model, get_tun_list_by_modelid, list_ventanaly_monitor_data_days, query_knowledge_base, get_device_kind_dict, get_device_list_by_kind, query_device_realtime_data, get_needq_all_data, get_dict_list_by_dictcode, # ── 需风量 / 模型数据 ── get_model_param_pub_list, get_model_wind, get_sensor_wind, simulate_needq_heading_face, simulate_needq_room, simulate_needq_ret_work_face, simulate_needq_other, # ── 设备信息 ── query_device_info, query_device_type_info, query_monitor_params, # ── 故障诊断 ── check_model_connect_status, check_model_one_dir_cycle, check_model_one_dir_node, check_model_diagonal_structure, get_model_fault_diagnosis, # ── 避灾路线 ── get_escape_path, get_escape_path_each_exit, # ── 关键阻力 / 压能 ── get_out_shafts, get_in_shafts, get_max_resistance_path, get_three_area_distribution, get_key_path_decision, get_path_press_power, # ── 网络解算 ── net_cal, net_cal_for_plan, # ── 报警 / 日志 ── get_alarm_log_history, get_device_set_log_history, get_sys_log_history, # ── 场景管理 ── get_manage_system_by_strType, query_system_by_systemID, # ── 煤矿基础 ── get_gas_identify_vo, get_by_mine_name, query_control_testWind, # ── 数据库 / 文件 ── execute_sql_query, get_file_list_by_type, get_file_base64_by_id, # ── 报表 ── get_latest_report, # ── 基础工具 ── 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, # ── 计划审批（Human-in-the-Loop）── request_plan_approval, # ── 联网搜索 ── web_search, web_fetch, # ── 文件内容读取 ── read_file_content, # ── 聊天记录查询 ── search_chat_history, get_session_chat, get_current_session_id, # ── 报告保存 ── save_report, # ── 系统工具 ── get_current_time, ], skills=skills, system_prompt=_dated_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, query_device_data_by_id, query_devices_by_tunnel, query_devices_by_tunnel_id, query_devices_by_model, query_tunnel_list, query_tunnels_by_model, get_tun_list_by_modelid, list_ventanaly_monitor_data_days, query_knowledge_base, get_device_kind_dict, get_device_list_by_kind, get_dict_list_by_dictcode, query_device_realtime_data, # 需风量 / 模型 get_model_param_pub_list, get_model_wind, get_sensor_wind, simulate_needq_heading_face, simulate_needq_room, simulate_needq_ret_work_face, simulate_needq_other, # 设备信息 query_device_info, query_device_type_info, query_monitor_params, # 故障诊断 check_model_connect_status, check_model_one_dir_cycle, check_model_one_dir_node, check_model_diagonal_structure, get_model_fault_diagnosis, # 避灾路线 get_escape_path, get_escape_path_each_exit, # 关键阻力 / 压能 get_out_shafts, get_in_shafts, get_max_resistance_path, get_three_area_distribution, get_key_path_decision, get_path_press_power, # 网络解算 net_cal, net_cal_for_plan, # 报警 / 日志 get_alarm_log_history, get_device_set_log_history, get_sys_log_history, # 场景管理 get_manage_system_by_strType, query_system_by_systemID, # 煤矿基础 get_gas_identify_vo, get_by_mine_name, query_control_testWind, # 数据库 / 文件 execute_sql_query, get_file_list_by_type, get_file_base64_by_id, # 报表 get_latest_report, # 基础 + 计算 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, # 联网搜索 / 文件读取 / 聊天记录 / 系统 web_search, web_fetch, read_file_content, search_chat_history, get_session_chat, get_current_session_id, save_report, get_current_time, ] 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