前言

电商客服最危险的错误不是回答不够流畅,而是编造价格、库存、品牌和优惠。即使已经接入RAG,模型仍可能把相似商品当成目标商品、修改检索结果中的数字,或者在没有数据时根据常识补充一个看似合理的答案。

本文基于Function Calling+RAG电商客服项目,分析一套程序级证据链:RAG只召回候选商品ID,SQLite按ID验证商品与促销,Decimal根据数据库规则计算价格,EvidenceLedger记录允许进入回答的事实,validate_answer()在输出前检查商品ID和人民币金额。

这套方案的目标是降低幻觉和拦截无证据回答,并不意味着能够100%消除所有模型错误。文章将同时说明已实现能力、测试覆盖和当前边界。



项目防幻觉效果与结构

在这里插入图片描述

电商客服四层防幻觉流程。候选召回、SQL复核、确定性计算和答案校验依次执行。

在这里插入图片描述

EvidenceLedger证据账本。最终回答只能引用工具记录过的商品ID和金额。

在这里插入图片描述

确定性价格计算流程。模型不能把自己生成的单价作为可信输入。

在这里插入图片描述

最终回答校验分支。证据不足时返回安全兜底回答。

一、电商客服常见的模型幻觉

编造不存在的商品

数据库没有无人机,模型却回答:

本店有登山无人机,目前售价999元,库存20台。

修改真实价格

数据库中的足球价格是120元,模型可能输出:

足球现在售价99元。

忽略优惠条件

羽毛球拍需要购买两支以上才享受8折,模型却把购买一支也算成240元。

混淆赠品和降价

瑜伽垫优惠是赠送教程视频,不代表商品价格减少50元。

这些错误不能只依靠Prompt中的“不要编造”解决。Prompt是行为引导,不是程序级验证。

二、为什么RAG仍然可能幻觉

RAG改善的是信息获取,不保证最终回答必然正确。

可能出现:

  • 向量检索召回相似但错误的商品。
  • 向量索引未及时更新。
  • 模型忽略检索结果中的条件。
  • 模型在组织语言时修改数字。
  • 没有检索结果时仍根据常识回答。

因此项目采用:

RAG负责“可能是谁”
SQLite负责“事实是什么”
Decimal负责“价格怎么算”
Validator负责“答案能否放行”

三、防幻觉相关目录

function_calling_rag_agent/
│
├── database.py
├── agent.py
│
├── tools/
│   ├── product_tools.py
│   ├── price_tools.py
│   └── registry.py
│
├── guardrails/
│   ├── evidence.py
│   └── validator.py
│
├── scripts/
│   └── initialize_database.py
│
└── tests/
    └── test_tools_and_guardrails.py
文件职责
database.py统一SQLite连接和行转换
product_tools.py商品、促销权威查询
price_tools.py重新查询数据并计算价格
evidence.py记录已验证ID和数字
validator.py校验最终回答
test_tools_and_guardrails.py验证正常与虚假场景

四、SQLite数据库结构

商品表:

CREATE TABLE products (
    product_id TEXT PRIMARY KEY,
    product_name TEXT NOT NULL,
    description TEXT NOT NULL,
    specifications TEXT NOT NULL,
    usage TEXT NOT NULL,
    brand TEXT NOT NULL,
    price REAL NOT NULL CHECK(price >= 0),
    stock_quantity INTEGER NOT NULL CHECK(stock_quantity >= 0)
);

促销表:

CREATE TABLE promotions (
    promotion_id TEXT PRIMARY KEY,
    product_id TEXT NOT NULL REFERENCES products(product_id),
    promotion_text TEXT NOT NULL,
    discount_type TEXT NOT NULL,
    discount_value REAL,
    minimum_quantity INTEGER NOT NULL DEFAULT 1,
    minimum_spend REAL,
    extra_condition TEXT
);

product_id是整条证据链的连接点:

RAG候选商品ID
→ products商品事实
→ promotions优惠规则
→ 确定性价格计算
→ 最终回答引用

五、数据库连接封装

import sqlite3
from contextlib import contextmanager

from config import DATABASE_PATH


@contextmanager
def connect():
    connection = sqlite3.connect(DATABASE_PATH)
    connection.row_factory = sqlite3.Row
    try:
        yield connection
    finally:
        connection.close()


def rows_to_dicts(rows):
    return [dict(row) for row in rows]

row_factory=sqlite3.Row让查询结果既可以按位置访问,也能转换为带字段名的字典。contextmanager确保工具执行完成或抛出异常后连接都能关闭。

调试时观察:

DATABASE_PATH
connection.row_factory
dict(row)

六、RAG结果为什么不是权威事实

search_products()返回:

{
    "status": "candidates_found",
    "source": "hybrid_rag_index",
    "warning": (
        "候选结果不是权威事实,"
        "必须用get_product_details进行SQL验证。"
    ),
    "candidates": [
        {
            "product_id": "001",
            "product_name": "足球",
            "retrieval_score": 0.91,
            "is_verified_fact": False
        }
    ]
}

向量相似度只能说明文本相关,不能证明价格、库存和品牌当前有效。下一步必须调用get_product_details()

七、商品SQL验证工具

def get_product_details(product_ids: list[str]) -> dict:
    unique_ids = list(
        dict.fromkeys(str(item) for item in product_ids)
    )[:10]

    placeholders = ",".join("?" for _ in unique_ids)

    with connect() as connection:
        rows = connection.execute(
            f"""
            SELECT * FROM products
            WHERE product_id IN ({placeholders})
            ORDER BY product_id
            """,
            unique_ids,
        ).fetchall()

    products = rows_to_dicts(rows)

    return {
        "status": "verified" if products else "not_found",
        "source": "sqlite.products",
        "products": products,
        "missing_product_ids": sorted(
            set(unique_ids)
            - {item["product_id"] for item in products}
        ),
    }

输入限制

  • 商品ID统一转换成字符串。
  • 使用dict.fromkeys()去重并保留顺序。
  • 最多处理10个商品ID。
  • SQL值使用?占位符。

返回状态

查到商品:

status = verified
source = sqlite.products

没有查到:

status = not_found

模型不能把not_found解释成“可能有货”。

八、促销SQL验证工具

def get_promotions(product_ids: list[str]) -> dict:
    unique_ids = list(
        dict.fromkeys(str(item) for item in product_ids)
    )[:10]
    placeholders = ",".join("?" for _ in unique_ids)

    with connect() as connection:
        rows = connection.execute(
            f"""
            SELECT * FROM promotions
            WHERE product_id IN ({placeholders})
            ORDER BY promotion_id
            """,
            unique_ids,
        ).fetchall()

    promotions = rows_to_dicts(rows)
    return {
        "status": "verified" if promotions else "not_found",
        "source": "sqlite.promotions",
        "promotions": promotions,
    }

促销不再通过商品名称读取文本,而是按商品ID关联,避免“足球”同时出现在其他商品描述中造成误匹配。

九、促销类型和条件

项目包含:

类型示例计算方式
percentage足球9折小计乘0.9
fixed_reduction满500减100小计减100
buy_x_get_y买三送一根据数量计算收费件数
gift赠送教程或会员不降低商品价格
free_shipping满200包邮不修改商品小计

条件字段包括:

  • minimum_quantity
  • minimum_spend
  • extra_condition

模型必须明确说明条件是否满足,不能只看到“8折”就直接计算。

十、为什么价格工具不接收单价

Function Calling Schema只让模型传:

{
  "product_id": "001",
  "quantity": 1,
  "is_first_purchase": false
}

模型不能传入:

{"price": 99}

价格工具根据product_id重新查询SQLite:

product = connection.execute(
    """
    SELECT product_id, product_name, price, stock_quantity
    FROM products
    WHERE product_id = ?
    """,
    (str(product_id),),
).fetchone()

因此模型自己编造的单价不会进入计算过程。

十一、Decimal确定性价格计算

unit_price = Decimal(str(product["price"]))
subtotal = unit_price * quantity
final_total = subtotal

百分比折扣:

if condition_met and discount_type == "percentage":
    final_total = subtotal * discount_value

满减:

elif condition_met and discount_type == "fixed_reduction":
    final_total = max(
        Decimal("0"),
        subtotal - discount_value
    )

金额统一保留两位小数:

def _money(value: Decimal) -> float:
    return float(
        value.quantize(
            Decimal("0.01"),
            rounding=ROUND_HALF_UP
        )
    )

相比旧版eval(),这套实现只执行明确允许的优惠规则,不会执行任意Python表达式。

十二、价格工具返回证据

return {
    "status": "verified",
    "source": [
        "sqlite.products",
        "sqlite.promotions",
        "deterministic_decimal_calculation"
    ],
    "product_id": product["product_id"],
    "product_name": product["product_name"],
    "quantity": quantity,
    "unit_price": _money(unit_price),
    "subtotal": _money(subtotal),
    "final_total": _money(final_total),
    "promotion_applied": applied,
    "promotion_note": note,
}

它不只返回108,还说明108来自哪个商品、多少件、什么优惠和哪些数据源。

十三、EvidenceLedger证据账本

class EvidenceLedger:
    def __init__(self) -> None:
        self.verified_product_ids: set[str] = set()
        self.allowed_numbers: set[str] = set()
        self.has_verified_product_data = False

三个核心状态:

属性含义
verified_product_idsSQL或价格工具确认过的商品ID
allowed_numbers工具结果中出现的合法数字
has_verified_product_data是否获得过权威商品记录

记录工具结果:

def record(self, tool_name: str, result: dict) -> None:
    if result.get("status") != "verified":
        return

    if tool_name == "get_product_details":
        products = result.get("products", [])
        self.has_verified_product_data = bool(products)
        for product in products:
            self.verified_product_ids.add(
                str(product["product_id"])
            )
            self._record_numbers(product)

    elif tool_name in {
        "get_promotions",
        "calculate_final_price"
    }:
        self._record_numbers(result)
        if result.get("product_id"):
            self.verified_product_ids.add(
                str(result["product_id"])
            )

search_products状态不是verified,因此RAG候选不会直接进入权威证据集合。

十四、数字规范化

回答可能写成:

120元
120.00元
¥120
¥120.0

它们应被视为相同金额。项目使用Decimal.normalize()统一格式:

def normalize_number(value) -> str:
    try:
        number = Decimal(str(value))
    except InvalidOperation:
        return str(value)

    normalized = format(number.normalize(), "f")
    return "0" if normalized in {"-0", ""} else normalized

例如:

120.00 → 120
108.0  → 108

十五、最终答案校验器

安全兜底回答:

SAFE_FALLBACK = (
    "当前商品数据库中没有足够的已验证信息来回答这个问题,"
    "请联系人工客服进一步核实。"
)

校验商品证据:

needs_product_evidence = any(
    word in f"{question}\n{answer}"
    for word in product_fact_words
)

if needs_product_evidence and not evidence.has_verified_product_data:
    return False, SAFE_FALLBACK

校验商品ID:

cited_ids = set(
    re.findall(
        r"商品ID[::]\s*([A-Za-z0-9_-]+)",
        answer
    )
)

if evidence.has_verified_product_data:
    if not cited_ids or not cited_ids.issubset(
        evidence.verified_product_ids
    ):
        return False, SAFE_FALLBACK

校验人民币金额:

money_matches = re.findall(
    r"[¥¥]\s*(\d+(?:\.\d+)?)|(\d+(?:\.\d+)?)\s*元",
    answer
)

money_values = [
    prefix_value or suffix_value
    for prefix_value, suffix_value in money_matches
]

if any(
    normalize_number(value) not in evidence.allowed_numbers
    for value in money_values
):
    return False, SAFE_FALLBACK

模型回答999元¥999.00时,如果999没有出现在证据集合中,回答就不会放行。

十六、Agent如何串联证据与校验

evidence = EvidenceLedger()

for _ in range(SETTINGS.max_iterations):
    completion = self.client.chat.completions.create(...)
    message = completion.choices[0].message

    if message.tool_calls:
        for tool_call in message.tool_calls:
            result = self.registry.execute(
                tool_call.function.name,
                tool_call.function.arguments,
            )
            evidence.record(tool_call.function.name, result)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "name": tool_call.function.name,
                "content": json.dumps(result, ensure_ascii=False),
            })
        continue

    answer = message.content or SAFE_FALLBACK
    _, safe_answer = validate_answer(question, answer, evidence)
    return safe_answer

候选回答生成后不会直接打印,而是先进入validate_answer()

十七、自动化测试覆盖

SQL商品验证:

result = get_product_details(["001"])
self.assertEqual(result["status"], "verified")
self.assertEqual(result["products"][0]["price"], 120.0)

足球九折:

result = calculate_final_price("001", 1)
self.assertEqual(result["final_total"], 108.0)
self.assertTrue(result["promotion_applied"])

首次购买条件:

normal = calculate_final_price(
    "004", 1, is_first_purchase=False
)
first = calculate_final_price(
    "004", 1, is_first_purchase=True
)
self.assertEqual(normal["final_total"], 500.0)
self.assertEqual(first["final_total"], 400.0)

虚假价格拦截:

invalid, fallback = validate_answer(
    "足球多少钱?",
    "足球(商品ID:001)价格999元。",
    evidence
)
self.assertFalse(invalid)
self.assertEqual(fallback, SAFE_FALLBACK)

项目完整测试集当前共9项通过,其中防幻觉测试覆盖无证据商品、虚假数字、人民币符号和优惠条件等场景。

十八、运行与调试

运行测试:

cd "D:\Jupyter_Projects\PythonProject\大模型学习\AI Agent\function_calling_rag_agent"
python -m unittest discover -s tests -v

推荐断点:

  1. get_product_details()执行SQL后。
  2. calculate_final_price()读取商品和促销后。
  3. condition_met计算后。
  4. evidence.record()入口。
  5. validate_answer()提取商品ID后。
  6. 人民币金额正则匹配后。

重点观察:

products
promotion
subtotal
final_total
condition_met
evidence.verified_product_ids
evidence.allowed_numbers
cited_ids
money_values

十九、常见问题与解决方法

1. RAG找到商品但答案被拦截

原因:只调用了search_products,没有调用get_product_details

解决:检查System Prompt和工具消息,确保候选ID继续经过SQL验证。

2. 数据库有商品却返回not_found

原因:商品ID类型、数据库路径或数据初始化存在问题。

解决:检查DATABASE_PATH和实际表内容。

3. 优惠没有应用

原因:数量、金额或首次购买条件不满足。

解决:观察minimum_quantityminimum_spendextra_conditioncondition_met

4. 120元被错误拦截

原因:数字格式没有统一。

解决:所有工具数字和回答数字都通过normalize_number()处理。

5. ¥999没有被识别

原因:金额正则只匹配了“元”后缀。

解决:当前代码同时匹配¥,修改时保留回归测试。

6. 回答没有商品ID

原因:模型没有遵守System Prompt。

解决:校验器会返回兜底回答;同时强化Prompt中的引用格式。

7. promotion状态verified但商品未验证

原因:优惠记录不能替代商品详情验证。

解决:has_verified_product_data只由get_product_details设置。

8. 赠品优惠导致价格错误降低

原因:把赠品价值误当作减价。

解决:gift只设置promotion_applied,不修改final_total

9. 买三送一数量理解不一致

原因:用户所说数量可能是购买数量或最终获得数量。

解决:界面和工具Schema应进一步明确数量语义,这是当前可改进点。

10. SQL占位符数量错误

原因:空商品ID列表导致IN ()

解决:Function Schema要求至少一个ID,程序端仍可增加空列表提前返回。

11. 模型使用“售价一百二十元”

原因:当前校验器主要识别阿拉伯数字金额。

解决:可增加中文数字规范化,这是当前尚未实现的边界。

12. 测试通过是否表示零幻觉

原因:自动化测试只能覆盖已设计场景。

解决:不能声称零幻觉,应持续收集真实失败案例并增加回归测试。

二十、总结

本文分析了电商客服Agent中的程序级防幻觉流程。RAG首先召回候选商品ID,但候选结果不会直接进入权威证据;get_product_detailsget_promotions使用SQLite按商品ID验证事实;calculate_final_price重新读取单价和优惠,并用Decimal执行确定性计算。

EvidenceLedger记录已验证商品ID和工具结果中的合法数字。模型生成答案后,validate_answer()检查商品事实是否有SQL证据、引用ID是否有效,以及“元”“¥”“¥”金额是否出现在证据集合中。校验失败时返回SAFE_FALLBACK

这套机制能够拦截项目测试覆盖的无证据商品和虚假价格,但仍不是万能的事实验证系统。下一篇将对比旧版ReAct与新版Function Calling+RAG,梳理整个项目的设计演进和适用场景。

更多推荐