第 8 课：完整项目实战

smart-support/
├── main.py                 ← 主程序入口
├── config.py               ← 系统配置
├── tools/
│   ├── __init__.py
│   ├── knowledge.py        ← 知识库工具
│   └── tickets.py          ← 工单工具
├── hooks/
│   ├── __init__.py
│   └── safety.py           ← 安全 Hook
├── agents/
│   ├── __init__.py
│   └── team.py             ← Agent 团队定义
├── data/
│   └── faq.json            ← FAQ 数据
└── requirements.txt

"""系统配置"""

SYSTEM_PROMPT = """你是智能客服系统的调度中心。

你管理一个客服团队：
- classifier: 工单分类员（快速判断类型和紧急度）
- support-agent: 客服专员（回答问题、查询数据）
- quality-checker: 质检员（检查回复质量）

你还有以下工具可以直接使用：
- search_faq: 搜索知识库
- get_ticket: 查询工单
- create_ticket: 创建工单
- update_ticket: 更新工单状态

工作流程：
1. 理解客户问题
2. 需要时先用 classifier 分类
3. 用 support-agent 或直接用工具回答
4. 重要回复用 quality-checker 检查
5. 遇到愤怒客户或复杂问题，建议转人工

规则：
- 全程用中文
- 回复友好、专业、简洁
- 不泄露其他客户信息
- 退款超过 5000 元建议转人工
"""

"""知识库检索工具"""

import json
from pathlib import Path
from claude_agent_sdk import tool

# 知识库数据
FAQ_FILE = Path(__file__).parent.parent / "data" / "faq.json"

def load_faq() -> list[dict]:
    """加载 FAQ 数据"""
    if FAQ_FILE.exists():
        return json.loads(FAQ_FILE.read_text(encoding="utf-8"))
    # 默认 FAQ
    return [
        {"question": "如何重置密码", "answer": "进入登录页，点击忘记密码，输入邮箱即可。",
         "keywords": ["密码", "重置", "忘记", "登录不了"]},
        {"question": "退款政策", "answer": "7天内全额退款，7-30天退未使用部分。",
         "keywords": ["退款", "退钱", "取消"]},
        {"question": "API调用限制", "answer": "免费版100次/天，基础版1万次/天，企业版无限。",
         "keywords": ["API", "限制", "429", "频率"]},
        {"question": "如何升级套餐", "answer": "登录后台 → 设置 → 套餐管理 → 选择套餐。",
         "keywords": ["升级", "企业版", "套餐", "价格"]},
        {"question": "数据导出", "answer": "数据管理页 → 选择范围 → 选择格式(CSV/Excel) → 点导出。",
         "keywords": ["导出", "下载", "数据", "Excel"]},
        {"question": "多人协作", "answer": "设置 → 团队管理 → 邀请成员，支持设定不同权限角色。",
         "keywords": ["协作", "团队", "邀请", "成员", "权限"]},
    ]


FAQ_DATA = load_faq()


@tool(
    "search_faq",
    "当客户问常见问题时，在知识库中搜索答案。返回匹配的FAQ条目。",
    {"keyword": {"type": "string", "description": "搜索关键词（如：密码、退款、API）"}}
)
async def search_faq(keyword: str) -> str:
    """搜索知识库"""
    results = []
    kw = keyword.lower()

    for faq in FAQ_DATA:
        searchable = (faq["question"] + faq["answer"]
                      + " ".join(faq.get("keywords", []))).lower()
        if kw in searchable:
            results.append(faq)

    if not results:
        return f"知识库中未找到与 '{keyword}' 相关的内容。建议人工处理。"

    output = f"找到 {len(results)} 条相关FAQ：\n\n"
    for faq in results:
        output += f"Q: {faq['question']}\nA: {faq['answer']}\n\n"
    return output

"""工单管理工具"""

import json
import datetime
from claude_agent_sdk import tool

# 模拟数据库
TICKET_DB: dict[str, dict] = {
    "TK-001": {
        "id": "TK-001", "customer": "张三", "phone": "13812345678",
        "subject": "登录异常", "description": "登录页面一直转圈",
        "status": "open", "priority": "high",
        "created_at": "2025-03-15 10:30",
        "history": ["[2025-03-15 10:30] 工单创建"],
    },
    "TK-002": {
        "id": "TK-002", "customer": "李四", "phone": "13987654321",
        "subject": "企业版咨询", "description": "想了解企业版功能和价格",
        "status": "pending", "priority": "low",
        "created_at": "2025-03-15 11:00",
        "history": ["[2025-03-15 11:00] 工单创建"],
    },
}

ORDER_DB: dict[str, dict] = {
    "ORD-001": {"id": "ORD-001", "customer": "张三", "product": "基础版年付",
                "amount": 1200, "status": "active", "date": "2025-01-15"},
    "ORD-002": {"id": "ORD-002", "customer": "李四", "product": "企业版月付",
                "amount": 500, "status": "active", "date": "2025-03-01"},
}

_counter = 100


@tool(
    "get_ticket",
    "根据工单号查询工单详细信息",
    {"ticket_id": {"type": "string", "description": "工单号（如 TK-001）"}}
)
async def get_ticket(ticket_id: str) -> str:
    ticket = TICKET_DB.get(ticket_id.upper())
    if not ticket:
        return f"未找到工单 {ticket_id}"
    return json.dumps(ticket, ensure_ascii=False, indent=2)


@tool(
    "create_ticket",
    "为客户创建新的工单",
    {"subject": {"type": "string", "description": "工单标题"},
     "description": {"type": "string", "description": "问题描述"},
     "customer": {"type": "string", "description": "客户姓名"},
     "priority": {"type": "string", "description": "优先级: critical/high/medium/low"}}
)
async def create_ticket(subject: str, description: str,
                        customer: str = "未知", priority: str = "medium") -> str:
    global _counter
    _counter += 1
    ticket_id = f"TK-{_counter}"
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")

    TICKET_DB[ticket_id] = {
        "id": ticket_id, "customer": customer,
        "subject": subject, "description": description,
        "status": "open", "priority": priority,
        "created_at": now,
        "history": [f"[{now}] 工单创建"],
    }
    return f"✅ 工单已创建: {ticket_id} | {subject} | 优先级: {priority}"


@tool(
    "update_ticket",
    "更新工单状态",
    {"ticket_id": {"type": "string", "description": "工单号"},
     "status": {"type": "string", "description": "新状态: open/pending/resolved/closed"},
     "note": {"type": "string", "description": "操作备注"}}
)
async def update_ticket(ticket_id: str, status: str, note: str = "") -> str:
    ticket = TICKET_DB.get(ticket_id.upper())
    if not ticket:
        return f"未找到工单 {ticket_id}"

    old = ticket["status"]
    ticket["status"] = status
    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
    ticket["history"].append(f"[{now}] {old} → {status}: {note}")
    return f"✅ {ticket_id}: {old} → {status}"


@tool(
    "get_order",
    "查询客户的订单信息",
    {"order_id": {"type": "string", "description": "订单号（如 ORD-001）"}}
)
async def get_order(order_id: str) -> str:
    order = ORDER_DB.get(order_id.upper())
    if not order:
        return f"未找到订单 {order_id}"
    return json.dumps(order, ensure_ascii=False, indent=2)


@tool(
    "search_customer",
    "按客户姓名搜索其工单和订单",
    {"name": {"type": "string", "description": "客户姓名"}}
)
async def search_customer(name: str) -> str:
    tickets = [t for t in TICKET_DB.values() if name in t.get("customer", "")]
    orders = [o for o in ORDER_DB.values() if name in o.get("customer", "")]

    if not tickets and not orders:
        return f"未找到客户 '{name}' 的任何记录"

    output = f"客户 '{name}' 的记录：\n"
    if tickets:
        output += f"\n工单 ({len(tickets)} 个):\n"
        for t in tickets:
            output += f"  {t['id']}: {t['subject']} [{t['status']}]\n"
    if orders:
        output += f"\n订单 ({len(orders)} 个):\n"
        for o in orders:
            output += f"  {o['id']}: {o['product']} ¥{o['amount']} [{o['status']}]\n"
    return output

"""安全 Hook 和权限控制"""

import re
import datetime
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny, ToolPermissionContext

# 审计日志
audit_log: list[dict] = []


async def pii_mask(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """自动脱敏 PII"""
    response = str(input_data.get("tool_response", ""))

    # 手机号
    masked = re.sub(r'(\d{3})\d{4}(\d{4})', r'\1****\2', response)
    # 邮箱
    masked = re.sub(
        r'([a-zA-Z0-9])[a-zA-Z0-9.]*([a-zA-Z0-9])@',
        r'\1***\2@', masked
    )

    if masked != response:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "updatedMCPToolOutput": masked,
            }
        }
    return {}


async def audit_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """审计日志"""
    audit_log.append({
        "time": datetime.datetime.now().strftime("%H:%M:%S"),
        "tool": input_data.get("tool_name", "?"),
        "input": str(input_data.get("tool_input", ""))[:100],
    })
    return {}


async def permission_callback(
    tool_name: str,
    input_data: dict,
    context: ToolPermissionContext
) -> PermissionResultAllow | PermissionResultDeny:
    """权限控制"""

    # 查询类：自动允许
    read_tools = [
        "mcp__support__search_faq",
        "mcp__support__get_ticket",
        "mcp__support__get_order",
        "mcp__support__search_customer",
        "Read", "Grep", "Glob",
    ]
    if tool_name in read_tools:
        return PermissionResultAllow()

    # 创建工单：允许，但 critical 降级为 high
    if tool_name == "mcp__support__create_ticket":
        if input_data.get("priority") == "critical":
            modified = input_data.copy()
            modified["priority"] = "high"
            return PermissionResultAllow(updated_input=modified)
        return PermissionResultAllow()

    # 更新工单：允许
    if tool_name == "mcp__support__update_ticket":
        return PermissionResultAllow()

    # 其他：默认允许
    return PermissionResultAllow()


def get_audit_summary() -> str:
    """获取审计摘要"""
    if not audit_log:
        return "本次会话无工具调用"
    lines = [f"共 {len(audit_log)} 次工具调用:"]
    for entry in audit_log[-15:]:
        lines.append(f"  [{entry['time']}] {entry['tool']}")
    return "\n".join(lines)

"""客服 Agent 团队定义"""

from claude_agent_sdk import AgentDefinition

classifier = AgentDefinition(
    description="快速分类工单：判断类型、紧急度、情绪、应分配团队",
    prompt="""你是工单分类专家。分析工单内容，输出 JSON：
{"category": "bug/feature_request/billing/account/consulting/complaint",
 "urgency": "critical/high/medium/low",
 "sentiment": "angry/frustrated/neutral/positive",
 "team": "engineering/support/billing/sales"}
只输出 JSON，不要其他内容。""",
    tools=["Read"],
    model="haiku",
)

support_agent = AgentDefinition(
    description="回答客户问题，查询知识库和工单系统",
    prompt="""你是专业客服"小智"。规则：
1. 友好、专业、简洁
2. 主动使用工具查询信息
3. 信息不够就追问
4. 遇到解决不了的问题说"我帮您转人工"
5. 用中文，每次不超过 3 句话""",
    tools=["Read", "Bash"],
    model="sonnet",
)

quality_checker = AgentDefinition(
    description="检查客服回复质量：礼貌性、准确性、完整性",
    prompt="""你是质检员。评估回复质量：
- 评分: 1-10
- 问题: 发现的问题
- 建议: 改进建议
用中文简洁输出。""",
    tools=["Read"],
    model="haiku",
)

ALL_AGENTS = {
    "classifier": classifier,
    "support-agent": support_agent,
    "quality-checker": quality_checker,
}

"""智能客服系统 —— 主程序"""

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, ResultMessage, TextBlock, ToolUseBlock,
    create_sdk_mcp_server,
)
from claude_agent_sdk.types import HookMatcher

from config import SYSTEM_PROMPT
from tools.knowledge import search_faq
from tools.tickets import (
    get_ticket, create_ticket, update_ticket,
    get_order, search_customer,
)
from hooks.safety import (
    pii_mask, audit_hook, permission_callback, get_audit_summary,
)
from agents.team import ALL_AGENTS

# 创建 MCP 服务器
support_server = create_sdk_mcp_server(
    name="support",
    version="1.0.0",
    tools=[search_faq, get_ticket, create_ticket,
           update_ticket, get_order, search_customer],
)


async def main():
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM_PROMPT,
        mcp_servers={"support": support_server},
        allowed_tools=[
            "mcp__support__search_faq",
            "mcp__support__get_ticket",
            "mcp__support__create_ticket",
            "mcp__support__update_ticket",
            "mcp__support__get_order",
            "mcp__support__search_customer",
        ],
        agents=ALL_AGENTS,
        hooks={
            "PostToolUse": [
                HookMatcher(matcher=None, hooks=[pii_mask]),
                HookMatcher(matcher=None, hooks=[audit_hook]),
            ],
        },
        can_use_tool=permission_callback,
    )

    print("=" * 50)
    print("🤖 智能客服系统 v1.0")
    print("=" * 50)
    print("输入 quit 退出 | 输入 log 查看操作日志")
    print()

    async with ClaudeSDKClient(options=options) as client:
        while True:
            user_input = input("客户: ").strip()

            if user_input.lower() in ['quit', 'exit', 'q']:
                print(f"\n📋 {get_audit_summary()}")
                print("👋 系统已关闭")
                break

            if user_input.lower() == 'log':
                print(f"\n📋 {get_audit_summary()}\n")
                continue

            if not user_input:
                continue

            await client.query(user_input)

            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"\n🤖 小智: {block.text}")
                        elif isinstance(block, ToolUseBlock):
                            print(f"  🔧 {block.name}")
                elif isinstance(msg, ResultMessage):
                    if msg.total_cost_usd and msg.total_cost_usd > 0:
                        print(f"  💰 ${msg.total_cost_usd:.4f}")
            print()


if __name__ == "__main__":
    asyncio.run(main)

# 安装依赖
pip install claude-agent-sdk

# 确保 API Key 已设置
export ANTHROPIC_API_KEY="sk-ant-你的密钥"

# 运行
python main.py

客户: 我密码忘了怎么办
  🔧 search_faq
🤖 小智: 您好！重置密码很简单：进入登录页，点击"忘记密码"...

客户: 帮我查一下工单 TK-001
  🔧 get_ticket
🤖 小智: 您的工单 TK-001 "登录异常"，当前状态是处理中...

客户: 我是张三，帮我看看我的订单
  🔧 search_customer
🤖 小智: 张三您好，找到您的记录：1个工单(TK-001)和1个订单(ORD-001)...

客户: 我要退款！你们服务太差了！！
  🔧 classifier
  🔧 support-agent
🤖 小智: 非常抱歉给您带来不好的体验。请问您的订单号是多少？...

客户: log
📋 共 4 次工具调用:
  [10:30:15] search_faq
  [10:30:45] get_ticket
  [10:31:12] search_customer
  [10:31:48] classifier

客户: quit
📋 审计摘要...
👋 系统已关闭

客户输入
  │
  ▼
主程序 (ClaudeSDKClient)
  │
  ├── 系统提示词 → 调度中心角色
  │
  ├── Agent 层 → 分类 / 客服 / 质检
  │   ├── classifier (Haiku) → 快速分类
  │   ├── support-agent (Sonnet) → 回答问题
  │   └── quality-checker (Haiku) → 质量检查
  │
  ├── 工具层 (MCP)
  │   ├── search_faq → 知识库
  │   ├── get_ticket / create / update → 工单
  │   ├── get_order → 订单
  │   └── search_customer → 客户搜索
  │
  ├── 安全层
  │   ├── PostToolUse: PII 脱敏
  │   ├── PostToolUse: 审计日志
  │   └── PermissionCallback: 权限控制
  │
  └── 输出 → 回复客户