AI项目实战（4）智能客服 Agent

@tool
def query_order(order_id: str) -> str:
    """根据订单号查询订单状态、商品和金额。"""
    conn = None
    try:
        conn = _get_db()
        row = conn.execute("SELECT ...").fetchone()
        if not row:
            return f"未找到订单 {order_id}，请核对订单号后重试。"
        return f"订单 {order_id}：{row['item']}..."
    except sqlite3.Error as e:
        logger.error("查询订单失败: order_id=%s error=%s", order_id, e)
        return "订单查询服务暂时不可用，请稍后重试或联系人工客服。"
    finally:
        if conn:
            conn.close()

"""领域模型层：定义订单、物流、政策等核心数据结构。"""
from __future__ import annotations

from dataclasses import dataclass
from enum import Enum


class OrderStatus(str, Enum):
    """订单状态枚举。"""
    PENDING_PAYMENT = "待付款"
    PAID = "已付款"
    SHIPPED = "已发货"
    DELIVERED = "已签收"
    CANCELLED = "已取消"


@dataclass
class Order:
    """订单领域模型。"""
    id: str
    item: str
    amount: float
    status: OrderStatus
    created_at: str

    def __str__(self) -> str:
        return f"订单 {self.id}：{self.item}，金额 {self.amount} 元，状态：{self.status}。"


@dataclass
class LogisticsInfo:
    """物流信息领域模型。"""
    order_id: str
    tracking_info: str
    updated_at: str

    def __str__(self) -> str:
        return f"订单 {self.order_id} 物流：{self.tracking_info}"


@dataclass
class Policy:
    """售后政策领域模型。"""
    id: int
    topic: str
    content: str

    def __str__(self) -> str:
        return f"【{self.topic}】{self.content}"

"""数据仓储层：SQLite 持久化存储。

这是三层架构的中间层，负责数据访问。
封装数据库操作，向上层提供领域模型对象。

> 设计说明：
> - SQLite 适合教学演示和单机部署，零配置零运维
> - 生产环境可替换为 PostgreSQL，只需修改连接字符串和驱动
> - 仓储层接口保持不变，上层代码无需修改
"""
from __future__ import annotations

import sqlite3
from functools import lru_cache
from pathlib import Path
from typing import List, Optional

from core.config import DATA_DIR
from core.logging_conf import get_logger

from .models import LogisticsInfo, Order, OrderStatus, Policy

logger = get_logger("p01.repository")
DB_PATH = DATA_DIR / "customer_service.db"


class CustomerServiceRepository:
    """客服数据仓储，封装所有数据库操作。

    所有方法均返回领域模型对象或 None，上层无需处理原始 Row。
    """

    def __init__(self, db_path: Path = DB_PATH) -> None:
        self._db_path = db_path
        self._init_db()

    def _get_db(self) -> sqlite3.Connection:
        """获取数据库连接（每次新建，避免跨线程共享问题）。"""
        conn = sqlite3.connect(str(self._db_path))
        conn.row_factory = sqlite3.Row
        return conn

    def _init_db(self) -> None:
        """初始化数据库表与种子数据（幂等）。"""
        conn = self._get_db()
        conn.executescript("""
            CREATE TABLE IF NOT EXISTS orders (
                id TEXT PRIMARY KEY,
                item TEXT NOT NULL,
                amount REAL NOT NULL,
                status TEXT NOT NULL DEFAULT '待付款',
                created_at TEXT NOT NULL DEFAULT (datetime('now'))
            );
            CREATE TABLE IF NOT EXISTS logistics (
                order_id TEXT PRIMARY KEY,
                tracking_info TEXT NOT NULL,
                updated_at TEXT NOT NULL DEFAULT (datetime('now')),
                FOREIGN KEY (order_id) REFERENCES orders(id)
            );
            CREATE TABLE IF NOT EXISTS policies (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                content TEXT NOT NULL
            );
        """)
        # 种子数据（仅当表为空时插入）
        cur = conn.execute("SELECT COUNT(*) FROM orders")
        if cur.fetchone()[0] == 0:
            conn.executescript("""
                INSERT INTO orders(id, item, amount, status) VALUES
                    ('10001', '无线耳机', 299, '已发货'),
                    ('10002', '机械键盘', 459, '待付款'),
                    ('10003', '便携充电宝', 129, '已签收');
                INSERT INTO logistics(order_id, tracking_info) VALUES
                    ('10001', '已到达【杭州转运中心】，预计明天送达');
                INSERT INTO policies(topic, content) VALUES
                    ('退换货',
                     '商品签收后 7 天内无理由退货（商品需保持完好）；'
                     '质量问题 15 天内可换货，运费由商家承担。'),
                    ('发票',
                     '电子发票在签收后 3 个工作日内发送至您的注册邮箱。');
            """)
        conn.commit()
        conn.close()

    def get_order(self, order_id: str) -> Optional[Order]:
        """根据订单号查询订单，返回领域模型或 None。"""
        conn = None
        try:
            conn = self._get_db()
            row = conn.execute(
                "SELECT id, item, amount, status, created_at FROM orders WHERE id = ?",
                (order_id,),
            ).fetchone()
            if not row:
                return None
            return Order(
                id=row["id"],
                item=row["item"],
                amount=row["amount"],
                status=OrderStatus(row["status"]),
                created_at=row["created_at"],
            )
        except sqlite3.Error as e:
            logger.error("查询订单失败: order_id=%s error=%s", order_id, e)
            raise  # 仓储层只记录日志，向上抛出异常
        finally:
            if conn:
                conn.close()

    def get_logistics(self, order_id: str) -> Optional[LogisticsInfo]:
        """根据订单号查询物流信息。"""
        conn = None
        try:
            conn = self._get_db()
            row = conn.execute(
                "SELECT order_id, tracking_info, updated_at FROM logistics WHERE order_id = ?",
                (order_id,),
            ).fetchone()
            if not row:
                return None
            return LogisticsInfo(
                order_id=row["order_id"],
                tracking_info=row["tracking_info"],
                updated_at=row["updated_at"],
            )
        except sqlite3.Error as e:
            logger.error("查询物流失败: order_id=%s error=%s", order_id, e)
            raise
        finally:
            if conn:
                conn.close()

    def search_policies(self, keyword: str) -> List[Policy]:
        """根据关键词搜索售后政策。"""
        conn = None
        try:
            conn = self._get_db()
            rows = conn.execute(
                "SELECT id, topic, content FROM policies "
                "WHERE topic LIKE ? OR content LIKE ?",
                (f"%{keyword}%", f"%{keyword}%"),
            ).fetchall()
            return [
                Policy(
                    id=r["id"],
                    topic=r["topic"],
                    content=r["content"],
                )
                for r in rows
            ]
        except sqlite3.Error as e:
            logger.error("查询政策失败: keyword=%s error=%s", keyword, e)
            raise
        finally:
            if conn:
                conn.close()


@lru_cache(maxsize=None)
def get_repository() -> CustomerServiceRepository:
    """获取仓储单例（线程安全）。

    使用 functools.lru_cache 实现单例模式，避免多线程环境下重复初始化。
    """
    return CustomerServiceRepository()

"""工具层：定义 Agent 可调用的工具函数。

这是三层架构的上层，依赖仓储层。
每个工具都带有错误处理和降级，保证生产稳定性。
"""
from __future__ import annotations

from langchain.tools import tool

from core.logging_conf import get_logger
from .repositories import get_repository

logger = get_logger("p01.tools")


@tool
def query_order(order_id: str) -> str:
    """根据订单号查询订单状态、商品和金额。"""
    try:
        repo = get_repository()
        order = repo.get_order(order_id)
        if not order:
            return f"未找到订单 {order_id}，请核对订单号后重试。"
        return str(order)
    except Exception as e:
        logger.exception("查询订单失败: order_id=%s error=%s", order_id, e)
        return "订单查询服务暂时不可用，请稍后重试或联系人工客服。"


@tool
def query_logistics(order_id: str) -> str:
    """根据订单号查询物流配送进度。"""
    try:
        repo = get_repository()
        logistics = repo.get_logistics(order_id)
        if not logistics:
            return f"订单 {order_id} 暂无物流信息（可能尚未发货）。"
        return str(logistics)
    except Exception as e:
        logger.exception("查询物流失败: order_id=%s error=%s", order_id, e)
        return "物流查询服务暂时不可用，请稍后重试。"


@tool
def query_policy(topic: str) -> str:
    """查询售后政策（退货、换货、发票等）。"""
    try:
        repo = get_repository()
        policies = repo.search_policies(topic)
        if not policies:
            return f"未找到与「{topic}」相关的政策，请尝试其他关键词或转人工客服。"
        return "\n\n".join(str(p) for p in policies)
    except Exception as e:
        logger.exception("查询政策失败: topic=%s error=%s", topic, e)
        return "政策查询服务暂时不可用，请稍后重试。"

"""Prompt 层：构建 System Prompt。"""
from __future__ import annotations


def build_system_prompt(brand: str = "云购商城") -> str:
    """按品牌生成 System Prompt。

    设计原则：
    1. 角色设定：第一句明确身份
    2. 工具使用约束：必须调用工具，不得凭空回答
    3. 语气风格：亲切、专业、简洁
    4. 边界设定：超出能力礼貌转人工
    5. 错误传播：工具错误如实告知用户
    """
    return f"""你是「{brand}」的金牌智能客服，热情、专业、简洁。

工作要求：
1. 涉及订单、物流、售后政策的问题，必须调用相应工具获取准确信息，不得凭空回答。
2. 回答用中文，语气亲切，先共情再给方案。
3. 如果用户问题超出你的能力（如投诉、退款审批、价格谈判），
   礼貌告知将转接人工客服，并说明原因。
4. 不要编造订单号、政策内容或价格信息。
5. 如果工具返回错误（如"服务不可用"），如实告知用户并建议稍后重试。"""

"""Agent 服务层：Agent 工厂。"""
from __future__ import annotations

from langchain_core.language_models import BaseChatModel
from langchain_core.tools import BaseTool

from core import build_chat_model, get_settings
from .prompts import build_system_prompt
from .tools import query_logistics, query_order, query_policy


class CustomerServiceAgentFactory:
    """客服 Agent 工厂，统一构建 Agent 实例。"""

    @staticmethod
    def get_tools() -> list[BaseTool]:
        """获取客服可用的所有工具。"""
        return [query_order, query_logistics, query_policy]

    @staticmethod
    def build_agent(
        model: BaseChatModel | None = None,
        system_prompt: str | None = None,
    ):
        """构建智能客服 Agent。"""
        from langchain.agents import create_agent

        if model is None:
            model = build_chat_model()
        if system_prompt is None:
            system_prompt = build_system_prompt()
        tools = CustomerServiceAgentFactory.get_tools()

        return create_agent(
            model=model,
            tools=tools,
            system_prompt=system_prompt,
        )

"""项目定义层：Project 子类。"""
from __future__ import annotations

from core import BaseProject
from .service import CustomerServiceAgentFactory


class CustomerServiceProject(BaseProject):
    """智能客服项目。

    继承 BaseProject，自动获得：
    - 模型调用自动重试
    - 流控检查
    - 指标收集
    - SSE 流式支持
    """

    id = "p01_customer_service"
    name = "智能客服 Agent"
    description = "电商客服：查订单、查物流、查售后政策，超范围转人工。"
    capabilities = ["Tool", "Prompt"]

    def build_agent(self):
        """构建 Agent 实例。"""
        return CustomerServiceAgentFactory.build_agent()

"""项目一：智能客服 Agent（生产级）。"""
from __future__ import annotations

from core import registry

from .project import CustomerServiceProject
from .tools import query_logistics, query_order, query_policy

# 注册到全局路由表
registry.register(CustomerServiceProject())

__all__ = [
    "CustomerServiceProject",
    "query_order",
    "query_logistics",
    "query_policy",
]

# 这行代码在重构前后完全一样
from projects.p01_customer_service import query_order, CustomerServiceProject

class TestQueryOrder:
    def test_order_found(self):
        out = query_order.invoke({"order_id": "10001"})
        assert "无线耳机" in out and "已发货" in out and "299" in out

    def test_order_not_found(self):
        out = query_order.invoke({"order_id": "99999"})
        assert "未找到" in out

    def test_order_db_error_graceful(self, monkeypatch):
        """验证数据库异常时工具优雅降级而非崩溃。"""
        def _boom(*a, **kw):
            raise sqlite3.Error("模拟数据库故障")
        monkeypatch.setattr(sqlite3, "connect", _boom)
        out = query_order.invoke({"order_id": "10001"})
        assert "暂时不可用" in out  # 降级提示，而非崩溃

class TestAgentIntegration:
    def test_order_query_uses_tool(self):
        project = CustomerServiceProject()
        reply = project.run("帮我查订单 10001 的状态")
        assert "10001" in reply

    def test_out_of_scope_escalates(self):
        project = CustomerServiceProject()
        reply = project.run("我要投诉你们发货太慢")
        assert any(w in reply for w in ["转接", "人工", "客服"])

    def test_metrics_incremented(self):
        project = CustomerServiceProject()
        before = project.metrics.total_runs
        project.run("查订单 10001")
        assert project.metrics.total_runs == before + 1