这次我们来看一个很典型的 AI 智能体落地场景数字图书管理员。它不是简单的图书检索系统而是把 SQL 关系型数据库和向量数据库组合起来让 LLM 自动完成“结构化查询 语义检索 回答生成”的完整链路。如果你正在做 AI 智能体开发或者想了解 SQL 与向量数据库协同工作流怎么落地这篇文章可以直接收藏。项目核心就三件事第一用 SQL 管理图书的结构化数据比如 ISBN、作者、出版年份、分类、库存状态第二用向量数据库管理非结构化语义比如书名含义、内容摘要、评论情感、正文片段第三通过大模型的 Function Calling 或工具调用能力把两类查询封装成智能体工具让 AI 根据用户问题自动选择查询方式。全文会用一套可落地的方案讲清楚数据模型设计、智能体路由逻辑、SQL 工具和向量检索工具的实现、API 接口封装、批量入库任务以及常见的故障排查。会给出可复制的 Python 和 SQL 示例核心代码都不长本地开发机就能跑通。1. 核心能力速览能力项说明项目类型AI 智能体应用 / RAG 检索增强生成系统核心功能图书结构化查询、语义检索、混合检索、智能问答技术栈Python、FastAPI、SQLite/PostgreSQL、ChromaDB/pgvector、LangChain 或自研 Agent数据库能力SQL 负责精确查询向量数据库负责语义检索智能体能力意图识别、工具路由、SQL 生成、结果融合、回答生成推荐部署环境8G 内存以上开发机CPU 可运行GPU 用于提升 LLM 推理速度显存占用取决于 LLM 服务需按实际模型版本测试启动方式命令行启动 Web 服务 / API 服务是否支持 API支持RESTful 接口是否支持批量任务支持可批量导入图书数据和向量化适合读者AI 应用开发者、RAG 系统开发者、图书管理系统开发者从材料看这个项目的重点不是复杂算法而是数据架构和工具编排能力。SQL 解决“精确”问题向量数据库解决“模糊”问题智能体解决“选择哪条路”的问题。2. 适用场景与使用边界数字图书管理员最适合以下场景图书馆、资料室的图书检索和借阅咨询。企业内部知识库的文档问答比如制度文件、操作手册。内容平台的书籍推荐和语义搜索。RAG 系统的教学演示适合作为智能体开发入门案例。它能解决的问题很明确用户问“《三体》是哪年出版的”“作者是谁”这种结构化问题直接走 SQL用户问“有没有类似《百年孤独》魔幻现实主义风格的书”这种语义问题走向量检索用户问“帮我找 2023 年以后出版、和人工智能相关的书”这种同时包含结构化条件和语义条件的就要走混合检索。使用边界也要说清楚。第一数字图书如果涉及版权保护内容只能建立索引和摘要不能直接存储和输出全文。第二涉及用户借阅历史、个人信息时要做好脱敏和权限控制。第三智能体自动生成 SQL 存在注入风险必须对输入和输出做双重校验。第四向量检索结果有概率性不能完全替代精确查询重要数据判断应该走 SQL 确认。3. 整体架构与工作流设计这个项目的工作流可以拆成四层。第一层是数据层。图书的元数据比如书名、作者、ISBN、分类、出版年份、库存量存放在关系型数据库里。图书的内容摘要、章节文本、评论、标签经过 Embedding 模型向量化之后存入向量数据库。两个库通过图书 ID 关联。第二层是工具层。智能体暴露几个工具search_books_sql用于精确查询search_books_vector用于语义查询search_books_hybrid用于混合查询。每个工具都封装成函数有明确的参数描述和返回值格式。第三层是智能体层。LLM 接收用户问题后先判断查询意图再决定调用哪个工具最后把工具返回的数据整理成自然语言回答。这一步可以用 LangChain 的 Agent 框架也可以自己写一个简单的路由函数。第四层是接入层。通过 FastAPI 提供 Web API前端可以是一个简单的聊天页面也可以接到企业微信、钉钉等 IM 机器人。下面是一个实用的工作流路由逻辑def route_query(user_query: str): rule_engine detect_structured_intent(user_query) semantic_engine detect_semantic_intent(user_query) if rule_engine and not semantic_engine: return sql elif semantic_engine and not rule_engine: return vector elif rule_engine and semantic_engine: return hybrid else: return llm_generatedetect_structured_intent可以检测用户问题中是否包含 ISBN、作者名、出版年份、分类名称等结构化字段。detect_semantic_intent可以检测是否包含“类似”“风格”“推荐”“相关”等语义类关键词。实际的智能体不会只靠关键词判断更可靠的方式是让 LLM 自己决定工具调用也就是 Function Calling。后面会给出代码示例。4. 环境准备与数据模型设计4.1 基础环境推荐环境清单如下具体版本需按实际项目调整操作系统Ubuntu 20.04 / CentOS 7 / Windows 10 均可。Python3.10 以上。数据库SQLite 适合单机测试PostgreSQL 适合生产。向量数据库ChromaDB 适合快速验证pgvector 适合和 PostgreSQL 统一管理Milvus 适合大规模生产。LLM 服务OpenAI API 兼容接口或本地部署的 Qwen、ChatGLM 等模型服务。安装基础依赖pip install fastapi uvicorn sqlalchemy chromadb openai pydantic如果使用 pgvector需要额外安装 PostgreSQL 扩展CREATE EXTENSION IF NOT EXISTS vector;4.2 图书表结构设计用 SQLite 做演示时图书主表可以这样建CREATE TABLE books ( id INTEGER PRIMARY KEY AUTOINCREMENT, isbn TEXT UNIQUE, title TEXT NOT NULL, author TEXT, publisher TEXT, category TEXT, publish_year INTEGER, stock INTEGER DEFAULT 0, summary TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_books_category ON books(category); CREATE INDEX idx_books_publish_year ON books(publish_year); CREATE INDEX idx_books_title ON books(title);这里summary字段可以放内容摘要向量表存的是 summary 的向量表示。生产环境中summary可能很长更适合放在独立的book_chunks表里按段落存储。4.3 向量数据表设计使用 ChromaDB 时集合设计如下import chromadb client chromadb.PersistentClient(path./chroma_data) collection client.get_or_create_collection( namebooks, metadata{hnsw:space: cosine} )集合中的每条记录包含id图书 ID对应 SQL 表中的主键。document摘要或正文片段。metadata包含 title、author、category、publish_year 等结构化字段。这样设计的好处是向量检索结果可以直接返回元数据不需要每次回表查 SQL。但需要注意向量库里的元数据是冗余副本更新图书信息时要同步更新。4.4 智能体表结构设计如果要做用户提问日志和反馈记录可以增加两张表CREATE TABLE query_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_query TEXT, route_type TEXT, answer TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE feedback_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_log_id INTEGER, rating INTEGER, comment TEXT );日志表不仅用于排查问题还可以作为后续微调智能体的训练数据。5. SQL 工具与向量检索工具实现5.1 SQL 查询工具智能体的第一个工具是search_books_sql负责把用户的自然语言转换为 SQL 并查询。这里推荐使用 LLM 生成 SQL 的方式但必须加一层安全校验。示例工具函数import sqlite3 import json def search_books_sql(query_params: dict): 根据结构化参数查询图书信息。 query_params 支持 isbn, title, author, category, publish_year, stock conn sqlite3.connect(library.db) cursor conn.cursor() conditions [] values [] if query_params.get(isbn): conditions.append(isbn ?) values.append(query_params[isbn]) if query_params.get(author): conditions.append(author LIKE ?) values.append(f%{query_params[author]}%) if query_params.get(category): conditions.append(category ?) values.append(query_params[category]) if query_params.get(publish_year): conditions.append(publish_year ?) values.append(query_params[publish_year]) if query_params.get(title): conditions.append(title LIKE ?) values.append(f%{query_params[title]}%) sql SELECT * FROM books if conditions: sql WHERE AND .join(conditions) sql LIMIT ? values.append(query_params.get(limit, 10)) cursor.execute(sql, values) rows cursor.fetchall() column_names [desc[0] for desc in cursor.description] conn.close() results [] for row in rows: results.append(dict(zip(column_names, row))) return json.dumps(results, ensure_asciiFalse)这里没有直接让 LLM 生成任意 SQL而是通过参数化查询的方式把用户请求先解析成结构化字段再拼接 SQL。这种方式能显著降低 SQL 注入风险。5.2 向量检索工具第二个工具是search_books_vector负责语义检索。用户输入一句话先做 Embedding再去向量库找最相近的图书。def search_books_vector(query_text: str, top_k: int 5): from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) response client.embeddings.create( modeltext-embedding-v1, inputquery_text ) query_embedding response.data[0].embedding results collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, metadatas, distances] ) books [] for i in range(len(results[ids][0])): books.append({ book_id: results[ids][0][i], title: results[metadatas][0][i].get(title), author: results[metadatas][0][i].get(author), distance: results[distances][0][i], snippet: results[documents][0][i][:100] }) return json.dumps(books, ensure_asciiFalse)向量检索的关键参数是distance也就是相似度分数。ChromaDB 默认按距离排序距离越小越相似。实际使用时要统计一下正常结果的分数分布设定一个阈值分数超过阈值的直接过滤掉避免无关结果混入。5.3 混合检索工具第三个工具是search_books_hybrid把 SQL 和向量检索结合起来。典型场景是“找一本人工智能类的书风格要通俗易读”。def search_books_hybrid(category: str, query_text: str, top_k: int 5): # 第一步SQL 按分类缩小范围 conn sqlite3.connect(library.db) cursor conn.cursor() cursor.execute( SELECT id, title, author, category, publish_year, stock FROM books WHERE category ? LIMIT 50, (category,) ) sql_rows cursor.fetchall() conn.close() allowed_ids [str(row[0]) for row in sql_rows] if not allowed_ids: return json.dumps([], ensure_asciiFalse) # 第二步嵌入查询文本 response client.embeddings.create( modeltext-embedding-v1, inputquery_text ) query_embedding response.data[0].embedding # 第三步在限定 ID 范围内检索 results collection.query( query_embeddings[query_embedding], n_resultstop_k, where{book_id: {$in: allowed_ids}}, include[documents, metadatas, distances] ) # 第四步结果融合排序 books [] for i in range(len(results[ids][0])): books.append({ book_id: results[ids][0][i], title: results[metadatas][0][i].get(title), category: category, distance: results[distances][0][i] }) books.sort(keylambda x: x[distance]) return json.dumps(books, ensure_asciiFalse)混合检索的融合策略有两种一种是先过滤再检索适合分类、年份、作者等结构化条件很强的情况另一种是分别检索再合并排序适合两者权重相等的场景。实际项目里建议先做 A/B 测试确定哪种策略更符合业务需求。6. 智能体路由与 Function Calling 实现上面三个工具都封装好了接下来要让 LLM 自动选择调用哪个工具。最稳定的方式是 Function Calling。6.1 工具声明tools [ { type: function, function: { name: search_books_sql, description: 根据结构化条件查询图书信息适用于作者、ISBN、分类、出版年份、库存储备等精确查询, parameters: { type: object, properties: { title: {type: string, description: 书名}, author: {type: string, description: 作者}, category: {type: string, description: 图书分类}, publish_year: {type: integer, description: 出版年份}, isbn: {type: string, description: ISBN}, limit: {type: integer, description: 返回条数默认10} } } } }, { type: function, function: { name: search_books_vector, description: 语义检索图书适用于推荐、相似风格、模糊描述等查询, parameters: { type: object, properties: { query_text: {type: string, description: 用户的语义描述}, top_k: {type: integer, description: 返回条数默认5} }, required: [query_text] } } } ]6.2 智能体主循环from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) def run_agent(user_query: str): messages [ {role: system, content: 你是数字图书管理员根据用户问题选择工具并回答问题。}, {role: user, content: user_query} ] response client.chat.completions.create( modelqwen2.5-7b-instruct, messagesmessages, toolstools, tool_choiceauto ) message response.choices[0].message if message.tool_calls: tool_call message.tool_calls[0] function_name tool_call.function.name arguments json.loads(tool_call.function.arguments) if function_name search_books_sql: result search_books_sql(arguments) elif function_name search_books_vector: result search_books_vector(arguments.get(query_text), arguments.get(top_k, 5)) else: result json.dumps({error: 未知工具}) messages.append(message) messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) final_response client.chat.completions.create( modelqwen2.5-7b-instruct, messagesmessages, toolstools, tool_choicenone ) return final_response.choices[0].message.content else: return message.content这套主循环就是智能体的核心框架。用户提问后LLM 判断是否需要调用工具如果需要就生成工具参数程序执行工具后把结果回传给 LLMLLM 再生成最终回答。这里的base_url可以指向 OpenAI API也可以指向本地部署的 vLLM、Ollama 等兼容服务。只需要把api_key换成实际密钥即可。7. API 接口与批量任务7.1 API 服务搭建用 FastAPI 把智能体封装成 HTTP 接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI(title数字图书管理员 API) class QueryRequest(BaseModel): query: str session_id: str default class QueryResponse(BaseModel): answer: str route_type: str app.post(/api/query, response_modelQueryResponse) async def query_books(req: QueryRequest): answer run_agent(req.query) return QueryResponse(answeranswer) app.get(/api/health) async def health_check(): return {status: ok}启动服务uvicorn main:app --host 0.0.0.0 --port 7860启动后可以直接在浏览器访问http://127.0.0.1:7860/docs查看 Swagger 文档调试接口非常方便。7.2 curl 调用示例curl -X POST http://127.0.0.1:7860/api/query \ -H Content-Type: application/json \ -d {query: 帮我找一本关于人工智能的科普书, session_id: test-001}7.3 Python 调用示例import requests url http://127.0.0.1:7860/api/query payload { query: 刘慈欣有哪些作品, session_id: test-002 } response requests.post(url, jsonpayload, timeout60) print(response.json())7.4 批量图书入库任务图书管理一个很大的需求是批量导入。假设有一份 CSV 文件格式如下isbn,title,author,publisher,category,publish_year,stock,summary 9787536692930,三体,刘慈欣,重庆出版社,科幻,2008,10,地球文明向宇宙发出的第一声啼鸣 9787536692947,三体2黑暗森林,刘慈欣,重庆出版社,科幻,2008,8,宇宙社会学核心设定批量入库脚本import csv import sqlite3 from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) def batch_import(csv_path: str, batch_size: int 20): conn sqlite3.connect(library.db) cursor conn.cursor() with open(csv_path, r, encodingutf-8) as f: reader csv.DictReader(f) batch [] for row in reader: cursor.execute( INSERT INTO books (isbn, title, author, publisher, category, publish_year, stock, summary) VALUES (?, ?, ?, ?, ?, ?, ?, ?), (row[isbn], row[title], row[author], row[publisher], row[category], int(row[publish_year]), int(row[stock]), row[summary]) ) book_id cursor.lastrowid batch.append({ id: str(book_id), document: row[summary], metadata: { title: row[title], author: row[author], category: row[category] } }) if len(batch) batch_size: collection.upsert(ids[b[id] for b in batch], documents[b[document] for b in batch], metadatas[b[metadata] for b in batch]) batch [] if batch: collection.upsert(ids[b[id] for b in batch], documents[b[document] for b in batch], metadatas[b[metadata] for b in batch]) conn.commit() conn.close() print(批量导入完成)批量任务的注意点有两个。第一SQL 批量插入要放在一个事务里失败时能回滚。第二向量化调用要控制并发。如果 Embedding 服务能力有限一次请求太多容易超时建议按批次提交批次大小根据 Embedding 服务的吞吐能力调整。7.5 批量任务队列设计生产环境批量导入可能上万条数据不能直接在主线程循环建议用消息队列CSV 文件 - 读取任务 - 数据清洗 - SQL 入库 - 向量化 - 向量入库 - 完成标记可以用 Celery、Redis Queue 或者简单的 Python 多进程实现。关键是要做断点续传记录每本书的处理状态失败的书重新入队避免全部重跑。8. 功能测试与效果验证8.1 结构化查询测试测试用例测试项输入示例预期结果精确查询“《三体》作者是谁”返回刘慈欣条件筛选“2020年以后出版的科幻书”返回符合年份和分类的书库存查询“有哪些书库存大于5”返回库存大于5的书记录判断标准SQL 工具返回的数据和直接执行 SQL 的结果一致。这里可以重点观察 LLM 是否把用户问题正确解析成了结构化参数。8.2 语义检索测试测试用例测试项输入示例预期结果推荐书籍“有没有类似《活着》那种沉重现实的书”返回与苦难、现实题材相关的书模糊描述“讲外星人和宇宙文明的书”返回科幻类图书跨语言检索“books about machine learning”返回机器学习相关图书判断标准返回结果在语义上相关而不是简单的关键词匹配。如果结果不相关优先检查 Embedding 模型的质量和图书摘要的文本长度。8.3 混合检索测试测试用例测试项输入示例预期结果分类语义“人工智能方面适合入门读的书”返回 AI 分类下、语义上偏向入门、通俗的图书年份语义“2022年以后出版的有深度的历史书”返回 2022 年后出版、历史类、深度内容优先判断标准每条结果既要满足结构化条件又要满足语义相关性。两者都满足的排在前面。8.4 全链路稳定性测试连续向智能体发起 100 个不同类型的测试请求统计请求成功率。平均响应时间。工具调用正确率。最终回答的可读性。如果成功率低于 95%需要检查是哪一步出错。可能是 LLM 生成参数异常也可能是工具函数返回格式不符合预期。8.5 安全测试必须测试的边界场景1. SQL 注入测试输入 1; DROP TABLE books; -- 2. 越权查询测试输入涉及隐私的查询 3. 超长输入测试输入几万字的文本 4. 空查询测试输入空白字符安全测试的目的是确认智能体不会把危险 SQL 透传到数据库。参数化查询已经能挡住大部分注入攻击但 LLM 生成工具参数时仍有可能“突发奇想”所以数据库账号权限也很关键。给智能体专用的数据库账号只开放SELECT权限禁止DROP、DELETE、UPDATE等高危操作这是底线要求。9. 资源占用与性能观察这个项目的资源消耗主要来自四个部分LLM 服务、Embedding 服务、关系型数据库、向量数据库。LLM 服务占资源最大。如果本地部署 7B 模型显存占用通常在 6G 到 16G 之间具体需要看模型量化和推理框架。如果本地没有 GPU可以改用 API 调用方式本机只跑数据库和工具层开发机只需要 8G 内存。Embedding 模型相对轻量CPU 也能跑。以文本向量化为例百条级别的查询在 CPU 上是毫秒到几十毫秒级别不影响交互体验。SQLite 在单机测试时几乎无感知但生产环境一旦并发上来建议切换到 PostgreSQL并对高频查询字段建索引。向量数据库的性能取决于索引类型和召回数量。HNSW 索引响应快但占内存Flat 索引准确但速度慢需要按数据量测试。性能观察建议用命令# 查看 GPU 显存占用 nvidia-smi -l 2 # 查看本机内存和 CPU htop # 查看进程端口占用 lsof -i:7860如果发现响应变慢先定位瓶颈。如果 SQL 查询慢用EXPLAIN QUERY PLAN看是否走了索引。如果向量检索慢减小top_k或者给集合重建索引。如果整条链路慢优先看 LLM 推理耗时可以在智能体主循环里加日志记录每个环节的耗时。这里还可以补充一个“分级缓存”的思路。热门问题结果可以缓存到 Redis缓存键用用户问题的 Embedding 做相似度匹配命中缓存后直接返回可以节省大量 LLM 调用成本。10. 常见问题与排查方法问题现象可能原因排查方式解决方案智能体返回“不知道”路由判断错误没有调用工具打印 tool_calls 数据优化工具描述增加示例SQL 工具返回空结果查询条件太严格单独执行 SQL 验证放宽条件使用模糊匹配向量检索结果不相关Embedding 模型与业务不匹配抽样检查距离分数更换 Embedding 模型或重写摘要向量数据库连接失败服务未启动或路径不对查看启动日志确认 persist 路径重启服务混合检索结果排序混乱融合策略选择错误分开对比两次检索结果使用 RRF 或加权融合算法API 请求超时LLM 推理时间过长查看调用耗时开启流式输出设置合理超时时间批量导入中断单条数据格式异常查看错误日志增加异常捕获标记失败数据后续重试LLM 生成 SQL 参数错误工具描述不清晰检查 Function Calling 的 arguments增加参数示例提高温度设为 0端口冲突7860 被占用lsof -i:7860更换端口启动内存持续增长向量库加载数据过多查看内存曲线限制集合大小使用服务端向量库关于 SQL 慢查询这里多说一句。智能体生成的查询条件有时候会不带索引比如LIKE %关键词%这种写法在数据量大时就是全表扫描。优化方案是配合 SQLite 的 FTS5 全文索引或者 PostgreSQL 的 pg_trgm 扩展把模糊查询从全表扫描变成索引扫描。11. 最佳实践与合规提醒第一数据模型设计上图书表要预留created_at、updated_at字段向量表要用图书 ID 关联主表避免只存向量不存原文。因为后续如果有数据修正需要根据 ID 定位更新。第二关于 SQL 注入防护推荐策略是“参数化查询 白名单校验”。凡是输入到 SQL 的条件只允许出现在白名单字典里。LLM 生成参数前也要先过滤一次字段名确保没有DROP、DELETE、ALTER等关键字。第三向量数据更新要和 SQL 数据保持一致。最简单的做法是建立一个“待同步队列”图书主表更新后把记录 ID 写入队列另一个消费者负责更新向量库。第四版权合规是数字图书项目的高压线。项目的定位是“图书管理员”负责的是图书的元数据和摘要信息不是提供全文下载。如果要索引正文内容建议只对公开版权书籍或企业自有文档做全文索引。涉及用户上传的书籍或评论时也需要在用户协议里明确授权范围。第五个人隐私保护。图书借阅记录、用户查询历史、评分反馈都属于个人信息。生产环境要加密存储、按角色控制访问权限、定期导出删除。API 层需要增加身份认证不能用裸接口直接暴露内网服务。第六给所有外部依赖加版本锁定。Python 依赖用requirements.txt锁定版本Docker 部署时锁定镜像 tag。否则 Embedding 模型更新后已经入库的向量和新的向量可能不在同一个语义空间导致检索效果突然变差。12. 总结与下一步数字图书管理员这个项目最有价值的地方是把 SQL 的精确查询能力、向量数据库的语义检索能力、LLM 的工具调用能力组合成了一个完整的智能体工作流。它不是玩具项目而是很多企业知识库问答系统的简化版。搭好这套框架之后换一个应用场景只需要改数据表和工具函数。建议最先验证三件事第一用 Function Calling 让智能体正确识别 SQL 查询和向量查询第二混合检索的排序效果第三批量导入上万条数据后的系统稳定性。最容易踩的坑也提前说LLM 工具参数解析偶发混乱需要加异常重试向量数据库的相似度阈值必须根据实际数据调SQL 模糊查询在数据量大时性能骤降需要提前上全文索引。下一步可以做三件扩展一是给智能体增加“多轮对话记忆”让用户能追问查询结果二是接入“用户借阅记录”实现“根据我借过的书推荐相似的”三是把回答评估加入反馈循环统计用户标注“没有帮助”的比例反过来优化工具描述和检索阈值。如果这篇文章对你有帮助建议收藏备用。后面遇到智能体开发、RAG 检索、SQL 与向量库协同的问题可以随时回来翻一翻也欢迎在评论区交流你的落地经验。 数字图书管理员：SQL与向量数据库融合的AI智能体实战
这次我们来看一个很典型的 AI 智能体落地场景数字图书管理员。它不是简单的图书检索系统而是把 SQL 关系型数据库和向量数据库组合起来让 LLM 自动完成“结构化查询 语义检索 回答生成”的完整链路。如果你正在做 AI 智能体开发或者想了解 SQL 与向量数据库协同工作流怎么落地这篇文章可以直接收藏。项目核心就三件事第一用 SQL 管理图书的结构化数据比如 ISBN、作者、出版年份、分类、库存状态第二用向量数据库管理非结构化语义比如书名含义、内容摘要、评论情感、正文片段第三通过大模型的 Function Calling 或工具调用能力把两类查询封装成智能体工具让 AI 根据用户问题自动选择查询方式。全文会用一套可落地的方案讲清楚数据模型设计、智能体路由逻辑、SQL 工具和向量检索工具的实现、API 接口封装、批量入库任务以及常见的故障排查。会给出可复制的 Python 和 SQL 示例核心代码都不长本地开发机就能跑通。1. 核心能力速览能力项说明项目类型AI 智能体应用 / RAG 检索增强生成系统核心功能图书结构化查询、语义检索、混合检索、智能问答技术栈Python、FastAPI、SQLite/PostgreSQL、ChromaDB/pgvector、LangChain 或自研 Agent数据库能力SQL 负责精确查询向量数据库负责语义检索智能体能力意图识别、工具路由、SQL 生成、结果融合、回答生成推荐部署环境8G 内存以上开发机CPU 可运行GPU 用于提升 LLM 推理速度显存占用取决于 LLM 服务需按实际模型版本测试启动方式命令行启动 Web 服务 / API 服务是否支持 API支持RESTful 接口是否支持批量任务支持可批量导入图书数据和向量化适合读者AI 应用开发者、RAG 系统开发者、图书管理系统开发者从材料看这个项目的重点不是复杂算法而是数据架构和工具编排能力。SQL 解决“精确”问题向量数据库解决“模糊”问题智能体解决“选择哪条路”的问题。2. 适用场景与使用边界数字图书管理员最适合以下场景图书馆、资料室的图书检索和借阅咨询。企业内部知识库的文档问答比如制度文件、操作手册。内容平台的书籍推荐和语义搜索。RAG 系统的教学演示适合作为智能体开发入门案例。它能解决的问题很明确用户问“《三体》是哪年出版的”“作者是谁”这种结构化问题直接走 SQL用户问“有没有类似《百年孤独》魔幻现实主义风格的书”这种语义问题走向量检索用户问“帮我找 2023 年以后出版、和人工智能相关的书”这种同时包含结构化条件和语义条件的就要走混合检索。使用边界也要说清楚。第一数字图书如果涉及版权保护内容只能建立索引和摘要不能直接存储和输出全文。第二涉及用户借阅历史、个人信息时要做好脱敏和权限控制。第三智能体自动生成 SQL 存在注入风险必须对输入和输出做双重校验。第四向量检索结果有概率性不能完全替代精确查询重要数据判断应该走 SQL 确认。3. 整体架构与工作流设计这个项目的工作流可以拆成四层。第一层是数据层。图书的元数据比如书名、作者、ISBN、分类、出版年份、库存量存放在关系型数据库里。图书的内容摘要、章节文本、评论、标签经过 Embedding 模型向量化之后存入向量数据库。两个库通过图书 ID 关联。第二层是工具层。智能体暴露几个工具search_books_sql用于精确查询search_books_vector用于语义查询search_books_hybrid用于混合查询。每个工具都封装成函数有明确的参数描述和返回值格式。第三层是智能体层。LLM 接收用户问题后先判断查询意图再决定调用哪个工具最后把工具返回的数据整理成自然语言回答。这一步可以用 LangChain 的 Agent 框架也可以自己写一个简单的路由函数。第四层是接入层。通过 FastAPI 提供 Web API前端可以是一个简单的聊天页面也可以接到企业微信、钉钉等 IM 机器人。下面是一个实用的工作流路由逻辑def route_query(user_query: str): rule_engine detect_structured_intent(user_query) semantic_engine detect_semantic_intent(user_query) if rule_engine and not semantic_engine: return sql elif semantic_engine and not rule_engine: return vector elif rule_engine and semantic_engine: return hybrid else: return llm_generatedetect_structured_intent可以检测用户问题中是否包含 ISBN、作者名、出版年份、分类名称等结构化字段。detect_semantic_intent可以检测是否包含“类似”“风格”“推荐”“相关”等语义类关键词。实际的智能体不会只靠关键词判断更可靠的方式是让 LLM 自己决定工具调用也就是 Function Calling。后面会给出代码示例。4. 环境准备与数据模型设计4.1 基础环境推荐环境清单如下具体版本需按实际项目调整操作系统Ubuntu 20.04 / CentOS 7 / Windows 10 均可。Python3.10 以上。数据库SQLite 适合单机测试PostgreSQL 适合生产。向量数据库ChromaDB 适合快速验证pgvector 适合和 PostgreSQL 统一管理Milvus 适合大规模生产。LLM 服务OpenAI API 兼容接口或本地部署的 Qwen、ChatGLM 等模型服务。安装基础依赖pip install fastapi uvicorn sqlalchemy chromadb openai pydantic如果使用 pgvector需要额外安装 PostgreSQL 扩展CREATE EXTENSION IF NOT EXISTS vector;4.2 图书表结构设计用 SQLite 做演示时图书主表可以这样建CREATE TABLE books ( id INTEGER PRIMARY KEY AUTOINCREMENT, isbn TEXT UNIQUE, title TEXT NOT NULL, author TEXT, publisher TEXT, category TEXT, publish_year INTEGER, stock INTEGER DEFAULT 0, summary TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_books_category ON books(category); CREATE INDEX idx_books_publish_year ON books(publish_year); CREATE INDEX idx_books_title ON books(title);这里summary字段可以放内容摘要向量表存的是 summary 的向量表示。生产环境中summary可能很长更适合放在独立的book_chunks表里按段落存储。4.3 向量数据表设计使用 ChromaDB 时集合设计如下import chromadb client chromadb.PersistentClient(path./chroma_data) collection client.get_or_create_collection( namebooks, metadata{hnsw:space: cosine} )集合中的每条记录包含id图书 ID对应 SQL 表中的主键。document摘要或正文片段。metadata包含 title、author、category、publish_year 等结构化字段。这样设计的好处是向量检索结果可以直接返回元数据不需要每次回表查 SQL。但需要注意向量库里的元数据是冗余副本更新图书信息时要同步更新。4.4 智能体表结构设计如果要做用户提问日志和反馈记录可以增加两张表CREATE TABLE query_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_query TEXT, route_type TEXT, answer TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE feedback_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_log_id INTEGER, rating INTEGER, comment TEXT );日志表不仅用于排查问题还可以作为后续微调智能体的训练数据。5. SQL 工具与向量检索工具实现5.1 SQL 查询工具智能体的第一个工具是search_books_sql负责把用户的自然语言转换为 SQL 并查询。这里推荐使用 LLM 生成 SQL 的方式但必须加一层安全校验。示例工具函数import sqlite3 import json def search_books_sql(query_params: dict): 根据结构化参数查询图书信息。 query_params 支持 isbn, title, author, category, publish_year, stock conn sqlite3.connect(library.db) cursor conn.cursor() conditions [] values [] if query_params.get(isbn): conditions.append(isbn ?) values.append(query_params[isbn]) if query_params.get(author): conditions.append(author LIKE ?) values.append(f%{query_params[author]}%) if query_params.get(category): conditions.append(category ?) values.append(query_params[category]) if query_params.get(publish_year): conditions.append(publish_year ?) values.append(query_params[publish_year]) if query_params.get(title): conditions.append(title LIKE ?) values.append(f%{query_params[title]}%) sql SELECT * FROM books if conditions: sql WHERE AND .join(conditions) sql LIMIT ? values.append(query_params.get(limit, 10)) cursor.execute(sql, values) rows cursor.fetchall() column_names [desc[0] for desc in cursor.description] conn.close() results [] for row in rows: results.append(dict(zip(column_names, row))) return json.dumps(results, ensure_asciiFalse)这里没有直接让 LLM 生成任意 SQL而是通过参数化查询的方式把用户请求先解析成结构化字段再拼接 SQL。这种方式能显著降低 SQL 注入风险。5.2 向量检索工具第二个工具是search_books_vector负责语义检索。用户输入一句话先做 Embedding再去向量库找最相近的图书。def search_books_vector(query_text: str, top_k: int 5): from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) response client.embeddings.create( modeltext-embedding-v1, inputquery_text ) query_embedding response.data[0].embedding results collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, metadatas, distances] ) books [] for i in range(len(results[ids][0])): books.append({ book_id: results[ids][0][i], title: results[metadatas][0][i].get(title), author: results[metadatas][0][i].get(author), distance: results[distances][0][i], snippet: results[documents][0][i][:100] }) return json.dumps(books, ensure_asciiFalse)向量检索的关键参数是distance也就是相似度分数。ChromaDB 默认按距离排序距离越小越相似。实际使用时要统计一下正常结果的分数分布设定一个阈值分数超过阈值的直接过滤掉避免无关结果混入。5.3 混合检索工具第三个工具是search_books_hybrid把 SQL 和向量检索结合起来。典型场景是“找一本人工智能类的书风格要通俗易读”。def search_books_hybrid(category: str, query_text: str, top_k: int 5): # 第一步SQL 按分类缩小范围 conn sqlite3.connect(library.db) cursor conn.cursor() cursor.execute( SELECT id, title, author, category, publish_year, stock FROM books WHERE category ? LIMIT 50, (category,) ) sql_rows cursor.fetchall() conn.close() allowed_ids [str(row[0]) for row in sql_rows] if not allowed_ids: return json.dumps([], ensure_asciiFalse) # 第二步嵌入查询文本 response client.embeddings.create( modeltext-embedding-v1, inputquery_text ) query_embedding response.data[0].embedding # 第三步在限定 ID 范围内检索 results collection.query( query_embeddings[query_embedding], n_resultstop_k, where{book_id: {$in: allowed_ids}}, include[documents, metadatas, distances] ) # 第四步结果融合排序 books [] for i in range(len(results[ids][0])): books.append({ book_id: results[ids][0][i], title: results[metadatas][0][i].get(title), category: category, distance: results[distances][0][i] }) books.sort(keylambda x: x[distance]) return json.dumps(books, ensure_asciiFalse)混合检索的融合策略有两种一种是先过滤再检索适合分类、年份、作者等结构化条件很强的情况另一种是分别检索再合并排序适合两者权重相等的场景。实际项目里建议先做 A/B 测试确定哪种策略更符合业务需求。6. 智能体路由与 Function Calling 实现上面三个工具都封装好了接下来要让 LLM 自动选择调用哪个工具。最稳定的方式是 Function Calling。6.1 工具声明tools [ { type: function, function: { name: search_books_sql, description: 根据结构化条件查询图书信息适用于作者、ISBN、分类、出版年份、库存储备等精确查询, parameters: { type: object, properties: { title: {type: string, description: 书名}, author: {type: string, description: 作者}, category: {type: string, description: 图书分类}, publish_year: {type: integer, description: 出版年份}, isbn: {type: string, description: ISBN}, limit: {type: integer, description: 返回条数默认10} } } } }, { type: function, function: { name: search_books_vector, description: 语义检索图书适用于推荐、相似风格、模糊描述等查询, parameters: { type: object, properties: { query_text: {type: string, description: 用户的语义描述}, top_k: {type: integer, description: 返回条数默认5} }, required: [query_text] } } } ]6.2 智能体主循环from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) def run_agent(user_query: str): messages [ {role: system, content: 你是数字图书管理员根据用户问题选择工具并回答问题。}, {role: user, content: user_query} ] response client.chat.completions.create( modelqwen2.5-7b-instruct, messagesmessages, toolstools, tool_choiceauto ) message response.choices[0].message if message.tool_calls: tool_call message.tool_calls[0] function_name tool_call.function.name arguments json.loads(tool_call.function.arguments) if function_name search_books_sql: result search_books_sql(arguments) elif function_name search_books_vector: result search_books_vector(arguments.get(query_text), arguments.get(top_k, 5)) else: result json.dumps({error: 未知工具}) messages.append(message) messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) final_response client.chat.completions.create( modelqwen2.5-7b-instruct, messagesmessages, toolstools, tool_choicenone ) return final_response.choices[0].message.content else: return message.content这套主循环就是智能体的核心框架。用户提问后LLM 判断是否需要调用工具如果需要就生成工具参数程序执行工具后把结果回传给 LLMLLM 再生成最终回答。这里的base_url可以指向 OpenAI API也可以指向本地部署的 vLLM、Ollama 等兼容服务。只需要把api_key换成实际密钥即可。7. API 接口与批量任务7.1 API 服务搭建用 FastAPI 把智能体封装成 HTTP 接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI(title数字图书管理员 API) class QueryRequest(BaseModel): query: str session_id: str default class QueryResponse(BaseModel): answer: str route_type: str app.post(/api/query, response_modelQueryResponse) async def query_books(req: QueryRequest): answer run_agent(req.query) return QueryResponse(answeranswer) app.get(/api/health) async def health_check(): return {status: ok}启动服务uvicorn main:app --host 0.0.0.0 --port 7860启动后可以直接在浏览器访问http://127.0.0.1:7860/docs查看 Swagger 文档调试接口非常方便。7.2 curl 调用示例curl -X POST http://127.0.0.1:7860/api/query \ -H Content-Type: application/json \ -d {query: 帮我找一本关于人工智能的科普书, session_id: test-001}7.3 Python 调用示例import requests url http://127.0.0.1:7860/api/query payload { query: 刘慈欣有哪些作品, session_id: test-002 } response requests.post(url, jsonpayload, timeout60) print(response.json())7.4 批量图书入库任务图书管理一个很大的需求是批量导入。假设有一份 CSV 文件格式如下isbn,title,author,publisher,category,publish_year,stock,summary 9787536692930,三体,刘慈欣,重庆出版社,科幻,2008,10,地球文明向宇宙发出的第一声啼鸣 9787536692947,三体2黑暗森林,刘慈欣,重庆出版社,科幻,2008,8,宇宙社会学核心设定批量入库脚本import csv import sqlite3 from openai import OpenAI client OpenAI(base_urlhttp://localhost:8000/v1, api_keyEMPTY) def batch_import(csv_path: str, batch_size: int 20): conn sqlite3.connect(library.db) cursor conn.cursor() with open(csv_path, r, encodingutf-8) as f: reader csv.DictReader(f) batch [] for row in reader: cursor.execute( INSERT INTO books (isbn, title, author, publisher, category, publish_year, stock, summary) VALUES (?, ?, ?, ?, ?, ?, ?, ?), (row[isbn], row[title], row[author], row[publisher], row[category], int(row[publish_year]), int(row[stock]), row[summary]) ) book_id cursor.lastrowid batch.append({ id: str(book_id), document: row[summary], metadata: { title: row[title], author: row[author], category: row[category] } }) if len(batch) batch_size: collection.upsert(ids[b[id] for b in batch], documents[b[document] for b in batch], metadatas[b[metadata] for b in batch]) batch [] if batch: collection.upsert(ids[b[id] for b in batch], documents[b[document] for b in batch], metadatas[b[metadata] for b in batch]) conn.commit() conn.close() print(批量导入完成)批量任务的注意点有两个。第一SQL 批量插入要放在一个事务里失败时能回滚。第二向量化调用要控制并发。如果 Embedding 服务能力有限一次请求太多容易超时建议按批次提交批次大小根据 Embedding 服务的吞吐能力调整。7.5 批量任务队列设计生产环境批量导入可能上万条数据不能直接在主线程循环建议用消息队列CSV 文件 - 读取任务 - 数据清洗 - SQL 入库 - 向量化 - 向量入库 - 完成标记可以用 Celery、Redis Queue 或者简单的 Python 多进程实现。关键是要做断点续传记录每本书的处理状态失败的书重新入队避免全部重跑。8. 功能测试与效果验证8.1 结构化查询测试测试用例测试项输入示例预期结果精确查询“《三体》作者是谁”返回刘慈欣条件筛选“2020年以后出版的科幻书”返回符合年份和分类的书库存查询“有哪些书库存大于5”返回库存大于5的书记录判断标准SQL 工具返回的数据和直接执行 SQL 的结果一致。这里可以重点观察 LLM 是否把用户问题正确解析成了结构化参数。8.2 语义检索测试测试用例测试项输入示例预期结果推荐书籍“有没有类似《活着》那种沉重现实的书”返回与苦难、现实题材相关的书模糊描述“讲外星人和宇宙文明的书”返回科幻类图书跨语言检索“books about machine learning”返回机器学习相关图书判断标准返回结果在语义上相关而不是简单的关键词匹配。如果结果不相关优先检查 Embedding 模型的质量和图书摘要的文本长度。8.3 混合检索测试测试用例测试项输入示例预期结果分类语义“人工智能方面适合入门读的书”返回 AI 分类下、语义上偏向入门、通俗的图书年份语义“2022年以后出版的有深度的历史书”返回 2022 年后出版、历史类、深度内容优先判断标准每条结果既要满足结构化条件又要满足语义相关性。两者都满足的排在前面。8.4 全链路稳定性测试连续向智能体发起 100 个不同类型的测试请求统计请求成功率。平均响应时间。工具调用正确率。最终回答的可读性。如果成功率低于 95%需要检查是哪一步出错。可能是 LLM 生成参数异常也可能是工具函数返回格式不符合预期。8.5 安全测试必须测试的边界场景1. SQL 注入测试输入 1; DROP TABLE books; -- 2. 越权查询测试输入涉及隐私的查询 3. 超长输入测试输入几万字的文本 4. 空查询测试输入空白字符安全测试的目的是确认智能体不会把危险 SQL 透传到数据库。参数化查询已经能挡住大部分注入攻击但 LLM 生成工具参数时仍有可能“突发奇想”所以数据库账号权限也很关键。给智能体专用的数据库账号只开放SELECT权限禁止DROP、DELETE、UPDATE等高危操作这是底线要求。9. 资源占用与性能观察这个项目的资源消耗主要来自四个部分LLM 服务、Embedding 服务、关系型数据库、向量数据库。LLM 服务占资源最大。如果本地部署 7B 模型显存占用通常在 6G 到 16G 之间具体需要看模型量化和推理框架。如果本地没有 GPU可以改用 API 调用方式本机只跑数据库和工具层开发机只需要 8G 内存。Embedding 模型相对轻量CPU 也能跑。以文本向量化为例百条级别的查询在 CPU 上是毫秒到几十毫秒级别不影响交互体验。SQLite 在单机测试时几乎无感知但生产环境一旦并发上来建议切换到 PostgreSQL并对高频查询字段建索引。向量数据库的性能取决于索引类型和召回数量。HNSW 索引响应快但占内存Flat 索引准确但速度慢需要按数据量测试。性能观察建议用命令# 查看 GPU 显存占用 nvidia-smi -l 2 # 查看本机内存和 CPU htop # 查看进程端口占用 lsof -i:7860如果发现响应变慢先定位瓶颈。如果 SQL 查询慢用EXPLAIN QUERY PLAN看是否走了索引。如果向量检索慢减小top_k或者给集合重建索引。如果整条链路慢优先看 LLM 推理耗时可以在智能体主循环里加日志记录每个环节的耗时。这里还可以补充一个“分级缓存”的思路。热门问题结果可以缓存到 Redis缓存键用用户问题的 Embedding 做相似度匹配命中缓存后直接返回可以节省大量 LLM 调用成本。10. 常见问题与排查方法问题现象可能原因排查方式解决方案智能体返回“不知道”路由判断错误没有调用工具打印 tool_calls 数据优化工具描述增加示例SQL 工具返回空结果查询条件太严格单独执行 SQL 验证放宽条件使用模糊匹配向量检索结果不相关Embedding 模型与业务不匹配抽样检查距离分数更换 Embedding 模型或重写摘要向量数据库连接失败服务未启动或路径不对查看启动日志确认 persist 路径重启服务混合检索结果排序混乱融合策略选择错误分开对比两次检索结果使用 RRF 或加权融合算法API 请求超时LLM 推理时间过长查看调用耗时开启流式输出设置合理超时时间批量导入中断单条数据格式异常查看错误日志增加异常捕获标记失败数据后续重试LLM 生成 SQL 参数错误工具描述不清晰检查 Function Calling 的 arguments增加参数示例提高温度设为 0端口冲突7860 被占用lsof -i:7860更换端口启动内存持续增长向量库加载数据过多查看内存曲线限制集合大小使用服务端向量库关于 SQL 慢查询这里多说一句。智能体生成的查询条件有时候会不带索引比如LIKE %关键词%这种写法在数据量大时就是全表扫描。优化方案是配合 SQLite 的 FTS5 全文索引或者 PostgreSQL 的 pg_trgm 扩展把模糊查询从全表扫描变成索引扫描。11. 最佳实践与合规提醒第一数据模型设计上图书表要预留created_at、updated_at字段向量表要用图书 ID 关联主表避免只存向量不存原文。因为后续如果有数据修正需要根据 ID 定位更新。第二关于 SQL 注入防护推荐策略是“参数化查询 白名单校验”。凡是输入到 SQL 的条件只允许出现在白名单字典里。LLM 生成参数前也要先过滤一次字段名确保没有DROP、DELETE、ALTER等关键字。第三向量数据更新要和 SQL 数据保持一致。最简单的做法是建立一个“待同步队列”图书主表更新后把记录 ID 写入队列另一个消费者负责更新向量库。第四版权合规是数字图书项目的高压线。项目的定位是“图书管理员”负责的是图书的元数据和摘要信息不是提供全文下载。如果要索引正文内容建议只对公开版权书籍或企业自有文档做全文索引。涉及用户上传的书籍或评论时也需要在用户协议里明确授权范围。第五个人隐私保护。图书借阅记录、用户查询历史、评分反馈都属于个人信息。生产环境要加密存储、按角色控制访问权限、定期导出删除。API 层需要增加身份认证不能用裸接口直接暴露内网服务。第六给所有外部依赖加版本锁定。Python 依赖用requirements.txt锁定版本Docker 部署时锁定镜像 tag。否则 Embedding 模型更新后已经入库的向量和新的向量可能不在同一个语义空间导致检索效果突然变差。12. 总结与下一步数字图书管理员这个项目最有价值的地方是把 SQL 的精确查询能力、向量数据库的语义检索能力、LLM 的工具调用能力组合成了一个完整的智能体工作流。它不是玩具项目而是很多企业知识库问答系统的简化版。搭好这套框架之后换一个应用场景只需要改数据表和工具函数。建议最先验证三件事第一用 Function Calling 让智能体正确识别 SQL 查询和向量查询第二混合检索的排序效果第三批量导入上万条数据后的系统稳定性。最容易踩的坑也提前说LLM 工具参数解析偶发混乱需要加异常重试向量数据库的相似度阈值必须根据实际数据调SQL 模糊查询在数据量大时性能骤降需要提前上全文索引。下一步可以做三件扩展一是给智能体增加“多轮对话记忆”让用户能追问查询结果二是接入“用户借阅记录”实现“根据我借过的书推荐相似的”三是把回答评估加入反馈循环统计用户标注“没有帮助”的比例反过来优化工具描述和检索阈值。如果这篇文章对你有帮助建议收藏备用。后面遇到智能体开发、RAG 检索、SQL 与向量库协同的问题可以随时回来翻一翻也欢迎在评论区交流你的落地经验。