""" Supabase 日志使用示例 演示如何在项目中使用 Supabase 日志记录器 """ from services.supabase_logger import ( get_supabase_logger, log_api_access, log_model_performance, LogPerformance, log_debug, log_error ) from functools import wraps import time # ============================================ # 示例 1: 基本日志记录 # ============================================ def example_basic_logging(): """基本日志记录示例""" logger = get_supabase_logger() # 记录 API 访问 logger.log_api_access( endpoint="/api/v1/chat", method="POST", session_id="user_123", response_time_ms=150, status_code=200 ) # 记录调试信息 logger.log_debug( level="INFO", logger="intelligence_analyzer", message="Scenario detected: policy_inquiry", session_id="user_123", data={ "scenario": "policy_inquiry", "confidence": 0.95, "keywords": ["年假", "多少"] } ) # 记录错误 logger.log_error( error_type="model_load", error_message="Failed to load sentiment model", error_stack="Traceback...\n File ...", endpoint="/api/v1/evaluate" ) # ============================================ # 示例 2: 使用装饰器 # ============================================ @log_api_access("/api/v1/chat") def process_chat_request(question: str, session_id: str): """ 使用 API 日志装饰器 自动记录 API 调用和响应时间 """ # 模拟处理 time.sleep(0.1) return {"answer": "这是回复内容"} @log_model_performance("sbert-hr-v2", "sbert") def encode_with_sbert(text: str): """ 使用模型性能装饰器 自动记录模型推理性能 """ # 模拟编码 time.sleep(0.05) return [0.1, 0.2, 0.3] # ============================================ # 示例 3: 使用上下文管理器 # ============================================ def example_context_manager(): """使用上下文管理器追踪性能""" with LogPerformance("chat_processing", "deepseek-chat", "llm"): # 执行耗时操作 result = call_llm_api("年假有多少天？") return result def call_llm_api(prompt: str): """模拟 LLM API 调用""" time.sleep(0.2) return "根据规定，年假天数与工龄挂钩..." # ============================================ # 示例 4: 在 intelligence_analyzer 中集成 # ============================================ def analyze_with_logging(employee_input: str, session_id: str): """ 在分析器中集成日志记录 """ logger = get_supabase_logger() # 记录输入 logger.log_debug( level="INFO", logger="intelligence_analyzer", message=f"Processing input: {employee_input[:50]}...", session_id=session_id ) try: # 场景识别 start_time = time.time() scenario = detect_scenario(employee_input) logger.log_debug( level="INFO", logger="intelligence_analyzer", message="Scenario detected", session_id=session_id, data={"scenario": scenario, "confidence": 0.9} ) # 记录用户行为 logger.log_user_behavior( session_id=session_id, action_type="chat", action_detail="scenario_detection", input_text=employee_input, scenario_detected=scenario, success=True ) return scenario except Exception as e: # 记录错误 logger.log_error( error_type="analysis_error", error_message=str(e), session_id=session_id, user_input=employee_input ) raise def detect_scenario(text: str) -> str: """模拟场景检测""" time.sleep(0.02) return "policy_inquiry" # ============================================ # 示例 5: 在 FastAPI 路由中集成 # ============================================ from fastapi import Request async def log_request_middleware(request: Request, call_next): """ FastAPI 中间件 - 记录所有请求 """ logger = get_supabase_logger() start_time = time.time() # 处理请求 response = await call_next(request) # 计算响应时间 process_time = int((time.time() - start_time) * 1000) # 记录日志 logger.log_api_access( endpoint=request.url.path, method=request.method, client_ip=request.client.host if request.client else None, user_agent=request.headers.get("user-agent"), session_id=request.headers.get("x-session-id"), response_time_ms=process_time, status_code=response.status_code ) # 添加响应头 response.headers["X-Process-Time"] = str(process_time) return response # ============================================ # 示例 6: 批量日志记录 # ============================================ def batch_log_examples(): """批量日志记录示例""" logger = get_supabase_logger() # 批量记录用户行为 behaviors = [ { "session_id": "user_1", "action_type": "chat", "scenario_detected": "policy_inquiry", "success": True }, { "session_id": "user_1", "action_type": "chat", "scenario_detected": "leave_application", "success": True }, { "session_id": "user_2", "action_type": "evaluate", "success": False, "error_message": "Invalid dialogue format" } ] for behavior in behaviors: logger.log_user_behavior(**behavior) # ============================================ # 示例 7: 错误追踪 # ============================================ def example_with_error_tracking(): """带错误追踪的示例""" logger = get_supabase_logger() try: # 模拟错误 result = 1 / 0 except Exception as e: import traceback # 记录详细错误信息 logger.log_error( error_type="calculation_error", error_message=str(e), error_stack=traceback.format_exc(), session_id="debug_session", request_data={"operation": "division", "operands": [1, 0]} ) # ============================================ # 示例 8: 模型性能追踪 # ============================================ def track_all_models(): """追踪所有模型的性能""" logger = get_supabase_logger() models = { "sbert": "KarenYYH/sbert-hr-v2", "sentiment": "KarenYYH/sentiment-hr", "llm": "deepseek-chat" } for model_type, model_name in models.items(): with LogPerformance(f"load_{model_type}", model_name, model_type): # 模拟模型加载 time.sleep(0.1) # ============================================ # 简化的日志记录 # ============================================ def example_simplified_logging(): """使用简化函数记录日志""" # 调试日志 log_debug("intelligence_analyzer", "处理用户输入", level="INFO", session_id="user_123", data={"input_length": 20}) # 错误日志 log_error("api_error", "请求超时", endpoint="/api/v1/chat", session_id="user_123", response_time_ms=5000) # ============================================ # 运行所有示例 # ============================================ if __name__ == "__main__": print("运行 Supabase 日志示例...") example_basic_logging() print("✓ 基本日志记录") process_chat_request("年假有多少天？", "user_123") print("✓ API 日志装饰器") encode_with_sbert("测试文本") print("✓ 模型性能装饰器") example_context_manager() print("✓ 上下文管理器") analyze_with_logging("年假有多少天？", "user_456") print("✓ 分析器集成") batch_log_examples() print("✓ 批量日志") example_with_error_tracking() print("✓ 错误追踪") track_all_models() print("✓ 模型性能追踪") example_simplified_logging() print("✓ 简化日志") print("\n所有示例运行完成！")