Python 实战：使用 Function Calling + RAG + SQLite 打造可验证的电商智能客服

一、项目背景

传统的大模型客服通常直接把用户问题交给模型，然后返回模型生成的文本。这种方式实现简单，但存在几个明显问题：

1. 商品价格、库存、品牌等数据可能被模型编造。
2. 促销规则涉及数量、折扣、满减和赠品，单靠模型心算容易出错。
3. 通过正则解析 Action:、Observation: 的 ReAct 实现不稳定。
4. 模型可能把用户输入的错误数字当成真实商品数据。
5. 用户的上下文追问，例如“那买两个呢”，容易丢失商品信息。

本文以一个运动用品电商客服项目为例，将内部逻辑改造成：

用户问题
   ↓
Function Calling
   ↓
RAG 检索商品和促销证据
   ↓
SQLite / 业务规则验证
   ↓
服务端校验引用和数字事实
   ↓
最终中文回答

项目代码中的核心文件如下：

.
├── main.py                         # CLI 启动入口
├── function_calling_agent.py       # Function Calling 代理
├── product_rag.py                  # SQLite + 促销文档检索和校验
├── op_llm_client.py                # Ollama /api/chat 客户端
├── SportsEquipment.db              # 商品数据库
├── store_promotions.txt            # 促销文档
└── test_function_calling.py        # 自动化测试

二、为什么不用字符串解析 ReAct

旧实现要求模型输出如下文本：

Thought: 我需要查询足球
Action: query_by_product_name: 足球

程序再使用正则表达式解析：

action_re = re.compile(r"^Action: (\w+): (.*)$")

这种方式有几个缺点：

• 模型可能输出多余文字，导致正则匹配失败。
• 参数是字符串，需要自己处理逗号、冒号和 JSON 转义。
• 工具名称拼写错误只能在运行时发现。
• 工具返回结果没有标准的 tool_call_id 关联。
• 不能很好地利用模型 API 原生的工具调用能力。

Function Calling 的核心思想是：由程序声明工具 schema，模型返回结构化的函数名和 JSON 参数，程序执行函数后，再把结果以 tool message 传回模型。

三、声明工具 Schema

在 function_calling_agent.py 中声明三个函数：

TOOL_SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "retrieve_product_knowledge",
            "description": "检索 SQLite 商品数据和促销文档证据",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "top_k": {
                        "type": "integer",
                        "minimum": 1,
                        "maximum": 20,
                        "default": 12,
                    },
                },
                "required": ["query"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "calculate_verified_total",
            "description": "使用 SQL 价格、库存和已验证优惠计算订单总价",
            "parameters": {
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "quantity": {"type": "integer", "minimum": 1},
                    "discount_rate": {"type": "number", "minimum": 0.01},
                },
                "required": ["product_id", "quantity"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "submit_grounded_answer",
            "description": "提交带 evidence_id 引用的最终中文回答",
            "parameters": {
                "type": "object",
                "properties": {
                    "answer": {"type": "string"},
                    "citation_ids": {
                        "type": "array",
                        "items": {"type": "string"},
                    },
                    "status": {
                        "type": "string",
                        "enum": ["grounded", "insufficient_evidence"],
                    },
                },
                "required": ["answer", "citation_ids", "status"],
                "additionalProperties": False,
            },
        },
    },
]

三个工具分别负责：

• retrieve_product_knowledge：只负责检索事实。
• calculate_verified_total：只负责读取 SQL 价格和库存并计算。
• submit_grounded_answer：只负责提交最终答案，不能直接绕过服务端校验。

这里把“最终回答”也设计成了一个函数调用，而不是直接信任模型的普通文本。这是防止模型绕过引用校验的关键。

四、Function Calling 主循环

代理的核心逻辑位于 GroundedFunctionCallingAgent.ask()。

messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    *self.conversation,
    {"role": "user", "content": user_message},
]

for iteration in range(self._max_iterations()):
    completion = self._complete(
        messages,
        force_retrieval=iteration == 0,
    )
    message = completion.choices[0].message
    tool_calls = message.tool_calls or []

    if not tool_calls:
        return "抱歉，我没有拿到可验证的商品数据。"

    messages.append({
        "role": "assistant",
        "content": message.content,
        "tool_calls": serialize_tool_calls(tool_calls),
    })

    for call in tool_calls:
        name = call.function.name
        arguments = json.loads(call.function.arguments)

        if name == "submit_grounded_answer":
            return self._submit(arguments)

        result = tools[name](**arguments)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result, ensure_ascii=False),
        })

实际项目进一步对 tool_choice 做了状态控制：

1. 第一轮强制调用 retrieve_product_knowledge。
2. 如果问题需要金额计算，后端优先准备经过验证的计算结果。
3. 检索或计算完成后，强制调用 submit_grounded_answer。
4. 校验失败时，把错误作为 tool result 反馈给模型，让模型重新提交。

这样可以避免模型在“检索”和“计算”之间无意义循环。

五、RAG 检索层设计

5.1 商品表结构

项目的 SportsEquipment.db 中有 products 表：

CREATE TABLE products (
    product_id TEXT,
    product_name TEXT,
    description TEXT,
    specifications TEXT,
    usage TEXT,
    brand TEXT,
    price REAL,
    stock_quantity INTEGER
);

5.2 参数化 SQL 查询

product_rag.py 中优先做商品名精确匹配，再做字段模糊匹配：

with closing(sqlite3.connect(self.db_path)) as connection:
    connection.row_factory = sqlite3.Row
    rows = connection.execute(
        """
        SELECT product_id, product_name, description,
               specifications, usage, brand, price, stock_quantity
        FROM products
        WHERE product_name LIKE ? COLLATE NOCASE
           OR description LIKE ? COLLATE NOCASE
           OR specifications LIKE ? COLLATE NOCASE
           OR usage LIKE ? COLLATE NOCASE
           OR brand LIKE ? COLLATE NOCASE
        """,
        (like, like, like, like, like),
    ).fetchall()

这里必须使用参数绑定：

connection.execute(sql, (like, like, like, like, like))

不要把用户输入直接拼接到 SQL 字符串中，否则会带来 SQL 注入风险。

5.3 促销文档检索

促销文档示例：

1. 足球 - 购买足球即可享受9折优惠。
2. 羽毛球拍 - 任意购买羽毛球拍两支以上，享8折优惠。
3. 篮球 - 单笔订单满300元，篮球半价。

对于商品查询，程序根据检索到的商品名匹配促销行。对于“有优惠吗”这类没有商品名的问题，则检索促销目录，而不是只返回文档免责声明。

这解决了两种场景：

足球有什么优惠？       -> 返回足球促销 evidence
有优惠吗？              -> 返回促销目录 evidence

5.4 Evidence 数据结构

每条事实都会包装成 Evidence：

@dataclass(frozen=True)
class Evidence:
    evidence_id: str
    source: str
    text: str
    product_id: str | None = None

示例结果：

{
  "evidence_id": "product:001",
  "source": "sqlite.products",
  "product_id": "001",
  "text": "商品ID=001; 商品名=足球; 品牌=耐克; 价格=120.0元; 库存=50"
}

促销证据示例：

{
  "evidence_id": "promotion:001:2",
  "source": "store_promotions.txt:2",
  "product_id": "001",
  "text": "1. 足球 - 购买足球即可享受9折优惠。"
}

evidence_id 的作用是建立完整的事实链：模型回答中引用什么，服务端就验证什么。

六、SQL 验证和订单计算

模型不能直接决定最终价格。calculate_verified_total() 会重新从 SQL 查询商品价格和库存：

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

if row is None:
    return {"ok": False, "error": "product_id not found in SQL database"}

if quantity > int(row["stock_quantity"]):
    return {"ok": False, "error": "requested quantity exceeds verified stock"}

基础价格计算：

base_total = round(float(row["price"]) * quantity, 2)
total = base_total

if discount_rate is not None:
    total = round(base_total * discount_rate, 2)

折扣率不能由模型随意传入。程序会检查这个折扣率是否出现在促销 evidence 中：

allowed_rates = self._discount_rates(promotion_text)
if discount_rate not in allowed_rates:
    return {
        "ok": False,
        "error": "discount rate is not present in the verified promotion document",
    }

中文折扣需要注意：

9折  -> 0.9
8折  -> 0.8
95折 -> 0.95
半价 -> 0.5

不能简单地把所有数字除以 10，否则 95折 会被错误解析成 9.5。

七、如何防止模型幻觉

7.1 引用必须真实存在

if any(citation_id not in evidence for citation_id in citation_ids):
    return False, "answer cites evidence that was not retrieved"

模型提交一个不存在的 evidence_id 时，服务端直接拒绝。

7.2 数字必须能在证据中找到

source_text = " ".join(
    evidence[citation_id].text
    for citation_id in citation_ids
)
source_numbers = _numeric_values(source_text)

for match in re.finditer(r"\d+(?:\.\d+)?", answer):
    number = Decimal(match.group())
    if number not in source_numbers:
        return False, f"numeric claim {number} is not present in cited evidence"

例如：

数据库证据：足球价格为120元
模型回答：足球价格为999元

校验器会拒绝 999。

7.3 支持等价折扣表达

模型可能把 9折 改写成 90%。系统不会简单地把 90 放行，而是只有当引用证据中存在 9折 时，才接受 90% 这种等价表达。

7.4 自动补充唯一证据

如果模型已经引用商品和促销证据，但忘记引用唯一的计算 evidence，例如回答中出现 108.0，程序可以自动补充唯一匹配的 sql.calculation 证据。

这个自动补充是有边界的：

• 只补充唯一匹配的证据。
• 普通数字不能任意补充。
• 促销数字必须匹配明确的 N折 或百分比。

7.5 用户输入的错误价格不能污染答案

当用户问：

足球售价999元吗？

系统不会把 999 当成数据库事实，而是基于已检索的商品证据返回：

不是。根据已验证数据，足球售价为120.0元。

7.6 商品数据只读

当前项目没有提供修改商品价格、库存的工具。因此对以下请求直接拒绝：

把足球价格改成1元

商品数据是只读的，我不能修改价格、库存或促销信息。

这比让模型“假装修改成功”更安全。

八、上下文追问处理

客服场景中经常出现：

足球多少钱？
有优惠吗？
那买两个呢？

代理只保存精简的用户和最终客服回答：

self.conversation.extend([
    {"role": "user", "content": user_message},
    {"role": "assistant", "content": answer},
])
self.conversation = self.conversation[-8:]

工具消息和 evidence 不跨请求复用，避免旧数据污染新请求；但商品上下文会保留，以支持“那买两个呢”这类追问。

对于明确追问，程序会把当前商品名补回检索 query：

模型 query：那买两个呢
实际检索：足球 那买两个呢

如果用户明确询问未知商品，例如“店里有网球吗”，则不会强行继承足球上下文。

九、配置和运行

9.1 安装依赖

pip install -r requirements.txt

9.2 配置 API Key

在 .env 文件中设置：

API_KEY=your_api_key

项目默认使用 OpenAI-compatible API：

{
  "openai": {
    "use_model": true,
    "model_name": "qwen-plus",
    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
    "temperature": 0.2,
    "max_iterations": 8
  }
}

启动：

python main.py

Windows 虚拟环境启动方式：

& F:\liting\lyl\vs\damoxing\8.3.2\.venv\Scripts\python.exe `
  F:\liting\lyl\vs\damoxing\8.3.2\main.py

9.3 使用 Ollama

关闭 openai.use_model，开启 ollama.use_model：

{
  "ollama": {
    "use_model": true,
    "model_name": "llama3.1:8b",
    "temperature": 0.2,
    "max_iterations": 8
  }
}

项目通过 /api/chat 发送 messages 和 tools。Ollama 模型及版本需要支持工具调用能力。

十、测试设计

项目使用标准库 unittest，测试不调用真实外部模型，而是使用 fake client 和临时 SQLite 数据库。

运行测试：

python -m unittest test_function_calling.py -v

测试覆盖：

商品检索
SQL 价格和库存验证
9折、95折、半价解析
订单数量约束
羽毛球拍两支 480 元
缺失计算引用自动恢复
缺失促销引用自动恢复
上下文追问
通用促销目录
未知商品拒答
999 元价格纠正
只读修改保护

一个典型测试：

def test_retrieval_prepares_verified_badminton_total(self):
    agent = GroundedFunctionCallingAgent(
        FakeClient(),
        {"model_name": "fake", "max_iterations": 4},
        self.kb,
    )
    agent.expected_quantity = 2
    agent.needs_calculation = True

    result = agent._retrieve("羽毛球拍买两支，优惠后多少钱？")

    self.assertEqual(result["verified_calculation"]["total"], 480.0)

测试的重点不是验证模型“说得像不像”，而是验证：

1. 数据是否来自数据库。
2. 促销是否来自文档。
3. 计算结果是否正确。
4. 错误数字是否被拦截。
5. 无证据时是否拒答。

十一、常见问题和改进方向

问题一：为什么没有使用向量数据库？

当前项目数据量较小，商品表只有几十条记录，促销文档也是短文本。使用参数化 SQLite 查询和关键词检索已经足够，并且更容易验证价格、库存等结构化事实。

生产环境可以增加：

• Embedding 模型
• Milvus、FAISS 或 pgvector
• 文档分块和向量检索
• 关键词检索与向量检索的混合排序

但无论是否使用向量数据库，最终价格和库存仍然应该回到业务数据库验证。

问题二：为什么不完全相信 RAG 返回的文本？

RAG 只能说明“检索到了某段文本”，不能自动保证交易计算正确。因此本项目采用双重验证：

非结构化促销文档 -> 检索优惠规则
结构化 SQL        -> 验证价格和库存
服务端校验        -> 验证最终答案数字和引用

问题三：数字校验是不是等于完全没有幻觉？

不是。当前方案重点防止商品价格、库存、折扣、总价等核心数字幻觉，并要求回答引用 evidence。对于复杂自然语言中的因果关系、主观推荐和长文本事实，还可以增加：

• 结构化答案 schema
• 事实级 claim extraction
• 第二个验证模型
• 规则引擎
• 人工审核

问题四：生产环境还需要哪些能力？

建议继续增加：

• 用户身份和权限控制
• 工具调用审计日志
• API 超时和重试策略
• 促销规则结构化存储
• 单元测试、集成测试和回归测试
• 商品价格变更版本号
• 证据有效期和缓存失效策略
• 订单创建与库存扣减事务

十二、总结

这个项目的核心不是简单地把大模型接入客服，而是建立一条可验证的数据链：

Function Calling 负责调用正确工具
RAG 负责找到相关证据
SQLite 负责验证结构化商品事实
业务规则负责验证折扣和订单计算
evidence_id 负责建立引用关系
服务端校验负责拦截最终幻觉

最终效果是：

• “足球多少钱？”可以基于 SQL 返回真实价格。
• “羽毛球拍买两支优惠后多少钱？”可以验证为 480 元。
• “足球售价999元吗？”不会把 999 当成真实价格。
• “把足球价格改成1元”会被只读保护拦截。
• “那买两个呢？”可以根据上下文继续回答。

对于商品客服、订单助手、售后问答、企业知识库问答等场景，这种“模型负责理解，工具负责取数，服务端负责验证”的架构，比单纯依赖 Prompt 更适合落地。