Kotaemon数据库Schema解释：新人快速上手-智慧文博士

Kotaemon数据库Schema解释：新人快速上手

在企业级智能对话系统的开发中，一个常见的挑战是：如何让大语言模型（LLM）不仅“能说”，还能“说得准、有依据、可操作”。许多团队最初尝试直接微调模型来嵌入业务知识，结果却发现更新成本高、答案不可追溯、系统难以维护。这正是检索增强生成（Retrieval-Augmented Generation, RAG）架构兴起的背景——而Kotaemon正是在这一理念下构建的高性能、模块化智能体框架。

它不依赖黑盒式的模型记忆，而是通过一套清晰的数据结构设计，将知识检索、上下文管理和外部交互解耦为独立可维护的组件。其中，数据库 Schema 就是这套系统的骨架。理解这些 Schema，等于掌握了 Kotaemon 的数据流动逻辑与扩展边界。

数据即设计：从三个核心模块看系统运作

Kotaemon 的强大之处，在于它把 AI 系统的关键能力都映射到了具体的数据表结构上。每一个 Schema 都不只是存储容器，更是一种行为契约。我们不妨从最常接触的三个模块切入：文档检索、对话状态和工具调用。

文档不是“喂”给模型的，而是“查”出来的

很多人误以为 LLM 要回答专业问题就得先学会所有内容，但现实是——你没法实时重训模型去适应每天变化的政策或产品信息。Kotaemon 的做法更聪明：把知识库变成可搜索的向量空间。

当你上传一份《员工手册》PDF，系统会自动将其切分为多个语义完整的段落（chunk），每个 chunk 经过嵌入模型（embedding model）转换成高维向量，并存入向量数据库（如 Chroma 或 Pinecone）。与此同时，原始文本和元数据（来源、分类、更新时间等）也一并保存，形成“文本-向量-元数据”三位一体的记录。

这样一来，用户提问“年假怎么申请？”时，系统不会靠猜测作答，而是：

将问题编码为向量；
在向量空间中查找最相似的几个文档片段；
把这些片段作为上下文注入 prompt，供大模型参考生成答案；
同时返回引用来源，实现可审计性。

这种机制的优势非常明显。相比微调模型，它的更新成本几乎为零——只需刷新数据库即可；而且天然支持多租户隔离（比如不同部门看到不同的政策版本），只需要在查询时加上tenant_id过滤条件就行。

from llama_index import Document, VectorStoreIndex from llama_index.vector_stores import ChromaVectorStore import chromadb client = chromadb.PersistentClient(path="/db/chroma") collection = client.create_collection("knowledge_base") doc = Document( text="员工请假流程需提交至HR系统审批。", metadata={ "source": "employee_handbook_v3.pdf", "category": "HR", "updated_at": "2024-05-01" } ) vector_store = ChromaVectorStore(chroma_collection=collection) index = VectorStoreIndex.from_documents([doc], vector_store=vector_store)

这段代码看似简单，实则暗藏工程考量。比如metadata字段的设计就决定了后续能否做精准过滤。如果你只存了个文件名，后期想按“发布日期 > 2024 年”来筛选就很困难；但如果提前规划好字段结构，未来就能轻松支持动态知识路由。

还有一个容易被忽视的点是分块策略。太短的 chunk 可能丢失上下文，太长的又会影响检索精度。实践中建议结合句子边界和段落主题进行智能切分，甚至可以引入滑动窗口重叠机制，避免关键信息刚好落在两个 chunk 之间被割裂。

对话不是一次性的，而是有记忆的旅程

传统聊天机器人常常陷入“每轮对话都是新开始”的困境。你说“帮我订会议室”，它问你要时间；你说了时间，它又忘了你要订的事。原因很简单：没有持久化的状态管理。

Kotaemon 用一张sessions表解决了这个问题。每次用户发起对话，系统都会创建或加载对应的 session 记录，里面不仅保存了聊天历史，还包括当前意图（intent）、已填充槽位（slots）、流程进度等结构化信息。

想象这样一个场景：用户说“我要请三天假”，系统识别出意图是“申请年假”，并开始收集必要参数——起止日期、理由、审批人等。这些中间状态都写入数据库。哪怕用户中途退出一天后再回来，系统也能准确接续上次的流程。

更重要的是，这种设计让复杂任务成为可能。比如跨系统操作：“先查我还有几天年假，再帮我提交申请。” 第一步调用内部 API 获取数据，第二步基于结果引导用户确认，整个过程由状态机驱动，而不是靠模型临时发挥。

下面是 SQLite 实现的一个轻量版示例：

import sqlite3 from datetime import datetime, timedelta import json def init_db(): conn = sqlite3.connect('sessions.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_id TEXT NOT NULL, messages TEXT, current_state TEXT, last_active TIMESTAMP, expires_at TIMESTAMP ) ''') conn.commit() return conn def update_session(session_id, user_id, new_message, current_intent, filled_slots): conn = init_db() cursor = conn.cursor() cursor.execute("SELECT messages, current_state FROM sessions WHERE session_id=?", (session_id,)) row = cursor.fetchone() if row: messages = json.loads(row[0]) + [new_message] state = json.loads(row[1]) else: messages = [new_message] state = {} state.update({ "current_intent": current_intent, "filled_slots": filled_slots }) expires_at = datetime.now() + timedelta(hours=24) cursor.execute(''' INSERT OR REPLACE INTO sessions (session_id, user_id, messages, current_state, last_active, expires_at) VALUES (?, ?, ?, ?, ?, ?) ''', ( session_id, user_id, json.dumps(messages), json.dumps(state), datetime.now(), expires_at )) conn.commit() conn.close()

这里用了JSONB类似的文本字段来存动态状态，灵活且兼容性强。不过在生产环境中，建议配合 Redis 做一层缓存，避免频繁读写磁盘影响响应速度。同时设置合理的 TTL（Time To Live），防止无效会话堆积导致数据库膨胀。

另一个值得强调的细节是：状态更新必须是原子性的。如果在写入过程中发生中断，可能导致上下文错乱。因此在高并发场景下，应使用事务机制或乐观锁保障一致性。

模型不仅能“说”，还能“做”

真正强大的 AI 系统不该停留在问答层面，而应具备执行能力。这就是工具调用（Tool Calling）的意义所在。

在 Kotaemon 中，所有外部功能都被注册为“工具”——无论是查询订单、发送邮件，还是调用 ERP 接口。每个工具都有标准化描述，包括名称、用途、输入参数格式（用 JSON Schema 定义）、调用地址等。这些信息集中存放在工具注册表中，构成一个“可用能力清单”。

当用户说“查一下我的订单状态”，模型不会自己去翻数据库，而是输出一个标准格式的调用指令，例如：

{ "tool_name": "get_order_status", "args": { "order_id": "ORD12345678" } }

系统解析后验证权限，调用对应服务，拿到结果再交还给模型总结回复。整个过程就像操作系统调度进程一样可控、可监控。

这种方式带来的好处远超技术本身：

安全性提升：所有操作都经过显式授权，避免模型随意访问敏感接口；
调试更容易：你可以清楚看到哪一步调用了哪个工具，失败时也能快速定位；
功能热插拔：新增一个插件只需注册即可，无需修改主逻辑。

class ToolRegistry: def __init__(self): self.tools = {} def register(self, name: str, description: str, endpoint: str, parameters: Dict): self.tools[name] = { "name": name, "description": description, "endpoint": endpoint, "parameters": parameters, "method": "POST" } def get_schema_for_llm(self) -> List[Dict]: return [ { "type": "function", "function": { "name": tool["name"], "description": tool["description"], "parameters": { "type": "object", "properties": tool["parameters"], "required": [k for k, v in tool["parameters"].items() if v.get("required", False)] } } } for tool in self.tools.values() ] def invoke(self, tool_name: str, args: Dict) -> Dict: if tool_name not in self.tools: raise ValueError(f"Unknown tool: {tool_name}") tool = self.tools[tool_name] try: response = requests.post(tool["endpoint"], json=args, timeout=10) response.raise_for_status() return {"result": response.json()} except Exception as e: return {"error": str(e)}

这个注册中心的设计看似基础，却是系统灵活性的核心。尤其要注意parameters的定义质量——如果描述模糊，模型很可能传错参数。建议遵循 OpenAPI 规范，明确字段类型、是否必填、取值范围等，并加入示例帮助模型理解。

此外，对于耗时较长的操作（如生成报告），建议支持异步模式：工具返回一个任务 ID，前端轮询状态，完成后触发回调通知。这样既能保持对话流畅，又能处理真实业务中的复杂流程。

架构全景：数据如何串联起整个系统

把这三个 Schema 放在一起，就能看到 Kotaemon 的完整数据流图景：

+------------------+ +---------------------+ | |<----->| LLM Inference API | | User Interface | +---------------------+ | (Web/Mobile/App) | ^ +------------------+ | | | v | +------------------+ +--------v---------+ | | | | | Kotaemon Core |<---->| Vector Database | | - Query Router | | (Chroma/Pinecone) | | - State Manager | +-------------------+ | - Tool Caller | +------------------+ | v +------------------+ | | | Plugin Gateway | | - REST APIs | | - Internal Tools | +------------------+

用户的每一次输入，都会触发一次闭环处理：