企业级 AI 应用开发实战：用 FastAPI 打造一个可落地的 AI 工单分流助手

云起川南

316人浏览 · 2026-06-12 14:45:00

云起川南 · 2026-06-12 14:45:00 发布

前言

很多企业做 AI 应用时，第一反应是做一个聊天机器人。

用户输入问题，AI 返回答案，看起来很智能，但真正进入业务系统后会发现：聊天机器人并不能直接解决企业的核心问题。

企业真正需要的 AI 应用，往往不是简单聊天，而是能把非结构化问题转化为结构化业务动作。

例如客服场景中，客户可能会发来这样一段话：

I received the tent yesterday, but one pole is broken and I need a replacement as soon as possible.

如果只是让 AI 回复一封邮件，价值有限。

更有价值的做法是让 AI 自动完成：

识别问题类型：配件损坏
判断紧急程度：高
提取关键信息：产品、问题、客户诉求
匹配售后政策：需要客户提供订单截图和破损照片
生成处理建议：建议创建补发配件工单
生成客服回复草稿：英文邮件草稿
是否需要人工确认：需要

这就是企业级 AI 应用真正有价值的方向：

AI 不只是回答问题，而是参与业务流程。

本文将用 FastAPI 实现一个简化版的 AI 工单分流助手，包含：

客户问题理解；
问题类型分类；
优先级判断；
售后政策匹配；
工单处理建议；
英文客服回复草稿生成；
人工确认标记；
可扩展的生产化架构。

这篇文章适合 AI 应用开发者、后端工程师、架构师，以及正在做企业 AI 助手、AI 客服、AI 工单系统的技术团队参考。

AI 工单处理流程图

以下是AI工单分流助手的核心处理流程：

一、为什么选择“AI 工单分流助手”作为实战案例？

相比普通 AI 聊天机器人，AI 工单分流助手更接近企业真实需求。

企业每天都会收到大量非结构化信息：

客户投诉；
售后问题；
退款请求；
配件缺失；
产品损坏；
物流异常；
使用咨询；
平台申诉；
内部任务流转。

这些信息如果完全靠人工处理，会有几个明显问题：

人工阅读成本高；
新员工判断标准不统一；
紧急问题容易被遗漏；
客服回复口径不一致；
工单分类不准确；
管理层很难统计问题类型。

AI 工单助手可以解决的不是“自动替代客服”，而是辅助客服快速完成初步判断。

一个合理的企业级 AI 工单系统应该这样设计：

客户问题
  ↓
AI 结构化理解
  ↓
规则引擎校验
  ↓
匹配售后政策
  ↓
生成处理建议
  ↓
生成回复草稿
  ↓
人工确认
  ↓
创建工单或回复客户

注意，AI 在这里不是最终决策者，而是业务助手。

二、本文项目目标

我们要实现一个接口：

POST /tickets/analyze

用户提交一段客户问题，例如：

{
  "customer_message": "I received the screen house yesterday, but one fiberglass pole is broken. Can you send me a replacement?",
  "platform": "Amazon",
  "order_id": "114-1234567-1234567"
}

系统返回：

{
  "category": "damaged_part",
  "priority": "high",
  "summary": "Customer reports a broken fiberglass pole and requests a replacement.",
  "suggested_action": "Ask customer to provide photos and confirm shipping address before arranging replacement parts.",
  "need_human_review": true,
  "reply_draft": "Hello,\n\nThank you for contacting us..."
}

这个结果已经不是单纯文本生成，而是把客户问题转换成了结构化业务结果。

三、项目技术栈

本文使用：

Python 3.10+
FastAPI
Pydantic
httpx
OpenAI-Compatible API
规则引擎
Prompt 模板

这里的大模型接口使用 OpenAI-Compatible 格式。

如果你使用的是其他兼容接口，例如国内大模型平台、私有化模型网关，也可以替换 base_url 和 model。

四、项目目录结构

项目目录如下：

ai-ticket-assistant/
│
├── app/
│   ├── main.py
│   ├── config.py
│   ├── schemas.py
│   │
│   ├── llm/
│   │   └── llm_client.py
│   │
│   ├── prompts/
│   │   └── ticket_prompt.py
│   │
│   ├── services/
│   │   ├── ticket_service.py
│   │   └── rule_engine.py
│   │
│   └── knowledge/
│       └── after_sales_policy.py
│
├── requirements.txt
├── .env
└── README.md

这个结构非常适合后续扩展。

例如后面可以继续加入：

数据库；
向量知识库；
工单系统；
权限控制；
日志系统；
消息队列；
人工审核后台。

五、安装依赖

requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
httpx==0.27.2
python-dotenv==1.0.1
pydantic==2.8.2

安装依赖：

pip install -r requirements.txt

六、配置环境变量

.env：

LLM_API_KEY=你的API_KEY
LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o-mini

说明：

LLM_API_KEY：大模型 API Key；
LLM_BASE_URL：大模型接口地址；
LLM_MODEL：模型名称。

如果你用的是其他兼容 OpenAI 格式的模型服务，只需要修改这三个配置。

七、读取配置

app/config.py：

import os
from dotenv import load_dotenv

load_dotenv()


class Settings:
    LLM_API_KEY: str = os.getenv("LLM_API_KEY", "")
    LLM_BASE_URL: str = os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")
    LLM_MODEL: str = os.getenv("LLM_MODEL", "gpt-4o-mini")


settings = Settings()

企业项目中，配置不要写死在代码里。

生产环境建议使用：

环境变量
配置中心
Kubernetes Secret
Vault
云厂商密钥管理服务

八、定义数据结构

app/schemas.py：

from pydantic import BaseModel, Field
from typing import Optional


class TicketAnalyzeRequest(BaseModel):
    customer_message: str = Field(..., description="客户原始问题")
    platform: Optional[str] = Field(default="Unknown", description="销售平台，例如 Amazon、Walmart、Shopify")
    order_id: Optional[str] = Field(default=None, description="订单号")


class TicketAnalysisResult(BaseModel):
    category: str
    priority: str
    summary: str
    customer_intent: str
    suggested_action: str
    need_human_review: bool
    reply_draft: str


class TicketAnalyzeResponse(BaseModel):
    success: bool
    data: TicketAnalysisResult

这里有一个很重要的设计点：AI 输出不能只是一段自然语言。

企业系统更需要结构化结果。

例如：

category
priority
summary
suggested_action
need_human_review
reply_draft

这些字段可以直接进入后续业务流程。

九、封装大模型客户端

app/llm/llm_client.py：

import json
import httpx
from typing import Dict, Any
from app.config import settings


class LLMClient:
    def __init__(self):
        self.api_key = settings.LLM_API_KEY
        self.base_url = settings.LLM_BASE_URL.rstrip("/")
        self.model = settings.LLM_MODEL

    async def chat_json(self, system_prompt: str, user_prompt: str) -> Dict[str, Any]:
        """
        调用大模型，并要求返回 JSON。
        """
        url = f"{self.base_url}/chat/completions"

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        payload = {
            "model": self.model,
            "temperature": 0.2,
            "messages": [
                {
                    "role": "system",
                    "content": system_prompt
                },
                {
                    "role": "user",
                    "content": user_prompt
                }
            ]
        }

        async with httpx.AsyncClient(timeout=60) as client:
            response = await client.post(url, headers=headers, json=payload)
            response.raise_for_status()
            result = response.json()

        content = result["choices"][0]["message"]["content"]

        return self._safe_parse_json(content)

    @staticmethod
    def _safe_parse_json(content: str) -> Dict[str, Any]:
        """
        尽量从模型返回内容中解析 JSON。
        防止模型返回 ```json 包裹内容。
        """
        content = content.strip()

        if content.startswith("```json"):
            content = content.replace("```json", "").replace("```", "").strip()

        if content.startswith("```"):
            content = content.replace("```", "").strip()

        try:
            return json.loads(content)
        except json.JSONDecodeError:
            raise ValueError(f"模型返回内容不是合法 JSON：{content}")


llm_client = LLMClient()

这里我们没有直接返回文本，而是让模型返回 JSON。

因为企业系统真正需要的是：

可解析
可存储
可流转
可统计
可审计

而不是一段不可控的自然语言。

十、设计 Prompt 模板

app/prompts/ticket_prompt.py：

def build_ticket_analysis_prompt(
    customer_message: str,
    platform: str,
    order_id: str,
    policy_text: str
) -> str:
    return f"""
请分析以下客户售后问题，并生成结构化工单分析结果。

【客户原始问题】
{customer_message}

【销售平台】
{platform}

【订单号】
{order_id or "未提供"}

【公司售后政策】
{policy_text}

请你完成以下任务：

1. 判断客户问题类型 category；
2. 判断客户意图 customer_intent；
3. 判断优先级 priority；
4. 总结客户问题 summary；
5. 给出客服处理建议 suggested_action；
6. 判断是否需要人工审核 need_human_review；
7. 生成一封英文客服回复草稿 reply_draft。

category 只能从以下选项中选择：
- damaged_part：配件损坏
- missing_part：配件缺失
- leaking_issue：漏水问题
- refund_request：退款请求
- warranty_activation：保修激活
- order_not_found：订单无法查询
- product_usage_question：产品使用咨询
- logistics_issue：物流问题
- other：其他

priority 只能从以下选项中选择：
- low
- medium
- high
- urgent

need_human_review 规则：
- 涉及退款，必须为 true；
- 涉及补发配件，必须为 true；
- 订单号缺失，必须为 true；
- 客户情绪强烈，必须为 true；
- 普通使用咨询可以为 false。

请严格返回 JSON，不要输出 Markdown，不要输出解释。

JSON 格式如下：

{{
  "category": "damaged_part",
  "priority": "high",
  "summary": "一句话总结客户问题",
  "customer_intent": "客户真实诉求",
  "suggested_action": "客服下一步应该怎么处理",
  "need_human_review": true,
  "reply_draft": "英文客服回复草稿"
}}
"""

这里的 Prompt 有几个关键点：

限定分类枚举；
限定优先级枚举；
明确人工审核规则；
强制 JSON 输出；
把售后政策作为上下文输入；
不让模型自由发挥。

企业级 Prompt 不是写得越长越好，而是要让模型输出稳定、可控、可解析。

十一、模拟售后政策知识库

为了让代码简单，我们先用一个 Python 文件模拟知识库。

app/knowledge/after_sales_policy.py：

AFTER_SALES_POLICY = """
公司售后政策：

1. 如果客户反馈产品损坏、配件缺失、漏水等质量问题，客服需要先要求客户提供订单截图、产品照片或视频。
2. 如果客户申请补发配件，需要确认订单号、产品型号、损坏配件名称和收货地址。
3. 如果订单无法在系统中查询到，需要请客户提供完整订单页面截图，截图中需要包含产品标题、订单号、购买日期和产品详情。
4. 客服不得在未确认订单和问题原因前，直接承诺全额退款。
5. 涉及退款、补发、赔偿等操作时，必须由人工客服确认后再执行。
6. 如果客户只是咨询产品使用方法，可以直接提供指导说明。
7. 回复客户时需要保持礼貌、专业，并感谢客户的耐心等待。
8. 英文回复中不要使用过度承诺，例如 guaranteed full refund、immediate compensation 等表达。
"""

真实生产环境中，这里可以替换为 RAG 知识库。

例如：

PDF 售后政策
客服 SOP 文档
产品说明书
订单规则
平台政策
内部知识库

十二、规则引擎：不能完全相信 AI 判断

企业级 AI 应用不能完全依赖模型。

例如，AI 可能把退款请求判断成普通咨询，也可能漏掉高风险问题。

所以我们需要增加一个规则引擎，对 AI 结果进行二次校验。

app/services/rule_engine.py：

from typing import Dict, Any


class TicketRuleEngine:
    """
    工单规则引擎。
    用于对 AI 输出结果进行二次修正。
    """

    HIGH_RISK_KEYWORDS = [
        "refund",
        "return",
        "angry",
        "disappointed",
        "complaint",
        "broken",
        "damaged",
        "leaking",
        "missing",
        "replacement",
        "defective"
    ]

    REFUND_KEYWORDS = [
        "refund",
        "money back",
        "return"
    ]

    @classmethod
    def apply_rules(
        cls,
        ai_result: Dict[str, Any],
        customer_message: str,
        order_id: str | None
    ) -> Dict[str, Any]:
        message_lower = customer_message.lower()

        # 规则 1：没有订单号，必须人工审核
        if not order_id:
            ai_result["need_human_review"] = True

        # 规则 2：涉及退款，必须人工审核
        if any(keyword in message_lower for keyword in cls.REFUND_KEYWORDS):
            ai_result["category"] = "refund_request"
            ai_result["need_human_review"] = True
            ai_result["priority"] = cls._max_priority(ai_result.get("priority"), "high")

        # 规则 3：涉及损坏、漏水、缺件、补发，必须人工审核
        if any(keyword in message_lower for keyword in cls.HIGH_RISK_KEYWORDS):
            ai_result["need_human_review"] = True
            ai_result["priority"] = cls._max_priority(ai_result.get("priority"), "high")

        # 规则 4：修正非法 priority
        if ai_result.get("priority") not in ["low", "medium", "high", "urgent"]:
            ai_result["priority"] = "medium"

        # 规则 5：修正非法 category
        valid_categories = [
            "damaged_part",
            "missing_part",
            "leaking_issue",
            "refund_request",
            "warranty_activation",
            "order_not_found",
            "product_usage_question",
            "logistics_issue",
            "other"
        ]

        if ai_result.get("category") not in valid_categories:
            ai_result["category"] = "other"

        return ai_result

    @staticmethod
    def _max_priority(current: str | None, target: str) -> str:
        priority_order = {
            "low": 1,
            "medium": 2,
            "high": 3,
            "urgent": 4
        }

        current_score = priority_order.get(current, 2)
        target_score = priority_order.get(target, 2)

        if target_score > current_score:
            return target

        return current or "medium"

这段代码体现了企业级 AI 应用一个非常重要的原则：

AI 负责理解，规则负责兜底。

不能把所有判断都交给 AI。

尤其是涉及退款、赔偿、合同、财务、法务、客户隐私等场景，一定要有规则引擎或人工审核。

十三、实现工单分析服务

app/services/ticket_service.py：

from app.llm.llm_client import llm_client
from app.prompts.ticket_prompt import build_ticket_analysis_prompt
from app.knowledge.after_sales_policy import AFTER_SALES_POLICY
from app.services.rule_engine import TicketRuleEngine
from app.schemas import TicketAnalysisResult


class TicketService:
    async def analyze_ticket(
        self,
        customer_message: str,
        platform: str,
        order_id: str | None
    ) -> TicketAnalysisResult:
        """
        分析客户问题，并返回结构化工单结果。
        """

        system_prompt = """
你是一个企业级 AI 工单分析助手。
你的任务是将客户的非结构化问题转化为结构化工单字段。
你必须遵守公司售后政策，不允许编造订单信息、退款承诺或补偿方案。
你只能返回合法 JSON。
"""

        user_prompt = build_ticket_analysis_prompt(
            customer_message=customer_message,
            platform=platform,
            order_id=order_id or "",
            policy_text=AFTER_SALES_POLICY
        )

        ai_result = await llm_client.chat_json(
            system_prompt=system_prompt,
            user_prompt=user_prompt
        )

        final_result = TicketRuleEngine.apply_rules(
            ai_result=ai_result,
            customer_message=customer_message,
            order_id=order_id
        )

        return TicketAnalysisResult(**final_result)


ticket_service = TicketService()

这个服务层是核心编排逻辑。

流程如下：

客户问题
  ↓
构造 Prompt
  ↓
调用大模型
  ↓
解析 JSON
  ↓
规则引擎二次校验
  ↓
返回结构化结果

这个流程很适合企业生产系统。

因为它不是简单依赖大模型，而是把模型能力放进一个可控流程中。

十四、实现 FastAPI 接口

app/main.py：

from fastapi import FastAPI
from app.schemas import TicketAnalyzeRequest, TicketAnalyzeResponse
from app.services.ticket_service import ticket_service

app = FastAPI(
    title="AI Ticket Assistant",
    description="企业级 AI 工单分流助手",
    version="1.0.0"
)


@app.get("/health")
async def health_check():
    return {
        "status": "ok",
        "message": "AI Ticket Assistant is running"
    }


@app.post("/tickets/analyze", response_model=TicketAnalyzeResponse)
async def analyze_ticket(request: TicketAnalyzeRequest):
    result = await ticket_service.analyze_ticket(
        customer_message=request.customer_message,
        platform=request.platform,
        order_id=request.order_id
    )

    return TicketAnalyzeResponse(
        success=True,
        data=result
    )

启动项目：

uvicorn app.main:app --reload

访问接口文档：

http://127.0.0.1:8000/docs

十五、测试接口

请求示例：

curl -X POST "http://127.0.0.1:8000/tickets/analyze" \
-H "Content-Type: application/json" \
-d '{
  "customer_message": "I received the screen house yesterday, but one fiberglass pole is broken. Can you send me a replacement?",
  "platform": "Amazon",
  "order_id": "114-1234567-1234567"
}'

可能返回：

{
  "success": true,
  "data": {
    "category": "damaged_part",
    "priority": "high",
    "summary": "Customer reports that one fiberglass pole of the screen house is broken and requests a replacement.",
    "customer_intent": "The customer wants a replacement fiberglass pole.",
    "suggested_action": "Ask the customer to provide photos of the broken pole and confirm the shipping address before arranging a replacement.",
    "need_human_review": true,
    "reply_draft": "Hello,\n\nThank you for contacting us. We are sorry to hear that one of the fiberglass poles for your screen house arrived broken.\n\nCould you please send us a photo of the broken pole and confirm your shipping address? Once we receive the information, we will further check and assist you with the appropriate replacement solution.\n\nThank you for your patience and understanding.\n\nBest regards,"
  }
}

这个接口返回的内容可以直接用于：

创建工单；
分配客服；
设置优先级；
生成回复草稿；
进入人工审核流程；
统计售后问题类型。

十六、再测试一个退款场景

请求：

curl -X POST "http://127.0.0.1:8000/tickets/analyze" \
-H "Content-Type: application/json" \
-d '{
  "customer_message": "This tent is leaking badly. I am very disappointed and want a full refund immediately.",
  "platform": "Walmart"
}'

注意这里没有订单号，而且客户要求退款。

系统应该返回：

{
  "category": "refund_request",
  "priority": "high",
  "need_human_review": true
}

即使大模型判断不准确，规则引擎也会兜底：

if not order_id:
    ai_result["need_human_review"] = True

if any(keyword in message_lower for keyword in cls.REFUND_KEYWORDS):
    ai_result["category"] = "refund_request"
    ai_result["need_human_review"] = True
    ai_result["priority"] = cls._max_priority(ai_result.get("priority"), "high")

这就是企业级 AI 应用和普通 Demo 的区别。

普通 Demo 只关注模型回答好不好。

企业级 AI 应用必须关注：

是否可控
是否安全
是否可追溯
是否能进入业务流程

十七、为什么这个案例适合企业落地？

这个 AI 工单分流助手虽然简单，但已经具备企业级 AI 应用的几个关键能力。

1. 非结构化信息结构化

客户原始问题是自然语言，系统把它转换成结构化字段：

category
priority
summary
customer_intent
suggested_action
need_human_review
reply_draft

结构化之后，才能进入业务系统。

2. AI + 规则引擎协同

AI 擅长理解语言，但不适合单独做高风险判断。

规则引擎可以弥补 AI 的不稳定性。

3. 可以直接嵌入客服系统

这个接口可以接入：

客服后台
CRM
工单系统
企业微信机器人
飞书机器人
邮件系统
售后管理系统

4. 支持人工确认

涉及退款、补发、投诉等场景，系统会标记：

"need_human_review": true

这符合企业真实流程。

5. 方便后续统计分析

因为输出是结构化字段，所以可以统计：

每天多少退款请求
多少配件损坏问题
哪个平台投诉最多
哪些产品售后问题最多
高优先级工单占比
AI 草稿被人工修改的比例

这些数据对管理层非常有价值。

十八、生产环境如何继续升级？

当前代码只是最小可用版本。

如果要进入生产环境，可以从以下几个方向升级。

1. 接入数据库保存工单

可以设计一张工单表：

CREATE TABLE ai_ticket (
    id BIGINT PRIMARY KEY,
    platform VARCHAR(50),
    order_id VARCHAR(100),
    customer_message TEXT,
    category VARCHAR(50),
    priority VARCHAR(20),
    summary TEXT,
    suggested_action TEXT,
    need_human_review BOOLEAN,
    reply_draft TEXT,
    status VARCHAR(30),
    created_at TIMESTAMP
);

每次 AI 分析后，把结果写入数据库。

2. 接入人工审核流程

工单状态可以设计为：

pending_review：待人工审核
approved：已确认
rejected：已驳回
processed：已处理
closed：已关闭

AI 只生成建议，最终操作由人工确认。

3. 接入真实订单系统

当前代码没有查询真实订单。

生产环境可以增加：

Amazon SP-API
Walmart API
Shopify API
ERP 系统
WMS 系统
自研订单系统

例如：

客户提供订单号
  ↓
系统查询订单
  ↓
确认购买日期
  ↓
确认是否在保修期
  ↓
确认产品型号
  ↓
AI 生成更准确的处理建议

4. 接入 RAG 知识库

当前售后政策是写在 Python 文件中的。

生产环境应该使用 RAG：

产品说明书
售后政策
平台政策
客服 SOP
配件清单
保修规则
历史高质量回复

AI 生成回复时，不应该只依赖固定 Prompt，而应该动态检索相关知识。

5. 增加日志和成本统计

每次调用 AI，都应该记录：

用户
时间
问题
模型
Prompt 版本
输入 Token
输出 Token
响应时间
调用成本
返回结果
人工是否采纳

这些数据可以帮助团队持续优化 AI 应用。

6. 增加 Prompt 版本管理

Prompt 不建议长期写死在代码里。

可以放到数据库中：

CREATE TABLE ai_prompt_template (
    id BIGINT PRIMARY KEY,
    name VARCHAR(100),
    scene VARCHAR(100),
    version VARCHAR(30),
    content TEXT,
    status VARCHAR(20),
    created_at TIMESTAMP
);

这样可以支持：

Prompt 灰度发布
Prompt 回滚
A/B 测试
不同业务线使用不同 Prompt

十九、企业级 AI 工单系统的完整演进架构

后续可以演进成这样的架构：

客户邮件 / 平台消息 / 在线客服
        ↓
消息接入层
        ↓
AI 工单分析服务
        ↓
订单系统查询
        ↓
RAG 售后知识库
        ↓
规则引擎校验
        ↓
生成工单分类和优先级
        ↓
生成客服回复草稿
        ↓
人工审核
        ↓
发送回复 / 补发配件 / 创建退款申请
        ↓
日志记录与数据分析

这个架构可以应用到很多企业场景：

电商客服；
售后工单；
IT Helpdesk；
内部行政工单；
财务报销初审；
供应链异常处理；
仓库问题反馈；
质量问题归类。

二十、AI 工单助手开发中的几个关键经验

1. 不要只让 AI 生成文本

企业系统更需要结构化数据。

自然语言结果看起来好，但很难进入系统。

结构化字段才是真正有业务价值的输出。

2. 不要完全相信 AI 判断

AI 可以理解问题，但规则引擎必须兜底。

尤其是涉及：

退款
赔偿
补发
投诉
差评
法律风险
客户隐私

这些场景必须人工确认。

3. Prompt 要限制输出范围

分类字段、优先级字段、审核规则都应该提前定义好。

不要让模型自己发明分类。

4. AI 应用要从具体场景切入

不要一开始就做一个万能 AI 助手。

更好的方式是从一个具体场景切入：

AI 工单分流
AI 客服回复
AI 知识库问答
AI 评论分析
AI 报表解读
AI 商品文案生成

场景越具体，越容易落地。

5. 企业 AI 应用必须有人机协同

AI 负责提升效率，人负责最终确认。

这是当前企业级 AI 应用最稳妥的落地方式。

二十二、补充：完整的测试用例与验证

在实际部署 AI 工单分流助手时，全面的测试验证是确保系统稳定可靠的关键。本节提供 4 个覆盖不同场景的完整 curl 请求示例和预期的 JSON 响应，并说明如何验证规则引擎是否生效。

场景一：普通咨询（低优先级）

请求示例：

curl -X POST "http://127.0.0.1:8000/analyze-ticket" \
  -H "Content-Type: application/json" \
  -d '{
    "customer_query": "你好，我昨天买的手机壳感觉有点松动，能换一个吗？",
    "order_id": "ORD20240611001"
  }'

预期响应：

{
  "customer_query": "你好，我昨天买的手机壳感觉有点松动，能换一个吗？",
  "order_id": "ORD20240611001",
  "category": "产品质量问题",
  "sub_category": "配件松动",
  "priority": "低",
  "needs_human_review": false,
  "policy_match": "7天无理由退换货",
  "response_draft": "Dear customer, we apologize for the inconvenience. Your case qualifies for our 7-day return/exchange policy. Please provide your shipping address and we will arrange a replacement.",
  "rule_engine_override": null,
  "final_priority": "低",
  "final_needs_human_review": false
}

验证点：

AI 正确识别为「产品质量问题」→「配件松动」
优先级为「低」
needs_human_review 为 false
规则引擎未触发覆盖（rule_engine_override 为 null）

场景二：高优先级损坏（规则引擎触发）

请求示例：

curl -X POST "http://127.0.0.1:8000/analyze-ticket" \
  -H "Content-Type: application/json" \
  -d '{
    "customer_query": "我刚买的笔记本电脑屏幕完全碎了，完全无法使用！",
    "order_id": "ORD20240611002"
  }'

预期响应：

{
  "customer_query": "我刚买的笔记本电脑屏幕完全碎了，完全无法使用！",
  "order_id": "ORD20240611002",
  "category": "产品质量问题",
  "sub_category": "屏幕损坏",
  "priority": "中",
  "needs_human_review": false,
  "policy_match": "电子产品质量问题",
  "response_draft": "Dear customer, we are sorry to hear about your laptop screen. This qualifies as a product quality issue. Please provide photos of the damage and we will arrange a repair or replacement.",
  "rule_engine_override": {
    "reason": "关键词触发：屏幕完全碎了，完全无法使用",
    "original_priority": "中",
    "new_priority": "高",
    "original_needs_human_review": false,
    "new_needs_human_review": true
  },
  "final_priority": "高",
  "final_needs_human_review": true
}

验证点：

AI 初始判断优先级为「中」
规则引擎检测到「完全碎了」「完全无法使用」等紧急关键词
rule_engine_override 显示优先级从「中」提升到「高」，且 needs_human_review 从 false 改为 true
最终结果使用规则引擎覆盖后的值

场景三：无订单号退款（规则引擎触发）

请求示例：

curl -X POST "http://127.0.0.1:8000/analyze-ticket" \
  -H "Content-Type: application/json" \
  -d '{
    "customer_query": "我要退款，但我找不到订单号了",
    "order_id": ""
  }'

预期响应：

{
  "customer_query": "我要退款，但我找不到订单号了",
  "order_id": "",
  "category": "退款申请",
  "sub_category": "无订单号退款",
  "priority": "中",
  "needs_human_review": false,
  "policy_match": "标准退款流程",
  "response_draft": "Dear customer, we can help you with the refund. Please provide your registered email or phone number so we can locate your order.",
  "rule_engine_override": {
    "reason": "订单号为空",
    "original_priority": "中",
    "new_priority": "高",
    "original_needs_human_review": false,
    "new_needs_human_review": true
  },
  "final_priority": "高",
  "final_needs_human_review": true
}

验证点：

AI 正确识别为「退款申请」→「无订单号退款」
规则引擎检测到 order_id 为空字符串
rule_engine_override 显示优先级提升且需要人工审核
验证了规则引擎对数据完整性的校验功能

场景四：物流问题（普通优先级）

请求示例：

curl -X POST "http://127.0.0.1:8000/analyze-ticket" \
  -H "Content-Type: application/json" \
  -d '{
    "customer_query": "我的包裹显示已签收，但我没有收到，物流信息显示签收人是\"门口\"",
    "order_id": "ORD20240611003"
  }'

预期响应：

{
  "customer_query": "我的包裹显示已签收，但我没有收到，物流信息显示签收人是\"门口\"",
  "order_id": "ORD20240611003",
  "category": "物流问题",
  "sub_category": "未收到货",
  "priority": "中",
  "needs_human_review": true,
  "policy_match": "物流异常处理流程",
  "response_draft": "Dear customer, we apologize for the delivery issue. We will contact the logistics provider to investigate the \"doorstep\" signature. Please allow 1-2 business days for an update.",
  "rule_engine_override": null,
  "final_priority": "中",
  "final_needs_human_review": true
}

验证点：

AI 正确识别为「物流问题」→「未收到货」
优先级为「中」，需要人工审核
规则引擎未触发覆盖
验证了 AI 对复杂物流场景的理解能力

如何验证规则引擎是否生效

查看 rule_engine_override 字段
- 如果为 null：规则引擎未触发，使用 AI 的原始判断
- 如果有值：规则引擎已触发，查看 reason 了解触发原因，比较 original_* 和 new_* 字段
对比 final_priority 和 final_needs_human_review
- 这两个字段是规则引擎覆盖后的最终结果
- 如果与 AI 初始的 priority、needs_human_review 不同，说明规则引擎生效
测试边界案例
- 订单号为空、格式错误
- 包含紧急关键词（着火、爆炸、完全损坏等）
- 高金额订单（可在规则引擎中配置金额阈值）
- VIP 客户标识（可在数据结构中扩展）
监控日志输出
在服务端查看规则引擎的决策日志，确认触发逻辑与预期一致。

通过以上测试用例，可以全面验证 AI 工单分流助手在不同场景下的表现，特别是规则引擎与 AI 判断的协同工作效果。

二十一、总结

本文实现了一个简化版的企业级 AI 工单分流助手。

它完成了以下能力：

接收客户原始问题；
使用大模型理解客户诉求；
生成结构化工单字段；
匹配公司售后政策；
判断工单优先级；
判断是否需要人工审核；
生成英文客服回复草稿；
使用规则引擎进行二次校验；
通过 FastAPI 对外提供接口。

这篇文章的重点不是炫技，而是说明一个企业级 AI 应用的核心思想：

AI 不应该只是聊天窗口，而应该成为业务流程中的智能节点。

从技术角度看，企业级 AI 应用开发的关键不只是大模型 API，而是：

Prompt 设计
结构化输出
规则引擎
业务系统集成
人工审核
日志审计
数据闭环

真正能落地的 AI 应用，一定是能接入企业业务、遵守业务规则、输出结构化结果、支持人工确认，并且可以持续优化的系统。

后续我会继续更新企业级 AI 应用开发实战系列，包括：

AI 工单系统如何接入真实数据库；
RAG 知识库如何接入售后 SOP；
AI 客服助手如何对接订单系统；
Prompt 模板如何做版本管理；
AI Agent 如何安全调用企业内部 API；
企业级 AI 应用如何做日志、监控和成本统计。

关注我，一起系统学习企业级 AI 应用开发。

AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

AtomGit开源社区

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

AtomGit开源社区

AtomCode 全新上线万亿代码模型 LongCat-2.0。限时 7 天免费体验！

AtomGit开源社区

所有评论(0)

查看更多评论

云起川南

@love7489

已为社区贡献5条内容

企业级 AI 应用开发实战：用 FastAPI 打造一个可落地的 AI 工单分流助手

云起川南

前言

AI 工单处理流程图

一、为什么选择“AI 工单分流助手”作为实战案例？

二、本文项目目标

三、项目技术栈

四、项目目录结构

五、安装依赖

六、配置环境变量

七、读取配置

八、定义数据结构

九、封装大模型客户端

十、设计 Prompt 模板

十一、模拟售后政策知识库

十二、规则引擎：不能完全相信 AI 判断

十三、实现工单分析服务

十四、实现 FastAPI 接口

十五、测试接口

十六、再测试一个退款场景

十七、为什么这个案例适合企业落地？

1. 非结构化信息结构化

2. AI + 规则引擎协同

3. 可以直接嵌入客服系统

4. 支持人工确认

5. 方便后续统计分析

十八、生产环境如何继续升级？

1. 接入数据库保存工单

2. 接入人工审核流程

3. 接入真实订单系统

4. 接入 RAG 知识库

5. 增加日志和成本统计

6. 增加 Prompt 版本管理

十九、企业级 AI 工单系统的完整演进架构

二十、AI 工单助手开发中的几个关键经验

1. 不要只让 AI 生成文本

2. 不要完全相信 AI 判断

3. Prompt 要限制输出范围

4. AI 应用要从具体场景切入

5. 企业 AI 应用必须有人机协同

二十二、补充：完整的测试用例与验证

场景一：普通咨询（低优先级）

场景二：高优先级损坏（规则引擎触发）

场景三：无订单号退款（规则引擎触发）

场景四：物流问题（普通优先级）

如何验证规则引擎是否生效

二十一、总结

所有评论(0)

温馨提示：您尚未绑定手机号

云起川南