diff --git a/src/langbot/libs/wecom_ai_bot_api/api.py b/src/langbot/libs/wecom_ai_bot_api/api.py index 44e4c2b1..bedae67f 100644 --- a/src/langbot/libs/wecom_ai_bot_api/api.py +++ b/src/langbot/libs/wecom_ai_bot_api/api.py @@ -228,6 +228,9 @@ class StreamSessionManager: msg_id = session.msg_id if msg_id and self._msg_index.get(msg_id) == stream_id: self._msg_index.pop(msg_id, None) + # Clean up feedback index for expired sessions + if session.feedback_id: + self._feedback_index.pop(session.feedback_id, None) def _decrypt_file(encrypted_data: bytes, aes_key_str: str) -> bytes: @@ -903,35 +906,38 @@ class WecomBotClient: ) session = self.stream_sessions.get_session_by_feedback_id(feedback_id) + if session: await self.logger.info( f'反馈关联到会话: stream_id={session.stream_id}, msg_id={session.msg_id}, user_id={session.user_id}' ) - for handler in self._message_handlers.get('feedback', []): - try: - await handler( - feedback_id=feedback_id, - feedback_type=feedback_type, - feedback_content=feedback_content, - inaccurate_reasons=inaccurate_reasons, - session=session, - ) - except Exception: - await self.logger.error(traceback.format_exc()) - - if self._feedback_callback: - try: - await self._feedback_callback( - feedback_id=feedback_id, - feedback_type=feedback_type, - feedback_content=feedback_content, - inaccurate_reasons=inaccurate_reasons, - session=session, - ) - except Exception: - await self.logger.error(traceback.format_exc()) else: - await self.logger.warning(f'未找到 feedback_id={feedback_id} 对应的会话') + await self.logger.warning(f'未找到 feedback_id={feedback_id} 对应的会话，仍将记录反馈') + + # Dispatch feedback event regardless of session availability + for handler in self._message_handlers.get('feedback', []): + try: + await handler( + feedback_id=feedback_id, + feedback_type=feedback_type, + feedback_content=feedback_content, + inaccurate_reasons=inaccurate_reasons, + session=session, + ) + except Exception: + await self.logger.error(traceback.format_exc()) + + if self._feedback_callback: + try: + await self._feedback_callback( + feedback_id=feedback_id, + feedback_type=feedback_type, + feedback_content=feedback_content, + inaccurate_reasons=inaccurate_reasons, + session=session, + ) + except Exception: + await self.logger.error(traceback.format_exc()) except Exception: await self.logger.error(traceback.format_exc()) diff --git a/src/langbot/pkg/api/http/service/monitoring.py b/src/langbot/pkg/api/http/service/monitoring.py index 7a1cad23..e1c60fec 100644 --- a/src/langbot/pkg/api/http/service/monitoring.py +++ b/src/langbot/pkg/api/http/service/monitoring.py @@ -1224,30 +1224,83 @@ class MonitoringService: """ import json - record_id = str(uuid.uuid4()) - record_data = { - 'id': record_id, - 'timestamp': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None), - 'feedback_id': feedback_id, - 'feedback_type': feedback_type, - 'feedback_content': feedback_content, - 'inaccurate_reasons': json.dumps(inaccurate_reasons, ensure_ascii=False) if inaccurate_reasons else None, - 'bot_id': bot_id, - 'bot_name': bot_name, - 'pipeline_id': pipeline_id, - 'pipeline_name': pipeline_name, - 'session_id': session_id, - 'message_id': message_id, - 'stream_id': stream_id, - 'user_id': user_id, - 'platform': platform, - } + now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None) + reasons_json = json.dumps(inaccurate_reasons, ensure_ascii=False) if inaccurate_reasons else None - await self.ap.persistence_mgr.execute_async( - sqlalchemy.insert(persistence_monitoring.MonitoringFeedback).values(record_data) + MonitoringFeedback = persistence_monitoring.MonitoringFeedback + + # Handle cancel feedback (type=3): delete existing record + if feedback_type == 3: + await self.ap.persistence_mgr.execute_async( + sqlalchemy.delete(MonitoringFeedback).where(MonitoringFeedback.feedback_id == feedback_id) + ) + return None + + # Check if record with this feedback_id already exists + existing_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(MonitoringFeedback).where(MonitoringFeedback.feedback_id == feedback_id) ) + existing_row = existing_result.first() - return record_id + if existing_row: + # UPDATE existing record + existing = existing_row[0] if isinstance(existing_row, tuple) else existing_row + await self.ap.persistence_mgr.execute_async( + sqlalchemy.update(MonitoringFeedback) + .where(MonitoringFeedback.feedback_id == feedback_id) + .values( + timestamp=now, + feedback_type=feedback_type, + feedback_content=feedback_content, + inaccurate_reasons=reasons_json, + bot_id=bot_id or existing.bot_id, + bot_name=bot_name or existing.bot_name, + pipeline_id=pipeline_id or existing.pipeline_id, + pipeline_name=pipeline_name or existing.pipeline_name, + session_id=session_id or existing.session_id, + message_id=message_id or existing.message_id, + stream_id=stream_id or existing.stream_id, + user_id=user_id or existing.user_id, + platform=platform or existing.platform, + ) + ) + return existing.id + else: + # INSERT new record with IntegrityError defense + record_id = str(uuid.uuid4()) + record_data = { + 'id': record_id, + 'timestamp': now, + 'feedback_id': feedback_id, + 'feedback_type': feedback_type, + 'feedback_content': feedback_content, + 'inaccurate_reasons': reasons_json, + 'bot_id': bot_id, + 'bot_name': bot_name, + 'pipeline_id': pipeline_id, + 'pipeline_name': pipeline_name, + 'session_id': session_id, + 'message_id': message_id, + 'stream_id': stream_id, + 'user_id': user_id, + 'platform': platform, + } + try: + await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(MonitoringFeedback).values(record_data)) + return record_id + except Exception: + # UNIQUE constraint conflict (concurrent feedback for same feedback_id) + await self.ap.persistence_mgr.execute_async( + sqlalchemy.update(MonitoringFeedback) + .where(MonitoringFeedback.feedback_id == feedback_id) + .values( + timestamp=now, + feedback_type=feedback_type, + feedback_content=feedback_content, + inaccurate_reasons=reasons_json, + ) + ) + return feedback_id async def get_feedback_stats( self, diff --git a/src/langbot/pkg/platform/sources/wecombot.py b/src/langbot/pkg/platform/sources/wecombot.py index d102f1ce..01af2929 100644 --- a/src/langbot/pkg/platform/sources/wecombot.py +++ b/src/langbot/pkg/platform/sources/wecombot.py @@ -328,6 +328,9 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): feedback_type = kwargs.get('feedback_type', 0) feedback_content = kwargs.get('feedback_content', '') or None inaccurate_reasons = kwargs.get('inaccurate_reasons', []) or None + # WeChat Work returns integer reason codes, but FeedbackEvent expects strings + if inaccurate_reasons: + inaccurate_reasons = [str(r) for r in inaccurate_reasons] session = kwargs.get('session') session_id = None diff --git a/web/src/app/home/monitoring/hooks/useFeedbackData.ts b/web/src/app/home/monitoring/hooks/useFeedbackData.ts index 17c17ceb..9948bee8 100644 --- a/web/src/app/home/monitoring/hooks/useFeedbackData.ts +++ b/web/src/app/home/monitoring/hooks/useFeedbackData.ts @@ -1,6 +1,7 @@ import { useState, useEffect, useCallback, useMemo } from 'react'; import { httpClient } from '@/app/infra/http'; import { FeedbackRecord, FeedbackStats } from '../types/monitoring'; +import { parseUTCTimestamp } from '../utils/dateUtils'; interface UseFeedbackDataParams { botIds?: string[]; @@ -142,7 +143,7 @@ export function useFeedbackData(params: UseFeedbackDataParams = {}) { const transformedFeedback: FeedbackRecord[] = result.feedback.map( (item) => ({ id: item.id, - timestamp: new Date(item.timestamp), + timestamp: parseUTCTimestamp(item.timestamp), feedbackId: item.feedback_id, feedbackType: item.feedback_type === 1 ? 'like' : 'dislike', feedbackContent: item.feedback_content,