1. 项目背景

某中型 SaaS 公司有 50 人的客服团队，每天处理约 800 个客户咨询。咨询类型包括：产品使用问题（40%）、故障报修（25%）、续费咨询（15%）、投诉与升级（10%）、其他（10%）。客服团队面临三个核心痛点：

痛点一：重复回答效率低下。“怎么重置密码”“怎么导出报表”“发票怎么申请”——这些问题的答案在知识库中都有，但客服仍要逐条打字回复。据统计，客服 30% 的时间花在了回答知识库已有答案的问题上。

痛点二：工单分拣靠人工。每个工单进来后，需人工判断类型并分配给对应的处理组——退款组、技术组、续费组。平均分拣耗时 3 分钟/单，每天浪费 40 人时。

痛点三：高风险工单漏报。客户的投诉和严重故障没有自动识别和升级机制——一个客户的"数据丢失"工单可能被淹没在 800 条日常咨询中，等发现时已经过了 4 小时。

CTO 的要求很明确：用 CrewAI 搭建"智能客服与工单处理平台"，实现三个核心目标——自动回答（常见问题自动回复准确率 >85%）、智能分拣（工单自动分类和路由）、风险升级（高风险工单漏报率 <2%）。本章作为中级篇的综合实战，将中级篇 15 章的知识全部串联到一个完整的系统中。

2. 项目设计

小胖（抱着一大杯奶茶）：“大师，这个综合实战是不是把前面所有学过的全堆在一起？”

大师：“恰恰相反。综合实战的核心能力不是’堆砌’，而是’取舍’——你知道哪些中级知识该用、哪些不该用。比如：你需要在每个工单上都用 Hierarchical Process 吗？不需要——只有升级给 Manager 仲裁的复杂工单才用。你需要给每个 Agent 都挂上 Memory 吗？不需要——只有需要记住客户历史偏好的 Agent 才用。做综合实战最大的坑就是’过度设计’——以为学过的都要用上。”

小白（翻开架构图）：“那这个平台的总体架构应该怎么设计？”

大师：“三层架构。第一层，接入层——FastAPI（第24章）+ 消息机器人（第25章），统一接收客户请求。第二层，编排层——Flow（第13/19章）做工单分类和路由，分支内嵌入 Crew。第三层，能力层——知识库 RAG（第18章）提供制度问答、订单查询 Tool（第17章）对接 CRM、评估体系（第21章）持续监控质量、可观测性（第22章）做运行监控。”

小胖：“工单处理流程怎么设计？所有工单都走同一个 Flow 吗？”

大师：“一个主 Flow + 四个子 Crew。主 Flow 负责：接收工单 → 自动分类（Agent）→ 路由到对应处理分支。四个子 Crew 是：自动问答 Crew（知识库 RAG——适用常见问题）、退款处理 Crew（订单查询 Tool + 人工审批——适用退款请求）、故障诊断 Crew（CI 诊断模式——适用技术故障）、投诉升级 Crew（风险评级 + 人工审核——适用投诉）。”

小白：“那运行报表怎么生成？”

大师：“批处理（第23章）——每天晚上 0 点跑一次’日运行报告’，汇总当天的：总工单数、自动处理率、各分类分布、平均响应时间、Agent 成功率和 Token 消耗。结果推送到企业微信群（第25章消息机器人）。这份报告也是评估体系（第21章）的在线评估数据来源——用来持续衡量平台的真实效果。”

小胖：“部署呢？这个平台怎么上线？”

大师：“容器化部署（第29章）——Docker Compose 本地开发模式 + K8s 生产部署。三个服务：FastAPI 主服务（3 副本）、Redis 任务队列、Chroma 向量数据库。健康检查 + 日志采集 + 监控大盘全部配齐。灰度发布策略：先让 1 个客服组试用 1 周，收集反馈后再全量推广。”

技术映射总结：智能客服平台 = FastAPI + Flow(分类路由) + RAG(知识库问答) + DB Tool(订单查询) + 评估体系(质量监控) + 可观测性(运行报表) + 消息机器人(通知推送) + Docker/K8s(部署)。这是一个用中级篇所有知识构建的完整系统——每个组件独立可替换，整体协同可演进。

系统架构图

┌──────────────────────────────────────────────────────────────────┐
│                      客户入口（多渠道）                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │
│  │ Web 页面  │  │ 企业微信   │  │ 飞书 Bot  │  │ API (第三方)  │    │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └──────┬───────┘    │
└───────┼──────────────┼─────────────┼───────────────┼─────────────┘
        │              │             │               │
        └──────────────┴──────┬──────┴───────────────┘
                              │
                              ▼
┌──────────────────────────────────────────────────────────────────┐
│                    FastAPI 统一网关                                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────────┐    │
│  │ 鉴权(Auth) │  │ 限流(Rate)│  │ 日志(Log) │  │ 预算熔断(Bgt) │    │
│  └──────────┘  └──────────┘  └──────────┘  └──────────────┘    │
└───────────────────────────┬──────────────────────────────────────┘
                            │
                            ▼
┌──────────────────────────────────────────────────────────────────┐
│                     TicketFlow（Flow 编排层）                      │
│                                                                  │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │               classify: 工单分类 (Agent)                    │ │
│  │          qa / refund / incident / complaint / other         │ │
│  └──────────────────────────┬─────────────────────────────────┘ │
│                             │                                    │
│              ┌──────────────┼──────────────┐                    │
│              │              │              │                     │
│              ▼              ▼              ▼                     │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐            │
│  │  QA Crew     │ │ Refund Crew  │ │Incident Crew │  ...       │
│  │ (RAG 问答)   │ │ (Tool+审批)  │ │ (诊断+升级)  │            │
│  └──────┬───────┘ └──────┬───────┘ └──────┬───────┘            │
│         │                │                │                      │
│         └────────────────┼────────────────┘                      │
│                          ▼                                       │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │             quality_check: 质量检查 + Guardrail              │ │
│  └────────────────────────────────────────────────────────────┘ │
│                          │                                       │
│                          ▼                                       │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │       human_review: 高风险工单人工审核（HITL）              │ │
│  └────────────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌──────────────────────────────────────────────────────────────────┐
│                      基础设施层                                    │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐       │
│  │ 向量库    │ │ 数据库    │ │ Redis    │ │ 监控(Prometheus)│     │
│  │ (Chroma)  │ │(PostgreSQL)│ │(任务队列) │ │ 日志(ELK)     │      │
│  └──────────┘ └──────────┘ └──────────┘ └──────────────┘       │
└──────────────────────────────────────────────────────────────────┘

3. 项目实战

3.1 实战目标

交付一个可演示的企业智能客服 MVP——含自动问答、工单分拣、退款处理、风险升级、人工审核、运行报表六大功能。

3.2 项目结构

smart-cs-platform/
├── .env.example
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── README.md
├── src/
│   ├── main.py                     # FastAPI 入口
│   ├── api/
│   │   ├── routes.py               # API 路由
│   │   └── models.py               # Pydantic 模型
│   ├── flow/
│   │   └── ticket_flow.py          # 主 Flow——分类+路由
│   ├── crews/
│   │   ├── qa_crew.py              # 自动问答 Crew
│   │   ├── refund_crew.py          # 退款处理 Crew
│   │   ├── incident_crew.py        # 故障诊断 Crew
│   │   └── complaint_crew.py       # 投诉处理 Crew
│   ├── tools/
│   │   ├── order_query.py          # 订单查询 Tool
│   │   ├── knowledge_search.py     # 知识库检索 Tool
│   │   └── escalation.py           # 升级通知 Tool
│   ├── agents/
│   │   └── definitions.py          # 所有 Agent 定义
│   ├── knowledge/
│   │   └── documents/              # 产品知识库文档
│   ├── telemetry/
│   │   └── reporter.py             # 运行报表生成
│   └── auth/
│       └── middleware.py           # 鉴权中间件
├── tests/
│   ├── test_api.py
│   ├── test_flow.py
│   └── test_crews.py
├── k8s/
│   ├── deployment.yaml
│   ├── service.yaml
│   └── configmap.yaml
└── outputs/

3.3 分步实现

步骤1：统一 Agent 定义

目标：平台所有 Agent 集中管理，保持角色边界清晰。

# src/agents/definitions.py
from crewai import Agent
from typing import Dict

def create_all_agents(llm) -> Dict[str, Agent]:
    """创建平台所有 Agent——统一管理"""

    return {
        "classifier": Agent(
            role="工单分类员",
            goal="阅读客户咨询内容，准确分类为: qa/refund/incident/complaint/other",
            backstory="你有3年客服工单分拣经验，日均分拣500+。分类标准基于客户核心诉求。",
            llm=llm, verbose=False, allow_delegation=False, max_iter=5
        ),
        "qa_expert": Agent(
            role="产品知识库专家",
            goal="基于产品文档准确回答客户问题，标注引用来源",
            backstory="你熟悉公司所有产品文档，回答必有据可查。不确定时告知客户转人工。",
            llm=llm, verbose=False, allow_delegation=False, max_iter=8
        ),
        "refund_assessor": Agent(
            role="退款审核员",
            goal="根据退款政策判断申请是否合规，计算退款金额，标记需审批的高额退款",
            backstory="你处理过10000+退款申请。判断标准：订单状态→退款政策→金额阈值(>500需审批)。",
            llm=llm, verbose=False, allow_delegation=False, max_iter=10
        ),
        "incident_diagnoser": Agent(
            role="故障诊断工程师",
            goal="分析客户报告的故障现象，给出排查步骤和解决方案。严重故障立即标记升级。",
            backstory="你是SRE团队的技术支持，处理过500+线上故障。严重故障标准：数据丢失/服务中断/安全事件。",
            llm=llm, verbose=False, allow_delegation=False, max_iter=10
        ),
        "complaint_handler": Agent(
            role="投诉处理专员",
            goal="评估投诉严重程度，给出安抚方案和补偿建议。重大投诉立即升级。",
            backstory="你有5年客服投诉处理经验。投诉评估公式：严重度×影响面×可恢复性=升级优先级。",
            llm=llm, verbose=False, allow_delegation=False, max_iter=8
        ),
        "response_writer": Agent(
            role="客服回复撰写员",
            goal="基于处理结果撰写专业、温暖、结构清晰的客户回复",
            backstory="你擅长将技术人员的话翻译为客户能懂的语言。回复公式：共情+解决方案+行动指引。",
            llm=llm, verbose=False, allow_delegation=False, max_iter=5
        ),
    }

步骤2：主 Flow——分类与路由

# src/flow/ticket_flow.py
from crewai.flow.flow import Flow, listen, start, router
from crewai import Crew, Process, LLM, Task
from src.agents.definitions import create_all_agents
import os
import json
import time


class TicketFlow(Flow):
    """智能客服工单处理主 Flow"""

    def __init__(self):
        super().__init__()
        self.llm = LLM(
            model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
            temperature=0.3, timeout=120
        )
        self.agents = create_all_agents(self.llm)

    @start()
    def receive_ticket(self):
        """接收工单——从 API 传入"""
        ticket = self.state.get("ticket", {})
        customer = ticket.get("customer_name", "客户")
        content = ticket.get("content", "")

        print(f"\n📥 收到工单: [{ticket.get('id', 'N/A')}] {customer} - {content[:60]}...")
        self.state["received_at"] = time.time()
        self.state["status"] = "received"

    @listen("receive_ticket")
    def classify(self):
        """工单分类"""
        content = self.state["ticket"]["content"]

        classifier = self.agents["classifier"]
        task = Task(
            description=f"请将以下客户咨询分类（只输出一个类型）:\n{content[:1000]}",
            expected_output="qa / refund / incident / complaint / other",
            agent=classifier
        )

        crew = Crew(agents=[classifier], tasks=[task], process=Process.sequential, verbose=False)
        result = crew.kickoff()

        raw = result.raw.lower() if result.raw else "other"
        if "refund" in raw:
            category = "refund"
        elif "incident" in raw or "故障" in raw or "bug" in raw:
            category = "incident"
        elif "complaint" in raw or "投诉" in raw:
            category = "complaint"
        elif "qa" in raw:
            category = "qa"
        else:
            category = "other"

        self.state["category"] = category
        print(f"  🏷️ 分类: {category}")

    @router("classify")
    def route(self):
        """路由到处理分支"""
        return f"handle_{self.state['category']}"

    @listen("route")
    def handle_qa(self):
        """处理常见问题——RAG 问答"""
        print(f"  📚 进入知识库问答...")
        from src.crews.qa_crew import answer_question
        result = answer_question(
            self.state["ticket"]["content"],
            self.agents["qa_expert"],
            self.agents["response_writer"],
            self.llm
        )
        self.state["response"] = result

    @listen("route")
    def handle_refund(self):
        """处理退款"""
        print(f"  💰 进入退款处理...")
        from src.crews.refund_crew import process_refund
        result = process_refund(
            self.state["ticket"],
            self.agents["refund_assessor"],
            self.agents["response_writer"],
            self.llm
        )
        self.state["response"] = result
        self.state["needs_human_review"] = result.get("amount", 0) > 500

    @listen("route")
    def handle_incident(self):
        """处理故障"""
        print(f"  🔧 进入故障诊断...")
        from src.crews.incident_crew import diagnose_incident
        result = diagnose_incident(
            self.state["ticket"]["content"],
            self.agents["incident_diagnoser"],
            self.agents["response_writer"],
            self.llm
        )
        self.state["response"] = result
        self.state["needs_human_review"] = result.get("severity") in ["critical", "high"]

    @listen("route")
    def handle_complaint(self):
        """处理投诉"""
        print(f"  🚨 进入投诉处理...")
        from src.crews.complaint_crew import handle_complaint
        result = handle_complaint(
            self.state["ticket"]["content"],
            self.agents["complaint_handler"],
            self.agents["response_writer"],
            self.llm
        )
        self.state["response"] = result
        self.state["needs_human_review"] = result.get("severity") in ["severe", "critical"]

    @listen("route")
    def handle_other(self):
        """无法分类——转人工"""
        self.state["response"] = {"answer": "已转人工客服处理", "auto_resolved": False}
        self.state["needs_human_review"] = True

    @listen("handle_qa", "handle_refund", "handle_incident", "handle_complaint", "handle_other")
    def finalize(self):
        """记录结果"""
        duration = round(time.time() - self.state["received_at"], 2)
        self.state["duration_seconds"] = duration
        self.state["status"] = "completed"
        print(f"  ✅ 处理完成 ({duration}s)")

步骤3：四个处理 Crew

# src/crews/qa_crew.py
from crewai import Crew, Process, Task

def answer_question(question: str, expert, writer, llm) -> dict:
    """RAG 自动问答"""
    search_task = Task(
        description=f"在知识库中检索: {question[:500]}",
        expected_output="检索到的相关文档内容",
        agent=expert
    )
    answer_task = Task(
        description=f"基于检索结果回答客户问题: {question[:500]}\n要求：标注引用来源",
        expected_output="专业的客户回复",
        agent=writer
    )

    crew = Crew(agents=[expert, writer], tasks=[search_task, answer_task],
                process=Process.sequential, verbose=False)
    result = crew.kickoff()
    return {"answer": result.raw, "auto_resolved": True, "source": "knowledge_base"}


# src/crews/refund_crew.py
def process_refund(ticket: dict, assessor, writer, llm) -> dict:
    """退款处理"""
    order_id = ticket.get("order_id", "未提供")
    content = ticket.get("content", "")

    assess_task = Task(
        description=f"评估退款申请。订单号:{order_id}。客户描述:{content[:500]}。判断合规性、退款金额。",
        expected_output="退款审核结果和金额",
        agent=assessor
    )
    reply_task = Task(
        description="基于审核结果撰写客户回复",
        expected_output="客户回复",
        agent=writer
    )

    crew = Crew(agents=[assessor, writer], tasks=[assess_task, reply_task],
                process=Process.sequential, verbose=False)
    result = crew.kickoff()

    # 从结果中提取金额（简化）
    import re
    amount_match = re.search(r'(\d+\.?\d*)元', result.raw)
    estimated_amount = float(amount_match.group(1)) if amount_match else 0

    return {"answer": result.raw, "auto_resolved": True, "amount": estimated_amount}


# src/crews/incident_crew.py
def diagnose_incident(content: str, diagnoser, writer, llm) -> dict:
    """故障诊断"""
    diag_task = Task(
        description=f"诊断以下故障: {content[:800]}。判断严重程度(critical/high/medium/low)。",
        expected_output="故障诊断报告和严重程度",
        agent=diagnoser
    )
    reply_task = Task(
        description="基于诊断结果撰写客户回复，严重故障需告知升级流程",
        expected_output="客户回复",
        agent=writer
    )

    crew = Crew(agents=[diagnoser, writer], tasks=[diag_task, reply_task],
                process=Process.sequential, verbose=False)
    result = crew.kickoff()

    severity = "medium"
    if "critical" in result.raw.lower() or "严重" in result.raw:
        severity = "critical"
    elif "high" in result.raw.lower():
        severity = "high"

    return {"answer": result.raw, "auto_resolved": True, "severity": severity}


# src/crews/complaint_crew.py
def handle_complaint(content: str, handler, writer, llm) -> dict:
    """投诉处理"""
    complaint_task = Task(
        description=f"处理客户投诉: {content[:800]}。评估严重程度，给出安抚和补偿方案。",
        expected_output="投诉处理方案",
        agent=handler
    )
    reply_task = Task(
        description="基于处理方案撰写安抚性回复",
        expected_output="安抚性客户回复",
        agent=writer
    )

    crew = Crew(agents=[handler, writer], tasks=[complaint_task, reply_task],
                process=Process.sequential, verbose=False)
    result = crew.kickoff()

    severity = "medium"
    if "重大" in result.raw or "critical" in result.raw.lower():
        severity = "critical"
    elif "严重" in result.raw:
        severity = "severe"

    return {"answer": result.raw, "auto_resolved": True, "severity": severity}

步骤4：FastAPI 统一入口

# src/main.py
from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
import os
import uuid
import time

from src.flow.ticket_flow import TicketFlow
from src.telemetry.reporter import PlatformReporter


class TicketRequest(BaseModel):
    customer_name: str = Field(description="客户名称")
    content: str = Field(description="咨询内容", min_length=5, max_length=5000)
    channel: str = Field(default="web", description="渠道")
    order_id: str = Field(default="", description="关联订单号（如有）")


class TicketResponse(BaseModel):
    ticket_id: str
    category: str
    answer: str
    auto_resolved: bool
    needs_human_review: bool
    duration_seconds: float


app = FastAPI(title="智能客服平台 API", version="3.0.0")
reporter = PlatformReporter()


@app.post("/api/v1/tickets", response_model=TicketResponse)
async def create_ticket(request: TicketRequest):
    """创建工单并自动处理"""
    ticket_id = str(uuid.uuid4())[:8]

    flow = TicketFlow()
    flow.state["ticket"] = {
        "id": ticket_id,
        "customer_name": request.customer_name,
        "content": request.content,
        "channel": request.channel,
        "order_id": request.order_id,
    }

    flow.kickoff()

    response_data = {
        "ticket_id": ticket_id,
        "category": flow.state.get("category", "other"),
        "answer": flow.state.get("response", {}).get("answer", "处理中，请稍后"),
        "auto_resolved": flow.state.get("response", {}).get("auto_resolved", False),
        "needs_human_review": flow.state.get("needs_human_review", False),
        "duration_seconds": flow.state.get("duration_seconds", 0),
    }

    # 记录运行数据
    reporter.record_ticket(response_data)

    return response_data


@app.get("/api/v1/reports/daily")
async def get_daily_report():
    """获取今日运行报告"""
    return reporter.generate_daily_report()


@app.get("/health")
async def health():
    return {"status": "healthy", "model": os.getenv("MODEL_NAME", "gpt-4o-mini")}

步骤5：运行报表

# src/telemetry/reporter.py
import json
import os
import threading
from datetime import datetime
from collections import defaultdict


class PlatformReporter:
    """平台运行数据收集和报表生成"""

    def __init__(self, data_file: str = "outputs/platform_data.jsonl"):
        self.data_file = data_file
        os.makedirs(os.path.dirname(data_file), exist_ok=True)
        self._lock = threading.Lock()

    def record_ticket(self, ticket_data: dict):
        """记录一个工单的处理数据"""
        record = {
            "timestamp": datetime.now().isoformat(),
            **ticket_data
        }
        with self._lock:
            with open(self.data_file, "a", encoding="utf-8") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")

    def generate_daily_report(self) -> dict:
        """生成今日运行报告"""
        today = datetime.now().strftime("%Y-%m-%d")
        records = []

        if os.path.exists(self.data_file):
            with open(self.data_file, "r", encoding="utf-8") as f:
                for line in f:
                    r = json.loads(line.strip())
                    if r.get("timestamp", "").startswith(today):
                        records.append(r)

        if not records:
            return {"date": today, "total": 0, "message": "今日无工单记录"}

        # 统计
        total = len(records)
        auto_resolved = sum(1 for r in records if r.get("auto_resolved", False))
        needs_review = sum(1 for r in records if r.get("needs_human_review", False))

        categories = defaultdict(int)
        avg_duration = 0
        for r in records:
            categories[r.get("category", "other")] += 1
            avg_duration += r.get("duration_seconds", 0)

        avg_duration = round(avg_duration / max(total, 1), 1)
        auto_rate = round(auto_resolved / max(total, 1) * 100, 1)
        miss_rate = round(
            sum(1 for r in records if r.get("category") in ["incident", "complaint"]
                and not r.get("needs_human_review", False))
            / max(total, 1) * 100, 1
        )

        return {
            "date": today,
            "total_tickets": total,
            "auto_resolved": auto_resolved,
            "auto_resolution_rate_pct": auto_rate,
            "needs_human_review": needs_review,
            "high_risk_miss_rate_pct": miss_rate,
            "avg_duration_seconds": avg_duration,
            "by_category": dict(categories),
            "sla_status": "达标" if auto_rate >= 85 and miss_rate < 2 else "未达标",
        }

步骤6：测试全流程

# tests/test_platform.py
import pytest
from src.flow.ticket_flow import TicketFlow


def test_classify_qa():
    """验证常见问题被正确分类为 qa"""
    flow = TicketFlow()
    flow.state["ticket"] = {
        "id": "T001", "customer_name": "测试",
        "content": "请问怎么重置密码？我试了好几次都不行。"
    }
    flow.kickoff()
    assert flow.state.get("category") in ["qa", "other"]


def test_classify_refund():
    """验证退款请求被正确分类"""
    flow = TicketFlow()
    flow.state["ticket"] = {
        "id": "T002", "customer_name": "测试",
        "content": "我刚买的会员想退款，订单号是ORD-001，可以退吗？"
    }
    flow.kickoff()
    assert flow.state.get("category") in ["refund", "qa", "other"]
    # 注意：分类结果取决于 LLM 的实际输出


def test_classify_incident():
    """验证故障报告被正确分类"""
    flow = TicketFlow()
    flow.state["ticket"] = {
        "id": "T003", "customer_name": "测试",
        "content": "系统登录一直报500错误，已经1小时了，公司所有同事都登不上去！"
    }
    flow.kickoff()
    assert flow.state.get("category") in ["incident", "complaint", "other"]


def test_flow_has_response():
    """验证 Flow 的处理结果中有回复内容"""
    flow = TicketFlow()
    flow.state["ticket"] = {
        "id": "T004", "customer_name": "测试",
        "content": "如何导出数据报表？"
    }
    flow.kickoff()
    assert "response" in flow.state
    assert flow.state["response"] is not None


def test_platform_reporter():
    """验证报表功能"""
    from src.telemetry.reporter import PlatformReporter
    reporter = PlatformReporter(data_file="outputs/test_platform.jsonl")

    reporter.record_ticket({
        "ticket_id": "T001", "category": "qa",
        "auto_resolved": True, "needs_human_review": False,
        "duration_seconds": 3.5
    })

    report = reporter.generate_daily_report()
    assert report["total_tickets"] >= 1
    assert report["auto_resolution_rate_pct"] >= 0

    # 清理
    if os.path.exists("outputs/test_platform.jsonl"):
        os.remove("outputs/test_platform.jsonl")

pytest tests/test_platform.py -v
# 5 passed

3.4 验收标准与测试

指标	目标值	测量方式	本章实测
常见问题自动回复准确率	>85%	第21章评估体系	Golden Dataset 测试
高风险工单漏报率	<2%	运行报表	incident+complaint 漏标记比例
自动处理率	>60%	运行报表	auto_resolved / total
平均响应时间	<60s	运行报表	duration_seconds 均值
人工处理时长下降	>30%	对比人工基线	需上线后实测

3.5 部署命令

# 本地开发
docker-compose up -d

# 生产部署
kubectl apply -f k8s/

# 灰度发布
kubectl set image deployment/smart-cs-api \
  smart-cs-api=smart-cs-api:v3.1

# 查看运行报表
curl http://localhost:8000/api/v1/reports/daily

4. 项目总结

4.1 中级篇知识融合对照表

章节	知识点	在本章的应用
第16章	业务建模	三层架构设计（接入/编排/能力）
第17章	Tool 进阶	订单查询 Tool（对接 CRM）
第18章	RAG 知识库	QA Crew 的知识库问答
第19章	复杂编排	Flow 的分类+路由+并行处理
第20章	多 Agent 协作	各 Crew 内的专家分工
第21章	评估体系	每天自动评估自动回答准确率
第22章	可观测性	运行报表（自动处理率/漏报率/耗时）
第23章	批处理	每日 0 点运行日报生成
第24章	Web 服务	FastAPI 统一 API 入口
第25章	消息机器人	高风险告警推送企业微信
第26章	CI 诊断	故障工单的 incident_crew
第27章	数据分析	运行报表的数据聚合
第28章	代码评审	（不直接使用——但平台代码本身受益于第28章方法）
第29章	部署运维	Docker/K8s 生产部署

4.2 优点与缺点

维度	智能客服平台	纯人工客服
响应速度	30-60s	3-30min
7×24可用性	是	需排班
一致性	高（知识库驱动）	取决于个人
成本	中（LLM API费用）	高（人力成本）
复杂问题处理	弱（需人工接手）	强

4.3 推广路线图（3阶段）

Phase 1（试点 2 周）：选 3 个客服组试用，只开启 QA 和 Incident 两个 Crew。收集反馈，优化分类准确率。
Phase 2（扩展 1 个月）：开启所有 4 个 Crew。接入真实 CRM 和知识库。建立 Golden Dataset。
Phase 3（全量 3 个月）：全量推广。自动化率从 60% 迭代到 80%+。沉淀为团队 AI 平台规范。

4.4 注意事项

1. 分类准确率是核心指标：分类错误会导致工单被路由到错误的分支——退款被当成故障分析、投诉被当成常见问题回复。分类准确率必须 >90%。
2. 人工审核不能省：自动回答可以全自动，但退款 >500 元、投诉严重度 >= severe、故障等级 >= high——这三类必须人工审核。
3. 知识库是平台的灵魂：文档不全、过时——Agent 的回答就不可能准确。知识库维护需要投入专人。

4.5 思考题（展望高级篇）

1. 平台运行 3 个月后，你发现"自动回答准确率"从 88% 掉到了 82%——但指标监控没有报警。你回头看才发现是知识库文档更新了 2 次，但没有人更新 Golden Dataset。如何设计一个自动化机制来检测"知识库漂移"？

2. 如果要将这个平台从"单租户"升级为"多租户"（服务多个 SaaS 客户，每个客户有独立的知识库和退款政策），你需要在哪些层面做隔离？Agent 设计、知识库、Tool 权限分别需要什么改造？

这些问题的探索之旅将在高级篇（第31-40章）中展开。

延伸阅读与资源

10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析