From f87222e6fbec427ea91f7ee78c082ed57782a7c9 Mon Sep 17 00:00:00 2001
From: 16337 <1633794139@qq.com>
Date: Fri, 20 Mar 2026 11:10:54 +0800
Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9AAgentDispatcher=20?=
 =?UTF-8?q?=E5=A4=9A=E6=A8=A1=E6=80=81=E9=87=8D=E5=86=99?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 统一使用 VLM 模型处理文字+图片
- 多轮对话上下文（SessionManager）
- 图片分析上报：VLM 分析 → 追问位置 → 创建工单
- 结单图片分析：VLM 确认异常消除 → 自动结单
- 意图识别嵌入对话回复中，不再单独调用
- 所有模型配置走 settings，无硬编码
---
 app/services/agent_dispatcher.py | 381 +++++++++++++++++++++++--------
 1 file changed, 288 insertions(+), 93 deletions(-)

diff --git a/app/services/agent_dispatcher.py b/app/services/agent_dispatcher.py
index 175ed76..6263dc0 100644
--- a/app/services/agent_dispatcher.py
+++ b/app/services/agent_dispatcher.py
@@ -1,44 +1,61 @@
 """
-交互Agent调度器
+交互Agent调度器（多模态版）
 
-接收企微用户消息，通过LLM识别意图，路由到对应处理器。
-
-支持意图：
-- create_work_order: 创建工单（"帮我创建XX工单"）
-- query_alarm: 查询告警（"今天有多少告警"）
-- export_report: 导出报表（"导出本周告警报表"）
-- general_chat: 兜底闲聊
+统一使用 VLM 模型处理文字+图片，支持：
+- 文字意图识别（创建工单/查询告警/导出报表/闲聊）
+- 图片分析上报（VLM 分析 → 追问位置 → 创建工单）
+- 结单图片分析（VLM 确认异常消除 → 自动结单）
+- 多轮对话上下文（每用户独立，10轮，10分钟TTL）
 """
 
 import json
+import time
 from datetime import timedelta
-from typing import Dict, Optional
+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
 from app.utils.timezone import beijing_now
 
-INTENT_SYSTEM_PROMPT = """你是物业安防AI助手。根据用户消息识别意图，仅输出JSON。
+SYSTEM_PROMPT = """你是物业安防AI助手。你可以：
+1. 分析用户上传的现场图片，识别安全隐患或异常情况
+2. 帮助创建安保工单（需要位置信息）
+3. 查询告警统计数据
+4. 导出告警报表
+5. 分析处理结果图片，确认异常是否消除
+
+规则：
+- 当用户发送图片时，分析图片内容，判断是否有安全隐患
+- 如果识别到异常，描述异常并询问具体位置
+- 回复简洁专业，不超过100字
+- 不要编造不存在的信息
+
+当需要识别意图时，在回复末尾附加JSON标记：
+<!--INTENT:{"intent":"...","params":{...}}-->
 
 可选意图：
 - create_work_order: 用户要创建工单或上报问题
-- query_alarm: 用户要查询告警数据或统计
-- export_report: 用户要导出报表或Excel
-- general_chat: 其他闲聊或无法识别
+- query_alarm: 用户要查询告警数据（params: time_range=today/week/month, alarm_type=leave_post/intrusion/all）
+- export_report: 用户要导出报表（params: time_range=today/week/month）
+- general_chat: 其他对话（无需附加标记）"""
 
-输出格式：{"intent":"...","params":{...}}
+IMAGE_ANALYZE_PROMPT = """分析这张图片，判断是否存在安全隐患或异常情况。
+请用JSON格式回复：
+{"has_anomaly": true/false, "description": "异常描述", "alarm_type": "告警类型(fire/intrusion/damage/leak/other/none)"}
+只输出JSON，不要其他内容。"""
 
-params说明：
-- create_work_order: {"title":"工单标题","description":"描述","priority":"low/medium/high/urgent"}
-- query_alarm: {"time_range":"today/week/month","alarm_type":"leave_post/intrusion/all"}
-- export_report: {"time_range":"today/week/month"}
-- general_chat: {"message":"友好的回复内容"}"""
+CLOSE_ANALYZE_PROMPT = """这是一张处理后的现场照片。请判断之前的异常是否已经消除。
+之前的异常是：{previous_issue}
+请用JSON格式回复：
+{"resolved": true/false, "description": "当前状态描述"}
+只输出JSON，不要其他内容。"""
 
 
 class AgentDispatcher:
-    """交互Agent调度器（单例）"""
+    """交互Agent调度器（多模态，单例）"""
 
     def __init__(self):
         self._client: Optional[AsyncOpenAI] = None
@@ -46,13 +63,13 @@ class AgentDispatcher:
 
     def init(self, config):
         """初始化Agent"""
-        self._enabled = config.enabled and bool(config.llm_api_key)
+        self._enabled = config.enabled and bool(config.vlm_api_key)
         if self._enabled:
             self._client = AsyncOpenAI(
-                api_key=config.llm_api_key,
-                base_url=config.llm_base_url,
+                api_key=config.vlm_api_key,
+                base_url=config.vlm_base_url,
             )
-            logger.info(f"交互Agent已启用: model={config.llm_model}")
+            logger.info(f"交互Agent已启用: model={config.vlm_model}")
         else:
             logger.info("交互Agent未启用（AGENT_ENABLED=false 或缺少 API Key）")
 
@@ -60,90 +77,275 @@ class AgentDispatcher:
     def enabled(self) -> bool:
         return self._enabled
 
+    # ==================== 消息入口 ====================
+
     async def handle_message(self, user_id: str, content: str) -> str:
-        """处理用户消息，返回回复文本"""
+        """处理文字消息"""
         if not self._enabled:
             return "AI助手未启用，请联系管理员配置。"
 
-        # 1. 意图识别
-        intent_result = await self._classify_intent(content)
-        intent = intent_result.get("intent", "general_chat")
-        params = intent_result.get("params", {})
+        session = get_session_manager().get(user_id)
 
-        logger.info(f"Agent意图识别: user={user_id}, intent={intent}, params={params}")
+        # 状态机：等待位置信息
+        if session.state == "waiting_location":
+            return await self._handle_location_reply(user_id, session, content)
 
-        # 2. 路由到对应 handler
-        handlers = {
-            "create_work_order": self._handle_create_work_order,
-            "query_alarm": self._handle_query_alarm,
-            "export_report": self._handle_export_report,
-            "general_chat": self._handle_general_chat,
-        }
+        # 状态机：等待确认
+        if session.state == "waiting_confirm":
+            return await self._handle_confirm_reply(user_id, session, content)
 
-        handler = handlers.get(intent, self._handle_general_chat)
+        # 正常对话：带上下文调用 VLM
+        session.add_history("user", content)
+
+        reply = await self._chat(session)
+        session.add_history("assistant", reply)
+
+        # 检查是否有嵌入的意图标记
+        intent_result = self._extract_intent(reply)
+        if intent_result:
+            clean_reply = reply.split("<!--INTENT:")[0].strip()
+            action_reply = await self._execute_intent(user_id, intent_result)
+            return f"{clean_reply}\n\n{action_reply}" if clean_reply else action_reply
+
+        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:
-            return await handler(user_id, params, content)
+            oss.upload_file(image_data, object_key, content_type="image/jpeg")
         except Exception as e:
-            logger.error(f"Agent handler异常: intent={intent}, error={e}", exc_info=True)
-            return "处理请求时出错，请稍后重试。"
+            logger.error(f"Agent图片上传COS失败: {e}")
+            return "图片保存失败，请重新发送。"
+
+        image_url = oss.get_presigned_url(object_key)
+
+        # 3. 如果在等待结单图片
+        if session.state == "waiting_close_photo":
+            return await self._handle_close_photo(user_id, session, image_url, object_key)
+
+        # 4. 正常上报：VLM 分析图片
+        session.pending_image_url = object_key
+        analysis = await self._analyze_image(image_url)
+
+        session.pending_analysis = analysis.get("description", "")
+        session.pending_alarm_type = analysis.get("alarm_type", "other")
+
+        session.add_history("user", [
+            {"type": "text", "text": "[用户上传了一张图片]"},
+        ])
+
+        if analysis.get("has_anomaly"):
+            session.state = "waiting_location"
+            desc = analysis.get("description", "异常情况")
+            reply = f"检测到异常：{desc}\n\n请问具体在什么位置？（如：A栋3层东侧走廊）"
+        else:
+            session.state = "waiting_location"
+            reply = "未检测到明显异常。如确需上报，请告知具体位置和情况描述。"
+
+        session.add_history("assistant", reply)
+        return reply
+
+    # ==================== 状态机处理 ====================
+
+    async def _handle_location_reply(self, user_id: str, session, location: str) -> str:
+        """用户回复位置 → 创建工单"""
+        session.add_history("user", location)
+
+        alarm_type_name = session.pending_alarm_type or "异常情况"
+        description = session.pending_analysis or "人工上报"
+        image_url = ""
+        if session.pending_image_url:
+            try:
+                from app.services.oss_storage import get_oss_storage
+                image_url = get_oss_storage().get_permanent_url(session.pending_image_url)
+            except Exception:
+                pass
+
+        # 通过 IoT 平台创建工单
+        try:
+            from app.services.work_order_client import get_work_order_client
+            wo_client = get_work_order_client()
+            if wo_client.enabled:
+                alarm_id = f"MANUAL_{user_id}_{int(time.time())}"
+                order_id = await wo_client.create_order(
+                    title=f"{alarm_type_name} - 人工上报",
+                    area_id=0,
+                    alarm_id=alarm_id,
+                    alarm_type=alarm_type_name,
+                    description=f"{description}\n位置：{location}",
+                    priority=2,
+                    trigger_source="人工上报",
+                    camera_id="",
+                    image_url=image_url,
+                )
+                session.pending_order_id = order_id or ""
+                session.reset()
+
+                if order_id:
+                    reply = f"工单已创建\n编号：{order_id}\n位置：{location}\n描述：{description}\n\n相关人员将尽快处理。"
+                else:
+                    reply = "工单创建失败，请联系管理员。"
+            else:
+                session.reset()
+                reply = f"已记录上报信息\n位置：{location}\n描述：{description}\n\n（工单系统未启用，请联系管理员）"
+        except Exception as e:
+            logger.error(f"Agent创建工单失败: {e}", exc_info=True)
+            session.reset()
+            reply = "工单创建失败，请稍后重试或联系管理员。"
+
+        session.add_history("assistant", reply)
+        return reply
+
+    async def _handle_confirm_reply(self, user_id: str, session, content: str) -> str:
+        """用户回复确认"""
+        content_lower = content.strip().lower()
+        if content_lower in ("是", "确认", "对", "yes", "y", "ok"):
+            session.reset()
+            return "已确认，感谢反馈。"
+        elif content_lower in ("否", "取消", "不是", "no", "n"):
+            session.reset()
+            return "已取消。如需帮助请随时联系。"
+        else:
+            return "请回复「是」确认或「否」取消。"
+
+    async def _handle_close_photo(self, user_id: str, session, image_url: str, object_key: str) -> str:
+        """分析结单图片"""
+        previous_issue = session.pending_analysis or "之前的异常"
 
-    async def _classify_intent(self, content: str) -> Dict:
-        """LLM意图分类"""
         try:
             resp = await self._client.chat.completions.create(
-                model=settings.agent.llm_model,
+                model=settings.agent.vlm_model,
                 messages=[
-                    {"role": "system", "content": INTENT_SYSTEM_PROMPT},
-                    {"role": "user", "content": content},
+                    {"role": "system", "content": CLOSE_ANALYZE_PROMPT.format(previous_issue=previous_issue)},
+                    {"role": "user", "content": [
+                        {"type": "image_url", "image_url": {"url": image_url}},
+                        {"type": "text", "text": "请判断异常是否已消除"},
+                    ]},
                 ],
-                timeout=settings.agent.llm_timeout,
+                timeout=settings.agent.vlm_timeout,
             )
             text = resp.choices[0].message.content.strip()
             if "```" in text:
-                text = text.split("```")[1]
+                text = text.split("```")[1].strip()
                 if text.startswith("json"):
-                    text = text[4:]
-                text = text.strip()
+                    text = text[4:].strip()
+            result = json.loads(text)
+        except Exception as e:
+            logger.error(f"结单图片分析失败: {e}")
+            session.reset()
+            return "图片分析失败，请重新拍照上传，或联系管理员手动结单。"
+
+        if result.get("resolved"):
+            # 自动结单
+            order_id = session.pending_order_id
+            if order_id:
+                try:
+                    from app.services.work_order_client import get_work_order_client
+                    wo_client = get_work_order_client()
+                    await wo_client.auto_complete_order(int(order_id), "处理完成-AI确认异常已消除")
+                except Exception as e:
+                    logger.error(f"自动结单失败: {e}")
+
+            session.reset()
+            desc = result.get("description", "现场已恢复正常")
+            return f"确认处理完成：{desc}\n工单已自动关闭。"
+        else:
+            desc = result.get("description", "仍有异常")
+            return f"检测到仍有异常：{desc}\n请继续处理后再次拍照上传。"
+
+    # ==================== VLM 调用 ====================
+
+    async def _chat(self, session) -> str:
+        """带上下文的 VLM 对话"""
+        try:
+            messages = [{"role": "system", "content": SYSTEM_PROMPT}]
+            messages.extend(session.get_history_for_vlm())
+
+            resp = await self._client.chat.completions.create(
+                model=settings.agent.vlm_model,
+                messages=messages,
+                timeout=settings.agent.vlm_timeout,
+            )
+            return resp.choices[0].message.content.strip()
+        except Exception as e:
+            logger.error(f"VLM对话失败: {e}")
+            return "抱歉，AI助手暂时无法响应，请稍后重试。"
+
+    async def _analyze_image(self, image_url: str) -> Dict:
+        """VLM 分析图片内容"""
+        try:
+            resp = await self._client.chat.completions.create(
+                model=settings.agent.vlm_model,
+                messages=[
+                    {"role": "system", "content": IMAGE_ANALYZE_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"意图识别失败: {e}")
-            return {"intent": "general_chat", "params": {"message": "抱歉，我暂时无法理解您的请求。"}}
+            logger.error(f"VLM图片分析失败: {e}")
+            return {"has_anomaly": False, "description": "", "alarm_type": ""}
 
-    async def _handle_create_work_order(self, user_id: str, params: Dict, raw: str) -> str:
-        """创建工单"""
-        from app.services.work_order_service import get_work_order_service
-        svc = get_work_order_service()
+    # ==================== 意图执行 ====================
 
-        title = params.get("title", "")
-        if not title:
-            title = raw[:50]
+    def _extract_intent(self, reply: str) -> Optional[Dict]:
+        """从回复中提取嵌入的意图标记"""
+        if "<!--INTENT:" not in reply:
+            return None
+        try:
+            json_str = reply.split("<!--INTENT:")[1].split("-->")[0]
+            return json.loads(json_str)
+        except Exception:
+            return None
 
-        order = svc.create_work_order(
-            title=title,
-            description=params.get("description", raw),
-            priority=params.get("priority", "medium"),
-            assignee_uid=user_id,
-        )
+    async def _execute_intent(self, user_id: str, intent_result: Dict) -> str:
+        """执行识别到的意图"""
+        intent = intent_result.get("intent", "")
+        params = intent_result.get("params", {})
 
-        if order:
-            priority_names = {"low": "低", "medium": "中", "high": "高", "urgent": "紧急"}
-            p_name = priority_names.get(order.priority.value, "中")
-            return (
-                f"工单已创建\n"
-                f"编号：{order.order_no}\n"
-                f"标题：{order.title}\n"
-                f"优先级：{p_name}\n"
-                f"状态：待处理"
-            )
-        return "工单创建失败，请稍后重试"
+        if intent == "query_alarm":
+            return await self._handle_query_alarm(user_id, params)
+        elif intent == "export_report":
+            return await self._handle_export_report(user_id, params)
+        elif intent == "create_work_order":
+            # 文字创建工单 → 进入等待位置状态
+            session = get_session_manager().get(user_id)
+            session.pending_analysis = params.get("description", "")
+            session.pending_alarm_type = params.get("title", "人工上报")
+            session.state = "waiting_location"
+            return "请提供具体位置（如：A栋3层东侧走廊）"
+        return ""
 
-    async def _handle_query_alarm(self, user_id: str, params: Dict, raw: str) -> str:
+    async def _handle_query_alarm(self, user_id: str, params: Dict) -> str:
         """查询告警统计"""
         from app.services.alarm_event_service import get_alarm_event_service
-
         svc = get_alarm_event_service()
 
-        # 解析时间范围
         time_range = params.get("time_range", "today")
         now = beijing_now()
         if time_range == "week":
@@ -169,7 +371,6 @@ class AgentDispatcher:
             page_size=10000,
         )
 
-        # 按类型统计
         type_count = {}
         status_count = {"NEW": 0, "CONFIRMED": 0, "FALSE": 0, "CLOSED": 0}
         for a in alarms:
@@ -177,7 +378,10 @@ class AgentDispatcher:
             if a.alarm_status in status_count:
                 status_count[a.alarm_status] += 1
 
-        type_names = {"leave_post": "人员离岗", "intrusion": "周界入侵"}
+        type_names = {
+            "leave_post": "人员离岗", "intrusion": "周界入侵",
+            "illegal_parking": "车辆违停", "vehicle_congestion": "车辆拥堵",
+        }
         type_lines = [f"  {type_names.get(t, t)}: {c}条" for t, c in type_count.items()]
 
         return (
@@ -189,7 +393,7 @@ class AgentDispatcher:
             f"误报过滤: {status_count['FALSE']}条"
         )
 
-    async def _handle_export_report(self, user_id: str, params: Dict, raw: str) -> str:
+    async def _handle_export_report(self, user_id: str, params: Dict) -> str:
         """导出Excel报表"""
         from app.services.report_generator import generate_alarm_report
         from app.services.oss_storage import get_oss_storage
@@ -199,11 +403,9 @@ class AgentDispatcher:
 
         if not result:
             range_names = {"today": "今日", "week": "本周", "month": "本月"}
-            return f"{range_names.get(time_range, '今日')}暂无告警数据，无法生成报表。"
+            return f"{range_names.get(time_range, '今日')}暂无告警数据。"
 
         filename, file_bytes = result
-
-        # 上传到 COS 获取下载链接
         oss = get_oss_storage()
         try:
             object_key = oss.upload_file(
@@ -215,14 +417,7 @@ class AgentDispatcher:
             return f"报表已生成\n文件：{filename}\n下载：{download_url}"
         except Exception as e:
             logger.warning(f"报表上传COS失败: {e}")
-            return f"报表已生成（{filename}），但上传失败，请联系管理员。"
-
-    async def _handle_general_chat(self, user_id: str, params: Dict, raw: str) -> str:
-        """兜底回复"""
-        msg = params.get("message", "")
-        if msg:
-            return msg
-        return "您好，我是安防AI助手。可以帮您：\n1. 创建工单\n2. 查询告警统计\n3. 导出告警报表\n\n请直接描述您的需求。"
+            return f"报表生成成功（{filename}），但上传失败，请联系管理员。"
 
 
 # 全局单例