最近好几个读者跟我聊RAG检索增强生成的时候都提到同一个尴尬场景文档丢进去了问一句“上个月某区域一共成交了多少单平均客单价是多少”大模型回答得倒是又快又自信结果一对数差得离谱。说白了普通RAG擅长“找信息”不擅长“算信息”。它把文档切成片段存进向量库用户提问时检索出最相关的片段再丢给大模型生成答案。这个过程对“某文档里有哪些处理流程”这类问题很管用可一旦涉及到精确的统计、求和、均值、多条件筛选大模型就露馅了——它本质上是一个语言模型不是在跑数据库查询。所以“AI数据库”这东西确实是神仙结合数据库负责精确计算AI负责理解意图和生成回答。这篇文章我会把整个“会计算的RAG系统”从思路到落地完整拆开给你一套可以直接跑通的最小实现再附上我实际调试中踩过的坑和排查套路。不管是刚接触RAG的新手还是已经在做智能问答的老手都应该能从这里拿到点东西。1. 为什么普通RAG一到计算就翻车问题拆解与方案选型1.1 三个典型翻车场景先说一个我印象很深的场景。有次给一个模拟项目做“销售周报问答”文档里明明写了“A区Q3销售额为1200万元B区为850万元”模型被问到“A区和B区哪个高”时回答对了但接着问“两者加起来是多少”就直接答成“2350万元”实际上应该是2050万元。这种问题不是它不会算而是它从文本里抽取数字时可能把“850万”理解成“800万50万”或者干脆自己“脑补”出了一个数字。另一个场景是表格类文档。很多制度文件、产品手册里都夹着大量表格比如“各档位折扣对照表”“故障代码说明表”。常规切分方式会把表格切成碎片表头是一块数据行是另一块向量检索时往往只召回其中几行结果模型拿着残缺数据强行生成答案数字怎么可能对得上。第三个场景是多条件统计。比如“帮我统计最近七天状态为已完成且金额大于5000的订单有多少笔”。这类问题需要精确过滤普通RAG根本不具备执行条件查询的能力。向量检索只会返回“语义上相似”的文本片段至于这些片段是否覆盖了所有满足条件的记录它完全无法保证。我把这些问题总结成一句话大模型是“语文考生”不是“算术考生”。它擅长从文字中概括、归纳、润色但碰到精确计算尤其是涉及多笔记录聚合时它的概率生成机制本身就是致命弱点。不能指望一个靠预测下一个词工作的系统去稳定输出一行正确的求和结果。1.2 “会计算”的两种实现路线要让RAG从“会找”变成“会算”业界主流有两条路线。路线A是工具调用也叫Function Calling。大模型在生成回答前先判断这个问题是否需要精确计算。如果需要它就输出一个结构化的“调用请求”比如指定一个叫“query_orders”的函数的参数系统收到后真正去数据库执行查询和聚合再把查询结果回传给大模型由大模型基于真实数字组织语言。路线B是Text2SQL。直接把数据库表结构、字段说明、枚举值写进系统提示词让大模型根据用户问题生成SQL语句然后交给数据库引擎执行最后把查询结果返回给大模型生成回答。我自己的选型经验是优先做工具调用把Text2SQL作为子能力藏在某个工具函数里。因为工具调用的可控性明显更好——每个工具对应一段固化的执行逻辑你可以在工具内部做各种防护而Text2SQL自由度太高模型一旦生成一个语义正确但逻辑错误的SQL排查起来很痛苦。对比项工具调用Function CallingText2SQL计算可控性计算逻辑固化在函数里更可靠SQL高度灵活容易生成错误语句适用问题统计类、固定口径类问题组合条件查询、探索式分析对模型要求需要模型支持工具调用普通模型可通过提示词生成SQL安全边界白名单函数天然隔离必须做只读限制和超时控制调试体验每一步工具调用都有日志需要看SQL和返回结果链路更长1.3 整体架构谁该做什么事整个“会计算的RAG系统”我习惯分成六层。文档层负责收集和清洗原始资料包括PDF、Word、Markdown、Excel等。索引层负责把文档切块、向量化、写入存储同时会把结构化的记录比如订单表、设备台账导入数据库。检索层负责任务分发是走向量检索还是查结构化数据库或者两者都查。规划层由大模型完成它负责理解用户问题、决定调哪个工具、确定参数。工具层是真正的算力来源包含查询函数、聚合函数、SQL执行模块。生成层最后把工具返回的结果用自然语言包装给用户。这个结构的关键在于大模型始终只做“决策”和“表达”所有数值计算都交给工具层和数据库。把这句话记住后面所有实现都是围绕它展开的。2. 动手前的技术选型与架构准备2.1 数据边界哪些进文档库哪些进数据库很多人拿到项目就急着装向量数据库其实第一步应该先分清数据边界。这个比选什么存储重要得多。我通常把数据分成三类。第一类是“知识型数据”比如操作手册、管理制度、产品说明这类数据以叙述和规范为主适合切分后进向量库第二类是“记录型数据”比如订单明细、工单记录、库存流水这类数据是结构化表格应该直接进关系型数据库第三类是“混合型数据”比如带统计表格的周报既保留叙述部分供检索又把表格数据抽出来单独建表。为什么要分开因为向量检索解决的是“语义相似”而精确计算解决的是“条件命中”。订单表这种东西按行切进向量库检索出来也是零散片段没有任何计算价值。正确做法是让大模型通过工具直接查库文档向量库只负责回答流程类、定义类问题。2.2 存储选型从轻量实现开始的理由存储方案上我强烈建议第一版不要上重组件直接用“Python标准库 一个轻量关系型数据库”把流程跑通。你完全可以自己实现一个简易向量检索文本向量用列表存查询时算余弦相似度数据量几千条以内性能完全可接受。这套东西跑通了再替换成专门的向量数据库也不迟。真正到了生产环境评估向量存储时我重点看四个维度。一是标量过滤能力能不能在检索时同时按“区域”“日期”“文档类型”过滤这个能力比很多花哨功能都重要二是索引类型是HNSW还是IVF不同索引在召回率和内存占用上差别很大三是并发能力团队内部用和对外提供服务是完全不同的量级四是部署复杂度是嵌入式还是要单独起服务直接影响运维成本。2.3 文档切分与Embedding的实操经验文档切分这块我踩过很多坑最简单的经验是不要按固定字符数硬切。按固定长度切块会把一个完整的表格拆得七零八落还会把一段话的逻辑从中间截断。我更推荐按语义边界切优先按标题层级其次是段落最后才是长度上限。块大小方面我常用512到1024个token具体看你的Embedding模型和场景。块太大检索精度下降块太小又会丢失上下文。真正要注意的是表格类内容务必整块保留。我会在预处理阶段检测到表格后先转换成Markdown或JSON格式作为一个独立块整体入库。这样检索时至少能保证表头和数据行在一起不会出现“只找到表头没找到数据”的尴尬情况。Embedding模型的选择标准我一般看四点中文语义效果好不好、向量维度是否适合当前存储、是本地化部署还是走接口调用、更新迭代是否频繁。正式项目里建议选一个在垂直领域表现稳定的模型而不是一味追求参数大。2.4 计算工具的设计原则工具调用是整个系统的核心工具定义得好不好直接决定模型“会不会用”。我总结了四条原则。第一工具名称要直白。函数名最好包含明确的动作和对象比如query_orders、calculate_revenue不要用do_stuff这类模糊名字。第二参数定义要严格。用JSON Schema把每个参数的类型、是否必填、可选项枚举写清楚给描述时把常见同义词也带上比如“日期范围”参数描述里写上“支持本月、上月、最近N天等自然语言”。第三输出要保持结构化。工具返回的不应该是大段文本应该是JSON包含数据和必要的说明字段这样模型解析时不容易出错。第四执行必须可控。所有工具函数内部要有异常捕获返回给模型的错误信息要准确这样模型才能根据错误重新生成参数。3. 核心环节实现手把手搭建一个会计算的RAG系统3.1 数据入库向量库和业务库并行写入下面进入代码环节。我用纯标准库实现配合一个轻量关系型数据库去掉外部依赖方便你完整跑通。先看入库部分的代码。import sqlite3 import json import hashlib conn sqlite3.connect(rag_demo.db) cur conn.cursor() # 创建向量存储表用于存放知识型文档块 cur.execute( CREATE TABLE IF NOT EXISTS doc_chunks ( id INTEGER PRIMARY KEY AUTOINCREMENT, chunk_text TEXT, source TEXT, embedding TEXT ) ) # 创建业务订单表用于精确计算 cur.execute( CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, region TEXT, category TEXT, amount REAL, status TEXT, order_date TEXT ) ) conn.commit()向量存储表里那列embedding直接存JSON序列化后的向量数组。单机演示、几千条数据完全没问题真要上生产再换真正的向量索引。接下来写一个简易Embedding函数。正式场景里这里要替换成真实的语义模型我在这里用一个伪向量只是为了让你在没有外部依赖的情况下把流程完整跑一遍。def embed_text(text): # 正式环境替换为真实的Embedding模型接口 digest hashlib.sha256(text.encode(utf-8)).digest() return [b / 255.0 for b in digest[:32]]入库时文档块向量化后写入doc_chunks结构化订单数据写入orders。def index_document(text, source): chunks split_into_chunks(text) for chunk in chunks: vec embed_text(chunk) cur.execute( INSERT INTO doc_chunks (chunk_text, source, embedding) VALUES (?, ?, ?), (chunk, source, json.dumps(vec)) ) conn.commit() def import_orders(orders_data): for order in orders_data: cur.execute( INSERT INTO orders (region, category, amount, status, order_date) VALUES (?, ?, ?, ?, ?), (order[region], order[category], order[amount], order[status], order[order_date]) ) conn.commit()split_into_chunks函数我就不完整贴了核心逻辑就是按标题和段落边界切分超过预设长度再截断同时保证重叠部分能衔接上下文。注意一个细节表格数据在切分前先转成Markdown作为一个独立块存储单独加一个标记。这样检索时能区分普通文本和表格块。3.2 混合检索向量检索和关键词检索双通道业务中很多数字检索问题向量检索并不擅长。你需要一个混合检索策略向量检索抓“语义相似”关键词检索抓“字面命中”最后按加权分数合并排序。import math def cosine_similarity(vec_a, vec_b): dot sum(x * y for x, y in zip(vec_a, vec_b)) norm_a math.sqrt(sum(x * x for x in vec_a)) norm_b math.sqrt(sum(x * x for x in vec_b)) if norm_a 0 or norm_b 0: return 0 return dot / (norm_a * norm_b) def vector_search(query_text, top_k5): query_vec embed_text(query_text) rows cur.execute(SELECT id, chunk_text, embedding FROM doc_chunks).fetchall() scored [] for row_id, text, embedding_json in rows: vec json.loads(embedding_json) score cosine_similarity(query_vec, vec) scored.append((score, row_id, text)) scored.sort(reverseTrue, keylambda x: x[0]) return [(row_id, text, score) for score, row_id, text in scored[:top_k]] def keyword_search(query_text, top_k5): terms [term for term in query_text.split() if term] rows cur.execute(SELECT id, chunk_text FROM doc_chunks).fetchall() scored [] for row_id, text in rows: score sum(1 for term in terms if term in text) if score 0: scored.append((score, row_id, text)) scored.sort(reverseTrue, keylambda x: x[0]) return [(row_id, text, score) for score, row_id, text in scored[:top_k]] def hybrid_search(query_text, top_k5): docs {} for row_id, text, score in vector_search(query_text, top_ktop_k * 2): docs[row_id] {text: text, vector_score: score, keyword_score: 0} for row_id, text, score in keyword_search(query_text, top_ktop_k * 2): item docs.setdefault(row_id, {text: text, vector_score: 0, keyword_score: 0}) item[keyword_score] score ranked [] for row_id, item in docs.items(): # 分数归一化后加权合并向量和关键词各占一半权重 total item[vector_score] * 0.6 item[keyword_score] * 0.4 item[total_score] total ranked.append((total, row_id, item[text])) ranked.sort(reverseTrue, keylambda x: x[0]) return [(row_id, text, score) for score, row_id, text in ranked[:top_k]]这里的加权比例你可以根据业务调如果主要是流程问答向量权重可以更高如果涉及大量专有名词、编号、型号关键词权重要提升。当前这套没有做分数归一化真实场景建议对两种分数先做min-max归一化再合并避免因为数值范围不同导致某一边主导排序。3.3 计算工具与Function Calling机制接下来是关键部分让大模型学会调用计算工具。先定义工具列表。这里用一个抽象的llm_chat函数来代表大模型接口你接任何支持工具调用的模型时把它替换成对应调用即可。TOOLS [ { type: function, function: { name: search_knowledge, description: 检索知识库文档查找流程说明、定义、制度等非结构化内容, parameters: { type: object, properties: { query: {type: string, description: 用户问题的关键词或原问题} }, required: [query] } } }, { type: function, function: { name: query_orders, description: 查询订单表数据支持按区域、类别、状态、日期范围过滤并计算总额、平均值、数量、最大值、最小值, parameters: { type: object, properties: { region: {type: string, description: 区域名称例如A区、B区不传表示全部}, category: {type: string, description: 商品类别不传表示全部}, status: {type: string, description: 订单状态例如已完成、进行中不传表示全部}, date_from: {type: string, description: 起始日期格式YYYY-MM-DD}, date_to: {type: string, description: 结束日期格式YYYY-MM-DD}, metric: {type: string, enum: [total_amount, avg_amount, count, max_amount, min_amount], description: 计算指标} }, required: [metric] } } } ]工具执行函数负责根据模型返回的调用参数执行真实计算。def execute_tool_call(tool_name, arguments): args json.loads(arguments) if tool_name search_knowledge: result hybrid_search(args.get(query, ), top_k5) return {docs: [{text: text, score: score} for _, text, score in result]} if tool_name query_orders: return run_order_query(args) return {error: f不认识的工具: {tool_name}} def run_order_query(args): conditions [] params [] if args.get(region): conditions.append(region ?) params.append(args[region]) if args.get(category): conditions.append(category ?) params.append(args[category]) if args.get(status): conditions.append(status ?) params.append(args[status]) if args.get(date_from): conditions.append(order_date ?) params.append(args[date_from]) if args.get(date_to): conditions.append(order_date ?) params.append(args[date_to]) where_sql if conditions: where_sql WHERE AND .join(conditions) metric args.get(metric, total_amount) # 这里用SQL参数化查询防止拼接注入 if metric total_amount: sql fSELECT SUM(amount) FROM orders {where_sql} label 总额 elif metric avg_amount: sql fSELECT AVG(amount) FROM orders {where_sql} label 平均客单价 elif metric count: sql fSELECT COUNT(*) FROM orders {where_sql} label 订单数量 elif metric max_amount: sql fSELECT MAX(amount) FROM orders {where_sql} label 最大单笔金额 elif metric min_amount: sql fSELECT MIN(amount) FROM orders {where_sql} label 最小单笔金额 else: return {error: f不支持的指标: {metric}} row cur.execute(sql, params).fetchone() return {label: label, value: row[0]}这里必须强调两点。第一所有参数必须走参数化SQL绝对不能直接把模型生成的字符串拼进SQL否则一旦模型被恶意提示词引导后果很严重。第二生产环境建议单独开一个只读数据库账户给这个工具用彻底杜绝写入可能。3.4 主流程模型决策、工具执行、结果生成最后把整个流程串起来。系统提示词要明确告诉模型统计类问题必须调用工具不要自己估算。SYSTEM_PROMPT 你是一个数据分析助手。你可以使用以下工具获取信息 - search_knowledge检索知识库文档。 - query_orders查询订单数据并执行统计计算。 当用户问题涉及数量、金额、比例、平均值等精确数值时必须先调用query_orders工具。 所有计算以工具返回结果为准严禁自行改数字或估算。 如果工具返回错误请根据错误信息调整输入参数后重试。 def llm_chat(messages, toolsNone): # 此处替换为真实大模型调用 # 入参和返回需要兼容类似 openai 风格的 chat completion 结构 pass def run_rag_agent(question, max_steps3): messages [{role: system, content: SYSTEM_PROMPT}] messages.append({role: user, content: question}) for step in range(max_steps): response llm_chat(messages, toolsTOOLS) if response.get(tool_calls): messages.append(response) for tool_call in response[tool_calls]: tool_result execute_tool_call( tool_call[function][name], tool_call[function][arguments] ) messages.append({ role: tool, tool_call_id: tool_call[id], content: json.dumps(tool_result, ensure_asciiFalse) }) else: return response[content] return 达到最大执行步数无法完成该问题。流程其实就是一个循环模型决定调工具系统执行并返回结果模型再基于结果继续。最多循环三轮已经覆盖绝大多数场景。实际调试中我发现超过三轮还在循环多半是参数生成有问题与其让模型反复试错不如直接把错误信息打上日志去分析。3.5 一个完整问答实例模型是如何“算”出答案的用“上个月A区完成订单的总金额是多少”来走一遍流程。第一步模型解析问题识别到“总金额”是统计指标“上个月”被换算成日期范围“A区”是区域过滤条件“完成订单”映射为status参数。第二步模型返回一个工具调用请求参数大概长这样{ region: A区, status: 已完成, metric: total_amount, date_from: 2024-11-01, date_to: 2024-11-30 }第三步系统执行query_orders在数据库里过滤出符合条件的订单算出SUM(amount)返回结果。第四步模型拿到类似{label: 总额, value: 1234567.0}的结果组织成自然语言回复“上个月A区完成订单的总金额为1234567.0元。”整个链路里没有任何一处数值是模型“想”出来的所以结果可复现、可审计。4. 常见问题与排查技巧实录4.1 检索召回缺少关键数值内容症状是回答内容看似合理但关键数字要么缺失要么不对。最常见的原因是只用向量检索而向量检索对精确数字非常不敏感。我遇到过一个情况用户问“B区设备的故障代码E204怎么处理”向量检索召回了“E204故障处理流程”的标题块但正文里对故障代码的描述在另一块文本里最终答案只有标题没有内容。解决办法就是上混合检索关键词命中能把“E204”这种精确标识直接拉出来。另一个很有效的方案是元数据过滤在入库时给每个块打上“文档类型”“适用区域”“版本号”等标签检索时直接用用户问题里解析出的元数据做硬过滤减少噪声。4.2 模型就是不调用工具这是最常见的坑。模型答得行云流水但一看就知道数值是编的完全没有走工具。我排查这个问题一般是按顺序看三个层面。第一看模型本身是否支持Function Calling。很多入门玩家用了一个不支持结构化工具调用的对话模型工具定义传进去直接被无视。这类模型可以考虑走Text2SQL路线或者换一个支持工具调用的大模型。第二看工具定义是否清晰。JSON Schema里的description字段非常重要模型就是靠这些文字来决定何时调用。一个模糊的工具描述模型根本不知道这个工具是干嘛的。我早期写的description是“查询订单数据”模型完全无感改成“当用户需要统计金额、数量、平均值等精确数值时必须调用”模型立刻就会用。第三看系统提示词是否强调规则。一定要在提示词里明确边界什么时候必须调用工具什么情况下允许不调用。我通常会加一条反例式的描述直接估算数字是错误行为必须基于工具结果回答。4.3 工具结果对了生成回答还是出错这种情况更隐蔽。工具返回了正确的数值大模型在组织语言时还是可能“发挥”改数字或者把单位写错。根源在于模型把工具结果当成了“参考资料”而不是“唯一事实来源”。解决办法有两个。第一是把温度调低统计问答场景我直接设成0到0.1温度越高编造概率越大。第二是在系统提示词里增加惩罚性约束比如“工具返回的数值是最终事实一字不改地引用你无权修改、省略或换算该数值”。同时把工具返回内容放到上下文里时用固定格式包裹例如“【工具返回】label总额, value1234567.0【工具返回结束】”让模型明确这是不可篡改的数据区。4.4 计算慢和超时的工程优化第一版流程跑通后很多人会发现数据库查询越来越慢。原因多半出在没加索引。orders表查询经常按order_date、region、status过滤这三个字段必须建索引否则全表扫描在数据量上来后就是灾难。CREATE INDEX idx_orders_date ON orders(order_date); CREATE INDEX idx_orders_region ON orders(region); CREATE INDEX idx_orders_status ON orders(status);另外工具执行一定要有超时控制。如果工具卡住了外层循环还在等整个接口就挂了。给工具执行套上一个超时熔断比如超过5秒就返回错误让模型换条件重试或者直接告诉用户查询超时。生产环境还要做缓存同一个问题加参数哈希后如果命中缓存直接返回历史结果能省一大笔模型调用费。4.5 安全问题只读SQL和工具白名单我在这个项目里吃过一次亏。当时图方便SQL直接用了表单拼接用户输入带了一个分号差点把orders表清掉。后来总结出血泪教训工具函数必须是白名单机制所有参数用参数化查询绝不做字符串拼接数据库连接使用只读模式工具日志完整记录每一次调用的入参和出参出了问题可以回放复盘。# 使用只读模式连接数据库 from urllib.parse import quote_plus readonly_conn sqlite3.connect( file: quote_plus(rag_demo.db) ?modero, uriTrue )这个只读连接只能在工具函数里使用入库导入数据用另一个单独连接。两权分离生产环境还要配合最小权限账户把写权限彻底收掉。5. 从“会算数”到“会干活”扩展与经验总结5.1 把工具集扩展成完整的数据助手“会计算”只是第一步。沿着同样的思路可以继续给系统加工具导出Excel、生成图表、对比报表、调用外部接口。核心模式完全一致工具负责执行模型负责决策。比如用户说“把A区这个月的订单按类别做成饼图”你只需要加一个generate_chart工具参数是数据和图表类型工具内部用绘图库生成图片文件再把图片路径返回给模型模型就能告诉用户“图表已生成参见图片”。这种扩展性就是工具调用模式最大的价值——每加一个工具系统能力就大一圈而且每个工具都是确定性的不会因为模型发挥不稳而突然失效。5.2 我实现过程中的几点真实体会第一不要试图让大模型自己算任何数字哪怕只是一个很简单的加法。概率模型做精确计算一次两次能对不代表上生产环境能稳定对。数字问题必须交给确定性系统这是这条链路里最核心的原则。第二先把最小闭环跑通再上重组件。我现在带着团队做这类项目第一步永远是SQLite加一个轻量向量索引把检索、工具调用、生成这三段链路验证通顺了再考虑替换成独立的向量数据库、任务编排框架。很多入门者上来就搭一堆组件出了问题根本分不清是哪个环节的锅。第三日志是调试的灵魂。工具调用的入参、返回结果、模型中间回复全部要打日志。我排查过的大部分诡异问题靠的都是回看日志而不是猜。加日志这个习惯比任何优化技巧都值钱。第四别忽略提示词工程。很多人觉得有了工具调用架构提示词就不重要了。实际上模型是否调工具、参数是否正确、结果是否被忠实引用每个环节都被提示词深深影响。把系统提示词当成一个功能模块来维护随着问题类型不断补充规则系统的稳定性和准确率都会肉眼可见地提升。这整套流程做下来“AI数据库”的结合点算是真正落地了。从一个只会翻文档的问答机器人升级成一个可以拍胸脯给出准确数字的数据助手。剩下的事情就是根据你自己的业务数据把工具函数越加越丰富让这个系统慢慢变成团队真正的数字业务伙伴。 
长河君 · 2026/10/23 3:23:22 · 阅读 1.2k 会计算的RAG：融合Function Calling与数据库查询的精确问答系统
最近好几个读者跟我聊RAG检索增强生成的时候都提到同一个尴尬场景文档丢进去了问一句“上个月某区域一共成交了多少单平均客单价是多少”大模型回答得倒是又快又自信结果一对数差得离谱。说白了普通RAG擅长“找信息”不擅长“算信息”。它把文档切成片段存进向量库用户提问时检索出最相关的片段再丢给大模型生成答案。这个过程对“某文档里有哪些处理流程”这类问题很管用可一旦涉及到精确的统计、求和、均值、多条件筛选大模型就露馅了——它本质上是一个语言模型不是在跑数据库查询。所以“AI数据库”这东西确实是神仙结合数据库负责精确计算AI负责理解意图和生成回答。这篇文章我会把整个“会计算的RAG系统”从思路到落地完整拆开给你一套可以直接跑通的最小实现再附上我实际调试中踩过的坑和排查套路。不管是刚接触RAG的新手还是已经在做智能问答的老手都应该能从这里拿到点东西。1. 为什么普通RAG一到计算就翻车问题拆解与方案选型1.1 三个典型翻车场景先说一个我印象很深的场景。有次给一个模拟项目做“销售周报问答”文档里明明写了“A区Q3销售额为1200万元B区为850万元”模型被问到“A区和B区哪个高”时回答对了但接着问“两者加起来是多少”就直接答成“2350万元”实际上应该是2050万元。这种问题不是它不会算而是它从文本里抽取数字时可能把“850万”理解成“800万50万”或者干脆自己“脑补”出了一个数字。另一个场景是表格类文档。很多制度文件、产品手册里都夹着大量表格比如“各档位折扣对照表”“故障代码说明表”。常规切分方式会把表格切成碎片表头是一块数据行是另一块向量检索时往往只召回其中几行结果模型拿着残缺数据强行生成答案数字怎么可能对得上。第三个场景是多条件统计。比如“帮我统计最近七天状态为已完成且金额大于5000的订单有多少笔”。这类问题需要精确过滤普通RAG根本不具备执行条件查询的能力。向量检索只会返回“语义上相似”的文本片段至于这些片段是否覆盖了所有满足条件的记录它完全无法保证。我把这些问题总结成一句话大模型是“语文考生”不是“算术考生”。它擅长从文字中概括、归纳、润色但碰到精确计算尤其是涉及多笔记录聚合时它的概率生成机制本身就是致命弱点。不能指望一个靠预测下一个词工作的系统去稳定输出一行正确的求和结果。1.2 “会计算”的两种实现路线要让RAG从“会找”变成“会算”业界主流有两条路线。路线A是工具调用也叫Function Calling。大模型在生成回答前先判断这个问题是否需要精确计算。如果需要它就输出一个结构化的“调用请求”比如指定一个叫“query_orders”的函数的参数系统收到后真正去数据库执行查询和聚合再把查询结果回传给大模型由大模型基于真实数字组织语言。路线B是Text2SQL。直接把数据库表结构、字段说明、枚举值写进系统提示词让大模型根据用户问题生成SQL语句然后交给数据库引擎执行最后把查询结果返回给大模型生成回答。我自己的选型经验是优先做工具调用把Text2SQL作为子能力藏在某个工具函数里。因为工具调用的可控性明显更好——每个工具对应一段固化的执行逻辑你可以在工具内部做各种防护而Text2SQL自由度太高模型一旦生成一个语义正确但逻辑错误的SQL排查起来很痛苦。对比项工具调用Function CallingText2SQL计算可控性计算逻辑固化在函数里更可靠SQL高度灵活容易生成错误语句适用问题统计类、固定口径类问题组合条件查询、探索式分析对模型要求需要模型支持工具调用普通模型可通过提示词生成SQL安全边界白名单函数天然隔离必须做只读限制和超时控制调试体验每一步工具调用都有日志需要看SQL和返回结果链路更长1.3 整体架构谁该做什么事整个“会计算的RAG系统”我习惯分成六层。文档层负责收集和清洗原始资料包括PDF、Word、Markdown、Excel等。索引层负责把文档切块、向量化、写入存储同时会把结构化的记录比如订单表、设备台账导入数据库。检索层负责任务分发是走向量检索还是查结构化数据库或者两者都查。规划层由大模型完成它负责理解用户问题、决定调哪个工具、确定参数。工具层是真正的算力来源包含查询函数、聚合函数、SQL执行模块。生成层最后把工具返回的结果用自然语言包装给用户。这个结构的关键在于大模型始终只做“决策”和“表达”所有数值计算都交给工具层和数据库。把这句话记住后面所有实现都是围绕它展开的。2. 动手前的技术选型与架构准备2.1 数据边界哪些进文档库哪些进数据库很多人拿到项目就急着装向量数据库其实第一步应该先分清数据边界。这个比选什么存储重要得多。我通常把数据分成三类。第一类是“知识型数据”比如操作手册、管理制度、产品说明这类数据以叙述和规范为主适合切分后进向量库第二类是“记录型数据”比如订单明细、工单记录、库存流水这类数据是结构化表格应该直接进关系型数据库第三类是“混合型数据”比如带统计表格的周报既保留叙述部分供检索又把表格数据抽出来单独建表。为什么要分开因为向量检索解决的是“语义相似”而精确计算解决的是“条件命中”。订单表这种东西按行切进向量库检索出来也是零散片段没有任何计算价值。正确做法是让大模型通过工具直接查库文档向量库只负责回答流程类、定义类问题。2.2 存储选型从轻量实现开始的理由存储方案上我强烈建议第一版不要上重组件直接用“Python标准库 一个轻量关系型数据库”把流程跑通。你完全可以自己实现一个简易向量检索文本向量用列表存查询时算余弦相似度数据量几千条以内性能完全可接受。这套东西跑通了再替换成专门的向量数据库也不迟。真正到了生产环境评估向量存储时我重点看四个维度。一是标量过滤能力能不能在检索时同时按“区域”“日期”“文档类型”过滤这个能力比很多花哨功能都重要二是索引类型是HNSW还是IVF不同索引在召回率和内存占用上差别很大三是并发能力团队内部用和对外提供服务是完全不同的量级四是部署复杂度是嵌入式还是要单独起服务直接影响运维成本。2.3 文档切分与Embedding的实操经验文档切分这块我踩过很多坑最简单的经验是不要按固定字符数硬切。按固定长度切块会把一个完整的表格拆得七零八落还会把一段话的逻辑从中间截断。我更推荐按语义边界切优先按标题层级其次是段落最后才是长度上限。块大小方面我常用512到1024个token具体看你的Embedding模型和场景。块太大检索精度下降块太小又会丢失上下文。真正要注意的是表格类内容务必整块保留。我会在预处理阶段检测到表格后先转换成Markdown或JSON格式作为一个独立块整体入库。这样检索时至少能保证表头和数据行在一起不会出现“只找到表头没找到数据”的尴尬情况。Embedding模型的选择标准我一般看四点中文语义效果好不好、向量维度是否适合当前存储、是本地化部署还是走接口调用、更新迭代是否频繁。正式项目里建议选一个在垂直领域表现稳定的模型而不是一味追求参数大。2.4 计算工具的设计原则工具调用是整个系统的核心工具定义得好不好直接决定模型“会不会用”。我总结了四条原则。第一工具名称要直白。函数名最好包含明确的动作和对象比如query_orders、calculate_revenue不要用do_stuff这类模糊名字。第二参数定义要严格。用JSON Schema把每个参数的类型、是否必填、可选项枚举写清楚给描述时把常见同义词也带上比如“日期范围”参数描述里写上“支持本月、上月、最近N天等自然语言”。第三输出要保持结构化。工具返回的不应该是大段文本应该是JSON包含数据和必要的说明字段这样模型解析时不容易出错。第四执行必须可控。所有工具函数内部要有异常捕获返回给模型的错误信息要准确这样模型才能根据错误重新生成参数。3. 核心环节实现手把手搭建一个会计算的RAG系统3.1 数据入库向量库和业务库并行写入下面进入代码环节。我用纯标准库实现配合一个轻量关系型数据库去掉外部依赖方便你完整跑通。先看入库部分的代码。import sqlite3 import json import hashlib conn sqlite3.connect(rag_demo.db) cur conn.cursor() # 创建向量存储表用于存放知识型文档块 cur.execute( CREATE TABLE IF NOT EXISTS doc_chunks ( id INTEGER PRIMARY KEY AUTOINCREMENT, chunk_text TEXT, source TEXT, embedding TEXT ) ) # 创建业务订单表用于精确计算 cur.execute( CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, region TEXT, category TEXT, amount REAL, status TEXT, order_date TEXT ) ) conn.commit()向量存储表里那列embedding直接存JSON序列化后的向量数组。单机演示、几千条数据完全没问题真要上生产再换真正的向量索引。接下来写一个简易Embedding函数。正式场景里这里要替换成真实的语义模型我在这里用一个伪向量只是为了让你在没有外部依赖的情况下把流程完整跑一遍。def embed_text(text): # 正式环境替换为真实的Embedding模型接口 digest hashlib.sha256(text.encode(utf-8)).digest() return [b / 255.0 for b in digest[:32]]入库时文档块向量化后写入doc_chunks结构化订单数据写入orders。def index_document(text, source): chunks split_into_chunks(text) for chunk in chunks: vec embed_text(chunk) cur.execute( INSERT INTO doc_chunks (chunk_text, source, embedding) VALUES (?, ?, ?), (chunk, source, json.dumps(vec)) ) conn.commit() def import_orders(orders_data): for order in orders_data: cur.execute( INSERT INTO orders (region, category, amount, status, order_date) VALUES (?, ?, ?, ?, ?), (order[region], order[category], order[amount], order[status], order[order_date]) ) conn.commit()split_into_chunks函数我就不完整贴了核心逻辑就是按标题和段落边界切分超过预设长度再截断同时保证重叠部分能衔接上下文。注意一个细节表格数据在切分前先转成Markdown作为一个独立块存储单独加一个标记。这样检索时能区分普通文本和表格块。3.2 混合检索向量检索和关键词检索双通道业务中很多数字检索问题向量检索并不擅长。你需要一个混合检索策略向量检索抓“语义相似”关键词检索抓“字面命中”最后按加权分数合并排序。import math def cosine_similarity(vec_a, vec_b): dot sum(x * y for x, y in zip(vec_a, vec_b)) norm_a math.sqrt(sum(x * x for x in vec_a)) norm_b math.sqrt(sum(x * x for x in vec_b)) if norm_a 0 or norm_b 0: return 0 return dot / (norm_a * norm_b) def vector_search(query_text, top_k5): query_vec embed_text(query_text) rows cur.execute(SELECT id, chunk_text, embedding FROM doc_chunks).fetchall() scored [] for row_id, text, embedding_json in rows: vec json.loads(embedding_json) score cosine_similarity(query_vec, vec) scored.append((score, row_id, text)) scored.sort(reverseTrue, keylambda x: x[0]) return [(row_id, text, score) for score, row_id, text in scored[:top_k]] def keyword_search(query_text, top_k5): terms [term for term in query_text.split() if term] rows cur.execute(SELECT id, chunk_text FROM doc_chunks).fetchall() scored [] for row_id, text in rows: score sum(1 for term in terms if term in text) if score 0: scored.append((score, row_id, text)) scored.sort(reverseTrue, keylambda x: x[0]) return [(row_id, text, score) for score, row_id, text in scored[:top_k]] def hybrid_search(query_text, top_k5): docs {} for row_id, text, score in vector_search(query_text, top_ktop_k * 2): docs[row_id] {text: text, vector_score: score, keyword_score: 0} for row_id, text, score in keyword_search(query_text, top_ktop_k * 2): item docs.setdefault(row_id, {text: text, vector_score: 0, keyword_score: 0}) item[keyword_score] score ranked [] for row_id, item in docs.items(): # 分数归一化后加权合并向量和关键词各占一半权重 total item[vector_score] * 0.6 item[keyword_score] * 0.4 item[total_score] total ranked.append((total, row_id, item[text])) ranked.sort(reverseTrue, keylambda x: x[0]) return [(row_id, text, score) for score, row_id, text in ranked[:top_k]]这里的加权比例你可以根据业务调如果主要是流程问答向量权重可以更高如果涉及大量专有名词、编号、型号关键词权重要提升。当前这套没有做分数归一化真实场景建议对两种分数先做min-max归一化再合并避免因为数值范围不同导致某一边主导排序。3.3 计算工具与Function Calling机制接下来是关键部分让大模型学会调用计算工具。先定义工具列表。这里用一个抽象的llm_chat函数来代表大模型接口你接任何支持工具调用的模型时把它替换成对应调用即可。TOOLS [ { type: function, function: { name: search_knowledge, description: 检索知识库文档查找流程说明、定义、制度等非结构化内容, parameters: { type: object, properties: { query: {type: string, description: 用户问题的关键词或原问题} }, required: [query] } } }, { type: function, function: { name: query_orders, description: 查询订单表数据支持按区域、类别、状态、日期范围过滤并计算总额、平均值、数量、最大值、最小值, parameters: { type: object, properties: { region: {type: string, description: 区域名称例如A区、B区不传表示全部}, category: {type: string, description: 商品类别不传表示全部}, status: {type: string, description: 订单状态例如已完成、进行中不传表示全部}, date_from: {type: string, description: 起始日期格式YYYY-MM-DD}, date_to: {type: string, description: 结束日期格式YYYY-MM-DD}, metric: {type: string, enum: [total_amount, avg_amount, count, max_amount, min_amount], description: 计算指标} }, required: [metric] } } } ]工具执行函数负责根据模型返回的调用参数执行真实计算。def execute_tool_call(tool_name, arguments): args json.loads(arguments) if tool_name search_knowledge: result hybrid_search(args.get(query, ), top_k5) return {docs: [{text: text, score: score} for _, text, score in result]} if tool_name query_orders: return run_order_query(args) return {error: f不认识的工具: {tool_name}} def run_order_query(args): conditions [] params [] if args.get(region): conditions.append(region ?) params.append(args[region]) if args.get(category): conditions.append(category ?) params.append(args[category]) if args.get(status): conditions.append(status ?) params.append(args[status]) if args.get(date_from): conditions.append(order_date ?) params.append(args[date_from]) if args.get(date_to): conditions.append(order_date ?) params.append(args[date_to]) where_sql if conditions: where_sql WHERE AND .join(conditions) metric args.get(metric, total_amount) # 这里用SQL参数化查询防止拼接注入 if metric total_amount: sql fSELECT SUM(amount) FROM orders {where_sql} label 总额 elif metric avg_amount: sql fSELECT AVG(amount) FROM orders {where_sql} label 平均客单价 elif metric count: sql fSELECT COUNT(*) FROM orders {where_sql} label 订单数量 elif metric max_amount: sql fSELECT MAX(amount) FROM orders {where_sql} label 最大单笔金额 elif metric min_amount: sql fSELECT MIN(amount) FROM orders {where_sql} label 最小单笔金额 else: return {error: f不支持的指标: {metric}} row cur.execute(sql, params).fetchone() return {label: label, value: row[0]}这里必须强调两点。第一所有参数必须走参数化SQL绝对不能直接把模型生成的字符串拼进SQL否则一旦模型被恶意提示词引导后果很严重。第二生产环境建议单独开一个只读数据库账户给这个工具用彻底杜绝写入可能。3.4 主流程模型决策、工具执行、结果生成最后把整个流程串起来。系统提示词要明确告诉模型统计类问题必须调用工具不要自己估算。SYSTEM_PROMPT 你是一个数据分析助手。你可以使用以下工具获取信息 - search_knowledge检索知识库文档。 - query_orders查询订单数据并执行统计计算。 当用户问题涉及数量、金额、比例、平均值等精确数值时必须先调用query_orders工具。 所有计算以工具返回结果为准严禁自行改数字或估算。 如果工具返回错误请根据错误信息调整输入参数后重试。 def llm_chat(messages, toolsNone): # 此处替换为真实大模型调用 # 入参和返回需要兼容类似 openai 风格的 chat completion 结构 pass def run_rag_agent(question, max_steps3): messages [{role: system, content: SYSTEM_PROMPT}] messages.append({role: user, content: question}) for step in range(max_steps): response llm_chat(messages, toolsTOOLS) if response.get(tool_calls): messages.append(response) for tool_call in response[tool_calls]: tool_result execute_tool_call( tool_call[function][name], tool_call[function][arguments] ) messages.append({ role: tool, tool_call_id: tool_call[id], content: json.dumps(tool_result, ensure_asciiFalse) }) else: return response[content] return 达到最大执行步数无法完成该问题。流程其实就是一个循环模型决定调工具系统执行并返回结果模型再基于结果继续。最多循环三轮已经覆盖绝大多数场景。实际调试中我发现超过三轮还在循环多半是参数生成有问题与其让模型反复试错不如直接把错误信息打上日志去分析。3.5 一个完整问答实例模型是如何“算”出答案的用“上个月A区完成订单的总金额是多少”来走一遍流程。第一步模型解析问题识别到“总金额”是统计指标“上个月”被换算成日期范围“A区”是区域过滤条件“完成订单”映射为status参数。第二步模型返回一个工具调用请求参数大概长这样{ region: A区, status: 已完成, metric: total_amount, date_from: 2024-11-01, date_to: 2024-11-30 }第三步系统执行query_orders在数据库里过滤出符合条件的订单算出SUM(amount)返回结果。第四步模型拿到类似{label: 总额, value: 1234567.0}的结果组织成自然语言回复“上个月A区完成订单的总金额为1234567.0元。”整个链路里没有任何一处数值是模型“想”出来的所以结果可复现、可审计。4. 常见问题与排查技巧实录4.1 检索召回缺少关键数值内容症状是回答内容看似合理但关键数字要么缺失要么不对。最常见的原因是只用向量检索而向量检索对精确数字非常不敏感。我遇到过一个情况用户问“B区设备的故障代码E204怎么处理”向量检索召回了“E204故障处理流程”的标题块但正文里对故障代码的描述在另一块文本里最终答案只有标题没有内容。解决办法就是上混合检索关键词命中能把“E204”这种精确标识直接拉出来。另一个很有效的方案是元数据过滤在入库时给每个块打上“文档类型”“适用区域”“版本号”等标签检索时直接用用户问题里解析出的元数据做硬过滤减少噪声。4.2 模型就是不调用工具这是最常见的坑。模型答得行云流水但一看就知道数值是编的完全没有走工具。我排查这个问题一般是按顺序看三个层面。第一看模型本身是否支持Function Calling。很多入门玩家用了一个不支持结构化工具调用的对话模型工具定义传进去直接被无视。这类模型可以考虑走Text2SQL路线或者换一个支持工具调用的大模型。第二看工具定义是否清晰。JSON Schema里的description字段非常重要模型就是靠这些文字来决定何时调用。一个模糊的工具描述模型根本不知道这个工具是干嘛的。我早期写的description是“查询订单数据”模型完全无感改成“当用户需要统计金额、数量、平均值等精确数值时必须调用”模型立刻就会用。第三看系统提示词是否强调规则。一定要在提示词里明确边界什么时候必须调用工具什么情况下允许不调用。我通常会加一条反例式的描述直接估算数字是错误行为必须基于工具结果回答。4.3 工具结果对了生成回答还是出错这种情况更隐蔽。工具返回了正确的数值大模型在组织语言时还是可能“发挥”改数字或者把单位写错。根源在于模型把工具结果当成了“参考资料”而不是“唯一事实来源”。解决办法有两个。第一是把温度调低统计问答场景我直接设成0到0.1温度越高编造概率越大。第二是在系统提示词里增加惩罚性约束比如“工具返回的数值是最终事实一字不改地引用你无权修改、省略或换算该数值”。同时把工具返回内容放到上下文里时用固定格式包裹例如“【工具返回】label总额, value1234567.0【工具返回结束】”让模型明确这是不可篡改的数据区。4.4 计算慢和超时的工程优化第一版流程跑通后很多人会发现数据库查询越来越慢。原因多半出在没加索引。orders表查询经常按order_date、region、status过滤这三个字段必须建索引否则全表扫描在数据量上来后就是灾难。CREATE INDEX idx_orders_date ON orders(order_date); CREATE INDEX idx_orders_region ON orders(region); CREATE INDEX idx_orders_status ON orders(status);另外工具执行一定要有超时控制。如果工具卡住了外层循环还在等整个接口就挂了。给工具执行套上一个超时熔断比如超过5秒就返回错误让模型换条件重试或者直接告诉用户查询超时。生产环境还要做缓存同一个问题加参数哈希后如果命中缓存直接返回历史结果能省一大笔模型调用费。4.5 安全问题只读SQL和工具白名单我在这个项目里吃过一次亏。当时图方便SQL直接用了表单拼接用户输入带了一个分号差点把orders表清掉。后来总结出血泪教训工具函数必须是白名单机制所有参数用参数化查询绝不做字符串拼接数据库连接使用只读模式工具日志完整记录每一次调用的入参和出参出了问题可以回放复盘。# 使用只读模式连接数据库 from urllib.parse import quote_plus readonly_conn sqlite3.connect( file: quote_plus(rag_demo.db) ?modero, uriTrue )这个只读连接只能在工具函数里使用入库导入数据用另一个单独连接。两权分离生产环境还要配合最小权限账户把写权限彻底收掉。5. 从“会算数”到“会干活”扩展与经验总结5.1 把工具集扩展成完整的数据助手“会计算”只是第一步。沿着同样的思路可以继续给系统加工具导出Excel、生成图表、对比报表、调用外部接口。核心模式完全一致工具负责执行模型负责决策。比如用户说“把A区这个月的订单按类别做成饼图”你只需要加一个generate_chart工具参数是数据和图表类型工具内部用绘图库生成图片文件再把图片路径返回给模型模型就能告诉用户“图表已生成参见图片”。这种扩展性就是工具调用模式最大的价值——每加一个工具系统能力就大一圈而且每个工具都是确定性的不会因为模型发挥不稳而突然失效。5.2 我实现过程中的几点真实体会第一不要试图让大模型自己算任何数字哪怕只是一个很简单的加法。概率模型做精确计算一次两次能对不代表上生产环境能稳定对。数字问题必须交给确定性系统这是这条链路里最核心的原则。第二先把最小闭环跑通再上重组件。我现在带着团队做这类项目第一步永远是SQLite加一个轻量向量索引把检索、工具调用、生成这三段链路验证通顺了再考虑替换成独立的向量数据库、任务编排框架。很多入门者上来就搭一堆组件出了问题根本分不清是哪个环节的锅。第三日志是调试的灵魂。工具调用的入参、返回结果、模型中间回复全部要打日志。我排查过的大部分诡异问题靠的都是回看日志而不是猜。加日志这个习惯比任何优化技巧都值钱。第四别忽略提示词工程。很多人觉得有了工具调用架构提示词就不重要了。实际上模型是否调工具、参数是否正确、结果是否被忠实引用每个环节都被提示词深深影响。把系统提示词当成一个功能模块来维护随着问题类型不断补充规则系统的稳定性和准确率都会肉眼可见地提升。这整套流程做下来“AI数据库”的结合点算是真正落地了。从一个只会翻文档的问答机器人升级成一个可以拍胸脯给出准确数字的数据助手。剩下的事情就是根据你自己的业务数据把工具函数越加越丰富让这个系统慢慢变成团队真正的数字业务伙伴。