这次我们来看一个比较典型的 AI 智能体落地场景:数字图书管理员。它不只是一个聊天机器人,而是一个真正把 SQL 和向量数据库结合起来做协同工作流的智能体系统。
先说结论:这个场景最大的价值,是把它拆开后你可以看懂 AI 智能体项目里最常用的两种数据通路——结构化数据走 SQL,非结构化内容走向量检索,再通过 LLM Agent 做意图理解和路由。如果你正在做 AI 智能体开发、RAG 检索增强生成,或者想把图书、文档、知识库类项目做得更可用,这篇文章可以直接收藏。
文章会依次拆解:核心架构、数据库设计、Agent 工作流实现、功能测试、API 接口设计、资源占用观察和常见问题排查。整个演示不依赖特定云服务,本地用开源组件就能搭出一版可运行的原型。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目类型 | 数字图书管理 AI 智能体,结构化数据与语义检索协同 |
| 核心组件 | LLM Agent + SQL 数据库 + 向量数据库 + Embedding 模型 |
| SQL 负责 | 书籍元数据、馆藏状态、借还事务、读者信息、统计报表 |
| 向量库负责 | 图书内容语义检索、相似推荐、自然语言问答 |
| Agent 负责 | 用户意图识别、查询路由、工具调用、结果汇总 |
| 推荐硬件 | 本地 CPU 可跑通流程,LLM 和 Embedding 推理建议 GPU 加速 |
| 支持平台 | Windows / Linux / macOS 均可,需按组件分别安装 |
| 启动方式 | 命令行启动,各服务独立运行 |
| API 能力 | 可通过 HTTP 接口封装查询、入库、同步任务 |
| 批量任务 | 支持批量书籍入库、批量索引同步、批量元数据更新 |
| 适合读者 | AI 智能体工程师、RAG 开发者、图书馆/知识库系统设计者 |
需要说明:这个项目的显存占用、具体接口路径和启动脚本并不是固定的,取决于你选择的 LLM、Embedding 模型和数据库版本。下面所有示例都是通用的架构演示,落地时需要按实际选型调整。
2. 整体架构:SQL 与向量数据库为什么要协同
数字图书管理员要处理的用户请求,天然分为两类。
一类是精确查询和事务操作,比如“《三体》还剩几本可借”“帮我借一本书,ISBN 是 978-7-302-xxxx”“这个月借阅量最高的是哪类书”。这类请求对准确率要求极高,不能用模糊检索来回答,必须落到 SQL 数据库上做精确关联和聚合统计。
另一类是语义检索,比如“找几本讲时间旅行的书”“这本书和《百年孤独》风格类似吗”。用户不会精确记忆书名和分类号,但你想要的答案隐藏在图书内容的语义特征里。这时候需要把书籍摘要、目录甚至正文片段向量化,存进向量数据库,用相似度检索召回。
问题来了:如果只用 SQL,语义检索会变得很笨。你只能依赖标题、标签、分类号做 LIKE 查询,遇到“时间旅行”这种不在标题里的语义就失效了。如果只用向量数据库,精确计数、事务更新、统计报表又做不了,向量库本身不适合做强一致性的结构化写入。
所以这里的协同工作流是:
- LLM Agent 接收用户自然语言请求。
- Agent 判断任务类型,决定调用 SQL 工具还是向量检索工具。
- SQL 层负责所有结构化数据和事务。
- 向量层负责内容语义召回。
- 更复杂的场景会把两步串联,先向量召回候选,再用 SQL 做条件过滤和排序。
- Agent 汇总结果,组织成用户可读的答案。
这个模式在材料里被称为“协同工作流”,本质上是把两种数据库定位成不同职责的组件,而不是互相替代。
2.1 查询路由设计
核心是让 Agent 学会“分诊”。最常见的做法是在 Agent 的工具描述里写清楚每个工具的适用场景,让 LLM 自己选择:
tools = [ { "name": "query_sql", "description": "用于查询书籍精确元数据、馆藏数量、借还状态、统计数据。当用户提到具体书名、ISBN、作者、分类号、借阅统计时使用。", "parameters": ["sql"] }, { "name": "search_vectors", "description": "用于按语义搜索图书内容,适合模糊描述、主题检索、相似推荐。当用户用自然语言描述感兴趣的内容时使用。", "parameters": ["query", "top_k"] } ]这里的关键点在于,工具描述不要写成含糊的“查询数据库”,而是要写成带明确触发条件的路由规则。实测下来,描述越具体,Agent 路由准确率越高。
3. 适用场景与使用边界
这套架构适合三类场景:
- 图书馆、资料室、企业知识库的智能检索入口,用户可以用自然语言查书、问内容、做推荐。
- 电商或内容平台的商品/文档检索,需要同时支持精确筛选和语义召回。
- RAG 项目的进阶版,在普通向量检索之上叠加结构化数据过滤。
需要注意的使用边界:
- 向量检索本身是“近似搜索”,不能代替 SQL 做精确计数和账务类操作。
- Agent 自动生成 SQL 存在一定风险,必须限制数据库账号权限,只允许只读或受控写入。
- 涉及读者借阅记录、个人信息时,必须做脱敏和权限控制,不能把隐私数据暴露给 LLM。
- 图书内容如果受版权保护,向量索引只能用于内部检索和个人学习测试,不能把全文内容未经授权地对外提供服务。
- 不要用这套架构处理需要强事务保证的金融、医疗核心业务,除非你额外引入完善的补偿机制和审计机制。
4. 环境准备与前置条件
搭建这个项目不需要特别夸张的硬件,但组件比较多。建议按下面的清单准备,避免装到一半才发现缺东西。
- 操作系统:Windows 10/11、Ubuntu 20.04+ 或 macOS 12+ 都可以。
- Python 版本:建议 3.10 或 3.11,兼容性最稳。
- LLM 推理环境:可以使用 OpenAI 兼容接口的在线模型,也可以部署本地模型。本地模型建议至少 8GB 显存,实际占用取决于模型尺寸。
- Embedding 模型:用于把文本转成向量,常见有基于 SentenceTransformer 的开源模型,CPU 也能跑,但速度慢。
- SQL 数据库:推荐 SQLite 做原型验证,PostgreSQL 做生产。如果使用 PostgreSQL,可以直接装 pgvector 插件同时支持向量检索,减少一个组件。
- 向量数据库:可选 ChromaDB、Milvus、Qdrant。原型阶段 ChromaDB 最简单,数据量大了再切 Milvus。
- Docker:如果你不想在宿主机装一堆依赖,可以用 Docker 隔离数据库组件。
- 磁盘空间:SQL 数据通常很小,向量索引和模型文件需要预留 5GB 到 20GB。
检查列表:
# Python 版本检查 python --version # CUDA 检查(使用本地 GPU 推理时) nvidia-smi # Docker 检查 docker --version5. 数据库设计与数据同步
这一节是整套工作流的地基。图书管理员的业务数据,我用两张核心表来演示。
5.1 SQL 表结构设计
CREATE TABLE books ( id INTEGER PRIMARY KEY AUTOINCREMENT, isbn TEXT UNIQUE, title TEXT NOT NULL, author TEXT, category TEXT, language TEXT, publish_year INTEGER, location TEXT, total_copies INTEGER DEFAULT 1, available_copies INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE borrow_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id INTEGER, reader_id TEXT, borrow_date TIMESTAMP, return_date TIMESTAMP, status TEXT DEFAULT 'borrowed', FOREIGN KEY (book_id) REFERENCES books(id) );这里我把书籍的基本信息、馆藏数量、借还状态放在 SQL 里。所有需要精确判断的操作,比如“可借数量减一”,都必须走 SQL 事务:
BEGIN; UPDATE books SET available_copies = available_copies - 1 WHERE id = ? AND available_copies > 0; INSERT INTO borrow_records (book_id, reader_id) VALUES (?, ?); COMMIT;这个写法能避免并发借书时把库存扣成负数,是 SQL 层必须守住的核心能力。
5.2 向量索引设计
向量库存储的是“内容语义”,也就是每本书的嵌入向量。为了控制成本,入库时我建议对每本书生成三类向量片段:
- 书名与作者组合向量。
- 书籍简介向量。
- 目录或关键章节摘要向量,可选。
每一段向量都关联 book_id,方便后面回表查询 SQL 元数据。ChromaDB 中的集合结构大致如下:
import chromadb client = chromadb.PersistentClient(path="./library_db") collection = client.get_or_create_collection( name="book_contents", metadata={"hnsw:space": "cosine"} ) collection.add( ids=["book_1_seg_title", "book_1_seg_desc"], embeddings=[[0.01, 0.02, ...], [0.03, 0.01, ...]], # 实际由 embedding 模型生成 metadatas=[ {"book_id": 1, "segment_type": "title"}, {"book_id": 1, "segment_type": "description"} ], documents=["《三体》刘慈欣", "地球文明与三体文明的信息交流故事"] )embedding 向量的具体维度取决于你选的模型,常见是 384、768 或 1024 维,不需要手动指定死。
5.3 数据同步任务
每次新增书籍,系统要同时更新 SQL 和向量库,这是最容易出错的地方。推荐实现一个同步任务模块:
- 在 SQL 中插入 books 记录,拿到 book_id。
- 调用 Embedding 模型生成向量。
- 把向量写入向量数据库。
- 如果第 3 步失败,需要回滚第 1 步插入,或进入重试队列。
- 全部成功后才算完成入库。
另一种做法是把向量索引构建做成异步任务,主流程先写 SQL 返回成功,后台队列再补向量。这种方式响应快,但会出现短暂的数据不一致,需要在读取时做兼容。
6. Agent 工作流实现
接下来是核心部分:让 Agent 能真正回答用户的自然语言问题。这里我给出一个不依赖特定框架的参考实现,核心逻辑是 ReAct 模式:思考 → 调用工具 → 观察结果 → 继续或输出。
6.1 工具函数封装
import sqlite3 import chromadb def query_sql(sql): conn = sqlite3.connect("library.db") conn.row_factory = sqlite3.Row cursor = conn.cursor() cursor.execute(sql) rows = [dict(row) for row in cursor.fetchall()] conn.close() return rows def search_books_semantic(query, top_k=5): client = chromadb.PersistentClient(path="./library_db") collection = client.get_or_create_collection(name="book_contents") results = collection.query( query_texts=[query], n_results=top_k ) return results注意,query_sql 这里只是演示。实际项目中千万不能直接把用户输入拼接进 SQL,要使用参数化查询,并对 LLM 生成的 SQL 做白名单校验。
6.2 安全校验
SQL 注入是绕不开的问题。AI 智能体自动生成 SQL 放大了这个风险,必须做三层防护:
import re ALLOWED_SQL_PATTERN = re.compile(r"^(SELECT|WITH)\s", re.IGNORECASE) BLOCK_KEYWORDS = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "ATTACH"] def validate_sql(sql: str): if not ALLOWED_SQL_PATTERN.match(sql): raise ValueError("只允许执行 SELECT 查询") for kw in BLOCK_KEYWORDS: if kw in sql.upper(): raise ValueError(f"禁止包含关键字 {kw}") return sql更稳妥的方案是给数据库单独建一个只读账号,从账号权限层面杜绝写操作。
6.3 工作流编排
我给出一个最简单的 Agent 执行逻辑:
def book_agent(user_input: str): # 第一步:让 LLM 决定调用哪个工具 plan = llm_route(user_input, tools) if plan["tool"] == "query_sql": sql = plan["sql"] validate_sql(sql) rows = query_sql(sql) return llm_generate_answer(user_input, rows) if plan["tool"] == "search_vectors": results = search_books_semantic(user_input, top_k=5) book_ids = extract_book_ids(results) # 第二步:用 SQL 回查这些候选书的精确信息 placeholders = ",".join("?" * len(book_ids)) sql = f"SELECT * FROM books WHERE id IN ({placeholders})" books = query_sql_params(sql, book_ids) return llm_generate_answer(user_input, books)这个流程里最有价值的一点是:向量检索的结果不会直接作为最终答案,而是作为候选集,再用 SQL 回表拿到完整的元信息。比如用户说“找几本时间旅行的书”,向量库召回 5 本候选书,SQL 再把这 5 本书的馆藏状态和位置查出来,最后 Agent 告诉用户“这三本在架可借,两本已借出”。这就是协同工作流的关键收益。
7. 功能测试与效果验证
搭建完成后,建议按以下顺序逐项验证。
7.1 精确查询测试
测试输入:
请帮我查一下《三体》有多少本可借。预期行为:
- Agent 应该调用 query_sql 工具。
- 生成类似
SELECT title, available_copies FROM books WHERE title LIKE '%三体%'的查询。 - 返回数字和书名。
判断标准:结果中必须包含准确的剩余数量,而不是“大约”“可能”。
7.2 语义检索测试
测试输入:
想找一些探讨人类记忆和身份认同的书。预期行为:
- Agent 调用 search_vectors 工具。
- 向量库召回与“记忆、身份认同”语义相近的书,即使书名里没有这些词。
判断标准:召回结果与输入语义相关,而不是简单按关键词匹配。
7.3 混合查询测试
测试输入:
找几本2020年后出版的机器学习相关书籍,只要中文的。这是最容易翻车的场景。模型需要先把“机器学习相关”这条语义条件交给向量检索,再把“2020年后出版”“中文”这两条结构化条件交给 SQL 过滤。
实际工作中我建议用固定 pipeline 而不是完全交给 LLM 自由发挥:
- 用 Embedding 模型对用户输入做关键词/语义拆分提示。
- 向量检索召回候选 20 本。
- 把候选 book_id 传给 SQL,用
publish_year >= 2020 AND language = '中文'过滤。 - 返回过滤后的结果。
7.4 批量入库测试
准备一个 CSV 文件,包含书名、作者、ISBN、分类、简介等字段,执行批量入库脚本:
import csv with open("books.csv", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: add_book_with_embedding(row)判断标准:CSV 中所有书籍在 SQL 和向量库中都存在,且数量一致。
7.5 失败场景测试
故意输入超长文本、乱码、未收录的书名,观察 Agent 是否会出现幻觉。建议在 System Prompt 中明确要求:“如果数据库中没有匹配结果,直接说未找到,不要编造书籍信息。”这一步必须由人工复核结果。
8. 接口 API 与批量任务设计
如果要把能力开放给前端页面或其他系统,建议封装一个轻量 HTTP 服务。使用 FastAPI 是最快的方式。
8.1 API 入口示例
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): question: str top_k: int = 5 class AddBookRequest(BaseModel): isbn: str title: str author: str = "" category: str = "" description: str = "" publish_year: int = None @app.post("/api/query") def handle_query(req: QueryRequest): return book_agent(req.question, top_k=req.top_k) @app.post("/api/books/add") def add_book(req: AddBookRequest): return add_book_with_embedding(req.dict()) @app.post("/api/books/sync") def sync_index(): return run_index_sync()8.2 curl 调用示例
curl -X POST http://127.0.0.1:8000/api/query \ -H "Content-Type: application/json" \ -d '{"question": "找一些关于人工智能历史的书", "top_k": 5}'8.3 Python 调用示例
import requests resp = requests.post( "http://127.0.0.1:8000/api/query", json={"question": "找一些关于人工智能历史的书", "top_k": 5}, timeout=120 ) print(resp.json())8.4 批量任务建议
批量任务一定要满足三个工程要求:
- 幂等性:同一本书重复同步不会产生重复向量和重复记录。
- 失败重试:建议用队列,任务失败后进入重试队列,最多重试 3 次。
- 日志追踪:每条任务记录 book_id、状态、错误信息、耗时。
{ "task_id": "sync_20250101_001", "batch_name": "第一批入库", "total": 1000, "success": 998, "failed": 2, "failed_items": [ { "book_id": "isbn_9787302xxxx", "error": "embedding 模型调用超时" } ] }批量任务的口径是:宁可失败显眼,不要静默吞掉错误。
9. 资源占用与性能观察
这个项目没有固定的显存占用指标,因为你可以选择不同大小的 LLM 和 Embedding 模型。但有几条通用的观察方法。
9.1 显存占用观察
如果你使用本地 GPU 跑 LLM:
nvidia-smi -l 1重点观察两个阶段:
- Embedding 批量入库时,显存占用通常不高,但 CPU 和磁盘 IO 会成瓶颈。
- LLM 推理回答时,显存占用会明显爬升,实际数字与模型参数量和上下文长度强相关。
9.2 主要性能瓶颈
- Embedding 模型推理速度:批量入库时最耗时,建议用 GPU 或并行 batch。
- 向量检索速度:数据量在百万级以下,ChromaDB 也能满足;量级增长后建议切 Milvus。
- 慢 SQL:最常见的坑是对 title 字段做 LIKE '%关键词%',无法走索引。生产环境建议增加全文索引或与向量检索配合减少扫描量。
- LLM 推理延迟:影响问答体验的决定性因素,建议流式输出。
9.3 降低资源占用的建议
- 向量入库时控制 batch_size,避免一次处理过多文本导致内存暴涨。
- 检索时固定 top_k,不要无限制返回。
- 对常用元数据查询增加缓存,减少 LLM 重复生成 SQL 的开销。
- LLM 上下文里只放必要的结果摘要,不要把整本书内容都塞进 prompt。
10. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| Agent 一直生成 SQL 而不是向量检索 | 工具描述不够清晰 | 打印 Agent 路由日志,查看它对工具的理解 | 调整工具 description,明确触发条件 |
| 向量检索结果与问题完全无关 | Embedding 模型能力不足或文本切分不合理 | 单独测试某一段文本的相似度 | 更换更强的 Embedding 模型,优化文本切块 |
| SQL 查询报语法错误 | LLM 生成的 SQL 不符合方言 | 记录生成的 SQL 并人工核对 | 在 prompt 中提供表结构和示例 SQL |
| 图书已入库但搜索不到 | 向量库同步失败或索引未构建完成 | 检查同步任务日志 | 重跑 sync 任务 |
| 批量入库中途卡住 | 单批 size 过大或模型服务超时 | 查看任务队列 | 减小 batch_size,增加超时时间 |
| 并发借书时库存变负数 | SQL 缺少条件更新和事务 | 检查更新语句 | 使用WHERE available_copies > 0条件更新 |
| API 服务响应慢 | LLM 推理串行且无缓存 | 观察请求耗时分布 | 增加并发队列和缓存层 |
| 数据库账号被注入风险 | LLM 生成的 SQL 未安全校验 | 开启 SQL 审计日志 | 仅授予只读权限,参数化查询 |
| 向量库和 SQL 数据不一致 | 双写缺少补偿机制 | 比对两库数量 | 增加同步任务和一致性校验脚本 |
11. 最佳实践与使用建议
11.1 先跑通最小闭环
不建议一上来就接大模型和分布式数据库。第一次实验建议用:
- 100 本书的本地测试数据。
- SQLite 做 SQL 存储。
- ChromaDB 做向量存储。
- 最小的开源 Embedding 模型。
- 通过模拟 LLM 或真实 LLM API 串联。
这个最小闭环能验证路由、双写、检索、回表查询四条核心链路是否顺畅。
11.2 目录与数据管理
建议目录分三层:
library-agent/ ├── data/ │ ├── sql/ # SQLite 文件 │ ├── vector/ # ChromaDB 持久化目录 │ └── raw/ # 原始书籍元数据和文本 ├── scripts/ │ ├── ingest.py # 批量入库 │ ├── sync.py # 索引同步 │ └── agent.py # Agent 主逻辑 ├── logs/ └── tests/ # 测试用例与测试数据11.3 合规与安全红线
最后要强调几条不能越过的红线:
- 数字图书内容如果要向量化检索,需确认版权授权范围,只做摘要检索,不对外传播全文。
- 读者借阅记录属于个人隐私,数据库必须加密存储,接口必须鉴权。
- Agent 生成的 SQL 必须限制为只读,写操作只能通过业务接口执行,不能直接暴露给模型。
- 涉及人脸、声音、肖像或版权素材的类似项目,必须在合法授权前提下测试。
- 发布或商用前,要人工复核一批测试问题,防止 LLM 编排错误导致错误结果流出。
12. 总结与下一步
这个项目的核心结论很明确:AI 智能体不是“一个模型打天下”,而是让模型学会编排不同的数据组件。SQL 负责精确、事务、统计,向量数据库负责语义、召回、推荐,LLM 负责把用户的自然语言翻译成一组工具调用,再把结果组织成人话。
最值得先验证的功能是三类:精确查询是否准、语义检索是否相关、混合查询是否能把两步链路串起来。最容易踩的坑是数据不一致和 SQL 注入,建议在架构层面提前加补偿任务和权限隔离。
下一步可以扩展的方向包括:增加多轮对话记忆,让图书管理员能记住用户的借阅偏好;引入更细粒度的权限体系,区分管理员和读者视图;把 Retrieval 链路升级为混合检索加重排,提升推荐质量;如果数据量超过百万级,把向量库迁移到 Milvus,并接入专业的任务队列服务。
建议收藏备用,实际搭建时按这里的环境清单、表结构和测试用例一步步过,遇到问题回到第 10 节的排查表对照处理即可。