""" 电话通话工具 - 实现电话实时语音对话功能 功能： 1. 发起电话呼叫 2. 处理来电 3. 实时语音对话（ASR + LLM + TTS） 4. 挂断电话 """ import json from typing import Optional, Dict from langchain.tools import tool from coze_coding_dev_sdk import ASRClient, TTSClient from coze_coding_utils.runtime_ctx.context import new_context import logging import asyncio from datetime import datetime logger = logging.getLogger(__name__) class PhoneCallManager: """电话通话管理器 - 管理所有通话会话""" _instance = None _calls: Dict[str, Dict] = {} # 存储活跃的通话 def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def create_call(self, call_id: str, phone_number: str, direction: str = "outbound") -> Dict: """创建新的通话会话""" call_session = { "call_id": call_id, "phone_number": phone_number, "direction": direction, # inbound/outbound "status": "ringing", # ringing/connected/ended "start_time": datetime.now(), "messages": [], # 对话历史 "user_info": None, # 用户信息 "context": {} # 上下文信息 } self._calls[call_id] = call_session logger.info(f"创建通话会话: {call_id}, 电话: {phone_number}, 方向: {direction}") return call_session def get_call(self, call_id: str) -> Optional[Dict]: """获取通话会话""" return self._calls.get(call_id) def update_call(self, call_id: str, updates: Dict): """更新通话会话""" if call_id in self._calls: self._calls[call_id].update(updates) logger.info(f"更新通话会话 {call_id}: {updates}") def end_call(self, call_id: str): """结束通话""" if call_id in self._calls: call = self._calls[call_id] call["status"] = "ended" call["end_time"] = datetime.now() duration = (call["end_time"] - call["start_time"]).total_seconds() logger.info(f"结束通话 {call_id}, 持续时间: {duration:.1f}秒") # 可以在这里保存通话记录到数据库 def get_active_calls(self) -> list: """获取所有活跃通话""" return [ call for call in self._calls.values() if call["status"] in ["ringing", "connected"] ] # 全局通话管理器 phone_manager = PhoneCallManager() @tool def initiate_phone_call( phone_number: str, greeting: Optional[str] = None ) -> str: """ 发起电话呼叫，与业主进行语音通话。 参数: phone_number: 要呼叫的电话号码 greeting: 开场白（可选），如果不提供将使用默认问候语 返回: 通话ID和状态信息 """ try: # 生成通话ID import uuid call_id = f"call_{uuid.uuid4().hex[:12]}" # 创建通话会话 call_session = phone_manager.create_call( call_id=call_id, phone_number=phone_number, direction="outbound" ) # 默认问候语 default_greeting = "您好，我是物业管家小何，有什么可以帮您的吗？" greeting_text = greeting or default_greeting # 生成问候语语音 ctx = new_context(method="phone.call") tts_client = TTSClient(ctx=ctx) greeting_audio_url, _ = tts_client.synthesize( uid=call_id, text=greeting_text, speaker="zh_female_xiaohe_uranus_bigtts", audio_format="mp3" ) # 更新通话状态 phone_manager.update_call(call_id, { "status": "connected", "greeting": greeting_text, "greeting_audio": greeting_audio_url }) logger.info(f"电话呼叫成功: {call_id} -> {phone_number}") return json.dumps({ "success": True, "call_id": call_id, "phone_number": phone_number, "status": "connected", "greeting": greeting_text, "greeting_audio_url": greeting_audio_url, "message": f"已成功呼叫 {phone_number}，通话已建立" }, ensure_ascii=False) except Exception as e: error_msg = f"发起电话呼叫失败: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def handle_incoming_call( phone_number: str ) -> str: """ 处理来电，建立通话会话。 参数: phone_number: 来电号码 返回: 通话ID和欢迎语音URL """ try: # 生成通话ID import uuid call_id = f"call_{uuid.uuid4().hex[:12]}" # 创建通话会话 call_session = phone_manager.create_call( call_id=call_id, phone_number=phone_number, direction="inbound" ) # 欢迎语 welcome_text = "您好，欢迎致电物业服务中心，我是智能管家小何，请问有什么可以帮您的？" # 生成欢迎语音 ctx = new_context(method="phone.incoming") tts_client = TTSClient(ctx=ctx) welcome_audio_url, _ = tts_client.synthesize( uid=call_id, text=welcome_text, speaker="zh_female_xiaohe_uranus_bigtts", audio_format="mp3" ) # 更新通话状态 phone_manager.update_call(call_id, { "status": "connected", "greeting": welcome_text, "greeting_audio": welcome_audio_url }) logger.info(f"来电处理成功: {call_id}, 来自: {phone_number}") return json.dumps({ "success": True, "call_id": call_id, "phone_number": phone_number, "status": "connected", "welcome_text": welcome_text, "welcome_audio_url": welcome_audio_url, "message": "来电已接通，欢迎语音已播放" }, ensure_ascii=False) except Exception as e: error_msg = f"处理来电失败: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def process_voice_in_call( call_id: str, audio_url: str ) -> str: """ 在通话中处理用户的语音输入，识别文字后返回响应。 参数: call_id: 通话ID audio_url: 用户语音的URL地址 返回: 识别的文字和回复语音URL """ try: # 获取通话会话 call = phone_manager.get_call(call_id) if not call: return json.dumps({ "success": False, "error": f"通话 {call_id} 不存在" }, ensure_ascii=False) if call["status"] != "connected": return json.dumps({ "success": False, "error": f"通话 {call_id} 未处于连接状态" }, ensure_ascii=False) # 语音识别 ctx = new_context(method="phone.asr") asr_client = ASRClient(ctx=ctx) recognized_text, _ = asr_client.recognize( uid=call_id, url=audio_url ) logger.info(f"通话 {call_id} 语音识别结果: {recognized_text}") # 记录用户消息 call["messages"].append({ "role": "user", "content": recognized_text, "timestamp": datetime.now().isoformat() }) # 这里应该调用Agent的LLM进行对话 # 在实际实现中，这会通过WebSocket实时调用主Agent # 这里先返回识别结果，让主Agent处理 return json.dumps({ "success": True, "call_id": call_id, "recognized_text": recognized_text, "message": "语音识别成功，等待Agent响应" }, ensure_ascii=False) except Exception as e: error_msg = f"处理通话语音失败: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def respond_in_call( call_id: str, text: str, speaker: str = "zh_female_xiaohe_uranus_bigtts" ) -> str: """ 在通话中用语音回复用户。 参数: call_id: 通话ID text: 要回复的文字内容 speaker: 语音角色（可选） 返回: 回复语音的URL """ try: # 获取通话会话 call = phone_manager.get_call(call_id) if not call: return json.dumps({ "success": False, "error": f"通话 {call_id} 不存在" }, ensure_ascii=False) if call["status"] != "connected": return json.dumps({ "success": False, "error": f"通话 {call_id} 未处于连接状态" }, ensure_ascii=False) # 生成回复语音 ctx = new_context(method="phone.tts") tts_client = TTSClient(ctx=ctx) audio_url, audio_size = tts_client.synthesize( uid=call_id, text=text, speaker=speaker, audio_format="mp3" ) # 记录助手消息 call["messages"].append({ "role": "assistant", "content": text, "audio_url": audio_url, "timestamp": datetime.now().isoformat() }) logger.info(f"通话 {call_id} 回复语音已生成: {text[:50]}...") return json.dumps({ "success": True, "call_id": call_id, "text": text, "audio_url": audio_url, "audio_size": audio_size, "message": "回复语音已生成，可以播放给用户" }, ensure_ascii=False) except Exception as e: error_msg = f"生成回复语音失败: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def end_phone_call( call_id: str, farewell: Optional[str] = None ) -> str: """ 结束电话通话。 参数: call_id: 通话ID farewell: 结束语（可选） 返回: 通话统计信息 """ try: # 获取通话会话 call = phone_manager.get_call(call_id) if not call: return json.dumps({ "success": False, "error": f"通话 {call_id} 不存在" }, ensure_ascii=False) # 默认结束语 default_farewell = "感谢您的来电，祝您生活愉快，再见！" farewell_text = farewell or default_farewell # 生成结束语语音 ctx = new_context(method="phone.end") tts_client = TTSClient(ctx=ctx) farewell_audio_url, _ = tts_client.synthesize( uid=call_id, text=farewell_text, speaker="zh_female_xiaohe_uranus_bigtts", audio_format="mp3" ) # 结束通话 phone_manager.end_call(call_id) # 计算通话时长 duration = (call.get("end_time", datetime.now()) - call["start_time"]).total_seconds() logger.info(f"通话 {call_id} 已结束，持续时间: {duration:.1f}秒") return json.dumps({ "success": True, "call_id": call_id, "status": "ended", "farewell": farewell_text, "farewell_audio_url": farewell_audio_url, "duration_seconds": duration, "message_count": len(call["messages"]), "phone_number": call["phone_number"], "direction": call["direction"] }, ensure_ascii=False) except Exception as e: error_msg = f"结束通话失败: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False) @tool def get_call_status( call_id: str ) -> str: """ 查询通话状态和信息。 参数: call_id: 通话ID 返回: 通话详细信息 """ try: call = phone_manager.get_call(call_id) if not call: return json.dumps({ "success": False, "error": f"通话 {call_id} 不存在" }, ensure_ascii=False) # 计算通话时长 if "end_time" in call: duration = (call["end_time"] - call["start_time"]).total_seconds() else: duration = (datetime.now() - call["start_time"]).total_seconds() return json.dumps({ "success": True, "call": { "call_id": call["call_id"], "phone_number": call["phone_number"], "direction": call["direction"], "status": call["status"], "duration_seconds": duration, "message_count": len(call["messages"]), "start_time": call["start_time"].isoformat(), "end_time": call.get("end_time", "").isoformat() if "end_time" in call else None } }, ensure_ascii=False) except Exception as e: error_msg = f"查询通话状态失败: {str(e)}" logger.error(error_msg, exc_info=True) return json.dumps({ "success": False, "error": error_msg }, ensure_ascii=False)