AI+医疗产品扣子客服智能体开发实战：从零构建高可用对话系统

最近在做一个医疗产品的智能客服项目，客户那边对专业性和合规性要求特别高。传统的规则脚本或者简单的问答机器人根本应付不来，比如用户问“阿司匹林能和布洛芬一起吃吗”，这种涉及药品配伍禁忌的问题，如果回答错了，后果很严重。所以，我们决定用更现代的“智能体”思路来构建这个系统，并选择了扣子（Coze）的智能体开发框架作为基础。今天这篇笔记，就和大家分享一下我们从零开始，构建一个高可用医疗客服对话系统的实战过程和踩过的坑。

1. 为什么医疗客服这么难做？—— 背景与核心痛点

在动手之前，我们花了很长时间去理解医疗客服场景的特殊性，这直接决定了技术方案的设计。总结下来，主要有三大挑战：

1. 专业术语理解难，歧义多：医疗领域有海量的专业名词、药品名（商品名、通用名、化学名）、检查项目、疾病名称等。比如“心梗”是“心肌梗死”的简称，“PCR”在医疗语境下是“聚合酶链式反应”（一种检测技术），而在其他领域可能是别的意思。简单的关键词匹配在这里会漏洞百出。
2. 隐私与合规要求极其严格：这可能是医疗AI项目最大的门槛。系统需要处理患者的症状描述、用药历史等敏感信息，必须符合像HIPAA（Health Insurance Portability and Accountability Act，健康保险携带和责任法案）这类法规。这意味着数据在传输、存储、日志记录的每一个环节都需要加密和脱敏，未经授权绝对不能泄露。
3. 服务必须连续、可靠，且能处理紧急情况：客服对话不能突然中断。用户可能从询问挂号流程，自然过渡到描述症状，再问到某个药的副作用。系统需要记住对话的上下文（对话状态跟踪）。更重要的是，当识别到用户描述“胸痛剧烈”、“呼吸困难”等可能属于急症的情况时，系统必须能触发紧急流程，比如结束AI对话，立即转接人工坐席。

2. 技术选型：为什么是智能体框架？

面对这些挑战，我们评估了几种常见方案：

• 纯规则引擎：用if-else或者正则表达式。优点是可控、解释性强，但维护成本随着规则数量指数级增长，无法理解语义，面对复杂、多变的用户问法会非常吃力。
• 传统NLP流水线：串联意图分类、实体识别、槽位填充等模块。灵活性比规则好，但各模块是孤立的，错误会逐级传递，且很难实现复杂的多轮对话管理和基于知识的推理。
• 智能体（Agent）框架：以扣子（Coze）为例，它提供了一个以“智能体”为核心的设计范式。智能体可以理解为具备一定自主决策能力的程序单元，它内部可以整合多种工具（Tools）、记忆（Memory）和知识（Knowledge）。对我们来说，这意味着：
  • 模块化与集成：我们可以把专业的医疗实体识别模型、内部的知识图谱、合规检查逻辑都封装成“工具”，让智能体在对话过程中按需调用。
  • 状态管理内置：框架通常提供了对话状态（Dialog State）管理的机制，方便我们跟踪用户问了什么、提供了什么信息。
  • 灵活的流程控制：易于实现“先问症状，再推荐科室，最后提供挂号指引”这样的多轮对话流程，以及在特定条件下（如识别到急症关键词）执行强制跳转（如转人工）。

基于以上对比，扣子智能体框架在灵活性、可扩展性以及对复杂对话流程的支持上，更契合我们医疗客服的场景，因此成为了我们的技术基座。

3. 核心模块实现拆解

确定了框架，接下来就是填充核心能力。我们主要建设了三个模块：

3.1 医疗实体识别模块：BERT+CRF

意图识别是基础，但医疗场景下，精确地抽取出用户话语中的实体（疾病、药品、症状）更为关键。我们采用经典的BERT+CRF（Conditional Random Field，条件随机场）序列标注模型。

• 思路：BERT负责获取每个汉字/词语的上下文相关语义表示，CRF层在此基础上学习标签之间的转移规则（比如“疾病”标签后面更可能跟“症状”标签，而不是“人名”标签），从而得到全局最优的标签序列。
• 数据预处理示例：医疗数据稀缺，我们采用“公开数据集+业务日志脱敏+数据增强”的方式构建训练集。数据增强包括同义词替换（如“发烧”替换为“发热”）、实体随机掩码等。

import json import random from typing import List, Dict def load_and_augment_medical_data(file_path: str, augmentation_factor: float = 0.3) -> List[Dict]: """ 加载并增强医疗NER训练数据。 Args: file_path: 训练数据文件路径，格式为每行一个JSON，包含`text`和`labels`。 augmentation_factor: 数据增强的比例。 Returns: 增强后的数据列表。 """ with open(file_path, 'r', encoding='utf-8') as f: raw_data = [json.loads(line) for line in f] augmented_data = [] medical_synonyms = { # 简化示例 "发烧": ["发热", "高热"], "头痛": ["头疼"], "阿司匹林": ["乙酰水杨酸"] } for item in raw_data: augmented_data.append(item) if random.random() < augmentation_factor: text = item['text'] new_text = text # 简单的同义词替换增强 for word, syn_list in medical_synonyms.items(): if word in text and syn_list: new_text = new_text.replace(word, random.choice(syn_list), 1) # 只替换第一个出现 if new_text != text: # 注意：实际应用中，替换实体后，对应的标签序列也需要相应调整，这里为简化省略 # 这里仅示意文本增强，生产环境需要更严谨的标签同步处理 augmented_item = item.copy() augmented_item['text'] = new_text augmented_data.append(augmented_item) return augmented_data # 使用示例 # train_data = load_and_augment_medical_data('medical_ner_train.jsonl')

3.2 对话状态管理：基于Redis的状态机

多轮对话的核心是记住上下文。我们在扣子智能体的记忆机制外，用Redis实现了一个更持久、可共享的对话状态机。

• 设计：每个会话（Session）一个唯一的ID，作为Redis的Key。Value存储一个结构化的JSON对象，包含当前对话阶段、已收集的槽位（Slots）信息（如用户症状、疑似疾病、已提及药品）、时间戳等。
• TTL与持久化：为每个会话Key设置TTL（Time-To-Live，例如30分钟），超时自动清除，释放资源。同时，重要的对话状态（如已确认的问诊记录）会异步持久化到数据库，供后续分析或人工查看。

import json import redis from datetime import datetime from typing import Optional, Any class DialogStateManager: """基于Redis的对话状态管理器。""" def __init__(self, redis_client: redis.Redis, ttl_seconds: int = 1800): self.redis = redis_client self.ttl = ttl_seconds def get_state(self, session_id: str) -> Optional[dict]: """获取指定会话的当前状态。""" state_json = self.redis.get(f"dialog_state:{session_id}") if state_json: return json.loads(state_json) return None def update_state(self, session_id: str, **updates) -> bool: """ 更新会话状态。如果会话不存在则创建。 Args: session_id: 会话唯一标识。 **updates: 要更新或添加的键值对。 Returns: 更新是否成功。 """ current_state = self.get_state(session_id) or {} current_state.update(updates) current_state['last_updated'] = datetime.now().isoformat() try: self.redis.setex( name=f"dialog_state:{session_id}", time=self.ttl, value=json.dumps(current_state, ensure_ascii=False) ) return True except redis.RedisError as e: # 生产环境应接入日志和监控 print(f"Redis操作失败: {e}") return False def clear_state(self, session_id: str) -> None: """清除指定会话的状态。""" self.redis.delete(f"dialog_state:{session_id}") # 初始化Redis连接（示例） # r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True) # state_mgr = DialogStateManager(r)

3.3 知识图谱问答融合

对于“两种药能不能一起吃”、“某个症状可能是什么病”这类问题，需要深度的知识推理。我们将内部构建的医疗知识图谱（Knowledge Graph）封装成问答工具。

• 对接方案：在扣子智能体中创建一个query_medical_kg工具。当用户问题被识别为需要专业知识解答时，智能体调用此工具。
• 工具内部：接收用户问题（或提取出的实体），将其转换为知识图谱查询语句（如Cypher语句查询Neo4j），执行查询，并将返回的路径、节点信息组织成自然语言回复。

4. 核心对话处理类代码示例

下面是一个整合了上述模块核心功能的对话处理类。它体现了生产代码应有的异常处理、合规校验和清晰的结构。

import functools import logging from typing import Callable, Dict, Any logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def handle_dialog_errors(func: Callable) -> Callable: """装饰器：统一处理对话流程中的异常，并记录日志。""" @functools.wraps(func) def wrapper(self, *args, **kwargs): session_id = kwargs.get('session_id') or (args[0] if args else 'unknown') try: return func(self, *args, **kwargs) except Exception as e: error_msg = f"对话处理失败 [Session: {session_id}]: {e}" logger.error(error_msg, exc_info=True) # 返回一个友好的降级回复，避免用户看到技术错误 return { "reply": "抱歉，系统暂时遇到了一点问题。您可以重新描述您的问题，或联系人工客服。", "should_transfer_to_human": True, # 触发转人工标志 "error": error_msg } return wrapper class MedicalDialogProcessor: """医疗客服对话核心处理器。""" def __init__(self, state_manager: DialogStateManager, ner_model, kg_client): self.state_manager = state_manager self.ner_model = ner_model # 假设是加载好的BERT+CRF模型 self.kg_client = kg_client # 知识图谱查询客户端 @handle_dialog_errors def process(self, user_input: str, session_id: str) -> Dict[str, Any]: """ 处理用户输入，返回系统回复和动作。 Args: user_input: 用户输入的文本。 session_id: 当前会话ID。 Returns: 包含回复文本、后续动作等信息的字典。 """ # 1. 合规性校验：检查输入是否包含不允许的敏感信息（如身份证号完整号码） if self._contains_sensitive_info(user_input): logger.warning(f"输入可能包含敏感信息，已拦截。Session: {session_id}") return { "reply": "您输入的内容可能包含个人敏感信息，为了您的隐私安全，请勿在此透露。您可以描述症状或咨询医疗问题。", "action": "sensitive_input_blocked" } # 2. 获取并更新对话状态 current_state = self.state_manager.get_state(session_id) or {} current_state.setdefault('conversation_stage', 'greeting') current_state.setdefault('slots', {}) # 3. 医疗实体识别 entities = self._extract_medical_entities(user_input) logger.info(f"Session {session_id} 识别到实体: {entities}") # 4. 根据对话阶段和实体决定回复逻辑 reply, next_action = self._decide_response( user_input, entities, current_state ) # 5. 更新状态（例如，填充槽位，推进阶段） current_state['slots'].update(self._extract_slots_from_entities(entities)) if next_action == 'move_to_diagnosis': current_state['conversation_stage'] = 'diagnosis_inquiry' elif next_action == 'transfer_human': current_state['conversation_stage'] = 'awaiting_human' self.state_manager.update_state(session_id, **current_state) # 6. 检查是否需要紧急转人工（例如，识别到急症关键词） if self._is_emergency_situation(user_input, entities): reply = "您描述的情况可能比较紧急，为了您的安全，我将立即为您转接人工医疗顾问。请稍候。" next_action = 'emergency_transfer' return { "reply": reply, "action": next_action, "updated_state": current_state # 实际返回可能不包含完整状态，这里仅为示意 } def _contains_sensitive_info(self, text: str) -> bool: """简易敏感信息检查（示例，生产环境需要更复杂的规则或模型）。""" import re # 简单正则示例，匹配18位身份证号（生产环境需更严谨） id_pattern = r'\b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[1-2]\d|3[0-1])\d{3}[\dXx]\b' if re.search(id_pattern, text): return True return False def _extract_medical_entities(self, text: str) -> List[Dict]: """调用NER模型提取医疗实体。""" # 这里应是调用真实模型的代码，返回如 [{'text':'头痛','type':'SYMPTOM','start':0,'end':2}, ...] # 示例返回 return self.ner_model.predict(text) # 假设的模型接口 def _decide_response(self, user_input: str, entities: List[Dict], state: Dict): """决策逻辑的核心。""" # 这是一个简化的决策树示例 stage = state['conversation_stage'] if stage == 'greeting': return "您好，我是医疗助手。请问您有什么不舒服吗？", 'ask_symptom' elif stage == 'diagnosis_inquiry' and entities: # 如果有疾病或症状实体，尝试查询知识图谱 kg_answer = self.kg_client.query(entities) return f"根据您描述的‘{entities[0][“text”]}’，{kg_answer}", 'provide_info' # ... 更多阶段判断 return "我还在学习中，您能换个说法再问一次吗？", 'fallback' def _is_emergency_situation(self, text: str, entities: List[Dict]) -> bool: """判断是否为紧急情况。""" emergency_keywords = ['胸痛', '窒息', '昏迷', '大出血', '呼吸困难'] for kw in emergency_keywords: if kw in text: return True # 也可以结合实体类型判断，如识别到“心肌梗死”实体 for ent in entities: if ent.get('type') == 'DISEASE' and ent.get('text') in ['心肌梗死', '脑卒中']: return True return False

5. 生产环境必须考虑的几点

代码跑通只是第一步，要真正上线，还有几个关键问题要解决：

1. 对话日志脱敏：所有经过系统的用户输入、AI回复、内部状态，在写入日志或分析系统前，必须进行脱敏处理。我们写了一个过滤器，自动将识别到的疾病名、药品名、身体部位等替换为通用标签，如[疾病实体]、[药品A]。
2. 并发与会话隔离：在高并发下，必须保证每个用户的会话状态绝对隔离。我们的session_id通常由前端生成并保证唯一（如UUID），结合Redis的Key-Value结构，天然支持隔离。需要注意Redis连接池的管理和可能出现的写冲突（虽然概率极低）。
3. 模型热更新：医疗知识更新快，NER模型或意图分类模型需要迭代。我们采用“模型版本目录+配置中心”的方式。服务从配置中心读取当前生效的模型版本路径，加载模型。更新时，先将新模型部署到新目录，然后在配置中心切换版本号，服务通过健康检查或定时拉取感知到变化后，平滑加载新模型，旧请求仍由旧模型处理直至完成。

6. 医疗场景特有的“坑”与对策

• 专业术语导致的意图漂移：用户说“我拉肚子”，模型可能识别为“腹泻”（正确），但也可能被泛化理解为“肠胃不适”。对策是构建高质量的领域词典和同义词库，并在模型训练时作为特征输入，强化领域相关性。定期用bad case（错误案例）更新训练数据。
• 坐席接管机制：这是安全底线。我们在系统中设置了多级触发条件：
  • 用户主动要求：用户输入“转人工”。
  • AI识别到紧急情况：如上述_is_emergency_situation方法。
  • AI多次无法理解：连续N轮对话落入fallback（回退）流程。
  • 用户情绪负面：通过简单的情感分析（如检测大量负面词）触发。 一旦触发，系统会立即锁定该会话状态，向人工坐席系统发送警报和完整的脱敏后对话历史，并将后续该会话的所有用户输入直接路由给坐席。

7. 挑战任务：完善药品配伍禁忌检查

现在，你已经有了一个基线系统。我们来设想一个更具体的功能增强点。

任务背景：在process方法中，我们提取了药品实体。但医疗客服经常需要回答“A药和B药能一起吃吗？”这类问题。

你的挑战： 请基于现有的MedicalDialogProcessor类，设计并实现一个check_drug_interaction方法。该方法应：

1. 接收一个包含药品实体（DRUG类型）的列表。
2. 查询知识图谱（或一个模拟的禁忌规则库），判断这些药品之间是否存在已知的配伍禁忌。
3. 如果存在禁忌，返回具体的警示信息（如“阿司匹林与华法林同用可能增加出血风险，请务必咨询医生或药师。”）；如果不存在已知禁忌，则返回安全提示（如“未发现明确的配伍禁忌，但合并用药仍需谨慎，请遵医嘱。”）。
4. 将这个方法整合到_decide_response或process的逻辑中，当用户输入中识别出两种或以上药品实体时，自动触发检查并将结果融入回复。

提示：你可以先实现一个本地的禁忌字典{('阿司匹林', '华法林'): '增加出血风险'， ...}来模拟知识图谱查询。思考如何优雅地处理药品名不同表述（如“拜阿司匹林” vs “阿司匹林肠溶片”）的归一化问题。

---

整个项目做下来，感觉智能体框架确实为构建复杂的领域对话系统提供了很好的抽象。最大的体会是，在医疗这类严肃领域，技术方案的“可靠性”和“可控性”远比“炫技”重要。每一步操作、每一个回复都要有据可查，有规可循。希望这篇笔记里分享的思路和代码片段，能给你带来一些启发。如果你也在做类似的项目，欢迎一起交流那些让人头疼的“坑”和有趣的解决方案。