第 3 课：工单自动分类
为什么先学分类？
客户发了一条消息过来，你的系统第一件事要干嘛？
分类。
分错了，技术问题发给了销售，账单问题发给了技术——客户等半天没人管，体验极差。
分类是智能客服的第一道关卡，也是最适合用 query() 来做的事：输入一个工单，输出一个分类结果，干净利落。
最简单的分类
import anyio
from claude_agent_sdk import (
query, ClaudeAgentOptions,
AssistantMessage, TextBlock,
)
async def classify_ticket(ticket_text: str):
"""给工单分个类"""
options = ClaudeAgentOptions(
system_prompt="""你是工单分类专家。根据工单内容，输出分类结果。
格式：
类别: xxx
紧急度: 高/中/低
分配: xxx团队""",
max_turns=1,
)
async for msg in query(prompt=ticket_text, options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(block.text)
# 测试
anyio.run(lambda: classify_ticket("你们的 API 返回 500 错误，已经影响线上业务了！"))
输出大概是：
类别: 技术故障
紧急度: 高
分配: 技术团队
结构化输出：用 JSON
纯文本不好解析。实际系统中，我们需要 JSON 格式的结果，方便程序处理。
import json
import anyio
from claude_agent_sdk import (
query, ClaudeAgentOptions,
AssistantMessage, ResultMessage, TextBlock,
)
CLASSIFIER_PROMPT = """你是工单分类系统。对每个工单进行分析，输出 JSON 格式的分类结果。
分类维度：
1. category（类别）: bug | feature_request | billing | account | consulting | complaint
2. urgency（紧急度）: critical | high | medium | low
3. team（分配团队）: engineering | support | billing | sales | account_mgmt
4. sentiment（用户情绪）: angry | frustrated | neutral | positive
5. summary（一句话摘要）: 20 字以内
只输出 JSON，不要其他内容。"""
async def classify_ticket(ticket_text: str) -> dict:
"""分类工单，返回结构化结果"""
options = ClaudeAgentOptions(
system_prompt=CLASSIFIER_PROMPT,
max_turns=1,
)
result_text = ""
async for msg in query(prompt=ticket_text, options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
result_text += block.text
# 解析 JSON
# AI 可能用 ```json 包裹，需要清理
clean = result_text.strip()
if clean.startswith("```"):
clean = clean.split("\n", 1)[1] # 去掉第一行
clean = clean.rsplit("```", 1)[0] # 去掉最后的 ```
return json.loads(clean)
async def main():
tickets = [
"你们的 API 返回 500 错误，已经影响线上业务了！赶紧处理！",
"请问企业版每年多少钱？有没有试用？",
"我上个月的账单多扣了 200 块，搞什么？",
"能不能加个批量导出功能？",
"App 在 iOS 17 上打开就闪退",
]
print("=" * 60)
print("工单自动分类系统")
print("=" * 60)
for i, ticket in enumerate(tickets, 1):
print(f"\n📋 工单 #{i}: {ticket}")
try:
result = await classify_ticket(ticket)
print(f" 类别: {result['category']}")
print(f" 紧急: {result['urgency']}")
print(f" 团队: {result['team']}")
print(f" 情绪: {result['sentiment']}")
print(f" 摘要: {result['summary']}")
except json.JSONDecodeError:
print(" ❌ JSON 解析失败")
except Exception as e:
print(f" ❌ 错误: {e}")
anyio.run(main)
用 output_format 强制 JSON
上面的方法依赖 AI "听话"地输出 JSON。更可靠的方式是用 SDK 的 output_format：
async def classify_ticket_strict(ticket_text: str) -> dict:
"""强制 JSON 输出的分类"""
options = ClaudeAgentOptions(
system_prompt=CLASSIFIER_PROMPT,
max_turns=1,
output_format={
"type": "json_schema",
"schema": {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["bug", "feature_request", "billing",
"account", "consulting", "complaint"],
},
"urgency": {
"type": "string",
"enum": ["critical", "high", "medium", "low"],
},
"team": {
"type": "string",
"enum": ["engineering", "support", "billing",
"sales", "account_mgmt"],
},
"sentiment": {
"type": "string",
"enum": ["angry", "frustrated", "neutral", "positive"],
},
"summary": {"type": "string"},
},
"required": ["category", "urgency", "team", "sentiment", "summary"],
},
},
)
result_text = ""
async for msg in query(prompt=ticket_text, options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
result_text += block.text
return json.loads(result_text)
output_format 的好处：AI 的输出一定是你定义的 JSON 格式，不会多输出废话。
批量处理工单
实际场景中，工单是源源不断来的。来做个批量处理器：
import json
import anyio
from claude_agent_sdk import (
query, ClaudeAgentOptions,
AssistantMessage, TextBlock,
)
CLASSIFIER_PROMPT = """你是工单分类系统。输出 JSON 格式：
{"category": "...", "urgency": "...", "team": "...", "sentiment": "...", "summary": "..."}
只输出 JSON。"""
async def classify_one(ticket_id: str, text: str) -> dict:
"""分类单个工单"""
options = ClaudeAgentOptions(
system_prompt=CLASSIFIER_PROMPT,
max_turns=1,
)
result_text = ""
async for msg in query(prompt=text, options=options):
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
result_text += block.text
clean = result_text.strip()
if clean.startswith("```"):
clean = clean.split("\n", 1)[1]
clean = clean.rsplit("```", 1)[0]
result = json.loads(clean)
result["ticket_id"] = ticket_id
return result
async def batch_classify(tickets: list[dict]) -> list[dict]:
"""批量分类工单"""
results = []
for ticket in tickets:
print(f" 处理 {ticket['id']}...", end=" ")
try:
result = await classify_one(ticket["id"], ticket["text"])
results.append(result)
print(f"✅ {result['category']} / {result['urgency']}")
except Exception as e:
print(f"❌ {e}")
results.append({"ticket_id": ticket["id"], "error": str(e)})
return results
async def main():
# 模拟工单队列
tickets = [
{"id": "TK-001", "text": "登录页面一直转圈，进不去"},
{"id": "TK-002", "text": "想了解一下你们的产品适不适合我们公司"},
{"id": "TK-003", "text": "上个月账单有问题，多收了钱"},
{"id": "TK-004", "text": "能加个数据导出为 Excel 的功能吗"},
{"id": "TK-005", "text": "线上服务全挂了！！！赶紧看！！"},
]
print("🚀 开始批量分类...\n")
results = await batch_classify(tickets)
# 按紧急度排序
urgency_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
results.sort(key=lambda x: urgency_order.get(x.get("urgency", "low"), 99))
print("\n" + "=" * 60)
print("分类结果（按紧急度排序）")
print("=" * 60)
for r in results:
if "error" in r:
print(f" {r['ticket_id']}: ❌ 分类失败")
else:
emoji = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🟢"}
print(f" {emoji.get(r['urgency'], '⚪')} {r['ticket_id']}: "
f"{r['category']} → {r['team']} | {r['summary']}")
anyio.run(main)
分类的最佳实践
1. 提示词要具体
# ❌ 太模糊
"给这个工单分个类"
# ✅ 具体明确
"""你是工单分类系统。分类维度：
- category: bug（软件故障）, feature_request（功能需求）, billing（账单问题）...
- urgency: critical（系统不可用）, high（严重影响使用）, medium（部分影响）, low（不影响使用）
..."""
2. 用 max_turns=1 控制轮次
分类是一次性的事，不需要多轮对话。max_turns=1 可以节省时间和费用。
3. 用便宜的模型做分类
分类不需要最强的模型，Haiku 就够了：
options = ClaudeAgentOptions(
system_prompt=CLASSIFIER_PROMPT,
max_turns=1,
model="claude-haiku-4-5", # 快速便宜
)
4. 错误处理
AI 不保证每次都完美输出 JSON。一定要有错误处理：
try:
result = json.loads(result_text)
except json.JSONDecodeError:
# 回退方案：用默认分类
result = {
"category": "unknown",
"urgency": "medium",
"team": "support",
"sentiment": "neutral",
"summary": "分类失败，需人工处理",
}
本课小结
query()非常适合工单分类这种"一进一出"的任务- 用 JSON 格式输出方便程序处理，
output_format可以强制 - 批量处理时逐个处理并记录结果
- 分类用便宜的模型就够了（Haiku）
- 一定要有错误处理和回退方案
课后练习
- 在分类维度中增加
language（语言）字段，自动检测工单是中文还是英文 - 写一个函数，把分类结果保存到 JSON 文件中
- 给批量处理加上计时器，统计平均每个工单的分类时间