chore: bump version 4.6.3

* feat: add voice and file supports for wecom

* feat: add   and  in query variables

* feat: supports for lark recv file message

* feat: kook recv voice msg

* feat: supports for Voice and File in discord

* chore: remove debug msg

* perf: remove unnecessary bot logs

* feat: implement bot log filtering and per label color (#1839)

* feat: add sender_name and group_name in query variables

docker/docker-compose.yaml

@@ -25,7 +25,6 @@ services:
     platform: linux/amd64 # For Apple Silicon compatibility
     volumes:
       - ./data:/app/data
-      - ./plugins:/app/plugins
     restart: on-failure
     environment:
       - TZ=Asia/Shanghai

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.6.2"
+version = "4.6.3"
 description = "Easy-to-use global IM bot platform designed for LLM era"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -63,11 +63,13 @@ dependencies = [
     "langchain-text-splitters>=0.0.1",
     "chromadb>=0.4.24",
     "qdrant-client (>=1.15.1,<2.0.0)",
-    "langbot-plugin==0.2.0",
+    "langbot-plugin==0.2.1",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "tboxsdk>=0.0.10",
     "boto3>=1.35.0",
+    "pymilvus>=2.6.4",
+    "pgvector>=0.4.1",
 ]
 keywords = [
     "bot",

src/langbot/__init__.py

@@ -1,3 +1,3 @@
 """LangBot - Easy-to-use global IM bot platform designed for LLM era"""
 
-__version__ = '4.6.2'
+__version__ = '4.6.3'

src/langbot/libs/wecom_ai_bot_api/api.py

@@ -394,7 +394,6 @@ class WecomBotClient:
         """
         try:
             self.wxcpt = WXBizMsgCrypt(self.Token, self.EnCodingAESKey, '')
-            await self.logger.info(f'{req.method} {req.url} {str(req.args)}')
 
             if req.method == 'GET':
                 return await self._handle_get_callback(req)
@@ -458,32 +457,174 @@ class WecomBotClient:
     async def get_message(self, msg_json):
         message_data = {}
 
+        msg_type = msg_json.get('msgtype', '')
+        if msg_type:
+            message_data['msgtype'] = msg_type
+
         if msg_json.get('chattype', '') == 'single':
             message_data['type'] = 'single'
         elif msg_json.get('chattype', '') == 'group':
             message_data['type'] = 'group'
 
-        if msg_json.get('msgtype') == 'text':
+        max_inline_file_size = 5 * 1024 * 1024  # avoid decoding very large payloads by default
+
+        async def _safe_download(url: str):
+            if not url:
+                return None
+            return await self.download_url_to_base64(url, self.EnCodingAESKey)
+
+        if msg_type == 'text':
             message_data['content'] = msg_json.get('text', {}).get('content')
-        elif msg_json.get('msgtype') == 'image':
+        elif msg_type == 'markdown':
+            message_data['content'] = msg_json.get('markdown', {}).get('content') or msg_json.get('text', {}).get(
+                'content', ''
+            )
+        elif msg_type == 'image':
             picurl = msg_json.get('image', {}).get('url', '')
-            base64 = await self.download_url_to_base64(picurl, self.EnCodingAESKey)
-            message_data['picurl'] = base64
-        elif msg_json.get('msgtype') == 'mixed':
+            base64_data = await _safe_download(picurl)
+            if base64_data:
+                message_data['picurl'] = base64_data
+                message_data['images'] = [base64_data]
+        elif msg_type == 'voice':
+            voice_info = msg_json.get('voice', {}) or {}
+            download_url = voice_info.get('url')
+            message_data['voice'] = {
+                'url': download_url,
+                'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
+                'filesize': voice_info.get('filesize') or voice_info.get('size'),
+                'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
+            }
+            # 企业微信智能转写文本（如果已有）直接复用，避免重复转写
+            if voice_info.get('content'):
+                message_data['content'] = voice_info.get('content')
+            if (message_data['voice'].get('filesize') or 0) <= max_inline_file_size:
+                voice_base64 = await _safe_download(download_url)
+                if voice_base64:
+                    message_data['voice']['base64'] = voice_base64
+        elif msg_type == 'video':
+            video_info = msg_json.get('video', {}) or {}
+            download_url = video_info.get('url')
+            video_data = {
+                'url': download_url,
+                'filesize': video_info.get('filesize') or video_info.get('size'),
+                'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
+                'md5sum': video_info.get('md5sum') or video_info.get('md5'),
+                'filename': video_info.get('filename') or video_info.get('name'),
+            }
+            if (video_data.get('filesize') or 0) <= max_inline_file_size:
+                video_base64 = await _safe_download(download_url)
+                if video_base64:
+                    video_data['base64'] = video_base64
+            message_data['video'] = video_data
+        elif msg_type == 'file':
+            file_info = msg_json.get('file', {}) or {}
+            download_url = file_info.get('url') or file_info.get('fileurl')
+            file_data = {
+                'filename': file_info.get('filename') or file_info.get('name'),
+                'filesize': file_info.get('filesize') or file_info.get('size'),
+                'md5sum': file_info.get('md5sum') or file_info.get('md5'),
+                'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
+                'download_url': download_url,
+                'extra': file_info,
+            }
+            if (file_data.get('filesize') or 0) <= max_inline_file_size:
+                file_base64 = await _safe_download(download_url)
+                if file_base64:
+                    file_data['base64'] = file_base64
+            message_data['file'] = file_data
+        elif msg_type == 'link':
+            message_data['link'] = msg_json.get('link', {})
+            if not message_data.get('content'):
+                title = message_data['link'].get('title', '')
+                desc = message_data['link'].get('description') or message_data['link'].get('digest', '')
+                message_data['content'] = '\n'.join(filter(None, [title, desc]))
+        elif msg_type == 'mixed':
             items = msg_json.get('mixed', {}).get('msg_item', [])
             texts = []
-            picurl = None
+            images = []
+            files = []
+            voices = []
+            videos = []
+            links = []
             for item in items:
-                if item.get('msgtype') == 'text':
+                item_type = item.get('msgtype')
+                if item_type == 'text':
                     texts.append(item.get('text', {}).get('content', ''))
-                elif item.get('msgtype') == 'image' and picurl is None:
-                    picurl = item.get('image', {}).get('url')
+                elif item_type == 'image':
+                    img_url = item.get('image', {}).get('url')
+                    base64_data = await _safe_download(img_url)
+                    if base64_data:
+                        images.append(base64_data)
+                elif item_type == 'file':
+                    file_info = item.get('file', {}) or {}
+                    download_url = file_info.get('url') or file_info.get('fileurl')
+                    file_data = {
+                        'filename': file_info.get('filename') or file_info.get('name'),
+                        'filesize': file_info.get('filesize') or file_info.get('size'),
+                        'md5sum': file_info.get('md5sum') or file_info.get('md5'),
+                        'sdkfileid': file_info.get('sdkfileid') or file_info.get('fileid'),
+                        'download_url': download_url,
+                        'extra': file_info,
+                    }
+                    if (file_data.get('filesize') or 0) <= max_inline_file_size:
+                        file_base64 = await _safe_download(download_url)
+                        if file_base64:
+                            file_data['base64'] = file_base64
+                    files.append(file_data)
+                elif item_type == 'voice':
+                    voice_info = item.get('voice', {}) or {}
+                    download_url = voice_info.get('url')
+                    voice_data = {
+                        'url': download_url,
+                        'md5sum': voice_info.get('md5sum') or voice_info.get('md5'),
+                        'filesize': voice_info.get('filesize') or voice_info.get('size'),
+                        'sdkfileid': voice_info.get('sdkfileid') or voice_info.get('fileid'),
+                    }
+                    if voice_info.get('content'):
+                        texts.append(voice_info.get('content'))
+                    if (voice_data.get('filesize') or 0) <= max_inline_file_size:
+                        voice_base64 = await _safe_download(download_url)
+                        if voice_base64:
+                            voice_data['base64'] = voice_base64
+                    voices.append(voice_data)
+                elif item_type == 'video':
+                    video_info = item.get('video', {}) or {}
+                    download_url = video_info.get('url')
+                    video_data = {
+                        'url': download_url,
+                        'filesize': video_info.get('filesize') or video_info.get('size'),
+                        'sdkfileid': video_info.get('sdkfileid') or video_info.get('fileid'),
+                        'md5sum': video_info.get('md5sum') or video_info.get('md5'),
+                        'filename': video_info.get('filename') or video_info.get('name'),
+                    }
+                    if (video_data.get('filesize') or 0) <= max_inline_file_size:
+                        video_base64 = await _safe_download(download_url)
+                        if video_base64:
+                            video_data['base64'] = video_base64
+                    videos.append(video_data)
+                elif item_type == 'link':
+                    links.append(item.get('link', {}))
 
             if texts:
-                message_data['content'] = ''.join(texts)  # 拼接所有 text
-            if picurl:
-                base64 = await self.download_url_to_base64(picurl, self.EnCodingAESKey)
-                message_data['picurl'] = base64  # 只保留第一个 image
+                message_data['content'] = ' '.join(texts)  # 拼接所有 text
+            if images:
+                message_data['images'] = images
+                message_data['picurl'] = images[0]  # 只保留第一个 image
+            if files:
+                message_data['files'] = files
+                message_data['file'] = files[0]
+            if voices:
+                message_data['voices'] = voices
+                message_data['voice'] = voices[0]
+            if videos:
+                message_data['videos'] = videos
+                message_data['video'] = videos[0]
+            if links:
+                message_data['link'] = links[0]
+            if items:
+                message_data['attachments'] = items
+        else:
+            message_data['raw_msg'] = msg_json
 
         # Extract user information
         from_info = msg_json.get('from', {})

src/langbot/libs/wecom_ai_bot_api/wecombotevent.py

@@ -17,6 +17,13 @@ class WecomBotEvent(dict):
         """
         return self.get('type', '')
 
+    @property
+    def msgtype(self) -> str:
+        """
+        消息 msgtype
+        """
+        return self.get('msgtype', '')
+
     @property
     def userid(self) -> str:
         """
@@ -52,6 +59,55 @@ class WecomBotEvent(dict):
         """
         return self.get('picurl', '')
 
+    @property
+    def images(self):
+        """
+        图片列表（兼容 mixed）
+        """
+        return self.get('images', [])
+
+    @property
+    def file(self):
+        """
+        文件信息
+        """
+        return self.get('file', {})
+
+    @property
+    def voice(self):
+        """
+        语音信息
+        """
+        return self.get('voice', {})
+
+    @property
+    def video(self):
+        """
+        视频信息
+        """
+        return self.get('video', {})
+
+    @property
+    def link(self):
+        """
+        链接消息信息
+        """
+        return self.get('link', {})
+
+    @property
+    def location(self):
+        """
+        位置信息
+        """
+        return self.get('location', {})
+
+    @property
+    def attachments(self):
+        """
+        原始 mixed 中的附件项
+        """
+        return self.get('attachments', [])
+
     @property
     def chatid(self) -> str:
         """

src/langbot/libs/wecom_api/api.py

@@ -139,6 +139,58 @@ class WecomClient:
                 await self.logger.error(f'发送图片失败:{data}')
                 raise Exception('Failed to send image: ' + str(data))
 
+    async def send_voice(self, user_id: str, agent_id: int, media_id: str):
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+        url = self.base_url + '/message/send?access_token=' + self.access_token
+        async with httpx.AsyncClient() as client:
+            params = {
+                'touser': user_id,
+                'msgtype': 'voice',
+                'agentid': agent_id,
+                'voice': {
+                    'media_id': media_id,
+                },
+                'safe': 0,
+                'enable_id_trans': 0,
+                'enable_duplicate_check': 0,
+                'duplicate_check_interval': 1800,
+            }
+            response = await client.post(url, json=params)
+            data = response.json()
+            if data['errcode'] == 40014 or data['errcode'] == 42001:
+                self.access_token = await self.get_access_token(self.secret)
+                return await self.send_voice(user_id, agent_id, media_id)
+            if data['errcode'] != 0:
+                await self.logger.error(f'发送语音失败:{data}')
+                raise Exception('Failed to send voice: ' + str(data))
+
+    async def send_file(self, user_id: str, agent_id: int, media_id: str):
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+        url = self.base_url + '/message/send?access_token=' + self.access_token
+        async with httpx.AsyncClient() as client:
+            params = {
+                'touser': user_id,
+                'msgtype': 'file',
+                'agentid': agent_id,
+                'file': {
+                    'media_id': media_id,
+                },
+                'safe': 0,
+                'enable_id_trans': 0,
+                'enable_duplicate_check': 0,
+                'duplicate_check_interval': 1800,
+            }
+            response = await client.post(url, json=params)
+            data = response.json()
+            if data['errcode'] == 40014 or data['errcode'] == 42001:
+                self.access_token = await self.get_access_token(self.secret)
+                return await self.send_file(user_id, agent_id, media_id)
+            if data['errcode'] != 0:
+                await self.logger.error(f'发送文件失败:{data}')
+                raise Exception('Failed to send file: ' + str(data))
+
     async def send_private_msg(self, user_id: str, agent_id: int, content: str):
         if not await self.check_access_token():
             self.access_token = await self.get_access_token(self.secret)
@@ -287,7 +339,7 @@ class WecomClient:
                 return ext
         return 'jpg'  # 默认返回jpg
 
-    async def upload_to_work(self, image: platform_message.Image):
+    async def upload_image_to_work(self, image: platform_message.Image):
         """
         获取 media_id
         """
@@ -304,7 +356,7 @@ class WecomClient:
                 file_bytes = await f.read()
                 file_name = image.path.split('/')[-1]
         elif image.url:
-            file_bytes = await self.download_image_to_bytes(image.url)
+            file_bytes = await self.download_media_to_bytes(image.url)
             file_name = image.url.split('/')[-1]
         elif image.base64:
             try:
@@ -339,7 +391,7 @@ class WecomClient:
             data = response.json()
             if data['errcode'] == 40014 or data['errcode'] == 42001:
                 self.access_token = await self.get_access_token(self.secret)
-                media_id = await self.upload_to_work(image)
+                media_id = await self.upload_image_to_work(image)
             if data.get('errcode', 0) != 0:
                 await self.logger.error(f'上传图片失败:{data}')
                 raise Exception('failed to upload file')
@@ -347,13 +399,128 @@ class WecomClient:
             media_id = data.get('media_id')
             return media_id
 
-    async def download_image_to_bytes(self, url: str) -> bytes:
+    async def upload_voice_to_work(self, voice: platform_message.Voice):
+        """
+        上传语音文件到企业微信
+        """
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+        url = self.base_url + '/media/upload?access_token=' + self.access_token + '&type=file'
+        file_bytes = None
+        file_name = 'voice.mp3'
+
+        if voice.path:
+            async with aiofiles.open(voice.path, 'rb') as f:
+                file_bytes = await f.read()
+                file_name = voice.path.split('/')[-1]
+        elif voice.url:
+            file_bytes = await self.download_media_to_bytes(voice.url)
+            file_name = voice.url.split('/')[-1]
+        elif voice.base64:
+            try:
+                base64_data = voice.base64
+                if ',' in base64_data:
+                    base64_data = base64_data.split(',', 1)[1]
+                padding = 4 - (len(base64_data) % 4) if len(base64_data) % 4 else 0
+                padded_base64 = base64_data + '=' * padding
+                file_bytes = base64.b64decode(padded_base64)
+            except binascii.Error as e:
+                raise ValueError(f'Invalid base64 string: {str(e)}')
+        else:
+            await self.logger.error('Voice对象出错')
+            raise ValueError('voice对象出错')
+
+        boundary = '-------------------------acebdf13572468'
+        headers = {'Content-Type': f'multipart/form-data; boundary={boundary}'}
+        body = (
+            (
+                f'--{boundary}\r\n'
+                f'Content-Disposition: form-data; name="media"; filename="{file_name}"; filelength={len(file_bytes)}\r\n'
+                f'Content-Type: application/octet-stream\r\n\r\n'
+            ).encode('utf-8')
+            + file_bytes
+            + f'\r\n--{boundary}--\r\n'.encode('utf-8')
+        )
+
+        # print(body)
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, headers=headers, content=body)
+            data = response.json()
+            if data['errcode'] == 40014 or data['errcode'] == 42001:
+                self.access_token = await self.get_access_token(self.secret)
+                media_id = await self.upload_voice_to_work(voice)
+            if data.get('errcode', 0) != 0:
+                await self.logger.error(f'上传语音文件失败:{data}')
+                raise Exception('failed to upload file')
+            media_id = data.get('media_id')
+            return media_id
+
+    async def upload_file_to_work(self, file: platform_message.File):
+        """
+        上传文件到企业微信
+        """
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+        url = self.base_url + '/media/upload?access_token=' + self.access_token + '&type=file'
+        file_bytes = None
+        file_name = 'file.txt'
+        if file.path:
+            async with aiofiles.open(file.path, 'rb') as f:
+                file_bytes = await f.read()
+                file_name = file.path.split('/')[-1]
+        elif file.url:
+            file_bytes = await self.download_media_to_bytes(file.url)
+            file_name = file.url.split('/')[-1]
+        elif file.base64:
+            try:
+                base64_data = file.base64
+                if ',' in base64_data:
+                    base64_data = base64_data.split(',', 1)[1]
+                padding = 4 - (len(base64_data) % 4) if len(base64_data) % 4 else 0
+                padded_base64 = base64_data + '=' * padding
+                file_bytes = base64.b64decode(padded_base64)
+            except binascii.Error as e:
+                raise ValueError(f'Invalid base64 string: {str(e)}')
+        else:
+            await self.logger.error('File对象出错')
+            raise ValueError('file对象出错')
+        boundary = '-------------------------acebdf13572468'
+        headers = {'Content-Type': f'multipart/form-data; boundary={boundary}'}
+        body = (
+            (
+                f'--{boundary}\r\n'
+                f'Content-Disposition: form-data; name="media"; filename="{file_name}"; filelength={len(file_bytes)}\r\n'
+                f'Content-Type: application/octet-stream\r\n\r\n'
+            ).encode('utf-8')
+            + file_bytes
+            + f'\r\n--{boundary}--\r\n'.encode('utf-8')
+        )
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, headers=headers, content=body)
+            data = response.json()
+            if data['errcode'] == 40014 or data['errcode'] == 42001:
+                self.access_token = await self.get_access_token(self.secret)
+                media_id = await self.upload_file_to_work(file)
+            if data.get('errcode', 0) != 0:
+                await self.logger.error(f'上传文件失败:{data}')
+                raise Exception('failed to upload file')
+            media_id = data.get('media_id')
+            return media_id
+
+    async def download_media_to_bytes(self, url: str) -> bytes:
         async with httpx.AsyncClient() as client:
             response = await client.get(url)
             response.raise_for_status()
             return response.content
 
     # 进行media_id的获取
-    async def get_media_id(self, image: platform_message.Image):
-        media_id = await self.upload_to_work(image=image)
+    async def get_media_id(self, media: platform_message.Image | platform_message.Voice | platform_message.File):
+        if isinstance(media, platform_message.Image):
+            media_id = await self.upload_image_to_work(image=media)
+        elif isinstance(media, platform_message.Voice):
+            media_id = await self.upload_voice_to_work(voice=media)
+        elif isinstance(media, platform_message.File):
+            media_id = await self.upload_file_to_work(file=media)
+        else:
+            raise ValueError('Unsupported media type')
         return media_id

src/langbot/pkg/api/http/controller/main.py

@@ -92,7 +92,11 @@ class HTTPController:
 
         @self.quart_app.route('/')
         async def index():
-            return await quart.send_from_directory(frontend_path, 'index.html', mimetype='text/html')
+            response = await quart.send_from_directory(frontend_path, 'index.html', mimetype='text/html')
+            response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
+            response.headers['Pragma'] = 'no-cache'
+            response.headers['Expires'] = '0'
+            return response
 
         @self.quart_app.route('/<path:path>')
         async def static_file(path: str):

src/langbot/pkg/persistence/migrations/dbm002_combine_quote_msg_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(2)
@@ -11,30 +10,45 @@ class DBMigrateCombineQuoteMsgConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure 'trigger' exists
+            if 'trigger' not in config:
+                config['trigger'] = {}
+
+            # Ensure 'misc' exists in 'trigger'
             if 'misc' not in config['trigger']:
                 config['trigger']['misc'] = {}
 
+            # Add 'combine-quote-message' if not exists
             if 'combine-quote-message' not in config['trigger']['misc']:
                 config['trigger']['misc']['combine-quote-message'] = False
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm003_n8n_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(3)
@@ -11,14 +10,23 @@ class DBMigrateN8nConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure 'ai' exists
+            if 'ai' not in config:
+                config['ai'] = {}
+
+            # Add 'n8n-service-api' if not exists
             if 'n8n-service-api' not in config['ai']:
                 config['ai']['n8n-service-api'] = {
                     'webhook-url': 'http://your-n8n-webhook-url',
@@ -33,16 +41,21 @@ class DBMigrateN8nConfig(migration.DBMigration):
                     'output-key': 'response',
                 }
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm004_rag_kb_uuid.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(4)
@@ -11,27 +10,43 @@ class DBMigrateRAGKBUUID(migration.DBMigration):
 
     async def upgrade(self):
         """升级"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure nested structure exists
+            if 'ai' not in config:
+                config['ai'] = {}
+            if 'local-agent' not in config['ai']:
+                config['ai']['local-agent'] = {}
+
+            # Add 'knowledge-base' if not exists
             if 'knowledge-base' not in config['ai']['local-agent']:
                 config['ai']['local-agent']['knowledge-base'] = ''
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """降级"""

src/langbot/pkg/persistence/migrations/dbm005_pipeline_remove_cot_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(5)
@@ -11,27 +10,43 @@ class DBMigratePipelineRemoveCotConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure nested structure exists
+            if 'output' not in config:
+                config['output'] = {}
+            if 'misc' not in config['output']:
+                config['output']['misc'] = {}
+
+            # Add 'remove-think' if not exists
             if 'remove-think' not in config['output']['misc']:
                 config['output']['misc']['remove-think'] = False
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm006_langflow_api_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(6)
@@ -11,14 +10,23 @@ class DBMigrateLangflowApiConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure 'ai' exists
+            if 'ai' not in config:
+                config['ai'] = {}
+
+            # Add 'langflow-api' if not exists
             if 'langflow-api' not in config['ai']:
                 config['ai']['langflow-api'] = {
                     'base-url': 'http://localhost:7860',
@@ -29,16 +37,21 @@ class DBMigrateLangflowApiConfig(migration.DBMigration):
                     'tweaks': '{}',
                 }
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm010_pipeline_multi_knowledge_base.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(10)
@@ -11,16 +10,20 @@ class DBMigratePipelineMultiKnowledgeBase(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
             # Convert knowledge-base from string to array
-            if 'local-agent' in config['ai']:
+            if 'ai' in config and 'local-agent' in config['ai']:
                 current_kb = config['ai']['local-agent'].get('knowledge-base', '')
 
                 # If it's already a list, skip
@@ -37,29 +40,38 @@ class DBMigratePipelineMultiKnowledgeBase(migration.DBMigration):
                 if 'knowledge-base' in config['ai']['local-agent']:
                     del config['ai']['local-agent']['knowledge-base']
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
-                )
-            )
+                # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+                if self.ap.persistence_mgr.db.name == 'postgresql':
+                    await self.ap.persistence_mgr.execute_async(
+                        sqlalchemy.text(
+                            'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                        ),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                    )
+                else:
+                    await self.ap.persistence_mgr.execute_async(
+                        sqlalchemy.text(
+                            'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                        ),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                    )
 
     async def downgrade(self):
         """Downgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
             # Convert knowledge-bases from array back to string
-            if 'local-agent' in config['ai']:
+            if 'ai' in config and 'local-agent' in config['ai']:
                 current_kbs = config['ai']['local-agent'].get('knowledge-bases', [])
 
                 # If it's already a string, skip
@@ -76,13 +88,18 @@ class DBMigratePipelineMultiKnowledgeBase(migration.DBMigration):
                 if 'knowledge-bases' in config['ai']['local-agent']:
                     del config['ai']['local-agent']['knowledge-bases']
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
-                )
-            )
+                # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+                if self.ap.persistence_mgr.db.name == 'postgresql':
+                    await self.ap.persistence_mgr.execute_async(
+                        sqlalchemy.text(
+                            'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                        ),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                    )
+                else:
+                    await self.ap.persistence_mgr.execute_async(
+                        sqlalchemy.text(
+                            'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                        ),
+                        {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                    )

src/langbot/pkg/persistence/migrations/dbm011_dify_base_prompt_config.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(11)
@@ -11,29 +10,45 @@ class DBMigrateDifyApiConfig(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            config = serialized_pipeline['config']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            config = json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
 
+            # Ensure nested structure exists
+            if 'ai' not in config:
+                config['ai'] = {}
+            if 'dify-service-api' not in config['ai']:
+                config['ai']['dify-service-api'] = {}
+
+            # Add 'base-prompt' if not exists
             if 'base-prompt' not in config['ai']['dify-service-api']:
                 config['ai']['dify-service-api']['base-prompt'] = (
                     'When the file content is readable, please read the content of this file. When the file is an image, describe the content of this image.',
                 )
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    {
-                        'config': config,
-                        'for_version': self.ap.ver_mgr.get_current_version(),
-                    }
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET config = :config, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {'config': json.dumps(config), 'for_version': current_version, 'uuid': uuid},
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/persistence/migrations/dbm012_pipeline_extensions_enable_all.py

@@ -1,8 +1,7 @@
 from .. import migration
 
 import sqlalchemy
-
-from ...entity.persistence import pipeline as persistence_pipeline
+import json
 
 
 @migration.migration_class(12)
@@ -11,14 +10,25 @@ class DBMigratePipelineExtensionsEnableAll(migration.DBMigration):
 
     async def upgrade(self):
         """Upgrade"""
-        # read all pipelines
-        pipelines = await self.ap.persistence_mgr.execute_async(sqlalchemy.select(persistence_pipeline.LegacyPipeline))
+        # Read all pipelines using raw SQL
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, extensions_preferences FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
 
-        for pipeline in pipelines:
-            serialized_pipeline = self.ap.persistence_mgr.serialize_model(persistence_pipeline.LegacyPipeline, pipeline)
+        current_version = self.ap.ver_mgr.get_current_version()
 
-            extensions_preferences = serialized_pipeline['extensions_preferences']
+        for pipeline_row in pipelines:
+            uuid = pipeline_row[0]
+            extensions_preferences = (
+                json.loads(pipeline_row[1]) if isinstance(pipeline_row[1], str) else pipeline_row[1]
+            )
 
+            # Ensure extensions_preferences is a dict
+            if extensions_preferences is None:
+                extensions_preferences = {}
+
+            # Add 'enable_all_plugins' if not exists
             if 'enable_all_plugins' not in extensions_preferences:
                 if 'plugins' in extensions_preferences:
                     extensions_preferences['enable_all_plugins'] = False
@@ -26,6 +36,7 @@ class DBMigratePipelineExtensionsEnableAll(migration.DBMigration):
                     extensions_preferences['enable_all_plugins'] = True
                     extensions_preferences['plugins'] = []
 
+            # Add 'enable_all_mcp_servers' if not exists
             if 'enable_all_mcp_servers' not in extensions_preferences:
                 if 'mcp_servers' in extensions_preferences:
                     extensions_preferences['enable_all_mcp_servers'] = False
@@ -33,14 +44,29 @@ class DBMigratePipelineExtensionsEnableAll(migration.DBMigration):
                     extensions_preferences['enable_all_mcp_servers'] = True
                     extensions_preferences['mcp_servers'] = []
 
-            await self.ap.persistence_mgr.execute_async(
-                sqlalchemy.update(persistence_pipeline.LegacyPipeline)
-                .where(persistence_pipeline.LegacyPipeline.uuid == serialized_pipeline['uuid'])
-                .values(
-                    extensions_preferences=extensions_preferences,
-                    for_version=self.ap.ver_mgr.get_current_version(),
+            # Update using raw SQL with compatibility for both SQLite and PostgreSQL
+            if self.ap.persistence_mgr.db.name == 'postgresql':
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET extensions_preferences = :extensions_preferences::jsonb, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {
+                        'extensions_preferences': json.dumps(extensions_preferences),
+                        'for_version': current_version,
+                        'uuid': uuid,
+                    },
+                )
+            else:
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.text(
+                        'UPDATE legacy_pipelines SET extensions_preferences = :extensions_preferences, for_version = :for_version WHERE uuid = :uuid'
+                    ),
+                    {
+                        'extensions_preferences': json.dumps(extensions_preferences),
+                        'for_version': current_version,
+                        'uuid': uuid,
+                    },
                 )
-            )
 
     async def downgrade(self):
         """Downgrade"""

src/langbot/pkg/pipeline/pipelinemgr.py

@@ -237,6 +237,7 @@ class RuntimePipeline:
                 launcher_type=query.launcher_type.value,
                 launcher_id=query.launcher_id,
                 sender_id=query.sender_id,
+                message_event=query.message_event,
                 message_chain=query.message_chain,
             )
 

src/langbot/pkg/pipeline/preproc/preproc.py

@@ -7,6 +7,7 @@ from langbot_plugin.api.entities.builtin.provider import message as provider_mes
 import langbot_plugin.api.entities.events as events
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
+import langbot_plugin.api.entities.builtin.platform.events as platform_events
 
 
 @stage.stage_class('PreProcessor')
@@ -74,12 +75,26 @@ class PreProcessor(stage.PipelineStage):
                 self.ap.logger.debug(f'Bound MCP servers: {bound_mcp_servers}')
                 self.ap.logger.debug(f'Use funcs: {query.use_funcs}')
 
+        sender_name = ''
+
+        if isinstance(query.message_event, platform_events.GroupMessage):
+            sender_name = query.message_event.sender.member_name
+        elif isinstance(query.message_event, platform_events.FriendMessage):
+            sender_name = query.message_event.sender.nickname
+
         variables = {
+            'launcher_type': query.session.launcher_type.value,
+            'launcher_id': query.session.launcher_id,
+            'sender_id': query.sender_id,
             'session_id': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
             'conversation_id': conversation.uuid,
             'msg_create_time': (
                 int(query.message_event.time) if query.message_event.time else int(datetime.datetime.now().timestamp())
             ),
+            'group_name': query.message_event.group.name
+            if isinstance(query.message_event, platform_events.GroupMessage)
+            else '',
+            'sender_name': sender_name,
         }
         query.variables.update(variables)
 
@@ -111,6 +126,12 @@ class PreProcessor(stage.PipelineStage):
                 ):
                     if me.base64 is not None:
                         content_list.append(provider_message.ContentElement.from_image_base64(me.base64))
+            elif isinstance(me, platform_message.Voice):
+                # 转成文件链接，让下游 runner 上传到目标模型
+                if me.base64:
+                    content_list.append(provider_message.ContentElement.from_file_base64(me.base64, 'voice.silk'))
+                elif me.url:
+                    content_list.append(provider_message.ContentElement.from_file_url(me.url, 'voice'))
             elif isinstance(me, platform_message.File):
                 # if me.url is not None:
                 content_list.append(provider_message.ContentElement.from_file_url(me.url, me.name))

src/langbot/pkg/pipeline/process/handlers/chat.py

@@ -40,6 +40,7 @@ class ChatMessageHandler(handler.MessageHandler):
             launcher_id=query.launcher_id,
             sender_id=query.sender_id,
             text_message=str(query.message_chain),
+            message_event=query.message_event,
             message_chain=query.message_chain,
             query=query,
         )
@@ -75,7 +76,7 @@ class ChatMessageHandler(handler.MessageHandler):
                         runner = r(self.ap, query.pipeline_config)
                         break
                 else:
-                    raise ValueError(f'未找到请求运行器: {query.pipeline_config["ai"]["runner"]["runner"]}')
+                    raise ValueError(f'Request Runner not found: {query.pipeline_config["ai"]["runner"]["runner"]}')
                 if is_stream:
                     resp_message_id = uuid.uuid4()
 
@@ -90,7 +91,9 @@ class ChatMessageHandler(handler.MessageHandler):
                             await query.adapter.create_message_card(str(resp_message_id), query.message_event)
                             is_create_card = True
                         query.resp_messages.append(result)
-                        self.ap.logger.info(f'对话({query.query_id})流式响应: {self.cut_str(result.readable_str())}')
+                        self.ap.logger.info(
+                            f'Conversation({query.query_id}) Streaming Response: {self.cut_str(result.readable_str())}'
+                        )
 
                         if result.content is not None:
                             text_length += len(result.content)
@@ -101,7 +104,9 @@ class ChatMessageHandler(handler.MessageHandler):
                     async for result in runner.run(query):
                         query.resp_messages.append(result)
 
-                        self.ap.logger.info(f'对话({query.query_id})响应: {self.cut_str(result.readable_str())}')
+                        self.ap.logger.info(
+                            f'Conversation({query.query_id}) Response: {self.cut_str(result.readable_str())}'
+                        )
 
                         if result.content is not None:
                             text_length += len(result.content)
@@ -112,7 +117,7 @@ class ChatMessageHandler(handler.MessageHandler):
 
                 query.session.using_conversation.messages.extend(query.resp_messages)
             except Exception as e:
-                self.ap.logger.error(f'对话({query.query_id})请求失败: {type(e).__name__} {str(e)}')
+                self.ap.logger.error(f'Conversation({query.query_id}) Request Failed: {type(e).__name__} {str(e)}')
                 traceback.print_exc()
 
                 hide_exception_info = query.pipeline_config['output']['misc']['hide-exception']

src/langbot/pkg/platform/sources/aiocqhttp.py

@@ -327,9 +327,6 @@ class AiocqhttpEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title=event.sender['title'] if 'title' in event.sender else '',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=yiri_chain,
                 time=event.time,

src/langbot/pkg/platform/sources/dingtalk.py

@@ -119,9 +119,6 @@ class DingTalkEventConverter(abstract_platform_adapter.AbstractEventConverter):
                     permission=platform_entities.Permission.Member,
                 ),
                 special_title='',
-                join_timestamp=0,
-                last_speak_timestamp=0,
-                mute_time_remaining=0,
             )
             time = event.incoming_message.create_at
             return platform_events.GroupMessage(

src/langbot/pkg/platform/sources/discord.py

@@ -8,6 +8,9 @@ import base64
 import uuid
 import os
 import datetime
+
+# 使用BytesIO创建文件对象，避免路径问题
+import io
 import asyncio
 from enum import Enum
 
@@ -594,7 +597,7 @@ class DiscordMessageConverter(abstract_platform_adapter.AbstractMessageConverter
                 break
 
         text_string = ''
-        image_files = []
+        files = []
 
         for ele in message_chain:
             if isinstance(ele, platform_message.Image):
@@ -668,22 +671,67 @@ class DiscordMessageConverter(abstract_platform_adapter.AbstractMessageConverter
                         continue  # 跳过读取失败的文件
 
                 if image_bytes:
-                    # 使用BytesIO创建文件对象，避免路径问题
-                    import io
-
-                    image_files.append(discord.File(fp=io.BytesIO(image_bytes), filename=filename))
+                    files.append(discord.File(fp=io.BytesIO(image_bytes), filename=filename))
             elif isinstance(ele, platform_message.Plain):
                 text_string += ele.text
+            elif isinstance(ele, platform_message.Voice):
+                file_bytes = None
+                filename = f'{uuid.uuid4()}.mp3'
+                if ele.base64:
+                    if ele.base64.startswith('data:'):
+                        data_header = ele.base64.split(',')[0]
+                        if 'wav' in data_header:
+                            filename = f'{uuid.uuid4()}.wav'
+                        elif 'mp3' in data_header:
+                            filename = f'{uuid.uuid4()}.mp3'
+                        elif 'ogg' in data_header:
+                            filename = f'{uuid.uuid4()}.ogg'
+                        elif 'm4a' in data_header:
+                            filename = f'{uuid.uuid4()}.m4a'
+                        elif 'aac' in data_header:
+                            filename = f'{uuid.uuid4()}.aac'
+                        elif 'flac' in data_header:
+                            filename = f'{uuid.uuid4()}.flac'
+                        elif 'alac' in data_header:
+                            filename = f'{uuid.uuid4()}.alac'
+                        elif 'opus' in data_header:
+                            filename = f'{uuid.uuid4()}.opus'
+                        elif 'webm' in data_header:
+                            filename = f'{uuid.uuid4()}.webm'
+
+                    file_base64 = ele.base64.split(',')[-1]
+                    file_bytes = base64.b64decode(file_base64)
+                elif ele.url:
+                    async with aiohttp.ClientSession() as session:
+                        async with session.get(ele.url) as response:
+                            file_bytes = await response.read()
+                if file_bytes:
+                    files.append(discord.File(fp=io.BytesIO(file_bytes), filename=filename))
+            elif isinstance(ele, platform_message.File):
+                file_bytes = None
+                filename = f'{uuid.uuid4()}.{ele.name.split(".")[-1]}'
+                if ele.base64:
+                    if ele.base64.startswith('data:'):
+                        file_base64 = ele.base64.split(',')[1]
+                        file_bytes = base64.b64decode(file_base64)
+                    else:
+                        file_bytes = base64.b64decode(ele.base64)
+                elif ele.url:
+                    async with aiohttp.ClientSession() as session:
+                        async with session.get(ele.url) as response:
+                            file_bytes = await response.read()
+                if file_bytes:
+                    files.append(discord.File(fp=io.BytesIO(file_bytes), filename=filename))
             elif isinstance(ele, platform_message.Forward):
                 for node in ele.node_list:
                     (
                         node_text,
-                        node_images,
+                        node_files,
                     ) = await DiscordMessageConverter.yiri2target(node.message_chain)
                     text_string += node_text
-                    image_files.extend(node_images)
+                    files.extend(node_files)
 
-        return text_string, image_files
+        return text_string, files
 
     @staticmethod
     async def target2yiri(message: discord.Message) -> platform_message.MessageChain:
@@ -769,9 +817,6 @@ class DiscordEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=message_chain,
                 time=event.created_at.timestamp(),
@@ -993,7 +1038,7 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             await self.voice_manager.cleanup_inactive_connections()
 
     async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
-        msg_to_send, image_files = await self.message_converter.yiri2target(message)
+        msg_to_send, files = await self.message_converter.yiri2target(message)
 
         try:
             # 获取频道对象
@@ -1006,8 +1051,8 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                 'content': msg_to_send,
             }
 
-            if len(image_files) > 0:
-                args['files'] = image_files
+            if len(files) > 0:
+                args['files'] = files
 
             await channel.send(**args)
 
@@ -1021,15 +1066,16 @@ class DiscordAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         message: platform_message.MessageChain,
         quote_origin: bool = False,
     ):
-        msg_to_send, image_files = await self.message_converter.yiri2target(message)
+        msg_to_send, files = await self.message_converter.yiri2target(message)
+
         assert isinstance(message_source.source_platform_object, discord.Message)
 
         args = {
             'content': msg_to_send,
         }
 
-        if len(image_files) > 0:
-            args['files'] = image_files
+        if len(files) > 0:
+            args['files'] = files
 
         if quote_origin:
             args['reference'] = message_source.source_platform_object

src/langbot/pkg/platform/sources/kook.py

@@ -137,7 +137,11 @@ class KookMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
             # For file messages, content is typically the file URL
             attachments = extra.get('attachments', {})
             file_name = attachments.get('name', 'file')
-            components.append(platform_message.Plain(text=f'[File: {file_name}]'))
+            components.append(platform_message.File(url=content, name=file_name))
+        elif msg_type == 8:  # Audio message
+            # For audio messages, content is typically the audio URL
+            attachments = extra.get('attachments', {})
+            components.append(platform_message.Voice(url=content))
         elif msg_type == 9:  # KMarkdown message
             # Note: content is already stripped of mention patterns above
             if content:
@@ -219,9 +223,6 @@ class KookEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=message_chain,
                 time=event_time,
@@ -320,9 +321,6 @@ class KookAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                     data = await response.json()
                     if data.get('code') == 0:
                         user_info = data['data']
-                        await self.logger.info(
-                            f'Retrieved bot user info: {user_info.get("username")} (ID: {user_info.get("id")})'
-                        )
                         return user_info
                     else:
                         raise Exception(f'Failed to get bot user info: {data.get("message")}')
@@ -346,11 +344,10 @@ class KookAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
 
         # Ignore messages from bot itself to prevent infinite loops
         if self.bot_account_id and str(author_id) == self.bot_account_id:
-            await self.logger.debug(f'Ignoring message from bot itself (author_id: {author_id})')
             return
 
-        # Only process text messages (type 1, 2, 4, 9, 10) in GROUP or PERSON channels
-        if event_type in [1, 2, 4, 9, 10] and channel_type in ['GROUP', 'PERSON']:
+        # Only process text messages (type 1, 2, 4, 8, 9, 10) in GROUP or PERSON channels
+        if event_type in [1, 2, 4, 8, 9, 10] and channel_type in ['GROUP', 'PERSON']:
             try:
                 # Convert to LangBot event
                 lb_event = await self.event_converter.target2yiri(data, self.bot_account_id)
@@ -380,7 +377,6 @@ class KookAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                             'sn': self.current_sn,
                         }
                         await self.ws.send(json.dumps(ping_msg))
-                        await self.logger.debug(f'Sent PING with sn={self.current_sn}')
                     except Exception:
                         # Connection closed or send failed, exit loop
                         break
@@ -401,10 +397,9 @@ class KookAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                     self.gateway_url = await self._get_gateway_url()
 
                 # Connect to WebSocket
-                await self.logger.info(f'Connecting to KOOK WebSocket: {self.gateway_url}')
                 async with websockets.connect(self.gateway_url) as ws:
+                    await self.logger.info(f'Connected to KOOK WebSocket: {self.gateway_url}')
                     self.ws = ws
-                    await self.logger.info('KOOK WebSocket connected')
 
                     # Start heartbeat
                     self.heartbeat_task = asyncio.create_task(self._heartbeat_loop())
@@ -455,10 +450,11 @@ class KookAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                             elif signal == 3:  # PONG
                                 await self._handle_pong(msg_data.get('d', {}))
                             elif signal == 5:  # RECONNECT
-                                await self.logger.info('Received RECONNECT signal')
+                                # await self.logger.info('Received RECONNECT signal')
                                 break  # Break to reconnect
                             elif signal == 6:  # RESUME ACK
-                                await self.logger.info('Resume successful')
+                                # await self.logger.info('Resume successful')
+                                pass
                         except json.JSONDecodeError:
                             await self.logger.error(f'Failed to parse message: {message}')
                         except Exception as e:
@@ -571,6 +567,8 @@ class KookAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         if quote_origin and msg_id:
             payload['quote'] = msg_id
 
+        payload['reply_msg_id'] = msg_id
+
         headers = {
             'Authorization': f'Bot {self.config["token"]}',
             'Content-Type': 'application/json',

src/langbot/pkg/platform/sources/lark.py

@@ -55,9 +55,7 @@ class AESCipher(object):
 
 class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
     @staticmethod
-    async def upload_image_to_lark(
-        msg: platform_message.Image, api_client: lark_oapi.Client
-    ) -> typing.Optional[str]:
+    async def upload_image_to_lark(msg: platform_message.Image, api_client: lark_oapi.Client) -> typing.Optional[str]:
         """Upload an image to Lark and return the image_key, or None if upload fails."""
         image_bytes = None
 
@@ -95,7 +93,9 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
                 return None
 
         if image_bytes is None:
-            print(f'No image data available for Image message (url={msg.url}, base64={bool(msg.base64)}, path={msg.path})')
+            print(
+                f'No image data available for Image message (url={msg.url}, base64={bool(msg.base64)}, path={msg.path})'
+            )
             return None
 
         try:
@@ -113,10 +113,7 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
                 request = (
                     CreateImageRequest.builder()
                     .request_body(
-                        CreateImageRequestBody.builder()
-                        .image_type('message')
-                        .image(open(temp_file_path, 'rb'))
-                        .build()
+                        CreateImageRequestBody.builder().image_type('message').image(open(temp_file_path, 'rb')).build()
                     )
                     .build()
                 )
@@ -143,7 +140,7 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
         message_chain: platform_message.MessageChain, api_client: lark_oapi.Client
     ) -> typing.Tuple[list, list]:
         """Convert message chain to Lark format.
-        
+
         Returns:
             Tuple of (text_elements, image_keys):
             - text_elements: List of paragraphs for post message format
@@ -159,24 +156,24 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
         async def process_text_with_images(text: str) -> typing.Tuple[str, list]:
             """Extract Markdown images from text and return cleaned text + image URLs."""
             extracted_urls = []
-            
+
             # Find all Markdown images
             matches = list(markdown_image_pattern.finditer(text))
             if not matches:
                 return text, []
-            
+
             # Extract URLs and remove image syntax from text
             cleaned_text = text
             for match in reversed(matches):  # Reverse to maintain correct positions
                 url = match.group(2)
                 extracted_urls.insert(0, url)  # Insert at beginning since we're going in reverse
                 # Replace image syntax with empty string or a placeholder
-                cleaned_text = cleaned_text[:match.start()] + cleaned_text[match.end():]
-            
+                cleaned_text = cleaned_text[: match.start()] + cleaned_text[match.end() :]
+
             # Clean up multiple consecutive newlines that might result from removing images
             cleaned_text = re.sub(r'\n{3,}', '\n\n', cleaned_text)
             cleaned_text = cleaned_text.strip()
-            
+
             return cleaned_text, extracted_urls
 
         for msg in message_chain:
@@ -189,14 +186,14 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
                         text = msg.text.encode('latin1').decode('utf-8')
                     except UnicodeError:
                         text = msg.text.encode('utf-8', errors='replace').decode('utf-8')
-                
+
                 # Check for and extract Markdown images from text
                 cleaned_text, extracted_urls = await process_text_with_images(text)
-                
+
                 # Add cleaned text if not empty
                 if cleaned_text:
                     pending_paragraph.append({'tag': 'md', 'text': cleaned_text})
-                
+
                 # Process extracted image URLs
                 for url in extracted_urls:
                     # Create a temporary Image message to upload
@@ -204,7 +201,7 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
                     image_key = await LarkMessageConverter.upload_image_to_lark(temp_image, api_client)
                     if image_key:
                         image_keys.append(image_key)
-                        
+
             elif isinstance(msg, platform_message.At):
                 pending_paragraph.append({'tag': 'at', 'user_id': msg.target, 'style': []})
             elif isinstance(msg, platform_message.AtAll):
@@ -300,6 +297,10 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
             message_content['content'] = new_list
         elif message.message_type == 'image':
             message_content['content'] = [{'tag': 'img', 'image_key': message_content['image_key'], 'style': []}]
+        elif message.message_type == 'file':
+            message_content['content'] = [
+                {'tag': 'file', 'file_key': message_content['file_key'], 'file_name': message_content['file_name']}
+            ]
 
         for ele in message_content['content']:
             if ele['tag'] == 'text':
@@ -330,6 +331,33 @@ class LarkMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
                 image_format = response.raw.headers['content-type']
 
                 lb_msg_list.append(platform_message.Image(base64=f'data:{image_format};base64,{image_base64}'))
+            elif ele['tag'] == 'file':
+                file_key = ele['file_key']
+                file_name = ele['file_name']
+
+                request: GetMessageResourceRequest = (
+                    GetMessageResourceRequest.builder()
+                    .message_id(message.message_id)
+                    .file_key(file_key)
+                    .type('file')
+                    .build()
+                )
+
+                response: GetMessageResourceResponse = await api_client.im.v1.message_resource.aget(request)
+
+                if not response.success():
+                    raise Exception(
+                        f'client.im.v1.message_resource.get failed, code: {response.code}, msg: {response.msg}, log_id: {response.get_log_id()}, resp: \n{json.dumps(json.loads(response.raw.content), indent=4, ensure_ascii=False)}'
+                    )
+
+                file_bytes = response.file.read()
+                file_base64 = base64.b64encode(file_bytes).decode()
+
+                file_format = response.raw.headers['content-type']
+
+                lb_msg_list.append(
+                    platform_message.File(base64=f'data:{file_format};base64,{file_base64}', name=file_name)
+                )
 
         return platform_message.MessageChain(lb_msg_list)
 
@@ -369,9 +397,6 @@ class LarkEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=message_chain,
                 time=event.event.message.create_time,

src/langbot/pkg/platform/sources/legacy/gewechat.py

@@ -437,9 +437,6 @@ class GewechatEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=message_chain,
                 time=event['Data']['CreateTime'],

src/langbot/pkg/platform/sources/legacy/nakuru.py

@@ -153,9 +153,6 @@ class NakuruProjectEventConverter(abstract_platform_adapter.AbstractEventConvert
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title=event.sender.title,
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=yiri_chain,
                 time=event.time,

src/langbot/pkg/platform/sources/legacy/qqbotpy.py

@@ -279,11 +279,6 @@ class OfficialEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=int(
-                        datetime.datetime.strptime(event.member.joined_at, '%Y-%m-%dT%H:%M:%S%z').timestamp()
-                    ),
-                    last_speak_timestamp=datetime.datetime.now().timestamp(),
-                    mute_time_remaining=0,
                 ),
                 message_chain=OfficialMessageConverter.extract_message_chain_from_obj(event, event.id),
                 time=int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp()),
@@ -312,9 +307,6 @@ class OfficialEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=int(0),
-                    last_speak_timestamp=datetime.datetime.now().timestamp(),
-                    mute_time_remaining=0,
                 ),
                 message_chain=OfficialMessageConverter.extract_message_chain_from_obj(event, event.id),
                 time=int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp()),

src/langbot/pkg/platform/sources/line.py

@@ -108,9 +108,6 @@ class LINEEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=message_chain,
                 time=event.timestamp,
@@ -262,19 +259,6 @@ class LINEAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         # 保持运行但不启动独立端口
 
         # 打印 webhook 回调地址
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f'http://127.0.0.1:{api_port}/bots/{self.bot_uuid}'
-                webhook_url_public = f'http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}'
-
-                await self.logger.info('LINE Webhook 回调地址:')
-                await self.logger.info(f'  本地地址: {webhook_url}')
-                await self.logger.info(f'  公网地址: {webhook_url_public}')
-                await self.logger.info('请在 LINE 后台配置此回调地址')
-            except Exception as e:
-                await self.logger.warning(f'无法生成 webhook URL: {e}')
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)

src/langbot/pkg/platform/sources/officialaccount.py

@@ -155,20 +155,6 @@ class OfficialAccountAdapter(abstract_platform_adapter.AbstractMessagePlatformAd
         # 统一 webhook 模式下，不启动独立的 Quart 应用
         # 保持运行但不启动独立端口
 
-        # 打印 webhook 回调地址
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f'http://127.0.0.1:{api_port}/bots/{self.bot_uuid}'
-                webhook_url_public = f'http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}'
-
-                await self.logger.info('微信公众号 Webhook 回调地址:')
-                await self.logger.info(f'  本地地址: {webhook_url}')
-                await self.logger.info(f'  公网地址: {webhook_url_public}')
-                await self.logger.info('请在微信公众号后台配置此回调地址')
-            except Exception as e:
-                await self.logger.warning(f'无法生成 webhook URL: {e}')
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)

src/langbot/pkg/platform/sources/qqofficial.py

@@ -94,9 +94,6 @@ class QQOfficialEventConverter(abstract_platform_adapter.AbstractEventConverter)
                     permission=platform_entities.Permission.Member,
                 ),
                 special_title='',
-                join_timestamp=0,
-                last_speak_timestamp=0,
-                mute_time_remaining=0,
             )
             time = int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp())
             return platform_events.GroupMessage(
@@ -117,9 +114,6 @@ class QQOfficialEventConverter(abstract_platform_adapter.AbstractEventConverter)
                     permission=platform_entities.Permission.Member,
                 ),
                 special_title='',
-                join_timestamp=0,
-                last_speak_timestamp=0,
-                mute_time_remaining=0,
             )
             time = int(datetime.datetime.strptime(event.timestamp, '%Y-%m-%dT%H:%M:%S%z').timestamp())
             return platform_events.GroupMessage(
@@ -247,20 +241,6 @@ class QQOfficialAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter
         # 统一 webhook 模式下，不启动独立的 Quart 应用
         # 保持运行但不启动独立端口
 
-        # 打印 webhook 回调地址
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f'http://127.0.0.1:{api_port}/bots/{self.bot_uuid}'
-                webhook_url_public = f'http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}'
-
-                await self.logger.info('QQ 官方机器人 Webhook 回调地址:')
-                await self.logger.info(f'  本地地址: {webhook_url}')
-                await self.logger.info(f'  公网地址: {webhook_url_public}')
-                await self.logger.info('请在 QQ 官方机器人后台配置此回调地址')
-            except Exception as e:
-                await self.logger.warning(f'无法生成 webhook URL: {e}')
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)

src/langbot/pkg/platform/sources/slack.py

@@ -76,9 +76,6 @@ class SlackEventConverter(abstract_platform_adapter.AbstractEventConverter):
                     id=event.channel_id, name='MEMBER', permission=platform_entities.Permission.Member
                 ),
                 special_title='',
-                join_timestamp=0,
-                last_speak_timestamp=0,
-                mute_time_remaining=0,
             )
             time = int(datetime.datetime.utcnow().timestamp())
             return platform_events.GroupMessage(
@@ -112,10 +109,7 @@ class SlackAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             raise command_errors.ParamNotEnoughError('Slack机器人缺少相关配置项，请查看文档或联系管理员')
 
         bot = SlackClient(
-            bot_token=config['bot_token'],
-            signing_secret=config['signing_secret'],
-            logger=logger,
-            unified_mode=True
+            bot_token=config['bot_token'], signing_secret=config['signing_secret'], logger=logger, unified_mode=True
         )
 
         super().__init__(
@@ -194,24 +188,10 @@ class SlackAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     async def run_async(self):
         # 统一 webhook 模式下，不启动独立的 Quart 应用
         # 保持运行但不启动独立端口
-
-        # 打印 webhook 回调地址
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f"http://127.0.0.1:{api_port}/bots/{self.bot_uuid}"
-                webhook_url_public = f"http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}"
-
-                await self.logger.info(f"Slack 机器人 Webhook 回调地址:")
-                await self.logger.info(f"  本地地址: {webhook_url}")
-                await self.logger.info(f"  公网地址: {webhook_url_public}")
-                await self.logger.info(f"请在 Slack 后台配置此回调地址")
-            except Exception as e:
-                await self.logger.warning(f"无法生成 webhook URL: {e}")
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)
+
         await keep_alive()
 
     async def kill(self) -> bool:

src/langbot/pkg/platform/sources/telegram.py

@@ -120,9 +120,6 @@ class TelegramEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=lb_message,
                 time=event.message.date.timestamp(),

src/langbot/pkg/platform/sources/websocket_adapter.py

@@ -97,8 +97,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
         # 推送到所有相关连接
         await self.outbound_message_queue.put(message_data)
 
-        await self.logger.info(f'Send message to {target_id}: {message}')
-
         return message_data
 
     async def reply_message(
@@ -242,7 +240,6 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
 
     async def run_async(self):
         """运行适配器"""
-        await self.logger.info('WebSocket适配器已启动')
 
         try:
             while True:
@@ -258,12 +255,11 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
 
                 await asyncio.sleep(0.1)
         except asyncio.CancelledError:
-            await self.logger.info('WebSocket适配器已停止')
             raise
 
     async def kill(self):
         """停止适配器"""
-        await self.logger.info('WebSocket适配器正在停止')
+        pass
 
     async def _process_image_components(self, message_chain_obj: list):
         """

src/langbot/pkg/platform/sources/wechatpad.py

@@ -501,9 +501,6 @@ class WeChatPadEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 ),
                 message_chain=message_chain,
                 time=event['create_time'],

src/langbot/pkg/platform/sources/wecom.py

@@ -35,6 +35,20 @@ class WecomMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
                         'media_id': await bot.get_media_id(msg),
                     }
                 )
+            elif type(msg) is platform_message.Voice:
+                content_list.append(
+                    {
+                        'type': 'voice',
+                        'media_id': await bot.get_media_id(msg),
+                    }
+                )
+            elif type(msg) is platform_message.File:
+                content_list.append(
+                    {
+                        'type': 'file',
+                        'media_id': await bot.get_media_id(msg),
+                    }
+                )
             elif type(msg) is platform_message.Forward:
                 for node in msg.node_list:
                     content_list.extend((await WecomMessageConverter.yiri2target(node.message_chain, bot)))
@@ -185,6 +199,10 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                 await self.bot.send_private_msg(fixed_user_id, Wecom_event.agent_id, content['content'])
             elif content['type'] == 'image':
                 await self.bot.send_image(fixed_user_id, Wecom_event.agent_id, content['media_id'])
+            elif content['type'] == 'voice':
+                await self.bot.send_voice(fixed_user_id, Wecom_event.agent_id, content['media_id'])
+            elif content['type'] == 'file':
+                await self.bot.send_file(fixed_user_id, Wecom_event.agent_id, content['media_id'])
 
     async def send_message(self, target_type: str, target_id: str, message: platform_message.MessageChain):
         content_list = await WecomMessageConverter.yiri2target(message, self.bot)
@@ -197,6 +215,10 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
                     await self.bot.send_private_msg(user_id, agent_id, content['content'])
                 if content['type'] == 'image':
                     await self.bot.send_image(user_id, agent_id, content['media'])
+                if content['type'] == 'voice':
+                    await self.bot.send_voice(user_id, agent_id, content['media'])
+                if content['type'] == 'file':
+                    await self.bot.send_file(user_id, agent_id, content['media'])
 
     def register_listener(
         self,
@@ -232,19 +254,6 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         return await self.bot.handle_unified_webhook(request)
 
     async def run_async(self):
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f'http://127.0.0.1:{api_port}/bots/{self.bot_uuid}'
-                webhook_url_public = f'http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}'
-
-                await self.logger.info('企业微信 Webhook 回调地址:')
-                await self.logger.info(f'  本地地址: {webhook_url}')
-                await self.logger.info(f'  公网地址: {webhook_url_public}')
-                await self.logger.info('请在企业微信后台配置此回调地址')
-            except Exception as e:
-                await self.logger.warning(f'无法生成 webhook URL: {e}')
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)

src/langbot/pkg/platform/sources/wecom.yaml

@@ -21,21 +21,21 @@ spec:
     - name: secret
       label:
         en_US: Secret
-        zh_Hans: 密钥
+        zh_Hans: 密钥 (Secret)
       type: string
       required: true
       default: ""
     - name: token
       label:
         en_US: Token
-        zh_Hans: 令牌
+        zh_Hans: 令牌 (Token)
       type: string
       required: true
       default: ""
     - name: EncodingAESKey
       label:
         en_US: EncodingAESKey
-        zh_Hans: 消息加解密密钥
+        zh_Hans: 消息加解密密钥 (EncodingAESKey)
       type: string
       required: true
       default: ""

src/langbot/pkg/platform/sources/wecombot.py

@@ -28,9 +28,105 @@ class WecomBotMessageConverter(abstract_platform_adapter.AbstractMessageConverte
         if event.type == 'group':
             yiri_msg_list.append(platform_message.At(target=event.ai_bot_id))
         yiri_msg_list.append(platform_message.Source(id=event.message_id, time=datetime.datetime.now()))
-        yiri_msg_list.append(platform_message.Plain(text=event.content))
-        if event.picurl != '':
-            yiri_msg_list.append(platform_message.Image(base64=event.picurl))
+
+        if event.content:
+            yiri_msg_list.append(platform_message.Plain(text=event.content))
+
+        images = []
+        if event.images:
+            images.extend([img for img in event.images if img])
+        if not images and event.picurl:
+            images.append(event.picurl)
+        for image_base64 in images:
+            if image_base64:
+                yiri_msg_list.append(platform_message.Image(base64=image_base64))
+
+        file_info = event.file or {}
+        if file_info:
+            file_url = (
+                file_info.get('download_url')
+                or file_info.get('url')
+                or file_info.get('fileurl')
+                or file_info.get('path')
+            )
+            file_base64 = file_info.get('base64')
+            file_name = file_info.get('filename') or file_info.get('name')
+            file_size = file_info.get('filesize') or file_info.get('size')
+            file_data = file_url or file_base64
+            if file_data or file_name:
+                file_kwargs = {}
+                if file_data:
+                    file_kwargs['url'] = file_data
+                if file_name:
+                    file_kwargs['name'] = file_name
+                if file_size is not None:
+                    file_kwargs['size'] = file_size
+                try:
+                    yiri_msg_list.append(platform_message.File(**file_kwargs))
+                except Exception:
+                    # 兜底
+                    yiri_msg_list.append(platform_message.Unknown(text='[file message unsupported]'))
+
+        voice_info = event.voice or {}
+        if voice_info:
+            voice_payload = voice_info.get('base64') or voice_info.get('url')
+            if voice_payload:
+                if voice_info.get('base64') and not voice_payload.startswith('data:'):
+                    voice_payload = f'data:audio/mpeg;base64,{voice_info.get("base64")}'
+                try:
+                    yiri_msg_list.append(platform_message.Voice(base64=voice_payload))
+                except Exception:
+                    try:
+                        voice_kwargs = {'url': voice_payload}
+                        voice_name = voice_info.get('filename') or voice_info.get('name')
+                        voice_size = voice_info.get('filesize') or voice_info.get('size')
+                        if voice_name:
+                            voice_kwargs['name'] = voice_name
+                        if voice_size is not None:
+                            voice_kwargs['size'] = voice_size
+                        yiri_msg_list.append(platform_message.File(**voice_kwargs))
+                    except Exception:
+                        yiri_msg_list.append(platform_message.Unknown(text='[voice message unsupported]'))
+
+        video_info = event.video or {}
+        if video_info:
+            video_payload = (
+                video_info.get('base64')
+                or video_info.get('url')
+                or video_info.get('download_url')
+                or video_info.get('fileurl')
+            )
+            if video_payload:
+                video_kwargs = {'url': video_payload}
+                video_name = video_info.get('filename') or video_info.get('name')
+                video_size = video_info.get('filesize') or video_info.get('size')
+                if video_name:
+                    video_kwargs['name'] = video_name
+                if video_size is not None:
+                    video_kwargs['size'] = video_size
+                try:
+                    # 没有专门的视频类型，沿用 File 传递给上层
+                    yiri_msg_list.append(platform_message.File(**video_kwargs))
+                except Exception:
+                    yiri_msg_list.append(platform_message.Unknown(text='[video message unsupported]'))
+
+        if event.msgtype == 'link' and event.link:
+            link = event.link
+            summary = '\n'.join(
+                filter(
+                    None,
+                    [link.get('title', ''), link.get('description') or link.get('digest', ''), link.get('url', '')],
+                )
+            )
+            if summary:
+                yiri_msg_list.append(platform_message.Plain(text=summary))
+
+        has_content_element = any(
+            not isinstance(element, (platform_message.Source, platform_message.At)) for element in yiri_msg_list
+        )
+        if not has_content_element:
+            fallback_type = event.msgtype or 'unknown'
+            yiri_msg_list.append(platform_message.Unknown(text=f'[unsupported wecom msgtype: {fallback_type}]'))
         chain = platform_message.MessageChain(yiri_msg_list)
 
         return chain
@@ -67,9 +163,6 @@ class WecomBotEventConverter(abstract_platform_adapter.AbstractEventConverter):
                         permission=platform_entities.Permission.Member,
                     ),
                     special_title='',
-                    join_timestamp=0,
-                    last_speak_timestamp=0,
-                    mute_time_remaining=0,
                 )
                 time = datetime.datetime.now().timestamp()
                 return platform_events.GroupMessage(
@@ -211,20 +304,6 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         # 统一 webhook 模式下，不启动独立的 Quart 应用
         # 保持运行但不启动独立端口
 
-        # 打印 webhook 回调地址
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f'http://127.0.0.1:{api_port}/bots/{self.bot_uuid}'
-                webhook_url_public = f'http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}'
-
-                await self.logger.info('企业微信机器人 Webhook 回调地址:')
-                await self.logger.info(f'  本地地址: {webhook_url}')
-                await self.logger.info(f'  公网地址: {webhook_url_public}')
-                await self.logger.info('请在企业微信后台配置此回调地址')
-            except Exception as e:
-                await self.logger.warning(f'无法生成 webhook URL: {e}')
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)

src/langbot/pkg/platform/sources/wecombot.yaml

@@ -21,14 +21,14 @@ spec:
     - name: Token
       label:
         en_US: Token
-        zh_Hans: 令牌
+        zh_Hans: 令牌 (Token)
       type: string
       required: true
       default: ""
     - name: EncodingAESKey
       label:
         en_US: EncodingAESKey
-        zh_Hans: 消息加解密密钥
+        zh_Hans: 消息加解密密钥 (EncodingAESKey)
       type: string
       required: true
       default: ""

src/langbot/pkg/platform/sources/wecomcs.py

@@ -213,23 +213,10 @@ class WecomCSAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         # 统一 webhook 模式下，不启动独立的 Quart 应用
         # 保持运行但不启动独立端口
 
-        # 打印 webhook 回调地址
-        if self.bot_uuid and hasattr(self.logger, 'ap'):
-            try:
-                api_port = self.logger.ap.instance_config.data['api']['port']
-                webhook_url = f"http://127.0.0.1:{api_port}/bots/{self.bot_uuid}"
-                webhook_url_public = f"http://<Your-Public-IP>:{api_port}/bots/{self.bot_uuid}"
-
-                await self.logger.info(f"企业微信客服 Webhook 回调地址:")
-                await self.logger.info(f"  本地地址: {webhook_url}")
-                await self.logger.info(f"  公网地址: {webhook_url_public}")
-                await self.logger.info(f"请在企业微信后台配置此回调地址")
-            except Exception as e:
-                await self.logger.warning(f"无法生成 webhook URL: {e}")
-
         async def keep_alive():
             while True:
                 await asyncio.sleep(1)
+
         await keep_alive()
 
     async def kill(self) -> bool:

src/langbot/pkg/plugin/handler.py

@@ -139,6 +139,8 @@ class RuntimeConnectionHandler(handler.Handler):
 
             message_chain_obj = platform_message.MessageChain.model_validate(message_chain)
 
+            self.ap.logger.debug(f'Reply message: {message_chain_obj.model_dump(serialize_as_any=False)}')
+
             await query.adapter.reply_message(
                 query.message_event,
                 message_chain_obj,
@@ -563,7 +565,7 @@ class RuntimeConnectionHandler(handler.Handler):
                 'event_context': event_context,
                 'include_plugins': include_plugins,
             },
-            timeout=60,
+            timeout=180,
         )
 
         return result

src/langbot/pkg/provider/runners/difysvapi.py

@@ -4,6 +4,7 @@ import typing
 import json
 import uuid
 import base64
+import mimetypes
 
 
 from langbot.pkg.provider import runner
@@ -12,6 +13,7 @@ import langbot_plugin.api.entities.builtin.provider.message as provider_message
 from langbot.pkg.utils import image
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 from langbot.libs.dify_service_api.v1 import client, errors
+import httpx
 
 
 @runner.runner_class('dify-service-api')
@@ -70,14 +72,43 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                 content = f'<think>\n{thinking_content}\n</think>\n{content}'.strip()
             return content, thinking_content
 
-    async def _preprocess_user_message(self, query: pipeline_query.Query) -> tuple[str, list[str]]:
-        """预处理用户消息，提取纯文本，并将图片上传到 Dify 服务
+    async def _preprocess_user_message(self, query: pipeline_query.Query) -> tuple[str, list[dict]]:
+        """预处理用户消息，提取纯文本，并将图片/文件上传到 Dify 服务
 
         Returns:
-            tuple[str, list[str]]: 纯文本和图片的 Dify 服务图片 ID
+            tuple[str, list[dict]]: 纯文本和上传后的文件描述（包含 type 与 id）
         """
         plain_text = ''
-        file_ids = []
+        upload_files: list[dict] = []
+        user_tag = f'{query.session.launcher_type.value}_{query.session.launcher_id}'
+
+        async def upload_file_bytes(file_name: str, file_bytes: bytes, content_type: str) -> str:
+            file_name = file_name or 'file'
+            content_type = content_type or 'application/octet-stream'
+            file = (file_name, file_bytes, content_type)
+            resp = await self.dify_client.upload_file(file, user_tag)
+            return resp['id']
+
+        async def download_file(file_url: str) -> tuple[bytes, str]:
+            """Download file from url (supports data url)."""
+
+            async with httpx.AsyncClient() as client_session:
+                resp = await client_session.get(file_url)
+                resp.raise_for_status()
+                content_type = (
+                    resp.headers.get('content-type') or mimetypes.guess_type(file_url)[0] or 'application/octet-stream'
+                )
+                return resp.content, content_type
+
+        def _detect_file_type(content_type: str) -> str:
+            """Map MIME to dify file type."""
+            if content_type and content_type.startswith('image/'):
+                return 'image'
+            if content_type and content_type.startswith('audio/'):
+                return 'audio'
+            if content_type and content_type.startswith('video/'):
+                return 'video'
+            return 'document'
 
         if isinstance(query.user_message.content, list):
             for ce in query.user_message.content:
@@ -86,30 +117,36 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                 elif ce.type == 'image_base64':
                     image_b64, image_format = await image.extract_b64_and_format(ce.image_base64)
                     file_bytes = base64.b64decode(image_b64)
-                    file = ('img.png', file_bytes, f'image/{image_format}')
-                    file_upload_resp = await self.dify_client.upload_file(
-                        file,
-                        f'{query.session.launcher_type.value}_{query.session.launcher_id}',
-                    )
-                    image_id = file_upload_resp['id']
-                    file_ids.append(image_id)
-                # elif ce.type == "file_url":
-                #     file_bytes = base64.b64decode(ce.file_url)
-                #     file_upload_resp = await self.dify_client.upload_file(
-                #         file_bytes,
-                #         f'{query.session.launcher_type.value}_{query.session.launcher_id}',
-                #     )
-                #     file_id = file_upload_resp['id']
-                #     file_ids.append(file_id)
+                    image_id = await upload_file_bytes(f'img.{image_format}', file_bytes, f'image/{image_format}')
+                    upload_files.append({'type': 'image', 'id': image_id})
+                elif ce.type == 'file_url':
+                    file_url = getattr(ce, 'file_url', None)
+                    file_name = getattr(ce, 'file_name', None) or 'file'
+                    try:
+                        file_bytes, content_type = await download_file(file_url)
+                        file_id = await upload_file_bytes(file_name, file_bytes, content_type)
+                        file_type = _detect_file_type(content_type)
+                        upload_files.append({'type': file_type, 'id': file_id})
+                    except Exception as e:
+                        self.ap.logger.warning(f'dify file upload failed: {e}')
+                elif ce.type == 'file_base64':
+                    file_name = getattr(ce, 'file_name', None) or 'file'
+
+                    header, b64_data = ce.file_base64.split(',', 1)
+                    content_type = 'application/octet-stream'
+                    if ';' in header:
+                        content_type = header.split(';')[0][5:] or content_type
+                    file_bytes = base64.b64decode(b64_data)
+                    file_id = await upload_file_bytes(file_name, file_bytes, content_type)
+                    file_type = _detect_file_type(content_type)
+                    upload_files.append({'type': file_type, 'id': file_id})
+
         elif isinstance(query.user_message.content, str):
             plain_text = query.user_message.content
-        # plain_text = "When the file content is readable, please read the content of this file. When the file is an image, describe the content of this image." if file_ids and not plain_text else plain_text
-        # plain_text = "The user message type cannot be parsed." if not file_ids and not plain_text else plain_text
-        # plain_text = plain_text if plain_text else "When the file content is readable, please read the content of this file. When the file is an image, describe the content of this image."
-        # print(self.pipeline_config['ai'])
+
         plain_text = plain_text if plain_text else self.pipeline_config['ai']['dify-service-api']['base-prompt']
 
-        return plain_text, file_ids
+        return plain_text, upload_files
 
     async def _chat_messages(
         self, query: pipeline_query.Query
@@ -118,14 +155,15 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         cov_id = query.session.using_conversation.uuid or ''
         query.variables['conversation_id'] = cov_id
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        plain_text, upload_files = await self._preprocess_user_message(query)
 
         files = [
             {
-                'type': 'image',
-                'upload_file_id': image_id,
+                'type': f['type'],
+                'transfer_method': 'local_file',
+                'upload_file_id': f['id'],
             }
-            for image_id in image_ids
+            for f in upload_files
         ]
 
         mode = 'basic'  # 标记是基础编排还是工作流编排
@@ -183,15 +221,15 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         cov_id = query.session.using_conversation.uuid or ''
         query.variables['conversation_id'] = cov_id
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        plain_text, upload_files = await self._preprocess_user_message(query)
 
         files = [
             {
-                'type': 'image',
+                'type': f['type'],
                 'transfer_method': 'local_file',
-                'upload_file_id': image_id,
+                'upload_file_id': f['id'],
             }
-            for image_id in image_ids
+            for f in upload_files
         ]
 
         ignored_events = []
@@ -280,15 +318,15 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.variables['conversation_id'] = query.session.using_conversation.uuid
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        plain_text, upload_files = await self._preprocess_user_message(query)
 
         files = [
             {
-                'type': 'image',
+                'type': f['type'],
                 'transfer_method': 'local_file',
-                'upload_file_id': image_id,
+                'upload_file_id': f['id'],
             }
-            for image_id in image_ids
+            for f in upload_files
         ]
 
         ignored_events = ['text_chunk', 'workflow_started']
@@ -352,15 +390,15 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         cov_id = query.session.using_conversation.uuid or ''
         query.variables['conversation_id'] = cov_id
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        plain_text, upload_files = await self._preprocess_user_message(query)
 
         files = [
             {
-                'type': 'image',
+                'type': f['type'],
                 'transfer_method': 'local_file',
-                'upload_file_id': image_id,
+                'upload_file_id': f['id'],
             }
-            for image_id in image_ids
+            for f in upload_files
         ]
 
         basic_mode_pending_chunk = ''
@@ -436,15 +474,15 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         cov_id = query.session.using_conversation.uuid or ''
         query.variables['conversation_id'] = cov_id
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        plain_text, upload_files = await self._preprocess_user_message(query)
 
         files = [
             {
-                'type': 'image',
+                'type': f['type'],
                 'transfer_method': 'local_file',
-                'upload_file_id': image_id,
+                'upload_file_id': f['id'],
             }
-            for image_id in image_ids
+            for f in upload_files
         ]
 
         ignored_events = []
@@ -558,15 +596,15 @@ class DifyServiceAPIRunner(runner.RequestRunner):
 
         query.variables['conversation_id'] = query.session.using_conversation.uuid
 
-        plain_text, image_ids = await self._preprocess_user_message(query)
+        plain_text, upload_files = await self._preprocess_user_message(query)
 
         files = [
             {
-                'type': 'image',
+                'type': f['type'],
                 'transfer_method': 'local_file',
-                'upload_file_id': image_id,
+                'upload_file_id': f['id'],
             }
-            for image_id in image_ids
+            for f in upload_files
         ]
 
         ignored_events = ['workflow_started']

src/langbot/pkg/provider/runners/langflowapi.py

@@ -94,7 +94,6 @@ class LangflowAPIRunner(runner.RequestRunner):
             if is_stream:
                 # 流式请求
                 async with client.stream('POST', url, json=payload, headers=headers, timeout=120.0) as response:
-                    print(response)
                     response.raise_for_status()
 
                     accumulated_content = ''

src/langbot/pkg/provider/session/sessionmgr.py

@@ -33,6 +33,7 @@ class SessionManager:
         session = provider_session.Session(
             launcher_type=query.launcher_type,
             launcher_id=query.launcher_id,
+            sender_id=query.sender_id,
         )
         session._semaphore = asyncio.Semaphore(session_concurrency)
         self.session_list.append(session)

src/langbot/pkg/vector/mgr.py

@@ -4,6 +4,8 @@ from ..core import app
 from .vdb import VectorDatabase
 from .vdbs.chroma import ChromaVectorDatabase
 from .vdbs.qdrant import QdrantVectorDatabase
+from .vdbs.milvus import MilvusVectorDatabase
+from .vdbs.pgvector_db import PgVectorDatabase
 
 
 class VectorDBManager:
@@ -16,12 +18,47 @@ class VectorDBManager:
     async def initialize(self):
         kb_config = self.ap.instance_config.data.get('vdb')
         if kb_config:
-            if kb_config.get('use') == 'chroma':
+            vdb_type = kb_config.get('use')
+
+            if vdb_type == 'chroma':
                 self.vector_db = ChromaVectorDatabase(self.ap)
                 self.ap.logger.info('Initialized Chroma vector database backend.')
-            elif kb_config.get('use') == 'qdrant':
+
+            elif vdb_type == 'qdrant':
                 self.vector_db = QdrantVectorDatabase(self.ap)
                 self.ap.logger.info('Initialized Qdrant vector database backend.')
+
+            elif vdb_type == 'milvus':
+                # Get Milvus configuration
+                milvus_config = kb_config.get('milvus', {})
+                uri = milvus_config.get('uri', './data/milvus.db')
+                token = milvus_config.get('token')
+                self.vector_db = MilvusVectorDatabase(self.ap, uri=uri, token=token)
+                self.ap.logger.info('Initialized Milvus vector database backend.')
+
+            elif vdb_type == 'pgvector':
+                # Get pgvector configuration
+                pgvector_config = kb_config.get('pgvector', {})
+                connection_string = pgvector_config.get('connection_string')
+                if connection_string:
+                    self.vector_db = PgVectorDatabase(self.ap, connection_string=connection_string)
+                else:
+                    # Use individual parameters
+                    host = pgvector_config.get('host', 'localhost')
+                    port = pgvector_config.get('port', 5432)
+                    database = pgvector_config.get('database', 'langbot')
+                    user = pgvector_config.get('user', 'postgres')
+                    password = pgvector_config.get('password', 'postgres')
+                    self.vector_db = PgVectorDatabase(
+                        self.ap,
+                        host=host,
+                        port=port,
+                        database=database,
+                        user=user,
+                        password=password
+                    )
+                self.ap.logger.info('Initialized pgvector database backend.')
+
             else:
                 self.vector_db = ChromaVectorDatabase(self.ap)
                 self.ap.logger.warning('No valid vector database backend configured, defaulting to Chroma.')

src/langbot/pkg/vector/vdbs/milvus.py

@@ -0,0 +1,249 @@
+from __future__ import annotations
+import asyncio
+from typing import Any, Dict
+from pymilvus import MilvusClient, DataType
+from langbot.pkg.vector.vdb import VectorDatabase
+from langbot.pkg.core import app
+
+
+class MilvusVectorDatabase(VectorDatabase):
+    """Milvus vector database implementation"""
+
+    def __init__(self, ap: app.Application, uri: str = "milvus.db", token: str = None):
+        """Initialize Milvus vector database
+
+        Args:
+            ap: Application instance
+            uri: Milvus connection URI. For local file: "milvus.db"
+                 For remote server: "http://localhost:19530"
+            token: Optional authentication token for remote connections
+        """
+        self.ap = ap
+        self.uri = uri
+        self.token = token
+        self.client = None
+        self._collections = {}
+        self._initialize_client()
+
+    def _initialize_client(self):
+        """Initialize Milvus client connection"""
+        try:
+            if self.token:
+                self.client = MilvusClient(uri=self.uri, token=self.token)
+            else:
+                self.client = MilvusClient(uri=self.uri)
+            self.ap.logger.info(f"Connected to Milvus at {self.uri}")
+        except Exception as e:
+            self.ap.logger.error(f"Failed to connect to Milvus: {e}")
+            raise
+
+    async def get_or_create_collection(self, collection: str):
+        """Get or create a Milvus collection
+
+        Args:
+            collection: Collection name (corresponds to knowledge base UUID)
+        """
+        if collection in self._collections:
+            return self._collections[collection]
+
+        # Check if collection exists
+        has_collection = await asyncio.to_thread(
+            self.client.has_collection, collection_name=collection
+        )
+
+        if not has_collection:
+            # Create collection with custom schema to support string IDs
+            from pymilvus import CollectionSchema, FieldSchema, DataType
+
+            fields = [
+                FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=255),
+                FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1536),
+                FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
+                FieldSchema(name="file_id", dtype=DataType.VARCHAR, max_length=255),
+                FieldSchema(name="chunk_uuid", dtype=DataType.VARCHAR, max_length=255),
+            ]
+
+            schema = CollectionSchema(fields=fields, description="LangBot knowledge base vectors")
+
+            await asyncio.to_thread(
+                self.client.create_collection,
+                collection_name=collection,
+                schema=schema,
+                metric_type="COSINE",
+            )
+
+            # Create index for vector field (required for loading/searching)
+            index_params = {
+                "metric_type": "COSINE",
+                "index_type": "AUTOINDEX",
+                "params": {}
+            }
+            await asyncio.to_thread(
+                self.client.create_index,
+                collection_name=collection,
+                field_name="vector",
+                index_params=index_params
+            )
+
+            self.ap.logger.info(f"Created Milvus collection '{collection}' with index")
+        else:
+            self.ap.logger.info(f"Milvus collection '{collection}' already exists")
+
+        self._collections[collection] = collection
+        return collection
+
+    async def add_embeddings(
+        self,
+        collection: str,
+        ids: list[str],
+        embeddings_list: list[list[float]],
+        metadatas: list[dict[str, Any]],
+    ) -> None:
+        """Add vector embeddings to Milvus collection
+
+        Args:
+            collection: Collection name
+            ids: List of unique IDs for each vector
+            embeddings_list: List of embedding vectors
+            metadatas: List of metadata dictionaries for each vector
+        """
+        await self.get_or_create_collection(collection)
+
+        # Prepare data in Milvus format
+        data = []
+        for i, vector_id in enumerate(ids):
+            entry = {
+                "id": vector_id,
+                "vector": embeddings_list[i],
+            }
+            # Add metadata fields
+            if metadatas and i < len(metadatas):
+                metadata = metadatas[i]
+                # Add common metadata fields
+                if "text" in metadata:
+                    entry["text"] = metadata["text"]
+                if "file_id" in metadata:
+                    entry["file_id"] = metadata["file_id"]
+                if "uuid" in metadata:
+                    entry["chunk_uuid"] = metadata["uuid"]
+            data.append(entry)
+
+        # Insert data into Milvus
+        await asyncio.to_thread(
+            self.client.insert,
+            collection_name=collection,
+            data=data
+        )
+
+        # Load collection for searching (Milvus requires this)
+        await asyncio.to_thread(
+            self.client.load_collection,
+            collection_name=collection
+        )
+
+        self.ap.logger.info(f"Added {len(ids)} embeddings to Milvus collection '{collection}'")
+
+    async def search(
+        self, collection: str, query_embedding: list[float], k: int = 5
+    ) -> Dict[str, Any]:
+        """Search for similar vectors in Milvus collection
+
+        Args:
+            collection: Collection name
+            query_embedding: Query vector
+            k: Number of top results to return
+
+        Returns:
+            Dictionary with search results in Chroma-compatible format
+        """
+        await self.get_or_create_collection(collection)
+
+        # Perform search
+        search_params = {
+            "metric_type": "COSINE",
+            "params": {}
+        }
+
+        results = await asyncio.to_thread(
+            self.client.search,
+            collection_name=collection,
+            data=[query_embedding],
+            limit=k,
+            search_params=search_params,
+            output_fields=["text", "file_id", "chunk_uuid"]
+        )
+
+        # Convert results to Chroma-compatible format
+        # Milvus returns: [[ {id, distance, entity: {...}} ]]
+        ids = []
+        distances = []
+        metadatas = []
+
+        if results and len(results) > 0:
+            for hit in results[0]:
+                ids.append(hit.get("id", ""))
+                distances.append(hit.get("distance", 0.0))
+
+                # Build metadata from entity fields
+                entity = hit.get("entity", {})
+                metadata = {}
+                if "text" in entity:
+                    metadata["text"] = entity["text"]
+                if "file_id" in entity:
+                    metadata["file_id"] = entity["file_id"]
+                if "chunk_uuid" in entity:
+                    metadata["uuid"] = entity["chunk_uuid"]
+                metadatas.append(metadata)
+
+        # Return in Chroma-compatible format (nested lists)
+        result = {
+            "ids": [ids],
+            "distances": [distances],
+            "metadatas": [metadatas]
+        }
+
+        self.ap.logger.info(
+            f"Milvus search in '{collection}' returned {len(ids)} results"
+        )
+        return result
+
+    async def delete_by_file_id(self, collection: str, file_id: str) -> None:
+        """Delete vectors from collection by file_id
+
+        Args:
+            collection: Collection name
+            file_id: File ID to filter deletion
+        """
+        await self.get_or_create_collection(collection)
+
+        # Delete entities matching the file_id
+        await asyncio.to_thread(
+            self.client.delete,
+            collection_name=collection,
+            filter=f'file_id == "{file_id}"'
+        )
+        self.ap.logger.info(
+            f"Deleted embeddings from Milvus collection '{collection}' with file_id: {file_id}"
+        )
+
+    async def delete_collection(self, collection: str):
+        """Delete a Milvus collection
+
+        Args:
+            collection: Collection name to delete
+        """
+        if collection in self._collections:
+            del self._collections[collection]
+
+        # Check if collection exists before attempting deletion
+        has_collection = await asyncio.to_thread(
+            self.client.has_collection, collection_name=collection
+        )
+
+        if has_collection:
+            await asyncio.to_thread(
+                self.client.drop_collection, collection_name=collection
+            )
+            self.ap.logger.info(f"Deleted Milvus collection '{collection}'")
+        else:
+            self.ap.logger.warning(f"Milvus collection '{collection}' not found")

src/langbot/pkg/vector/vdbs/pgvector_db.py

@@ -0,0 +1,286 @@
+from __future__ import annotations
+import asyncio
+from typing import Any, Dict
+from sqlalchemy import create_engine, text, Column, String, Text
+from sqlalchemy.orm import declarative_base, sessionmaker, Session
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
+from pgvector.sqlalchemy import Vector
+from langbot.pkg.vector.vdb import VectorDatabase
+from langbot.pkg.core import app
+import uuid
+
+Base = declarative_base()
+
+
+class PgVectorEntry(Base):
+    """SQLAlchemy model for pgvector entries"""
+    __tablename__ = 'langbot_vectors'
+
+    id = Column(String, primary_key=True)
+    collection = Column(String, index=True, nullable=False)
+    embedding = Column(Vector(1536))  # Default dimension, will be created dynamically
+    text = Column(Text)
+    file_id = Column(String, index=True)
+    chunk_uuid = Column(String)
+
+
+class PgVectorDatabase(VectorDatabase):
+    """PostgreSQL with pgvector extension database implementation"""
+
+    def __init__(
+        self,
+        ap: app.Application,
+        connection_string: str = None,
+        host: str = "localhost",
+        port: int = 5432,
+        database: str = "langbot",
+        user: str = "postgres",
+        password: str = "postgres"
+    ):
+        """Initialize pgvector database
+
+        Args:
+            ap: Application instance
+            connection_string: Full PostgreSQL connection string (overrides other params)
+            host: PostgreSQL host
+            port: PostgreSQL port
+            database: Database name
+            user: Database user
+            password: Database password
+        """
+        self.ap = ap
+
+        # Build connection string if not provided
+        if connection_string:
+            self.connection_string = connection_string
+        else:
+            self.connection_string = (
+                f"postgresql+psycopg://{user}:{password}@{host}:{port}/{database}"
+            )
+
+        self.async_connection_string = self.connection_string.replace(
+            "postgresql://", "postgresql+asyncpg://"
+        ).replace(
+            "postgresql+psycopg://", "postgresql+asyncpg://"
+        )
+
+        self.engine = None
+        self.async_engine = None
+        self.SessionLocal = None
+        self.AsyncSessionLocal = None
+        self._collections = set()
+        self._initialize_db()
+
+    def _initialize_db(self):
+        """Initialize database connection and create tables"""
+        try:
+            # Create async engine for async operations
+            self.async_engine = create_async_engine(
+                self.async_connection_string,
+                echo=False,
+                pool_pre_ping=True
+            )
+            self.AsyncSessionLocal = async_sessionmaker(
+                self.async_engine,
+                class_=AsyncSession,
+                expire_on_commit=False
+            )
+
+            # Create sync engine for table creation
+            sync_connection_string = self.connection_string.replace(
+                "postgresql+asyncpg://", "postgresql+psycopg://"
+            )
+            self.engine = create_engine(sync_connection_string, echo=False)
+
+            # Create pgvector extension and tables
+            with self.engine.connect() as conn:
+                # Enable pgvector extension
+                conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+                conn.commit()
+
+            # Create tables
+            Base.metadata.create_all(self.engine)
+
+            self.ap.logger.info(f"Connected to PostgreSQL with pgvector")
+        except Exception as e:
+            self.ap.logger.error(f"Failed to connect to PostgreSQL: {e}")
+            raise
+
+    async def get_or_create_collection(self, collection: str):
+        """Get or create a collection (logical grouping in pgvector)
+
+        Args:
+            collection: Collection name (knowledge base UUID)
+        """
+        # In pgvector, collections are logical - we just track them
+        if collection not in self._collections:
+            self._collections.add(collection)
+            self.ap.logger.info(f"Registered pgvector collection '{collection}'")
+        return collection
+
+    async def add_embeddings(
+        self,
+        collection: str,
+        ids: list[str],
+        embeddings_list: list[list[float]],
+        metadatas: list[dict[str, Any]],
+    ) -> None:
+        """Add vector embeddings to pgvector
+
+        Args:
+            collection: Collection name
+            ids: List of unique IDs for each vector
+            embeddings_list: List of embedding vectors
+            metadatas: List of metadata dictionaries
+        """
+        await self.get_or_create_collection(collection)
+
+        async with self.AsyncSessionLocal() as session:
+            try:
+                for i, vector_id in enumerate(ids):
+                    metadata = metadatas[i] if i < len(metadatas) else {}
+
+                    entry = PgVectorEntry(
+                        id=vector_id,
+                        collection=collection,
+                        embedding=embeddings_list[i],
+                        text=metadata.get("text", ""),
+                        file_id=metadata.get("file_id", ""),
+                        chunk_uuid=metadata.get("uuid", "")
+                    )
+                    session.add(entry)
+
+                await session.commit()
+                self.ap.logger.info(
+                    f"Added {len(ids)} embeddings to pgvector collection '{collection}'"
+                )
+            except Exception as e:
+                await session.rollback()
+                self.ap.logger.error(f"Error adding embeddings to pgvector: {e}")
+                raise
+
+    async def search(
+        self, collection: str, query_embedding: list[float], k: int = 5
+    ) -> Dict[str, Any]:
+        """Search for similar vectors using cosine distance
+
+        Args:
+            collection: Collection name
+            query_embedding: Query vector
+            k: Number of top results to return
+
+        Returns:
+            Dictionary with search results in Chroma-compatible format
+        """
+        await self.get_or_create_collection(collection)
+
+        async with self.AsyncSessionLocal() as session:
+            try:
+                # Use cosine distance for similarity search
+                from sqlalchemy import select, func
+
+                # Query for similar vectors
+                stmt = (
+                    select(
+                        PgVectorEntry.id,
+                        PgVectorEntry.text,
+                        PgVectorEntry.file_id,
+                        PgVectorEntry.chunk_uuid,
+                        PgVectorEntry.embedding.cosine_distance(query_embedding).label('distance')
+                    )
+                    .filter(PgVectorEntry.collection == collection)
+                    .order_by(PgVectorEntry.embedding.cosine_distance(query_embedding))
+                    .limit(k)
+                )
+
+                result = await session.execute(stmt)
+                rows = result.fetchall()
+
+                # Convert to Chroma-compatible format
+                ids = []
+                distances = []
+                metadatas = []
+
+                for row in rows:
+                    ids.append(row.id)
+                    distances.append(float(row.distance))
+                    metadatas.append({
+                        "text": row.text or "",
+                        "file_id": row.file_id or "",
+                        "uuid": row.chunk_uuid or ""
+                    })
+
+                result_dict = {
+                    "ids": [ids],
+                    "distances": [distances],
+                    "metadatas": [metadatas]
+                }
+
+                self.ap.logger.info(
+                    f"pgvector search in '{collection}' returned {len(ids)} results"
+                )
+                return result_dict
+
+            except Exception as e:
+                self.ap.logger.error(f"Error searching pgvector: {e}")
+                raise
+
+    async def delete_by_file_id(self, collection: str, file_id: str) -> None:
+        """Delete vectors by file_id
+
+        Args:
+            collection: Collection name
+            file_id: File ID to filter deletion
+        """
+        await self.get_or_create_collection(collection)
+
+        async with self.AsyncSessionLocal() as session:
+            try:
+                from sqlalchemy import delete
+
+                stmt = delete(PgVectorEntry).where(
+                    PgVectorEntry.collection == collection,
+                    PgVectorEntry.file_id == file_id
+                )
+                await session.execute(stmt)
+                await session.commit()
+
+                self.ap.logger.info(
+                    f"Deleted embeddings from pgvector collection '{collection}' with file_id: {file_id}"
+                )
+            except Exception as e:
+                await session.rollback()
+                self.ap.logger.error(f"Error deleting from pgvector: {e}")
+                raise
+
+    async def delete_collection(self, collection: str):
+        """Delete all vectors in a collection
+
+        Args:
+            collection: Collection name to delete
+        """
+        if collection in self._collections:
+            self._collections.remove(collection)
+
+        async with self.AsyncSessionLocal() as session:
+            try:
+                from sqlalchemy import delete
+
+                stmt = delete(PgVectorEntry).where(
+                    PgVectorEntry.collection == collection
+                )
+                await session.execute(stmt)
+                await session.commit()
+
+                self.ap.logger.info(f"Deleted pgvector collection '{collection}'")
+            except Exception as e:
+                await session.rollback()
+                self.ap.logger.error(f"Error deleting pgvector collection: {e}")
+                raise
+
+    async def close(self):
+        """Close database connections"""
+        if self.async_engine:
+            await self.async_engine.dispose()
+        if self.engine:
+            self.engine.dispose()

src/langbot/templates/config.yaml

@@ -36,6 +36,15 @@ vdb:
         host: localhost
         port: 6333
         api_key: ''
+    milvus:
+        uri: 'http://127.0.0.1:19530'
+        token: ''
+    pgvector:
+        host: '127.0.0.1'
+        port: 5433
+        database: 'langbot'
+        user: 'postgres'
+        password: 'postgres'
 storage:
     use: local
     s3:
@@ -49,4 +58,4 @@ plugin:
     runtime_ws_url: 'ws://langbot_plugin_runtime:5400/control/ws'
     enable_marketplace: true
     cloud_service_url: 'https://space.langbot.app'
-    display_plugin_debug_url: 'http://localhost:5401'
+    display_plugin_debug_url: 'ws://localhost:5401/plugin/debug/ws'

tests/unit_tests/pipeline/conftest.py

@@ -14,6 +14,8 @@ from unittest.mock import AsyncMock, Mock
 
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
+import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 
 from langbot.pkg.pipeline import entities as pipeline_entities
@@ -159,12 +161,18 @@ def sample_message_chain():
 
 @pytest.fixture
 def sample_message_event(sample_message_chain):
-    """Provides sample message event"""
-    event = Mock()
-    event.sender = Mock()
-    event.sender.id = 12345
-    event.time = 1609459200  # 2021-01-01 00:00:00
-    return event
+    """Provides sample message event (FriendMessage)"""
+    sender = platform_entities.Friend(
+        id=12345,
+        nickname='TestUser',
+        remark=None,
+    )
+    return platform_events.FriendMessage(
+        type='FriendMessage',
+        sender=sender,
+        message_chain=sample_message_chain,
+        time=1609459200,  # 2021-01-01 00:00:00
+    )
 
 
 @pytest.fixture

web/src/app/home/bots/components/bot-log/view/BotLogCard.tsx

@@ -44,12 +44,35 @@ export function BotLogCard({ botLog }: { botLog: BotLog }) {
     const strArr = str.split('');
     return strArr;
   }
+
+  // 根据日志级别返回对应的样式类
+  function getLevelStyles(level: string) {
+    switch (level.toLowerCase()) {
+      case 'error':
+        return 'bg-red-100 text-red-800 dark:bg-red-900/30 dark:text-red-400';
+      case 'warning':
+        return 'bg-orange-100 text-orange-800 dark:bg-orange-900/30 dark:text-orange-400';
+      case 'info':
+        return 'bg-blue-100 text-blue-800 dark:bg-blue-900/30 dark:text-blue-400';
+      case 'debug':
+        return 'bg-gray-100 text-gray-800 dark:bg-gray-900/30 dark:text-gray-400';
+      default:
+        return 'bg-gray-100 text-gray-800 dark:bg-gray-900/30 dark:text-gray-400';
+    }
+  }
+
   return (
     <div className={`${styles.botLogCardContainer}`}>
       {/* 头部标签，时间 */}
       <div className={`${styles.cardTitleContainer}`}>
         <div className={`flex flex-row gap-2 items-center`}>
-          <div className={`${styles.tag}`}>{botLog.level}</div>
+          <div
+            className={`px-2 py-1 rounded text-xs font-medium uppercase ${getLevelStyles(
+              botLog.level,
+            )}`}
+          >
+            {botLog.level}
+          </div>
           {botLog.message_session_id && (
             <div
               className={`${styles.tag} ${styles.chatTag}`}

web/src/app/home/bots/components/bot-log/view/BotLogListComponent.tsx

@@ -1,11 +1,19 @@
 'use client';
 
 import { BotLogManager } from '@/app/home/bots/components/bot-log/BotLogManager';
-import { useCallback, useEffect, useRef, useState } from 'react';
+import { useCallback, useEffect, useRef, useState, useMemo } from 'react';
 import { BotLog } from '@/app/infra/http/requestParam/bots/GetBotLogsResponse';
 import { BotLogCard } from '@/app/home/bots/components/bot-log/view/BotLogCard';
 import styles from './botLog.module.css';
 import { Switch } from '@/components/ui/switch';
+import {
+  Popover,
+  PopoverContent,
+  PopoverTrigger,
+} from '@/components/ui/popover';
+import { Button } from '@/components/ui/button';
+import { Checkbox } from '@/components/ui/checkbox';
+import { ChevronDownIcon } from 'lucide-react';
 import { debounce } from 'lodash';
 import { useTranslation } from 'react-i18next';
 
@@ -14,9 +22,21 @@ export function BotLogListComponent({ botId }: { botId: string }) {
   const manager = useRef(new BotLogManager(botId)).current;
   const [botLogList, setBotLogList] = useState<BotLog[]>([]);
   const [autoFlush, setAutoFlush] = useState(true);
+  const [selectedLevels, setSelectedLevels] = useState<string[]>([
+    'info',
+    'warning',
+    'error',
+  ]);
   const listContainerRef = useRef<HTMLDivElement>(null);
   const botLogListRef = useRef<BotLog[]>(botLogList);
 
+  const logLevels = [
+    { value: 'error', label: 'ERROR' },
+    { value: 'warning', label: 'WARNING' },
+    { value: 'info', label: 'INFO' },
+    { value: 'debug', label: 'DEBUG' },
+  ];
+
   useEffect(() => {
     initComponent();
     return () => {
@@ -28,6 +48,42 @@ export function BotLogListComponent({ botId }: { botId: string }) {
     botLogListRef.current = botLogList;
   }, [botLogList]);
 
+  // 根据级别过滤日志
+  const filteredLogs = useMemo(() => {
+    if (selectedLevels.length === 0) {
+      return botLogList;
+    }
+    return botLogList.filter((log) => selectedLevels.includes(log.level));
+  }, [botLogList, selectedLevels]);
+
+  const handleLevelToggle = (levelValue: string) => {
+    setSelectedLevels((prev) => {
+      if (prev.includes(levelValue)) {
+        return prev.filter((l) => l !== levelValue);
+      } else {
+        return [...prev, levelValue];
+      }
+    });
+  };
+
+  const getDisplayText = () => {
+    if (selectedLevels.length === 0) {
+      return t('bots.selectLevel');
+    }
+    if (selectedLevels.length === logLevels.length) {
+      return t('bots.allLevels');
+    }
+    // 如果选中3个或以上，显示数量
+    if (selectedLevels.length >= 3) {
+      return `${selectedLevels.length} ${t('bots.levelsSelected')}`;
+    }
+    // 显示选中级别的标签（大写形式）
+    return logLevels
+      .filter((level) => selectedLevels.includes(level.value))
+      .map((level) => level.label)
+      .join(', ');
+  };
+
   // 观测自动刷新状态
   useEffect(() => {
     if (autoFlush) {
@@ -116,9 +172,43 @@ export function BotLogListComponent({ botId }: { botId: string }) {
       <div className={`${styles.listHeader}`}>
         <div className={'mr-2'}>{t('bots.enableAutoRefresh')}</div>
         <Switch checked={autoFlush} onCheckedChange={(e) => setAutoFlush(e)} />
+        <div className={'ml-4 mr-2'}>{t('bots.logLevel')}</div>
+        <Popover>
+          <PopoverTrigger asChild>
+            <Button
+              variant="outline"
+              size="sm"
+              className="w-[180px] flex items-center justify-between"
+            >
+              <span className="text-sm truncate flex-1 text-left">
+                {getDisplayText()}
+              </span>
+              <ChevronDownIcon className="ml-2 h-4 w-4 flex-shrink-0" />
+            </Button>
+          </PopoverTrigger>
+          <PopoverContent className="w-[180px] p-2">
+            <div className="flex flex-col gap-2">
+              {logLevels.map((level) => (
+                <div key={level.value} className="flex items-center space-x-2">
+                  <Checkbox
+                    id={level.value}
+                    checked={selectedLevels.includes(level.value)}
+                    onCheckedChange={() => handleLevelToggle(level.value)}
+                  />
+                  <label
+                    htmlFor={level.value}
+                    className="text-sm font-medium leading-none peer-disabled:cursor-not-allowed peer-disabled:opacity-70 cursor-pointer"
+                  >
+                    {level.label}
+                  </label>
+                </div>
+              ))}
+            </div>
+          </PopoverContent>
+        </Popover>
       </div>
 
-      {botLogList.map((botLog) => {
+      {filteredLogs.map((botLog) => {
         return <BotLogCard botLog={botLog} key={botLog.seq_id} />;
       })}
     </div>

web/src/i18n/locales/en-US.ts

@@ -192,6 +192,10 @@ const enUS = {
     webhookUrlCopied: 'Webhook URL copied',
     webhookUrlHint:
       'Click the input to select all, then press Ctrl+C (Mac: Cmd+C) to copy, or click the button',
+    logLevel: 'Log Level',
+    allLevels: 'All Levels',
+    selectLevel: 'Select Level',
+    levelsSelected: 'levels selected',
   },
   plugins: {
     title: 'Extensions',

web/src/i18n/locales/ja-JP.ts

@@ -194,6 +194,10 @@ const jaJP = {
     webhookUrlCopied: 'Webhook URL をコピーしました',
     webhookUrlHint:
       '入力ボックスをクリックして全選択し、Ctrl+C (Mac: Cmd+C) でコピーするか、右側のボタンをクリックしてください',
+    logLevel: 'ログレベル',
+    allLevels: 'すべてのレベル',
+    selectLevel: 'レベルを選択',
+    levelsSelected: 'レベル選択済み',
   },
   plugins: {
     title: '拡張機能',

web/src/i18n/locales/zh-Hans.ts

@@ -187,6 +187,10 @@ const zhHans = {
     webhookUrlCopied: 'Webhook 地址已复制',
     webhookUrlHint:
       '点击输入框自动全选，然后按 Ctrl+C (Mac: Cmd+C) 复制，或点击右侧按钮',
+    logLevel: '日志级别',
+    allLevels: '全部级别',
+    selectLevel: '选择级别',
+    levelsSelected: '个级别已选',
   },
   plugins: {
     title: '插件扩展',

web/src/i18n/locales/zh-Hant.ts

@@ -187,6 +187,10 @@ const zhHant = {
     webhookUrlCopied: 'Webhook 位址已複製',
     webhookUrlHint:
       '點擊輸入框自動全選，然後按 Ctrl+C (Mac: Cmd+C) 複製，或點擊右側按鈕',
+    logLevel: '日誌級別',
+    allLevels: '全部級別',
+    selectLevel: '選擇級別',
+    levelsSelected: '個級別已選',
   },
   plugins: {
     title: '外掛擴展',