基于 Azure 与 OpenAI 构建智能语音客服系统

接收来电

将你在 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 应用的入口，定义了接收来电和事件回调的核心端点。这里使用了后台任务来处理耗时的呼叫流程，避免阻塞 API 响应。

# 文件：app/main.py (片段)
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from http import HTTPStatus

app = FastAPI()

@app.post("/events")
async def events(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")
async def outbound_call(
    request: OutboundCallRequest,
    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 的响应转换为语音。注意这里的异步队列设计，用于解耦音频输入输出与 LLM 推理过程。

# 文件：app/helpers/call_llm.py (片段)
@start_as_current_span("call_load_llm_chat")
async def load_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 队列。
    """
    async with SttClient(call=call, sample_rate=audio_sample_rate) as stt_client, \
               use_tts_client(call.locale) as tts_synthesizer:
        # 注册 TTS 回调，将合成的音频放入输出队列
        def tts_callback(text: str):
            asyncio.create_task(handle_realtime_tts(text, tts_synthesizer, audio_out))

        tools_plugin = DefaultPlugin(
            call=call,
            client=automation_client,
            post_callback=post_callback,
            scheduler=scheduler,
            tts_callback=tts_callback,
            tts_client=tts_synthesizer,
        )

        messages = await _init_messages(call)
        messages.append(MessageModel(persona=PersonaEnum.ASSISTANT, text=call.greeting))

        while True:
            transcript = await stt_client.wait_for_transcript()
            if transcript:
                messages.append(MessageModel(persona=PersonaEnum.USER, text=transcript))

            async for delta in completion_stream(
                max_tokens=call.max_tokens,
                messages=messages,
                system=call.prompts,
                tools=tools_plugin.tools_definitions,
            ):
                if delta.content:
                    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 缓存装饰器来优化异步和同步函数的性能，减少重复计算和外部服务调用。针对异步场景，我们特别记录了事件循环 ID 以确保缓存键的唯一性。

# 文件：app/helpers/cache.py
from collections import OrderedDict
from functools import wraps
import asyncio
from collections.abc import Awaitable

def lru_acache(maxsize: int = 128):
    """异步函数的 LRU 缓存装饰器。
    缓存函数的返回值。当达到最大大小时，最久未使用的缓存项会被移除。
    """
    def decorator(func):
        cache: OrderedDict[tuple, Awaitable] = OrderedDict()

        @wraps(func)
        async def wrapper(*args, **kwargs):
            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)

            if len(cache) > maxsize:
                cache.popitem(last=False)
            return value
        return wrapper
    return decorator

def lru_cache(maxsize: int = 128):
    """同步函数的 LRU 缓存装饰器，原理同上。"""
    pass