Python SQL RAG Function Calling 构建审计数据分析智能体本文以“审计数据异常分析”为例设计并实现一个可落地的 AI 智能体方案。系统通过 Python 完成数据清洗通过 SQL 完成指标查询通过 RAG 检索制度依据通过 Function Calling 调用分析工具最终输出异常线索、原因解释和处理建议。摘要AI 智能体并不是简单地把问题丢给大模型而是让大模型具备“理解任务、选择工具、调用函数、整合结果、生成结论”的能力。在数据分析类场景中智能体尤其适合处理这类问题某类费用是否存在异常波动某个供应商是否存在集中采购风险是否存在重复报销、超标准报销、跨期入账等情况当前异常是否违反制度条款能否自动生成一份可读的分析结论本文围绕一个“审计数据分析智能体”展开使用 Python、Pandas、SQLite、RAG 和 Function Calling 实现一个完整技术方案。重点不在概念介绍而是拆解一个能落地的工程流程。目录一、智能体适合解决什么问题二、整体架构设计三、核心技术选型四、样例数据设计五、数据清洗流程六、SQL 指标分析七、RAG 制度检索八、Function Calling 工具设计九、智能体执行流程十、完整代码示例十一、运行与测试十二、工程优化方向十三、总结一、智能体适合解决什么问题传统数据分析一般是这样的流程拿到数据 ↓ 人工清洗 ↓ 写 SQL 或 Python 分析 ↓ 人工查看异常 ↓ 人工整理报告如果问题比较固定这种方式没有问题。但在实际分析中需求往往是动态的帮我看一下 6 月份差旅费有没有异常找出金额较高且审批周期异常短的报销记录这个供应商最近交易频率是不是突然变高了根据制度判断这笔费用有没有超标准风险这些问题有一个共同点用户不是直接写 SQL而是用自然语言描述分析目标。智能体要做的事情就是把自然语言问题转换成可执行的分析步骤。一个可用的数据分析智能体至少要具备四种能力能力说明理解问题判断用户想查数据、找异常、看制度还是生成报告调用工具根据任务选择 SQL 查询、异常检测、制度检索等工具整合结果把多个工具返回的数据组织成清晰结论可追溯输出依据包括数据口径、命中规则和参考制度二、整体架构设计本文设计的智能体架构如下用户问题 ↓ 意图识别 ↓ 任务规划 ↓ 工具调用 ├── 数据查询工具 ├── 异常检测工具 ├── 制度检索工具 └── 报告生成工具 ↓ 结果汇总 ↓ 输出分析结论从工程角度看可以分成四层数据层CSV / Excel / SQLite 分析层Pandas / SQL / 异常规则 知识层制度文档 / RAG 检索 智能体层大模型 Function Calling 工作流本文为了方便演示使用 SQLite 作为本地数据库使用简单文本检索模拟 RAG重点展示工程流程。后续可以替换为 MySQL、PostgreSQL、向量数据库或企业知识库。三、核心技术选型技术作用Python编写数据处理、工具函数和智能体流程Pandas清洗 CSV / Excel 数据计算统计指标SQLite存储结构化数据执行 SQL 查询SQL查询交易明细、聚合指标和异常记录RAG从制度文本中检索相关规则Function Calling让大模型按需调用工具函数LangChain / LlamaIndex可用于更复杂的智能体编排Dify / Coze可用于低代码工作流编排和应用发布通义千问 / GLM / DeepSeek可作为智能体的大模型底座这里要注意一个关键点智能体开发不是只会调大模型 API而是要把数据、工具、规则和模型组织成稳定流程。四、样例数据设计为了便于理解本文使用一份费用报销数据作为示例。字段设计如下字段含义record_id报销记录 IDemployee员工姓名department部门expense_type费用类型amount金额vendor供应商submit_date提交日期approve_date审批日期invoice_no发票号description费用说明示例数据record_id,employee,department,expense_type,amount,vendor,submit_date,approve_date,invoice_no,description R001,张三,销售部,差旅费,860,携程商旅,2026-06-01,2026-06-03,INV001,客户拜访交通住宿 R002,李四,市场部,招待费,3200,某餐饮公司,2026-06-02,2026-06-02,INV002,客户接待 R003,王五,销售部,差旅费,860,携程商旅,2026-06-03,2026-06-04,INV001,重复提交测试 R004,赵六,研发部,办公用品,1200,办公优选,2026-06-04,2026-06-10,INV004,采购键盘鼠标 R005,钱七,市场部,招待费,9800,某餐饮公司,2026-06-05,2026-06-05,INV005,大型活动接待这份数据中包含几个常见风险点发票号重复金额较高审批时间过短同一供应商交易集中费用类型可能超过制度标准五、数据清洗流程数据分析前必须先清洗数据。常见清洗步骤包括去除空行统一日期格式金额字段转数值去除字符串前后空格处理缺失值生成衍生字段例如审批耗时示例代码importpandasaspddefclean_expense_data(csv_path:str)-pd.DataFrame:dfpd.read_csv(csv_path)text_columns[record_id,employee,department,expense_type,vendor,invoice_no,description]forcolumnintext_columns:df[column]df[column].astype(str).str.strip()df[amount]pd.to_numeric(df[amount],errorscoerce).fillna(0)df[submit_date]pd.to_datetime(df[submit_date],errorscoerce)df[approve_date]pd.to_datetime(df[approve_date],errorscoerce)df[approve_days](df[approve_date]-df[submit_date]).dt.days dfdf.dropna(subset[record_id,submit_date,approve_date])returndf清洗后的数据可以写入 SQLite方便后续 SQL 查询importsqlite3defsave_to_sqlite(df:pd.DataFrame,db_path:str):connsqlite3.connect(db_path)df.to_sql(expense_records,conn,if_existsreplace,indexFalse)conn.close()六、SQL 指标分析SQL 适合处理结构化数据分析例如聚合、筛选、排序和分组统计。1. 查询金额最高的报销记录SELECTrecord_id,employee,department,expense_type,amount,vendor,submit_dateFROMexpense_recordsORDERBYamountDESCLIMIT10;2. 检查重复发票SELECTinvoice_no,COUNT(*)AScount_records,SUM(amount)AStotal_amountFROMexpense_recordsGROUPBYinvoice_noHAVINGCOUNT(*)1;3. 统计供应商交易集中度SELECTvendor,COUNT(*)AStransaction_count,SUM(amount)AStotal_amount,AVG(amount)ASavg_amountFROMexpense_recordsGROUPBYvendorORDERBYtotal_amountDESC;4. 查询审批耗时异常短的记录SELECTrecord_id,employee,department,expense_type,amount,approve_daysFROMexpense_recordsWHEREapprove_days0ORDERBYamountDESC;这些 SQL 可以封装成工具函数让智能体在需要时调用。七、RAG 制度检索结构化数据只能告诉我们“发生了什么”制度文本可以帮助判断“是否合理”。例如制度中可能写着差旅费单笔报销金额超过 1000 元时需要提供行程单和审批说明。 招待费单笔金额超过 3000 元时需要补充接待对象、事由和部门负责人审批。 同一发票号不得重复报销。 供应商交易金额短期内异常集中时应进行进一步核查。当用户问招待费 9800 元有没有风险智能体应该先检索相关制度片段再结合数据给出判断。为了演示本文用一个简单的关键词检索函数模拟 RAGPOLICY_DOCS[差旅费单笔报销金额超过 1000 元时需要提供行程单和审批说明。,招待费单笔金额超过 3000 元时需要补充接待对象、事由和部门负责人审批。,同一发票号不得重复报销。,供应商交易金额短期内异常集中时应进行进一步核查。,]defretrieve_policy(query:str,top_k:int2)-list[str]:scored_docs[]fordocinPOLICY_DOCS:score0forcharinset(query):ifcharindoc:score1scored_docs.append((score,doc))scored_docs.sort(reverseTrue,keylambdaitem:item[0])return[docforscore,docinscored_docs[:top_k]ifscore0]生产环境中可以升级为制度文档切分 ↓ Embedding 向量化 ↓ 存入 FAISS / Milvus / Chroma / pgvector ↓ 根据用户问题检索相关制度 ↓ 交给大模型生成有依据的回答八、Function Calling 工具设计Function Calling 的核心思想是大模型不直接执行所有事情而是根据任务选择合适的工具函数。例如用户问帮我找出重复报销的记录并说明可能违反了哪条制度智能体可以拆成两步1. 调用 check_duplicate_invoice 查询重复发票 2. 调用 retrieve_policy 检索重复报销相关制度可以设计以下工具工具名称作用query_top_expenses查询高金额报销check_duplicate_invoice检查重复发票analyze_vendor_concentration分析供应商集中度check_fast_approval查询审批过快记录retrieve_policy检索制度依据工具定义示例TOOLS[{type:function,function:{name:check_duplicate_invoice,description:检查是否存在同一发票号重复报销的记录,parameters:{type:object,properties:{},required:[]}}},{type:function,function:{name:retrieve_policy,description:根据问题检索相关制度条款,parameters:{type:object,properties:{query:{type:string,description:需要检索的制度问题}},required:[query]}}}]实际接入通义千问、GLM、DeepSeek 或其他模型时需要根据对应平台的工具调用格式进行适配。工程上建议把模型适配层单独封装避免业务代码和模型接口强耦合。九、智能体执行流程一个简化版智能体可以按下面流程执行用户输入问题 ↓ 判断问题类型 ↓ 选择工具 ↓ 执行工具函数 ↓ 整理数据结果 ↓ 检索制度依据 ↓ 生成分析结论例如用户输入请检查是否存在重复报销并给出处理建议智能体执行过程1. 识别关键词重复、报销、处理建议 2. 调用 check_duplicate_invoice 3. 调用 retrieve_policy(重复报销) 4. 汇总重复记录、金额、发票号 5. 输出风险说明和建议这就是一个最小可用的 Agent 工作流。十、完整代码示例下面给出一个不依赖外部大模型的最小可运行版本用规则模拟智能体的工具选择过程。这样可以先验证数据处理、SQL 分析和制度检索流程。新建audit_agent_demo.pyimportsqlite3frompathlibimportPathimportpandasaspd DATA_PATHPath(expense_records.csv)DB_PATHPath(audit_demo.db)SAMPLE_DATArecord_id,employee,department,expense_type,amount,vendor,submit_date,approve_date,invoice_no,description R001,张三,销售部,差旅费,860,携程商旅,2026-06-01,2026-06-03,INV001,客户拜访交通住宿 R002,李四,市场部,招待费,3200,某餐饮公司,2026-06-02,2026-06-02,INV002,客户接待 R003,王五,销售部,差旅费,860,携程商旅,2026-06-03,2026-06-04,INV001,重复提交测试 R004,赵六,研发部,办公用品,1200,办公优选,2026-06-04,2026-06-10,INV004,采购键盘鼠标 R005,钱七,市场部,招待费,9800,某餐饮公司,2026-06-05,2026-06-05,INV005,大型活动接待 R006,孙八,销售部,差旅费,1680,携程商旅,2026-06-06,2026-06-06,INV006,跨城客户拜访 POLICY_DOCS[差旅费单笔报销金额超过 1000 元时需要提供行程单和审批说明。,招待费单笔金额超过 3000 元时需要补充接待对象、事由和部门负责人审批。,同一发票号不得重复报销。,供应商交易金额短期内异常集中时应进行进一步核查。,审批流程应保留必要审核时间金额较高且当天审批通过的记录需要关注。,]definit_sample_data():ifnotDATA_PATH.exists():DATA_PATH.write_text(SAMPLE_DATA,encodingutf-8)defclean_expense_data(csv_path:Path)-pd.DataFrame:dfpd.read_csv(csv_path)text_columns[record_id,employee,department,expense_type,vendor,invoice_no,description,]forcolumnintext_columns:df[column]df[column].astype(str).str.strip()df[amount]pd.to_numeric(df[amount],errorscoerce).fillna(0)df[submit_date]pd.to_datetime(df[submit_date],errorscoerce)df[approve_date]pd.to_datetime(df[approve_date],errorscoerce)df[approve_days](df[approve_date]-df[submit_date]).dt.days dfdf.dropna(subset[record_id,submit_date,approve_date])returndfdefsave_to_sqlite(df:pd.DataFrame):connsqlite3.connect(DB_PATH)df.to_sql(expense_records,conn,if_existsreplace,indexFalse)conn.close()defquery_sql(sql:str)-pd.DataFrame:connsqlite3.connect(DB_PATH)resultpd.read_sql_query(sql,conn)conn.close()returnresultdefcheck_duplicate_invoice()-pd.DataFrame:sql SELECT invoice_no, COUNT(*) AS count_records, SUM(amount) AS total_amount FROM expense_records GROUP BY invoice_no HAVING COUNT(*) 1 ORDER BY total_amount DESC; returnquery_sql(sql)defquery_high_amount_expenses(threshold:float3000)-pd.DataFrame:sqlf SELECT record_id, employee, department, expense_type, amount, vendor, submit_date, approve_date, approve_days, invoice_no FROM expense_records WHERE amount {threshold}ORDER BY amount DESC; returnquery_sql(sql)defanalyze_vendor_concentration()-pd.DataFrame:sql SELECT vendor, COUNT(*) AS transaction_count, SUM(amount) AS total_amount, AVG(amount) AS avg_amount FROM expense_records GROUP BY vendor ORDER BY total_amount DESC; returnquery_sql(sql)defcheck_fast_approval()-pd.DataFrame:sql SELECT record_id, employee, department, expense_type, amount, vendor, approve_days, invoice_no FROM expense_records WHERE approve_days 0 ORDER BY amount DESC; returnquery_sql(sql)defretrieve_policy(query:str,top_k:int2)-list[str]:scored_docs[]fordocinPOLICY_DOCS:score0forcharinset(query):ifcharindoc:score1scored_docs.append((score,doc))scored_docs.sort(reverseTrue,keylambdaitem:item[0])return[docforscore,docinscored_docs[:top_k]ifscore0]defformat_dataframe(df:pd.DataFrame)-str:ifdf.empty:return未发现符合条件的记录。returndf.to_string(indexFalse)defaudit_agent(question:str)-str:questionquestion.strip()outputs[]if重复inquestionor发票inquestion:duplicate_dfcheck_duplicate_invoice()policiesretrieve_policy(重复发票 重复报销)outputs.append(【分析任务】重复发票检查)outputs.append(【数据结果】)outputs.append(format_dataframe(duplicate_df))outputs.append(【制度依据】)outputs.extend(policies)ifnotduplicate_df.empty:outputs.append(【建议】存在同一发票号对应多条报销记录的情况建议核查原始发票、报销附件和提交人说明。)elif供应商inquestionor集中inquestion:vendor_dfanalyze_vendor_concentration()policiesretrieve_policy(供应商 交易 集中)outputs.append(【分析任务】供应商交易集中度分析)outputs.append(【数据结果】)outputs.append(format_dataframe(vendor_df))outputs.append(【制度依据】)outputs.extend(policies)outputs.append(【建议】重点关注交易金额高、交易次数多的供应商并结合采购合同、验收记录和付款审批进一步核查。)elif审批inquestionor过快inquestion:approval_dfcheck_fast_approval()policiesretrieve_policy(审批 过快 金额较高)outputs.append(【分析任务】审批耗时异常检查)outputs.append(【数据结果】)outputs.append(format_dataframe(approval_df))outputs.append(【制度依据】)outputs.extend(policies)outputs.append(【建议】金额较高且当天审批通过的记录应关注是否存在审批流于形式或补录审批的问题。)else:high_amount_dfquery_high_amount_expenses()policiesretrieve_policy(question)outputs.append(【分析任务】高金额费用检查)outputs.append(【数据结果】)outputs.append(format_dataframe(high_amount_df))outputs.append(【制度依据】)outputs.extend(policies)outputs.append(【建议】建议优先核查金额较高记录的业务真实性、审批完整性和附件充分性。)return\n.join(outputs)defmain():init_sample_data()dfclean_expense_data(DATA_PATH)save_to_sqlite(df)questions[请检查是否存在重复报销,分析供应商交易是否集中,找出审批过快的异常记录,查看高金额费用风险,]forquestioninquestions:print(*80)print(f用户问题{question})print(audit_agent(question))if__name____main__:main()十一、运行与测试1. 安装依赖python-m pip install pandas2. 运行程序python audit_agent_demo.py运行后会自动生成expense_records.csv audit_demo.db并输出几类分析结果重复发票检查供应商集中度分析审批耗时异常检查高金额费用检查输出示例用户问题请检查是否存在重复报销 【分析任务】重复发票检查 【数据结果】 invoice_no count_records total_amount INV001 2 1720 【制度依据】 同一发票号不得重复报销。 【建议】 存在同一发票号对应多条报销记录的情况建议核查原始发票、报销附件和提交人说明。这个版本虽然没有接入真实大模型但已经把智能体的核心流程跑通了自然语言问题 ↓ 工具选择 ↓ SQL 查询 ↓ 制度检索 ↓ 结论生成后续接入大模型时只需要把audit_agent()中的规则判断替换为模型的 Function Calling 即可。十二、工程优化方向1. 接入真实大模型可以使用通义千问、GLM、DeepSeek 等模型作为智能体推理层。大模型负责理解用户问题并决定调用哪个工具。工程上建议封装统一接口classLLMClient:defchat(self,messages:list[dict])-str:passdeftool_call(self,messages:list[dict],tools:list[dict])-dict:pass这样后续切换模型时不需要大改业务逻辑。2. 使用向量数据库实现真正 RAG本文用简单关键词方式模拟制度检索。实际项目可以升级为制度 PDF / Word ↓ 文本切分 ↓ Embedding 向量化 ↓ FAISS / Chroma / Milvus / pgvector ↓ 语义检索 ↓ 大模型生成有依据的解释这样用户即使用不同表述提问也能检索到相关制度。3. 增加工作流编排复杂任务可以拆成多个步骤问题理解 ↓ 数据查询 ↓ 异常检测 ↓ 制度检索 ↓ 风险评级 ↓ 报告生成如果使用 Dify、Coze 或类似平台可以把这些步骤配置成可视化工作流。如果使用代码实现可以用 LangChain、LlamaIndex 或自定义状态机管理流程。4. 增加多智能体协作可以拆成多个角色智能体职责数据分析智能体负责 SQL 查询和统计分析制度检索智能体负责查找相关制度风险判断智能体负责判断风险等级报告生成智能体负责生成最终分析报告多智能体不是越多越好只有当任务边界清晰、工具复杂度较高时才有必要拆分。5. 增加前端或接口服务当前示例是命令行程序。后续可以增加Streamlit 页面FastAPI 后端接口Excel 上传功能分析报告导出用户权限控制分析任务日志这样就能从 Demo 逐步演化成一个完整应用。十三、总结本文以审计数据异常分析为例拆解了一个数据分析智能体的落地方案。核心流程是数据清洗 → SQL 分析 → 异常识别 → 制度检索 → 工具调用 → 结论生成从技术角度看这类智能体项目的关键不只是调用大模型 API而是把以下能力组合起来Python 数据处理能力SQL 查询和指标分析能力RAG 知识检索能力Function Calling 工具调用能力工作流编排能力结果解释和可追溯能力当这些模块组合起来后大模型才真正从“聊天工具”变成“能处理业务任务的智能体”。参考资料阿里云百炼 Model Studio 文档https://help.aliyun.com/zh/model-studio/getting-started/what-is-model-studioLangChain 官方文档https://docs.langchain.com/oss/python/langchain/overviewDify 官方文档https://docs.dify.ai/Pandas 官方文档https://pandas.pydata.org/docs/SQLite 官方文档https://www.sqlite.org/docs.html
)
