update:单模块部署xiaozhi-server，支持mcp调用摄像头识图

2026-07-27 01:23:55 +08:00 · 2025-06-01 02:26:19 +08:00
parent 3c3be950e9
commit c4f2411fee
15 changed files with 426 additions and 70 deletions
@@ -0,0 +1,16 @@
+from aiohttp import web
+from config.logger import setup_logging
+
+
+class BaseHandler:
+    def __init__(self, config: dict):
+        self.config = config
+        self.logger = setup_logging()
+
+    def _add_cors_headers(self, response):
+        """添加CORS头信息"""
+        response.headers["Access-Control-Allow-Headers"] = (
+            "client-id, content-type, device-id"
+        )
+        response.headers["Access-Control-Allow-Credentials"] = "true"
+        response.headers["Access-Control-Allow-Origin"] = "*"
@@ -1,18 +1,15 @@
 import json
 import time
-import asyncio
 from aiohttp import web
-from config.logger import setup_logging
 from core.utils.util import get_local_ip
-from core.utils.modules_initialize import initialize_modules
+from core.api.base_handler import BaseHandler

 TAG = __name__


-class SimpleOtaServer:
+class OTAHandler(BaseHandler):
    def __init__(self, config: dict):
-        self.config = config
-        self.logger = setup_logging()
+        super().__init__(config)

    def _get_websocket_url(self, local_ip: str, port: int) -> str:
        """获取websocket地址
@@ -25,41 +22,15 @@ class SimpleOtaServer:
            str: websocket地址
        """
        server_config = self.config["server"]
-        websocket_config = server_config.get("websocket")
+        websocket_config = server_config.get("websocket", "")

-        if websocket_config and "你" not in websocket_config:
+        if "你的" not in websocket_config:
            return websocket_config
        else:
            return f"ws://{local_ip}:{port}/xiaozhi/v1/"

-    async def start(self):
-        server_config = self.config["server"]
-        host = server_config.get("ip", "0.0.0.0")
-        port = int(server_config.get("ota_port"))
-
-        if port:
-            app = web.Application()
-            # 添加路由
-            app.add_routes(
-                [
-                    web.get("/xiaozhi/ota/", self._handle_ota_get_request),
-                    web.post("/xiaozhi/ota/", self._handle_ota_request),
-                    web.options("/xiaozhi/ota/", self._handle_ota_request),
-                ]
-            )
-
-            # 运行服务
-            runner = web.AppRunner(app)
-            await runner.setup()
-            site = web.TCPSite(runner, host, port)
-            await site.start()
-
-            # 保持服务运行
-            while True:
-                await asyncio.sleep(3600)  # 每隔 1 小时检查一次
-
-    async def _handle_ota_request(self, request):
-        """处理 /xiaozhi/ota/ 的 POST 请求"""
+    async def handle_post(self, request):
+        """处理 OTA POST 请求"""
        try:
            data = await request.text()
            self.logger.bind(tag=TAG).debug(f"OTA请求方法: {request.method}")
@@ -75,11 +46,9 @@ class SimpleOtaServer:
            data_json = json.loads(data)

            server_config = self.config["server"]
-            host = server_config.get("ip", "0.0.0.0")
            port = int(server_config.get("port", 8000))
            local_ip = get_local_ip()

-            # OTA基础信息
            return_json = {
                "server_time": {
                    "timestamp": int(round(time.time() * 1000)),
@@ -104,16 +73,11 @@ class SimpleOtaServer:
                content_type="application/json",
            )
        finally:
-            # 添加header，允许跨域访问
-            response.headers["Access-Control-Allow-Headers"] = (
-                "client-id, content-type, device-id"
-            )
-            response.headers["Access-Control-Allow-Credentials"] = "true"
-            response.headers["Access-Control-Allow-Origin"] = "*"
+            self._add_cors_headers(response)
            return response

-    async def _handle_ota_get_request(self, request):
-        """处理 /xiaozhi/ota/ 的 GET 请求"""
+    async def handle_get(self, request):
+        """处理 OTA GET 请求"""
        try:
            server_config = self.config["server"]
            local_ip = get_local_ip()
@@ -125,10 +89,5 @@ class SimpleOtaServer:
            self.logger.bind(tag=TAG).error(f"OTA GET请求异常: {e}")
            response = web.Response(text="OTA接口异常", content_type="text/plain")
        finally:
-            # 添加header，允许跨域访问
-            response.headers["Access-Control-Allow-Headers"] = (
-                "client-id, content-type, device-id"
-            )
-            response.headers["Access-Control-Allow-Credentials"] = "true"
-            response.headers["Access-Control-Allow-Origin"] = "*"
+            self._add_cors_headers(response)
            return response
@@ -0,0 +1,121 @@
+import json
+from aiohttp import web
+from config.logger import setup_logging
+from core.utils.util import get_vision_url, is_valid_image_file
+from core.utils.vllm import create_instance
+import base64
+
+TAG = __name__
+
+# 设置最大文件大小为5MB
+MAX_FILE_SIZE = 5 * 1024 * 1024
+
+
+class VisionHandler:
+    def __init__(self, config: dict):
+        self.config = config
+        self.logger = setup_logging()
+
+    def _create_error_response(self, message: str) -> dict:
+        """创建统一的错误响应格式"""
+        return {"success": False, "message": message}
+
+    async def handle_post(self, request):
+        """处理 MCP Vision POST 请求"""
+        try:
+            # 获取请求头信息
+            device_id = request.headers.get("Device-Id", "")
+            client_id = request.headers.get("Client-Id", "")
+            self.logger.bind(tag=TAG).debug(f"Device-Id: {device_id}")
+            self.logger.bind(tag=TAG).debug(f"Client-Id: {client_id}")
+
+            # 解析multipart/form-data请求
+            reader = await request.multipart()
+
+            # 读取question字段
+            question_field = await reader.next()
+            if question_field is None:
+                raise ValueError("缺少问题字段")
+            question = await question_field.text()
+            self.logger.bind(tag=TAG).debug(f"Question: {question}")
+
+            # 读取图片文件
+            image_field = await reader.next()
+            if image_field is None:
+                raise ValueError("缺少图片文件")
+
+            # 读取图片数据
+            image_data = await image_field.read()
+            if not image_data:
+                raise ValueError("图片数据为空")
+
+            # 检查文件大小
+            if len(image_data) > MAX_FILE_SIZE:
+                raise ValueError(
+                    f"图片大小超过限制，最大允许{MAX_FILE_SIZE/1024/1024}MB"
+                )
+
+            # 检查文件格式
+            if not is_valid_image_file(image_data):
+                raise ValueError(
+                    "不支持的文件格式，请上传有效的图片文件（支持JPEG、PNG、GIF、BMP、TIFF、WEBP格式）"
+                )
+
+            # 将图片转换为base64编码
+            image_base64 = base64.b64encode(image_data).decode("utf-8")
+
+            vllm_config = self.config["VLLM"]["ChatGLMVLLM"]
+            provider = create_instance("openai", vllm_config)
+            response = provider.response(question, image_base64)
+
+            return_json = {
+                "success": True,
+                "result": response,
+            }
+
+            response = web.Response(
+                text=json.dumps(return_json, separators=(",", ":")),
+                content_type="application/json",
+            )
+        except ValueError as e:
+            self.logger.bind(tag=TAG).error(f"MCP Vision POST请求异常: {e}")
+            return_json = self._create_error_response(str(e))
+            response = web.Response(
+                text=json.dumps(return_json, separators=(",", ":")),
+                content_type="application/json",
+            )
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"MCP Vision POST请求异常: {e}")
+            return_json = self._create_error_response("处理请求时发生错误")
+            response = web.Response(
+                text=json.dumps(return_json, separators=(",", ":")),
+                content_type="application/json",
+            )
+        finally:
+            self._add_cors_headers(response)
+            return response
+
+    async def handle_get(self, request):
+        """处理 MCP Vision GET 请求"""
+        try:
+            vision_explain = get_vision_url(self.config)
+            message = f"MCP Vision 接口运行正常，视觉解释接口地址是：{vision_explain}"
+            response = web.Response(text=message, content_type="text/plain")
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"MCP Vision GET请求异常: {e}")
+            return_json = self._create_error_response("服务器内部错误")
+            response = web.Response(
+                text=json.dumps(return_json, separators=(",", ":")),
+                content_type="application/json",
+            )
+        finally:
+            self._add_cors_headers(response)
+            return response
+
+    def _add_cors_headers(self, response):
+        """添加CORS头信息"""
+        response.headers["Access-Control-Allow-Headers"] = (
+            "client-id, content-type, device-id"
+        )
+        response.headers["Access-Control-Allow-Credentials"] = "true"
+        response.headers["Access-Control-Allow-Origin"] = "*"
@@ -708,15 +708,24 @@ class ConnectionHandler:
                # 处理Server端MCP工具调用
                if self.mcp_manager.is_mcp_tool(function_name):
                    result = self._handle_mcp_tool_call(function_call_data)
-                elif hasattr(self, "mcp_client") and self.mcp_client.has_tool(function_name):
-                    # 如果是小智端MCP工具调用 
+                elif hasattr(self, "mcp_client") and self.mcp_client.has_tool(
+                    function_name
+                ):
+                    # 如果是小智端MCP工具调用
                    self.logger.bind(tag=TAG).debug(
                        f"调用小智端MCP工具: {function_name}, 参数: {function_arguments}"
                    )
                    try:
-                        result = asyncio.run_coroutine_threadsafe(call_mcp_tool(self, self.mcp_client, function_name, function_arguments), self.loop).result()
+                        result = asyncio.run_coroutine_threadsafe(
+                            call_mcp_tool(
+                                self, self.mcp_client, function_name, function_arguments
+                            ),
+                            self.loop,
+                        ).result()
                        self.logger.bind(tag=TAG).debug(f"MCP工具调用结果: {result}")
-                        result = ActionResponse(action=Action.REQLLM, result=result, response="")
+                        result = ActionResponse(
+                            action=Action.REQLLM, result=result, response=""
+                        )
                    except Exception as e:
                        self.logger.bind(tag=TAG).error(f"MCP工具调用失败: {e}")
                        result = ActionResponse(
@@ -4,10 +4,14 @@ import json
 import random
 import shutil
 import asyncio
-from core.handle.mcpHandle import MCPClient, send_mcp_initialize_message, send_mcp_tools_list_request
 from core.handle.sendAudioHandle import send_stt_message
 from core.utils.util import remove_punctuation_and_length
 from core.providers.tts.dto.dto import ContentType, InterfaceType
+from core.handle.mcpHandle import (
+    MCPClient,
+    send_mcp_initialize_message,
+    send_mcp_tools_list_request,
+)


 TAG = __name__
@@ -1,6 +1,7 @@
 import json
 import asyncio
 from concurrent.futures import Future
+from core.utils.util import get_vision_url

 TAG = __name__

@@ -205,6 +206,17 @@ async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict):

 async def send_mcp_initialize_message(conn):
    """发送MCP初始化消息"""
+
+    # 智控台暂时不启动视觉分析
+    if conn.read_config_from_api:
+        vision = {}
+    else:
+        vision_url = get_vision_url(conn.config)
+        vision = {
+            "url": vision_url,
+            "token": "test_token",
+        }
+
    payload = {
        "jsonrpc": "2.0",
        "id": 1,  # mcpInitializeID
@@ -214,6 +226,7 @@ async def send_mcp_initialize_message(conn):
            "capabilities": {
                "roots": {"listChanged": True},
                "sampling": {},
+                "vision": vision,
            },
            "clientInfo": {
                "name": "XiaozhiClient",
@@ -0,0 +1,62 @@
+import asyncio
+from aiohttp import web
+from config.logger import setup_logging
+from core.api.ota_handler import OTAHandler
+from core.api.vision_handler import VisionHandler
+
+TAG = __name__
+
+
+class SimpleHttpServer:
+    def __init__(self, config: dict):
+        self.config = config
+        self.logger = setup_logging()
+        self.ota_handler = OTAHandler(config)
+        self.vision_handler = VisionHandler(config)
+
+    def _get_websocket_url(self, local_ip: str, port: int) -> str:
+        """获取websocket地址
+
+        Args:
+            local_ip: 本地IP地址
+            port: 端口号
+
+        Returns:
+            str: websocket地址
+        """
+        server_config = self.config["server"]
+        websocket_config = server_config.get("websocket")
+
+        if websocket_config and "你" not in websocket_config:
+            return websocket_config
+        else:
+            return f"ws://{local_ip}:{port}/xiaozhi/v1/"
+
+    async def start(self):
+        server_config = self.config["server"]
+        host = server_config.get("ip", "0.0.0.0")
+        port = int(server_config.get("http_port"))
+
+        if port:
+            app = web.Application()
+            # 添加路由
+            app.add_routes(
+                [
+                    web.get("/xiaozhi/ota/", self.ota_handler.handle_get),
+                    web.post("/xiaozhi/ota/", self.ota_handler.handle_post),
+                    web.options("/xiaozhi/ota/", self.ota_handler.handle_post),
+                    web.get("/mcp/vision/explain", self.vision_handler.handle_get),
+                    web.post("/mcp/vision/explain", self.vision_handler.handle_post),
+                    web.options("/mcp/vision/explain", self.vision_handler.handle_post),
+                ]
+            )
+
+            # 运行服务
+            runner = web.AppRunner(app)
+            await runner.setup()
+            site = web.TCPSite(runner, host, port)
+            await site.start()
+
+            # 保持服务运行
+            while True:
+                await asyncio.sleep(3600)  # 每隔 1 小时检查一次
@@ -0,0 +1,12 @@
+from abc import ABC, abstractmethod
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class VLLMProviderBase(ABC):
+    @abstractmethod
+    def response(self, question, base64_image):
+        """VLLM response generator"""
+        pass
@@ -0,0 +1,63 @@
+import openai
+import json
+from config.logger import setup_logging
+from core.utils.util import check_model_key
+from core.providers.vllm.base import VLLMProviderBase
+
+TAG = __name__
+logger = setup_logging()
+
+
+class VLLMProvider(VLLMProviderBase):
+    def __init__(self, config):
+        self.model_name = config.get("model_name")
+        self.api_key = config.get("api_key")
+        if "base_url" in config:
+            self.base_url = config.get("base_url")
+        else:
+            self.base_url = config.get("url")
+
+        param_defaults = {
+            "max_tokens": (500, int),
+            "temperature": (0.7, lambda x: round(float(x), 1)),
+            "top_p": (1.0, lambda x: round(float(x), 1)),
+        }
+
+        for param, (default, converter) in param_defaults.items():
+            value = config.get(param)
+            try:
+                setattr(
+                    self,
+                    param,
+                    converter(value) if value not in (None, "") else default,
+                )
+            except (ValueError, TypeError):
+                setattr(self, param, default)
+
+        check_model_key("VLLM", self.api_key)
+        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url)
+
+    def response(self, question, base64_image):
+        try:
+            messages = [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": question},
+                        {
+                            "type": "image_url",
+                            "image_url": {"url": f"{base64_image}"},
+                        },
+                    ],
+                }
+            ]
+
+            response = self.client.chat.completions.create(
+                model=self.model_name, messages=messages, stream=False
+            )
+
+            return response.choices[0].message.content
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"Error in response generation: {e}")
+            raise
@@ -882,3 +882,51 @@ def filter_sensitive_info(config: dict) -> dict:
        return filtered

    return _filter_dict(copy.deepcopy(config))
+
+
+def get_vision_url(config: dict) -> str:
+    """获取 vision URL
+
+    Args:
+        config: 配置字典
+
+    Returns:
+        str: vision URL
+    """
+    server_config = config["server"]
+    vision_explain = server_config.get("vision_explain", "")
+    if "你的" in vision_explain:
+        local_ip = get_local_ip()
+        port = int(server_config.get("http_port", 8003))
+        vision_explain = f"http://{local_ip}:{port}/mcp/vision/explain"
+    return vision_explain
+
+
+def is_valid_image_file(file_data: bytes) -> bool:
+    """
+    检查文件数据是否为有效的图片格式
+
+    Args:
+        file_data: 文件的二进制数据
+
+    Returns:
+        bool: 如果是有效的图片格式返回True，否则返回False
+    """
+    # 常见图片格式的魔数（文件头）
+    image_signatures = {
+        b"\xff\xd8\xff": "JPEG",
+        b"\x89PNG\r\n\x1a\n": "PNG",
+        b"GIF87a": "GIF",
+        b"GIF89a": "GIF",
+        b"BM": "BMP",
+        b"II*\x00": "TIFF",
+        b"MM\x00*": "TIFF",
+        b"RIFF": "WEBP",
+    }
+
+    # 检查文件头是否匹配任何已知的图片格式
+    for signature in image_signatures:
+        if file_data.startswith(signature):
+            return True
+
+    return False
@@ -0,0 +1,23 @@
+import os
+import sys
+
+# 添加项目根目录到Python路径
+current_dir = os.path.dirname(os.path.abspath(__file__))
+project_root = os.path.abspath(os.path.join(current_dir, "..", ".."))
+sys.path.insert(0, project_root)
+
+from config.logger import setup_logging
+import importlib
+
+logger = setup_logging()
+
+
+def create_instance(class_name, *args, **kwargs):
+    # 创建LLM实例
+    if os.path.exists(os.path.join("core", "providers", "vllm", f"{class_name}.py")):
+        lib_name = f"core.providers.vllm.{class_name}"
+        if lib_name not in sys.modules:
+            sys.modules[lib_name] = importlib.import_module(f"{lib_name}")
+        return sys.modules[lib_name].VLLMProvider(*args, **kwargs)
+
+    raise ValueError(f"不支持的VLLM类型: {class_name}，请检查该配置的type是否设置正确")