先理解：Agent 项目先建工作流，再谈智能
工单 Agent 不是“把工单丢给大模型”。它要先理解工单有哪些字段、有哪些状态、哪些动作允许自动执行、哪些动作必须人工审核。
本篇先定义 Schema 和处理流程。这样后续工具调用、知识检索、护栏、评测都有稳定输入输出。
为什么 Schema 很重要
没有 Schema，Agent 每次可能用不同格式表达分类、优先级、处理建议。系统无法统计，也无法评测。Schema 的意义是把模型输出变成工程系统能消费的数据。
工作流先简单
第一版只做分诊：识别类别、判断优先级、给出下一步动作。不要一开始就让 Agent 自动退款、自动改数据。自动化权限应该随着评测结果逐步放开。
本系列会完成一个工单分诊 Agent：读取工单、分类、判断优先级、调用工具、生成处理草稿，并对高风险工单转人工。
本篇成品
text
ticket-agent/
src/schema.py
src/workflow.py
data/tickets.jsonl初始化
bash
mkdir -p ticket-agent/src ticket-agent/data
cd ticket-agent
python -m venv .venv
source .venv/bin/activate
pip install pydantic fastapi uvicorn样例工单
data/tickets.jsonl：
json
{"ticket_id":"T001","title":"付款后一直没发货","body":"客户已等待 5 天，要求处理","customer_level":"vip","product_line":"headphone","history_count":2}
{"ticket_id":"T002","title":"想开发票","body":"用户咨询电子发票入口","customer_level":"normal","product_line":"all","history_count":0}
{"ticket_id":"T003","title":"要求赔偿","body":"客户威胁投诉并要求赔偿 500 元","customer_level":"normal","product_line":"speaker","history_count":3}Schema
src/schema.py：
python
from pydantic import BaseModel
class Ticket(BaseModel):
ticket_id: str
title: str
body: str
customer_level: str = "normal"
product_line: str = "all"
history_count: int = 0
class TriageResult(BaseModel):
ticket_id: str
category: str
priority: str
team: str
risk_level: str
next_action: str工作流规则
src/workflow.py：
python
from src.schema import Ticket, TriageResult
def classify(ticket: Ticket) -> str:
text = ticket.title + ticket.body
if "发票" in text:
return "invoice"
if "发货" in text or "物流" in text:
return "shipping"
if "赔偿" in text or "投诉" in text:
return "complaint"
return "other"
def priority(ticket: Ticket, category: str) -> str:
if category == "complaint" or ticket.customer_level == "vip":
return "P1"
if ticket.history_count >= 2:
return "P2"
return "P3"
def team_for(category: str) -> str:
return {
"invoice": "finance_support",
"shipping": "customer_service",
"complaint": "senior_support",
}.get(category, "customer_service")
def risk_for(category: str) -> str:
return "high" if category == "complaint" else "low"
def triage(ticket: Ticket) -> TriageResult:
category = classify(ticket)
risk = risk_for(category)
return TriageResult(
ticket_id=ticket.ticket_id,
category=category,
priority=priority(ticket, category),
team=team_for(category),
risk_level=risk,
next_action="prepare_draft" if risk == "low" else "human_review",
)验收
bash
python - <<'PY'
import json
from pathlib import Path
from src.schema import Ticket
from src.workflow import triage
for line in Path("data/tickets.jsonl").read_text().splitlines():
print(triage(Ticket(**json.loads(line))))
PY确认：
T001是shipping且P1。T002是invoice。T003是complaint且human_review。
