每天面对大量书籍、读者借阅记录、库存信息的管理员,会发现一个现实困境:传统的 SQL 数据库擅长处理精确查询,比如“查 ISBN 为 X 的书籍在哪个书架”,但当读者问“有没有类似《三体》那样的科幻小说,最好节奏快一点”时,SQL 无能为力。关键词匹配永远无法理解“类似”和“节奏快一点”背后的语义。
这正是数字图书管理员 AI 智能体要解决的问题:把 SQL 的结构化查询能力和向量数据库的语义检索能力组合起来,形成一条完整的“意图理解 → 语义召回 → 精确过滤 → 业务校验”工作流。本文会从概念、架构、代码实现到排错思路,完整拆解这套协同工作流,并给出可直接复用的 Python 实现。如果你想动手做一个人 RAG 图书检索系统,或者想理解 SQL 与向量数据库为什么不是替代关系而是协作关系,这篇文章值得读完。
1. 这篇文章真正要解决的问题
在做图书管理类系统时,最常见的技术选型纠结是:用 SQL 还是用向量数据库?
如果只用 SQL 数据库(比如 MySQL、PostgreSQL、SQLite),你只能做精确匹配和简单的模糊查询。你可以写WHERE title LIKE '%三体%',但用户如果只记得“那本讲外星人入侵的中文科幻”,你就需要构造一长串关键词组合,效果还很不稳定。更麻烦的是,图书管理系统不只有检索,还有借阅状态、库存位置、借还记录这些强结构化数据。这些数据放在向量数据库里并不合适。
如果只用向量数据库(比如 ChromaDB、Milvus、Qdrant),语义检索确实解决了,但你要回答“这本书现在能不能借”的时候,向量数据库帮不上忙。它本质上擅长“找相似”,而不是管理事务和关联关系。
一个合格的“数字图书管理员”智能体,需要同时掌握两种能力:理解用户语义,走向量检索;执行精确业务判断,走 SQL 查询。两者的关系不是“二选一”,而是按工作流编排。真正值得研究的问题不是“向量数据库能不能取代 SQL”,而是“SQL 和向量数据库如何在一条工作流里各司其职”。
本文要解决的问题就是这套协同机制:
- 什么时候走 SQL,什么时候走向量检索;
- 两者如何编排成一条完整工作流;
- 如何在一个实际项目里用代码跑通;
- 运行中容易出现哪些坑,如何排查。
2. SQL 与向量数据库的核心概念与差异
先建立共同语言。我们要把两个数据库放进同一条工作流,至少要理解它们的定位差异。
SQL 数据库SQL(结构化查询语言)数据库存储的是具有固定模式(Schema)的表数据。它的最大优势是:关联查询、事务、约束、精确匹配、聚合统计。比如“统计 2024 年借阅次数最多的 10 本书”,这是典型的 SQL 工作。你定义好表结构,MySQL 会保证数据的一致性和查询的高效性。
向量数据库向量数据库存储的是嵌入向量(Embedding)——也就是把文本、图片等内容转换成的数值数组。它解决的问题是“语义相似度检索”。比如把“类似《三体》的科幻小说”转换为向量,再在书库的标题、简介、评论向量里找最相似的若干条。这种检索不依赖关键词完全匹配,而是依赖语义接近程度。
两者不是同一个维度上的替代品
| 维度 | SQL 数据库 | 向量数据库 |
|---|---|---|
| 核心能力 | 精确查询、事务、关联分析 | 语义相似度检索 |
| 数据组织方式 | 表 + 行 + 列,强 Schema | 集合 + 向量 + 元数据 |
| 适合场景 | 借阅记录、库存、业务交易 | 内容推荐、模糊检索、AI 问答 |
| 查询方式 | SQL 语句 | 向量距离计算(余弦、欧氏等) |
| 数据一致性 | 强一致 | 大多是最终一致或依赖外部同步 |
| 对 AI 的适配 | 需要额外处理语义 | 天然支持 Embedding |
从这张表可以看出:图书管理员的日常业务数据(读者、库存、借阅记录)天然适合 SQL 数据库;而“读者用自然语言描述自己想看的书”这一需求,天然需要向量数据库。两者协同的关键在于,用工作流把“语义理解”和“业务判断”串起来。
工作流的含义
在这里,工作流不是一个图形化拖拽工具,而是一段可编排的代码逻辑:接收用户意图 → 分解任务 → 调用不同数据库 → 汇总结果 → 返回给用户。你可以用简单的 Python 函数实现,也可以用 LangChain、Dify 这类框架编排得更复杂,但核心套路是一样的。
3. 系统架构设计:数字图书管理员智能体如何工作
一个典型的数字图书管理员 AI 智能体,架构上可以拆成四层:
第一层:用户交互层接收读者提问,可能是精确查询,也可能是模糊描述。例如:
- “帮我查一下《深入理解计算机系统》在哪个书架”
- “有没有轻松一点的散文集,适合睡前看”
- “我想借 Python 数据分析相关的书,有现货吗”
第二层:意图识别与任务拆解层用大模型(LLM)判断用户的查询类型:
- 精确查询:提取书名、作者、ISBN,交给 SQL;
- 语义查询:提取描述性内容,交给向量检索;
- 混合查询:既要语义相似,又要检查库存状态,需要两个数据库联合。
第三层:数据访问层这一层包含两个客户端:
- SQL Client:连接 MySQL/SQLite,执行精确查询;
- Vector Client:连接 ChromaDB/Milvus/Qdrant,执行相似度检索。
第四层:业务聚合层将 SQL 结果和向量结果合并,过滤掉已借出或下架的书籍,按相关性排序,生成最终回答。
这套架构设计的关键判断是:不要把业务数据(库存、借阅状态)塞进向量数据库,也不要把语义检索逻辑硬编码成 SQL。两种数据库承担不同职责,各取所长。
用工作流图来理解,流程是这样的(注意不是 Mermaid,是文字描述任务流):
用户问题 → LLM意图识别 → 精确查询:SQL 查询书籍元数据 → 返回结果 → 语义查询:向量召回 TopK → 返回结果 → 混合查询:向量召回候选 → SQL 过滤状态 → 排序返回从这个流程可以看到,“SQL 与向量数据库协同”不是简单地把两个查询结果拼在一起,而是按业务需要编排调用顺序。语义查询负责扩大范围,SQL 负责收紧精度。
4. 环境准备与基础配置
开始实践之前,需要准备环境和依赖。先强调一下:本文的代码以通用思路为主,具体版本以实际项目为准。不要盲目追求最新版本,稳定可用优先。
4.1 基础运行环境
建议使用 Python 3.9 或更高版本。需要安装的 Python 依赖如下:
pip install openai chromadb sqlalchemy fastapi uvicorn pydantic如果使用 OpenAI 兼容接口,还需要配置环境变量。如果不方便调用大模型接口,可以先使用固定规则完成意图识别,不影响理解整个工作流。
4.2 SQL 数据库准备
本文示例使用 SQLite 作为 SQL 数据库,原因是零配置、单文件,适合做最小演示。实际生产环境一般使用 MySQL 或 PostgreSQL,连接方式只是换一下连接串。
创建数据库文件的步骤后面会讲。这里只需要确认 Python 环境里能执行sqlite3相关代码即可。SQLAlchemy 是 ORM 层,用于管理数据库连接和表结构。
4.3 向量数据库准备
选择 ChromaDB 做演示,因为它是本地文件型向量数据库,安装简单,不需要单独部署服务。如果你使用 Milvus 或 Qdrant,思路相同,只是客户端调用方式不同。
ChromaDB 安装成功后,创建持久化目录:
import chromadb client = chromadb.PersistentClient(path="./library_vector_db") collection = client.get_or_create_collection("books")这里要注意:PersistentClient会把向量数据持久化到本地目录,重启后数据不丢失。这个目录要加入版本管理忽略列表,不要提交到 Git。
4.4 Embedding 模型选择
向量检索的效果取决于 Embedding 模型。如果使用 OpenAI 接口,需要准备 API Key:
export OPENAI_API_KEY=your_key_here如果无法使用云端模型,可以选择本地 Embedding 模型,比如text2vec或m3e,效果略降但可控。为了演示,本文留出一个get_embedding接口,你可以按自己的模型替换实现。
5. 数据库表结构与向量集合设计
这一节开始落地。我们的核心目标是:把图书元数据放在 SQL 表里,把语义向量放在向量集合里,两边通过一个统一的book_id关联。
5.1 SQL 表结构:books 表
创建books表,包含书籍的基本信息、库存状态和位置信息。
-- 文件路径:schema.sql CREATE TABLE IF NOT EXISTS books ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, author TEXT NOT NULL, isbn TEXT, category TEXT, description TEXT, location TEXT, status TEXT DEFAULT 'available', created_at DATETIME DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_books_title ON books(title); CREATE INDEX IF NOT EXISTS idx_books_category ON books(category);这里几个字段的作用:
title、author、isbn用于精确匹配;category用于过滤图书分类;description用于生成向量;location存放书架位置;status标记available(可借)或borrowed(已借出)。
设计判断:向量搜索时应看重title和description,SQL 过滤时看status和category。不要把status这种频繁变化的值放进向量数据库,否则每次借还书都要更新向量,成本高且容易不一致。
5.2 SQLAlchemy 建表代码
使用 SQLAlchemy 在 Python 中建表:
# 文件路径:db.py from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker DATABASE_URL = "sqlite:///library.db" engine = create_engine(DATABASE_URL, echo=False) Base = declarative_base() class Book(Base): __tablename__ = "books" id = Column(Integer, primary_key=True, autoincrement=True) title = Column(String(255), nullable=False) author = Column(String(100), nullable=False) isbn = Column(String(50)) category = Column(String(50)) description = Column(Text) location = Column(String(50)) status = Column(String(20), default="available") # 初始化表 Base.metadata.create_all(engine) # 创建会话 SessionLocal = sessionmaker(bind=engine) session = SessionLocal()5.3 向量集合设计
在 ChromaDB 中创建books集合,并指定 metadata 字段用来做过滤:
# 文件路径:vector_store.py import chromadb client = chromadb.PersistentClient(path="./library_vector_db") collection = client.get_or_create_collection( name="books", metadata={"hnsw:space": "cosine"} )hnsw:space指定为cosine,表示使用余弦相似度计算。如果书籍数量很大,也可以使用l2,但需要根据 Embedding 模型的情况调节阈值。
设计判断:向量集合中不保存库存状态等易变业务数据,但可以保存category和book_id。这样我们可以用向量检索先召回候选,再根据book_id去 SQL 里查准确状态。
6. 核心工作流实现:数据入库与向量化
系统运行的第一步,是把已有图书数据写入两个数据库。这个步骤通常是一次性初始化和增量更新结合。
6.1 定义 Embedding 接口
先定义一个获取向量的函数。如果你配置了 OpenAI API,可以用官方 Embedding 接口;否则用本地模型替代。
# 文件路径:embedding_service.py import os from openai import OpenAI client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) def get_embedding(text: str) -> list: """ 将文本转换为向量。 生产环境中可以替换为本地模型或私有化部署的 Embedding 服务。 """ text = text.replace("\n", " ") resp = client.embeddings.create( model="text-embedding-ada-002", input=text ) return resp.data[0].embedding注意:不同 Embedding 模型的向量维度可能不同。入库和查询时,必须使用同样的模型,否则向量无法比较。这是新手最容易踩的坑。
6.2 数据入库完整流程
这里设计一个入库函数:先把书籍信息写入 SQL,再把文本向量写入 ChromaDB。
# 文件路径:ingest.py from db import Book, session from vector_store import collection from embedding_service import get_embedding def add_book(title, author, category, description, location, isbn=None): """ 添加一本新书,同时写入 SQL 和向量数据库。 """ # 第一步:写入 SQL 数据库 book = Book( title=title, author=author, category=category, description=description, location=location, isbn=isbn, status="available" ) session.add(book) session.commit() session.refresh(book) # 第二步:生成向量 text_for_embedding = f"书名:{title},作者:{author},分类:{category},简介:{description}" vector = get_embedding(text_for_embedding) # 第三步:写入向量数据库 collection.add( ids=[str(book.id)], embeddings=[vector], documents=[text_for_embedding], metadatas=[{"category": category, "book_id": book.id}] ) return book.id关键逻辑:
- 先写 SQL 得到自增
id; - 用文本生成同一套 Embedding;
ids使用字符串形式的book_id,保证两侧数据能关联。
这里有一个容易忽视的问题:SQL 写入和向量写入不是原子的。如果向量写入失败,SQL 里会多出一条“幽灵书籍”。解决办法是加入补偿逻辑:捕获异常后回滚 SQL 记录,或者使用本地消息队列做异步重试。演示阶段可以简化,但生产环境必须考虑。
6.3 批量导入
实际图书系统不可能一本一本添加。批量导入时,建议分批处理,避免一次性生成太多向量导致内存溢出。
# 文件路径:batch_ingest.py def batch_add_books(books: list): """ 批量添加图书。 books: [{"title": ..., "author": ..., "category": ..., "description": ..., "location": ...}] """ for book_data in books: try: add_book(**book_data) except Exception as e: print(f"导入失败: {book_data.get('title')},错误: {e}") # 实际项目中,这里应该记录失败日志,后续补偿重试批量导入时,建议每 50~100 本做一次展示进度,方便观察状态。
7. 工作流核心实现:SQL 与向量检索的协同查询
数据入库之后,核心工作流才算真正展开。这里设计三种查询模式,对应读者真实提问。
7.1 精确查询:走 SQL
当用户给出明确的书名或 ISBN 时,走纯 SQL 查询即可,不需要向量检索。
# 文件路径:search_service.py from db import Book, session def search_by_sql(title: str = None, author: str = None, category: str = None): """ 精确条件查询书目信息。 注意:这里使用了参数化查询,避免 SQL 注入风险。 """ query = session.query(Book) if title: query = query.filter(Book.title.ilike(f"%{title}%")) if author: query = query.filter(Book.author.ilike(f"%{author}%")) if category: query = query.filter(Book.category == category) results = query.all() return [ { "id": b.id, "title": b.title, "author": b.author, "category": b.category, "location": b.location, "status": b.status } for b in results ]安全提示:这里使用ilike时,看似在拼接%,但实际上title变量是作为参数传给 SQLAlchemy 的,不会产生 SQL 注入问题。如果是手写 SQL,务必使用?占位符,绝不直接拼接用户输入。
7.2 语义查询:走向量检索
当用户用自然语言描述“像《百年孤独》那样魔幻现实主义的书”时,走向量检索。
# 文件路径:search_service.py from vector_store import collection from embedding_service import get_embedding def search_by_vector(query_text: str, top_k: int = 5): """ 基于向量相似度的语义检索。 """ query_vector = get_embedding(query_text) results = collection.query( query_embeddings=[query_vector], n_results=top_k ) output = [] for i in range(len(results["ids"][0])): book_id = results["ids"][0][i] distance = results["distances"][0][i] metadata = results["metadatas"][0][i] output.append({ "book_id": book_id, "score": distance, "category": metadata.get("category") }) return output向量检索的输出只是候选和相似度分数,业务状态(是否借出、在哪个书架)还需要去 SQL 查询。
7.3 混合查询:先向量召回,再 SQL 过滤
这是整个“协同工作流”的核心场景。读者问:“我想找讲深度学习的书,最好现在就能借。” 这个需求已经超出了单纯的语义检索,因为“现在就能借”是业务条件,要求从向量召回结果中过滤出status = 'available'的。
# 文件路径:search_service.py from db import Book, session def search_hybrid(query_text: str, top_k: int = 10, only_available: bool = False): """ 混合检索流程: 1. 向量数据库语义召回候选集; 2. 根据 book_id 去 SQL 数据库查询精确状态; 3. 按业务条件过滤和排序。 """ # 第一步:向量召回 vector_results = search_by_vector(query_text, top_k=top_k) # 第二步:提取 book_id 列表 candidate_ids = [int(item["book_id"]) for item in vector_results] if not candidate_ids: return [] # 第三步:去 SQL 查询候选集的详细状态 books = session.query(Book).filter(Book.id.in_(candidate_ids)).all() book_map = {b.id: b for b in books} # 第四步:合并结果并进行业务过滤 merged_results = [] for item in vector_results: book = book_map.get(int(item["book_id"])) if not book: continue if only_available and book.status != "available": continue merged_results.append({ "id": book.id, "title": book.title, "author": book.author, "category": book.category, "location": book.location, "status": book.status, "similarity_score": item["score"] }) # 第五步:按相似度分数排序 merged_results.sort(key=lambda x: x["similarity_score"]) return merged_results这段代码是整个工作流的枢纽,理解它就能理解 SQL 与向量数据库协同的本质:向量数据库负责“理解语义”,SQL 数据库负责“确认事实”。两者通过book_id关联,形成一个管道:召回 → 过滤 → 排序 → 输出。
7.4 意图识别:智能体如何决定走哪条分支
要让智能体自动决定使用哪种检索方式,最简单的方式是用大模型做意图分类。提示词可以这样设计:
# 文件路径:intent_service.py def detect_intent(user_query: str) -> str: """ 根据用户问题判断检索模式。 返回:exact / semantic / hybrid """ # 实际项目可用 LLM 做意图分类 prompt = f""" 判断下面这个图书查询请求的意图类型: 1. exact:包含明确书名、作者或ISBN,目的是精确查找。 2. semantic:描述风格、主题、感觉,没有精确条件。 3. hybrid:既有内容描述,又有库存状态要求。 用户问题:{user_query} 请只返回 exact、semantic 或 hybrid。 """ # 调用大模型接口 # response = llm.chat(prompt) # 这里为了演示,用简单规则代替 if any(keyword in user_query for keyword in ["能借吗", "有货", "位置", "ISBN", "作者"]): return "hybrid" if any(keyword in user_query for keyword in ["像", "类似", "那种", "一点"]): return "semantic" return "exact"生产环境中,推荐用大模型做更细粒度的意图识别和槽位提取(例如抽出书名、作者、分类、借阅状态要求),然后填充到查询参数里。这里的规则版本只是为了展示工作流的完整性。
7.5 统一查询入口
最后,把这些函数封装成一个统一的查询入口,供 API 层调用:
# 文件路径:agent_service.py from search_service import search_by_sql, search_hybrid from intent_service import detect_intent def query_book(user_query: str): """ 数字图书管理员智能体统一入口。 """ intent = detect_intent(user_query) if intent == "exact": # 精确查询走 SQL results = search_by_sql(title=user_query) return {"intent": "exact", "results": results} if intent == "semantic": # 语义查询走向量检索 results = search_hybrid(user_query, top_k=5, only_available=False) return {"intent": "semantic", "results": results} # hybrid 模式下,既做语义召回,又过滤借阅状态 results = search_hybrid(user_query, top_k=10, only_available=True) return {"intent": "hybrid", "results": results}这个统一入口的设计意义在于:上层 API 不需要关心内部走的是 SQL、向量还是混合流程,智能体自己完成路由。
8. API 层实现与运行验证
为了让上面的工作流可被外部系统调用,我们用 FastAPI 封装一个 REST API。
8.1 FastAPI 服务代码
# 文件路径:api.py from fastapi import FastAPI from pydantic import BaseModel from agent_service import query_book app = FastAPI(title="数字图书管理员 AI 智能体") class QueryRequest(BaseModel): question: str class QueryResponse(BaseModel): intent: str results: list @app.post("/api/query", response_model=QueryResponse) def handle_query(req: QueryRequest): """ 图书查询接口。 请求示例: POST /api/query {"question": "我想找一本关于AI的科普书,最好现在能借"} """ result = query_book(req.question) return result @app.get("/health") def health_check(): return {"status": "ok"}8.2 运行服务
uvicorn api:app --host 0.0.0.0 --port 8000启动后,可以用 curl 验证:
curl -X POST http://localhost:8000/api/query \ -H "Content-Type: application/json" \ -d '{"question": "我想找一本讲深度学习的书"}'预期输出:返回一条 JSON,包含intent字段和results数组。如果intent是semantic,说明走了向量检索分支;如果返回结果里有status字段,说明经过 SQL 业务过滤。
8.3 验证策略
判断工作流是否成功,可以从三个层面验证:
- 精确查询是否命中:输入完整书名,看是否返回正确的位置和状态。
- 语义查询是否命中:输入风格化描述,看是否召回内容相似的书籍。
- 混合查询是否过滤:先手动把某本书的
status改为borrowed,再通过混合查询看它是否被过滤掉。
手动修改状态的 SQL:
UPDATE books SET status = 'borrowed' WHERE id = 1;然后重新发起混合查询,如果返回结果里没有 id=1 的书,说明 SQL 过滤生效。
9. 常见问题与排查方法
在实际运行这套工作流时,容易遇到下面这些问题。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 向量检索返回空结果 | 向量集合为空,或集合名称错误 | 检查 ChromaDB 持久化目录是否有数据;打印collection.count() | 确认入库流程已执行;检查集合名称 |
| 相似度分数分布不合理 | Embedding 模型不一致,或文本没有预处理 | 对比入库和查询用的模型名;检查文本是否包含过多噪音 | 统一模型;清理文本 |
| 查询结果里书籍状态不准确 | 向量数据库与 SQL 数据库中 book_id 没有对应 | 对比 collection metadata 中的 book_id 与 SQL 表主键 | 入库时确保用同一个 id;建立同步补偿机制 |
| SQL 查询慢 | 缺少索引,或使用LIKE '%xxx%'导致全表扫描 | 查看执行计划;检查索引 | 对常用字段建索引;引入 Elasticsearch 或全文检索 |
| API 响应慢 | 向量检索耗时或 Embedding 调用耗时 | 分段打印耗时日志 | 对 Embedding 加缓存;批量向量召回 |
| 向量写入成功但 SQL 回滚 | 两个数据库写入不是原子操作 | 检查调用链中的异常处理 | 引入事务补偿或消息队列 |
一个很关键的排查思路是:把“语义检索问题”和“业务数据问题”分开排查。如果向量召回结果看起来合理,但最终结果不对,问题往往出在 SQL 环节;反之,如果向量召回结果本身就不相关,就应该检查 Embedding 模型和入库文本质量。
另外补充一点,如果你使用 ChromaDB 时出现网络或持久化异常,优先检查本地目录的读写权限。ChromaDB 的本地模式对中文支持良好,但在 Windows 系统上偶尔会遇到路径分隔符问题,建议统一使用绝对路径。
10. 最佳实践与工程建议
这套工作流看起来简洁,但进入生产环境后,有大量细节需要关注。
10.1 数据一致性优先设计
SQL 数据库和向量数据库是两套存储,天然存在数据一致性问题。建议遵循两个原则:
- 以 SQL 数据库为源数据(Source of Truth),向量数据库是 SQL 数据的“语义投影”。
- 更新顺序:先改 SQL,再重新生成向量并写入向量数据库。如果向量写入失败,启动补偿任务重试。
对于图书更新场景,可以维护一张同步状态表,记录每本书的向量版本号。SQL 更新时版本号 +1,后台任务发现向量版本落后时自动重建向量。
10.2 查询性能优化
- 向量召回数量不要贪多。
top_k一般取 10~20 就足够,SQL 再过滤掉不满足条件的记录。召回太多,后续 SQL 过滤的IN查询会很大。 - 书籍数量达到百万级时,需要考虑给 ChromaDB 或 Milvus 配置 GPU 或 HNSW 参数优化;SQL 层则使用分库分表或读写分离。
- Embedding 调用是耗时大户。对同一段高频查询文本,加一层 Redis 缓存。
# 伪代码:Embedding 缓存逻辑 def get_embedding_cached(text: str) -> list: cache_key = f"embedding:{hash(text)}" cached = redis.get(cache_key) if cached: return json.loads(cached) vector = get_embedding(text) redis.set(cache_key, json.dumps(vector), ex=3600) return vector10.3 提示词工程与意图识别
意图识别环节,建议不要只用简单规则。大模型能做到更好。但要注意:大模型的输出不稳定,需要把输出约束在固定枚举值内,同时做好异常兜底。例如大模型返回了不属于exact/semantic/hybrid的内容,默认走hybrid模式,而不是直接报错。
10.4 安全边界
- SQL 查询必须使用参数化查询或 ORM,防止 SQL 注入。
- 如果向量数据库部署为独立服务,务必配置访问认证,避免数据被未授权访问。
- 对话接口要加限流,防止恶意请求消耗大量 Embedding 算力。
- 日志中不要记录完整的用户问题,涉及个人隐私的信息需要脱敏。
10.5 可观测性
把智能体的每次查询记录成日志,至少包含:
- 用户原始输入;
- 识别出的意图;
- 向量召回条数;
- SQL 过滤掉多少条;
- 最终返回条数;
- 总耗时。
这组数据能帮助你分析系统瓶颈,也能定位用户从“提问模糊”到“结果不满意的原因”。可以选用 Prometheus + Grafana 做指标监控,也可以简单先用 JSON 日志加可视化工具。
11. 总结与后续深入学习方向
这一路下来,你应该能体会到 SQL 与向量数据库协同并不是一个抽象概念,而是一套可以落地的工程模式。数字图书管理员 AI 智能体的核心逻辑,可以总结为:用向量召回解决“语义理解问题”,用 SQL 过滤解决“业务事实问题”,两个环节通过统一 ID 关联,由意图识别层做路由。
如果要动手实践,建议按下面顺序走一遍:
- 准备好 Python 环境和依赖;
- 用 SQLite + ChromaDB 做最小入库;
- 实现纯 SQL 查询、纯向量查询、混合查询三种分支;
- 封装 FastAPI 接口;
- 改进意图识别,用大模型替换规则版本。
下一步值得深入的方向包括:
- 用大模型做书籍摘要和标签补充,提升向量检索质量;
- 引入 Redis 缓存,降低 Embedding 调用成本;
- 把工作流迁移到生产级向量数据库(Milvus、Qdrant、pgvector),并做好分片策略;
- 为图书管理员智能体增加还书提醒、借阅历史推荐等功能,扩展 SQL 业务能力。
技术选型上没有银弹。SQL 和向量数据库不是竞争关系,而是互补关系。谁能把这两者按业务场景编排好,谁就能做出一套真正理解用户需求的智能系统。建议把本文的核心代码收藏起来,动手搭一个最小版本,比反复看概念有效得多。