Notebook

第12章 学生Notebook：客户咨询分流与待回复工作台

把AI能力组织成可交付流程

本Notebook完成以下任务：

  1. 加载咨询数据和服务指南
  2. 完成可行性四维评分
  3. 填写六段设计书
  4. 实现主题/紧急度分类路由
  5. 从服务指南检索证据
  6. 生成回复草稿并标记人工复核
  7. 处理边界案例和失败兜底
  8. 生成HTML工作台
  9. B档：修改一个分类规则或AI介入层级

准备：确认 data/inquiries.json 和 data/service_guide.json 存在。

1. 加载数据

读取客户咨询和服务指南。

import json
import re
import html
from pathlib import Path
from datetime import datetime

# 加载数据
data_dir = Path('../data')

with open(data_dir / 'inquiries.json', 'r', encoding='utf-8') as f:
    inquiries_data = json.load(f)
inquiries = inquiries_data['inquiries']

with open(data_dir / 'service_guide.json', 'r', encoding='utf-8') as f:
    guide = json.load(f)

print(f'咨询数量: {len(inquiries)}')
print(f'服务分类数量: {len(guide["categories"])}')
print()
print('前3条咨询:')
for inq in inquiries[:3]:
    print(f'  {inq["id"]} [{inq["urgency"]}] {inq["customer"]}: {inq["text"][:40]}...')
print()
print('服务指南分类:')
for cat in guide['categories']:
    print(f'  {cat["id"]}: {cat["name"]} (AI级别: {cat["ai_level"]})')
print()
print('升级规则:')
for level, topics in guide['escalation_rules'].items():
    print(f'  {level}: {topics}')

2. 可行性四维评分

在开始设计工作流之前，先评估这个任务的可行性。对每个维度打1-5分。

# 可行性四维评分
# 请根据你的理解填写评分（1-5分）和理由

feasibility = {
    '数据充分性': {
        'score': 4,  # 修改为你的评分
        'reason': '服务指南覆盖常见场景，但无法覆盖所有情况'
    },
    '技术可行性': {
        'score': 5,  # 修改为你的评分
        'reason': ''  # 填写你的理由
    },
    '风险可控性': {
        'score': 3,  # 修改为你的评分
        'reason': ''  # 填写你的理由
    },
    '可交接性': {
        'score': 4,  # 修改为你的评分
        'reason': ''  # 填写你的理由
    }
}

print('可行性四维评分:')
total = 0
for dim, info in feasibility.items():
    print(f'  {dim}: {info["score"]}/5')
    print(f'    理由: {info["reason"]}')
    total += info['score']
print(f'  总分: {total}/20')
print()
if total >= 14:
    print('评估结论: 可行性较高，可以推进')
elif total >= 10:
    print('评估结论: 有一定风险，需要重点关注薄弱环节')
else:
    print('评估结论: 风险较高，建议重新评估方案')

3. 六段设计书

填写完整的六段设计书。这是工作流工程化的核心文档。

design_doc = {
    '第1段_任务声明': {
        '做什么': '把客户咨询分流到正确的处理路径，生成回复草稿',
        '不做什么': '不真实发送、不做最终决策、不处理服务指南之外的问题',
        '输入': '客户咨询文本',
        '输出': '分类结果 + 证据 + 回复草稿 + 责任人'
    },
    '第2段_可行性评估': {
        '参考': '见上方四维评分'
    },
    '第3段_四段式工作流': {
        '输入': '客户咨询文本',
        'AI动作': '主题分类 + 紧急度判断 + 证据检索 + 草稿生成',
        '人工复核': '边界案例和失败案例由人工确认',
        '输出': '待回复清单（含状态、证据、责任人）'
    },
    '第4段_责任链': {
        '数据读取': '系统负责',
        '隐私脱敏': '系统自动 + 人工确认',
        '分类': 'AI提供候选，人工可覆盖',
        '证据检索': 'AI负责',
        '草稿生成': 'AI负责',
        '复核': '人工负责（边界/失败案例）',
        '最终发送': '人工负责（AI不发送）'
    },
    '第5段_失效模式': [
        {'场景': '分类错误', '影响': '咨询进错队列', '对策': '人工复核覆盖'},
        {'场景': '检索无结果', '影响': '无法生成草稿', '对策': '进入人工队列'},
        {'场景': '模型不可用', '影响': '无法生成草稿', '对策': '使用模板兜底'},
        {'场景': '客户发送敏感信息', '影响': '隐私泄露', '对策': '自动脱敏'},
        {'场景': '客户情绪激动', '影响': '回复不当激化矛盾', '对策': '强制人工队列'}
    ],
    '第6段_对外可解释性': {
        '模板': '您的咨询被分类为{类别}，根据规则{规则编号}，回复由AI生成并经{责任人}复核。如不满意可{升级路径}。'
    }
}

print('六段设计书:')
for section, content in design_doc.items():
    print(f'\n{section}:')
    if isinstance(content, dict):
        for k, v in content.items():
            print(f'  {k}: {v}')
    elif isinstance(content, list):
        for item in content:
            print(f'  - {item["场景"]}: {item["对策"]}')

4. 路由：主题/紧急度分类

根据服务指南的关键词，对每条咨询进行主题分类和紧急度判断。

运行：观察每条咨询的分类结果。

def classify_topic(text, categories):
    """根据关键词匹配服务指南分类"""
    scores = {}
    for cat in categories:
        score = 0
        for kw in cat.get('keywords', []):
            if kw in text:
                score += 1
        if score > 0:
            scores[cat['id']] = score
    
    if not scores:
        return 'CAT_UNKNOWN', 0
    
    best = max(scores, key=scores.get)
    return best, scores[best]

def classify_urgency(text, declared_urgency):
    """判断紧急度：结合声明紧急度和文本情绪词"""
    urgent_words = ['投诉', '欺诈', '赶紧', '马上', '立刻', '不满', '态度差',
                    '举报', '消费者协会', '差评']
    
    has_urgent_word = any(w in text for w in urgent_words)
    
    if declared_urgency == '紧急' or has_urgent_word:
        return '紧急'
    return '普通'

# 对所有咨询进行分类
classification_results = []
for inq in inquiries:
    topic_id, topic_score = classify_topic(inq['text'], guide['categories'])
    urgency = classify_urgency(inq['text'], inq['urgency'])
    
    # 找到对应的分类信息
    cat_info = next((c for c in guide['categories'] if c['id'] == topic_id), None)
    
    result = {
        'inquiry_id': inq['id'],
        'customer': inq['customer'],
        'text': inq['text'],
        'topic_id': topic_id,
        'topic_name': cat_info['name'] if cat_info else '未知',
        'topic_score': topic_score,
        'urgency': urgency,
        'ai_level': cat_info['ai_level'] if cat_info else 'human_only',
        'original_category': inq['category']
    }
    classification_results.append(result)

print(f'{"咨询ID":<8} {"主题":<12} {"紧急度":<6} {"AI级别":<18} {"匹配度"}')
print('-' * 65)
for r in classification_results:
    print(f'{r["inquiry_id"]:<8} {r["topic_name"]:<12} {r["urgency"]:<6} {r["ai_level"]:<18} {r["topic_score"]}')

思考：

  1. 哪些咨询的分类匹配度为0？它们应该进哪个队列？
  2. INQ05（信息不足）被分到了哪个类别？为什么？
  3. INQ08（含敏感信息）的分类正确吗？还需要做什么额外处理？

在下方写下你的观察。

# 我的观察：
# 
# 匹配度为0的咨询：
# 
# INQ05 被分到了：
# 
# INQ08 还需要：

5. 证据检索：从服务指南找依据

对每条咨询，从服务指南中匹配最相关的规则作为回复依据。

运行：观察每条咨询匹配到的规则。

def retrieve_evidence(text, topic_id, categories):
    """从服务指南检索匹配的规则"""
    cat_info = next((c for c in categories if c['id'] == topic_id), None)
    if not cat_info:
        return None, '无匹配分类'
    
    rules = cat_info.get('rules', [])
    if not rules:
        return None, '分类下无规则'
    
    # 简单策略：返回第一条规则作为候选
    # 实际系统可以用更复杂的匹配
    # 对于退换货，检查是否包含质量/发错关键词
    if topic_id == 'CAT_RETURN':
        if any(w in text for w in ['发错', '质量', '不对', '破损']):
            return next((r for r in rules if r['rule_id'] == 'RET002'), rules[0]), '匹配商家责任规则'
    
    # 对于物流，检查是否有订单号
    if topic_id == 'CAT_LOGISTICS':
        has_order = bool(re.search(r'\d{10,}', text))
        if has_order:
            return rules[0], '有订单号，匹配物流查询规则'
    
    # 默认返回第一条规则
    return rules[0], f'匹配{cat_info["name"]}默认规则'

# 为每条咨询检索证据
for r in classification_results:
    rule, match_reason = retrieve_evidence(r['text'], r['topic_id'], guide['categories'])
    r['evidence_rule'] = rule
    r['match_reason'] = match_reason

print(f'{"咨询ID":<8} {"匹配规则":<10} {"需要人工":<8} {"匹配原因"}')
print('-' * 70)
for r in classification_results:
    rule = r['evidence_rule']
    rule_id = rule['rule_id'] if rule else '无'
    need_human = '是' if (rule and rule.get('need_human')) else '否'
    print(f'{r["inquiry_id"]:<8} {rule_id:<10} {need_human:<8} {r["match_reason"]}')

6. 隐私脱敏

检查咨询文本中是否包含敏感信息，如有则脱敏处理。

运行：观察哪些咨询触发了脱敏。

def check_and_mask_privacy(text, privacy_rules):
    """检查并脱敏敏感信息"""
    masked_text = text
    found_sensitive = []
    
    patterns = privacy_rules.get('sensitive_patterns', [])
    
    # 手机号
    phone_pattern = r'1[3-9]\d{9}'
    if re.search(phone_pattern, masked_text):
        masked_text = re.sub(phone_pattern, '[已脱敏:手机号]', masked_text)
        found_sensitive.append('手机号')
    
    # 身份证号
    id_pattern = r'\d{17}[\dXx]'
    if re.search(id_pattern, masked_text):
        masked_text = re.sub(id_pattern, '[已脱敏:身份证号]', masked_text)
        found_sensitive.append('身份证号')
    
    # 银行卡号（16-19位数字）
    bank_pattern = r'\d{16,19}'
    if re.search(bank_pattern, masked_text) and '已脱敏' not in masked_text:
        masked_text = re.sub(bank_pattern, '[已脱敏:银行卡号]', masked_text)
        found_sensitive.append('银行卡号')
    
    return masked_text, found_sensitive

# 对所有咨询进行脱敏检查
for r in classification_results:
    masked, found = check_and_mask_privacy(r['text'], guide['privacy_rules'])
    r['masked_text'] = masked
    r['sensitive_found'] = found

print('隐私脱敏结果:')
for r in classification_results:
    if r['sensitive_found']:
        print(f'  {r["inquiry_id"]}: 发现 {", ".join(r["sensitive_found"])}')
        print(f'    原文: {r["text"][:50]}...')
        print(f'    脱敏: {r["masked_text"][:50]}...')
        print()

if not any(r['sensitive_found'] for r in classification_results):
    print('  未发现敏感信息')

7. 草稿生成与人工复核队列

根据AI介入级别和规则模板生成回复草稿。边界案例和失败案例强制进入人工队列。

运行：观察每条咨询的处理路径。

def generate_draft(r, guide):
    """根据分类和规则生成回复草稿"""
    result = dict(r)  # copy
    
    # 判断是否需要强制人工
    force_human = False
    force_reason = ''
    
    # 边界案例：信息不足
    if r['topic_score'] == 0:
        force_human = True
        force_reason = '分类匹配度为0，信息可能不足'
    
    # 边界案例：含敏感信息
    if r['sensitive_found']:
        force_human = True
        force_reason = f'含敏感信息({", ".join(r["sensitive_found"])})，需人工确认脱敏'
    
    # 边界案例：原始分类为boundary或failure
    if r['original_category'] == 'boundary':
        # 检查是否属于升级规则中的立即人工类别
        if r['topic_name'] in guide['escalation_rules'].get('immediate_human', []):
            force_human = True
            force_reason = f'{r["topic_name"]}属于立即人工处理类别'
    
    if r['original_category'] == 'failure':
        force_human = True
        force_reason = '技术故障类咨询，AI系统可能不可用，需人工兜底'
    
    # 规则要求人工
    rule = r.get('evidence_rule')
    if rule and rule.get('need_human'):
        force_human = True
        force_reason = force_reason or rule.get('human_reason', '规则要求人工处理')
    
    # 生成草稿
    if force_human:
        result['queue'] = '人工队列'
        result['force_reason'] = force_reason
        if rule:
            result['draft'] = rule.get('response_template', '需要人工处理')
        else:
            result['draft'] = '感谢您的留言。为了更快帮您解决问题，请补充以下信息：1）订单号；2）具体问题描述。'
        result['status'] = '待人工处理'
        result['responsible'] = '人工客服'
    else:
        result['queue'] = 'AI草拟'
        result['force_reason'] = ''
        if rule:
            result['draft'] = rule.get('response_template', '正在为您处理')
        else:
            result['draft'] = '正在为您查询，请稍候'
        
        if r['ai_level'] == 'auto_draft':
            result['status'] = '待发送'
            result['responsible'] = 'AI（已记录日志）'
        elif r['ai_level'] == 'draft_with_review':
            result['status'] = '待人工复核'
            result['responsible'] = '人工复核'
        else:
            result['status'] = '待人工处理'
            result['responsible'] = '人工客服'
            result['queue'] = '人工队列'
    
    return result

# 生成所有草稿
draft_results = [generate_draft(r, guide) for r in classification_results]

print(f'{"咨询ID":<8} {"队列":<10} {"状态":<12} {"责任人":<14} {"强制原因"}')
print('-' * 80)
for r in draft_results:
    print(f'{r["inquiry_id"]:<8} {r["queue"]:<10} {r["status"]:<12} {r["responsible"]:<14} {r["force_reason"][:30]}')

思考：

  1. 有多少条咨询进入了人工队列？占比多少？
  2. INQ03（退款争议）和INQ06（投诉）为什么必须人工处理？
  3. 如果模型不可用，哪些咨询还能处理？哪些不能？

在下方写下你的分析。

# 我的分析：
# 
# 进入人工队列的咨询数量和占比：
# 
# INQ03和INQ06必须人工的原因：
# 
# 模型不可用时的影响：

8. HTML工作台

生成一个HTML工作台，展示所有咨询的处理状态、证据和责任人。

运行：生成HTML文件并在浏览器中查看。

def generate_html_workbench(draft_results, output_path):
    """生成HTML工作台"""
    rows = ''
    for r in draft_results:
        rule_id = r['evidence_rule']['rule_id'] if r.get('evidence_rule') else '无'
        
        # 状态颜色
        status_colors = {
            '待发送': '#4CAF50',
            '待人工复核': '#FF9800',
            '待人工处理': '#f44336'
        }
        color = status_colors.get(r['status'], '#999')
        
        # 脱敏标记
        sensitive_mark = ' ⚠️含敏感信息' if r['sensitive_found'] else ''
        
        rows += f'''
        <tr>
            <td>{r['inquiry_id']}</td>
            <td>{html.escape(r['customer'])}</td>
            <td>{html.escape(r['text'][:50])}...</td>
            <td>{r['topic_name']}</td>
            <td>{r['urgency']}</td>
            <td>{rule_id}</td>
            <td style="color:{color};font-weight:bold">{r['status']}{sensitive_mark}</td>
            <td>{r['responsible']}</td>
            <td>{html.escape(r.get('force_reason', '')[:30])}</td>
        </tr>'''
    
    html_content = f'''<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>客户咨询分流工作台 - 第12章</title>
    <style>
        body {{ font-family: "Microsoft YaHei", sans-serif; margin: 20px; background: #f5f5f5; }}
        h1 {{ color: #333; }}
        .summary {{ background: #fff; padding: 15px; border-radius: 8px; margin-bottom: 20px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        table {{ width: 100%; border-collapse: collapse; background: #fff; border-radius: 8px; overflow: hidden; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }}
        th {{ background: #1976D2; color: white; padding: 12px 8px; text-align: left; font-size: 14px; }}
        td {{ padding: 10px 8px; border-bottom: 1px solid #eee; font-size: 13px; }}
        tr:hover {{ background: #f0f7ff; }}
        .footer {{ margin-top: 20px; color: #666; font-size: 12px; }}
        .warning {{ color: #f44336; }}
    </style>
</head>
<body>
    <h1>📋 客户咨询分流工作台</h1>
    <div class="summary">
        <p><strong>店铺：</strong>{guide['store_name']}</p>
        <p><strong>咨询总数：</strong>{len(draft_results)}</p>
        <p><strong>待发送：</strong>{sum(1 for r in draft_results if r['status'] == '待发送')}</p>
        <p><strong>待人工复核：</strong>{sum(1 for r in draft_results if r['status'] == '待人工复核')}</p>
        <p><strong>待人工处理：</strong>{sum(1 for r in draft_results if r['status'] == '待人工处理')}</p>
        <p><strong>生成时间：</strong>{datetime.now().strftime('%Y-%m-%d %H:%M')}</p>
        <p class="warning"><strong>⚠️ 注意：本工作台仅展示草稿，绝不真实发送任何回复。</strong></p>
    </div>
    <table>
        <tr>
            <th>编号</th>
            <th>客户</th>
            <th>咨询内容</th>
            <th>主题</th>
            <th>紧急度</th>
            <th>规则</th>
            <th>状态</th>
            <th>责任人</th>
            <th>备注</th>
        </tr>
        {rows}
    </table>
    <div class="footer">
        <p>第12章教学资源 | 所有数据为虚构 | AI不承担最终决策责任</p>
    </div>
</body>
</html>'''
    
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(html_content)
    return output_path

output_path = generate_html_workbench(draft_results, '../outputs/ch12_workbench.html')
print(f'HTML工作台已生成: {output_path}')
print()
print('请在浏览器中打开查看。')
print()

# 统计
print('工作台统计:')
print(f'  待发送: {sum(1 for r in draft_results if r["status"] == "待发送")} 条')
print(f'  待人工复核: {sum(1 for r in draft_results if r["status"] == "待人工复核")} 条')
print(f'  待人工处理: {sum(1 for r in draft_results if r["status"] == "待人工处理")} 条')

9. 审计日志

生成每条咨询的处理日志，用于追溯和交接。

def generate_audit_log(draft_results):
    """生成审计日志"""
    logs = []
    for r in draft_results:
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'inquiry_id': r['inquiry_id'],
            'customer': r['customer'],
            'topic': r['topic_name'],
            'urgency': r['urgency'],
            'classification_score': r['topic_score'],
            'sensitive_info': r['sensitive_found'],
            'evidence_rule': r['evidence_rule']['rule_id'] if r.get('evidence_rule') else None,
            'ai_level': r['ai_level'],
            'queue': r['queue'],
            'status': r['status'],
            'responsible': r['responsible'],
            'force_reason': r.get('force_reason', ''),
            'draft_generated': bool(r.get('draft'))
        }
        logs.append(log_entry)
    return logs

audit_logs = generate_audit_log(draft_results)

print('审计日志（前3条）:')
for log in audit_logs[:3]:
    print(f'  [{log["timestamp"][:19]}] {log["inquiry_id"]} → {log["topic"]} → {log["status"]} ({log["responsible"]})')
    if log['sensitive_info']:
        print(f'    ⚠️ 敏感信息: {log["sensitive_info"]}')
    if log['force_reason']:
        print(f'    强制原因: {log["force_reason"]}')
print(f'  ... 共 {len(audit_logs)} 条日志')

10. B档修改：修改一个分类规则或AI介入层级

从以下选项中选择一项修改，修改前先写预测。

选项A：修改一个分类类别

  • 例如：把"退换货"的关键词增加或减少
  • 预测：哪些咨询的分类会改变？

选项B：修改一个AI介入层级

  • 例如：把"退换货"从 draft_with_review 改为 human_only
  • 预测：哪些咨询的路径会改变？误分成本有什么影响？

选项C：修改边界规则

  • 例如：把投诉的升级规则从"立即人工"改为"AI草拟+人工复核"
  • 预测：风险和效率如何变化？
# 我选择：选项___
# 
# 修改前预测：
# 
# 修改内容：

# === 示例：选项B - 修改AI介入层级 ===
# 取消下面的注释并修改

# # 修改前：记录原始状态
# original_levels = {cat['id']: cat['ai_level'] for cat in guide['categories']}
# print('修改前AI介入级别:')
# for cid, level in original_levels.items():
#     print(f'  {cid}: {level}')
# 
# # 修改：把退换货改为human_only
# for cat in guide['categories']:
#     if cat['id'] == 'CAT_RETURN':
#         cat['ai_level'] = 'human_only'  # 原来是 draft_with_review
# 
# # 重新运行分类和草稿生成
# # ... (重新调用上面的函数)
# 
# # 对比修改前后的差异
# print('\n修改后受影响咨询:')
# # 找出路径改变的咨询

# 修改后的影响分析：
# 
# 哪些咨询路径改变了：
# 
# 误分成本影响：
# 
# 责任链变化：

11. 人工改写草稿

选择至少一条AI生成的回复草稿，进行人工改写，并标记修改依据。

# 选择一条草稿进行人工改写
# 请选择一条你认为需要改进的回复

print('可改写的草稿:')
for i, r in enumerate(draft_results):
    print(f'  {i+1}. {r["inquiry_id"]} ({r["topic_name"]}):')
    print(f'     客户: {r["text"][:40]}...')
    print(f'     草稿: {r["draft"][:60]}...')
    print()

# 我选择改写第___条草稿
# 
# 原始草稿：
# 
# 改写后：
# 
# 修改依据：
# （例如：原草稿语气过于机械/缺少共情/信息不完整/不符合实际情况）

12. 验收

运行下面的检查，确认你完成了所有必做项。

checks = {
    '加载了咨询数据和服务指南': True,
    '完成了可行性四维评分': False,  # 检查第2节是否填写了理由
    '填写了六段设计书': True,  # 已预填，学生可补充
    '运行了主题/紧急度分类': True,
    '运行了证据检索': True,
    '运行了隐私脱敏': True,
    '生成了回复草稿': True,
    '识别了边界案例': True,
    '生成了HTML工作台': True,
    '完成了B档修改': False,  # 检查第10节
    '人工改写了至少一条草稿': False,  # 检查第11节
}

print('验收清单:')
for check, status in checks.items():
    print(f'  [{"✓" if status else " "}] {check}')

print()
print('请手动把 False 改为 True，确认你完成了对应项目。')
print()
print('关键概念检查:')
print('  1. 四段式工作流的四个阶段是 ___')
print('  2. AI介入级别有三种，分别是 ___')
print('  3. 责任链的作用是 ___')
print('  4. 失败兜底的含义是 ___')
print('  5. 为什么AI不能真实发送回复？___')