fix: wrong agentid

- Add get_user_info() to WecomClient to fetch user name via /user/get API
- Update WecomEventConverter.target2yiri to accept bot param and fetch real user name
- Update register_listener call to pass self.bot for user name lookup
- URL-encode userid parameter for safety

Co-authored-by: RockChinQ <45992437+RockChinQ@users.noreply.github.com>

pyproject.toml

@@ -61,7 +61,7 @@ dependencies = [
     "html2text>=2024.2.26",
     "langchain>=0.2.0",
     "langchain-text-splitters>=0.0.1",
-    "chromadb>=0.4.24",
+    "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
     "pyseekdb==1.1.0.post3",
     "langbot-plugin==0.3.0",

src/langbot/libs/wecom_api/api.py

@@ -4,6 +4,7 @@ import base64
 import binascii
 import httpx
 import traceback
+from urllib.parse import quote
 from quart import Quart
 import xml.etree.ElementTree as ET
 from typing import Callable, Dict, Any
@@ -67,6 +68,31 @@ class WecomClient:
                 await self.logger.error(f'获取accesstoken失败:{response.json()}')
                 raise Exception(f'未获取access token: {data}')
 
+    async def get_user_info(self, userid: str) -> dict:
+        """
+        Get user information by user ID using the application secret.
+
+        Args:
+            userid: The user ID to look up.
+
+        Returns:
+            dict: User information including 'name' field.
+        """
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+
+        url = self.base_url + '/user/get?access_token=' + self.access_token + '&userid=' + quote(userid)
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url)
+            data = response.json()
+            if data.get('errcode') == 40014 or data.get('errcode') == 42001:
+                self.access_token = await self.get_access_token(self.secret)
+                return await self.get_user_info(userid)
+            if data.get('errcode', 0) != 0:
+                await self.logger.error(f'获取用户信息失败:{data}')
+                return {}
+            return data
+
     async def get_users(self):
         if not self.check_access_token_for_contacts():
             self.access_token_for_contacts = await self.get_access_token(self.secret_for_contacts)

src/langbot/libs/wecom_customer_service_api/api.py

@@ -10,6 +10,7 @@ from typing import Callable
 from .wecomcsevent import WecomCSEvent
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
 import aiofiles
+import time
 
 
 class WecomCSClient:
@@ -34,6 +35,10 @@ class WecomCSClient:
         self.unified_mode = unified_mode
         self.app = Quart(__name__)
 
+        # Customer info cache: {external_userid: (info_dict, timestamp)}
+        self._customer_cache: dict[str, tuple[dict, float]] = {}
+        self._cache_ttl = 60  # Cache TTL in seconds (1 minute)
+
         # 只有在非统一模式下才注册独立路由
         if not self.unified_mode:
             self.app.add_url_rule(
@@ -378,3 +383,53 @@ class WecomCSClient:
     async def get_media_id(self, image: platform_message.Image):
         media_id = await self.upload_to_work(image=image)
         return media_id
+
+    async def get_customer_info(self, external_userid: str) -> dict | None:
+        """
+        Get customer information by external_userid with caching.
+
+        Uses a 1-minute cache to avoid repeated API calls for the same user.
+
+        Args:
+            external_userid: The external user ID of the customer.
+
+        Returns:
+            Customer info dict with 'nickname', 'avatar', etc., or None if not found.
+        """
+        # Check cache first
+        current_time = time.time()
+        if external_userid in self._customer_cache:
+            cached_info, cached_time = self._customer_cache[external_userid]
+            if current_time - cached_time < self._cache_ttl:
+                return cached_info
+
+        # Cache miss or expired, fetch from API
+        if not await self.check_access_token():
+            self.access_token = await self.get_access_token(self.secret)
+
+        url = f'{self.base_url}/kf/customer/batchget?access_token={self.access_token}'
+
+        payload = {
+            'external_userid_list': [external_userid],
+        }
+
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, json=payload)
+            data = response.json()
+
+            if data.get('errcode') in [40014, 42001]:
+                self.access_token = await self.get_access_token(self.secret)
+                return await self.get_customer_info(external_userid)
+
+            if data.get('errcode', 0) != 0:
+                if self.logger:
+                    await self.logger.warning(f'Failed to get customer info: {data}')
+                return None
+
+            customer_list = data.get('customer_list', [])
+            if customer_list:
+                customer_info = customer_list[0]
+                # Store in cache
+                self._customer_cache[external_userid] = (customer_info, current_time)
+                return customer_info
+            return None

src/langbot/pkg/api/http/service/monitoring.py

@@ -30,6 +30,7 @@ class MonitoringService:
         level: str = 'info',
         platform: str | None = None,
         user_id: str | None = None,
+        user_name: str | None = None,
         runner_name: str | None = None,
         variables: str | None = None,
         role: str = 'user',
@@ -49,6 +50,7 @@ class MonitoringService:
             'level': level,
             'platform': platform,
             'user_id': user_id,
+            'user_name': user_name,
             'runner_name': runner_name,
             'variables': variables,
             'role': role,
@@ -152,6 +154,7 @@ class MonitoringService:
         pipeline_name: str,
         platform: str | None = None,
         user_id: str | None = None,
+        user_name: str | None = None,
     ) -> None:
         """Record a new session"""
         session_data = {
@@ -166,6 +169,7 @@ class MonitoringService:
             'is_active': True,
             'platform': platform,
             'user_id': user_id,
+            'user_name': user_name,
         }
 
         await self.ap.persistence_mgr.execute_async(

src/langbot/pkg/entity/persistence/monitoring.py

@@ -20,6 +20,7 @@ class MonitoringMessage(Base):
     level = sqlalchemy.Column(sqlalchemy.String(50), nullable=False)  # info, warning, error, debug
     platform = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
     user_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
+    user_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)  # User display name
     runner_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)  # Runner name for this query
     variables = sqlalchemy.Column(sqlalchemy.Text, nullable=True)  # Query variables as JSON string
     role = sqlalchemy.Column(sqlalchemy.String(50), nullable=True, default='user')  # user, assistant
@@ -64,6 +65,7 @@ class MonitoringSession(Base):
     is_active = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False, default=True, index=True)
     platform = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
     user_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
+    user_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)  # User display name
 
 
 class MonitoringError(Base):

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

@@ -0,0 +1,74 @@
+from .. import migration
+
+import sqlalchemy
+import json
+
+
+@migration.migration_class(21)
+class DBMigrateMergeExceptionHandling(migration.DBMigration):
+    """Merge hide-exception and block-failed-request-output into a single exception-handling select option,
+    and add failure-hint field.
+
+    Conversion logic:
+    - block-failed-request-output=true  ->  exception-handling: hide
+    - hide-exception=true               ->  exception-handling: show-hint
+    - hide-exception=false              ->  exception-handling: show-error
+    """
+
+    async def upgrade(self):
+        """Upgrade"""
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text('SELECT uuid, config FROM legacy_pipelines')
+        )
+        pipelines = result.fetchall()
+
+        current_version = self.ap.ver_mgr.get_current_version()
+
+        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]
+
+            if 'output' not in config:
+                config['output'] = {}
+            if 'misc' not in config['output']:
+                config['output']['misc'] = {}
+
+            misc = config['output']['misc']
+
+            # Determine new exception-handling value from legacy fields
+            hide_exception = misc.get('hide-exception', True)
+            block_failed = misc.get('block-failed-request-output', False)
+
+            if block_failed:
+                exception_handling = 'hide'
+            elif hide_exception:
+                exception_handling = 'show-hint'
+            else:
+                exception_handling = 'show-error'
+
+            misc['exception-handling'] = exception_handling
+
+            # Add failure-hint with default value
+            misc['failure-hint'] = 'Request failed.'
+
+            # Remove legacy fields
+            misc.pop('hide-exception', None)
+
+            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"""
+        pass

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

@@ -0,0 +1,73 @@
+import sqlalchemy
+from .. import migration
+
+
+@migration.migration_class(22)
+class DBMigrateMonitoringUserId(migration.DBMigration):
+    """Add user_id and user_name columns to monitoring_sessions table
+
+    This migration adds the missing user_id column and also ensures user_name
+    column exists (in case migration 21 failed or was skipped).
+    """
+
+    async def _table_exists(self, table_name: str) -> bool:
+        """Check if a table exists (works for both SQLite and PostgreSQL)."""
+        if self.ap.persistence_mgr.db.name == 'postgresql':
+            result = await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.text(
+                    'SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = :table_name);'
+                ).bindparams(table_name=table_name)
+            )
+            return bool(result.scalar())
+        else:
+            result = await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.text("SELECT name FROM sqlite_master WHERE type='table' AND name=:table_name;").bindparams(
+                    table_name=table_name
+                )
+            )
+            return result.first() is not None
+
+    async def _get_table_columns(self, table_name: str) -> list[str]:
+        """Get column names from a table (works for both SQLite and PostgreSQL)."""
+        if self.ap.persistence_mgr.db.name == 'postgresql':
+            result = await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.text(
+                    'SELECT column_name FROM information_schema.columns WHERE table_name = :table_name;'
+                ).bindparams(table_name=table_name)
+            )
+            return [row[0] for row in result.fetchall()]
+        else:
+            if not table_name.isidentifier():
+                raise ValueError(f'Invalid table name: {table_name}')
+            result = await self.ap.persistence_mgr.execute_async(sqlalchemy.text(f'PRAGMA table_info({table_name});'))
+            return [row[1] for row in result.fetchall()]
+
+    async def _add_column_if_not_exists(self, table_name: str, column_name: str, column_type: str):
+        """Add a column to a table if it does not already exist."""
+        columns = await self._get_table_columns(table_name)
+        if column_name in columns:
+            self.ap.logger.debug('%s column already exists in %s.', column_name, table_name)
+            return
+        await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.text(f'ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type};')
+        )
+        self.ap.logger.info('Added %s column to %s table.', column_name, table_name)
+
+    async def upgrade(self):
+        # Check if monitoring_sessions table exists
+        if not await self._table_exists('monitoring_sessions'):
+            self.ap.logger.warning('monitoring_sessions table does not exist, skipping migration.')
+            return
+
+        # Add user_id column to monitoring_sessions table
+        await self._add_column_if_not_exists('monitoring_sessions', 'user_id', 'VARCHAR(255)')
+
+        # Add user_name column to monitoring_sessions table (in case migration 21 failed)
+        await self._add_column_if_not_exists('monitoring_sessions', 'user_name', 'VARCHAR(255)')
+
+        # Add user_name column to monitoring_messages table (in case migration 21 failed)
+        if await self._table_exists('monitoring_messages'):
+            await self._add_column_if_not_exists('monitoring_messages', 'user_name', 'VARCHAR(255)')
+
+    async def downgrade(self):
+        pass

src/langbot/pkg/pipeline/monitoring_helper.py

@@ -34,6 +34,15 @@ class MonitoringHelper:
             # Check if session exists, if not, record session start
             session_id = f'{query.launcher_type}_{query.launcher_id}'
 
+            # Get sender name from message event
+            sender_name = None
+            if hasattr(query, 'message_event'):
+                if hasattr(query.message_event, 'sender'):
+                    if hasattr(query.message_event.sender, 'nickname'):
+                        sender_name = query.message_event.sender.nickname
+                    elif hasattr(query.message_event.sender, 'member_name'):
+                        sender_name = query.message_event.sender.member_name
+
             # Try to record message
             # Use JSON serialization to preserve message chain structure (including image URLs, etc.)
             if hasattr(query, 'message_chain') and hasattr(query.message_chain, 'model_dump'):
@@ -57,6 +66,7 @@ class MonitoringHelper:
                 if hasattr(query.launcher_type, 'value')
                 else str(query.launcher_type),
                 user_id=query.sender_id,
+                user_name=sender_name,
                 runner_name=runner_name,
                 variables=None,  # Will be updated in record_query_success
             )
@@ -80,6 +90,7 @@ class MonitoringHelper:
                     if hasattr(query.launcher_type, 'value')
                     else str(query.launcher_type),
                     user_id=query.sender_id,
+                    user_name=sender_name,
                 )
 
             return message_id
@@ -128,6 +139,15 @@ class MonitoringHelper:
         try:
             session_id = f'{query.launcher_type}_{query.launcher_id}'
 
+            # Get sender name from message event
+            sender_name = None
+            if hasattr(query, 'message_event'):
+                if hasattr(query.message_event, 'sender'):
+                    if hasattr(query.message_event.sender, 'nickname'):
+                        sender_name = query.message_event.sender.nickname
+                    elif hasattr(query.message_event.sender, 'member_name'):
+                        sender_name = query.message_event.sender.member_name
+
             # Extract response content from resp_message_chain
             if hasattr(query, 'resp_message_chain') and query.resp_message_chain:
                 # Serialize the last response message chain
@@ -162,6 +182,7 @@ class MonitoringHelper:
                 if hasattr(query.launcher_type, 'value')
                 else str(query.launcher_type),
                 user_id=query.sender_id,
+                user_name=sender_name,
                 runner_name=runner_name,
                 role='assistant',
             )
@@ -183,6 +204,15 @@ class MonitoringHelper:
         try:
             session_id = f'{query.launcher_type}_{query.launcher_id}'
 
+            # Get sender name from message event
+            sender_name = None
+            if hasattr(query, 'message_event'):
+                if hasattr(query.message_event, 'sender'):
+                    if hasattr(query.message_event.sender, 'nickname'):
+                        sender_name = query.message_event.sender.nickname
+                    elif hasattr(query.message_event.sender, 'member_name'):
+                        sender_name = query.message_event.sender.member_name
+
             # Record error message
             message_id = await ap.monitoring_service.record_message(
                 bot_id=bot_id,
@@ -197,6 +227,7 @@ class MonitoringHelper:
                 if hasattr(query.launcher_type, 'value')
                 else str(query.launcher_type),
                 user_id=query.sender_id,
+                user_name=sender_name,
                 runner_name=runner_name,
             )
 

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

@@ -149,12 +149,19 @@ class ChatMessageHandler(handler.MessageHandler):
                 self.ap.logger.error(f'Conversation({query.query_id}) Request Failed: {error_info}')
                 traceback.print_exc()
 
-                hide_exception_info = query.pipeline_config['output']['misc']['hide-exception']
+                exception_handling = query.pipeline_config['output']['misc'].get('exception-handling', 'show-hint')
+
+                if exception_handling == 'show-error':
+                    user_notice = f'{e}'
+                elif exception_handling == 'show-hint':
+                    user_notice = query.pipeline_config['output']['misc'].get('failure-hint', 'Request failed.')
+                else:  # hide
+                    user_notice = None
 
                 yield entities.StageProcessResult(
                     result_type=entities.ResultType.INTERRUPT,
                     new_query=query,
-                    user_notice='请求失败' if hide_exception_info else f'{e}',
+                    user_notice=user_notice,
                     error_notice=f'{e}',
                     debug_notice=traceback.format_exc(),
                 )

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

@@ -148,51 +148,54 @@ class WecomEventConverter(abstract_platform_adapter.AbstractEventConverter):
             pass
 
         if type(event) is platform_events.FriendMessage:
-            payload = {
+            return event.source_platform_object
-                'MsgType': 'text',
-                'Content': '',
-                'FromUserName': event.sender.id,
-                'ToUserName': bot_account_id,
-                'CreateTime': int(datetime.datetime.now().timestamp()),
-                'AgentID': event.sender.nickname,
-            }
-            wecom_event = WecomEvent.from_payload(payload=payload)
-            if not wecom_event:
-                raise ValueError('无法从 message_data 构造 WecomEvent 对象')
-
-            return wecom_event
 
     @staticmethod
-    async def target2yiri(event: WecomEvent):
+    async def target2yiri(event: WecomEvent, bot: WecomClient = None):
         """
         将 WecomEvent 转换为平台的 FriendMessage 对象。
 
         Args:
             event (WecomEvent): 企业微信事件。
+            bot (WecomClient): 企业微信客户端，用于获取用户信息。
 
         Returns:
             platform_events.FriendMessage: 转换后的 FriendMessage 对象。
         """
+        # Try to get the user's real name from the WeCom API
+        nickname = str(event.user_id)
+        if bot and event.user_id:
+            try:
+                user_info = await bot.get_user_info(event.user_id)
+                if user_info and user_info.get('name'):
+                    nickname = user_info.get('name')
+            except Exception:
+                pass  # Fall back to user_id as nickname
+
         # 转换消息链
         if event.type == 'text':
             yiri_chain = await WecomMessageConverter.target2yiri(event.message, event.message_id)
             friend = platform_entities.Friend(
                 id=f'u{event.user_id}',
-                nickname=str(event.agent_id),
+                nickname=nickname,
                 remark='',
             )
 
-            return platform_events.FriendMessage(sender=friend, message_chain=yiri_chain, time=event.timestamp)
+            return platform_events.FriendMessage(
+                sender=friend, message_chain=yiri_chain, time=event.timestamp, source_platform_object=event
+            )
         elif event.type == 'image':
             friend = platform_entities.Friend(
                 id=f'u{event.user_id}',
-                nickname=str(event.agent_id),
+                nickname=nickname,
                 remark='',
             )
 
             yiri_chain = await WecomMessageConverter.target2yiri_image(picurl=event.picurl, message_id=event.message_id)
 
-            return platform_events.FriendMessage(sender=friend, message_chain=yiri_chain, time=event.timestamp)
+            return platform_events.FriendMessage(
+                sender=friend, message_chain=yiri_chain, time=event.timestamp, source_platform_object=event
+            )
 
 
 class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
@@ -210,7 +213,6 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             'secret',
             'token',
             'EncodingAESKey',
-            'contacts_secret',
         ]
 
         missing_keys = [key for key in required_keys if key not in config]
@@ -223,7 +225,7 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             secret=config['secret'],
             token=config['token'],
             EncodingAESKey=config['EncodingAESKey'],
-            contacts_secret=config['contacts_secret'],
+            contacts_secret=config.get('contacts_secret', ''),  # Optional, kept for backward compatibility
             logger=logger,
             unified_mode=True,
             api_base_url=config.get('api_base_url', 'https://qyapi.weixin.qq.com/cgi-bin'),
@@ -248,18 +250,17 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
     ):
         Wecom_event = await WecomEventConverter.yiri2target(message_source, self.bot_account_id, self.bot)
         content_list = await WecomMessageConverter.yiri2target(message, self.bot)
-        fixed_user_id = Wecom_event.user_id
+        # user_id is the original FromUserName from WecomEvent
-        # 删掉开头的u
+        user_id = Wecom_event.user_id
-        fixed_user_id = fixed_user_id[1:]
         for content in content_list:
             if content['type'] == 'text':
-                await self.bot.send_private_msg(fixed_user_id, Wecom_event.agent_id, content['content'])
+                await self.bot.send_private_msg(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'])
+                await self.bot.send_image(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'])
+                await self.bot.send_voice(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'])
+                await self.bot.send_file(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)
@@ -287,7 +288,7 @@ class WecomAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         async def on_message(event: WecomEvent):
             self.bot_account_id = event.receiver_id
             try:
-                return await callback(await self.event_converter.target2yiri(event), self)
+                return await callback(await self.event_converter.target2yiri(event, self.bot), self)
             except Exception:
                 await self.logger.error(f'Error in wecom callback: {traceback.format_exc()}')
 

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

@@ -39,13 +39,6 @@ spec:
       type: string
       required: true
       default: ""
-    - name: contacts_secret
-      label:
-        en_US: Contacts Secret
-        zh_Hans: 通讯录密钥
-      type: string
-      required: true
-      default: ""
     - name: api_base_url
       label:
         en_US: API Base URL

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

@@ -81,22 +81,33 @@ class WecomEventConverter(abstract_platform_adapter.AbstractEventConverter):
             return event.source_platform_object
 
     @staticmethod
-    async def target2yiri(event: WecomCSEvent):
+    async def target2yiri(event: WecomCSEvent, bot: WecomCSClient = None):
         """
         将 WecomEvent 转换为平台的 FriendMessage 对象。
 
         Args:
             event (WecomEvent): 企业微信客服事件。
+            bot (WecomCSClient): 企业微信客服客户端，用于获取用户信息。
 
         Returns:
             platform_events.FriendMessage: 转换后的 FriendMessage 对象。
         """
+        # Try to get customer nickname from WeChat API
+        nickname = str(event.user_id)
+        if bot and event.user_id:
+            try:
+                customer_info = await bot.get_customer_info(event.user_id)
+                if customer_info and customer_info.get('nickname'):
+                    nickname = customer_info.get('nickname')
+            except Exception:
+                pass  # Fall back to user_id as nickname
+
         # 转换消息链
         if event.type == 'text':
             yiri_chain = await WecomMessageConverter.target2yiri(event.message, event.message_id)
             friend = platform_entities.Friend(
                 id=f'u{event.user_id}',
-                nickname=str(event.user_id),
+                nickname=nickname,
                 remark='',
             )
 
@@ -106,7 +117,7 @@ class WecomEventConverter(abstract_platform_adapter.AbstractEventConverter):
         elif event.type == 'image':
             friend = platform_entities.Friend(
                 id=f'u{event.user_id}',
-                nickname=str(event.user_id),
+                nickname=nickname,
                 remark='',
             )
 
@@ -187,7 +198,7 @@ class WecomCSAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
         async def on_message(event: WecomCSEvent):
             self.bot_account_id = event.receiver_id
             try:
-                return await callback(await self.event_converter.target2yiri(event), self)
+                return await callback(await self.event_converter.target2yiri(event, self.bot), self)
             except Exception:
                 await self.logger.error(f'Error in wecomcs callback: {traceback.format_exc()}')
 

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

@@ -441,6 +441,7 @@ class DifyServiceAPIRunner(runner.RequestRunner):
         is_final = False
         think_start = False
         think_end = False
+        yielded_final = False
 
         remove_think = self.pipeline_config['output'].get('misc', '').get('remove-think')
 
@@ -493,13 +494,19 @@ class DifyServiceAPIRunner(runner.RequestRunner):
                     if answer:
                         basic_mode_pending_chunk = answer
 
-            if (is_final or message_idx % 8 == 0) and (basic_mode_pending_chunk != '' or is_final):
+            if (
+                not yielded_final
+                and (is_final or message_idx % 8 == 0)
+                and (basic_mode_pending_chunk != '' or is_final)
+            ):
                 # content, _ = self._process_thinking_content(basic_mode_pending_chunk)
                 yield provider_message.MessageChunk(
                     role='assistant',
                     content=basic_mode_pending_chunk,
                     is_final=is_final,
                 )
+                if is_final:
+                    yielded_final = True
 
         if chunk is None:
             raise errors.DifyAPIError('Dify API 没有返回任何响应，请检查网络连接和API配置')

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

@@ -74,7 +74,13 @@ class LocalAgentRunner(runner.RequestRunner):
                     self.ap.logger.warning(f'Knowledge base {kb_uuid} not found, skipping')
                     continue
 
-                result = await kb.retrieve(user_message_text)
+                result = await kb.retrieve(
+                    user_message_text,
+                    settings={
+                        'sender_id': str(query.sender_id),
+                        'session_name': f'{query.session.launcher_type.value}_{query.session.launcher_id}',
+                    },
+                )
 
                 if result:
                     all_results.extend(result)

src/langbot/pkg/rag/knowledge/kbmgr.py

@@ -321,13 +321,19 @@ class RuntimeKnowledgeBase(KnowledgeBaseInterface):
         if not plugin_id:
             raise ValueError(f'No RAG plugin ID configured for KB {kb.uuid}. Retrieval failed.')
 
+        # Session context (e.g. session_name) stays in retrieval_settings
+        # for plugins that need it. Do NOT move them into filters, as filters
+        # are passed directly to vector_search by some plugins (e.g. LangRAG)
+        # and would cause empty results when the metadata field doesn't exist.
+        filters = settings.pop('filters', {})
+
         retrieval_context = {
             'query': query,
             'knowledge_base_id': kb.uuid,
             'collection_id': kb.collection_id or kb.uuid,
             'retrieval_settings': settings,
             'creation_settings': kb.creation_settings or {},
-            'filters': settings.pop('filters', {}),
+            'filters': filters,
         }
 
         result = await self.ap.plugin_connector.call_rag_retrieve(

src/langbot/pkg/utils/constants.py

@@ -2,7 +2,7 @@ import langbot
 
 semantic_version = f'v{langbot.__version__}'
 
-required_database_version = 20
+required_database_version = 22
 """Tag the version of the database schema, used to check if the database needs to be migrated"""
 
 debug_mode = False

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

@@ -2,11 +2,14 @@ from __future__ import annotations
 import asyncio
 from typing import Any
 from chromadb import PersistentClient
-from langbot.pkg.vector.vdb import VectorDatabase
+from langbot.pkg.vector.vdb import VectorDatabase, SearchType
 from langbot.pkg.core import app
 import chromadb
 import chromadb.errors
 
+# RRF smoothing constant (standard value from the literature)
+_RRF_K = 60
+
 
 class ChromaVectorDatabase(VectorDatabase):
     def __init__(self, ap: app.Application, base_path: str = './data/chroma'):
@@ -14,6 +17,10 @@ class ChromaVectorDatabase(VectorDatabase):
         self.client = PersistentClient(path=base_path)
         self._collections = {}
 
+    @classmethod
+    def supported_search_types(cls) -> list[SearchType]:
+        return [SearchType.VECTOR, SearchType.FULL_TEXT, SearchType.HYBRID]
+
     async def get_or_create_collection(self, collection: str) -> chromadb.Collection:
         if collection not in self._collections:
             self._collections[collection] = await asyncio.to_thread(
@@ -34,8 +41,8 @@ class ChromaVectorDatabase(VectorDatabase):
         kwargs: dict[str, Any] = dict(embeddings=embeddings_list, ids=ids, metadatas=metadatas)
         if documents is not None:
             kwargs['documents'] = documents
-        await asyncio.to_thread(col.add, **kwargs)
+        await asyncio.to_thread(col.upsert, **kwargs)
-        self.ap.logger.info(f"Added {len(ids)} embeddings to Chroma collection '{collection}'.")
+        self.ap.logger.info(f"Upserted {len(ids)} embeddings to Chroma collection '{collection}'.")
 
     async def search(
         self,
@@ -47,6 +54,23 @@ class ChromaVectorDatabase(VectorDatabase):
         filter: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         col = await self.get_or_create_collection(collection)
+
+        if search_type == SearchType.FULL_TEXT:
+            return await self._full_text_search(col, collection, k, query_text, filter)
+        elif search_type == SearchType.HYBRID:
+            return await self._hybrid_search(col, collection, query_embedding, k, query_text, filter)
+
+        # Default: vector search
+        return await self._vector_search(col, collection, query_embedding, k, filter)
+
+    async def _vector_search(
+        self,
+        col: chromadb.Collection,
+        collection: str,
+        query_embedding: list[float],
+        k: int,
+        filter: dict[str, Any] | None,
+    ) -> dict[str, Any]:
         query_kwargs: dict[str, Any] = dict(
             query_embeddings=query_embedding,
             n_results=k,
@@ -55,9 +79,137 @@ class ChromaVectorDatabase(VectorDatabase):
         if filter:
             query_kwargs['where'] = filter
         results = await asyncio.to_thread(col.query, **query_kwargs)
-        self.ap.logger.info(f"Chroma search in '{collection}' returned {len(results.get('ids', [[]])[0])} results.")
+        self.ap.logger.info(
+            f"Chroma vector search in '{collection}' returned {len(results.get('ids', [[]])[0])} results."
+        )
         return results
 
+    async def _full_text_search(
+        self,
+        col: chromadb.Collection,
+        collection: str,
+        k: int,
+        query_text: str,
+        filter: dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        if not query_text:
+            return {'ids': [[]], 'metadatas': [[]], 'distances': [[]], 'documents': [[]]}
+
+        get_kwargs: dict[str, Any] = dict(
+            where_document={'$contains': query_text},
+            include=['metadatas', 'documents'],
+            limit=k,
+        )
+        if filter:
+            get_kwargs['where'] = filter
+        results = await asyncio.to_thread(col.get, **get_kwargs)
+
+        # col.get returns flat lists; wrap into column-major format.
+        # Distances are all 0.0 because Chroma's local $contains is a boolean
+        # filter with no relevance scoring.  Chroma's BM25 sparse embedding
+        # function (ChromaBm25EmbeddingFunction) can generate scored sparse
+        # vectors, but sparse vector *indexing* is only available on Chroma
+        # Cloud, not locally.  For ranked results, use hybrid mode or apply a
+        # reranker in a downstream stage.
+        ids = results.get('ids', [])
+        metadatas = results.get('metadatas', []) or [None] * len(ids)
+        documents = results.get('documents', []) or [None] * len(ids)
+        distances = [0.0] * len(ids)
+
+        self.ap.logger.info(f"Chroma full-text search in '{collection}' returned {len(ids)} results.")
+        return {'ids': [ids], 'metadatas': [metadatas], 'distances': [distances], 'documents': [documents]}
+
+    async def _hybrid_search(
+        self,
+        col: chromadb.Collection,
+        collection: str,
+        query_embedding: list[float],
+        k: int,
+        query_text: str,
+        filter: dict[str, Any] | None,
+    ) -> dict[str, Any]:
+        # Fall back to pure vector search when no text is provided
+        if not query_text:
+            return await self._vector_search(col, collection, query_embedding, k, filter)
+
+        # Run vector search and full-text search in parallel
+        vector_task = self._vector_search(col, collection, query_embedding, k, filter)
+        text_task = self._full_text_search(col, collection, k, query_text, filter)
+        vector_results, text_results = await asyncio.gather(vector_task, text_task)
+
+        vector_ids = vector_results.get('ids', [[]])[0]
+        text_ids = text_results.get('ids', [[]])[0]
+
+        if not vector_ids and not text_ids:
+            return {'ids': [[]], 'metadatas': [[]], 'distances': [[]], 'documents': [[]]}
+
+        # RRF fusion
+        fused = self._rrf_fuse([vector_ids, text_ids], k)
+        if not fused:
+            return {'ids': [[]], 'metadatas': [[]], 'distances': [[]], 'documents': [[]]}
+
+        fused_ids = [doc_id for doc_id, _ in fused]
+
+        # Fetch full metadata and documents for fused results
+        fetched = await asyncio.to_thread(col.get, ids=fused_ids, include=['metadatas', 'documents'])
+
+        # col.get returns results in arbitrary order; re-order to match fused ranking
+        fetched_map: dict[str, tuple] = {}
+        for i, fid in enumerate(fetched.get('ids', [])):
+            meta = (fetched.get('metadatas') or [None] * len(fetched['ids']))[i]
+            doc = (fetched.get('documents') or [None] * len(fetched['ids']))[i]
+            fetched_map[fid] = (meta, doc)
+
+        ordered_ids = []
+        ordered_metas = []
+        ordered_docs = []
+        ordered_dists = []
+
+        # Normalize RRF scores to 0~1 distances via min-max scaling.
+        # Raw RRF scores are tiny (e.g. 0.016~0.033 with k=60) so a naive
+        # ``1 - score`` would compress all distances into a narrow 0.96~0.98
+        # band with almost no discriminative power.  Min-max normalization
+        # spreads them across the full 0~1 range (0.0 = best match).
+        max_score = fused[0][1]
+        min_score = fused[-1][1]
+        score_range = max_score - min_score
+
+        for doc_id, score in fused:
+            if doc_id in fetched_map:
+                meta, doc = fetched_map[doc_id]
+                ordered_ids.append(doc_id)
+                ordered_metas.append(meta)
+                ordered_docs.append(doc)
+                if score_range > 0:
+                    ordered_dists.append(1.0 - (score - min_score) / score_range)
+                else:
+                    ordered_dists.append(0.0)
+
+        self.ap.logger.info(
+            f"Chroma hybrid search in '{collection}' returned {len(ordered_ids)} results "
+            f'(vector={len(vector_ids)}, text={len(text_ids)}).'
+        )
+        return {
+            'ids': [ordered_ids],
+            'metadatas': [ordered_metas],
+            'distances': [ordered_dists],
+            'documents': [ordered_docs],
+        }
+
+    @staticmethod
+    def _rrf_fuse(result_lists: list[list[str]], k: int) -> list[tuple[str, float]]:
+        """Reciprocal Rank Fusion over multiple ranked ID lists.
+
+        Returns a list of (doc_id, rrf_score) sorted by descending score,
+        truncated to *k* entries.
+        """
+        scores: dict[str, float] = {}
+        for ranked_ids in result_lists:
+            for rank, doc_id in enumerate(ranked_ids):
+                scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (_RRF_K + rank + 1)
+        sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
+        return sorted_results[:k]
+
     async def delete_by_file_id(self, collection: str, file_id: str) -> None:
         col = await self.get_or_create_collection(collection)
         await asyncio.to_thread(col.delete, where={'file_id': file_id})

src/langbot/templates/default-pipeline-config.json

@@ -95,11 +95,12 @@
             "max": 0
         },
         "misc": {
-            "hide-exception": true,
+            "exception-handling": "show-hint",
+            "failure-hint": "Request failed.",
             "at-sender": true,
             "quote-origin": true,
             "track-function-calls": false,
             "remove-think": false
         }
     }
-}
+}

src/langbot/templates/metadata/pipeline/output.yaml

@@ -78,13 +78,39 @@ stages:
       en_US: Misc
       zh_Hans: 杂项
     config:
-      - name: hide-exception
+      - name: exception-handling
         label:
-          en_US: Hide Exception
+          en_US: Exception Handling Strategy
-          zh_Hans: 不输出异常信息给用户
+          zh_Hans: 异常处理策略
-        type: boolean
+        description:
+          en_US: Controls how error messages are displayed to the user when an AI request fails
+          zh_Hans: 控制 AI 请求失败时向用户展示错误信息的方式
+        type: select
         required: true
-        default: true
+        default: show-hint
+        options:
+          - name: show-error
+            label:
+              en_US: Show Full Error
+              zh_Hans: 显示完整报错信息
+          - name: show-hint
+            label:
+              en_US: Show Failure Hint
+              zh_Hans: 仅文字提示
+          - name: hide
+            label:
+              en_US: Hide All
+              zh_Hans: 不显示任何异常信息
+      - name: failure-hint
+        label:
+          en_US: Failure Hint Text
+          zh_Hans: 失败提示文本
+        description:
+          en_US: The text to display when a request fails. Only effective when Exception Handling Strategy is set to "Show Failure Hint"
+          zh_Hans: 请求失败时显示的提示文本，仅在异常处理策略设置为"仅文字提示"时生效
+        type: string
+        required: false
+        default: 'Request failed.'
       - name: at-sender
         label:
           en_US: At Sender
@@ -119,3 +145,4 @@ stages:
         type: boolean
         required: true
         default: false
+

web/src/app/home/bots/components/bot-session/BotSessionMonitor.tsx

@@ -6,6 +6,7 @@ import { httpClient } from '@/app/infra/http/HttpClient';
 import { ScrollArea } from '@/components/ui/scroll-area';
 import { Button } from '@/components/ui/button';
 import { cn } from '@/lib/utils';
+import { Copy, Check } from 'lucide-react';
 import {
   MessageChainComponent,
   Plain,
@@ -27,6 +28,7 @@ interface SessionInfo {
   is_active: boolean;
   platform?: string | null;
   user_id?: string | null;
+  user_name?: string | null;
 }
 
 interface SessionMessage {
@@ -60,8 +62,29 @@ export default function BotSessionMonitor({ botId }: BotSessionMonitorProps) {
   const [messages, setMessages] = useState<SessionMessage[]>([]);
   const [loadingSessions, setLoadingSessions] = useState(false);
   const [loadingMessages, setLoadingMessages] = useState(false);
+  const [copiedUserId, setCopiedUserId] = useState(false);
   const messagesContainerRef = useRef<HTMLDivElement>(null);
 
+  const parseSessionType = (sessionId: string): string | null => {
+    const idx = sessionId.indexOf('_');
+    if (idx === -1) return null;
+    const type = sessionId.slice(0, idx);
+    if (type === 'person' || type === 'group') return type;
+    return null;
+  };
+
+  const abbreviateId = (id: string): string => {
+    if (id.length <= 10) return id;
+    return `${id.slice(0, 4)}..${id.slice(-4)}`;
+  };
+
+  const copyUserId = (userId: string) => {
+    navigator.clipboard.writeText(userId).then(() => {
+      setCopiedUserId(true);
+      setTimeout(() => setCopiedUserId(false), 2000);
+    });
+  };
+
   const loadSessions = useCallback(async () => {
     setLoadingSessions(true);
     try {
@@ -338,24 +361,36 @@ export default function BotSessionMonitor({ botId }: BotSessionMonitorProps) {
                   >
                     <div className="flex items-center justify-between mb-0.5">
                       <span className="text-sm font-medium truncate mr-2">
-                        {session.user_id || session.session_id.slice(0, 12)}
+                        {session.user_name ||
+                          session.user_id ||
+                          session.session_id.slice(0, 12)}
                       </span>
                       <span className="text-[11px] text-muted-foreground tabular-nums flex-shrink-0">
                         {formatRelativeTime(session.last_activity)}
                       </span>
                     </div>
                     <div className="flex items-center gap-1.5 text-xs text-muted-foreground">
+                      {parseSessionType(session.session_id) && (
+                        <span className="px-1 py-0.5 rounded bg-muted text-[10px]">
+                          {parseSessionType(session.session_id)}
+                        </span>
+                      )}
                       {session.platform && (
                         <span className="px-1 py-0.5 rounded bg-muted text-[10px]">
                           {session.platform}
                         </span>
                       )}
+                      {session.user_id && (
+                        <span className="truncate text-[10px]">
+                          {abbreviateId(session.user_id)}
+                        </span>
+                      )}
                       {session.is_active && (
                         <span className="flex items-center gap-0.5 text-green-600 dark:text-green-400">
                           <span className="w-1.5 h-1.5 rounded-full bg-green-500 inline-block" />
                         </span>
                       )}
-                      <span>{session.pipeline_name}</span>
+                      <span className="truncate">{session.pipeline_name}</span>
                     </div>
                   </button>
                 );
@@ -377,15 +412,42 @@ export default function BotSessionMonitor({ botId }: BotSessionMonitorProps) {
             <div className="px-6 py-3 border-b shrink-0 flex items-center justify-between">
               <div className="min-w-0">
                 <div className="text-sm font-medium truncate">
-                  {selectedSession?.user_id || selectedSessionId.slice(0, 20)}
+                  {selectedSession?.user_name ||
+                    selectedSession?.user_id ||
+                    selectedSessionId.slice(0, 20)}
                 </div>
                 <div className="flex items-center gap-2 text-xs text-muted-foreground">
+                  {parseSessionType(selectedSessionId) && (
+                    <span>{parseSessionType(selectedSessionId)}</span>
+                  )}
                   {selectedSession?.platform && (
-                    <span>{selectedSession.platform}</span>
+                    <>
+                      {parseSessionType(selectedSessionId) && <span>·</span>}
+                      <span>{selectedSession.platform}</span>
+                    </>
+                  )}
+                  {selectedSession?.user_id && (
+                    <>
+                      <span>·</span>
+                      <span className="font-mono">
+                        {selectedSession.user_id}
+                      </span>
+                      <button
+                        onClick={() => copyUserId(selectedSession.user_id!)}
+                        className="inline-flex items-center text-muted-foreground hover:text-foreground transition-colors"
+                        title={t('common.copy')}
+                      >
+                        {copiedUserId ? (
+                          <Check className="w-3 h-3 text-green-600" />
+                        ) : (
+                          <Copy className="w-3 h-3" />
+                        )}
+                      </button>
+                    </>
                   )}
                   {selectedSession?.pipeline_name && (
                     <>
-                      {selectedSession?.platform && <span>·</span>}
+                      <span>·</span>
                       <span>{selectedSession.pipeline_name}</span>
                     </>
                   )}

web/src/app/infra/http/BackendClient.ts

@@ -356,6 +356,7 @@ export class BackendClient extends BaseHttpClient {
       is_active: boolean;
       platform: string | null;
       user_id: string | null;
+      user_name: string | null;
     }>;
     total: number;
   }> {
@@ -384,6 +385,7 @@ export class BackendClient extends BaseHttpClient {
       level: string;
       platform: string | null;
       user_id: string | null;
+      user_name: string | null;
       runner_name: string | null;
       variables: string | null;
       role: string | null;

web/src/app/login/page.tsx

@@ -284,6 +284,27 @@ export default function Login() {
               </form>
             </Form>
           )}
+
+          <p className="text-xs text-center text-muted-foreground">
+            {t('common.agreementNotice')}{' '}
+            <a
+              href="https://langbot.app/privacy"
+              target="_blank"
+              rel="noopener noreferrer"
+              className="underline hover:text-foreground transition-colors"
+            >
+              {t('common.privacyPolicy')}
+            </a>{' '}
+            {t('common.and')}{' '}
+            <a
+              href={t('common.dataCollectionPolicyUrl')}
+              target="_blank"
+              rel="noopener noreferrer"
+              className="underline hover:text-foreground transition-colors"
+            >
+              {t('common.dataCollectionPolicy')}
+            </a>
+          </p>
         </CardContent>
       </Card>
     </div>

web/src/app/register/page.tsx

@@ -253,6 +253,27 @@ export default function Register() {
               </Button>
             </form>
           </Form>
+
+          <p className="text-xs text-center text-muted-foreground">
+            {t('common.agreementNotice')}{' '}
+            <a
+              href="https://langbot.app/privacy"
+              target="_blank"
+              rel="noopener noreferrer"
+              className="underline hover:text-foreground transition-colors"
+            >
+              {t('common.privacyPolicy')}
+            </a>{' '}
+            {t('common.and')}{' '}
+            <a
+              href={t('common.dataCollectionPolicyUrl')}
+              target="_blank"
+              rel="noopener noreferrer"
+              className="underline hover:text-foreground transition-colors"
+            >
+              {t('common.dataCollectionPolicy')}
+            </a>
+          </p>
         </CardContent>
       </Card>
     </div>

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

@@ -47,6 +47,12 @@ const enUS = {
     copyFailed: 'Copy Failed',
     test: 'Test',
     forgotPassword: 'Forgot Password?',
+    agreementNotice: 'By continuing, you agree to our',
+    privacyPolicy: 'Privacy Policy',
+    and: 'and',
+    dataCollectionPolicy: 'Data Collection Policy',
+    dataCollectionPolicyUrl:
+      'https://docs.langbot.app/en/insight/data-collection-policy',
     loading: 'Loading...',
     fieldRequired: 'This field is required',
     or: 'or',

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

@@ -48,6 +48,12 @@ const jaJP = {
     copyFailed: 'コピーに失敗しました',
     test: 'テスト',
     forgotPassword: 'パスワードを忘れた？',
+    agreementNotice: '続行することで、以下に同意したものとみなされます：',
+    privacyPolicy: 'プライバシーポリシー',
+    and: 'および',
+    dataCollectionPolicy: 'データ収集ポリシー',
+    dataCollectionPolicyUrl:
+      'https://docs.langbot.app/ja/insight/data-collection-policy',
     loading: '読み込み中...',
     fieldRequired: 'この項目は必須です',
     or: 'または',

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

@@ -47,6 +47,12 @@ const zhHans = {
     copyFailed: '复制失败',
     test: '测试',
     forgotPassword: '忘记密码？',
+    agreementNotice: '继续即表示您同意我们的',
+    privacyPolicy: '隐私政策',
+    and: '和',
+    dataCollectionPolicy: '数据收集政策',
+    dataCollectionPolicyUrl:
+      'https://docs.langbot.app/zh/insight/data-collection-policy',
     loading: '加载中...',
     fieldRequired: '此字段为必填项',
     or: '或',

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

@@ -47,6 +47,12 @@ const zhHant = {
     copyFailed: '複製失敗',
     test: '測試',
     forgotPassword: '忘記密碼？',
+    agreementNotice: '繼續即表示您同意我們的',
+    privacyPolicy: '隱私政策',
+    and: '和',
+    dataCollectionPolicy: '數據收集政策',
+    dataCollectionPolicyUrl:
+      'https://docs.langbot.app/zh/insight/data-collection-policy',
     loading: '載入中...',
     fieldRequired: '此欄位為必填',
     or: '或',