Call Center AI:智能语音机器人,快速构建AI客服系统
Call Center AI:智能语音机器人,快速构建AI客服系统
Call Center AI 是一个基于 Azure 和 OpenAI 的智能呼叫中心解决方案。它允许你通过 API 发起由 AI 代理拨打的电话,或直接接听来自配置号码的来电。该项目旨在帮助开发者在几小时内(而非几周)为保险、IT支持、客户服务等场景定制和部署智能语音助手。
# 示例:让AI机器人拨打电话data='{ "bot_company": "Contoso", "bot_name": "Amélie", "phone_number": "+11234567890", "task": "帮助客户解决数字化工作场所问题。助手为IT支持部门工作,目标是帮助客户解决问题并收集工单信息。", "agent_phone_number": "+33612345678", "claim": [ {"name": "硬件信息", "type": "text"}, {"name": "首次发现时间", "type": "datetime"}, {"name": "建筑位置", "type": "text"} ] }'curl\ --header 'Content-Type: application/json'\ --request POST \ --url https://your-instance/call \ --data "$data"功能特性
- 🤖 AI驱动的语音交互:集成了Azure OpenAI服务,实现与客户的自然语言对话,自动处理复杂的对话流程和任务。
- 📞 完整的通话生命周期管理:支持呼入和呼出电话,具备实时音频流、断线重连、通话录音和事件回调,确保通话的稳定性和连续性。
- ⚡️ 实时响应与低延迟:通过WebSocket实现音频流的实时双向传输,结合智能语音活动检测(VAD)和文本转语音(TTS)流式处理,提供流畅的对话体验。
- 📋 动态工单收集:可在通话开始前动态定义需要收集的信息(如文本、日期时间等),AI助手会在对话中自然地引导客户提供,并结构化输出。
- 🔧 高度可扩展的工具调用:内置插件系统,允许AI助手在对话中执行特定操作,如结束通话、创建新工单、转接人工坐席等,并可轻松扩展自定义工具。
- 📱 多渠道支持:除了语音通话,还集成了SMS功能,可以在通话前后通过短信与客户互动,例如发送通话摘要或询问额外信息。
安装指南
系统要求
- Python 3.10+
- 一个Azure订阅(需开通Communication Services、OpenAI、Speech Services等服务)
- (可选)一个Twilio账户(如果使用Twilio作为SMS提供商)
分步安装
启动服务
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload 配置环境变量或配置文件
项目支持从环境变量或YAML文件加载配置。复制示例配置文件并根据你的Azure资源进行修改。
cp config.example.yaml config.yaml # 编辑 config.yaml 文件,填入你的Azure Communication Services、OpenAI、Speech Services等的连接信息和密钥或者,你也可以将所有配置放入名为 CONFIG_JSON 的环境变量中(JSON格式)。
安装依赖
pip install -r requirements.txt 克隆仓库
git clone https://github.com/your-repo/call-center-ai.git cd call-center-ai 使用说明
发起一个由AI代理拨打的电话
通过API端点 /call 发起一个POST请求,即可让AI机器人主动呼叫指定的电话号码。
import requests import json url ="https://your-instance/call" payload ={"bot_company":"Contoso","bot_name":"Amélie","phone_number":"+11234567890","task":"帮助客户解决网络连接问题。","agent_phone_number":"+33612345678",# 人工坐席号码,用于转接"claim":[# 需要收集的工单信息{"name":"用户ID","type":"text"},{"name":"问题描述","type":"text"}]} headers ={'Content-Type':'application/json'} response = requests.post(url, headers=headers, data=json.dumps(payload))print(response.json())接收来电
将你在Azure Communication Services或Twilio中配置的电话号码指向你的服务端点。当有来电时,服务会自动应答,并启动AI助手与客户对话。
核心API概览
POST /call: 发起一个新的由AI代理拨打的电话。POST /events: 用于接收来自Azure Communication Services的通话事件回调(如通话已连接、已断开、DTMF识别结果等)。- WebSocket
/ws: 用于与实时音频流进行双向通信,实现低延迟的语音交互。 POST /sms: 用于接收SMS消息的回调端点。
核心代码
1. 呼叫入口与事件处理 (app/main.py - 片段)
这是FastAPI应用的入口,定义了接收来电和事件回调的核心端点。
# 文件: app/main.py (片段)from fastapi import FastAPI, Request # ... 其他导入 app = FastAPI()@app.post("/events")asyncdefevents(request: Request)-> JSONResponse:""" 处理来自 Azure Communication Services 的回调事件。 包括新呼入、通话已连接、已断开、DTMF识别等。 """# 解析云事件 cloud_events =await request.json()for cloud_event in cloud_events: event = CloudEvent.from_dict(cloud_event) logger.info("收到事件: %s", event.type)# 根据事件类型分发给不同的处理函数if event.type=="Microsoft.Communication.CallLegStateChanged":if event.data["state"]=="Connected":# 处理通话已连接事件,开始媒体流await on_call_connected(...)elif event.type=="Microsoft.Communication.PlayCompleted":await on_automation_play_completed(...)# ... 其他事件处理return JSONResponse(status_code=HTTPStatus.OK, content={})@app.post("/call")asyncdefoutbound_call( request: OutboundCallRequest,# Pydantic模型,包含呼出参数 background_tasks: BackgroundTasks,)-> JSONResponse:""" 发起一个由AI代理拨打的呼出电话。 """# 创建通话状态模型 call_state = CallStateModel(# ... 从请求中初始化)# 保存通话状态到数据库await _db.call_create(call_state)# 在后台任务中启动呼叫流程,避免阻塞API响应 background_tasks.add_task( start_outbound_call, callback_url=str(request.url_for("events")), call=call_state,# ...)return JSONResponse(status_code=HTTPStatus.ACCEPTED, content={"call_id":str(call_state.call_id)})2. AI对话核心逻辑 (app/helpers/call_llm.py - 片段)
这是AI对话管理的核心,负责接收用户语音识别结果,调用LLM,并将LLM的响应转换为语音。
# 文件: app/helpers/call_llm.py (片段)@start_as_current_span("call_load_llm_chat")asyncdefload_llm_chat( audio_in: asyncio.Queue[bytes],# 从用户接收的音频流 audio_out: asyncio.Queue[bytes|bool],# 发送给用户的音频流 audio_sample_rate:int, automation_client: CallAutomationClient, call: CallStateModel, post_callback: Callable[[CallStateModel], Awaitable[None]], scheduler: Scheduler, training_callback: Callable[[CallStateModel], Awaitable[None]],)->None:""" 加载并运行LLM聊天循环。从audio_in队列获取用户的语音输入, 将其转换为文本,发送给LLM,然后将LLM的文本响应转换为语音放入audio_out队列。 """# 初始化语音识别(STT)和语音合成(TTS)客户端asyncwith SttClient(call=call, sample_rate=audio_sample_rate)as stt_client, \ use_tts_client(call.locale)as tts_synthesizer:# 注册TTS回调,将合成的音频放入输出队列deftts_callback(text:str): asyncio.create_task(handle_realtime_tts(text, tts_synthesizer, audio_out))# 初始化LLM工具插件 tools_plugin = DefaultPlugin( call=call, client=automation_client, post_callback=post_callback, scheduler=scheduler, tts_callback=tts_callback, tts_client=tts_synthesizer,)# 创建LLM对话消息历史 messages =await _init_messages(call) messages.append(MessageModel(persona=PersonaEnum.ASSISTANT, text=call.greeting))# 主对话循环whileTrue:# 等待并获取用户的下一条语音消息 transcript =await stt_client.wait_for_transcript()if transcript: messages.append(MessageModel(persona=PersonaEnum.USER, text=transcript))# 将对话历史发送给LLM获取回复asyncfor delta in completion_stream( max_tokens=call.max_tokens, messages=messages, system=call.prompts, tools=tools_plugin.tools_definitions,):# 处理LLM的流式响应,包括文本和工具调用if delta.content:# 将文本响应通过TTS播报出去await tts_callback(delta.content)if delta.tool_calls:# 执行工具调用await tools_plugin.execute_tool_call(delta.tool_calls[0])# 将助手的最新回复添加到消息历史# ...3. 缓存装饰器 (app/helpers/cache.py)
项目使用自定义的LRU缓存装饰器来优化异步和同步函数的性能,减少重复计算和外部服务调用。
# 文件: app/helpers/cache.pyfrom collections import OrderedDict from functools import wraps import asyncio from collections.abc import Awaitable deflru_acache(maxsize:int=128):""" 异步函数的LRU缓存装饰器。 缓存函数的返回值。当达到最大大小时,最久未使用的缓存项会被移除。 """defdecorator(func): cache: OrderedDict[tuple, Awaitable]= OrderedDict()@wraps(func)asyncdefwrapper(*args,**kwargs):# 创建一个包含事件循环ID、位置参数和关键字参数的缓存键 key =(id(asyncio.get_event_loop()), args,frozenset(kwargs.items()),)if key in cache:# 如果缓存命中,将该键移动到末尾(表示最近使用) cache.move_to_end(key)return cache[key]# 计算新值并存入缓存 value =await func(*args,**kwargs) cache[key]= value cache.move_to_end(key)# 如果超出最大大小,移除最久未使用的项(第一个)iflen(cache)> maxsize: cache.popitem(last=False)return value return wrapper return decorator # 同步函数的LRU缓存deflru_cache(maxsize:int=128):"""同步函数的LRU缓存装饰器,原理同上。"""# ... 实现 ```FINISHED AHFTVYTuX6S/IgIPsbXOvjn9f7SQ1VezSmjKy6Zrwto= 更多精彩内容 请关注我的个人公众号 公众号(办公AI智能小助手) 对网络安全、黑客技术感兴趣的朋友可以关注我的安全公众号(网络安全技术点滴分享)