智能客服 Agent 专项设计方案（内部评审稿）

0. 文档说明与评审指引

0.1 定位与范围

本文档是《智能财富管家系统-业务与技术全景设计说明书》中「智能客服 Agent」的深化专项设计，聚焦对外客户服务入口这一条链路（用户提问 → 意图识别 → 知识检索 → 答案生成 → 合规校验 → 人工转接 → 结果反馈），供内部方案评审使用。

范围边界：只覆盖智能客服 Agent（自然语言问答 / FAQ / 产品一般性介绍 / 政策解读 / 转人工）。投顾助手、风控监测、数据分析三个 Agent 不在本文档范围（详见全景说明书 §5.3/§5.4/§5.5）。

设计原则：所有结论都落到项目实际内容（意图、集合、阈值、表、接口），并对照主流平台真实落地做法标注来源；信息不确定处标注「需核实」，不做臆测。

0.2 依据文件（优先级从高到低）

0.3 合规基线速览（本文档反复引用）

0.4 评审关注点（读前提示）

1. §2 闭环流程中「合规校验」是否应前置到「答案生成」之前（当前为生成后置校验 + 二次打回）。
2. §3.3「费率与收益计算」的收益口径：字段名 expected_return，对外一律表述「业绩比较基准」并附测算依据，不得出现「预期收益率」。
3. §5 转人工阈值（意图 0.6、RAG 混合判定）是否需按业务容忍度二次校准。
4. §6 回复话术的「负面词 7 个」是否要扩充（如「零风险」「稳赚不赔」等变体）。

1. 定位与设计目标

1.1 一句话定位

智能客服 Agent 是系统的唯一对外自然语言入口，负责把「客户想问的」映射到「合规能答的」，答不了的果断转人工——价值在于 7×24 秒回标准问题，边界在于绝不越合规红线。

1.2 服务对象与边界

1.3 业务目标与度量指标

1.4 合规红线（客服 Agent 视角）

客服 Agent 的职责是「信息提供」，因此它只读不写、只答不荐、只解释不下结论：

行业对标：华云天下提出的金融 AI 客服「十条铁律」首条即「不能承诺收益」——任何含「左右/大约/预期」的收益数字都不允许〔S13〕。招商基金总经理级口径强调「AI 不能代替信任、责任与人文关怀」〔S7〕，与本项目「不代客交易」定位同源。

1.5 AI 接入前后的工作流程对比

本节用两张流程对比图说明：AI 智能客服解决了什么问题、在哪些环节提升了效率。量化结论见 §1.3 度量指标。

1.5.1 未接入 AI 智能客服时的工作流程

flowchart TD
    A[用户有问题] --> B[电话/邮件/线下网点求助]
    B --> C[排队等待人工客服]
    C --> D{人工客服在线?}
    D -- 否（夜间/节假日） --> E[等到工作日再处理]
    E --> C
    D -- 是 --> F[人工逐条答疑
口径依赖个人经验]
    F --> G{能解决?}
    G -- 是 --> H[解决 + 人工记录]
    G -- 否 --> I[转资深客服/持证投顾]
    I --> F
    H --> J[人工汇总工单/报表
易遗漏]

痛点：响应慢（排队 + 非工作时间无人值守）、口径不一致（靠个人经验）、重复劳动高、留痕靠人工易遗漏。

1.5.2 接入 AI 智能客服后的工作流程

flowchart TD
    A[用户提问] --> B[AI 7×24 秒级响应]
    B --> C{意图识别 + 置信度}
    C -- 标准问题 --> D[三集合检索 → 秒回]
    C -- 低置信/情绪/投诉 --> E[自动转人工
携带完整上下文]
    D --> F[合规校验
负面词 + 免责声明]
    F --> G[结果反馈 SSE]
    E --> H[人工接管
聚焦复杂问题]
    G --> I[自动归档 + 脱敏留痕]
    H --> I

1.5.3 两者对比（AI 解决了什么 / 在哪提升效率）

2. 整体流程闭环

2.1 闭环流程图

flowchart TD
    A[用户提问] --> B[① 输入接收与校验鉴权]
    B --> B1{鉴权通过?}
    B1 -- 否 --> B2[返回登录引导]
    B1 -- 是 --> C[② 记忆召回
读 Redis 短期会话]
    C --> D[③ 意图路由
qwen-turbo 分类·阈值0.6]
    D --> D1{置信度>=0.6?}
    D1 -- 否 --> D2[澄清提问]
    D2 --> D3{仍无法识别?}
    D3 -- 是 --> H[转人工]
    D3 -- 否 --> D
    D1 -- 是 --> E[④ 知识检索
Milvus 三集合路由]
    E --> F[⑤ 答案生成
temperature 0.3 + 来源引用]
    F --> G[⑥ 合规校验
负面词后置 + 免责声明注入]
    G --> G1{命中负面词?}
    G1 -- 是 --> G2[打回重生成]
    G2 --> G3{二次仍命中?}
    G3 -- 是 --> G4[返回安全话术 + 告警]
    G3 -- 否 --> F
    G1 -- 否 --> I[⑦ 结果反馈 SSE 流式]
    I --> J{需转人工?}
    J -- 是 --> H
    J -- 否 --> K[数据沉淀：写回会话+归档脱敏]
    H --> K
    K --> L[事件广播：命中敏感意图 publish suspicious_intent]

2.2 七步拆解（与统一执行骨架对齐）

对齐说明：⑥⑦ 由基类强制，防止「某个 Agent 忘了归档/忘了广播」——这是合规留痕与跨 Agent 协作的结构性保障〔全景说明书 §8〕。

2.3 关键决策点：置信度与三档兜底

2.4 三集合路由代码与 Milvus 关键概念（带详细中文注释）

§2.1 的流程图里多次出现「Milvus 三集合路由」——这里给出对应工程模块（tool/milvus_tool.py，与本文档 §0.2 / §4.1 引用一致）的完整代码与逐行中文注释，方便评审时逐字段、逐分支核对。

阅读顺序：先看 2.4.1 概念与原理（建立直觉）→ 2.4.2 三级路由职责速查 → 2.4.3 Milvus 概念速查（扫盲）→ 2.4.4 完整代码（带注释的实现）→ 2.4.6 操作类查询规划。

2.4.1 三集合路由的概念与原理（先看懂，再看代码）

一句话定义：三集合路由 = 把知识库分成三个「书架」（FAQ / 产品 / 政策），用户每问一个问题，系统先判断「该去哪个/哪几个书架找」，再去对应书架里用「语义相似度」找出最相关的资料。

用一个图书馆的类比理解：

为什么需要「路由」而不是一个大集合：如果把 FAQ、产品、政策全塞进一个大集合，问「赎回费怎么算」时，可能把「反洗钱政策」也检索出来，干扰答案。分成三个集合 + 路由，等于先缩小查找范围，再精确定位——既快又准。

原理（三步走）：

flowchart LR
    A[用户问题] --> B[意图识别
这是哪类问题?]
    B --> C{路由决策
去哪个集合}
    C -- 高频问答 --> D1[fin_faq_collection]
    C -- 产品咨询 --> D2[fin_product_collection]
    C -- 政策解读 --> D3[fin_policy_collection]
    D1 & D2 & D3 --> E[集合内向量检索
找最相关的 topK 条]
    E --> F[重排去重
输出最终素材]

注意：这里的「路由」不是网络路由器（转发数据包），而是「按问题类型分发检索目标」的业务路由——两者同名，含义完全不同。

2.4.2 三级路由职责速查表

「三集合路由」= 在三个 Milvus collection（fin_faq / fin_product / fin_policy）之间，按用户意图选集合，再在集合内做向量检索。写入路径另有 upsert_document（往集合里塞新数据/更新旧数据）。

2.4.3 Milvus 关键概念速查（给不熟悉 Milvus 的同学）

2.4.4 完整代码（带详细中文注释）

下面的代码可直接保存为 tool/milvus_tool.py。每一行 # 后的中文注释都对不熟悉 Milvus / RAG 的读者做了通俗解释。

# -*- coding: utf-8 -*-
"""
milvus_tool.py
==============
智能客服 Agent 的「三集合路由」核心模块。

【三集合路由的定义和作用】
    智能客服 Agent 回答用户问题时，需要先在「知识库」里找到相关资料。
    我们的知识库分成了三个独立的 Milvus 集合（你可以理解为三张表），
    分别存放不同类型的资料：
        1. fin_faq_collection     —— 高频操作类问答（如「怎么重置密码」）
        2. fin_product_collection —— 产品说明书、费率、风险等级
        3. fin_policy_collection  —— 监管法规、适当性规则、销售办法
    为什么要分三个？
        - 不同类型资料的「更新频率」不同（FAQ 周更，政策即时更新）
        - 不同意图需要「检索不同的内容」（问政策就不该去产品表里找）
        - 权限/审计可以按集合隔离

【三级路由的协作流程】
    用户提问 → 意图分类器判定意图 →
        第一级（意图路由）→ 决定去哪些集合 →
        第二级（集合内检索）→ 每个集合内做向量相似度搜索 →
        第三级（重排与融合）→ 合并多集合候选 → 输出最终 topK
    写入路径（更新/新增知识）：upsert_document() → 写入对应 collection
"""
from __future__ import annotations  # 让类型注解在 Python 3.9 之前也能用
import time                        # 用 time.strftime 取当前日期
import hashlib                     # 用 md5 做内容指纹（去重用）
from typing import List, Dict, Any, Optional  # 类型注解，让代码可读性更强
from dataclasses import dataclass  # 用装饰器快速定义「数据类」（只装数据的简单类）

# 下面这行是 pymilvus 的官方 SDK（Milvus 团队维护的 Python 客户端）。
# 如果项目里还没装，请先：pip install pymilvus
from pymilvus import (
    connections,         # 用来连 Milvus 服务（建立网络连接）
    Collection,          # 集合对象 = 指向 Milvus 某张「表」的引用
    CollectionSchema,    # 集合的「表结构」= 字段列表
    FieldSchema,         # 单个字段的「列定义」
    DataType,            # 字段类型枚举：VARCHAR / FLOAT_VECTOR / INT64 ...
    utility,             # 工具函数：判断 collection 是否存在、查版本等
)


# ============================================================
# 一、常量区：把会变的配置集中起来（单一真相源）
# ============================================================
# 「单一真相源」= 全项目只有这一处定义这个值，避免散落各处导致改一处忘改另一处。
# 比如你想把端口换了，只改这里就行。

# Milvus 服务器地址。本地开发用 standalone，集群用 cluster 地址。
MILVUS_HOST = "127.0.0.1"
MILVUS_PORT = "19530"

# 三个集合的名字（= 三张「表」）。改这里就够了，不要在代码里到处写字符串字面量。
FAQ_COLLECTION      = "fin_faq_collection"
PRODUCT_COLLECTION   = "fin_product_collection"
POLICY_COLLECTION    = "fin_policy_collection"
ALL_COLLECTION_NAMES = [FAQ_COLLECTION, PRODUCT_COLLECTION, POLICY_COLLECTION]

# 意图 → 集合 的映射关系（= 「第一级路由」的配置表）
# 含义：用户问的是这个意图时，去这几个集合里找资料。
# 用列表而不是单个字符串，是允许「一个意图跨多个集合」的灵活设计。
INTENT_TO_COLLECTIONS: Dict[str, List[str]] = {
    "faq":              [FAQ_COLLECTION],                                       # 高频问题只在 FAQ 表
    "product_inquiry":  [PRODUCT_COLLECTION],                                   # 产品咨询去产品表
    "policy_explain":   [POLICY_COLLECTION],                                    # 政策解读去政策表
    "chitchat":         [],                                                     # 闲聊不去任何表（走 LLM 自由生成）
    "transfer_human":   [],                                                     # 转人工也不查表
    # 兜底：意图分类没把握时，三个集合都查，取并集后再重排
    "low_confidence":   [FAQ_COLLECTION, PRODUCT_COLLECTION, POLICY_COLLECTION],
}

# 每个集合检索时取 topK 条候选。TopK 越大召回越全但越慢。
COLLECTION_TOPK: Dict[str, int] = {
    FAQ_COLLECTION:     3,   # FAQ 是「一问一答」标准答案，取 3 条够用
    PRODUCT_COLLECTION: 5,   # 产品信息需要更多候选让 LLM 选
    POLICY_COLLECTION:  5,   # 政策条款同理
}

# 向量维度。要和 embedding 模型输出一致（如 Qwen Embedding 通常 1024 或 1536 维）。
# 「维度」= 一个向量有多少个数字。文本越长不一定维度越高，是模型决定的。
VECTOR_DIM = 1024

# 向量距离度量。COSINE = 余弦相似度（最常用，忽略向量长度只看方向）。
# 选 COSINE 而不是 L2（欧氏距离），因为我们关心「语义方向」而不是「距离绝对值」。
METRIC_TYPE = "COSINE"


# ============================================================
# 二、三个集合的「表结构」定义（schema）
# ============================================================
# Milvus 的 schema 就像建表语句：定义有哪些字段、各自什么类型。
# 用工厂函数（返回对象的函数）生成，方便统一管理 + 后续扩展。

def build_faq_schema() -> CollectionSchema:
    """FAQ 集合的表结构：高频问答。"""
    fields = [
        # 主键：用 md5(question) 做主键，保证同一问题只有一条记录
        FieldSchema("id",            DataType.VARCHAR, max_length=64, is_primary=True),
        FieldSchema("question",      DataType.VARCHAR, max_length=512),    # 标准问法
        FieldSchema("answer",        DataType.VARCHAR, max_length=2048),  # 标准答案
        FieldSchema("tags",          DataType.VARCHAR, max_length=256),   # 标签，JSON 字符串
        FieldSchema("effective_date",DataType.VARCHAR, max_length=32),    # 生效日期 "2026-01-01"
        FieldSchema("expire_date",   DataType.VARCHAR, max_length=32),    # 失效日期，空=长期有效
        # 向量字段：把 question 文本转成 embedding 存进来
        FieldSchema("embedding",     DataType.FLOAT_VECTOR, dim=VECTOR_DIM),
    ]
    # description 给这个集合写个说明，方便在 Milvus 管理界面（Attu）里辨认
    return CollectionSchema(fields, description="FAQ 集合：高频操作类问答")


def build_product_schema() -> CollectionSchema:
    """产品集合的表结构：产品说明书/费率/风险等级。"""
    fields = [
        FieldSchema("id",             DataType.VARCHAR, max_length=64, is_primary=True),
        FieldSchema("product_code",   DataType.VARCHAR, max_length=32),   # 产品代码，如 "000001"
        FieldSchema("product_name",   DataType.VARCHAR, max_length=128),
        FieldSchema("product_type",   DataType.VARCHAR, max_length=32),   # 股票/混合/债券
        FieldSchema("risk_level",     DataType.VARCHAR, max_length=8),    # R1-R5
        FieldSchema("content",        DataType.VARCHAR, max_length=4096), # 文档片段原文
        FieldSchema("source_url",     DataType.VARCHAR, max_length=512),
        FieldSchema("effective_date", DataType.VARCHAR, max_length=32),
        FieldSchema("embedding",      DataType.FLOAT_VECTOR, dim=VECTOR_DIM),
    ]
    return CollectionSchema(fields, description="产品集合：产品说明书/费率/风险")


def build_policy_schema() -> CollectionSchema:
    """政策集合的表结构：监管法规/适当性/销售办法。"""
    fields = [
        FieldSchema("id",             DataType.VARCHAR, max_length=64, is_primary=True),
        FieldSchema("doc_id",         DataType.VARCHAR, max_length=64),   # 法规文号，如 "银发2016261号"
        FieldSchema("title",          DataType.VARCHAR, max_length=256),
        FieldSchema("content",        DataType.VARCHAR, max_length=4096),
        FieldSchema("issuing_org",    DataType.VARCHAR, max_length=64),   # 发文机构：银保监会 / 证监会 ...
        FieldSchema("effective_date", DataType.VARCHAR, max_length=32),
        FieldSchema("embedding",      DataType.FLOAT_VECTOR, dim=VECTOR_DIM),
    ]
    return CollectionSchema(fields, description="政策集合：监管法规/适当性/销售办法")


# schema 字典：集合名 → schema 工厂函数
SCHEMA_FACTORIES: Dict[str, Any] = {
    FAQ_COLLECTION:     build_faq_schema,
    PRODUCT_COLLECTION: build_product_schema,
    POLICY_COLLECTION:  build_policy_schema,
}


# ============================================================
# 三、检索结果的数据类（让返回值有结构，方便后续处理）
# ============================================================
# @dataclass 装饰器自动生成 __init__ / __repr__ / __eq__ 等方法，代码更短。

@dataclass
class Hit:
    """单条检索命中结果。相当于「一行记录」+「相似度分」。"""
    collection: str              # 来自哪个集合
    doc_id:     str              # 该集合内的主键
    score:      float            # 相似度分（越大越像；范围约 0~1）
    content:    str              # 文本内容（问题/答案/产品/政策原文）
    metadata:   Dict[str, Any]   # 其他元数据（tags、product_code、raw_distance 等）


@dataclass
class RoutePlan:
    """第一级路由的输出：决定去哪些集合、每个取多少条。"""
    target_collections:  List[str]            # 要去查的集合名列表
    per_collection_topk: Dict[str, int]       # 每个集合分别取 topK


# ============================================================
# 四、三集合路由器（核心类）
# ============================================================

class ThreeCollectionRouter:
    """
    三集合路由器的核心类。
    一个实例 = 一个 Milvus 客户端连接。
    设计要点：复用连接、缓存 collection 句柄，避免每次检索都重新建连。
    """

    def __init__(self, host: str = MILVUS_HOST, port: str = MILVUS_PORT):
        # 1. 连 Milvus 服务
        self._connect(host, port)
        # 2. 懒加载：集合句柄按需获取（用 self.get_collection(name) 时才建），
        #    避免启动时全部加载占内存。
        # 「懒加载」= 用到时再加载，不用时不加载。
        self._collection_cache: Dict[str, Collection] = {}

    # ---------- 连接与初始化 ----------

    def _connect(self, host: str, port: str) -> None:
        """连 Milvus 服务。alias='default' 是默认连接名，后面 Collection 用默认连接。"""
        connections.connect(alias="default", host=host, port=port)
        # 用一个轻量调用确认连接成功（get_server_version() 返回类似 "v2.4.x"）
        if not utility.get_server_version():
            # 抛异常而不是静默失败，让上层知道出问题了
            raise RuntimeError(f"无法连接 Milvus 服务 {host}:{port}")

    def get_collection(self, name: str) -> Collection:
        """
        获取一个集合的句柄（带缓存）。
        Collection 对象 = 指向 Milvus 某张「表」的引用，搜索/写入都要用它。
        缓存的好处：第二次拿同一集合不用重新查 Milvus，省一次 RPC。
        """
        # 第一步：先看缓存里有没有
        if name not in self._collection_cache:
            # 第二步：缓存里没有，检查 Milvus 里有没有这个集合
            if not utility.has_collection(name):
                # 都没有 → 首次启动，自动建表
                self._create_collection(name)
            # 第三步：拿句柄
            coll = Collection(name)
            # 第四步：load() 把集合从磁盘加载到内存，否则搜索会报「collection not loaded」
            # Milvus 是「磁盘省钱、内存才快」的设计，不 load 就搜不了
            coll.load()
            # 放进缓存，下次直接拿
            self._collection_cache[name] = coll
        return self._collection_cache[name]

    def _create_collection(self, name: str) -> None:
        """首次启动时创建集合（建表 + 建索引）。"""
        # 用 schema 工厂拿表结构定义
        schema = SCHEMA_FACTORIES[name]()
        # 创建集合：参数 = 名字 + 表结构
        Collection(name, schema=schema)
        # 为向量字段建索引。
        # IVF_FLAT：常用索引，简单稳；HNSW 更精确但费内存；DISKANN 大数据量省钱。
        coll = Collection(name)
        coll.create_index(
            field_name="embedding",     # 给哪个字段建索引
            index_params={
                "metric_type": METRIC_TYPE,    # 用余弦相似度（前面常量区定义的）
                "index_type":  "IVF_FLAT",
                # nlist：聚类中心数。越大越精确但越慢；128 是常用起点
                "params":      {"nlist": 128},
            },
        )

    # ---------- 写入路径：往集合里塞一条新数据 / 更新旧数据 ----------

    def upsert_document(
        self,
        collection_name: str,
        doc_id: str,
        text: str,
        embedding: List[float],
        extra_fields: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        写入或更新一条文档。
        调用时机：运营上传新知识、费率变更、政策更新时。
        参数：
            collection_name: 写入哪个集合（faq / product / policy）
            doc_id:          主键（自己生成，md5 或雪花 ID 都行）
            text:            文本内容
            embedding:       text 转成的向量（调用 embedding 模型得到）
            extra_fields:    其他字段（product_code、effective_date 等）
        """
        coll = self.get_collection(collection_name)
        # 拼装一行数据。Milvus 要求：字段顺序和 schema 一致，且值都是列表（支持批量）。
        row = {
            "id":        [doc_id],
            "content":   [text],   # 简化：假设 schema 都有 content 字段
            "embedding": [embedding],
        }
        # 把额外字段补进去（如果 schema 里有）
        if extra_fields:
            for k, v in extra_fields.items():
                row[k] = [v]
        # upsert：存在则更新、不存在则插入（Milvus 用 upsert 统一两种操作）
        coll.upsert(row)
        # 写完刷一下：让数据立即可见、开始建索引。
        # flush 较重（要写盘），高频写入场景可以用 flush_buffer 批量刷
        coll.flush()

    # ---------- 读取路径：三集合路由的核心（对外入口） ----------

    def query(
        self,
        intent: str,
        query_text: str,
        query_embedding: List[float],
        user_id: Optional[int] = None,
        partition_name: Optional[str] = None,
    ) -> List[Hit]:
        """
        三级路由的对外入口——客服 Agent 每次问问题，都会调这个函数。
        参数：
            intent:         意图分类的结果（"faq" / "product_inquiry" / "policy_explain" / "low_confidence"）
            query_text:     用户的原始问题（中文）
            query_embedding: query_text 经 embedding 模型转成的向量（长度 = VECTOR_DIM）
            user_id:        当前用户 ID（用于权限过滤，例如只让看本人产品）
            partition_name: 可选——只查某个分区的数据（产品集合可按 product_type 分区）
        返回：经过三级路由后的候选结果列表（已重排）
        """
        # ============ 第一级：意图路由 → 决定去哪些集合 ============
        route_plan = self._step1_route_intent(intent, query_text)
        # 如果第一级判定不需要查表（闲聊/转人工），直接返回空，让上层走别的逻辑
        if not route_plan.target_collections:
            return []

        # ============ 第二级：集合内检索（多集合可并行）============
        # 多个集合之间没有依赖关系，可以并行查。这里用 for 串行演示；真实环境用线程池/异步并行。
        per_collection_hits: Dict[str, List[Hit]] = {}
        for coll_name in route_plan.target_collections:
            # 每个集合的 topK 从 route_plan 里取
            topk = route_plan.per_collection_topk.get(coll_name, 3)
            hits = self._step2_search_in_collection(
                coll_name=coll_name,
                query_embedding=query_embedding,
                top_k=topk,
                user_id=user_id,
                partition_name=partition_name,
            )
            per_collection_hits[coll_name] = hits

        # ============ 第三级：重排与融合 → 输出最终 topK ============
        final_hits = self._step3_rerank_and_merge(
            per_collection_hits=per_collection_hits,
            query_text=query_text,
        )
        return final_hits

    # ---------- 第一级：意图路由 ----------

    def _step1_route_intent(self, intent: str, query_text: str) -> RoutePlan:
        """
        第一级路由：决定去哪些集合、每个集合取多少条。
        决策逻辑：
            1. 查 INTENT_TO_COLLECTIONS 配置，看该意图默认对应哪些集合
            2. 如果意图不在配置里 → 走兜底：三个集合都查
        """
        # 默认配置里的目标集合列表
        targets = INTENT_TO_COLLECTIONS.get(intent)
        # 没找到意图配置 → 兜底：全集合都查
        if targets is None:
            # list() 是为了拷贝一份，避免后面修改时改了常量
            targets = list(ALL_COLLECTION_NAMES)

        # 构造每个集合的 topK 配置
        per_topk: Dict[str, int] = {}
        for name in targets:
            # 用 COLLECTION_TOPK 配置表查每个集合的 topK，找不到默认 3
            per_topk[name] = COLLECTION_TOPK.get(name, 3)
        return RoutePlan(target_collections=targets, per_collection_topk=per_topk)

    # ---------- 第二级：集合内检索 ----------

    def _step2_search_in_collection(
        self,
        coll_name: str,
        query_embedding: List[float],
        top_k: int,
        user_id: Optional[int],
        partition_name: Optional[str] = None,
    ) -> List[Hit]:
        """
        第二级路由：在某一个集合内做「向量相似度检索」。
        流程：
            1. 构造过滤表达式（标量过滤，比如「未过期」+「本人可见」）
            2. 用 embedding 向量在该集合内搜 topK（向量化最近邻）
            3. 把 Milvus 返回的原始结果转成 Hit 对象
        """
        coll = self.get_collection(coll_name)

        # 构造标量过滤条件（标量 = 非向量字段）。
        # 语法类似 SQL 的 WHERE 条件。含义：expire_date 为空 OR expire_date > 今天
        today = time.strftime("%Y-%m-%d")
        expr = f'expire_date == "" OR expire_date > "{today}"'

        # 产品集合再加权限过滤（简化示意：实际应该是 OR 公开产品 OR user_id 匹配）
        if coll_name == PRODUCT_COLLECTION and user_id is not None:
            expr += f' OR user_id == "{user_id}"'

        # 向量搜索参数：
        #   nprobe：搜索时探查的聚类数，越大越精确但越慢
        search_params = {"metric_type": METRIC_TYPE, "params": {"nprobe": 16}}

        # 准备 partition_names 参数（如果指定了 partition）
        # 搜索时指定 partition 可以大幅缩范围、提速
        search_kwargs: Dict[str, Any] = {
            "data": [query_embedding],      # 查询向量，外层是列表因为支持一次查多个
            "anns_field": "embedding",      # 用哪个向量字段做相似度
            "param": search_params,         # 搜索参数
            "limit": top_k,                 # topK
            "expr": expr,                   # 标量过滤条件
            "output_fields": ["id", "content"],  # 这些标量字段也一起返回
        }
        if partition_name:
            # 如果传了分区名，只在该分区内搜（产品集合可按 product_type 分区）
            search_kwargs["partition_names"] = [partition_name]

        # 执行搜索（anns_field = Approximate Nearest Neighbor Search field）
        results = coll.search(**search_kwargs)

        # 解析结果。results 是 [[Hit1, Hit2, ...]]，外层列表是因为可以一次查多个 query
        hits: List[Hit] = []
        for query_hits in results:
            for h in query_hits:
                # h.distance 是距离（越小越像）。
                # 余弦距离 = 1 - 余弦相似度，所以 score = 1 - distance
                # 我们把 score 统一成「越大越像」，方便后续按相似度排序
                score = 1.0 - h.distance
                hits.append(Hit(
                    collection=coll_name,
                    doc_id=h.id,                                  # 命中记录的主键
                    score=score,                                  # 相似度分（0~1）
                    content=h.entity.get("content", ""),         # 命中的文本内容
                    metadata={"raw_distance": h.distance},        # 保留原始距离方便排查
                ))
        return hits

    # ---------- 第三级：重排与融合 ----------

    def _step3_rerank_and_merge(
        self,
        per_collection_hits: Dict[str, List[Hit]],
        query_text: str,
    ) -> List[Hit]:
        """
        第三级路由：把多个集合的结果合并、重排、去重。
        策略：
            1. 把所有集合的 hits 合并到一个列表
            2. 按 score 降序排（相似度高的排前面）
            3. 简单去重：内容前 100 字相同的视为重复
            4. 截断到最终 topK（这里固定 5，可改为参数）
        进阶：可以接 rerank 模型（如 bge-reranker）做精排，这里简化。
        """
        # 1. 合并：把所有集合的 hits 展平成一个列表
        all_hits: List[Hit] = []
        for hits in per_collection_hits.values():
            all_hits.extend(hits)

        # 2. 按相似度分数降序排（lambda h: h.score 意思是「取 h 的 score 作为排序键」）
        all_hits.sort(key=lambda h: h.score, reverse=True)

        # 3. 去重（用内容前 100 字符的 md5 做指纹）
        # 为什么要去重？因为同一个答案可能在 FAQ 和产品集合里都出现（比如关于费率的说明）
        seen: set = set()
        deduped: List[Hit] = []
        for h in all_hits:
            # md5(内容前100字符) → 32 位字符串指纹
            fingerprint = hashlib.md5(h.content[:100].encode("utf-8")).hexdigest()
            if fingerprint in seen:
                continue  # 见过这个指纹了，跳过
            seen.add(fingerprint)
            deduped.append(h)

        # 4. 截断到最终 topK（这里固定 5，可改为参数或按业务调整）
        return deduped[:5]


# ============================================================
# 五、demo：怎么用
# ============================================================
# 下面是「怎么调用路由器」的最小示例。
# 真实 Agent 里：query_text 先送意图分类器（拿到 intent），再送 embedding 模型（拿到 vector），
# 然后调 router.query(...) 拿到候选 hits。

if __name__ == "__main__":
    # 1. 初始化路由器（自动连 Milvus、自动建表、自动 load）
    router = ThreeCollectionRouter()

    # 2. 模拟一次用户提问
    intent = "policy_explain"                       # 假设意图分类器说这是「政策解读」
    query_text = "什么叫冷静期？"                    # 用户原话
    # 真实场景：query_embedding = embedding_model.encode(query_text)
    # 这里用全 0 占位（真实项目必须用真实 embedding，否则相似度全为 0）
    query_embedding = [0.0] * VECTOR_DIM

    # 3. 三级路由检索
    hits = router.query(
        intent=intent,
        query_text=query_text,
        query_embedding=query_embedding,
        user_id=10001,                              # 当前用户 ID（用于权限过滤）
        # partition_name="股票型",                  # 可选：只查「股票型」分区
    )

    # 4. 打印结果
    for i, h in enumerate(hits, 1):
        print(f"[{i}] 集合={h.collection} 分数={h.score:.3f}")
        print(f"    内容: {h.content[:80]}...")
        print()

2.4.5 三级路由协作时序（一句话总结）

用户提问 ──► 意图分类 ──► 第一级(去哪个集合)
                          ──► 第二级(在集合里找 topK)
                              ──► 第三级(合并去重排序)
                                  ──► 交给 LLM 生成最终回复

写入路径（与读取路径独立）：upsert_document() 在运营/系统需要更新知识时调用，往指定 collection 塞一条或更新一条；不参与用户问问题的实时链路。

2.4.6 操作类查询的现状与实施规划（客户端未完成，暂缓）

本节回应一个常见疑问：为什么现在只有三个「知识类」集合，没有「操作类」（账户/持仓/交易查询）的向量库？

关键澄清：知识类 vs 操作类

为什么操作类暂缓：操作类查询需要「客户真实账户数据」，而这些数据由客户端（前端）+ 账户系统在真实业务运行中产生。客户端尚未开发完成，账户数据还不存在，因此现在无法实现操作类查询。

重要修正——操作类本质不是「向量库」：账户/持仓/交易查询是「精确查询」（查"我"的具体持仓、具体流水），应当通过 API 直连账户系统实现，而不是用向量库做「语义模糊检索」。向量库只服务于「答案散落在文档里的知识问答」。因此：

• 知识类（FAQ/产品/政策）→ 用 Milvus 向量库（本项目三集合）
• 操作类（账户/持仓/交易）→ 用账户系统 API 精确查询（待客户端完成后接入）

分阶段规划：

对 §3.1 账户与交易查询的影响：§3.1 将该场景列入 6 条业务路径，但当前其「可答范围」需标注为「Phase 2 待实现」——现阶段用户问账户问题，客服应引导转人工或说明「该功能即将上线」。

3. 细分流程（按意图类型）

3.0 意图体系总览（5 类）

本方案将 5 类「技术意图」按业务场景细分为 6 条处理路径（§3.1-§3.6），便于评审时逐条核对；faq/product_inquiry/policy_explain 是检索主路径，chitchat/transfer_human 是控制路径。

3.1 账户与交易查询（持仓 / 收益 / 交易记录）

⚠️ 实施状态：本场景属「操作类」查询，依赖客户端 + 账户系统产生真实账户数据，当前为 Phase 2 待实现（客户端尚未完成）。现阶段用户问账户问题，客服应引导转人工或说明「该功能即将上线」。详见 §2.4.6。

参考做法：天天基金智能客服对「什么时候能看到基金份额」这类账务问题，直接给出「T+1 工作日确认，[我的]→持仓查看，未确认的在[在途交易]查看」的标准流程答案〔S3〕。富途帮助中心以「我的 > 我的客服 > 全部问题搜索 → 仍未解决再转电话/在线」的三步式自助+兜底〔S8〕。

3.2 产品信息咨询（一般性介绍）

参考做法：招商基金智能客服对「基金下跌怎么办」分析净值下跌 4 个原因（市场变化、分红、折算、转型）并提示异常可查公告〔S6〕——只解释现象、不给结论，与本项目「只答不荐」一致。华夏基金「机智小牛智能体」主打「会倾听、能共情、懂陪伴」，同样止步于陪伴与解答〔S7〕。

3.3 费率与收益计算

参考做法：百度智能云证券基金方案内置「申购费/赎回费/管理费三维费率计算器，支持前端/后端收费切换」〔S15〕。华云天下铁律一给出的标准应答为「根据产品说明书，该产品的业绩比较基准为 XX%。过往业绩不代表未来表现，投资有风险，请您仔细阅读产品说明书」〔S13〕——本项目固定免责话术与之同构。

3.4 赎回与到账时效

参考做法：天天基金「基金吧答疑」对「基金赎回多久到账」给出分类型标准答案：货币 1-2 工作日、债券 2-4、混合/股票 2-4、QDII 4-13 工作日（具体以产品为准）〔S3〕。这正是本项目 fin_faq_collection 标准答案的组织范式——高频规则类问题固化为标准话术直返。

3.5 投诉与情绪安抚

参考做法：捷通华声明确「涉及投诉/纠纷关键词」纳入转人工触发条件，「宁可早转，不可硬撑」〔S14〕。国标《顾客联络服务 人工与智能客户服务协同要求》明确「感知到客户负面情绪…系统应自动转接人工」〔S11〕。

3.6 风险揭示与合规禁答

参考做法：华云天下铁律四标准应答「理财产品是非保本浮动收益产品，您的本金和收益可能会因市场波动而产生损失」〔S13〕。蚂蚁财富《智能理财助理服务协议》明确「输出可能出现错误或遗漏…不构成投资建议/推介/要约」〔S1〕——本项目的免责声明与「AI 生成内容风险提示」双保险与此对齐。

3.7 各意图数据流向图（汇总）

下图把 §3.1-§3.6 的六类业务场景，映射回 §3.0 的五类技术意图，展示不同意图下数据从「提问」到「结果」的完整流转路径。账户/持仓/交易查询（操作类）当前为 Phase 2 待实现（见 §2.4.6），图中以虚线标出。

flowchart TD
    U[用户提问] --> A[① 输入接收与校验鉴权]
    A --> B[② 记忆召回：读 Redis 会话]
    B --> INT{③ 意图路由
qwen-turbo · 阈值0.6}

    INT -- faq 高频问答 --> F1[检索 fin_faq_collection · TopK=3]
    F1 --> F2[直返标准答案]

    INT -- product_inquiry 产品咨询 --> P1[检索 fin_product_collection · TopK=5]
    P1 --> P2[RAG + LLM 生成 + 来源引用]

    INT -- policy_explain 政策解读 --> PL1[检索 fin_policy_collection · TopK=5]
    PL1 --> PL2[RAG + 解释]

    INT -- chitchat 闲聊 --> C1[不查集合
LLM 引导业务话题]

    INT -- transfer_human 转人工 --> T1[生成转接工单
携带完整上下文]
    T1 --> T2[运营/投顾后台接管]

    INT -- 置信度<0.6 --> L1[澄清提问 → 仍低则转人工]

    O1[账户/持仓/交易查询
Phase 2 · API 直连账户系统] -.-> G

    F2 --> G[⑤ 合规校验
负面词 + 免责声明]
    P2 --> G
    PL2 --> G
    C1 --> G
    T2 --> G
    L1 --> G

    G --> H[⑥ 数据沉淀：归档 + 脱敏]
    H --> I[⑦ 事件广播：敏感意图 → 风控]
    H --> J[结果反馈 SSE]

各意图数据路径说明：

4. 知识库清单

4.1 三集合总览

4.2 字段结构（建议统一 schema）

字段设计依据：阿里云开发者社区强调「知识条目需标注生效日期与适用范围，费率/期限/政策调整时旧条目及时下线，避免过期信息被检索命中」〔S16〕；富途微藤 AI 提供「知识库健康度检查」能力，识别并更正知识库瑕疵〔S9〕。

4.3 更新频率与数据来源

参考做法：招商基金智能在线客服「集成招商银行小招X引擎及知识库」，并将「文字聊天记录生成工单内容总结、自动匹配受理产品和渠道」，实现知识沉淀的自动化〔S5〕。富途微藤 AI 的「语料智能扩写」可在冷启动阶段快速丰富语料〔S9〕。

4.4 知识运营机制

• 设立专职知识运营岗位，明确更新责任与频率——「上线后无人维护」是金融产品知识库快速失效的头号原因〔S16〕。
• 同义问法补充：从历史工单提取真实表达（用户说「到期能拿回多少钱」而非「现金价值」），作为同义问法补充进知识条目与意图示例〔S16〕。
• 按条款完整性切分：把「同一条款的适用条件与例外情况」放在同一片段，避免拆散导致回答不准〔S16〕。

4.5 FAQ 示例（fin_faq_collection 内容样例）

下表给出 fin_faq_collection 的典型问答对示例，覆盖不同意图类型。实际入库时按 §4.2 字段结构存储（question / answer / tags / effective_date / expire_date / embedding）。

说明：「是否可直返」= 是否走「检索直返标准答案」路径（faq 意图）。涉及投诉/转人工的，直接走 transfer_human 意图，不检索 FAQ。

5. 人工审核机制

5.1 转人工触发条件（分级）

触发条件整合了项目既有设计（全景说明书场景 2 的四条件）与国标/行业要求：国标明确「交互失败达到设定阈值、感知负面情绪、涉信息安全人身安全紧急场景时自动转接人工」〔S11〕；捷通华声「连续 2 轮未理解意图、涉及投诉/纠纷关键词、客户主动要求人工」〔S14〕。

5.2 事后复核与质检（兜底保障）

即使 AI 未触发转人工，仍设事后质检：

参考：金融 AI 客服「双轨评测」——一条测业务准确率、一条测合规通过率，任何一条不通过不发布〔S14〕；监管要求客户沟通记录保存 ≥ 20 年且不得含诱导性表述，传统录音抽检覆盖率不足 5%，AI 全量留痕可支撑 100% 质检〔S15〕。

5.3 升级路径

一线客服（运营/客户经理接管）
   ↓ 无法解决或涉合规
高级坐席（持证投顾/资深客服）
   ↓ 涉监管敏感、法律、重大投诉
合规/法务（合规官）

参考做法：招商基金「金谘机构」投顾服务强调「从『顾』到『投』再到『顾』的闭环」，高净值客户「AI 初筛 + 投顾人工审核」正是分级升级的实践〔S4〕。富途按「在线客服 7×24 / 电话人工交易日 24h / 紧急交易柜台」分层〔S8〕。

5.4 国标对齐检查（《顾客联络服务 人工与智能客户服务协同要求》，2026-09-01 实施）

6. 回复策略

6.1 标准回复模板（分场景）

场景 A：产品信息 / 风险等级客观转述

「XX 基金为【股票型/混合型/债券型】基金，风险等级为 R【1-5】（【低/中低/中/中高/高】风险），适合【风险等级】及以上投资者。您可以在完成风险测评后查看是否匹配。」〔S13 铁律三模板〕

场景 B：费率 / 业绩比较基准

「根据产品说明书，该产品的【申购费率/赎回费率/管理费率】为 X%。该产品的业绩比较基准为 X%，过往业绩不代表未来表现，投资有风险，请您仔细阅读产品说明书。」〔S13 铁律一模板〕

场景 C：风险揭示（非保本）

「理财产品/基金是非保本浮动收益产品，您的本金和收益可能会因市场波动而产生损失，请在充分了解风险后做出决策。」〔S13 铁律四模板〕

场景 D：到账时效

「【货币 1-2 / 债券 2-4 / 混合·股票 2-4 / QDII 4-13】个工作日到账，具体以产品为准。」〔S3〕

场景 E：情绪安抚 + 投诉受理

「非常理解您的心情，很抱歉给您带来不好的体验。我已为您登记并转接人工专员跟进处理，稍后会有专人联系您。」

6.2 语气与措辞规范

6.3 必备风险提示与免责声明

固定免责声明（基类强制注入，子类不得绕过）：

「本内容仅为投资分析参考，不构成任何直接投资建议，不构成对任何产品的收益承诺，据此操作风险自负，请谨慎对待。」

AI 生成标识（建议补充，见 §0.4 评审点 4）：对 AI 生成的开放式回答，前置「本回答由 AI 生成，仅供参考」，对齐国标「保留显著 AI 生成标识和风险提示」〔S11〕。

6.4 拒答与兜底话术

6.5 负面词清单（7 个，零容忍）

评审建议：可将「零风险」「稳赚不赔」「躺着赚」「坐享收益」等变体纳入模糊匹配规则（见 §0.4 评审点 4）。

7. 数据与状态流转 / 接口与异常

7.1 接口约定

// POST /api/chat/customer   请求
{ "session_id": "...", "user_id": 10001, "message": "..." }
// 响应（统一信封）
{ "code": 200, "message": "success", "trace_id": "uuid",
  "data": { "reply": "...", "source_references": [...], "intent": "faq",
            "confidence": 0.86, "tool_calls": [...], "suggestions": [...] } }

7.2 会话状态机

7.3 异常处理表

8. 度量与验收（双轨评测）

参考：捷通华声「上线前必须跑满合规用例，任何一条不通过都不发布」〔S14〕；百度智能云分层目标「试点期意图识别准确率 ≥85%、推广期多轮完成率 ≥70%、成熟期 NPS ≥40」〔S15〕。

9. 完善方向与最终交付要求

本章按六大维度逐项指出当前差距 → 完善标准（可量化）→ 交付要求，作为评审后的迭代排期依据。每条完善标准都对应可验收的量化门槛；文末给出上线门禁（Go/No-Go）清单作为最终交付要求。

9.0 差距总览（一页速览）

9.1 功能完整性

9.2 用户体验

9.3 性能优化

9.4 错误处理

9.5 安全性

9.6 代码可读性

9.7 最终交付要求（上线门禁 Go/No-Go 清单）

以下为必须全部满足才可进入联调/上线的硬性门禁；任一未达标即 No-Go。

功能门禁（Must）

安全与合规门禁（Must）

质量门禁（Must）

交付物清单（最终交付要求）

10. 附录

10.1 参考来源清单

10.2 需核实项清单