WiseAgent智能体框架实战之CrewAI篇(五）— 构建完整的医院智能体导诊系统与异常处理

引言

大家好，我是Wise玩转 AI。在前三期的实践中，我们逐步构建了一个具备查询、预约、症状分析、记忆功能的医院导诊智能体。今天，我们将进入第4期的工程实践，完成整个导诊系统的整合，并重点关注异常处理和系统鲁棒性。

正如我们在理论文章中强调的，异常处理是智能体工程化落地的关键。一个没有完善异常处理机制的智能体，在实际使用中很容易 "崩溃" 或 "迷路"，无法提供可靠的服务。在企业级应用中，稳定性和可靠性往往比功能丰富性更为重要。

在本期中，我们将整合前三期的所有功能，实现一个完整的医院导诊系统，并为其添加全面的异常处理机制、用户反馈功能和系统监控能力，使我们的智能体真正具备生产环境部署的条件。

本期新增功能概述

1. 系统整合：将前几期实现的功能有机整合，形成完整的导诊流程

2. 异常处理：实现全面的错误检测和恢复机制，处理各种可能的异常情况

3. 用户反馈：添加用户反馈收集功能，用于持续改进系统

4. 系统监控：实现基本的系统监控和日志记录，便于问题排查和性能优化

5. 安全控制：添加基本的安全检查，防止恶意使用

这些功能将使我们的导诊智能体从一个功能原型进化为一个可部署、可维护、可靠的企业级应用。

最终项目结构

在前三期的基础上，我们的最终项目结构如下：

hospital_agent/ ├── agents/ │ ├── base_agent.py │ ├── advanced_agents.py │ └── memory_agents.py ├── tasks/ │ ├── base_tasks.py │ ├── advanced_tasks.py │ └── memory_tasks.py ├── tools/ │ ├── hospital_tools.py │ ├── appointment_tools.py │ ├── knowledge_tools.py │ └── feedback_tools.py # 新增：用户反馈工具 ├── data/ │ ├── hospital_data.py │ ├── appointment_data.py │ ├── medical_knowledge.py │ ├── long_term_memory.json │ └── feedbacks.json # 新增：用户反馈数据 ├── memory/ │ ├── short_term_memory.py │ └── long_term_memory.py ├── planner/ │ └── task_planner.py ├── exception/ # 新增：异常处理 │ ├── exceptions.py │ └── handler.py ├── monitor/ # 新增：监控与日志 │ └── monitor.py ├── security/ # 新增：安全控制 │ └── security_checker.py ├── main.py └── .env

新增的模块主要关注异常处理、监控、安全和用户反馈。

实现异常处理机制

首先，我们定义系统可能遇到的异常和相应的处理机制。

创建exception/exceptions.py：

class HospitalAgentException(Exception): """医院智能体系统基础异常类""" def __init__(self, message, error_code=None): super().__init__(message) self.message = message self.error_code = error_code class ConfigurationError(HospitalAgentException): """配置错误异常""" def __init__(self, message): super().__init__(message, "CONFIG_ERROR") class DatabaseConnectionError(HospitalAgentException): """数据库连接异常""" def __init__(self, message): super().__init__(message, "DB_CONNECTION_ERROR") class ExternalServiceError(HospitalAgentException): """外部服务调用异常""" def __init__(self, message): super().__init__(message, "EXTERNAL_SERVICE_ERROR") class SecurityError(HospitalAgentException): """安全相关异常""" def __init__(self, message): super().__init__(message, "SECURITY_ERROR") class BusinessLogicError(HospitalAgentException): """业务逻辑异常""" def __init__(self, message): super().__init__(message, "BUSINESS_LOGIC_ERROR")

创建exception/handler.py：

import functools import logging from phase4_enterprise_guide.exception.exceptions import HospitalAgentException, ConfigurationError, DatabaseConnectionError # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class ExceptionHandler: """异常处理装饰器类""" @staticmethod def handle_exceptions(func): """装饰器：处理函数异常""" @functools.wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except HospitalAgentException as e: logger.error(f"业务异常: {e.message}, 错误码: {e.error_code}") # 可以在这里添加特定的处理逻辑 raise except Exception as e: logger.error(f"未预期的系统异常: {str(e)}") # 转换为业务异常 raise HospitalAgentException(f"系统内部错误: {str(e)}") return wrapper @staticmethod def handle_tool_exceptions(tool_name): """装饰器：处理工具调用异常""" def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except ConfigurationError as e: logger.error(f"工具[{tool_name}]配置错误: {e.message}") raise except DatabaseConnectionError as e: logger.error(f"工具[{tool_name}]数据库连接失败: {e.message}") raise except Exception as e: logger.error(f"工具[{tool_name}]执行异常: {str(e)}") raise HospitalAgentException(f"工具[{tool_name}]执行失败: {str(e)}") return wrapper return decorator

这个异常处理机制定义了系统可能遇到的各种异常，并提供了友好的错误提示，符合我们之前提到的 "异常处理是智能体稳定性保障" 的原则。

实现用户反馈工具

创建tools/feedback_tools.py：

import json import os from datetime import datetime from crewai.tools import BaseTool from phase4_enterprise_guide.exception.handler import ExceptionHandler class FeedbackTool(BaseTool): """用户反馈收集工具""" name: str = "用户反馈收集工具" description: str = "收集用户对智能体服务的评分和评论反馈，并存储到JSON文件中" feedback_file: str = "./data/user_feedback.json" @ExceptionHandler.handle_tool_exceptions("FeedbackTool") def _run(self, rating: int, comment: str = "", user_id: str = "") -> str: """ 收集用户反馈 参数: rating: 评分 (1-5星) comment: 用户评论 (可选) user_id: 用户ID (可选) 返回: 反馈收集结果消息 """ # 验证评分范围 if not 1 <= rating <= 5: raise ValueError("评分必须在1-5之间") # 创建反馈数据 feedback_data = { "timestamp": datetime.now().isoformat(), "user_id": user_id or "anonymous", "rating": rating, "comment": comment } # 确保数据目录存在 os.makedirs(os.path.dirname(self.feedback_file), exist_ok=True) # 读取现有反馈数据 feedbacks = [] if os.path.exists(self.feedback_file): with open(self.feedback_file, 'r', encoding='utf-8') as f: try: feedbacks = json.load(f) except json.JSONDecodeError: feedbacks = [] # 添加新反馈 feedbacks.append(feedback_data) # 保存反馈数据 with open(self.feedback_file, 'w', encoding='utf-8') as f: json.dump(feedbacks, f, ensure_ascii=False, indent=2) return f"感谢您的反馈！您给了{rating}星评价。"

这个工具允许用户提交对服务的评分和评论，有助于持续改进系统。

实现系统监控和日志

创建monitor/monitor.py：

import logging import time from datetime import datetime from functools import wraps # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('./logs/system_monitor.log'), logging.StreamHandler() ] ) logger = logging.getLogger("AgentMonitor") class AgentMonitor: """系统监控和日志记录类""" @staticmethod def log_request_response(agent_name, request, response, execution_time): """ 记录请求和响应日志 参数: agent_name: 智能体名称 request: 请求内容 response: 响应内容 execution_time: 执行时间(秒) """ logger.info(f"智能体: {agent_name}") logger.info(f"请求: {request}") logger.info(f"响应: {response}") logger.info(f"执行时间: {execution_time:.4f}秒") logger.info("-" * 50) @staticmethod def log_performance(metric_name, value, unit=""): """ 记录性能指标 参数: metric_name: 指标名称 value: 指标值 unit: 单位 """ logger.info(f"性能指标 - {metric_name}: {value}{unit}") @staticmethod def monitor_execution(agent_name): """ 装饰器：监控函数执行时间和异常 参数: agent_name: 智能体名称 """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): start_time = time.time() try: result = func(*args, **kwargs) execution_time = time.time() - start_time # 记录执行成功的日志 logger.info(f"{agent_name}执行成功，耗时: {execution_time:.4f}秒") return result except Exception as e: execution_time = time.time() - start_time # 记录异常日志 logger.error(f"{agent_name}执行失败，耗时: {execution_time:.4f}秒，错误: {str(e)}") raise return wrapper return decorator @staticmethod def get_system_stats(): """ 获取系统统计信息 返回: 包含统计信息的字典 """ # 这里可以实现具体的系统统计逻辑 # 例如：CPU使用率、内存使用情况等 stats = { "timestamp": datetime.now().isoformat(), "status": "running", "uptime": "N/A", # 需要实现具体的运行时间计算 } return stats

这个监控器负责记录系统的运行日志和性能指标，便于后续的问题排查和系统优化。

实现安全控制

创建security/security_checker.py：

import re import logging from phase4_enterprise_guide.exception.exceptions import SecurityError # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class SecurityChecker: """安全检查类""" # 敏感操作关键词列表 SENSITIVE_OPERATIONS = [ "delete", "drop", "truncate", "shutdown", "restart", "kill", "rm -rf", "format" ] # SQL注入攻击模式 SQL_INJECTION_PATTERNS = [ r"(\%27)|(\')|(\-\-)", # 单引号和注释符 r"(\%22)|(\")", # 双引号 r"(\%3B)|(;)", # 分号 r"(\%2D)|(\-\-)", # 双连字符 r"(\%23)|(#)", # 井号 r"(union\s+select)", # UNION SELECT r"(exec|execute)", # 执行命令 r"(cmd|command)" # 命令执行 ] @classmethod def check_sensitive_operation(cls, operation): """ 检查敏感操作 参数: operation: 操作字符串 异常: SecurityError: 当检测到敏感操作时抛出 """ operation_lower = operation.lower() for sensitive_op in cls.SENSITIVE_OPERATIONS: if sensitive_op in operation_lower: logger.warning(f"检测到敏感操作: {operation}") raise SecurityError(f"操作被拒绝：检测到敏感操作 '{sensitive_op}'") @classmethod def check_sql_injection(cls, query): """ 检查SQL注入攻击 参数: query: 查询字符串 异常: SecurityError: 当检测到SQL注入攻击时抛出 """ query_lower = query.lower() for pattern in cls.SQL_INJECTION_PATTERNS: if re.search(pattern, query_lower, re.IGNORECASE): logger.warning(f"检测到潜在SQL注入攻击: {query}") raise SecurityError("查询被拒绝：检测到潜在的SQL注入攻击") @classmethod def sanitize_input(cls, user_input): """ 清理用户输入 参数: user_input: 用户输入字符串 返回: 清理后的字符串 """ # 移除潜在危险字符 sanitized = re.sub(r'[<>"\']', '', user_input) # 移除多余的空白字符 sanitized = re.sub(r'\s+', ' ', sanitized).strip() logger.debug(f"输入清理: '{user_input}' -> '{sanitized}'") return sanitized @classmethod def validate_user_id(cls, user_id): """ 验证用户ID格式 参数: user_id: 用户ID 异常: SecurityError: 当用户ID格式不正确时抛出 """ if not user_id or not isinstance(user_id, str): raise SecurityError("无效的用户ID：用户ID不能为空") # 检查用户ID长度 if len(user_id) > 50: raise SecurityError("无效的用户ID：用户ID过长") # 检查用户ID是否包含非法字符 if re.search(r'[^a-zA-Z0-9_\-@.]', user_id): raise SecurityError("无效的用户ID：用户ID包含非法字符")

这个安全检查器能够防止常见的恶意输入和未授权访问，提高系统的安全性。

整合所有功能，优化工具和智能体

更新工具类，添加异常处理：

# 以预约工具为例，其他工具类似更新 class AppointmentTool(BaseTool): # ... 保留之前的代码 ... @ExceptionHandler.handle_tool_exceptions("AppointmentTool") def _run(self, action: str, **kwargs) -> str: # ... 保留之前的代码 ... # 在适当的地方抛出自定义异常 if not doctor_name: raise BusinessLogicError("医生姓名不能为空") if action not in ["check_available", "make_appointment"]: raise BusinessLogicError(f"不支持的操作类型：{action}") if action == "check_available" and not date: raise BusinessLogicError("查询可用时段需要提供日期") if action == "make_appointment": if not all([date, time, patient_name]): raise BusinessLogicError("预约需要提供日期、时间和患者姓名") if doctor_name not in AVAILABLE_SLOTS or date not in AVAILABLE_SLOTS[doctor_name]: raise BusinessLogicError(f"{doctor_name}在{date}的出诊安排") if time not in AVAILABLE_SLOTS[doctor_name][date]: raise BusinessLogicError(f"{doctor_name}在{date}的{time}没有可用时段")

更新智能体，添加安全检查：

# 在enterprise_agents.py中更新 class EnterpriseHospitalAgents: @staticmethod def enterprise_department_agent(): # ... 保留之前的代码 ... # 重写工具调用方法，添加安全检查 def safe_tool_caller(tool, *args, **kwargs): # 检查工具访问权限 SecurityChecker.check_sensitive_operation(tool.name) # 执行工具 return tool.run(*args, **kwargs) agent.tools = [lambda *args, **kwargs: safe_tool_caller(tool, *args, **kwargs) for tool in all_tools] return agent

更新任务规划器，添加性能监控：

# 在EnterpriseHospitalTaskPlanner类中 @staticmethod @AgentMonitor.monitor_execution("EnterpriseHospitalTaskPlanner") def plan_and_execute_with_feedback(user_query, session_id, user_id=None): """带监控和反馈的任务规划与执行""" # 记录开始时间 start_time = time.time() try: # ... 保留之前的代码 ... result = crew.kickoff() # 计算执行时间 execution_time = time.time() - start_time # 记录成功执行 AgentMonitor.log_request_response( "EnterpriseHospitalTaskPlanner", user_query, result, execution_time ) return result except Exception as e: # 计算执行时间 execution_time = time.time() - start_time # 记录异常执行 AgentMonitor.log_request_response( "EnterpriseHospitalTaskPlanner", user_query, f"执行失败: {str(e)}", execution_time ) raise e

最终主程序实现

更新main.py，整合所有功能：

import os import uuid import time from datetime import datetime from dotenv import load_dotenv from phase4_enterprise_guide.planner.enterprise_planner import EnterpriseHospitalTaskPlanner from phase4_enterprise_guide.agents.enterprise_agents import EnterpriseHospitalAgents from phase4_enterprise_guide.tasks.enterprise_tasks import EnterpriseHospitalTasks from phase4_enterprise_guide.feedback.feedback_tools import FeedbackTool from phase4_enterprise_guide.monitor.monitor import AgentMonitor from phase4_enterprise_guide.security.security_checker import SecurityChecker from phase4_enterprise_guide.exception.handler import ExceptionHandler # 加载环境变量 load_dotenv() @ExceptionHandler.handle_exceptions def main(): print("欢迎使用医院智能导诊系统V4.0！") print("这是一个完整的医院导诊智能体，具备科室查询、医生查询、预约挂号、") print("症状分析、医疗咨询等功能，并能记住您的信息和对话历史。") print("使用完毕后，欢迎提供您的反馈，帮助我们改进服务。") print("请输入您的问题，输入'退出'结束服务，输入'反馈'提交使用体验。") # 初始化系统组件 feedback_tool = FeedbackTool() monitor = AgentMonitor() # 生成会话ID session_id = str(uuid.uuid4())[:8] print(f"\n您的会话ID：{session_id}（用于本次会话识别）") # 询问用户ID user_id = input("请输入您的用户ID（如没有可直接回车）：").strip() if not user_id: user_id = "anonymous" print("将使用匿名模式，不会保存您的长期信息") # 初始化智能体和任务 agents = EnterpriseHospitalAgents() tasks = EnterpriseHospitalTasks() while True: user_input = input("\n请输入您的问题：") if user_input == "退出": print("感谢使用，再见！") # 打印会话统计信息 stats = monitor.get_system_stats() print(f"\n会话统计：{stats}") break if user_input == "反馈": # 处理用户反馈 try: rating = int(input("请为本次服务评分（1-5分）：")) comment = input("请输入您的反馈意见：") result = feedback_tool._run(rating, comment, user_id) print(result) except ValueError: print("评分必须是1-5之间的数字，请重新提交反馈。") continue try: # 安全检查 SecurityChecker.check_sensitive_operation(user_input) user_input = SecurityChecker.sanitize_input(user_input) # 处理查询 result = EnterpriseHospitalTaskPlanner.plan_and_execute_with_feedback( user_input, session_id, user_id ) # 输出结果 print("\n导诊结果：") print(result) except Exception as e: print(f"处理您的请求时出现错误：{str(e)}") print("请尝试重新表述您的问题。") if __name__ == "__main__": main()

这个最终版本的主程序整合了所有功能，包括完整的导诊服务、记忆系统、异常处理、用户反馈、系统监控和安全控制，形成了一个真正可部署的企业级智能体应用。

系统测试与验证

运行最终版本的程序，测试各种功能和异常情况：

python main.py

测试场景包括：

1. 正常的科室查询、医生查询

2. 预约挂号流程

3. 症状分析和科室推荐

4. 医疗知识问答

5. 多轮对话，验证记忆功能

6. 输入错误信息，验证异常处理

7. 提交用户反馈

8. 尝试输入敏感内容，验证安全控制

通过这些测试，我们可以验证系统的功能完整性和鲁棒性。

第 4 期实现的核心要点总结

1. 系统整合：将前几期的功能有机整合，形成完整的导诊流程

2. 异常处理：实现了全面的异常定义和处理机制，提高了系统的稳定性

3. 用户反馈：添加了用户反馈收集功能，为系统改进提供依据

4. 系统监控：实现了日志记录和性能统计，便于系统维护和优化

5. 安全控制：添加了输入安全检查和权限控制，提高了系统的安全性

这些实现使我们的智能体系统具备了企业级应用所需的完整性、稳定性和安全性，真正实现了从原型到产品的跨越。

结语

到本期为止，我们的医院智能体导诊系统已经完成。通过四期的工程实践，我们从一个简单的查询工具开始，逐步添加了预约功能、记忆系统、知识问答，并最终实现了一个完整的、具备异常处理和安全控制的企业级智能体应用。

回顾整个过程，我们严格遵循了智能体工程化的核心原则：结构化设计、模块化开发、完善的工具系统、强大的任务规划和记忆能力，以及全面的异常处理。这些原则正是我们在系列理论文章中强调的核心观点，通过这个实际项目，我们验证了这些观点的正确性和实用性。

作为一名坚信 "AI 不是玩具，而是能10倍提升效率的助手和员工" 的开发者，我希望这个系列实践能够帮助大家掌握智能体技术的工程化落地方法。AI 智能体技术正在快速发展，未来将在各行各业发挥重要作用，掌握这门技术，将为我们的工作和生活带来巨大的变革。

感谢大家的关注，我们后续还会有更多关于 AI 智能体工程化落地的实践内容，敬请期待！