Kotaemon RSS订阅集成：实时获取最新资讯-智慧文博士

Kotaemon RSS订阅集成：实时获取最新资讯

在智能客服、虚拟助手和企业知识系统日益普及的今天，一个常被忽视但至关重要的问题浮出水面：如何让AI说“新话”？

我们见过太多这样的场景——用户问：“最近有没有关于数据安全的新政策？”系统翻遍静态知识库，给出一条三个月前的答案。看似合理，实则滞后。这种“知识老化”不仅降低用户体验，更可能引发合规风险。

Kotaemon 的出现，正是为了解决这一痛点。它不是一个简单的问答机器人框架，而是一个能“呼吸”的智能体——通过集成 RSS 订阅机制，持续吸入外部世界的最新动态，并将其转化为可检索、可追溯、可生成的知识血液。

从“被动响应”到“主动感知”：RSS为何是关键拼图？

传统 RAG（检索增强生成）系统大多依赖预先构建的知识库，更新周期以天甚至周计。但在新闻、政策、技术更新等领域，信息窗口期极短。一条突发公告发布两小时后才被纳入知识库？那已经太迟了。

RSS 协议虽然诞生已久，却因其轻量、标准化和广泛支持，成为实时内容分发的事实标准。无论是 TechCrunch 的科技快讯，还是政府官网的政策通告，几乎都提供 RSS 输出。更重要的是，这些内容自带结构化元数据：标题、摘要、发布时间、唯一标识符——这正是自动化处理的理想输入。

Kotaemon 没有重新发明轮子，而是巧妙地将 RSS 变成它的“耳朵”，实现了从“定期查资料”到“随时听消息”的转变。

抓得住：稳定可靠的采集能力是怎么炼成的？

抓取 RSS 看似简单，但真正落地时会遇到各种现实挑战：网络抖动、源站限流、格式不规范、时间戳混乱……如果每次失败都导致任务中断，那再好的设计也无法长期运行。

Kotaemon 的采集模块采用feedparser库作为核心解析引擎，它不仅能自动识别 RSS 2.0 和 Atom 1.0 格式，还能在 XML 结构轻微异常时尽力恢复数据，而不是直接抛错退出。这一点在面对非专业运维的博客站点时尤为重要。

更关键的是增量采集逻辑。以下代码片段展示了其精髓：

import feedparser from datetime import datetime from typing import List, Dict class RSSCollector: def __init__(self, sources: List[str], last_fetch_times: Dict[str, datetime]): self.sources = sources self.last_fetch_times = last_fetch_times def fetch_new_entries(self) -> List[Dict]: new_entries = [] for url in self.sources: try: feed = feedparser.parse(url) if feed.bozo and feed.bozo_exception: print(f"Feed parsing error for {url}: {feed.bozo_exception}") continue for entry in feed.entries: pub_time = datetime(*entry.published_parsed[:6]) if hasattr(entry, 'published_parsed') else datetime.now() # 仅处理比上次抓取时间更新的内容 if url in self.last_fetch_times and pub_time <= self.last_fetch_times[url]: continue new_entries.append({ "title": entry.title, "link": entry.link, "summary": entry.summary, "published_at": pub_time, "source": url }) self.last_fetch_times[url] = datetime.now() except Exception as e: print(f"Error fetching {url}: {e}") return new_entries

这里有几个工程上的小心思值得借鉴：

时间戳优先于顺序判断：有些 Feed 不保证条目按时间倒序排列，所以不能只取前几条，必须逐个比较发布时间。
回退机制：当某次抓取失败时，last_fetch_times不更新，下次重试仍能覆盖遗漏条目。
错误容忍而非崩溃：单个源的解析错误不会影响其他源的采集，系统整体健壮性大幅提升。

这个模块可以作为独立服务运行，也可以嵌入 Celery 或 Airflow 调度系统中，实现每15分钟一次的低频高稳抓取。

整得明白：文本清洗与向量化背后的细节考量

拿到原始 RSS 条目只是第一步。真正的挑战在于：如何把一段可能包含 HTML 标签、乱码字符甚至广告脚本的摘要，变成 LLM 能理解、向量数据库能索引的高质量语义单元？

这就是文档预处理模块的舞台。

from sentence_transformers import SentenceTransformer import numpy as np import re from typing import List, Dict class DocumentProcessor: def __init__(self, model_name='all-MiniLM-L6-v2', chunk_size=512): self.model = SentenceTransformer(model_name) self.chunk_size = chunk_size def clean_text(self, text: str) -> str: text = re.sub(r'<[^>]+>', '', text) # 去除HTML标签 text = re.sub(r'\s+', ' ', text) # 合并多余空白 return text.strip() def chunk_text(self, text: str) -> List[str]: words = text.split() chunks = [' '.join(words[i:i+self.chunk_size]) for i in range(0, len(words), self.chunk_size)] return chunks def process_entries(self, entries: List[Dict]) -> List[Dict]: processed = [] for entry in entries: cleaned_summary = self.clean_text(entry["summary"]) chunks = self.chunk_text(cleaned_summary) embeddings = self.model.encode(chunks) for i, chunk in enumerate(chunks): processed.append({ "text": chunk, "embedding": embeddings[i], "metadata": { "title": entry["title"], "source_url": entry["link"], "published_at": entry["published_at"], "origin_feed": entry["source"] } }) return processed

这段代码背后藏着几个实用经验：

词粒度切块优于字符：按空格分词再组合，避免在单词中间断裂，影响语义完整性。
滑动窗口不是万能的：虽然能缓解边界问题，但会显著增加存储和计算开销。对于 RSS 摘要这类本身较短的内容，固定步长已足够。
模型选择讲求平衡：all-MiniLM-L6-v2在速度与效果之间取得了良好折衷，适合高频更新场景。若追求更高精度，可换用text-embedding-3-large，但需评估延迟成本。

此外，每个向量片段都绑定了完整的元数据，这是实现“可追溯回答”的基础。用户看到的回答不再是一句孤立的话，而是有据可依的信息节点。

问得精准：对话中如何唤醒“新鲜记忆”？

很多系统做到了“采得进、存得下”，却卡在最后一步：怎么让用户真正用起来？

想象这样一个多轮对话：

用户：上个月发布了哪些AI相关政策？
系统：列举三条近期政策……
用户：第一条是谁发布的？

如果系统此时无法关联上下文，就会陷入“你说哪条？”的尴尬境地。

Kotaemon 的对话管理模块通过维护一个轻量级上下文状态机解决了这个问题。更重要的是，它引入了时效性加权检索机制，让“新”成为一种竞争力。

import numpy as np from sklearn.metrics.pairwise import cosine_similarity from datetime import datetime def retrieve_from_rss(user_query: str, vector_db: List[np.ndarray], metadata_db: List[Dict], processor: DocumentProcessor, top_k=3) -> List[Dict]: query_vec = processor.model.encode([user_query]) similarities = cosine_similarity(query_vec, np.array([v["embedding"] for v in vector_db])) ranked_indices = np.argsort(similarities[0])[::-1] results = [] now = datetime.now() for idx in ranked_indices[:top_k * 2]: item = metadata_db[idx] days_diff = (now - item["metadata"]["published_at"]).days time_weight = max(0.5, 1 - 0.05 * days_diff) # 每天衰减5%，最低保留50% adjusted_score = similarities[0][idx] * time_weight results.append({ "score": float(adjusted_score), "content": vector_db[idx]["text"], "metadata": item["metadata"] }) results.sort(key=lambda x: x["score"], reverse=True) return results[:top_k]

这个看似简单的公式其实非常有效：

相似度确保语义相关；
时间权重防止旧闻干扰；
综合评分实现“又准又新”。

比如用户问“最近有什么变化？”，即使某个老政策在语义上更匹配，也会因时间衰减而排名靠后，真正体现“最近”的含义。

同时，系统设有防幻觉机制：若 top-k 结果的综合得分均低于阈值，则明确告知“暂无最新相关信息”，而不是强行编造答案。

架构之美：松耦合如何成就高可用

整个系统的数据流动如下：

[RSS Feeds] ↓ (HTTP Polling) [RSS Collector] → [Message Queue (Optional)] ↓ (New Entries) [Document Processor] ↓ (Embedded Chunks + Metadata) [Vector Database] ←→ [LLM Gateway] ↑ ↗ [User Query] → [Dialogue Manager]

各组件之间尽可能解耦：