第 9 章 RAGTelegramBot

第 9 章：第七周（下）：Telegram Bot，把 RAG 装进口袋

这一章解决什么问题？上一章把 RAG 改造成会决策的 Agentic 流程，但入口仍是 HTTP 端点或本地 Gradio 页面，你得坐在电脑前才能用。Week 7 README 开头写明这一周做了 TWO major enhancements：Agentic RAG with LangGraph，以及 Telegram Bot Integration，定位是 conversational interface for mobile/desktop access。这一章只讲后者：为什么选 Telegram，bot 怎么接回前六周的服务层，消息如何分流，异步与错误处理怎么设计，以及怎么验证它真能用。

为什么把入口放在 Telegram

README 的 Week 7 一行摘要写作 Agentic RAG with LangGraph and Telegram Bot for mobile access；Week 7 README 把产物称为 mobile-first, conversational research assistant accessible anywhere via Telegram。这个选择绕开了三件麻烦事。

不写客户端。Telegram 覆盖 phone 与 desktop，你只要用 @BotFather 注册 bot 拿到 token，不必自建 App、上架审核、分别处理 iOS 和 Android，"移动端"的成本因此压到一个 token 的量级。

部署门槛低。源 README 的 FAQ 明确回答：成本上 Telegram bot API is completely free with no limits；公网 IP 上 No! Polling mode works perfectly for development and low-traffic deployments. Webhook requires HTTPS. 所以本地 docker compose up 起来就能用，移动端入口是默认形态，不是"以后再说"的加分项。

多人隔离是内建的：每个用户有独立的 settings 和 session，可选地用 TELEGRAM__ALLOWED_USER_IDS 把 bot 限制成私有。这三条加起来，"把 RAG 装进口袋"才变成当晚能跑通的东西。

Bot 架构与服务目录

Telegram 层没有检索能力，它是新入口管道，末端接回 Week 5-6 的 RAG 服务：

Telegram User
    ↓
Telegram Bot (Polling/Webhook)
    ↓
TelegramService + Handlers
    ↓ [Langfuse Tracing]
Cache Check (Redis)
    ├─ Hit → Instant Response (~100ms)
    └─ Miss → Full RAG Pipeline
        ↓
Hybrid Search (OpenSearch BM25 + Vector)
        ↓
LLM Generation (Ollama)
        ↓
Cache Store (Redis)
        ↓
Format Response → Send to Telegram

这条链上没有新组件做检索或生成。README 的 Service Integration 一节列全了复用关系：OpenSearch 负责 Hybrid Search，Jina Embeddings 提供语义检索，Ollama 生成答案，Redis 给重复查询带来 150-400x 加速，Langfuse 记录完整的 Telegram 交互 trace，PostgreSQL 的论文元数据经 OpenSearch 被检索到。Telegram 只贡献入口、会话与呈现。

新增代码集中在两个目录：

src/services/telegram/
├── client.py           # Main Telegram bot service
├── handlers.py         # Command and message handlers
├── formatters.py       # Message formatting utilities
├── keyboards.py        # Interactive inline keyboards
├── user_manager.py     # User settings and sessions
└── factory.py          # Factory function

src/schemas/telegram/
├── messages.py         # Message validation schemas
├── commands.py         # Command schemas
└── user_settings.py    # User preferences schema

分层意图清楚：client.py 只管生命周期，handlers.py 管路由与业务动作，formatters.py 把数据变成可发送文本，keyboards.py 管交互控件，user_manager.py 管每个用户的偏好与历史，factory.py 负责依赖注入；src/schemas/telegram/ 的 Pydantic schema 先把消息、命令、用户设置校验一遍。服务入口在 src/main.py：

# Entry point: src/main.py
telegram_service = make_telegram_service(...)
await telegram_service.start()

# Service: src/services/telegram/client.py
class TelegramService:
    async def start() -> Start bot in polling/webhook mode
    async def stop() -> Stop bot gracefully
    async def health_check() -> Check bot status

start() 兼顾 polling 与 webhook 两种模式，stop() 优雅退出，health_check() 挂进 FastAPI 健康检查——curl http://localhost:8000/api/v1/health 应能看到 telegram_service 状态。

Week 7 完整架构图：Telegram Bot 作为新入口接在 Agentic RAG 与 Hybrid Search 服务层之上，链路由 Telegram 用户经 OpenSearch、Ollama、Redis 落到 Langfuse trace

命令处理器与消息分流

handlers.py 是这层的核心，README 的 Code Structure 列出了它暴露的处理函数：

# Handlers: src/services/telegram/handlers.py
class TelegramHandlers:
    async def start_command() -> /start
    async def help_command() -> /help
    async def ask_command() -> /ask
    async def search_command() -> /search
    async def settings_command() -> /settings
    async def handle_message() -> Regular text messages

命令是显式入口，普通文本走 handle_message()，也就是 README 说的 automatic query routing (commands vs regular messages)。分流的意义在于：/search 只走 OpenSearch，2-3s 返回论文列表；/ask 与自然语言消息才走完整 RAG 流水线；/status、/settings 完全不碰 LLM。

命令	作用
`/start`	欢迎消息与 bot 能力
`/help`	详细使用说明
`/ask`	显式提问
`/search`	按关键词检索论文
`/settings`	自定义偏好
`/status`	检查系统健康与统计
`/clear`	清空对话历史

/settings 不靠敲参数，而是 keyboards.py 的 inline keyboard 按钮：搜索模式在 Hybrid (BM25 + semantic) 与 BM25 only 间切换，结果数量选 3、5、10 篇，分类过滤支持 All、cs.AI、cs.LG、cs.CL、cs.CV、cs.NE，还能选 LLM 模型与 streaming、显示来源等开关。README 记录的效果是按钮 <100ms 即回、改动持久保存，但也提醒 settings 存在内存里，生产要迁到 Redis/PostgreSQL。

格式化是独立关注点：

# Formatters: src/services/telegram/formatters.py
format_rag_response() -> Rich Markdown formatting
format_search_results() -> Search result display
format_welcome_message() -> /start message
escape_markdown_v2() -> Telegram MarkdownV2 escaping
split_long_message() -> Auto-split >4000 chars

五个函数对应五个真实的坑：Telegram 用 MarkdownV2，特殊字符必须转义，否则消息直接发送失败；单条消息有长度上限，超过 4000 字符要切分（TELEGRAM__MAX_MESSAGE_LENGTH=4000，注释写明 Telegram limit is 4096）；arXiv 链接要做成可点击标题链接。

消息流程与异步处理

README 用一段伪代码把一次问答摊开：

# Simplified message flow
async def handle_message(update, context):
    1. Extract user_id, chat_id, text
    2. Check rate limits
    3. Get/create user settings
    4. Show "typing..." indicator
    5. Check cache (Redis)
    6. If cache hit → format and send
    7. If cache miss:
        a. Generate embedding (Jina)
        b. Search papers (OpenSearch)
        c. Build prompt with context
        d. Generate answer (Ollama)
        e. Cache result (Redis)
        f. Trace interaction (Langfuse)
    8. Format response (Markdown)
    9. Split if too long
    10. Send to Telegram
    11. Update conversation history

两个细节值得停一下。第四步的 typing indicator 必须在检索之前发出——它在性能表里标着 <500ms，而首答要 15-20 秒；同一个 20 秒，有没有提示是两种感受。第三步与第十一步说明会话有状态：user_manager.py 为每个用户维护设置和最近 10 条消息的历史，默认 30 分钟不活动超时清理（TELEGRAM__SESSION_TIMEOUT_MINUTES=30）。第六步的缓存命中路径收益最直接：不生成 Embedding、不检索、不调 LLM。

README 的 Performance Benchmarks 表口径如下（课程 README 自身测量，非绝对承诺）：

指标	数值	说明
First Query	15-20s	完整 RAG 流水线执行
Cached Query	50-100ms	即 README 所称的 150-400x faster
Typing Indicator	<500ms	立即显示
Search Only	2-3s	`/search` 命令
Status Check	<1s	`/status` 命令
Settings Update	<100ms	按钮即时响应
Concurrent Users	10+	同时测试

README 还给了期望缓存命中率的三个口径：重复的完全相同查询 100%，热门问题 60-80%，唯一查询首次 0%。资源占用上，每个活跃 session 约 50MB 内存，会话不清理会随用户数线性增长。

错误处理与优雅降级

移动端最怕把内部异常直接甩给用户。README 的 Error Handling 列了五条降级规则：

Markdown 格式化失败 → 回退纯文本
Embedding 生成失败 → 回退 BM25，Hybrid Search 退化成关键词检索
Cache 不可用 → 跳过缓存继续跑
Langfuse tracing 失败 → 记 warning 后继续
服务异常 → 给用户可读的错误消息，不外泄 stack trace

速率限制是另一层保护：每用户每分钟 20 条（TELEGRAM__RATE_LIMIT_MESSAGES_PER_MINUTE=20），由 Telegram 库自动节流，外加 30 分钟 session 超时。高频排障场景：

现象	处理
Bot 不响应	检查 `TELEGRAM__ENABLED=true` 与 BOT_TOKEN
报 "Unauthorized"	token 无效，用 `@BotFather` 重新生成
有回复但没有答案	检查 OpenSearch、Ollama、embeddings
响应慢	首答本身慢，后续查询走缓存
报 "Forbidden"	检查 `ALLOWED_USER_IDS` 限制
内存问题	session 存在内存，监控 RAM

定位时打开 debug 日志并按 telegram 过滤：

# In .env
DEBUG=true

# Check bot logs
docker compose logs -f api | grep telegram

缓存没生效导致的慢，README 给的顺序是先确认 Redis 活着，再看缓存命中率，最后怀疑 Langfuse tracing 阻塞——docker exec rag-redis redis-cli ping 应返回 PONG，LANGFUSE__ENABLED=false 可排除该变量。

与 Agentic RAG 端点对接，以及 Telegram 配置

Telegram 层对接的是 Week 7 新增的 Agentic 端点，而不是 Week 5 那对 /api/v1/ask + /api/v1/stream。端点定义在 src/routers/agentic_ask.py：

POST /api/v1/ask-agentic

// Request
{
  "query": "What are transformers in ML?",
  "top_k": 3,
  "use_hybrid": true
}

// Response
{
  "query": "What are transformers in ML?",
  "answer": "Transformers are neural network architectures...",
  "sources": ["https://arxiv.org/pdf/1706.03762.pdf"],
  "chunks_used": 3,
  "search_mode": "hybrid",
  "reasoning_steps": [
    "Decided to retrieve relevant papers",
    "Retrieved documents from database",
    "Generated answer from relevant documents"
  ],
  "retrieval_attempts": 1
}

响应里最有价值的是 reasoning_steps 和 retrieval_attempts。README 把 Reasoning Transparency 列在 Week 7 的 Key Innovations 里，聊天界面恰好是展示它的地方。主 README 还提到相关 guardrail：Out-of-domain detection prevents hallucination。

配置全部走 .env：

# Enable/Disable Bot
TELEGRAM__ENABLED=true  # Set to false to disable

# Bot Token (Required)
TELEGRAM__BOT_TOKEN=your_token_here

# Deployment Mode
TELEGRAM__USE_WEBHOOK=false  # true for production
TELEGRAM__WEBHOOK_URL=https://your-domain.com
TELEGRAM__WEBHOOK_PATH=/telegram/webhook

# Access Control (Optional)
TELEGRAM__ALLOWED_USER_IDS=123456789,987654321  # Empty = allow all

# Behavior Settings
TELEGRAM__MAX_MESSAGE_LENGTH=4000  # Telegram limit is 4096
TELEGRAM__ENABLE_STREAMING=true
TELEGRAM__SESSION_TIMEOUT_MINUTES=30
TELEGRAM__RATE_LIMIT_MESSAGES_PER_MINUTE=20

# Default User Preferences
TELEGRAM__DEFAULT_TOP_K=3
TELEGRAM__DEFAULT_USE_HYBRID=true
TELEGRAM__DEFAULT_MODEL=llama3.2:1b

TELEGRAM__BOT_TOKEN 是唯一必填项，来自 @BotFather 的 /newbot 流程。开发默认 polling；生产建议开 webhook，README 列的前置条件是带有效证书的 HTTPS 域名、Nginx/Caddy 做 TLS 终止、公网 IP 或反代——polling 零网络配置，webhook 更实时但要先搭好 HTTPS。多实例扩展路径也点了：用户存储从内存迁到 Redis/PostgreSQL、增加 Redis 内存配额、起多个 API 实例做负载均衡、按业务调速率限制、给错误与延迟配告警。

端到端验证与测试场景

从零到能对话只要五步。先在 Telegram 里找 @BotFather 发 /newbot，起名字和一个以 bot 结尾的 username，拿到形如 1234567890:ABCdefGHIjklMNOpqrsTUVwxyz-1234567 的 token，再写进 .env：

# Enable Telegram bot
TELEGRAM__ENABLED=true
TELEGRAM__BOT_TOKEN=your_token_from_botfather_here

# Optional: Restrict to specific users (comma-separated Telegram user IDs)
# Leave empty to allow all users
TELEGRAM__ALLOWED_USER_IDS=

# Use polling mode for development (webhook requires HTTPS)
TELEGRAM__USE_WEBHOOK=false

第三步 uv sync 装 python-telegram-bot，第四步 docker compose up --build -d 并确认日志：

docker compose logs -f api

应当依次看到 INFO - Telegram bot started successfully、INFO - Starting Telegram bot in polling mode、INFO - Bot commands set successfully。第五步搜到你的 bot，发 /start，问一句 "What are transformers in machine learning?"，检查返回是否带 arXiv 来源链接。

六个手动场景覆盖主要分支。命令类：/start 回欢迎消息，/help 回命令文档，/status 回 ✅ OPENSEARCH / ✅ OLLAMA / ✅ CACHE 逐项状态。问答类：首次问 attention mechanisms 需等 15-20s，返回答案、带分数与 arXiv 链接的来源、以及 ⚙️ Mode: hybrid；同一问题再问一遍应约 100ms 返回并多出 ⚡ Cached 标记，这就是 150-400x 加速的手感验证。检索类：/search transformer neural networks 返回形如 📖 Found 145 papers (showing top 10) 的列表。设置类：点 "5 Results" 回 ✅ Results per query: 5，点 "cs.AI" 回 ✅ Category filter: cs.AI。多轮类：先问 BERT 再问 "How does it differ from GPT?" 要接得上上下文，/clear 清空历史但保留设置。异常类：发 asdfghjkl 应得到 ❌ No relevant papers found.，服务挂掉时得到可读提示而非 stack trace。

README 的验收清单有 12 条，最能暴露问题的是：全部命令可响应、自然语言提问返回带来源的答案、缓存生效（出现 ⚡）、设置跨会话保持、inline 按钮可点、长回复被正确切分、Markdown 渲染正常、Langfuse 在 http://localhost:3000 能看到 Telegram 事件、错误消息对用户友好。其中 Langfuse 那条要单独强调——trace 里看不到 Telegram 请求，说明这层还没接进 Week 6 的可观测性。

把入口搬到手机上，产品形态的变化比"多一个界面"更大：Gradio 是开发者坐在桌前调试用的，Telegram 常驻在手机里，同一个问题可以在地铁上问。README 列出的后续增强——图片、语音消息、群聊、多语言、按分类推送新论文等——大多建立在"入口已经随身"这个前提上；Integration Ideas 里的 Slack、Discord、WhatsApp、Web Widget 也说明同一件事：handlers.py / formatters.py / user_manager.py 这层切分可以整体换壳。

参考链接

课程仓库（Week 7 代码与 notebook）：jamwithai/production-agentic-rag-course
Telegram 服务目录：src/services/telegram/
Telegram schema 目录：src/schemas/telegram/
Agentic 端点：src/routers/agentic_ask.py
交互式 notebook：notebooks/week7/week7_agentic_rag.ipynb
Week 7 notebook 说明：notebooks/week7/README.md（其中提到的 docs/AGENTIC_RAG_IMPLEMENTATION_PLAN.md、docs/AGENTIC_RAG_TESTING_PLAN.md 在仓库文件树里不存在，故不提供链接）
对应博客：Agentic RAG with LangGraph and Telegram
Week 7 代码 release：week7.0
一手文档：Telegram Bot API、python-telegram-bot、@BotFather
依赖服务：Langfuse Docs、Redis Docs

自己跑的话，从 @BotFather 开始，把 token 填进 .env，其余照上面的顺序验证一遍即可。

第 9 章：第七周（下）：Telegram Bot，把 RAG 装进口袋

第 9 章：第七周（下）：Telegram Bot，把 RAG 装进口袋

为什么把入口放在 Telegram

Bot 架构与服务目录

命令处理器与消息分流

消息流程与异步处理

错误处理与优雅降级

与 Agentic RAG 端点对接，以及 Telegram 配置

端到端验证与测试场景

参考链接

📑 生产级 Agentic RAG 实战：从 arXiv 论文助手到 Telegram Bot