SQL与向量数据库协同实践:构建数字图书管理员AI智能体
2026/9/21 20:12:19 网站建设 项目流程

如果你正在做一个“图书管理”类产品,很可能会有一种很纠结的体验。用传统的关系型数据库(SQLite、MySQL、PostgreSQL)管理图书,优点是借阅记录、库存扣减、ISBN 精确检索非常可靠,但一遇到“帮我找一本讲分布式系统、但不要太偏理论的入门书”这种语义模糊的请求,SQL 就会陷入“关键词猜谜”的困境。反过来,如果你把所有图书信息全部塞进向量数据库,虽然有了语义搜索能力,但随之而来的是事务处理、多条件筛选、统计报表这些基础能力变得别扭甚至不可用。

这个矛盾的背后,是一个在 AI 应用开发里被反复讨论、却很少有人真正拆开讲清楚的问题:SQL 数据库和向量数据库不是替代关系,而是互补关系;真正决定体验的,是两者之间那条“协同工作流”怎么设计。

这篇文章要写的,是一个我称为“数字图书管理员”的 AI 智能体。它不依赖某一个“超级模型”,而是一套组合:用 SQL 负责精确事实和事务,用向量数据库负责语义理解和相似推荐,再用 Agent 工作流把两者编排起来。我们会从需求痛点出发,完成数据库建模、环境搭建、核心代码、运行验证、常见问题排查和工程最佳实践。读完你可以直接照着搭一个最小可运行的版本,也能理解如何在更大规模的项目里复用这套设计。

1. 这篇文章真正要解决的问题

先说一个非常实际的场景。假设你有一个小型图书馆,藏书几千本,读者会在小程序里提问:

  • “《深入理解计算机系统》有库存吗?”
  • “作者是卡斯特罗的那本讲操作系统的书,在哪个书架?”
  • “最近有哪些 AI 相关的书可以借?”
  • “我刚看完《SQL 必知必会》,有没有类似但更深入的书推荐?”

看前三个问题:它们都带有精确条件,书名、作者、分类、库存状态。这类查询极适合 SQL,因为答案必须“一个都不能错”。再看最后一个问题:“类似但更深入”。这里没有唯一的正确答案,需要理解用户的真实意图,并根据图书内容之间的语义距离做推荐。这恰恰是向量数据库擅长的事情。

如果只用 SQL,第三个问题只能靠关键词匹配;如果只用向量数据库,前两个问题会因为“向量检索的软匹配特性”产生错误——也许返回一本作者名字差不多的书,但这不是用户想要的。

所以我的判断是:数字图书管理员的本质,不是“用一个数据库替代另一个数据库”,而是通过工作流把两种数据库作为各自擅长领域里的工具,统一暴露给 AI Agent 调用。这篇文章的读者,不需要已经熟悉向量数据库,但最好具备基本的 SQL 知识和一点 Python 经验。你将学会:

  1. 如何为图书管理场景设计 SQL 表和向量集合。
  2. 如何理解 SQL 和向量数据库各自的能力边界。
  3. 如何用 Agent 工作流把两者编排成一次自然语言查询的完整链路。
  4. 如何验证体系是否可用,以及生产落地时的常见坑。

2. 基础概念:SQL、向量数据库与 Agent 工作流

很多人第一次听到“向量数据库”时,容易把它想成一个特别神秘的东西。实际上,它的核心抽象非常朴素:把一段文本、一张图或一条记录通过 Embedding 模型转换成一串浮点数,即“向量”,然后在这串数字的数学坐标空间里做“相似度排名”。

2.1 SQL 数据库的定位

SQL 数据库处理的是结构化数据,核心能力是精确查询。典型特点:

  • 强约束:主键、外键、唯一索引、非空约束都能保证数据的一致性。
  • 事务:通过 ACID 特性,保证一借一还、库存扣减这类操作要么全部成功,要么全部失败。
  • 聚合统计:GROUP BYCOUNTSUM等能力适合报表。
  • 成熟度极高:所有主流语言都有成熟驱动,运维工具链完整。

在图书管理员的场景里,图书的基本档案、读者档案、借阅流水、库存数量,这些都属于 SQL 的“管辖范围”。它们要求确定性:同一本书的 ISBN 不能变,可借数量必须是精确数字。

2.2 向量数据库的定位

向量数据库解决的是“语义理解”问题。它的存储对象是 Embedding 向量,查询方式不是WHERE条件,而是“找到与输入向量最接近的 K 个向量”。

拿 ChromaDB、Milvus、pgvector、Qdrant 做对比,它们各有侧重:

数据库运行方式典型适用场景与 SQL 的关系
ChromaDB轻量级、本地文件、开发友好原型验证、小规模语义检索独立运行,需要自行同步
Milvus分布式、高并发、海量向量生产级大厂、百万级向量独立集群,需要数据同步管道
pgvectorPostgreSQL 扩展Postgres 生态内做向量检索与 SQL 同库,事务一致性较易保证
QdrantRust 实现、过滤能力强需要复杂 metadata 过滤的检索独立服务,与 SQL 互补

这里要先说透一个容易踩坑的点:向量数据库不是“可以替代 SQL 的下一代数据库”,而是“对非精确匹配检索能力的一种扩展”。在做数字图书管理员时,我不建议你幻想“把所有书都向量化,然后用相似度回答一切问题”,因为图书馆系统里但凡涉及数量、日期、状态、权限的查询,向量检索的误差是无法接受的。

2.3 Agent 工作流与传统 API 调用的区别

所谓工作流,在 AI Agent 语境下,指的是“意图理解、路由决策、工具调用、结果组装”的编排过程。它不是写死的if-else,也不是简单的“调完大模型再调数据库”,而是把每一次用户请求看成一次任务,让一个主控模块决定该调用哪些工具、按什么顺序调用、怎么合并结果。

传统 API 调用往往是你知道固定的数据接口,比如GET /books?keyword=xx。但 AI Agent 工作流多了一层“自由对话 → 结构化工具调用”的转换。例如用户说“帮我找一本零基础能看懂的机器学习书”,Agent 需要先理解“零基础”这个约束,再决定是调用 SQL 做分类过滤,还是调用向量检索做语义匹配。

在这个项目里,工作流会比 Flowable、ComfyUI 这类偏“重流程编排”的系统更轻,也更接近 Dify、n8n 里的 AI Agent 节点式工作流思路。我们先用代码原生实现一套最简路由,后续很容易迁移到成熟的 AI 工作流平台。

3. 数字图书管理员的系统架构设计

在写代码之前,最好先把整体架构在脑子里画清楚。数字图书管理员并不复杂,但它要求每个模块各司其职。

3.1 模块划分

整个系统分成四层:

  1. 用户交互层:接收自然语言问题,并把答案组织成可读文本。
  2. 意图路由层:也叫 Agent 主控,决定请求走 SQL、走向量,还是两者融合。
  3. 工具层:SQL 工具、向量检索工具,各自封装为 Agent 可调用的函数。
  4. 数据层:SQLite + ChromaDB,或者任何 SQL 与向量数据库的组合。

这种分层的好处在于:当你想把 SQLite 换成 MySQL,或者把 ChromaDB 换成 Milvus,只需要改工具层内部的实现,意图路由层不需要大规模改动。

3.2 工作流如何协同两种数据库

我把核心工作流绘制成一个顺序链路,但这里不用时序图,用文字描述:

输入问题进入意图路由。意图路由的决策结果有三类:

  • 精确查询:书名、作者、ISBN、库存状态、借阅流水,走 SQL。
  • 语义检索:模糊描述、相似书籍推荐、按内容主题找书,走向量数据库。
  • 混合查询:既要求精确条件(如“AI 分类下有库存的书”),又要求语义排序(如“和《SQL 必知必会》内容相近”),则需要先 SQL 过滤,再向量排序,或先向量召回,再 SQL 过滤。

这里的核心判断是:协同不是“先执行一个、再执行另一个”那么简单,而是要设计好过滤条件放在哪一侧。例如:

  • SQL 有库存且分类属于 AI,然后在向量库里对候选集做语义排序。这种方式适用于候选集已经很小的情况。
  • 先用向量库召回 TOP 100 相似书籍,再用 SQL 判断其中哪些有库存。这种方式适用于语义性比较强、但精确约束比较弱的场景。

实际项目里,我建议默认采用“SQL 过滤 + 向量排序”,因为 SQL 过滤后的结果集通常可以控制在一个合理范围内,避免向量库在大候选集上做不必要的计算。

3.3 数据同步问题

一个必须直面的问题是:书加进系统时,要同时写入 SQL 表和向量集合。如果只用 SQL 表做事实层、用向量集合做语义索引,那两者之间的一致性就依赖写入方。最稳妥的做法是,在add_book这个业务方法里,同时完成 SQL 插入和向量写入,并把“向量写入失败”当作业务失败处理,保证要么都成功、要么都失败。这样虽然不能解决全量同步,但在最小系统里已经足够。

4. 环境准备与前置依赖

为了把精力集中在核心逻辑上,我们选择一套轻量、无需额外服务端的组合:

  • 操作系统:Windows / macOS / Linux 均可。
  • Python:3.9 或更高版本,建议 3.10+。
  • SQL 数据库:SQLite,Python 内置,零安装。
  • 向量数据库:ChromaDB,使用本地持久化文件。
  • Embedding 模型:先用 ChromaDB 默认的all-MiniLM-L6-v2或兼容的本地模型。如果你希望中文效果更好,后续可以替换为中文 Embedding 模型或远程 embedding API。

4.1 创建项目目录

mkdir digital-librarian cd digital-librarian

4.2 安装依赖

pip install chromadb

如果网络环境受限,无法自动下载模型文件,可以先配置国内镜像,或手动把模型文件下载到本地缓存目录。注意,对于生产项目,更推荐把 Embedding 服务和主服务解耦,统一通过 API 调用。

4.3 初始化代码结构

digital-librarian/ ├── data/ │ ├── library.db │ └── chroma_data/ ├── agent/ │ ├── __init__.py │ ├── librarian.py │ └── workflow.py ├── main.py └── README.md

data目录存放 SQLite 文件和向量数据目录。agent目录放 Agent 核心逻辑。main.py是命令行交互入口。

5. SQL 与向量数据库的首次建模

这一节动手建两个“数据底座”。先用 SQLite 建一张图书表、一张借阅流水表,再在 ChromaDB 中创建向量集合。

5.1 SQL 表结构设计

agent/librarian.py中初始化 SQL 连接并建表:

# 文件路径:agent/librarian.py import sqlite3 import uuid def get_connection(db_path: str = "data/library.db") -> sqlite3.Connection: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row return conn def init_sql_tables(conn: sqlite3.Connection) -> None: conn.executescript( """ CREATE TABLE IF NOT EXISTS books ( book_id TEXT PRIMARY KEY, title TEXT NOT NULL, author TEXT, isbn TEXT, category TEXT, publication_year INTEGER, total_copies INTEGER DEFAULT 1, available_copies INTEGER DEFAULT 1, location TEXT ); CREATE TABLE IF NOT EXISTS borrowers ( borrower_id TEXT PRIMARY KEY, name TEXT NOT NULL, email TEXT, registered_date TEXT DEFAULT CURRENT_DATE ); CREATE TABLE IF NOT EXISTS borrow_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, book_id TEXT NOT NULL, borrower_id TEXT NOT NULL, borrow_date TEXT DEFAULT CURRENT_DATE, return_date TEXT, status TEXT DEFAULT 'borrowed' ); """ ) conn.commit()

这里要注意几点:

  • book_id使用 UUID 字符串,避免自增主键在数据迁移时的冲突。
  • borrow_records记录每位读者每本书的借阅流水,业务上不能只靠books.available_copies一个数字,否则无法追溯历史。
  • 真正查询“某本书是否可借”时,以available_copies > 0为准。

5.2 ChromaDB 向量集合设计

在同一文件里初始化 ChromaDB:

import chromadb from chromadb.utils import embedding_functions chroma_client = chromadb.PersistentClient(path="data/chroma_data") embedding_fn = embedding_functions.DefaultEmbeddingFunction() collection = chroma_client.get_or_create_collection( name="book_content", embedding_function=embedding_fn, )

这里使用PersistentClient,向 ChromaDB 的早期版本写法做了一点区分:新版推荐PersistentClient,数据会持久化到本地目录。向量集合的“文档”建议同时包含书名、作者、分类和内容摘要,这样用户输入“一本讲 XX 主题的书”时,向量检索能在更完整的语义上下文里做匹配。

5.3 添加一本书的协同写入

新增图书时,SQL 与向量必须同步。我们封装一个add_book方法,把两步写进同一个业务操作:

def add_book( conn: sqlite3.Connection, collection, title: str, author: str, isbn: str, category: str, publication_year: int, total_copies: int, location: str, summary: str, ) -> str: book_id = str(uuid.uuid4())[:8] conn.execute( """ INSERT INTO books (book_id, title, author, isbn, category, publication_year, total_copies, available_copies, location) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (book_id, title, author, isbn, category, publication_year, total_copies, total_copies, location), ) conn.commit() doc_text = ( f"title: {title}\n" f"author: {author}\n" f"category: {category}\n" f"summary: {summary}" ) collection.upsert( ids=[book_id], documents=[doc_text], metadatas=[{"title": title, "category": category, "author": author}], ) return book_id

这个方法的意义在于:事实数据(可借数量、位置、ISBN)交给 SQL,语义索引(摘要、主题、分类描述)交给向量库。真正决定系统可用性的,不是单个数据库的能力,而是这两个写入动作是否总是一起完成。如果你在项目里用了事件消息队列,也可以用异步方式同步,但最小系统里同步写入更简单可控。

6. 核心工作流实现:意图路由与混合检索

有了数据层,接下来实现 Agent 工作流的“大脑”。

6.1 路由判断

为了演示通用思路,我先把路由设计成基于规则的轻量实现。这样做的好处是不增加 LLM 调用成本,适合最小的可运行系统;在生产环境里,你可以把这个位置替换成一个 LLM 意图识别函数,让它输出 JSON 格式的路由决策。

核心路由逻辑如下:

def route_query(query: str) -> str: exact_keywords = ["库存", "作者", "出版社", "ISBN", "分类", "在哪", "能不能借", "还有没有"] semantic_keywords = ["推荐", "类似", "有关", "入门", "深入", "适合", "讲什么", "主题"] has_exact = any(k in query for k in exact_keywords) has_semantic = any(k in query for k in semantic_keywords) if has_exact and has_semantic: return "hybrid" if has_exact: return "sql" if has_semantic: return "vector" return "sql"

这个路由很粗,但足够说明问题。真实系统里,你应该让 LLM 来输出结构化决策,而不是维护一堆中文关键词。因为用户提问方式千变万化,规则很难覆盖。

6.2 SQL 工具

路由到 SQL 分支时,执行精确查询。这里要点是:所有拼接 SQL 值的地方必须使用参数化查询,避免出现安全问题。例如:

def search_books_sql(conn: sqlite3.Connection, keyword: str): sql = """ SELECT * FROM books WHERE title LIKE ? OR author LIKE ? OR isbn LIKE ? OR category LIKE ? ORDER BY title """ pattern = f"%{keyword}%" rows = conn.execute(sql, (pattern, pattern, pattern, pattern)).fetchall() return [dict(row) for row in rows]

不要写出f"SELECT * FROM books WHERE title LIKE '%{keyword}%'"这类字符串拼接。这是常识,但在 AI Agent 生成 SQL 的代码里反而经常被忽略。Agent 生成 SQL 时,如果最终要执行,也必须经过严格的参数注入检测和最小权限授权。

6.3 向量检索工具

向量检索工具封装 ChromaDB 的查询逻辑:

def semantic_search(conn: sqlite3.Connection, collection, query_text: str, top_k: int = 5): results = collection.query(query_texts=[query_text], n_results=top_k) books = [] for i in range(len(results["ids"][0])): book_id = results["ids"][0][i] distance = results["distances"][0][i] row = conn.execute( "SELECT * FROM books WHERE book_id = ?", (book_id,) ).fetchone() if row: books.append({**dict(row), "similarity": round(1 - distance, 4)}) return books

这里距离越小越相似,我把它转换成similarity分数,便于后续展示。向量检索返回的候选集不一定都有库存,如果用户关心“能不能借”,你必须再用 SQL 做一次过滤。

6.4 混合检索工作流

混合检索是这次协同的核心。默认策略是“先 SQL 过滤,再向量排序”。比如用户问“AI 分类下有库存的、和《SQL 必知必会》类似的书”,流程如下:

def hybrid_search(conn, collection, query_text, category=None, only_available=True, top_k=5): # 第一步:SQL 精确过滤 sql = "SELECT * FROM books WHERE 1=1" params = [] if category: sql += " AND category = ?" params.append(category) if only_available: sql += " AND available_copies > 0" sql += " ORDER BY title" candidate_rows = conn.execute(sql, params).fetchall() if not candidate_rows: return [] candidate_ids = [row["book_id"] for row in candidate_rows] # 第二步:只对候选集做向量检索 results = collection.query( query_texts=[query_text], n_results=min(top_k, len(candidate_ids)), where={"book_id": {"$in": candidate_ids}}, ) ...

不过这里有一个需要注意的细节:ChromaDB 的where过滤依赖 metadata。如果刚才写入时没有在 metadata 里存book_id,过滤就会失效。所以前面的add_book最好把book_id也放入 metadata。这是一个典型的数据建模坑。

更新的写法是,把候选集向量结果重新映射回 SQL 记录:

def hybrid_search(conn, collection, query_text, candidate_ids, top_k=5): if not candidate_ids: return [] results = collection.query( query_texts=[query_text], n_results=min(top_k, len(candidate_ids)), ) id_to_rank = {book_id: idx for idx, book_id in enumerate(results["ids"][0])} sorted_ids = sorted(candidate_ids, key=lambda x: id_to_rank.get(x, len(candidate_ids))) books = [] for book_id in sorted_ids[:top_k]: row = conn.execute( "SELECT * FROM books WHERE book_id = ?", (book_id,) ).fetchone() if row: books.append(dict(row)) return books

这段代码的思路是:先召回向量库中全局最相近的一批书,再把它们和 SQL 过滤出的候选集做交集,最后按向量相似度排序。这种方式更加通用,避免把candidate_ids传到向量查询的where里造成不必要的限制。

6.5 工作流编排

工作流入口函数负责接收用户输入,调用路由,再调用不同工具:

def run_workflow(conn, collection, user_query: str): route = route_query(user_query) if route == "sql": books = search_books_sql(conn, user_query) return { "route": "sql", "books": books, "message": "通过 SQL 精确查询完成。", } if route == "vector": books = semantic_search(conn, collection, user_query) return { "route": "vector", "books": books, "message": "通过向量语义检索完成。", } # hybrid books = search_books_sql(conn, user_query) candidate_ids = [book["book_id"] for book in books] books = hybrid_search(conn, collection, user_query, candidate_ids) return { "route": "hybrid", "books": books, "message": "通过 SQL 过滤 + 向量排序完成。", }

这个实现已经是一个可运行的 Agent 工作流雏形。生产环境可以在此基础上增加 LLM 意图路由、多轮对话记忆、工具返回结果的结构化解析,以及失败重试机制。

7. 完整运行示例与效果验证

现在编译一个可执行入口,把我们前面的模块串起来。

7.1 写入演示数据

先准备几本书,方便测试:

# 文件路径:seed.py from agent.librarian import add_book, get_connection, init_sql_tables from agent.workflow import run_workflow import chromadb from chromadb.utils import embedding_functions conn = get_connection("data/library.db") init_sql_tables(conn) chroma_client = chromadb.PersistentClient(path="data/chroma_data") collection = chroma_client.get_or_create_collection( name="book_content", embedding_function=embedding_functions.DefaultEmbeddingFunction(), ) add_book( conn, collection, title="SQL必知必会", author="Ben Forta", isbn="9787111378080", category="数据库", publication_year=2010, total_copies=3, location="A-01-03", summary="SQL入门经典,适合零基础读者快速掌握查询、过滤、联结和子查询。", ) add_book( conn, collection, title="高性能MySQL", author="Baron Schwartz", isbn="9787111411657", category="数据库", publication_year=2013, total_copies=2, location="A-01-05", summary="MySQL性能优化的高阶读物,包含索引设计、慢查询分析、复制与扩展。", ) add_book( conn, collection, title="机器学习实战", author="Peter Harrington", isbn="9787115317956", category="AI", publication_year=2013, total_copies=2, location="B-03-01", summary="通过代码实践讲解机器学习算法,适合具备基础编程经验的读者。", )

如果 Embedding 模型下载较慢,程序会卡在这一步,这是正常的。第一次创建向量集合后会生成模型缓存,后续启动会快很多。

7.2 运行命令行交互

# 文件路径:main.py from agent.librarian import get_connection, init_sql_tables from agent.workflow import run_workflow import chromadb from chromadb.utils import embedding_functions def main(): conn = get_connection("data/library.db") init_sql_tables(conn) chroma_client = chromadb.PersistentClient(path="data/chroma_data") collection = chroma_client.get_or_create_collection( name="book_content", embedding_function=embedding_functions.DefaultEmbeddingFunction(), ) print("数字图书管理员已启动,输入问题开始查询,输入 exit 退出。") while True: user_query = input("你:").strip() if user_query == "exit": break result = run_workflow(conn, collection, user_query) print(f"\n路由分支:{result['route']}") if not result["books"]: print("没有找到相关图书。") continue for book in result["books"]: print( f"- {book['title']} | {book['author']} | " f"分类:{book['category']} | 可借:{book['available_copies']}" ) print() if __name__ == "__main__": main()

启动运行:

python main.py

7.3 预期效果

假设你输入“数据库相关的入门书”,意图路由会判定为“语义查询”或“混合查询”。如果路由结果是hybrid,它会先通过 SQL 找到分类含“数据库”或标题含“数据库”的候选书,再在向量库中做语义排序。最终输出应该优先出现《SQL必知必会》,因为它同时满足“入门”这个语义约束和“数据库”这个精确分类。

假设你输入“ISBN 9787111411657 的书”,路由会走sql,返回精确的一本《高性能MySQL》。

验证成功的关键判断标准是:精确问题回答不能错,语义问题回答要合理。如果发现精确问题被路由到了向量分支,多半是路由规则里精确关键词覆盖不全,或者用户输入里没有可识别的精确词,需要你不依赖路由就返回全量并兜底。

8. 常见问题与排查思路

任何涉及两个数据源的系统,坑通常都出现在“一致性”“查询性能”“依赖环境”这三个方向。

问题现象可能原因排查方式解决方案
向量检索结果为空数据没写入 collection,写入时 Embedding 失败检查 collection.count(),确认返回数量重新执行 add_book,查看模型下载日志
SQL 有数据,但向量检索查不到SQL 与向量写入不同步,向量集合缺失该记录对比 SQL books 表和 collection.count()用 book_id 做 key 定期校验同步
Semantic Search 结果不符合预期默认 Embedding 模型对中文/领域术语理解较弱测试不同提问,观察 Top5 结果换成中文 Embedding 模型,或改用远程 embedding API
启动很慢第一次加载 Embedding 模型观察日志,确认在下载模型权重提前预下载模型,设置本地缓存目录
混合检索候选集太大SQL 过滤太宽泛,或没有加上 only_available 条件打印候选集长度增加分类、年份、库存等强制过滤条件
数据存在,但 SQL LIKE 查不到用户输入了同义词或模糊表达检查录入值,确认分词方式多字段模糊匹配 + 向量召回兜底
生产环境使用 MySQL 后程序报错SQLite 的 SQL 方言与 MySQL 不一致查看错误日志中的 SQL 语句使用 ORM 或统一 SQL 方言层

这里提醒一点:所有涉及向量数据库和 SQL 数据库同步的场景,最优先要做的是“数据对账”。比如每天跑一次离线任务,找出 SQL 中存在但向量集合中缺失的记录,自动补写。这在生产环境里是必须的基础设施。

9. 最佳实践与工程建议

9.1 数据写入:先 SQL 后向量,并处理失败

在最小系统里,我们选择同步写入。但生产环境建议采用“写 SQL → 发消息 → 异步写向量”的方式。一方面减少用户请求的等待时间,另一方面允许向量索引系统独立伸缩。不过要注意:异步化会引入最终一致性,因此查询侧必须具备“向量查不到时回退 SQL 关键词搜索”的兜底逻辑。

9.2 检索策略:能用 SQL 过滤就不要全靠向量过滤

向量数据库的where过滤能力在近几年进步很大,但从架构和扩展性来看,把精确的、枚举型的约束交给 SQL,把语义相关性交给向量库,是最不容易出错的搭配。混合检索时,优先在 SQL 侧缩小候选集,再对候选集做语义排序。候选集控制在几百条以内,向量检索的速度和精度都会更稳定。

9.3 安全的几个优先事项

  • 参数化查询必须写进团队规范。即便 Agent 生成 SQL,也不能直接拼接用户输入。
  • 给“Agent 能执行的 SQL”设置严格权限。在 MySQL/PostgreSQL 里创建只读账号,禁止执行写操作,除非用户明确要求借书还书。
  • 向量数据库和 SQL 数据库都要放在内网服务,不直接暴露公网。
  • 所有借还书操作,必须走事务并在业务层记录操作人信息。

9.4 慢 SQL 与性能优化

如果某天系统用户量上来了,不要第一时间怀疑向量数据库,先看慢 SQL。给books.categorybooks.titlebooks.authorborrow_records.book_id建索引。对于 SQLite,可以使用EXPLAIN QUERY PLAN查看 SQL 执行计划;对于 MySQL,使用EXPLAIN。这个习惯能帮你避开“把所有问题都归咎于 AI 组件”的陷阱。

9.5 工作流平台化:从代码到 Dify/n8n

当你验证了这套最小工作流,下一步可以把它迁移到 Dify、n8n 等 AI 工作流平台。迁移时,只需把run_workflow拆成独立工具节点:一个 SQL 查询工具、一个向量检索工具、一个路由节点。在平台里配置好每个节点的输入输出字段,就可以获得可视化编排、历史回溯、版本管理、多用户并发等能力。核心算法和数据结构设计不变,变的只是编排外壳。

10. 总结与后续学习方向

这篇文章从“SQL 精确查询”和“向量语义检索”的矛盾出发,实现了一个名为“数字图书管理员”的最小 AI 智能体。核心结论可以浓缩成三句话:

  • SQL 负责精确事实,向量数据库负责语义理解,两者协同才能真正支撑自然语言图书管理。
  • 工作流的价值在于路由和编排,不是调用次数多,而是“在正确的地方调用正确的工具”。
  • 最小系统要同步写入、参数化查询、先 SQL 过滤再向量排序,这是最稳妥的默认策略。

如果你接下来想继续深入,可以按三个方向推进:

  1. 将意图路由从规则替换为 LLM 结构化输出,让系统支持更复杂的自然语言提问。
  2. 引入真实的大模型,让回答不只返回书单,还能生成推荐理由和对比说明。
  3. 把 SQLite + ChromaDB 替换为 PostgreSQL + pgvector,在同一个数据库实例里同时管理精确数据和向量数据,这会显著降低数据同步的复杂度。

数字图书管理员只是一个缩影,类似的“SQL + 向量 + 工作流”组合,完全可以复制到企业文档检索、商品推荐、工单分诊等场景。先把这个最小闭环跑通,再逐步替换组件,你会收获一套能适应变化的架构。建议收藏备用,实际动手时对照着参数和数据模型排查,比重新查资料快得多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询