第 9 课：进阶扩展

第 1-2 课：了解全貌 + 环境搭建
第 3 课：query() 工单自动分类
第 4 课：MCP 工具（知识库、工单数据库）
第 5 课：ClaudeSDKClient 多轮对话
第 6 课：Hook 安全管控 + 权限回调
第 7 课：多 Agent 角色协作
第 8 课：完整项目实战

"""用 SQLite 做工单存储（生产环境换成 PostgreSQL/MySQL）"""

import sqlite3
import json
from claude_agent_sdk import tool

DB_PATH = "support.db"

def init_db():
    """初始化数据库"""
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS tickets (
            id TEXT PRIMARY KEY,
            customer TEXT NOT NULL,
            subject TEXT NOT NULL,
            description TEXT,
            status TEXT DEFAULT 'open',
            priority TEXT DEFAULT 'medium',
            category TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS ticket_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ticket_id TEXT NOT NULL,
            action TEXT NOT NULL,
            note TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (ticket_id) REFERENCES tickets(id)
        )
    """)
    conn.commit()
    conn.close()


@tool(
    "db_search_tickets",
    "在数据库中搜索工单，支持按状态、优先级、关键词筛选",
    {
        "status": {"type": "string", "description": "状态筛选: open/pending/resolved/closed，留空查全部"},
        "keyword": {"type": "string", "description": "关键词搜索（标题或描述）"},
        "limit": {"type": "number", "description": "最多返回几条，默认 10"},
    }
)
async def db_search_tickets(status: str = "", keyword: str = "", limit: int = 10) -> str:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row

    query = "SELECT * FROM tickets WHERE 1=1"
    params = []

    if status:
        query += " AND status = ?"
        params.append(status)
    if keyword:
        query += " AND (subject LIKE ? OR description LIKE ?)"
        params.extend([f"%{keyword}%", f"%{keyword}%"])

    query += " ORDER BY created_at DESC LIMIT ?"
    params.append(limit)

    rows = conn.execute(query, params).fetchall()
    conn.close()

    if not rows:
        return "没有找到匹配的工单"

    results = [dict(row) for row in rows]
    return json.dumps(results, ensure_ascii=False, indent=2)

"""HTTP API 接口 —— 供网页/App/微信等渠道调用"""

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio

from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, TextBlock,
)

app = FastAPI(title="智能客服 API")

# 请求格式
class ChatRequest(BaseModel):
    message: str
    session_id: str = ""
    customer_name: str = "访客"

# 响应格式
class ChatResponse(BaseModel):
    reply: str
    session_id: str
    tools_used: list[str]


@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    """处理客户消息"""
    options = ClaudeAgentOptions(
        system_prompt="你是客服助手小智，用中文简洁回复。",
        max_turns=5,
        # 添加 MCP 服务器和工具配置...
    )

    reply_parts = []
    tools_used = []

    async for msg in query(prompt=req.message, options=options):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    reply_parts.append(block.text)
                elif hasattr(block, "name"):
                    tools_used.append(block.name)

    return ChatResponse(
        reply="".join(reply_parts),
        session_id=req.session_id or "new",
        tools_used=tools_used,
    )


# 启动: uvicorn api:app --host 0.0.0.0 --port 8000

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "我密码忘了", "customer_name": "张三"}'

from fastapi import WebSocket

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    """WebSocket 实时对话"""
    await websocket.accept()

    options = ClaudeAgentOptions(
        system_prompt="你是客服助手，用中文回复。",
    )

    async with ClaudeSDKClient(options=options) as client:
        try:
            while True:
                # 接收客户消息
                data = await websocket.receive_text()

                # 发给 AI
                await client.query(data)

                # 流式返回
                async for msg in client.receive_response():
                    if isinstance(msg, AssistantMessage):
                        for block in msg.content:
                            if isinstance(block, TextBlock):
                                await websocket.send_text(block.text)
        except Exception:
            await websocket.close()

"""客服系统监控"""

import datetime
from dataclasses import dataclass, field

@dataclass
class SupportMetrics:
    """客服系统指标"""
    total_tickets: int = 0
    resolved_tickets: int = 0
    avg_response_time_ms: float = 0
    total_cost_usd: float = 0
    tickets_by_category: dict = field(default_factory=dict)
    tickets_by_urgency: dict = field(default_factory=dict)
    customer_satisfaction: list = field(default_factory=list)

    def record_ticket(self, category: str, urgency: str,
                      response_time_ms: float, cost_usd: float):
        self.total_tickets += 1
        self.tickets_by_category[category] = (
            self.tickets_by_category.get(category, 0) + 1
        )
        self.tickets_by_urgency[urgency] = (
            self.tickets_by_urgency.get(urgency, 0) + 1
        )
        # 移动平均
        n = self.total_tickets
        self.avg_response_time_ms = (
            (self.avg_response_time_ms * (n - 1) + response_time_ms) / n
        )
        self.total_cost_usd += cost_usd

    def report(self) -> str:
        lines = [
            "📊 客服系统运营报告",
            "=" * 40,
            f"总工单数: {self.total_tickets}",
            f"已解决: {self.resolved_tickets}",
            f"解决率: {self.resolved_tickets / max(self.total_tickets, 1) * 100:.1f}%",
            f"平均响应: {self.avg_response_time_ms:.0f}ms",
            f"总费用: ${self.total_cost_usd:.4f}",
            "",
            "类别分布:",
        ]
        for cat, count in sorted(self.tickets_by_category.items(),
                                  key=lambda x: -x[1]):
            lines.append(f"  {cat}: {count}")

        lines.append("\n紧急度分布:")
        for urg, count in sorted(self.tickets_by_urgency.items()):
            lines.append(f"  {urg}: {count}")

        return "\n".join(lines)


# 全局指标
metrics = SupportMetrics()

from claude_agent_sdk import tool

@tool(
    "escalate_to_human",
    "将工单升级给人工客服处理。当遇到AI无法解决的问题时使用。",
    {
        "ticket_id": {"type": "string", "description": "工单号"},
        "reason": {"type": "string", "description": "升级原因"},
        "suggested_team": {"type": "string", "description": "建议分配的团队"},
    }
)
async def escalate_to_human(ticket_id: str, reason: str,
                             suggested_team: str = "support") -> str:
    """升级到人工"""
    # 实际项目中，这里会：
    # 1. 更新工单状态
    # 2. 通知人工客服（邮件/Slack/钉钉）
    # 3. 记录升级原因
    print(f"\n🚨 工单升级通知:")
    print(f"   工单: {ticket_id}")
    print(f"   原因: {reason}")
    print(f"   团队: {suggested_team}")

    return (
        f"已将工单 {ticket_id} 升级给 {suggested_team} 团队。"
        f"原因: {reason}。人工客服会在 30 分钟内跟进。"
    )

SYSTEM_PROMPT += """
自动升级规则：
- 客户连续 3 次表达不满 → 升级
- 涉及法律/投诉/安全 → 升级
- 退款超过 ¥5000 → 升级
- 客户明确要求人工 → 升级
- AI 3 轮对话内无法解决 → 升级
"""

# 在系统提示词中加入多语言规则
MULTILINGUAL_PROMPT = """
语言规则：
- 自动检测客户使用的语言
- 用客户的语言回复
- 如果分不清，默认用中文
- 专业术语保留英文原文
"""

# 或者用一个翻译 Agent
translator = AgentDefinition(
    description="检测语言并翻译工单内容",
    prompt="""你是翻译专家。
1. 检测输入文本的语言
2. 如果不是中文，翻译成中文
3. 输出格式: {"original_lang": "...", "translated": "..."}""",
    tools=["Read"],
    model="haiku",
)

options = ClaudeAgentOptions(
    # 单次对话预算上限
    max_budget_usd=1.00,

    # 限制最大轮次
    max_turns=10,

    # 用便宜模型做简单任务
    model="claude-haiku-4-5",

    # ...其他配置
)

1. API Key 安全
   ├── 不要硬编码在代码里
   ├── 用环境变量或 Secret Manager
   └── 定期轮换

2. 错误处理
   ├── AI 服务不可用 → 降级到关键词匹配
   ├── 工具调用失败 → 重试 + 告警
   └── 超时 → 自动道歉 + 转人工

3. 性能优化
   ├── 简单问题用 Haiku（快、便宜）
   ├── 知识库检索加缓存
   └── 批量处理用异步并发

4. 监控告警
   ├── 响应时间 > 5s → 告警
   ├── 错误率 > 5% → 告警
   ├── 每日费用 > 预算 → 告警
   └── 客户满意度 < 3 分 → 告警

5. 数据安全
   ├── PII 脱敏（Hook 自动处理）
   ├── 审计日志保存 90 天
   ├── 符合数据保护法规
   └── 定期安全审查