Kotaemon扩展应用：连接外部数据库实现动态知识检索

Kotaemon扩展应用：连接外部数据库实现动态知识检索

1. 技术背景与应用场景

随着大语言模型（LLM）在自然语言处理领域的广泛应用，基于检索增强生成（Retrieval-Augmented Generation, RAG）的问答系统逐渐成为企业级知识管理的核心工具。Kotaemon 是由 Cinnamon 开发的开源项目，定位为一个面向终端用户的 RAG UI 页面，旨在降低构建和使用 RAG pipeline 的技术门槛。

该平台不仅支持文档上传、索引构建与语义检索等基础功能，还提供了可视化界面，使非技术人员也能快速搭建 DocQA（Document-based Question Answering）系统。然而，在实际业务场景中，静态文档无法满足对实时性数据的需求，例如客户订单状态、库存信息或用户行为日志等动态内容。因此，将 Kotaemon 扩展以连接外部数据库，实现动态知识检索，是提升其工业适用性的关键一步。

本文属于实践应用类技术文章，重点介绍如何通过自定义组件集成外部关系型数据库（如 PostgreSQL、MySQL），使 Kotaemon 能够在生成回答时实时查询结构化数据源，从而实现“静态文档 + 动态数据”的混合检索能力。

2. Kotaemon 架构概览与扩展机制

2.1 核心架构解析

Kotaemon 基于模块化设计思想构建，其核心流程包括：

• 文档加载器（Document Loader）：支持 PDF、TXT、DOCX 等格式文件的解析。
• 文本分割器（Text Splitter）：将长文本切分为适合嵌入模型处理的片段。
• 向量存储（Vector Store）：利用 FAISS 或 Chroma 存储文本块的向量表示。
• 检索器（Retriever）：根据用户问题进行相似度匹配，返回相关上下文。
• 生成器（Generator）：调用 LLM（如通过 Ollama 部署的 Llama3）生成最终答案。

整个流程可通过图形化界面配置，形成可复用的 pipeline。

2.2 可扩展性设计

Kotaemon 支持通过 Python 插件机制注册自定义节点（Node），允许开发者注入新的数据源、处理逻辑或输出方式。这一特性为接入外部数据库提供了技术基础。我们可以通过实现一个DatabaseRetrieverNode类，将其注册到 pipeline 中，作为标准检索路径之外的补充数据通道。

3. 实现步骤详解：集成外部数据库

本节将演示如何在 Kotaemon 中添加一个 PostgreSQL 数据库连接，并在问答过程中自动执行 SQL 查询以获取最新数据。

3.1 环境准备

确保以下环境已就绪：

# 安装必要依赖 pip install psycopg2-binary sqlalchemy pandas

同时确认 Kotaemon 的插件目录结构如下：

kotaemon/plugins/ └── custom_retrievers/ └── __init__.py └── db_retriever.py

3.2 自定义数据库检索节点实现

以下是db_retriever.py的完整代码实现：

# plugins/custom_retrievers/db_retriever.py from typing import List, Dict, Any import os from sqlalchemy import create_engine, text import pandas as pd from kotaemon.base import BaseComponent, Document class DatabaseRetrieverNode(BaseComponent): """ 自定义节点：从外部数据库检索结构化数据 """ connection_string: str = os.getenv("DATABASE_URL", "postgresql://user:pass@localhost:5432/mydb") query_templates: Dict[str, str] = { "order_status": "SELECT status, updated_at FROM orders WHERE order_id = '{order_id}'", "user_info": "SELECT name, email, join_date FROM users WHERE user_id = '{user_id}'" } def invoke(self, input_text: str, **kwargs) -> List[Document]: """ 根据输入文本提取参数并执行查询 """ try: engine = create_engine(self.connection_string) results = [] # 示例：简单关键词匹配提取订单号 if "订单" in input_text and any(c.isdigit() for c in input_text): order_id = ''.join(filter(str.isdigit, input_text)) sql = self.query_templates["order_status"].format(order_id=order_id) with engine.connect() as conn: result = conn.execute(text(sql)) df = pd.DataFrame(result.fetchall(), columns=result.keys()) if not df.empty: content = df.to_json(orient='records', ensure_ascii=False) results.append(Document( text=content, metadata={"source": "database", "query_type": "order_status"} )) return results except Exception as e: return [Document(text=f"数据库查询失败: {str(e)}", metadata={"error": True})] async def run(self, input_text: str, **kwargs) -> List[Document]: return self.invoke(input_text, **kwargs)

说明：该节点监听用户提问中的关键词（如“订单”），提取数字作为订单 ID，然后执行预设 SQL 模板进行查询。结果以 JSON 格式封装为Document对象，供后续 LLM 使用。

3.3 注册插件

在plugins/custom_retrievers/__init__.py中注册新节点：

from .db_retriever import DatabaseRetrieverNode __all__ = ["DatabaseRetrieverNode"]

重启 Kotaemon 后台服务即可在 UI 的节点库中看到新组件。

3.4 在 Pipeline 中配置数据库检索

1. 登录 Kotaemon UI，默认账号密码为admin/admin。
2. 创建新 Pipeline，拖入 “Custom Retriever” 节点（即DatabaseRetrieverNode）。
3. 将其与主检索器并联，输出合并后送入 Generator。
4. 保存并运行 Pipeline。

此时，当用户提问“订单12345的状态是什么？”时，系统会：

• 从向量库中检索相关文档；
• 并行触发数据库查询；
• 将两者结果拼接后提交给 LLM 综合生成回答。

4. 实践难点与优化方案

4.1 参数提取准确性问题

原始实现依赖正则表达式或关键词匹配提取数据库查询参数，容易误判。改进方向包括：

• 引入轻量级 NER 模型识别实体（如订单号、用户ID）
• 使用意图分类模型判断是否需要访问数据库

# 示例：使用简单规则增强鲁棒性 import re def extract_order_id(query: str) -> str | None: patterns = [ r"订单[号|ID]*[:：\s]*(\d+)", r"单号[是为:：\s]*(\d+)", r"\b\d{6,}\b" # 连续6位以上数字视为订单号 ] for pattern in patterns: match = re.search(pattern, query) if match: return match.group(1) return None

4.2 安全性控制

直接拼接 SQL 存在注入风险。应改用参数化查询：

from sqlalchemy import bindparam # 修改模板为参数化形式 query_templates = { "order_status": "SELECT status, updated_at FROM orders WHERE order_id = :order_id" } # 执行时传参 stmt = text(sql).bindparams(bindparam("order_id")) result = conn.execute(stmt, {"order_id": order_id})

4.3 性能优化建议

• 缓存高频查询结果：对短时间内重复请求的数据设置 Redis 缓存
• 异步查询：使用asyncio和asyncpg提升并发性能
• 连接池管理：通过 SQLAlchemy 连接池避免频繁建立连接

5. 应用效果展示

完成上述配置后，可在 Kotaemon UI 中测试混合检索效果：

• 输入：“请告诉我订单 880235 的当前状态。”
• 系统响应示例：

  订单 880235 当前状态为“已发货”，更新时间为 2025-04-03 14:22:10。

与此同时，若该订单涉及退换货政策说明，系统还会从本地文档中检索相关政策条款一并呈现。

这种“动静结合”的检索模式显著提升了问答系统的实用性和时效性。

6. 总结

6.1 实践价值总结

本文介绍了如何通过扩展 Kotaemon 的插件系统，集成外部数据库实现动态知识检索。核心成果包括：

• 成功实现了基于 SQL 的实时数据查询能力；
• 构建了“文档检索 + 数据库查询”双通道 RAG 架构；
• 提供了一套可复用的插件开发模板，适用于多种结构化数据源。

6.2 最佳实践建议

1. 优先保障安全性：禁止 SQL 拼接，使用参数化查询；
2. 明确职责边界：数据库仅用于获取事实数据，复杂推理仍由 LLM 完成；
3. 做好错误降级：数据库不可用时应优雅回退，不影响主流程。

通过合理扩展，Kotaemon 不再局限于静态文档问答，而是演变为支持多源异构数据融合的企业级智能助手平台。

---

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。