""" 交互Agent调度器 基于 LangGraph StateGraph 的企微交互 Agent。 企微入口适配层：处理图片上传、VLM分析等企微特有逻辑， 核心对话由 LangGraph 图处理。 """ import json import time from typing import Dict, List, Optional from openai import AsyncOpenAI from app.config import settings from app.services.session_manager import get_session_manager from app.utils.logger import logger class AgentDispatcher: """交互Agent调度器（单例）""" def __init__(self): self._vlm_client: Optional[AsyncOpenAI] = None self._enabled = False self._graph = None self._pending_images: Dict[str, List[str]] = {} def init(self, config): """初始化Agent""" self._enabled = config.enabled and bool(config.vlm_api_key) if not self._enabled: logger.info("交互Agent未启用（AGENT_ENABLED=false 或缺少 API Key）") return # VLM 客户端（图片分析专用） self._vlm_client = AsyncOpenAI( api_key=config.vlm_api_key, base_url=config.vlm_base_url, ) from app.services.agent.graph import create_default_graph self._graph = create_default_graph() logger.info(f"交互Agent已启用(LangGraph): model={config.model}") @property def enabled(self) -> bool: return self._enabled # ==================== 消息入口 ==================== async def handle_message(self, user_id: str, content: str) -> str: """处理文字消息""" if not self._enabled: return "AI助手未启用，请联系管理员配置。" self._pending_images[user_id] = [] session = get_session_manager().get(user_id) normalized_content = content.strip() try: reply = self._maybe_start_manual_order_from_text(session, normalized_content) if reply is None: reply = await self._handle_manual_order_message(user_id, session, normalized_content) if reply is None: reply = await self._langgraph_chat(user_id, content) except Exception as e: logger.error(f"Agent对话失败: {e}", exc_info=True) reply = "抱歉，AI助手暂时无法响应，请稍后重试。" # 发送待发图片 pending = self._pending_images.pop(user_id, []) if pending: await self._send_images_to_user(user_id, pending) return reply async def handle_image(self, user_id: str, media_id: str) -> str: """处理图片消息""" if not self._enabled: return "AI助手未启用，请联系管理员配置。" session = get_session_manager().get(user_id) # 1. 下载图片 from app.services.wechat_service import get_wechat_service wechat = get_wechat_service() image_data = await wechat.download_media(media_id) if not image_data: return "图片下载失败，请重新发送。" # 2. 上传 COS 持久化 from app.services.oss_storage import get_oss_storage oss = get_oss_storage() object_key = f"agent/{user_id}/{int(time.time())}.jpg" try: oss.upload_file(image_data, object_key, content_type="image/jpeg") except Exception as e: logger.error(f"Agent图片上传COS失败: {e}") return "图片保存失败，请重新发送。" permanent_url = oss.get_permanent_url(object_key) presign_url = oss.get_presigned_url(object_key) if session.state == "waiting_manual_order_image": del presign_url return await self._start_manual_order_flow(session, permanent_url) # 3. 检查用户是否有待处理工单 handling_alarm_id = self._find_handling_alarm(user_id) if handling_alarm_id: session.pending_images.append(permanent_url) reply = f"收到图片，是否作为【告警 {handling_alarm_id[:20]}...】的处理结果提交？\n回复「是」确认提交，或继续发送更多图片。" session.pending_alarm_id = handling_alarm_id return reply if session.state in { "waiting_manual_order_area", "waiting_manual_order_remark", "waiting_manual_order_confirm", } and session.pending_manual_order_images: session.pending_manual_order_images.append(permanent_url) if session.state == "waiting_manual_order_area": return ( f"已追加图片，当前共 {len(session.pending_manual_order_images)} 张。\n" f"请选择区域：\n{self._format_area_options(session.pending_manual_order_area_options)}\n" "请回复区域编号。" ) if session.state == "waiting_manual_order_remark": return ( f"已追加图片，当前共 {len(session.pending_manual_order_images)} 张。\n" "请继续补充备注信息，可不填，回复“无”即可。" ) return ( f"已追加图片，当前共 {len(session.pending_manual_order_images)} 张。\n" "请回复“确认”创建工单，或回复“取消”放弃。" ) del presign_url return await self._start_manual_order_flow(session, permanent_url) # ==================== LangGraph 对话 ==================== async def _langgraph_chat(self, user_id: str, content: str) -> str: """LangGraph 图调用""" config = { "configurable": { "thread_id": f"wechat-{user_id}", "user_id": user_id, } } result = await self._graph.ainvoke( { "messages": [{"role": "user", "content": content}], "user_id": user_id, "pending_images": [], "user_uploaded_images": [], }, config=config, ) # 从工具返回中提取截图 URL（get_alarm_detail 返回的 snapshot_url） self._extract_pending_images(user_id, result) # 获取最终回复 last_msg = result["messages"][-1] reply = last_msg.content if hasattr(last_msg, "content") else str(last_msg) return reply.strip() if reply else "处理完成" def _extract_pending_images(self, user_id: str, result): """从 LangGraph 结果中提取需要发送的截图""" for msg in result.get("messages", []): if hasattr(msg, "type") and msg.type == "tool" and msg.name == "get_alarm_detail": try: data = json.loads(msg.content) if isinstance(msg.content, str) else msg.content url = data.get("snapshot_url", "") if url: if user_id not in self._pending_images: self._pending_images[user_id] = [] self._pending_images[user_id].append(url) except Exception: pass # ==================== 共用方法 ==================== async def _start_manual_order_flow(self, session, image_url: str) -> str: """启动手动工单创建流程。""" session.pending_manual_order_images.append(image_url) session.pending_manual_order_remark = "" session.pending_manual_order_area_id = "" session.pending_manual_order_area_name = "" session.pending_manual_order_area_options = self._list_notify_areas() if not session.pending_manual_order_area_options: session.pending_manual_order_area_options = [self._get_demo_area()] session.state = "waiting_manual_order_area" return ( "已收到图片，准备创建手动工单。\n" f"请选择区域：\n{self._format_area_options(session.pending_manual_order_area_options)}\n" "请回复区域编号。" ) @staticmethod def _maybe_start_manual_order_from_text(session, content: str) -> Optional[str]: """用户明确提出创建工单时，优先进入手动建单流程。""" if session.state != "idle": return None trigger_phrases = ( "创建工单", "新建工单", "手动工单", "手动上报", "我要创建工单", "我要上报", "上报工单", ) if not any(phrase in content for phrase in trigger_phrases): return None session.reset() session.state = "waiting_manual_order_image" return "请先上传现场图片，我会在收到图片后引导您选择区域并补充备注。" async def _handle_manual_order_message(self, user_id: str, session, content: str) -> Optional[str]: """处理手动工单创建状态机。""" if session.state == "waiting_manual_order_image": if content in {"取消", "算了", "不用了"}: session.reset() return "已取消本次手动工单创建。" return "请先上传现场图片。若不需要创建工单，回复“取消”即可。" if session.state == "waiting_manual_order_area": area = self._match_area_option(content, session.pending_manual_order_area_options) if not area: return ( "未识别到有效区域，请回复区域编号。\n" f"{self._format_area_options(session.pending_manual_order_area_options)}" ) assignees = self._get_area_assignees(area["area_id"]) if not assignees and area["area_id"] != self._get_demo_area()["area_id"]: return f"区域【{area['area_name']}】当前未绑定责任人，请重新选择其他区域。" session.pending_manual_order_area_id = area["area_id"] session.pending_manual_order_area_name = area["area_name"] session.state = "waiting_manual_order_remark" return f"已选择区域：【{area['area_name']}】。\n请补充备注信息，可不填，回复“无”即可。" if session.state == "waiting_manual_order_remark": session.pending_manual_order_remark = "" if content in {"无", "没有", "none", "None"} else content session.state = "waiting_manual_order_confirm" assignees = self._get_area_assignees(session.pending_manual_order_area_id) if not assignees and session.pending_manual_order_area_id == self._get_demo_area()["area_id"]: assignees = [{"person_name": "演示用户", "wechat_uid": user_id, "role": "demo"}] assignee_names = "、".join(person["person_name"] for person in assignees) remark = session.pending_manual_order_remark or "无" return ( "请确认是否创建工单：\n" f"区域：【{session.pending_manual_order_area_name}】\n" f"备注：{remark}\n" f"图片：{len(session.pending_manual_order_images)} 张\n" f"派发对象：{assignee_names}\n" "回复“确认”创建，回复“取消”放弃。" ) if session.state == "waiting_manual_order_confirm": if content in {"取消", "算了", "不用了"}: session.reset() return "已取消本次手动工单创建。" if content not in {"确认", "是", "创建", "提交"}: return "请回复“确认”创建工单，或回复“取消”放弃。" result = await self._create_manual_order(user_id, session) session.reset() return result return None @staticmethod def _list_notify_areas() -> List[Dict[str, str]]: from app.models import NotifyArea, get_session db = get_session() try: areas = ( db.query(NotifyArea) .filter(NotifyArea.enabled == 1) .order_by(NotifyArea.area_name.asc()) .all() ) return [ {"index": str(idx), "area_id": area.area_id, "area_name": area.area_name} for idx, area in enumerate(areas, start=1) ] finally: db.close() @staticmethod def _get_demo_area() -> Dict[str, str]: return {"index": "1", "area_id": "demo-area", "area_name": "\u6f14\u793a\u533a\u57df"} @staticmethod def _format_area_options(areas: List[Dict[str, str]]) -> str: return "\n".join(f"{item['index']}. {item['area_name']}" for item in areas) @staticmethod def _match_area_option(content: str, areas: List[Dict[str, str]]) -> Optional[Dict[str, str]]: normalized = content.strip() for item in areas: if normalized == item["index"] or normalized == item["area_name"]: return item return None @staticmethod def _get_area_assignees(area_id: str) -> List[Dict[str, str]]: from app.models import AreaPersonBinding, get_session db = get_session() try: persons = ( db.query(AreaPersonBinding) .filter( AreaPersonBinding.area_id == area_id, AreaPersonBinding.enabled == 1, ) .order_by(AreaPersonBinding.notify_level.asc(), AreaPersonBinding.id.asc()) .all() ) return [ { "person_name": person.person_name, "wechat_uid": person.wechat_uid, "role": person.role, } for person in persons ] finally: db.close() async def _create_manual_order(self, user_id: str, session) -> str: """创建手动工单并通知区域绑定人员。""" assignees = self._get_area_assignees(session.pending_manual_order_area_id) if not assignees and session.pending_manual_order_area_id != self._get_demo_area()["area_id"]: return f"区域【{session.pending_manual_order_area_name}】当前未绑定责任人，工单未创建。" if not assignees: assignees = [{"person_name": "演示用户", "wechat_uid": user_id, "role": "demo"}] from app.services.wechat_service import get_wechat_service from app.services.work_order_service import get_work_order_service area_name = session.pending_manual_order_area_name remark = session.pending_manual_order_remark or "无" title = f"【手动上报】{area_name}异常情况" description_lines = [ "来源：企微手动上报", f"上报人：{user_id}", f"区域：{area_name}", f"备注：{remark}", f"图片数量：{len(session.pending_manual_order_images)}", ] if session.pending_manual_order_images: description_lines.append("图片链接：") description_lines.extend(session.pending_manual_order_images) attachments = [ {"type": "image", "url": image_url} for image_url in session.pending_manual_order_images ] primary_assignee = assignees[0] work_order = get_work_order_service().create_work_order( title=title, description="\n".join(description_lines), priority="medium", assignee_uid=primary_assignee["wechat_uid"], assignee_name=primary_assignee["person_name"], attachments=attachments, department=area_name, ) if not work_order: return "工单创建失败，请稍后重试。" notify_message = ( f"收到新的手动工单：\n" f"工单号：{work_order.order_no}\n" f"区域：{area_name}\n" f"备注：{remark}" ) wechat = get_wechat_service() for assignee in assignees: try: await wechat.send_text_message(assignee["wechat_uid"], notify_message) except Exception as e: logger.error( f"手动工单通知失败: order={work_order.order_no}, " f"user={assignee['wechat_uid']}, error={e}" ) return ( f"工单已创建。\n" f"工单号：{work_order.order_no}\n" f"区域：【{area_name}】\n" f"已派发给：{'、'.join(person['person_name'] for person in assignees)}" ) async def _analyze_image(self, image_url: str, prompt: str) -> Dict: """VLM 分析图片内容""" try: resp = await self._vlm_client.chat.completions.create( model=settings.agent.vlm_model, temperature=0.1, messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": [ {"type": "image_url", "image_url": {"url": image_url}}, {"type": "text", "text": "请分析这张图片"}, ]}, ], timeout=settings.agent.vlm_timeout, ) text = resp.choices[0].message.content.strip() if "```" in text: text = text.split("```")[1].strip() if text.startswith("json"): text = text[4:].strip() return json.loads(text) except Exception as e: logger.error(f"VLM图片分析失败: {e}") return {"has_anomaly": False, "description": "", "alarm_type": ""} async def _send_images_to_user(self, user_id: str, image_urls: List[str]): """通过企微发送图片消息给用户""" from app.services.wechat_service import get_wechat_service wechat = get_wechat_service() if not wechat.enabled: return for url in image_urls: try: media_id = await wechat.upload_media_from_url(url) if media_id: access_token = await wechat._get_access_token() import httpx msg = { "touser": user_id, "msgtype": "image", "agentid": wechat.agent_id_int, "image": {"media_id": media_id}, } api_url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={access_token}" async with httpx.AsyncClient(timeout=10) as client: await client.post(api_url, json=msg) except Exception as e: logger.error(f"发送告警截图失败: user={user_id}, error={e}") @staticmethod def _find_handling_alarm(user_id: str) -> str: """查找用户正在处理的告警ID""" from app.models import get_session, AlarmEvent db = get_session() try: alarm = db.query(AlarmEvent).filter( AlarmEvent.handler == user_id, AlarmEvent.handle_status == "HANDLING", ).order_by(AlarmEvent.event_time.desc()).first() return alarm.alarm_id if alarm else "" except Exception as e: logger.error(f"查询待处理告警失败: user={user_id}, error={e}") return "" finally: db.close() # 全局单例 _agent_dispatcher: Optional[AgentDispatcher] = None def get_agent_dispatcher() -> AgentDispatcher: global _agent_dispatcher if _agent_dispatcher is None: _agent_dispatcher = AgentDispatcher() return _agent_dispatcher