通过工单分类Agent的开发过程,总结了系统提示词工程、输出验证、Agent设计等关键环节的常见陷阱和解决方案。
我最近上线了一个内部工单分诊机器人,用来减少值班告警噪音。它对传入的工单进行分类、起草内部备注,并将其路由到正确的团队。在开发过程中,我踩中了大多数 LLM 项目都会遇到的坑,所以这篇教程把我最终做成的 Agent 掰开揉碎,一个坑一个坑地讲。
需要准备的环境:一个 Oxlo.ai API Key(从 https://portal.oxlo.ai 获取),以及 OpenAI SDK 和几个辅助库:pip install openai pydantic tenacity
我第一版用的系统提示词非常模糊,模型在 markdown、JSON 和纯文本之间来回横跳。解决方案是写一份刚性很强的提示词,提前定义好角色定位、输出规则和边界情况。
import os
from openai import OpenAI
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
SYSTEM_PROMPT = """You are a support ticket triage agent.
Your job is to read a customer ticket and produce a structured analysis.
Rules:
- Classify urgency as low, medium, or high.
- Assign to exactly one team: billing, technical, or account-management.
- Write a concise internal note summarizing the issue and suggested next step.
- Set category to spam if the message lacks a concrete support request.
- Respond with valid JSON containing these keys: urgency, team, category, internal_note, confidence.
- confidence must be a float between 0.0 and 1.0 representing your certainty."""
ticket = "I was charged twice for my subscription this month. Please fix this immediately."
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": ticket},
],
)
print(response.choices[0].message.content)
第二个经典坑是用正则表达式解析自由文本。我切换到 JSON 模式并加了 Pydantic 模型,任何格式不匹配都会在进入路由逻辑之前快速失败。
import os
import json
from openai import OpenAI
from pydantic import BaseModel, Field
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
class TicketAnalysis(BaseModel):
urgency: str = Field(pattern="^(low|medium|high)$")
team: str = Field(pattern="^(billing|technical|account-management)$")
category: str = Field(pattern="^(spam|actionable)$")
internal_note: str
confidence: float = Field(ge=0.0, le=1.0)
SYSTEM_PROMPT = """You are a support ticket triage agent.
Your job is to read a customer ticket and produce a structured analysis.
Rules:
- Classify urgency as low, medium, or high.
- Assign to exactly one team: billing, technical, or account-management.
- Write a concise internal note summarizing the issue and suggested next step.
- Set category to spam if the message lacks a concrete support request.
- Respond with valid JSON containing these keys: urgency, team, category, internal_note, confidence.
- confidence must be a float between 0.0 and 1.0 representing your certainty."""
ticket = "I was charged twice for my subscription this month. Please fix this immediately."
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Analyze this ticket and return JSON:\n\n{ticket}"},
],
response_format={"type": "json_object"},
)
parsed = TicketAnalysis.model_validate_json(response.choices[0].message.content)
print(parsed.model_dump_json(indent=2))
第三个坑是无脑截断。我以前为了不超过 token 预算会把日志切片丢弃。由于 Oxlo.ai 采用扁平化的按请求计费方式,成本不随输入长度增长,所以我不再预先截断,而是开始发送完整的对话历史。不过我仍然会设置一个安全的字符数上限,以尊重模型的上下文窗口,但成本不再是截断的理由。
import os
import json
from openai import OpenAI
from pydantic import BaseModel, Field
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
class TicketAnalysis(BaseModel):
urgency: str = Field(pattern="^(low|medium|high)$")
team: str = Field(pattern="^(billing|technical|account-management)$")
category: str = Field(pattern="^(spam|actionable)$")
internal_note: str
confidence: float = Field(ge=0.0, le=1.0)
SYSTEM_PROMPT = """You are a support ticket triage agent.
Your job is to read a customer ticket and produce a structured analysis.
Rules:
- Classify urgency as low, medium, or high.
- Assign to exactly one team: billing, technical, or account-management.
- Write a concise internal note summarizing the issue and suggested next step.
- Set category to spam if the message lacks a concrete support request.
- Respond with valid JSON containing these keys: urgency, team, category, internal_note, confidence.
- confidence must be a float between 0.0 and 1.0 representing your certainty."""
def analyze_ticket(ticket_text: str) -> TicketAnalysis:
MAX_CHARS = 100000
if len(ticket_text) > MAX_CHARS:
ticket_text = ticket_text[:MAX_CHARS] + "\n[truncated]"
response = client.chat.completions.create(
model="qwen-3-32b",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Analyze this ticket and return JSON:\n\n{ticket_text}"},
],
response_format={"type": "json_object"},
)
return TicketAnalysis.model_validate_json(response.choices[0].message.content)
ticket = "I was charged twice for my subscription this month. Please fix this immediately."
print(analyze_ticket(ticket).model_dump_json(indent=2))
单次超时或限速错误不应该导致工单丢失。我用 tenacity 包装了每一次调用,偶发错误会自动重试。
import os
import json
from openai import OpenAI
from pydantic import BaseModel, Field
from tenacity import retry, stop_after_attempt, wait_exponential
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
class TicketAnalysis(BaseModel):
urgency: str = Field(pattern="^(low|medium|high)$")
team: str = Field(pattern="^(billing|technical|account-management)$")
category: str = Field(pattern="^(spam|actionable)$")
internal_note: str
confidence: float = Field(ge=0.0, le=1.0)
SYSTEM_PROMPT = """You are a support ticket triage agent.
Your job is to read a customer ticket and produce a structured analysis.
Rules:
- Classify urgency as low, medium, or high.
- Assign to exactly one team: billing, technical, or account-management.
- Write a concise internal note summarizing the issue and suggested next step.
- Set category to spam if the message lacks a concrete support request.
- Respond with valid JSON containing these keys: urgency, team, category, internal_note, confidence.
- confidence must be a float between 0.0 and 1.0 representing your certainty."""
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_oxlo(messages, response_format=None):
return client.chat.completions.create(
model="qwen-3-32b",
messages=messages,
response_format=response_format,
)
def analyze_ticket(ticket_text: str) -> TicketAnalysis:
MAX_CHARS = 100000
if len(ticket_text) > MAX_CHARS:
ticket_text = ticket_text[:MAX_CHARS] + "\n[truncated]"
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Analyze this ticket and return JSON:\n\n{ticket_text}"},
]
response = call_oxlo(messages, response_format={"type": "json_object"})
return TicketAnalysis.model_validate_json(response.choices[0].message.content)
ticket = "I was charged twice for my subscription this month. Please fix this immediately."
print(analyze_ticket(ticket).model_dump_json(indent=2))
最后一个坑是把 LLM 当确定性 API 看待。即使有了 JSON 模式,confidence 分数也可能很低。我对所有低于 0.7 或者高紧急程度的工单都加上人工审核门控。
import os
import json
from openai import OpenAI
from pydantic import BaseModel, Field
from tenacity import retry, stop_after_attempt, wait_exponential
client = OpenAI(
base_url="https://api.oxlo.ai/v1",
api_key=os.getenv("OXLO_API_KEY")
)
class TicketAnalysis(BaseModel):
urgency: str = Field(pattern="^(low|medium|high)$")
team: str = Field(pattern="^(billing|technical|account-management)$")
category: str = Field(pattern="^(spam|actionable)$")
internal_note: str
confidence: float = Field(ge=0.0, le=1.0)
SYSTEM_PROMPT = """You are a support ticket triage agent.
Your job is to read a customer ticket and produce a structured analysis.
Rules:
- Classify urgency as low, medium, or high.
- Assign to exactly one team: billing, technical, or account-management.
- Write a concise internal note summarizing the issue and suggested next step.
- Set category to spam if the message lacks a concrete support request.
- Respond with valid JSON containing these keys: urgency, team, category, internal_note, confidence.
- confidence must be a float between 0.0 and 1.0 representing your certainty."""
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_oxlo(messages, response_format=None):
return client.chat.completions.create(
model="qwen-3-32b",
messages=messages,
response_format=response_format,
)
def analyze_ticket(ticket_text: str) -> TicketAnalysis:
MAX_CHARS = 100000
if len(ticket_text) > MAX_CHARS:
ticket_text = ticket_text[:MAX_CHARS] + "\n[truncated]"
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": f"Analyze this ticket and return JSON:\n\n{ticket_text}"},
]
response = call_oxlo(messages, response_format={"type": "json_object"})
return TicketAnalysis.model_validate_json(response.choices[0].message.content)
def triage_ticket(ticket_text: str) -> dict:
analysis = analyze_ticket(ticket_text)
result = {
"urgency": analysis.urgency,
"team": analysis.team,
"internal_note": analysis.internal_note,
"confidence": analysis.confidence,
"requires_human_review": False,
}
if analysis.confidence < 0.7 or analysis.urgency == "high":
result["requires_human_review"] = True
return result
ticket = "I was charged twice for my subscription this month. Please fix this immediately."
print(json.dumps(triage_ticket(ticket), indent=2))
经过用一百条线上工单调优之后,这是我最终用下来的提示词。它足够具体,能消除歧义,但又足够短,最小化了注入攻击面。
SYSTEM_PROMPT = """You are a support ticket triage agent.
Your job is to read a customer ticket and produce a structured analysis.
Rules:
- Classify urgency as low, medium, or high.
- Assign to exactly one team: billing, technical, or account-management.
- Write a concise internal note summarizing the issue and suggested next step.
- Set category to spam if the message lacks a concrete support request.
- Respond with valid JSON containing these keys: urgency, team, category, internal_note, confidence.
- confidence must be a float between 0.0 and 1.0 representing your certainty."""
下面是把所有逻辑串起来的完整脚本。导出你的 Oxlo.ai Key,然后用一条真实感十足的工单跑一遍全流程。
if __name__ == "__main__":
ticket = """Subject: Urgent - API returning 502s since 09:00 UTC
Our production webhook endpoint has been failing since this morning.
Every POST to /v1/events returns a 502 Bad Gateway.
This is blocking our checkout flow. We need an ETA on the fix immediately.
- Account: ACME Corp
- Region: us-east-1
- Error rate: 100% since 09:00 UTC
"""
decision = triage_ticket(ticket)
print(json.dumps(decision, indent=2))
运行结果:
{
"urgency": "high",
"team": "technical",
"internal_note": "Customer reports total outage on POST /v1/events since 09:00 UTC. Route to infrastructure on-call immediately and provide customer ETA.",
"confidence": 0.96,
"requires_human_review": true
}
把这个函数接入你的工单系统入站 Webhook,让每条新消息都经过它的处理。如果你想实验不同的模型行为,可以换用 Llama 3.3 70B 来提升通用场景的准确率,或者用 DeepSeek V3.2 处理编码相关的工单。两者都可以在 Oxlo.ai 上使用同样的扁平按请求计价和零冷启动,因此你可以随意 A/B 测试,而无需重构成本模型。