14d - 语音 Agent 用例

┌─────────────────────────────────────────────────────────────────┐
│                    语音 Agent 行业应用地图                        │
├──────────────┬──────────────┬──────────────┬───────────────────┤
│   客户服务    │    销售营销   │   医疗健康    │    开发者工具      │
├──────────────┼──────────────┼──────────────┼───────────────────┤
│ · 智能客服热线│ · 外呼销售    │ · 预约挂号    │ · 语音编程        │
│ · 订单查询    │ · 线索筛选    │ · 用药提醒    │ · 语音 DevOps     │
│ · 投诉处理    │ · 预约确认    │ · 随访回访    │ · 语音数据查询    │
│ · 技术支持    │ · 满意度调查  │ · 健康咨询    │ · 语音代码审查    │
│ · FAQ 自动应答│ · 催收提醒    │ · 报告解读    │ · 语音 CI/CD 触发 │
├──────────────┼──────────────┼──────────────┼───────────────────┤
│   金融保险    │    房地产     │   教育培训    │    餐饮零售       │
├──────────────┼──────────────┼──────────────┼───────────────────┤
│ · 账户查询    │ · 房源推荐    │ · 课程咨询    │ · 订餐/外卖      │
│ · 理赔报案    │ · 看房预约    │ · 学习辅导    │ · 预订管理        │
│ · 贷款咨询    │ · 租约提醒    │ · 考试提醒    │ · 会员服务        │
│ · 风险提醒    │ · 客户跟进    │ · 家长沟通    │ · 库存查询        │
└──────────────┴──────────────┴──────────────┴───────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                    智能客服语音 Agent 架构                            │
│                                                                     │
│  ┌──────────┐    ┌──────────────────────────────────────────────┐   │
│  │ 客户来电  │───→│              Vapi.ai 编排层                   │   │
│  │ (PSTN)   │    │                                              │   │
│  └──────────┘    │  ┌────────┐  ┌────────┐  ┌────────────────┐ │   │
│                  │  │Deepgram│→ │GPT-4o  │→ │ ElevenLabs     │ │   │
│                  │  │Nova-3  │  │/Claude │  │ Flash v2.5     │ │   │
│                  │  │(STT)   │  │(LLM)   │  │ (TTS)          │ │   │
│                  │  └────────┘  └───┬────┘  └────────────────┘ │   │
│                  │                  │ Function Calling           │   │
│                  └──────────────────┼───────────────────────────┘   │
│                                     │                               │
│                    ┌────────────────┼────────────────┐              │
│                    ↓                ↓                ↓              │
│              ┌──────────┐    ┌──────────┐    ┌──────────┐          │
│              │ 订单系统  │    │ 物流 API  │    │ 知识库    │          │
│              │ (CRM/ERP)│    │ (快递100) │    │ (RAG)    │          │
│              └──────────┘    └──────────┘    └──────────┘          │
│                                     │                               │
│                    ┌────────────────┼────────────────┐              │
│                    ↓                ↓                ↓              │
│              ┌──────────┐    ┌──────────┐    ┌──────────┐          │
│              │ 人工转接  │    │ 工单系统  │    │ 数据分析  │          │
│              │ (Twilio)  │    │ (Zendesk)│    │ (日报表)  │          │
│              └──────────┘    └──────────┘    └──────────┘          │
└─────────────────────────────────────────────────────────────────────┘

import requests
import os
 
VAPI_API_KEY = os.getenv("VAPI_API_KEY")
 
def create_customer_service_agent():
    """创建电商客服语音 Agent"""
    response = requests.post(
        "https://api.vapi.ai/assistant",
        headers={
            "Authorization": f"Bearer {VAPI_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "name": "电商智能客服",
            "model": {
                "provider": "openai",
                "model": "gpt-4o",
                "messages": [
                    {
                        "role": "system",
                        "content": CUSTOMER_SERVICE_PROMPT
                    }
                ],
                "tools": [
                    {
                        "type": "function",
                        "function": {
                            "name": "query_order",
                            "description": "根据订单号或手机号查询订单状态",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "order_id": {
                                        "type": "string",
                                        "description": "订单号"
                                    },
                                    "phone": {
                                        "type": "string",
                                        "description": "手机号后四位"
                                    }
                                }
                            }
                        },
                        "server": {
                            "url": "https://your-api.com/vapi/tools"
                        }
                    },
                    {
                        "type": "function",
                        "function": {
                            "name": "track_logistics",
                            "description": "查询物流信息",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "tracking_number": {
                                        "type": "string",
                                        "description": "快递单号"
                                    }
                                },
                                "required": ["tracking_number"]
                            }
                        },
                        "server": {
                            "url": "https://your-api.com/vapi/tools"
                        }
                    },
                    {
                        "type": "function",
                        "function": {
                            "name": "create_return_request",
                            "description": "创建退换货申请",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "order_id": {
                                        "type": "string",
                                        "description": "订单号"
                                    },
                                    "reason": {
                                        "type": "string",
                                        "description": "退换货原因"
                                    },
                                    "type": {
                                        "type": "string",
                                        "enum": ["refund", "exchange"],
                                        "description": "退款或换货"
                                    }
                                },
                                "required": ["order_id", "reason", "type"]
                            }
                        },
                        "server": {
                            "url": "https://your-api.com/vapi/tools"
                        }
                    },
                    {
                        "type": "function",
                        "function": {
                            "name": "transfer_to_human",
                            "description": "转接人工客服",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "reason": {
                                        "type": "string",
                                        "description": "转接原因"
                                    },
                                    "department": {
                                        "type": "string",
                                        "enum": ["general", "refund", "complaint", "vip"],
                                        "description": "转接部门"
                                    }
                                },
                                "required": ["reason"]
                            }
                        }
                    },
                    {
                        "type": "function",
                        "function": {
                            "name": "search_knowledge_base",
                            "description": "搜索产品知识库回答常见问题",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "query": {
                                        "type": "string",
                                        "description": "用户的问题"
                                    }
                                },
                                "required": ["query"]
                            }
                        },
                        "server": {
                            "url": "https://your-api.com/vapi/tools"
                        }
                    }
                ]
            },
            "voice": {
                "provider": "11labs",
                "voiceId": "21m00Tcm4TlvDq8ikWAM",
                "stability": 0.5,
                "similarityBoost": 0.75
            },
            "firstMessage": "您好，欢迎致电星辰商城客服中心，我是智能客服小星。请问有什么可以帮您的？",
            "endCallMessage": "感谢您的来电，祝您购物愉快，再见！",
            "transcriber": {
                "provider": "deepgram",
                "model": "nova-3",
                "language": "zh"
            },
            "silenceTimeoutSeconds": 20,
            "maxDurationSeconds": 600,
            "endCallFunctionEnabled": True,
            "hipaaEnabled": False
        }
    )
    return response.json()
 
 
# 客服 Agent 系统提示词
CUSTOMER_SERVICE_PROMPT = """你是星辰商城的 AI 电话客服"小星"。
 
## 身份与风格
- 语气：温暖专业，像一位耐心的客服老手
- 语速：适中偏慢，确保客户听清
- 每次回复：控制在 2-3 句话
- 语言：简洁口语化中文，避免书面语
 
## 核心能力
1. 订单查询：根据订单号或手机号查询订单状态
2. 物流追踪：查询快递物流信息
3. 退换货：协助创建退换货申请（需二次确认）
4. 常见问题：通过知识库回答产品相关问题
5. 人工转接：无法解决时转接人工客服
 
## 对话规则
- 先确认客户身份（手机尾号或订单号）
- 敏感操作（退款、取消）必须二次确认："您确认要申请退款吗？"
- 不确定的信息说"我帮您查一下"，不要编造
- 客户情绪激动时先安抚："非常理解您的心情，我来帮您解决"
- 连续 2 次无法理解时主动转人工
 
## 电话特殊规则
- 不使用 Markdown 格式、表情符号
- 数字用口语（"一百二十三" 而非 "123"）
- 金额明确说"元"（"三百五十元"）
- 需要列举时分多轮说明
"""

# tools_server.py — 客服工具处理服务器
from fastapi import FastAPI, Request
from pinecone import Pinecone
from openai import OpenAI
import json
import httpx
 
app = FastAPI()
openai_client = OpenAI()
pc = Pinecone(api_key="your-pinecone-key")
index = pc.Index("product-knowledge")
 
@app.post("/vapi/tools")
async def handle_tool_call(request: Request):
    """处理 Vapi 工具调用"""
    body = await request.json()
    tool_call = body["message"]["toolCalls"][0]
    func_name = tool_call["function"]["name"]
    args = json.loads(tool_call["function"]["arguments"])
 
    result = ""
 
    if func_name == "query_order":
        result = await query_order(args)
    elif func_name == "track_logistics":
        result = await track_logistics(args)
    elif func_name == "create_return_request":
        result = await create_return(args)
    elif func_name == "search_knowledge_base":
        result = await search_kb(args)
 
    return {
        "results": [{
            "toolCallId": tool_call["id"],
            "result": result
        }]
    }
 
async def query_order(args: dict) -> str:
    """查询订单状态（对接 ERP 系统）"""
    order_id = args.get("order_id")
    phone = args.get("phone")
    # 实际项目中对接你的 ERP/CRM API
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://your-erp.com/api/orders",
            params={"order_id": order_id, "phone_last4": phone}
        )
        if resp.status_code == 200:
            order = resp.json()
            return (
                f"订单 {order['id']} 状态：{order['status']}。"
                f"商品：{order['product_name']}，"
                f"金额：{order['amount']} 元。"
                f"{'预计 ' + order['eta'] + ' 送达。' if order.get('eta') else ''}"
            )
        return "抱歉，没有找到该订单，请确认订单号是否正确。"
 
async def track_logistics(args: dict) -> str:
    """查询物流信息"""
    tracking = args["tracking_number"]
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.kuaidi100.com/query",
            params={"num": tracking}
        )
        if resp.status_code == 200:
            data = resp.json()
            latest = data["data"][0] if data.get("data") else None
            if latest:
                return f"最新物流：{latest['time']}，{latest['context']}"
        return "暂时查不到物流信息，可能快递刚发出，请稍后再查。"
 
async def create_return(args: dict) -> str:
    """创建退换货申请"""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://your-erp.com/api/returns",
            json={
                "order_id": args["order_id"],
                "reason": args["reason"],
                "type": args["type"]
            }
        )
        if resp.status_code == 201:
            ret = resp.json()
            type_text = "退款" if args["type"] == "refund" else "换货"
            return (
                f"{type_text}申请已提交，申请编号 {ret['return_id']}。"
                f"预计 1 到 3 个工作日内处理完成，届时会短信通知您。"
            )
        return "申请提交失败，我帮您转接人工客服处理。"
 
async def search_kb(args: dict) -> str:
    """RAG 知识库检索"""
    query = args["query"]
    # 生成查询向量
    embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding
 
    # Pinecone 检索
    results = index.query(
        vector=embedding,
        top_k=3,
        include_metadata=True
    )
 
    if results["matches"]:
        contexts = [m["metadata"]["text"] for m in results["matches"]]
        return "根据我们的产品资料：" + " ".join(contexts[:2])
    return "这个问题我不太确定，我帮您转接专业同事来解答。"

# analytics.py — 通话数据收集与分析
from fastapi import FastAPI, Request
from datetime import datetime
import json
 
app = FastAPI()
 
# Vapi 通话结束 Webhook
@app.post("/vapi/call-ended")
async def handle_call_ended(request: Request):
    """收集通话数据用于分析"""
    body = await request.json()
    call = body.get("message", {}).get("call", {})
 
    analytics_data = {
        "call_id": call.get("id"),
        "duration_seconds": call.get("duration"),
        "ended_reason": call.get("endedReason"),
        "cost": call.get("cost"),
        "transcript": call.get("transcript"),
        "timestamp": datetime.now().isoformat(),
        # 从转录中提取关键指标
        "intent": extract_intent(call.get("transcript", "")),
        "resolved": not call.get("transferredTo"),
        "sentiment": analyze_sentiment(call.get("transcript", ""))
    }
 
    # 存入数据库（示例用 print）
    print(f"通话分析: {json.dumps(analytics_data, ensure_ascii=False)}")
    # 实际项目：写入 PostgreSQL / BigQuery / ClickHouse
    return {"status": "ok"}
 
def extract_intent(transcript: str) -> str:
    """从转录文本提取用户意图"""
    intent_keywords = {
        "订单查询": ["订单", "查询", "到哪了", "发货"],
        "退换货": ["退货", "换货", "退款", "不想要"],
        "物流追踪": ["快递", "物流", "送到", "配送"],
        "产品咨询": ["怎么用", "功能", "区别", "推荐"],
        "投诉": ["投诉", "不满意", "差评", "态度"]
    }
    for intent, keywords in intent_keywords.items():
        if any(kw in transcript for kw in keywords):
            return intent
    return "其他"
 
def analyze_sentiment(transcript: str) -> str:
    """简单情绪分析"""
    negative = ["生气", "投诉", "差评", "不满", "垃圾", "骗子"]
    positive = ["谢谢", "满意", "不错", "很好", "感谢"]
    if any(w in transcript for w in negative):
        return "negative"
    if any(w in transcript for w in positive):
        return "positive"
    return "neutral"

┌─────────────────────────────────────────────────────────────────────┐
│                    销售外呼 Agent 架构                               │
│                                                                     │
│  ┌──────────┐    ┌──────────────────────────────────────────────┐   │
│  │ CRM 线索  │───→│           Bland.ai 外呼引擎                  │   │
│  │ 数据库    │    │                                              │   │
│  └──────────┘    │  ┌─────────────────────────────────────────┐ │   │
│       ↑          │  │         批量外呼调度器                    │ │   │
│       │          │  │  · 并发控制（50 路同时拨打）              │ │   │
│       │          │  │  · 时段管理（9:00-12:00, 14:00-18:00）  │ │   │
│       │          │  │  · 重试策略（未接听最多重试 2 次）        │ │   │
│       │          │  └─────────────────────────────────────────┘ │   │
│       │          │                                              │   │
│       │          │  ┌────────┐  ┌────────┐  ┌────────────────┐ │   │
│       │          │  │Deepgram│→ │Claude  │→ │ ElevenLabs     │ │   │
│       │          │  │(STT)   │  │Sonnet  │  │ (TTS)          │ │   │
│       │          │  └────────┘  └───┬────┘  └────────────────┘ │   │
│       │          └──────────────────┼───────────────────────────┘   │
│       │                             │                               │
│       │            ┌────────────────┼────────────────┐              │
│       │            ↓                ↓                ↓              │
│       │      ┌──────────┐    ┌──────────┐    ┌──────────┐          │
│       │      │ 线索评分  │    │ 日程预约  │    │ 通话录音  │          │
│       │      │ (AI 打分) │    │ (Calendly)│    │ (分析)   │          │
│       │      └─────┬────┘    └──────────┘    └──────────┘          │
│       │            │                                                │
│       └────────────┘  回写 CRM（线索状态 + 评分 + 摘要）            │
│                                                                     │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                    结果处理流水线                              │   │
│  │  高意向(A级) → Slack 通知销售 → 自动创建跟进任务              │   │
│  │  中意向(B级) → 加入培育序列 → 3天后再次外呼                   │   │
│  │  低意向(C级) → 标记归档 → 30天后重新评估                      │   │
│  │  未接听     → 24小时后重试 → 最多重试2次                      │   │
│  └──────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘

# outbound_dialer.py — 批量外呼调度
import requests
import asyncio
import csv
from datetime import datetime, time
from typing import List, Dict
import os
 
BLAND_API_KEY = os.getenv("BLAND_API_KEY")
 
SALES_PROMPT = """你是 CloudFlow（一家 SaaS 公司）的销售代表"小李"。
你正在给潜在客户打电话，介绍我们的项目管理工具。
 
## 你的目标
1. 确认对方身份和职位
2. 了解他们当前的项目管理痛点
3. 简要介绍 CloudFlow 的核心价值
4. 如果对方有兴趣，预约一次 15 分钟的产品演示
5. 如果对方不感兴趣，礼貌结束并记录原因
 
## 对话策略
- 开场白要简短有力，30 秒内说明来意
- 多问开放式问题，让客户说话
- 不要硬推销，重点是了解需求
- 提到具体数据："帮助 500+ 团队提升 30% 效率"
- 如果客户说忙，主动提出换个时间再聊
 
## 关键话术
- 开场："您好，请问是 [客户姓名] 吗？我是 CloudFlow 的小李，
  注意到您之前在我们官网留过信息，想花两分钟跟您聊聊。"
- 痛点挖掘："您团队现在用什么工具管理项目？有没有遇到什么不方便的地方？"
- 价值传递："我们的客户平均节省了 30% 的项目沟通时间"
- 预约："要不我们约个 15 分钟的线上演示，我给您看看具体怎么用？"
- 拒绝处理："完全理解，那我先把资料发到您邮箱，您有空看看？"
 
## 电话规则
- 不使用 Markdown 格式
- 每次说话不超过 3 句
- 语气自然友好，不要像念稿
"""
 
async def load_leads(csv_path: str) -> List[Dict]:
    """从 CSV 加载线索数据"""
    leads = []
    with open(csv_path, "r", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for row in reader:
            leads.append({
                "name": row["name"],
                "phone": row["phone"],
                "company": row["company"],
                "title": row.get("title", ""),
                "source": row.get("source", "website")
            })
    return leads
 
def is_calling_hours() -> bool:
    """检查是否在允许拨打的时间段"""
    now = datetime.now().time()
    morning = (time(9, 0), time(12, 0))
    afternoon = (time(14, 0), time(18, 0))
    return (morning[0] <= now <= morning[1] or
            afternoon[0] <= now <= afternoon[1])
 
async def make_call(lead: Dict) -> Dict:
    """发起单通外呼"""
    response = requests.post(
        "https://api.bland.ai/v1/calls",
        headers={
            "Authorization": BLAND_API_KEY,
            "Content-Type": "application/json"
        },
        json={
            "phone_number": lead["phone"],
            "task": SALES_PROMPT.replace(
                "[客户姓名]", lead["name"]
            ),
            "voice": "mason",  # 自然男声
            "first_sentence": (
                f"您好，请问是{lead['name']}吗？"
                f"我是 CloudFlow 的小李。"
            ),
            "wait_for_greeting": True,
            "max_duration": 300,  # 最长 5 分钟
            "language": "zh",
            "tools": [
                {
                    "name": "book_demo",
                    "description": "当客户同意预约产品演示时调用",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "preferred_time": {
                                "type": "string",
                                "description": "客户偏好的时间"
                            },
                            "email": {
                                "type": "string",
                                "description": "客户邮箱"
                            }
                        }
                    },
                    "url": "https://your-api.com/bland/book-demo"
                },
                {
                    "name": "score_lead",
                    "description": "通话结束时对线索进行评分",
                    "input_schema": {
                        "type": "object",
                        "properties": {
                            "interest_level": {
                                "type": "string",
                                "enum": ["high", "medium", "low", "not_interested"],
                                "description": "客户兴趣程度"
                            },
                            "pain_points": {
                                "type": "string",
                                "description": "客户提到的痛点"
                            },
                            "next_action": {
                                "type": "string",
                                "description": "建议的下一步动作"
                            }
                        }
                    },
                    "url": "https://your-api.com/bland/score-lead"
                }
            ],
            "metadata": {
                "lead_name": lead["name"],
                "company": lead["company"],
                "source": lead["source"]
            }
        }
    )
    return response.json()

# batch_executor.py — 批量外呼执行
import asyncio
from typing import List, Dict
 
async def batch_dial(
    leads: List[Dict],
    concurrency: int = 50,
    delay_between: float = 1.0
) -> List[Dict]:
    """批量外呼，控制并发数"""
    semaphore = asyncio.Semaphore(concurrency)
    results = []
 
    async def dial_with_limit(lead):
        async with semaphore:
            if not is_calling_hours():
                print("当前不在拨打时段，等待...")
                while not is_calling_hours():
                    await asyncio.sleep(60)
 
            result = await make_call(lead)
            results.append({
                "lead": lead,
                "call_id": result.get("call_id"),
                "status": result.get("status")
            })
            await asyncio.sleep(delay_between)
 
    tasks = [dial_with_limit(lead) for lead in leads]
    await asyncio.gather(*tasks)
    return results
 
# 主执行流程
async def run_campaign():
    """执行外呼活动"""
    leads = await load_leads("leads.csv")
    print(f"共 {len(leads)} 条线索，开始外呼...")
 
    results = await batch_dial(leads, concurrency=50)
 
    # 统计结果
    connected = [r for r in results if r["status"] == "completed"]
    no_answer = [r for r in results if r["status"] == "no-answer"]
    print(f"接通: {len(connected)}, 未接: {len(no_answer)}")
 
if __name__ == "__main__":
    asyncio.run(run_campaign())

# lead_scoring.py — 线索评分与 CRM 集成
from fastapi import FastAPI, Request
import httpx
import os
 
app = FastAPI()
HUBSPOT_TOKEN = os.getenv("HUBSPOT_TOKEN")
SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")
 
@app.post("/bland/score-lead")
async def score_lead(request: Request):
    """接收 Bland.ai 的线索评分回调"""
    body = await request.json()
    interest = body.get("interest_level", "low")
    pain_points = body.get("pain_points", "")
    next_action = body.get("next_action", "")
    metadata = body.get("metadata", {})
 
    # 评分映射
    score_map = {"high": 90, "medium": 60, "low": 30, "not_interested": 10}
    score = score_map.get(interest, 0)
 
    # 回写 HubSpot CRM
    await update_hubspot_contact(
        name=metadata.get("lead_name"),
        company=metadata.get("company"),
        score=score,
        notes=f"痛点: {pain_points}\n下一步: {next_action}"
    )
 
    # 高意向线索实时通知销售
    if interest == "high":
        await notify_sales_team(
            lead_name=metadata.get("lead_name"),
            company=metadata.get("company"),
            pain_points=pain_points
        )
 
    return {"status": "scored", "score": score}
 
async def update_hubspot_contact(
    name: str, company: str, score: int, notes: str
):
    """更新 HubSpot 联系人"""
    async with httpx.AsyncClient() as client:
        # 搜索联系人
        search = await client.post(
            "https://api.hubapi.com/crm/v3/objects/contacts/search",
            headers={"Authorization": f"Bearer {HUBSPOT_TOKEN}"},
            json={
                "filterGroups": [{
                    "filters": [{
                        "propertyName": "company",
                        "operator": "EQ",
                        "value": company
                    }]
                }]
            }
        )
        contacts = search.json().get("results", [])
        if contacts:
            contact_id = contacts[0]["id"]
            await client.patch(
                f"https://api.hubapi.com/crm/v3/objects/contacts/{contact_id}",
                headers={"Authorization": f"Bearer {HUBSPOT_TOKEN}"},
                json={
                    "properties": {
                        "lead_score": str(score),
                        "ai_call_notes": notes,
                        "ai_call_status": "completed"
                    }
                }
            )
 
async def notify_sales_team(
    lead_name: str, company: str, pain_points: str
):
    """Slack 通知销售团队"""
    async with httpx.AsyncClient() as client:
        await client.post(SLACK_WEBHOOK, json={
            "text": (
                f"🔥 *高意向线索*\n"
                f"姓名: {lead_name}\n"
                f"公司: {company}\n"
                f"痛点: {pain_points}\n"
                f"请尽快跟进！"
            )
        })
 
@app.post("/bland/book-demo")
async def book_demo(request: Request):
    """处理预约 Demo 请求"""
    body = await request.json()
    preferred_time = body.get("preferred_time", "")
    email = body.get("email", "")
 
    # 调用 Calendly API 创建预约
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.calendly.com/scheduling_links",
            headers={
                "Authorization": f"Bearer {os.getenv('CALENDLY_TOKEN')}",
                "Content-Type": "application/json"
            },
            json={
                "max_event_count": 1,
                "owner": "https://api.calendly.com/event_types/YOUR_EVENT_TYPE",
                "owner_type": "EventType"
            }
        )
        link = resp.json().get("resource", {}).get("booking_url", "")
 
    # 发送预约邮件（简化示例）
    print(f"发送预约链接 {link} 到 {email}")
 
    return {
        "status": "booked",
        "message": f"好的，我已经把预约链接发到您的邮箱 {email} 了，"
                   f"您可以选择方便的时间。"
    }

你是 [公司名称] 的销售代表 [Agent名称]，正在给潜在客户打电话。

## 目标
1. 确认对方身份（30 秒内）
2. 了解当前痛点（1 分钟）
3. 传递产品价值（30 秒）
4. 推动下一步行动（预约 Demo / 发送资料）

## 开场白策略
- 直接说明来意，不要绕弯子
- 提到客户的公司名或行业，显示你做了功课
- 用一句话说明为什么打这个电话

## 异议处理
- "没时间" → "完全理解，我只占用您两分钟"
- "不需要" → "请问您目前用什么方案解决 [痛点]？"
- "发资料" → "好的，您方便告诉我邮箱吗？"
- "多少钱" → "根据团队规模不同，我们有灵活的方案，
  预约个 Demo 我给您详细介绍"

## 禁止行为
- 不要过度推销或施压
- 不要贬低竞品
- 不要承诺无法兑现的优惠
- 不要在客户明确拒绝后继续纠缠

┌─────────────────────────────────────────────────────────────────────┐
│                 语音控制开发工具架构                                  │
│                                                                     │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                    开发者语音输入                              │   │
│  │  "创建一个 React 组件，接收 name 和 age 两个 props"           │   │
│  └──────────────────────┬───────────────────────────────────────┘   │
│                         ↓                                           │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │              Wispr Flow / SuperWhisper                        │   │
│  │         (本地 AI 语音转文本，理解开发者上下文)                  │   │
│  │                                                              │   │
│  │  · 识别 camelCase、snake_case 等命名规范                     │   │
│  │  · 理解编程术语（"括号"→()、"花括号"→{}）                    │   │
│  │  · 自动格式化为结构化 prompt                                  │   │
│  └──────────────────────┬───────────────────────────────────────┘   │
│                         ↓                                           │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                AI 编码助手 (IDE 集成)                          │   │
│  │                                                              │   │
│  │  ┌────────────┐  ┌────────────┐  ┌────────────────────────┐ │   │
│  │  │   Cursor    │  │   Kiro     │  │   Claude Code (CLI)   │ │   │
│  │  │  (AI IDE)   │  │  (AI IDE)  │  │   (终端)              │ │   │
│  │  └──────┬─────┘  └──────┬─────┘  └──────────┬────────────┘ │   │
│  │         │               │                    │              │   │
│  │         └───────────────┼────────────────────┘              │   │
│  │                         ↓                                    │   │
│  │              LLM (Claude / GPT-4o / Gemini)                 │   │
│  │                         ↓                                    │   │
│  │              代码生成 / 修改 / 审查 / 解释                    │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                         ↓                                           │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                  扩展能力 (MCP / 工具)                        │   │
│  │                                                              │   │
│  │  "查一下生产环境的错误日志"  → MCP → Datadog API             │   │
│  │  "部署到 staging"          → MCP → GitHub Actions            │   │
│  │  "这个 PR 有什么问题"      → MCP → GitHub PR Review          │   │
│  │  "数据库里有多少活跃用户"   → MCP → PostgreSQL Query          │   │
│  └──────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘

# macOS（推荐）
# 从 https://wisprflow.ai 下载安装
# 或通过 Homebrew
brew install --cask wispr-flow
 
# 配置
# 1. 打开 Wispr Flow → Settings
# 2. 选择语言模型（推荐 Flow Pro 模式）
# 3. 设置快捷键（默认：按住 Fn 键说话）
# 4. 启用 "Developer Mode" 以识别代码术语

1. 打开 Cursor 编辑器
2. 按 Cmd+K 打开 AI 编辑面板
3. 按住 Fn 键（Wispr Flow 快捷键）开始说话：
   "创建一个 TypeScript 函数，接收用户列表，
    按注册时间排序，返回最近 10 个活跃用户"
4. Wispr Flow 将语音转为文本，自动填入 Cursor 的 prompt 框
5. Cursor AI 生成代码
6. 语音审查："把排序改成降序，加上类型注解"

# 在终端中使用 Claude Code
# 按住 Fn 说话，Wispr Flow 自动将语音转为文本输入终端
 
# 语音："claude 帮我重构 src/utils/auth.ts，
#        把所有的 callback 改成 async await"
# Wispr Flow 输出：
claude "帮我重构 src/utils/auth.ts，把所有的 callback 改成 async/await"
 
# 语音："查看最近的 git 提交记录，找出哪个提交引入了这个 bug"
claude "查看最近的 git log，找出哪个 commit 引入了 TypeError bug"

# voice_devops_mcp.py — 语音 DevOps MCP Server 示例
# 这个 MCP Server 让 AI IDE 能通过语音执行 DevOps 操作
 
from mcp.server import Server
from mcp.types import Tool, TextContent
import subprocess
import httpx
 
server = Server("voice-devops")
 
@server.tool()
async def check_deployment_status(environment: str) -> list[TextContent]:
    """
    查询部署状态。
    语音示例："staging 环境现在是什么状态"
    """
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.github.com/repos/your-org/your-repo/deployments",
            headers={"Authorization": f"token {GITHUB_TOKEN}"},
            params={"environment": environment}
        )
        deployments = resp.json()
        if deployments:
            latest = deployments[0]
            return [TextContent(
                type="text",
                text=f"{environment} 环境最新部署：\n"
                     f"  状态: {latest['status']}\n"
                     f"  分支: {latest['ref']}\n"
                     f"  时间: {latest['created_at']}"
            )]
        return [TextContent(type="text", text=f"{environment} 没有找到部署记录")]
 
@server.tool()
async def query_error_logs(
    service: str,
    time_range: str = "1h"
) -> list[TextContent]:
    """
    查询错误日志。
    语音示例："查一下 user-service 最近一小时的错误日志"
    """
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.datadoghq.com/api/v2/logs/events/search",
            headers={
                "DD-API-KEY": DATADOG_API_KEY,
                "DD-APPLICATION-KEY": DATADOG_APP_KEY
            },
            json={
                "filter": {
                    "query": f"service:{service} status:error",
                    "from": f"now-{time_range}",
                    "to": "now"
                },
                "sort": "-timestamp",
                "page": {"limit": 10}
            }
        )
        logs = resp.json().get("data", [])
        if logs:
            summary = f"{service} 最近 {time_range} 有 {len(logs)} 条错误：\n"
            for log in logs[:5]:
                msg = log["attributes"]["message"][:100]
                summary += f"  - {msg}\n"
            return [TextContent(type="text", text=summary)]
        return [TextContent(
            type="text",
            text=f"{service} 最近 {time_range} 没有错误日志，一切正常"
        )]
 
@server.tool()
async def trigger_deployment(
    environment: str,
    branch: str = "main"
) -> list[TextContent]:
    """
    触发部署。
    语音示例："把 main 分支部署到 staging"
    """
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://api.github.com/repos/your-org/your-repo/dispatches",
            headers={"Authorization": f"token {GITHUB_TOKEN}"},
            json={
                "event_type": "deploy",
                "client_payload": {
                    "environment": environment,
                    "branch": branch
                }
            }
        )
        if resp.status_code == 204:
            return [TextContent(
                type="text",
                text=f"已触发部署：{branch} → {environment}，"
                     f"预计 3-5 分钟完成。"
            )]
        return [TextContent(
            type="text",
            text=f"部署触发失败：{resp.status_code}"
        )]

┌─────────────────────────────────────────────────────────────┐
│              语音编程一日工作流                                │
│                                                             │
│  09:00  🎤 "查看今天的 GitHub Issues 和 PR"                  │
│         → MCP 调用 GitHub API → 显示待处理列表               │
│                                                             │
│  09:15  🎤 "开始处理 Issue 42，先看一下相关代码"              │
│         → AI 定位相关文件 → 展示代码上下文                    │
│                                                             │
│  09:30  🎤 "在 UserService 里加一个 getActiveUsers 方法，    │
│            查询最近 30 天有登录记录的用户，分页返回"           │
│         → AI 生成代码 → 开发者语音审查修改                    │
│                                                             │
│  10:00  🎤 "给这个方法写单元测试，覆盖空结果和分页边界"       │
│         → AI 生成测试 → 运行测试                             │
│                                                             │
│  10:30  🎤 "提交代码，PR 标题写修复 Issue 42"                │
│         → 自动 git add/commit/push → 创建 PR                │
│                                                             │
│  11:00  🎤 "staging 环境的错误日志有什么异常吗"              │
│         → MCP 查询 Datadog → 返回错误摘要                    │
│                                                             │
│  14:00  🎤 "把 PR 123 合并到 main，然后部署到 staging"       │
│         → 合并 PR → 触发 CI/CD → 部署                       │
│                                                             │
│  16:00  🎤 "生成今天的工作日报，包括完成的任务和明天计划"     │
│         → AI 汇总 git log + Issues → 生成日报                │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                  医疗预约语音 Agent 架构                              │
│                                                                     │
│  ┌──────────┐    ┌──────────────────────────────────────────────┐   │
│  │ 患者来电  │───→│          Retell AI / Twilio + 自建           │   │
│  │ (PSTN)   │    │                                              │   │
│  └──────────┘    │  ┌────────┐  ┌────────┐  ┌────────────────┐ │   │
│                  │  │Deepgram│→ │GPT-4o  │→ │ ElevenLabs     │ │   │
│  ┌──────────┐    │  │Nova-3  │  │(LLM)   │  │ (TTS)          │ │   │
│  │ AI 外呼  │───→│  │(STT)   │  │        │  │                │ │   │
│  │ (提醒)   │    │  └────────┘  └───┬────┘  └────────────────┘ │   │
│  └──────────┘    └──────────────────┼───────────────────────────┘   │
│                                     │ Function Calling               │
│                    ┌────────────────┼────────────────┐              │
│                    ↓                ↓                ↓              │
│              ┌──────────┐    ┌──────────┐    ┌──────────┐          │
│              │ 排班系统  │    │ 患者档案  │    │ 短信通知  │          │
│              │ (日程API) │    │ (EMR/HIS)│    │ (Twilio) │          │
│              └──────────┘    └──────────┘    └──────────┘          │
│                                     │                               │
│              ┌──────────────────────┼──────────────────────┐        │
│              │          合规与安全层                        │        │
│              │  · 通话录音加密存储                          │        │
│              │  · 患者信息脱敏处理                          │        │
│              │  · HIPAA / 个人信息保护法合规                │        │
│              │  · 敏感操作人工确认                          │        │
│              └─────────────────────────────────────────────┘        │
└─────────────────────────────────────────────────────────────────────┘

# medical_agent.py — 医疗预约语音 Agent
import os
import json
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
from datetime import datetime, timedelta
from openai import AsyncOpenAI
import httpx
 
app = FastAPI()
openai_client = AsyncOpenAI()
 
MEDICAL_SYSTEM_PROMPT = """你是阳光诊所的 AI 预约助手"小阳"。
 
## 身份与风格
- 语气：温暖、耐心、专业
- 语速：偏慢，确保患者（尤其老年人）听清
- 每次回复：1-2 句话，简洁明了
- 称呼：使用"您"，称对方为"先生/女士"
 
## 核心能力
1. 新预约：帮患者预约门诊
2. 改约：修改已有预约时间
3. 取消预约：取消已有预约
4. 查询预约：查询预约详情
5. 科室咨询：介绍各科室和医生
 
## 预约流程
1. 确认患者身份（姓名 + 手机号后四位）
2. 了解就诊需求（哪个科室、什么症状）
3. 推荐合适的医生和时段
4. 确认预约信息（医生、时间、科室）
5. 发送短信确认
 
## 医疗特殊规则
- 不提供任何医疗诊断或用药建议
- 紧急情况（胸痛、呼吸困难等）立即建议拨打 120
- 患者描述症状时，只用于推荐科室，不做判断
- 涉及检查报告解读，转接医生
- 所有操作需患者口头确认
 
## 电话规则
- 不使用 Markdown 格式
- 时间用口语："明天上午十点"而非"2025-01-15 10:00"
- 医生姓名要说全："张明华主任"而非"张医生"
- 重要信息重复确认
"""
 
# Retell AI Webhook 处理
@app.post("/retell/webhook")
async def handle_retell_webhook(request: Request):
    """处理 Retell AI 的工具调用"""
    body = await request.json()
    event = body.get("event")
 
    if event == "call_started":
        return JSONResponse({"status": "ok"})
 
    elif event == "tool_call":
        tool_name = body["tool_call"]["name"]
        args = body["tool_call"]["arguments"]
 
        if tool_name == "check_availability":
            result = await check_doctor_availability(args)
        elif tool_name == "book_appointment":
            result = await book_appointment(args)
        elif tool_name == "cancel_appointment":
            result = await cancel_appointment(args)
        elif tool_name == "lookup_patient":
            result = await lookup_patient(args)
        elif tool_name == "send_confirmation_sms":
            result = await send_sms(args)
        else:
            result = "未知操作"
 
        return JSONResponse({
            "tool_call_id": body["tool_call"]["id"],
            "result": result
        })
 
    return JSONResponse({"status": "ok"})

# scheduling.py — 排班和预约业务逻辑
from datetime import datetime, timedelta
from typing import Optional
import asyncpg
 
# 数据库连接池（生产环境使用连接池）
DB_URL = "postgresql://user:pass@localhost/clinic"
 
async def check_doctor_availability(args: dict) -> str:
    """查询医生可用时段"""
    department = args.get("department", "")
    doctor_name = args.get("doctor_name", "")
    preferred_date = args.get("date", "")
 
    conn = await asyncpg.connect(DB_URL)
    try:
        # 查询指定科室/医生的可用时段
        query = """
            SELECT d.name, d.title, s.slot_date, s.slot_time,
                   s.is_available
            FROM doctors d
            JOIN schedule_slots s ON d.id = s.doctor_id
            WHERE d.department = $1
              AND s.slot_date >= $2
              AND s.is_available = true
        """
        params = [department, preferred_date or datetime.now().date()]
        if doctor_name:
            query += " AND d.name LIKE $3"
            params.append(f"%{doctor_name}%")
        query += " ORDER BY s.slot_date, s.slot_time LIMIT 6"
 
        rows = await conn.fetch(query, *params)
 
        if not rows:
            return (
                f"抱歉，{department}近期暂时没有可用的预约时段。"
                f"需要我帮您查看其他科室或其他时间吗？"
            )
 
        result = f"{department}可用时段：\n"
        for row in rows:
            date_str = row["slot_date"].strftime("%m月%d日")
            time_str = row["slot_time"].strftime("%H:%M")
            result += (
                f"{row['name']}{row['title']}，"
                f"{date_str} {time_str}\n"
            )
        return result
    finally:
        await conn.close()
 
async def book_appointment(args: dict) -> str:
    """创建预约"""
    patient_id = args.get("patient_id")
    doctor_name = args.get("doctor_name")
    slot_date = args.get("date")
    slot_time = args.get("time")
 
    conn = await asyncpg.connect(DB_URL)
    try:
        # 检查时段是否仍然可用（防止并发冲突）
        slot = await conn.fetchrow("""
            SELECT s.id, d.name, d.department
            FROM schedule_slots s
            JOIN doctors d ON d.id = s.doctor_id
            WHERE d.name LIKE $1
              AND s.slot_date = $2
              AND s.slot_time = $3
              AND s.is_available = true
            FOR UPDATE
        """, f"%{doctor_name}%", slot_date, slot_time)
 
        if not slot:
            return "抱歉，这个时段刚刚被预约了，需要我帮您看看其他时间吗？"
 
        # 创建预约记录
        appointment_id = await conn.fetchval("""
            INSERT INTO appointments
                (patient_id, slot_id, status, created_at)
            VALUES ($1, $2, 'confirmed', NOW())
            RETURNING id
        """, patient_id, slot["id"])
 
        # 标记时段为不可用
        await conn.execute("""
            UPDATE schedule_slots SET is_available = false
            WHERE id = $1
        """, slot["id"])
 
        date_str = slot_date if isinstance(slot_date, str) else slot_date.strftime("%m月%d日")
        return (
            f"预约成功！预约编号 {appointment_id}。"
            f"{slot['name']}，{date_str} {slot_time}，"
            f"{slot['department']}。"
            f"我马上给您发一条确认短信。"
        )
    finally:
        await conn.close()
 
async def cancel_appointment(args: dict) -> str:
    """取消预约"""
    appointment_id = args.get("appointment_id")
    patient_id = args.get("patient_id")
 
    conn = await asyncpg.connect(DB_URL)
    try:
        appt = await conn.fetchrow("""
            SELECT a.id, a.slot_id, s.slot_date, s.slot_time, d.name
            FROM appointments a
            JOIN schedule_slots s ON s.id = a.slot_id
            JOIN doctors d ON d.id = s.doctor_id
            WHERE a.id = $1 AND a.patient_id = $2
              AND a.status = 'confirmed'
        """, appointment_id, patient_id)
 
        if not appt:
            return "没有找到这个预约记录，请确认预约编号是否正确。"
 
        # 取消预约并释放时段
        await conn.execute("""
            UPDATE appointments SET status = 'cancelled' WHERE id = $1
        """, appointment_id)
        await conn.execute("""
            UPDATE schedule_slots SET is_available = true WHERE id = $1
        """, appt["slot_id"])
 
        return (
            f"已取消您在 {appt['name']} 处的预约。"
            f"如需重新预约，随时给我们打电话。"
        )
    finally:
        await conn.close()
 
async def lookup_patient(args: dict) -> str:
    """查询患者信息（用于身份验证）"""
    name = args.get("name", "")
    phone_last4 = args.get("phone_last4", "")
 
    conn = await asyncpg.connect(DB_URL)
    try:
        patient = await conn.fetchrow("""
            SELECT id, name, phone
            FROM patients
            WHERE name = $1
              AND RIGHT(phone, 4) = $2
        """, name, phone_last4)
 
        if patient:
            return json.dumps({
                "found": True,
                "patient_id": patient["id"],
                "name": patient["name"]
            })
        return json.dumps({
            "found": False,
            "message": "没有找到匹配的患者记录"
        })
    finally:
        await conn.close()
 
async def send_sms(args: dict) -> str:
    """发送预约确认短信"""
    phone = args.get("phone")
    message = args.get("message")
 
    async with httpx.AsyncClient() as client:
        await client.post(
            f"https://api.twilio.com/2010-04-01/Accounts/"
            f"{os.getenv('TWILIO_SID')}/Messages.json",
            auth=(os.getenv("TWILIO_SID"), os.getenv("TWILIO_TOKEN")),
            data={
                "To": phone,
                "From": os.getenv("TWILIO_PHONE"),
                "Body": message
            }
        )
    return "确认短信已发送。"

# reminder_caller.py — 预约提醒自动外呼
import asyncio
from datetime import datetime, timedelta
import asyncpg
import requests
import os
 
RETELL_API_KEY = os.getenv("RETELL_API_KEY")
 
REMINDER_PROMPT = """你是阳光诊所的预约提醒助手"小阳"。
你正在给患者打电话提醒明天的预约。
 
## 任务
1. 确认对方身份
2. 提醒预约时间和医生
3. 确认是否按时就诊
4. 如需改约，帮助处理
 
## 话术
- "您好，请问是 [患者姓名] 吗？"
- "提醒您明天 [时间] 在 [科室] 有一个预约，
   是 [医生姓名] 的门诊。"
- "请问您能按时来吗？"
- 如果确认："好的，请记得带上医保卡，提前 15 分钟到。"
- 如果要改约："好的，我帮您看看其他时间。"
- 如果要取消："好的，已帮您取消，需要时随时预约。"
 
## 规则
- 最多尝试确认 2 次
- 如果对方说忙，简短说明后结束
- 不提供任何医疗建议
"""
 
async def get_tomorrow_appointments() -> list:
    """获取明天的预约列表"""
    conn = await asyncpg.connect(DB_URL)
    try:
        tomorrow = datetime.now().date() + timedelta(days=1)
        rows = await conn.fetch("""
            SELECT a.id, p.name, p.phone,
                   d.name as doctor_name, d.department,
                   s.slot_time
            FROM appointments a
            JOIN patients p ON p.id = a.patient_id
            JOIN schedule_slots s ON s.id = a.slot_id
            JOIN doctors d ON d.id = s.doctor_id
            WHERE s.slot_date = $1
              AND a.status = 'confirmed'
              AND a.reminder_sent = false
        """, tomorrow)
        return [dict(row) for row in rows]
    finally:
        await conn.close()
 
async def make_reminder_call(appointment: dict):
    """发起提醒电话"""
    prompt = REMINDER_PROMPT.replace(
        "[患者姓名]", appointment["name"]
    ).replace(
        "[时间]", appointment["slot_time"].strftime("%H:%M")
    ).replace(
        "[科室]", appointment["department"]
    ).replace(
        "[医生姓名]", appointment["doctor_name"]
    )
 
    response = requests.post(
        "https://api.retellai.com/v2/create-phone-call",
        headers={
            "Authorization": f"Bearer {RETELL_API_KEY}",
            "Content-Type": "application/json"
        },
        json={
            "from_number": os.getenv("CLINIC_PHONE"),
            "to_number": appointment["phone"],
            "agent_id": os.getenv("REMINDER_AGENT_ID"),
            "retell_llm_dynamic_variables": {
                "patient_name": appointment["name"],
                "appointment_time": appointment["slot_time"].strftime("%H:%M"),
                "doctor_name": appointment["doctor_name"],
                "department": appointment["department"]
            },
            "metadata": {
                "appointment_id": appointment["id"]
            }
        }
    )
    return response.json()
 
async def run_daily_reminders():
    """每天下午 4 点执行提醒外呼"""
    appointments = await get_tomorrow_appointments()
    print(f"明天共 {len(appointments)} 个预约需要提醒")
 
    for appt in appointments:
        result = await make_reminder_call(appt)
        print(f"提醒 {appt['name']}: {result.get('status')}")
        await asyncio.sleep(2)  # 间隔 2 秒
 
if __name__ == "__main__":
    asyncio.run(run_daily_reminders())