如何在Kotaemon中自定义检索器和生成器组件？

如何在Kotaemon中自定义检索器和生成器组件？

在企业级智能问答系统日益复杂的今天，一个“通用”的大模型加向量库的架构往往难以满足真实业务场景的需求。比如，某银行客服系统需要从上千份合规文档中精准提取条款，同时确保回答不泄露客户隐私；又或者某医疗机构希望基于最新诊疗指南生成建议，但必须过滤掉尚未获批的实验性疗法——这些都不是开箱即用的 RAG 流程能解决的问题。

正是在这样的背景下，像Kotaemon这类强调模块化与可扩展性的框架展现出独特价值。它不像某些黑盒系统那样将检索和生成牢牢绑定，而是通过清晰的接口设计，让开发者可以自由替换或增强关键组件。其中，自定义检索器（Retriever）与生成器（Generator）成为企业构建高可信、强可控 AI 助手的核心突破口。

以一个典型的企业知识库应用为例：用户提问“年假如何申请？”如果使用标准向量检索，可能会返回三年前已作废的旧制度文档；而若生成器不具备内容审查能力，则可能无意中输出包含审批人姓名或内部流程编号等敏感信息。这类问题看似细小，却足以动摇整个系统的可用性与合规基础。

要真正掌控系统的输出质量，就必须深入到底层逻辑中去干预两个关键环节：

1. 我能查到什么？—— 检索阶段决定了上下文来源的准确性、时效性和权限边界；
2. 我该说什么？—— 生成阶段则关系到最终回复的安全性、结构化程度以及是否具备可解释性。

幸运的是，Kotaemon 提供了足够灵活的抽象机制来支持这两方面的深度定制。

自定义检索器：不只是找文本，更是做决策

在 Kotaemon 中，所有检索器都继承自BaseRetriever抽象类，并实现统一的retrieve(query: str) -> List[Document]接口。这个看似简单的签名背后，隐藏着极大的自由度——你完全可以在retrieve()方法中接入数据库查询、调用外部 API、执行正则匹配，甚至融合多种策略进行结果重排序。

举个实际例子：假设你的组织有一套 HR 政策管理系统，提供 RESTful 接口供实时查询。你可以编写如下自定义检索器：

from kotaemon.retrievers import BaseRetriever from kotaemon.documents import Document from typing import List import requests class CustomAPIRetriever(BaseRetriever): """ 自定义检索器：从企业内部API获取实时政策文档 """ def __init__(self, api_url: str, auth_token: str, timeout: int = 5): self.api_url = api_url self.auth_token = auth_token self.timeout = timeout def retrieve(self, query: str) -> List[Document]: headers = { "Authorization": f"Bearer {self.auth_token}", "Content-Type": "application/json" } payload = { "query": query, "filters": { "department": "HR", "valid_until": ">now" # 只返回仍在有效期的文档 } } try: response = requests.post( f"{self.api_url}/search", json=payload, headers=headers, timeout=self.timeout ) response.raise_for_status() results = response.json().get("results", []) except Exception as e: print(f"[警告] API检索失败: {e}") return [] documents = [] for item in results: doc = Document( text=item["content"], metadata={ "title": item["title"], "source": item["url"], "score": item["relevance_score"], "timestamp": item["updated_at"] } ) documents.append(doc) documents.sort(key=lambda x: x.metadata.get("score", 0), reverse=True) return documents[:5]

这段代码的价值远不止“连接 API”这么简单。它实现了几个关键控制点：

• 业务规则内嵌：自动附加部门和有效性时间过滤，避免过期内容被召回；
• 安全前置：在检索层就完成权限校验（如 token 鉴权），防止越权访问；
• 结果标准化：统一转换为Document对象，便于后续处理；
• 降级容错：异常时不中断主流程，而是返回空列表，保证系统健壮性。

更进一步，你还可以组合多个子检索器形成“混合检索”策略。例如，并行调用向量库、全文搜索引擎和上述 API 检索器，再通过加权打分或交叉验证的方式融合结果。这种多源协同的做法，在面对复杂语义或低资源领域时尤为有效。

自定义生成器：从“生成文字”到“交付答案”

如果说检索决定了“原材料”的质量，那么生成就是决定“成品”形态的最后一道工序。默认情况下，生成器只是简单地把上下文拼成 prompt 发给 LLM，但这对于生产环境来说远远不够。

真正的挑战在于：我们不仅要得到一段通顺的回答，还要确保它是安全的、有依据的、符合格式要求的。这就需要对生成过程实施精细化控制。

以下是一个经过强化的生成器示例，集成了引用标注与敏感词过滤功能：

from kotaemon.generators import BaseGenerator from typing import Dict, Any import openai import re class SafeRAGGenerator(BaseGenerator): """ 安全增强型生成器：结合GPT生成与关键词过滤 """ def __init__(self, model_name: str = "gpt-3.5-turbo", blocked_terms: list = None, enable_citation: bool = True): self.model_name = model_name self.blocked_terms = set(blocked_terms or ["密码", "身份证", "机密"]) self.enable_citation = enable_citation openai.api_key = "your-api-key" def _build_prompt(self, query: str, documents: list) -> str: context = "\n".join([ f"[{i+1}] {doc.text}" for i, doc in enumerate(documents) ]) citation_hint = ( "请在引用处用 [1]、[2] 等标记对应文档编号。" if self.enable_citation else "" ) prompt = f""" 你是一个专业助手，请根据以下资料回答问题。 若无法从资料中找到答案，请回答“暂无相关信息”。 资料： {context} 问题：{query} 要求：简洁准确，{citation_hint} 回答： """ return prompt def _contains_blocked_content(self, text: str) -> bool: return any(term in text for term in self.blocked_terms) def generate(self, query: str, documents: list) -> str: prompt = self._build_prompt(query, documents) try: response = openai.ChatCompletion.create( model=self.model_name, messages=[{"role": "user", "content": prompt}], temperature=0.3, max_tokens=512 ) raw_output = response.choices[0].message["content"].strip() if self._contains_blocked_content(raw_output): return "回答包含受限内容，已被屏蔽。" return raw_output except Exception as e: print(f"[错误] 生成失败: {e}") return "抱歉，当前服务不可用。"

这个生成器的设计体现了几个重要的工程思维：

• 提示工程结构化：明确告诉模型“怎么答”，并引导其使用[1]格式引用原文，极大提升了回答的可追溯性；
• 安全防线内置：在输出端设置黑名单扫描，作为最后一道防护屏障；
• 用户体验优先：即使后端出错也不暴露技术细节，而是返回友好提示；
• 配置可外置：敏感词列表、模型名称等均可从配置文件加载，便于运维管理。

值得注意的是，这里的generate()方法签名虽然只接受query和documents，但在实际项目中，你完全可以扩展参数，例如传入用户角色、会话历史或上下文状态，从而实现个性化生成逻辑。

当我们将这两个自定义组件整合进系统时，整体工作流变得既强大又可控：

[用户提问] ↓ [NLU预处理] → [CustomAPIRetriever] → [内部API/文档库] ↓ ↘ ↙ ↓ [融合检索结果] ↓ [SafeRAGGenerator] → [LLM + 安全过滤] ↓ [最终回答]

在这个架构中：

• 检索层不再是单纯的“相似度匹配”，而是融合了业务规则、权限控制和时效判断的智能筛选器；
• 生成层也不再是“盲信模型输出”，而是嵌入了审核逻辑、格式规范和异常兜底的可靠引擎；
• 整个系统具备了日志追踪、性能监控和灰度发布的可能性，为上线后的持续优化打下基础。

实践中常见的痛点也得以解决：

当然，这一切的前提是你愿意走出“默认配置”的舒适区。在实施过程中，以下几个经验值得参考：

• 不要追求一步到位：先实现基本功能，再逐步叠加过滤、缓存、熔断等高级特性；
• 做好降级预案：当 API 不可用时，应能切换至本地备份索引或静态知识库；
• 加强可观测性：记录每次检索的输入、命中文档 ID、生成耗时等指标，用于后期分析；
• 测试覆盖边界情况：如空查询、超长文本、特殊字符、乱码输入等；
• 避免硬编码：将 API 地址、token、黑名单等敏感项移至环境变量或配置中心。

真正有价值的 AI 系统，从来不是靠堆砌模型参数赢得比赛的。它的核心竞争力，往往体现在那些“不起眼”的细节里：一次精准的文档过滤、一句被成功拦截的敏感输出、一个带引用标记的答案链接……正是这些由自定义组件带来的微小确定性，累积成了企业在高风险场景下敢于部署的信心。

掌握如何在 Kotaemon 中定制检索与生成逻辑，意味着你不再只是一个使用者，而是一名系统设计者。你可以根据行业特性注入合规要求，可以根据用户角色调整信息粒度，甚至可以让 AI 主动识别模糊提问并发起澄清对话。

未来，随着个性化需求的增长，这类“深度定制”能力将不再是加分项，而是构建可信 AI 应用的基本功。而今天的每一次接口重写、每一条规则添加，都是在为那个更智能、更可靠、更贴近真实世界的交互范式铺路。