14c - AI 电话 Agent 构建

┌─────────────────────────────────────────────────────┐
│                   电话网络层                          │
│  PSTN / SIP Trunk / WebRTC                          │
│  (Twilio / Vonage / 运营商)                          │
├─────────────────────────────────────────────────────┤
│                   语音处理层                          │
│  STT (Deepgram Nova-3) ←→ TTS (ElevenLabs Flash)   │
│  中断检测 · VAD · 噪声消除                           │
├─────────────────────────────────────────────────────┤
│                   对话编排层                          │
│  状态机 · 意图路由 · 上下文管理 · 工具调用            │
│  (Vapi.ai / 自建 WebSocket 服务)                     │
├─────────────────────────────────────────────────────┤
│                   业务逻辑层                          │
│  LLM 推理 · CRM 集成 · 日程 API · 知识库检索         │
│  (OpenAI / Claude / Gemini + 自定义工具)             │
└─────────────────────────────────────────────────────┘

来电 → Vapi 电话网关 → STT (Deepgram) → LLM (GPT-4/Claude) → TTS (ElevenLabs) → 语音回复
                ↕                                    ↕
         Twilio/Vonage 号码                    自定义工具 (Function Calling)

# 设置环境变量
export VAPI_API_KEY="vapi_xxxxxxxxxxxxxxxx"

import requests
import os
 
VAPI_API_KEY = os.getenv("VAPI_API_KEY")
 
# 创建 Assistant
response = requests.post(
    "https://api.vapi.ai/assistant",
    headers={
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "name": "客服 Agent",
        "model": {
            "provider": "openai",
            "model": "gpt-4o",
            "messages": [
                {
                    "role": "system",
                    "content": """你是一位专业的客服代表，代表 [公司名称] 接听客户来电。
 
## 你的职责
- 友好地问候客户并确认身份
- 回答产品和服务相关问题
- 处理预约和订单查询
- 无法解决的问题转接人工客服
 
## 对话规则
- 使用简洁、口语化的中文
- 每次回复控制在 2-3 句话以内
- 主动确认客户需求，避免误解
- 敏感操作（退款、取消）需二次确认"""
                }
            ]
        },
        "voice": {
            "provider": "11labs",
            "voiceId": "21m00Tcm4TlvDq8ikWAM",  # Rachel 语音
            "stability": 0.5,
            "similarityBoost": 0.75
        },
        "firstMessage": "您好，这里是 [公司名称] 客服中心，请问有什么可以帮您的？",
        "endCallMessage": "感谢您的来电，祝您生活愉快，再见！",
        "transcriber": {
            "provider": "deepgram",
            "model": "nova-3",
            "language": "zh"
        },
        "silenceTimeoutSeconds": 30,
        "maxDurationSeconds": 600,
        "endCallFunctionEnabled": True
    }
)
 
assistant = response.json()
print(f"Assistant ID: {assistant['id']}")

// TypeScript 版本
import axios from 'axios';
 
const VAPI_API_KEY = process.env.VAPI_API_KEY;
 
async function createAssistant() {
  const response = await axios.post(
    'https://api.vapi.ai/assistant',
    {
      name: '客服 Agent',
      model: {
        provider: 'openai',
        model: 'gpt-4o',
        messages: [
          {
            role: 'system',
            content: `你是一位专业的客服代表，代表 [公司名称] 接听客户来电。
使用简洁口语化的中文，每次回复 2-3 句话。
无法解决的问题转接人工客服。`
          }
        ]
      },
      voice: {
        provider: '11labs',
        voiceId: '21m00Tcm4TlvDq8ikWAM',
        stability: 0.5,
        similarityBoost: 0.75
      },
      firstMessage: '您好，这里是客服中心，请问有什么可以帮您的？',
      transcriber: {
        provider: 'deepgram',
        model: 'nova-3',
        language: 'zh'
      },
      silenceTimeoutSeconds: 30,
      maxDurationSeconds: 600
    },
    {
      headers: {
        Authorization: `Bearer ${VAPI_API_KEY}`,
        'Content-Type': 'application/json'
      }
    }
  );
 
  console.log(`Assistant ID: ${response.data.id}`);
  return response.data;
}

# 购买 Vapi 号码
response = requests.post(
    "https://api.vapi.ai/phone-number",
    headers={
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "provider": "vapi",
        "assistantId": assistant["id"],
        "numberDesiredAreaCode": "415"  # 旧金山区号
    }
)
phone = response.json()
print(f"电话号码: {phone['number']}")

# 导入已有的 Twilio 号码
response = requests.post(
    "https://api.vapi.ai/phone-number",
    headers={
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "provider": "twilio",
        "number": "+14155551234",
        "twilioAccountSid": os.getenv("TWILIO_ACCOUNT_SID"),
        "twilioAuthToken": os.getenv("TWILIO_AUTH_TOKEN"),
        "assistantId": assistant["id"]
    }
)

# 为 Assistant 添加工具
requests.patch(
    f"https://api.vapi.ai/assistant/{assistant['id']}",
    headers={
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "model": {
            "provider": "openai",
            "model": "gpt-4o",
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "query_order",
                        "description": "根据订单号查询订单状态",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "order_id": {
                                    "type": "string",
                                    "description": "客户提供的订单号"
                                }
                            },
                            "required": ["order_id"]
                        }
                    },
                    "server": {
                        "url": "https://your-api.com/vapi/tool-handler"
                    }
                },
                {
                    "type": "function",
                    "function": {
                        "name": "transfer_to_human",
                        "description": "当客户要求或问题无法解决时，转接人工客服",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "reason": {
                                    "type": "string",
                                    "description": "转接原因"
                                },
                                "department": {
                                    "type": "string",
                                    "enum": ["sales", "support", "billing"],
                                    "description": "转接部门"
                                }
                            },
                            "required": ["reason"]
                        }
                    }
                }
            ]
        }
    }
)

# tool_handler.py — FastAPI 工具处理服务器
from fastapi import FastAPI, Request
import json
 
app = FastAPI()
 
@app.post("/vapi/tool-handler")
async def handle_tool_call(request: Request):
    """处理 Vapi 发来的工具调用请求"""
    body = await request.json()
    
    # Vapi 发送的请求结构
    tool_call = body.get("message", {}).get("toolCalls", [{}])[0]
    function_name = tool_call.get("function", {}).get("name")
    arguments = json.loads(tool_call.get("function", {}).get("arguments", "{}"))
    
    if function_name == "query_order":
        order_id = arguments.get("order_id")
        # 查询数据库（示例）
        order_info = await query_order_from_db(order_id)
        return {
            "results": [
                {
                    "toolCallId": tool_call["id"],
                    "result": f"订单 {order_id} 状态：{order_info['status']}，"
                              f"预计 {order_info['eta']} 送达。"
                }
            ]
        }
    
    elif function_name == "transfer_to_human":
        reason = arguments.get("reason")
        department = arguments.get("department", "support")
        # 返回转接指令
        return {
            "results": [
                {
                    "toolCallId": tool_call["id"],
                    "result": "正在为您转接人工客服，请稍候..."
                }
            ]
        }
 
async def query_order_from_db(order_id: str) -> dict:
    """模拟订单查询"""
    return {"status": "已发货", "eta": "明天下午"}

# 发起外呼电话
response = requests.post(
    "https://api.vapi.ai/call/phone",
    headers={
        "Authorization": f"Bearer {VAPI_API_KEY}",
        "Content-Type": "application/json"
    },
    json={
        "assistantId": assistant["id"],
        "phoneNumberId": phone["id"],
        "customer": {
            "number": "+8613800138000",  # 目标号码
            "name": "张先生"
        },
        "assistantOverrides": {
            "firstMessage": "张先生您好，我是 [公司] 的客服小助手，"
                           "想跟您确认一下明天的预约时间，方便吗？"
        }
    }
)
call = response.json()
print(f"通话 ID: {call['id']}, 状态: {call['status']}")

你是 [公司名称] 的 AI 电话客服代表 [Agent名称]。

## 身份与风格
- 语气：专业但亲切，像一位经验丰富的客服
- 语速：适中，不要太快
- 每次回复：控制在 2-3 句话，避免长篇大论
- 语言：[目标语言]

## 对话流程
1. 问候并确认客户身份（姓名或手机尾号）
2. 询问来电目的
3. 根据意图分类处理：
   - 订单查询 → 调用 query_order 工具
   - 预约安排 → 调用 book_appointment 工具
   - 投诉建议 → 记录后转人工
   - 其他问题 → 尝试回答，无法回答则转人工

## 关键规则
- 涉及退款、取消订单等敏感操作，必须二次确认
- 客户情绪激动时，先安抚再处理
- 超出能力范围时，主动说"我帮您转接专业同事"
- 不要编造信息，不确定的说"我帮您查一下"

## 结束通话
- 确认问题已解决："请问还有其他需要帮助的吗？"
- 礼貌告别："感谢您的来电，祝您生活愉快！"

来电 → Twilio PSTN → TwiML <Connect>
                         ↓
              <ConversationRelay>
                    ↕ WebSocket
              你的 AI 应用服务器
                    ↕
              LLM (OpenAI/Claude)
                    ↕
              业务工具 (CRM/日程/知识库)

# 安装 Twilio CLI
npm install -g twilio-cli
 
# 登录
twilio login
 
# 购买电话号码（美国号码示例）
twilio phone-numbers:buy:local --area-code 415
 
# 安装 Python SDK
pip install twilio fastapi uvicorn websockets openai

# 设置环境变量
export TWILIO_ACCOUNT_SID="ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
export TWILIO_AUTH_TOKEN="your_auth_token"
export TWILIO_PHONE_NUMBER="+14155551234"
export OPENAI_API_KEY="sk-xxxxxxxx"

# server.py — Twilio ConversationRelay 服务器
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import Response
from twilio.twiml.voice_response import VoiceResponse, Connect
import json
import os
from openai import AsyncOpenAI
 
app = FastAPI()
openai_client = AsyncOpenAI()
 
# ========== 1. 来电 Webhook ==========
@app.post("/incoming-call")
async def handle_incoming_call(request: Request):
    """处理来电，返回 TwiML 指令启动 ConversationRelay"""
    response = VoiceResponse()
    connect = Connect()
    
    # 配置 ConversationRelay
    conversation_relay = connect.conversation_relay(
        url=f"wss://{request.headers['host']}/ws",  # WebSocket 地址
        welcome_greeting="您好，欢迎致电客服中心，请问有什么可以帮您？",
        welcome_greeting_interruptible="speech",
        tts_provider="ElevenLabs",
        voice="21m00Tcm4TlvDq8ikWAM",  # ElevenLabs Rachel
        transcription_provider="Deepgram",
        speech_model="nova-3-general",
        language="zh-CN",
        interruptible="true"
    )
    
    response.append(connect)
    return Response(content=str(response), media_type="application/xml")

# server.py（续）— WebSocket 处理
 
# ========== 2. WebSocket 对话处理 ==========
@app.websocket("/ws")
async def websocket_handler(ws: WebSocket):
    """处理 ConversationRelay 的 WebSocket 连接"""
    await ws.accept()
    
    # 对话历史
    conversation_history = [
        {
            "role": "system",
            "content": """你是一位专业的电话客服代表。
规则：
- 回复简洁，每次 1-3 句话
- 使用口语化中文
- 不确定的信息说"我帮您查一下"
- 需要转人工时说"我帮您转接专业同事"
"""
        }
    ]
    
    try:
        while True:
            # 接收 ConversationRelay 消息
            data = await ws.receive_text()
            message = json.loads(data)
            
            msg_type = message.get("type")
            
            if msg_type == "setup":
                # 连接建立，可获取通话元数据
                call_sid = message.get("callSid")
                print(f"通话已连接: {call_sid}")
                
            elif msg_type == "prompt":
                # 收到用户语音转录文本
                user_text = message.get("voicePrompt", "")
                print(f"用户: {user_text}")
                
                if not user_text.strip():
                    continue
                
                # 添加到对话历史
                conversation_history.append({
                    "role": "user",
                    "content": user_text
                })
                
                # 调用 LLM 获取回复
                llm_response = await openai_client.chat.completions.create(
                    model="gpt-4o",
                    messages=conversation_history,
                    max_tokens=150,
                    temperature=0.7
                )
                
                assistant_text = llm_response.choices[0].message.content
                print(f"Agent: {assistant_text}")
                
                # 添加到对话历史
                conversation_history.append({
                    "role": "assistant",
                    "content": assistant_text
                })
                
                # 发送回复给 ConversationRelay（会自动 TTS 播放）
                await ws.send_json({
                    "type": "text",
                    "token": assistant_text,
                    "last": True  # 标记为最后一个文本块
                })
                
            elif msg_type == "interrupt":
                # 用户打断了 Agent 说话
                print("用户打断，停止当前回复")
                
            elif msg_type == "dtmf":
                # 用户按了电话按键
                digit = message.get("digit")
                print(f"用户按键: {digit}")
                if digit == "0":
                    # 按 0 转人工
                    await ws.send_json({
                        "type": "text",
                        "token": "正在为您转接人工客服，请稍候...",
                        "last": True
                    })
                    # 这里可以触发 Twilio 转接逻辑
                    
            elif msg_type == "error":
                print(f"错误: {message.get('description')}")
                
    except WebSocketDisconnect:
        print("通话结束")

// server.ts — TypeScript 版本（Express + ws）
import express from 'express';
import { WebSocketServer, WebSocket } from 'ws';
import { createServer } from 'http';
import OpenAI from 'openai';
import twilio from 'twilio';
 
const app = express();
const server = createServer(app);
const wss = new WebSocketServer({ server, path: '/ws' });
const openai = new OpenAI();
 
// 来电 Webhook
app.post('/incoming-call', (req, res) => {
  const VoiceResponse = twilio.twiml.VoiceResponse;
  const response = new VoiceResponse();
  const connect = response.connect();
  
  connect.conversationRelay({
    url: `wss://${req.headers.host}/ws`,
    welcomeGreeting: '您好，欢迎致电客服中心，请问有什么可以帮您？',
    ttsProvider: 'ElevenLabs',
    voice: '21m00Tcm4TlvDq8ikWAM',
    transcriptionProvider: 'Deepgram',
    speechModel: 'nova-3-general',
    language: 'zh-CN',
    interruptible: 'true'
  });
  
  res.type('text/xml').send(response.toString());
});
 
// WebSocket 对话处理
wss.on('connection', (ws: WebSocket) => {
  const history: Array<{role: string; content: string}> = [
    {
      role: 'system',
      content: '你是一位专业的电话客服代表。回复简洁，每次 1-3 句话。'
    }
  ];
  
  ws.on('message', async (data: Buffer) => {
    const message = JSON.parse(data.toString());
    
    switch (message.type) {
      case 'setup':
        console.log(`通话已连接: ${message.callSid}`);
        break;
        
      case 'prompt': {
        const userText = message.voicePrompt || '';
        if (!userText.trim()) return;
        
        console.log(`用户: ${userText}`);
        history.push({ role: 'user', content: userText });
        
        // 调用 LLM
        const completion = await openai.chat.completions.create({
          model: 'gpt-4o',
          messages: history,
          max_tokens: 150
        });
        
        const reply = completion.choices[0].message.content || '';
        console.log(`Agent: ${reply}`);
        history.push({ role: 'assistant', content: reply });
        
        // 发送回复
        ws.send(JSON.stringify({
          type: 'text',
          token: reply,
          last: true
        }));
        break;
      }
      
      case 'interrupt':
        console.log('用户打断');
        break;
        
      case 'dtmf':
        console.log(`按键: ${message.digit}`);
        break;
    }
  });
  
  ws.on('close', () => console.log('通话结束'));
});
 
server.listen(3000, () => console.log('服务器运行在 :3000'));

# 使用 ngrok 暴露本地服务（开发环境）
ngrok http 3000
 
# 配置 Twilio 号码的 Webhook
twilio phone-numbers:update +14155551234 \
  --voice-url="https://your-ngrok-url.ngrok.io/incoming-call" \
  --voice-method="POST"

# 流式 LLM 回复 — 更低延迟
async def stream_llm_response(ws: WebSocket, conversation_history: list):
    """流式发送 LLM 回复，实现逐句播放"""
    stream = await openai_client.chat.completions.create(
        model="gpt-4o",
        messages=conversation_history,
        max_tokens=150,
        temperature=0.7,
        stream=True
    )
    
    full_response = ""
    buffer = ""
    
    async for chunk in stream:
        delta = chunk.choices[0].delta.content or ""
        full_response += delta
        buffer += delta
        
        # 按句子边界发送（遇到句号、问号、感叹号时发送）
        sentence_endings = ["。", "？", "！", ".", "?", "!"]
        for ending in sentence_endings:
            if ending in buffer:
                parts = buffer.split(ending, 1)
                sentence = parts[0] + ending
                buffer = parts[1] if len(parts) > 1 else ""
                
                # 发送一个句子
                await ws.send_json({
                    "type": "text",
                    "token": sentence,
                    "last": False
                })
                break
    
    # 发送剩余内容
    if buffer.strip():
        await ws.send_json({
            "type": "text",
            "token": buffer,
            "last": True
        })
    else:
        # 标记结束
        await ws.send_json({
            "type": "text",
            "token": "",
            "last": True
        })
    
    return full_response

你是通过电话与客户对话的 AI 客服代表。

## 重要：电话对话特殊规则
1. 你的回复会被 TTS 朗读，所以：
   - 不要使用 Markdown 格式（**加粗**、- 列表等）
   - 不要使用表情符号
   - 数字用中文读法（"一百二十三" 而非 "123"）
   - URL 用口语描述（"我们的官网" 而非 "www.example.com"）
2. 每次回复控制在 30 字以内，电话中长回复体验很差
3. 需要列举多项时，分多轮对话说明
4. 用"嗯"、"好的"等语气词让对话更自然

## 对话流程
[根据业务需求定制]

## 转人工触发条件
- 客户明确要求
- 连续 2 次无法理解客户意图
- 涉及投诉或敏感操作

[Start Node] → [Speak: 欢迎语] → [Collect Input: 意图识别]
                                         ↓
                    ┌────────────────────┼────────────────────┐
                    ↓                    ↓                    ↓
            [订单查询流程]        [预约安排流程]        [转人工流程]
                    ↓                    ↓                    ↓
            [Webhook: 查询API]   [Webhook: 日程API]   [Transfer Node]
                    ↓                    ↓
            [Speak: 查询结果]    [Speak: 确认预约]
                    ↓                    ↓
            [End Node]           [End Node]

┌──────────────────────────────────────────────────────────┐
│                    呼叫流生命周期                          │
│                                                          │
│  ┌─────────┐   ┌─────────┐   ┌─────────┐   ┌─────────┐ │
│  │ 接入阶段 │ → │ 识别阶段 │ → │ 处理阶段 │ → │ 结束阶段 │ │
│  │         │   │         │   │         │   │         │ │
│  │ · 问候   │   │ · 身份验证│   │ · 意图路由│   │ · 确认   │ │
│  │ · 语言选择│   │ · 意图识别│   │ · 工具调用│   │ · 满意度 │ │
│  │ · 排队等候│   │ · 上下文  │   │ · 人工转接│   │ · 告别   │ │
│  └─────────┘   └─────────┘   └─────────┘   └─────────┘ │
│       ↑                                         │       │
│       └─────────── 循环（多轮对话）──────────────┘       │
└──────────────────────────────────────────────────────────┘

问候 → 收集信息 → 处理请求 → 确认结果 → 结束

问候 → 意图识别 → ┬→ 意图A处理 → 确认 → 结束
                  ├→ 意图B处理 → 确认 → 结束
                  ├→ 意图C处理 → 确认 → 结束
                  └→ 未知意图 → 澄清/转人工

┌─────────┐     ┌──────────┐     ┌──────────┐
│ 身份验证  │ ──→ │ 需求收集   │ ──→ │ 方案推荐  │
└─────────┘     └──────────┘     └──────────┘
     ↑               ↑ ↓              ↓
     │          ┌──────────┐     ┌──────────┐
     └──────── │ 信息补充   │     │ 确认下单  │
               └──────────┘     └──────────┘

规则引擎（身份验证、敏感操作）
    ↕
LLM 自由对话（闲聊、复杂问答）
    ↕
工具调用（数据查询、操作执行）

请为 [业务场景] 设计一个 AI 电话 Agent 的呼叫流。

## 业务背景
- 公司：[公司名称和行业]
- 目标用户：[用户画像]
- 主要来电原因：[列出 3-5 个主要意图]
- 工作时间：[服务时间]

## 设计要求
1. 画出完整的呼叫流程图（使用 Mermaid 语法）
2. 每个节点标注：
   - Agent 说什么（话术脚本）
   - 期望用户回复什么
   - 超时/异常处理
3. 标注人工转接触发条件
4. 标注需要调用的外部 API/工具
5. 考虑以下边界情况：
   - 用户沉默超过 [N] 秒
   - 用户连续说"听不懂"
   - 用户情绪激动
   - 网络/系统故障

## 输出格式
- Mermaid 流程图
- 每个节点的详细话术脚本
- 异常处理策略表

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from datetime import datetime
 
class CallPhase(Enum):
    """通话阶段"""
    GREETING = "greeting"           # 问候
    AUTHENTICATION = "auth"         # 身份验证
    INTENT_DETECTION = "intent"     # 意图识别
    INFORMATION_GATHERING = "gather" # 信息收集
    PROCESSING = "processing"       # 处理中
    CONFIRMATION = "confirmation"   # 确认
    TRANSFER = "transfer"           # 转接
    CLOSING = "closing"             # 结束
 
class UserSentiment(Enum):
    """用户情绪"""
    POSITIVE = "positive"
    NEUTRAL = "neutral"
    FRUSTRATED = "frustrated"
    ANGRY = "angry"
 
@dataclass
class ConversationState:
    """对话状态数据模型"""
    # 通话元数据
    call_id: str = ""
    caller_number: str = ""
    start_time: datetime = field(default_factory=datetime.now)
    
    # 当前阶段
    phase: CallPhase = CallPhase.GREETING
    
    # 用户信息（身份验证后填充）
    user_id: Optional[str] = None
    user_name: Optional[str] = None
    is_authenticated: bool = False
    
    # 意图和上下文
    detected_intent: Optional[str] = None
    intent_confidence: float = 0.0
    collected_info: dict = field(default_factory=dict)
    
    # 对话质量追踪
    turn_count: int = 0
    clarification_count: int = 0  # 澄清次数
    sentiment: UserSentiment = UserSentiment.NEUTRAL
    
    # 转接相关
    transfer_requested: bool = False
    transfer_reason: Optional[str] = None
    
    # 工具调用记录
    tool_calls: list = field(default_factory=list)
    
    def should_transfer_to_human(self) -> bool:
        """判断是否应该转人工"""
        return (
            self.transfer_requested
            or self.clarification_count >= 3
            or self.sentiment == UserSentiment.ANGRY
            or self.turn_count >= 20
        )
    
    def advance_phase(self, next_phase: CallPhase):
        """推进对话阶段"""
        self.phase = next_phase
        self.turn_count += 1

# conversation_fsm.py — 对话状态机
from typing import Callable, Optional
from dataclasses import dataclass
 
@dataclass
class Transition:
    """状态转换"""
    trigger: str           # 触发条件
    source: str            # 源状态
    dest: str              # 目标状态
    condition: Optional[Callable] = None  # 守卫条件
    action: Optional[Callable] = None     # 转换动作
 
class ConversationFSM:
    """对话有限状态机"""
    
    def __init__(self):
        self.current_state = "greeting"
        self.context = ConversationState()
        self.transitions = self._define_transitions()
        self.state_handlers = self._define_state_handlers()
    
    def _define_transitions(self) -> list[Transition]:
        """定义状态转换规则"""
        return [
            # 问候 → 身份验证
            Transition(
                trigger="greeting_done",
                source="greeting",
                dest="authentication"
            ),
            # 身份验证 → 意图识别（验证成功）
            Transition(
                trigger="auth_success",
                source="authentication",
                dest="intent_detection",
                condition=lambda ctx: ctx.is_authenticated
            ),
            # 身份验证 → 意图识别（跳过验证）
            Transition(
                trigger="auth_skip",
                source="authentication",
                dest="intent_detection"
            ),
            # 身份验证失败 → 重试或转人工
            Transition(
                trigger="auth_failed",
                source="authentication",
                dest="transfer",
                condition=lambda ctx: ctx.clarification_count >= 3
            ),
            # 意图识别 → 信息收集
            Transition(
                trigger="intent_detected",
                source="intent_detection",
                dest="information_gathering",
                condition=lambda ctx: ctx.intent_confidence > 0.7
            ),
            # 意图不明确 → 澄清（留在当前状态）
            Transition(
                trigger="intent_unclear",
                source="intent_detection",
                dest="intent_detection",
                action=lambda ctx: setattr(
                    ctx, 'clarification_count', 
                    ctx.clarification_count + 1
                )
            ),
            # 信息收集完成 → 处理
            Transition(
                trigger="info_complete",
                source="information_gathering",
                dest="processing"
            ),
            # 处理完成 → 确认
            Transition(
                trigger="process_done",
                source="processing",
                dest="confirmation"
            ),
            # 确认 → 结束
            Transition(
                trigger="confirmed",
                source="confirmation",
                dest="closing"
            ),
            # 确认 → 修改（回到信息收集）
            Transition(
                trigger="needs_modification",
                source="confirmation",
                dest="information_gathering"
            ),
            # 任意状态 → 转人工（全局转换）
            Transition(
                trigger="transfer_requested",
                source="*",  # 任意状态
                dest="transfer"
            ),
            # 任意状态 → 结束（用户挂断）
            Transition(
                trigger="hangup",
                source="*",
                dest="closing"
            ),
        ]
    
    def _define_state_handlers(self) -> dict:
        """定义每个状态的处理逻辑"""
        return {
            "greeting": self._handle_greeting,
            "authentication": self._handle_authentication,
            "intent_detection": self._handle_intent_detection,
            "information_gathering": self._handle_info_gathering,
            "processing": self._handle_processing,
            "confirmation": self._handle_confirmation,
            "transfer": self._handle_transfer,
            "closing": self._handle_closing,
        }
    
    def trigger(self, event: str) -> bool:
        """触发状态转换"""
        for t in self.transitions:
            if t.trigger == event and (
                t.source == "*" or t.source == self.current_state
            ):
                # 检查守卫条件
                if t.condition and not t.condition(self.context):
                    continue
                
                # 执行转换动作
                if t.action:
                    t.action(self.context)
                
                # 切换状态
                old_state = self.current_state
                self.current_state = t.dest
                print(f"状态转换: {old_state} → {self.current_state}")
                return True
        
        return False
    
    async def process_input(self, user_input: str) -> str:
        """处理用户输入，返回 Agent 回复"""
        handler = self.state_handlers.get(self.current_state)
        if handler:
            return await handler(user_input)
        return "抱歉，系统出现了问题，我帮您转接人工客服。"
    
    # ========== 状态处理器 ==========
    
    async def _handle_greeting(self, user_input: str) -> str:
        self.trigger("greeting_done")
        return "请问您贵姓？我需要验证一下您的身份。"
    
    async def _handle_authentication(self, user_input: str) -> str:
        # 简化的身份验证逻辑
        if self._extract_name(user_input):
            self.context.is_authenticated = True
            self.trigger("auth_success")
            return f"{self.context.user_name}您好，请问今天有什么可以帮您的？"
        else:
            self.context.clarification_count += 1
            if self.context.clarification_count >= 3:
                self.trigger("auth_failed")
                return "验证遇到困难，我帮您转接人工客服。"
            return "抱歉没听清，请再说一次您的姓名。"
    
    async def _handle_intent_detection(self, user_input: str) -> str:
        intent, confidence = await self._classify_intent(user_input)
        self.context.detected_intent = intent
        self.context.intent_confidence = confidence
        
        if confidence > 0.7:
            self.trigger("intent_detected")
            return self._get_info_gathering_prompt(intent)
        else:
            self.trigger("intent_unclear")
            return "抱歉，我没太理解您的意思，能再详细说一下吗？"
    
    async def _handle_info_gathering(self, user_input: str) -> str:
        # 根据意图收集所需信息
        missing = self._get_missing_info()
        if not missing:
            self.trigger("info_complete")
            return await self._handle_processing(user_input)
        return f"好的，还需要确认一下您的{missing[0]}。"
    
    async def _handle_processing(self, user_input: str) -> str:
        # 调用业务 API 处理请求
        result = await self._execute_business_logic()
        self.trigger("process_done")
        return f"已经为您处理好了。{result} 请问信息正确吗？"
    
    async def _handle_confirmation(self, user_input: str) -> str:
        if self._is_affirmative(user_input):
            self.trigger("confirmed")
            return "好的，还有其他需要帮助的吗？"
        else:
            self.trigger("needs_modification")
            return "好的，请告诉我需要修改什么。"
    
    async def _handle_transfer(self, user_input: str) -> str:
        return "正在为您转接人工客服，请稍候。"
    
    async def _handle_closing(self, user_input: str) -> str:
        return "感谢您的来电，祝您生活愉快，再见！"
    
    # ========== 辅助方法 ==========
    
    def _extract_name(self, text: str) -> bool:
        """从文本中提取姓名（简化实现）"""
        # 实际项目中使用 NER 或 LLM 提取
        if len(text) >= 2:
            self.context.user_name = text.strip()
            return True
        return False
    
    async def _classify_intent(self, text: str) -> tuple[str, float]:
        """意图分类（简化实现）"""
        # 实际项目中使用 LLM 或专门的 NLU 模型
        intent_keywords = {
            "order_query": ["订单", "快递", "物流", "发货"],
            "appointment": ["预约", "挂号", "安排", "时间"],
            "complaint": ["投诉", "不满", "退款", "差评"],
        }
        for intent, keywords in intent_keywords.items():
            if any(kw in text for kw in keywords):
                return intent, 0.85
        return "unknown", 0.3
    
    def _get_missing_info(self) -> list:
        """获取还缺少的信息字段"""
        required = {"order_query": ["order_id"], "appointment": ["date", "time"]}
        needed = required.get(self.context.detected_intent, [])
        return [f for f in needed if f not in self.context.collected_info]
    
    async def _execute_business_logic(self) -> str:
        """执行业务逻辑（简化实现）"""
        return "您的订单已确认"
    
    def _is_affirmative(self, text: str) -> bool:
        """判断是否为肯定回复"""
        affirmative = ["是", "对", "好", "没问题", "正确", "可以", "嗯"]
        return any(word in text for word in affirmative)
    
    def _get_info_gathering_prompt(self, intent: str) -> str:
        """根据意图返回信息收集提示"""
        prompts = {
            "order_query": "好的，请告诉我您的订单号。",
            "appointment": "好的，您想预约什么时间？",
            "complaint": "很抱歉给您带来不好的体验，能详细说说发生了什么吗？",
        }
        return prompts.get(intent, "好的，请详细说说您的需求。")

// conversationFSM.ts — TypeScript 版对话状态机
import { z } from 'zod';
 
// 状态定义
type CallState = 
  | 'greeting' 
  | 'authentication' 
  | 'intent_detection'
  | 'info_gathering' 
  | 'processing' 
  | 'confirmation'
  | 'transfer' 
  | 'closing';
 
// 事件定义
type CallEvent = 
  | 'greeting_done'
  | 'auth_success' 
  | 'auth_failed'
  | 'intent_detected'
  | 'intent_unclear'
  | 'info_complete'
  | 'process_done'
  | 'confirmed'
  | 'needs_modification'
  | 'transfer_requested'
  | 'hangup';
 
// 对话上下文
interface ConversationContext {
  callId: string;
  callerNumber: string;
  userName?: string;
  isAuthenticated: boolean;
  detectedIntent?: string;
  intentConfidence: number;
  collectedInfo: Record<string, string>;
  turnCount: number;
  clarificationCount: number;
  sentiment: 'positive' | 'neutral' | 'frustrated' | 'angry';
}
 
// 状态转换表
const transitionTable: Record<string, Partial<Record<CallEvent, CallState>>> = {
  greeting: {
    greeting_done: 'authentication',
    transfer_requested: 'transfer',
    hangup: 'closing'
  },
  authentication: {
    auth_success: 'intent_detection',
    auth_failed: 'transfer',
    transfer_requested: 'transfer',
    hangup: 'closing'
  },
  intent_detection: {
    intent_detected: 'info_gathering',
    intent_unclear: 'intent_detection',
    transfer_requested: 'transfer',
    hangup: 'closing'
  },
  info_gathering: {
    info_complete: 'processing',
    transfer_requested: 'transfer',
    hangup: 'closing'
  },
  processing: {
    process_done: 'confirmation',
    transfer_requested: 'transfer',
    hangup: 'closing'
  },
  confirmation: {
    confirmed: 'closing',
    needs_modification: 'info_gathering',
    transfer_requested: 'transfer',
    hangup: 'closing'
  }
};
 
class ConversationStateMachine {
  private state: CallState = 'greeting';
  private context: ConversationContext;
  
  constructor(callId: string, callerNumber: string) {
    this.context = {
      callId,
      callerNumber,
      isAuthenticated: false,
      intentConfidence: 0,
      collectedInfo: {},
      turnCount: 0,
      clarificationCount: 0,
      sentiment: 'neutral'
    };
  }
  
  getState(): CallState {
    return this.state;
  }
  
  getContext(): ConversationContext {
    return { ...this.context };
  }
  
  transition(event: CallEvent): boolean {
    const stateTransitions = transitionTable[this.state];
    if (!stateTransitions) return false;
    
    const nextState = stateTransitions[event];
    if (!nextState) return false;
    
    const prevState = this.state;
    this.state = nextState;
    this.context.turnCount++;
    
    console.log(`[FSM] ${prevState} --${event}--> ${this.state}`);
    return true;
  }
  
  shouldTransferToHuman(): boolean {
    return (
      this.context.clarificationCount >= 3 ||
      this.context.sentiment === 'angry' ||
      this.context.turnCount >= 20
    );
  }
  
  async processInput(userInput: string): Promise<string> {
    const handlers: Record<CallState, (input: string) => Promise<string>> = {
      greeting: async () => {
        this.transition('greeting_done');
        return '请问您贵姓？我需要验证一下您的身份。';
      },
      authentication: async (input) => {
        if (input.length >= 2) {
          this.context.userName = input.trim();
          this.context.isAuthenticated = true;
          this.transition('auth_success');
          return `${this.context.userName}您好，请问有什么可以帮您？`;
        }
        this.context.clarificationCount++;
        if (this.shouldTransferToHuman()) {
          this.transition('auth_failed');
          return '验证遇到困难，我帮您转接人工客服。';
        }
        return '抱歉没听清，请再说一次您的姓名。';
      },
      intent_detection: async (input) => {
        const { intent, confidence } = classifyIntent(input);
        this.context.detectedIntent = intent;
        this.context.intentConfidence = confidence;
        
        if (confidence > 0.7) {
          this.transition('intent_detected');
          return getInfoPrompt(intent);
        }
        this.context.clarificationCount++;
        this.transition('intent_unclear');
        return '抱歉，我没太理解，能再详细说一下吗？';
      },
      info_gathering: async (input) => {
        // 收集信息逻辑
        this.transition('info_complete');
        return '好的，正在为您处理...';
      },
      processing: async () => {
        this.transition('process_done');
        return '已经处理好了，请确认信息是否正确？';
      },
      confirmation: async (input) => {
        if (['是', '对', '好', '没问题'].some(w => input.includes(w))) {
          this.transition('confirmed');
          return '感谢您的来电，祝您生活愉快，再见！';
        }
        this.transition('needs_modification');
        return '好的，请告诉我需要修改什么。';
      },
      transfer: async () => '正在为您转接人工客服，请稍候。',
      closing: async () => '感谢您的来电，再见！'
    };
    
    const handler = handlers[this.state];
    return handler(userInput);
  }
}
 
// 辅助函数
function classifyIntent(text: string): { intent: string; confidence: number } {
  const intentMap: Record<string, string[]> = {
    order_query: ['订单', '快递', '物流'],
    appointment: ['预约', '挂号', '安排'],
    complaint: ['投诉', '退款', '不满']
  };
  
  for (const [intent, keywords] of Object.entries(intentMap)) {
    if (keywords.some(kw => text.includes(kw))) {
      return { intent, confidence: 0.85 };
    }
  }
  return { intent: 'unknown', confidence: 0.3 };
}
 
function getInfoPrompt(intent: string): string {
  const prompts: Record<string, string> = {
    order_query: '好的，请告诉我您的订单号。',
    appointment: '好的，您想预约什么时间？',
    complaint: '很抱歉给您带来不好的体验，能详细说说吗？'
  };
  return prompts[intent] || '好的，请详细说说您的需求。';
}

# state_store.py — Redis 状态持久化
import redis
import json
from datetime import timedelta
 
class ConversationStateStore:
    """基于 Redis 的对话状态存储"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url)
        self.ttl = timedelta(hours=1)  # 状态过期时间
    
    def save(self, call_id: str, state: ConversationState):
        """保存对话状态"""
        key = f"call_state:{call_id}"
        data = {
            "current_state": state.phase.value,
            "user_id": state.user_id,
            "user_name": state.user_name,
            "is_authenticated": state.is_authenticated,
            "detected_intent": state.detected_intent,
            "collected_info": state.collected_info,
            "turn_count": state.turn_count,
            "clarification_count": state.clarification_count,
            "sentiment": state.sentiment.value,
            "tool_calls": state.tool_calls,
        }
        self.redis.setex(key, self.ttl, json.dumps(data))
    
    def load(self, call_id: str) -> ConversationState | None:
        """加载对话状态"""
        key = f"call_state:{call_id}"
        data = self.redis.get(key)
        if not data:
            return None
        
        parsed = json.loads(data)
        state = ConversationState()
        state.call_id = call_id
        state.phase = CallPhase(parsed["current_state"])
        state.user_id = parsed.get("user_id")
        state.user_name = parsed.get("user_name")
        state.is_authenticated = parsed.get("is_authenticated", False)
        state.detected_intent = parsed.get("detected_intent")
        state.collected_info = parsed.get("collected_info", {})
        state.turn_count = parsed.get("turn_count", 0)
        state.clarification_count = parsed.get("clarification_count", 0)
        state.sentiment = UserSentiment(parsed.get("sentiment", "neutral"))
        return state
    
    def delete(self, call_id: str):
        """删除对话状态（通话结束后）"""
        self.redis.delete(f"call_state:{call_id}")

# 方式 1：冷转接（直接转接，不带上下文）
async def cold_transfer(ws: WebSocket, target_number: str):
    """冷转接 — 直接将通话转给人工"""
    await ws.send_json({
        "type": "end",
        "handoffData": json.dumps({
            "reasonCode": "customer_request",
            "conversationSummary": "客户咨询订单问题，需要人工处理"
        })
    })
    # ConversationRelay 结束后，使用 Twilio REST API 转接
    # 或在 TwiML 中配置 action URL 处理后续
 
# 方式 2：温转接（带上下文摘要）
async def warm_transfer(ws: WebSocket, state: ConversationState):
    """温转接 — 生成对话摘要后转接"""
    # 先生成对话摘要
    summary = await generate_conversation_summary(state)
    
    await ws.send_json({
        "type": "text",
        "token": "我现在帮您转接专业同事，已经把您的情况告诉他了，请稍候。",
        "last": True
    })
    
    # 将摘要存储，供人工客服查看
    await save_handoff_context(state.call_id, {
        "customer_name": state.user_name,
        "intent": state.detected_intent,
        "collected_info": state.collected_info,
        "summary": summary,
        "sentiment": state.sentiment.value,
        "turn_count": state.turn_count
    })
 
async def generate_conversation_summary(state: ConversationState) -> str:
    """使用 LLM 生成对话摘要"""
    response = await openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "用一段话总结以下客服对话的关键信息。"},
            {"role": "user", "content": f"""
客户：{state.user_name}
意图：{state.detected_intent}
已收集信息：{json.dumps(state.collected_info, ensure_ascii=False)}
对话轮次：{state.turn_count}
客户情绪：{state.sentiment.value}
"""}
        ],
        max_tokens=100
    )
    return response.choices[0].message.content

# error_handling.py — 电话 Agent 错误处理
import asyncio
from enum import Enum
 
class ErrorType(Enum):
    STT_FAILURE = "stt_failure"         # 语音识别失败
    LLM_TIMEOUT = "llm_timeout"         # LLM 响应超时
    LLM_ERROR = "llm_error"             # LLM 返回错误
    TTS_FAILURE = "tts_failure"         # 语音合成失败
    TOOL_FAILURE = "tool_failure"       # 工具调用失败
    WEBSOCKET_ERROR = "ws_error"        # WebSocket 连接错误
    SILENCE_TIMEOUT = "silence_timeout" # 用户长时间沉默
 
class ErrorHandler:
    """电话 Agent 错误处理器"""
    
    def __init__(self):
        self.error_counts: dict[str, int] = {}
        self.max_retries = 2
    
    async def handle(
        self, 
        error_type: ErrorType, 
        ws, 
        state: ConversationState
    ) -> str:
        """统一错误处理入口"""
        key = f"{state.call_id}:{error_type.value}"
        self.error_counts[key] = self.error_counts.get(key, 0) + 1
        count = self.error_counts[key]
        
        handlers = {
            ErrorType.STT_FAILURE: self._handle_stt_failure,
            ErrorType.LLM_TIMEOUT: self._handle_llm_timeout,
            ErrorType.LLM_ERROR: self._handle_llm_error,
            ErrorType.TTS_FAILURE: self._handle_tts_failure,
            ErrorType.TOOL_FAILURE: self._handle_tool_failure,
            ErrorType.SILENCE_TIMEOUT: self._handle_silence,
        }
        
        handler = handlers.get(error_type, self._handle_generic)
        return await handler(count, state)
    
    async def _handle_stt_failure(self, count: int, state) -> str:
        if count <= 2:
            return "抱歉，我没有听清，能请您再说一遍吗？"
        return "通话质量不太好，我帮您转接人工客服。"
    
    async def _handle_llm_timeout(self, count: int, state) -> str:
        if count == 1:
            return "请稍等，我正在查询中..."
        return "系统响应较慢，我帮您转接人工客服处理。"
    
    async def _handle_llm_error(self, count: int, state) -> str:
        return "系统遇到了一点问题，我帮您转接人工客服。"
    
    async def _handle_tts_failure(self, count: int, state) -> str:
        # TTS 失败时，尝试使用备用 TTS 或 DTMF 提示
        return "（系统语音异常，尝试备用方案）"
    
    async def _handle_tool_failure(self, count: int, state) -> str:
        if count == 1:
            return "查询系统暂时繁忙，我再试一次。"
        return "系统暂时无法查询，我帮您记录下来，稍后会有同事回复您。"
    
    async def _handle_silence(self, count: int, state) -> str:
        if count == 1:
            return "您好，请问还在吗？"
        elif count == 2:
            return "如果没有其他问题，我就先挂断了。祝您生活愉快！"
        return ""  # 空字符串表示挂断
    
    async def _handle_generic(self, count: int, state) -> str:
        return "抱歉出了点问题，我帮您转接人工客服。"

# call_monitor.py — 通话质量监控
from dataclasses import dataclass, field
from datetime import datetime
import statistics
 
@dataclass
class CallMetrics:
    """单次通话质量指标"""
    call_id: str
    start_time: datetime = field(default_factory=datetime.now)
    end_time: datetime | None = None
    
    # 延迟指标
    stt_latencies: list[float] = field(default_factory=list)   # STT 延迟（ms）
    llm_latencies: list[float] = field(default_factory=list)   # LLM 延迟（ms）
    tts_latencies: list[float] = field(default_factory=list)   # TTS 延迟（ms）
    e2e_latencies: list[float] = field(default_factory=list)   # 端到端延迟（ms）
    
    # 对话质量
    total_turns: int = 0
    successful_turns: int = 0
    interruptions: int = 0          # 用户打断次数
    silence_timeouts: int = 0       # 沉默超时次数
    stt_errors: int = 0             # STT 识别错误
    
    # 结果
    resolution: str = "unknown"     # resolved / transferred / abandoned
    transfer_reason: str | None = None
    
    def get_summary(self) -> dict:
        """生成通话质量摘要"""
        duration = (self.end_time - self.start_time).total_seconds() if self.end_time else 0
        
        return {
            "call_id": self.call_id,
            "duration_seconds": round(duration, 1),
            "total_turns": self.total_turns,
            "avg_e2e_latency_ms": round(
                statistics.mean(self.e2e_latencies), 0
            ) if self.e2e_latencies else 0,
            "p95_e2e_latency_ms": round(
                sorted(self.e2e_latencies)[int(len(self.e2e_latencies) * 0.95)]
                if self.e2e_latencies else 0, 0
            ),
            "interruption_rate": round(
                self.interruptions / max(self.total_turns, 1), 2
            ),
            "success_rate": round(
                self.successful_turns / max(self.total_turns, 1), 2
            ),
            "resolution": self.resolution,
        }
 
class CallQualityMonitor:
    """通话质量监控器"""
    
    def __init__(self):
        self.active_calls: dict[str, CallMetrics] = {}
        self.alert_thresholds = {
            "e2e_latency_ms": 2000,     # 端到端延迟超过 2 秒告警
            "stt_error_rate": 0.2,       # STT 错误率超过 20% 告警
            "transfer_rate": 0.3,        # 转人工率超过 30% 告警
        }
    
    def start_call(self, call_id: str) -> CallMetrics:
        metrics = CallMetrics(call_id=call_id)
        self.active_calls[call_id] = metrics
        return metrics
    
    def record_turn(
        self, call_id: str, 
        stt_ms: float, llm_ms: float, tts_ms: float,
        success: bool = True
    ):
        """记录一轮对话的延迟"""
        metrics = self.active_calls.get(call_id)
        if not metrics:
            return
        
        metrics.stt_latencies.append(stt_ms)
        metrics.llm_latencies.append(llm_ms)
        metrics.tts_latencies.append(tts_ms)
        metrics.e2e_latencies.append(stt_ms + llm_ms + tts_ms)
        metrics.total_turns += 1
        if success:
            metrics.successful_turns += 1
        
        # 检查是否需要告警
        e2e = stt_ms + llm_ms + tts_ms
        if e2e > self.alert_thresholds["e2e_latency_ms"]:
            self._alert(call_id, f"端到端延迟过高: {e2e:.0f}ms")
    
    def end_call(self, call_id: str, resolution: str):
        """结束通话，生成报告"""
        metrics = self.active_calls.get(call_id)
        if metrics:
            metrics.end_time = datetime.now()
            metrics.resolution = resolution
            summary = metrics.get_summary()
            # 发送到监控系统（Prometheus/Grafana/自定义）
            self._export_metrics(summary)
            del self.active_calls[call_id]
            return summary
    
    def _alert(self, call_id: str, message: str):
        """发送告警"""
        print(f"⚠️ 告警 [{call_id}]: {message}")
    
    def _export_metrics(self, summary: dict):
        """导出指标到监控系统"""
        print(f"📊 通话报告: {summary}")

传统流水线：语音 → STT (150ms) → LLM (500ms) → TTS (100ms) → 语音
                                总延迟: ~750ms+

Realtime API：语音 → OpenAI Realtime (端到端) → 语音
                        总延迟: ~300-500ms

# realtime_server.py — Twilio + OpenAI Realtime API
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import Response
import websockets
import asyncio
import json
import base64
 
app = FastAPI()
 
OPENAI_API_KEY = "sk-xxxxxxxx"
OPENAI_REALTIME_URL = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
 
SYSTEM_PROMPT = """你是一位专业的电话客服代表。
- 使用简洁的口语化中文
- 每次回复 1-3 句话
- 语气友好专业"""
 
@app.post("/incoming-call")
async def handle_incoming_call(request: Request):
    """来电 Webhook — 使用 Twilio Media Streams"""
    response = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
    <Say language="zh-CN">请稍等，正在为您接通智能客服。</Say>
    <Connect>
        <Stream url="wss://{request.headers['host']}/media-stream" />
    </Connect>
</Response>"""
    return Response(content=response, media_type="application/xml")
 
@app.websocket("/media-stream")
async def media_stream_handler(twilio_ws: WebSocket):
    """桥接 Twilio Media Streams 和 OpenAI Realtime API"""
    await twilio_ws.accept()
    
    stream_sid = None
    
    # 连接 OpenAI Realtime API
    async with websockets.connect(
        OPENAI_REALTIME_URL,
        additional_headers={
            "Authorization": f"Bearer {OPENAI_API_KEY}",
            "OpenAI-Beta": "realtime=v1"
        }
    ) as openai_ws:
        
        # 配置 OpenAI 会话
        await openai_ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "turn_detection": {"type": "server_vad"},
                "input_audio_format": "g711_ulaw",
                "output_audio_format": "g711_ulaw",
                "voice": "alloy",
                "instructions": SYSTEM_PROMPT,
                "modalities": ["text", "audio"],
                "temperature": 0.7
            }
        }))
        
        async def forward_twilio_to_openai():
            """Twilio 音频 → OpenAI"""
            try:
                while True:
                    data = await twilio_ws.receive_text()
                    message = json.loads(data)
                    
                    if message["event"] == "start":
                        nonlocal stream_sid
                        stream_sid = message["start"]["streamSid"]
                        
                    elif message["event"] == "media":
                        # 转发音频到 OpenAI
                        audio_data = message["media"]["payload"]
                        await openai_ws.send(json.dumps({
                            "type": "input_audio_buffer.append",
                            "audio": audio_data
                        }))
                        
                    elif message["event"] == "stop":
                        break
                        
            except WebSocketDisconnect:
                pass
        
        async def forward_openai_to_twilio():
            """OpenAI 音频 → Twilio"""
            try:
                async for message in openai_ws:
                    data = json.loads(message)
                    
                    if data["type"] == "response.audio.delta":
                        # 转发音频到 Twilio
                        audio_delta = data["delta"]
                        await twilio_ws.send_json({
                            "event": "media",
                            "streamSid": stream_sid,
                            "media": {
                                "payload": audio_delta
                            }
                        })
                    
                    elif data["type"] == "response.audio_transcript.done":
                        # Agent 回复的文本转录
                        print(f"Agent: {data.get('transcript', '')}")
                    
                    elif data["type"] == "conversation.item.input_audio_transcription.completed":
                        # 用户语音的文本转录
                        print(f"用户: {data.get('transcript', '')}")
                        
            except Exception as e:
                print(f"OpenAI 连接错误: {e}")
        
        # 并行运行双向转发
        await asyncio.gather(
            forward_twilio_to_openai(),
            forward_openai_to_twilio()
        )

来电 → Twilio → ConversationRelay → WebSocket 服务器
                                         ↓
                                    对话状态机
                                    ↓         ↓
                              GPT-4o 推理   工具调用
                                              ↓
                                    ┌─────────┼─────────┐
                                    ↓         ↓         ↓
                              查询预约    创建预约    取消预约
                              (CRM API)  (日程 API)  (CRM API)

# dental_agent.py — 牙科诊所预约 Agent
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import Response
from openai import AsyncOpenAI
import json
from datetime import datetime, timedelta
 
app = FastAPI()
openai_client = AsyncOpenAI()
 
# ========== 诊所信息 ==========
CLINIC_INFO = {
    "name": "阳光牙科诊所",
    "address": "北京市朝阳区建国路 88 号",
    "hours": "周一至周六 9:00-18:00，周日休息",
    "phone": "010-12345678",
    "services": ["洗牙", "补牙", "拔牙", "正畸", "种植牙", "美白"]
}
 
# ========== 工具定义 ==========
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "check_available_slots",
            "description": "查询指定日期的可用预约时段",
            "parameters": {
                "type": "object",
                "properties": {
                    "date": {
                        "type": "string",
                        "description": "日期，格式 YYYY-MM-DD"
                    },
                    "service": {
                        "type": "string",
                        "description": "服务类型，如洗牙、补牙等"
                    }
                },
                "required": ["date"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "create_appointment",
            "description": "创建新预约",
            "parameters": {
                "type": "object",
                "properties": {
                    "patient_name": {"type": "string", "description": "患者姓名"},
                    "phone": {"type": "string", "description": "联系电话"},
                    "date": {"type": "string", "description": "预约日期 YYYY-MM-DD"},
                    "time": {"type": "string", "description": "预约时间 HH:MM"},
                    "service": {"type": "string", "description": "服务类型"}
                },
                "required": ["patient_name", "phone", "date", "time", "service"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "query_appointment",
            "description": "根据姓名或电话查询已有预约",
            "parameters": {
                "type": "object",
                "properties": {
                    "patient_name": {"type": "string"},
                    "phone": {"type": "string"}
                }
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "cancel_appointment",
            "description": "取消预约",
            "parameters": {
                "type": "object",
                "properties": {
                    "appointment_id": {"type": "string", "description": "预约编号"}
                },
                "required": ["appointment_id"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "transfer_to_human",
            "description": "转接人工前台",
            "parameters": {
                "type": "object",
                "properties": {
                    "reason": {"type": "string", "description": "转接原因"}
                },
                "required": ["reason"]
            }
        }
    }
]
 
SYSTEM_PROMPT = f"""你是{CLINIC_INFO['name']}的 AI 预约助手，通过电话与患者对话。
 
## 诊所信息
- 地址：{CLINIC_INFO['address']}
- 营业时间：{CLINIC_INFO['hours']}
- 服务项目：{', '.join(CLINIC_INFO['services'])}
 
## 对话规则
1. 回复简洁，每次 1-3 句话（电话场景）
2. 使用口语化中文，不用 Markdown 格式
3. 数字用中文表达（"下午两点" 而非 "14:00"）
4. 主动引导对话，减少用户思考负担
5. 涉及取消预约时，必须二次确认
 
## 对话流程
1. 问候并询问需求
2. 根据需求调用相应工具
3. 确认操作结果
4. 询问是否还有其他需要
 
## 转人工条件
- 患者明确要求
- 涉及费用咨询（你不知道具体价格）
- 紧急牙痛需要当天加号
- 连续 2 次无法理解患者需求"""
 
# ========== 工具执行 ==========
async def execute_tool(name: str, args: dict) -> str:
    """执行工具调用（模拟实现）"""
    if name == "check_available_slots":
        date = args.get("date", "")
        return json.dumps({
            "date": date,
            "available_slots": [
                {"time": "09:00", "doctor": "王医生"},
                {"time": "10:30", "doctor": "李医生"},
                {"time": "14:00", "doctor": "王医生"},
                {"time": "15:30", "doctor": "李医生"},
            ]
        }, ensure_ascii=False)
    
    elif name == "create_appointment":
        return json.dumps({
            "success": True,
            "appointment_id": "APT-20250715-001",
            "message": f"预约成功：{args['patient_name']}，"
                       f"{args['date']} {args['time']}，{args['service']}",
        }, ensure_ascii=False)
    
    elif name == "query_appointment":
        return json.dumps({
            "appointments": [
                {
                    "id": "APT-20250710-003",
                    "date": "2025-07-20",
                    "time": "10:30",
                    "service": "洗牙",
                    "doctor": "王医生",
                    "status": "已确认"
                }
            ]
        }, ensure_ascii=False)
    
    elif name == "cancel_appointment":
        return json.dumps({
            "success": True,
            "message": f"预约 {args['appointment_id']} 已取消"
        }, ensure_ascii=False)
    
    elif name == "transfer_to_human":
        return json.dumps({
            "transferred": True,
            "message": "正在转接前台"
        }, ensure_ascii=False)
    
    return json.dumps({"error": "未知工具"})
 
# ========== 来电处理 ==========
@app.post("/incoming-call")
async def handle_incoming_call(request: Request):
    response = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
    <Connect>
        <ConversationRelay
            url="wss://{request.headers['host']}/ws"
            welcomeGreeting="您好，这里是{CLINIC_INFO['name']}，我是 AI 预约助手，请问有什么可以帮您？"
            ttsProvider="ElevenLabs"
            voice="21m00Tcm4TlvDq8ikWAM"
            transcriptionProvider="Deepgram"
            speechModel="nova-3-general"
            language="zh-CN"
            interruptible="true"
        />
    </Connect>
</Response>"""
    return Response(content=response, media_type="application/xml")
 
@app.websocket("/ws")
async def websocket_handler(ws: WebSocket):
    await ws.accept()
    
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    
    try:
        while True:
            data = await ws.receive_text()
            message = json.loads(data)
            
            if message.get("type") == "prompt":
                user_text = message.get("voicePrompt", "").strip()
                if not user_text:
                    continue
                
                messages.append({"role": "user", "content": user_text})
                
                # 调用 LLM（带工具）
                response = await openai_client.chat.completions.create(
                    model="gpt-4o",
                    messages=messages,
                    tools=TOOLS,
                    tool_choice="auto",
                    max_tokens=200,
                    temperature=0.7
                )
                
                assistant_msg = response.choices[0].message
                messages.append(assistant_msg.model_dump())
                
                # 处理工具调用
                if assistant_msg.tool_calls:
                    for tool_call in assistant_msg.tool_calls:
                        func_name = tool_call.function.name
                        func_args = json.loads(tool_call.function.arguments)
                        
                        # 执行工具
                        result = await execute_tool(func_name, func_args)
                        
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tool_call.id,
                            "content": result
                        })
                    
                    # 让 LLM 根据工具结果生成回复
                    follow_up = await openai_client.chat.completions.create(
                        model="gpt-4o",
                        messages=messages,
                        max_tokens=200,
                        temperature=0.7
                    )
                    
                    reply = follow_up.choices[0].message.content
                    messages.append({"role": "assistant", "content": reply})
                else:
                    reply = assistant_msg.content
                
                # 发送回复
                await ws.send_json({
                    "type": "text",
                    "token": reply,
                    "last": True
                })
                
    except WebSocketDisconnect:
        print("通话结束")