🤖 Telegram Agent Bot：一天搭好、一周能给别人用的 Agent 项目

📌 这篇适合谁

• 学完 ReAct/Function Calling，想接一个真实交互界面的开发者
• 需要一个能放进简历的"端到端" Agent 项目的求职者
• 想做一个"真的有人用"的 Agent 但不知道从哪开始的人
• 对 Bot 开发感兴趣但没碰过 Telegram API 的新手

这是案例库「个人项目」系列的第一篇。

我是🍋AI小柠檬。Telegram Bot 是我做的第一个"有真实用户"的 Agent 项目。从 0 到上线花了一天，第一周就有 30 个人在用。这篇把完整过程交给你。

---

写在前面：为什么 Telegram Bot 是 Agent 项目的最佳起点 🎯

⚠️ 我见过的"第一个项目"选择失误

很多人学完 Agent 想做项目，第一反应是"做一个 SaaS 产品"或"做一个网站"。然后：

• 花 2 周搞前端（React/Vue/Tailwind）
• 花 1 周搞后端（FastAPI/数据库/认证）
• 花 3 天搞部署（Docker/Nginx/SSL）
• Agent 本身只写了 2 小时

80% 的时间花在了和 Agent 无关的事情上。

Telegram Bot 的优势：Telegram 帮你做了前端、消息推送、用户认证、移动端适配。你只需要写 Agent 逻辑 + 一个接收消息的 endpoint。

📊 Telegram Bot vs 其他方案

方案	前端工作量	部署复杂度	用户获取	适合阶段
Telegram Bot	0（Telegram 自带）	低（一个 Python 进程）	发链接就能用	⭐ 第一个项目
Discord Bot	0	低	需要服务器邀请	第二个项目
Web App	高（需要写前端）	中	需要域名+SEO	第三个项目
微信小程序	高	高（审核）	高（微信生态）	产品化阶段

---

1️⃣ 最终效果：用户视角 📱

用户在 Telegram 里和你的 Bot 对话，Bot 背后是一个完整的 ReAct Agent：

用户: 帮我查一下北京明天的天气，如果下雨就提醒我带伞

Bot: 🔍 正在查询北京明天的天气...

Bot: 📊 查询结果：
北京明天：小雨转多云，15-22°C
🌂 建议带伞！上午有小雨，下午转多云。

---
用户: 帮我总结一下这篇文章 https://example.com/article

Bot: 📖 正在阅读文章...

Bot: 📝 文章摘要：
这篇文章讲了 3 个核心观点：
1. ...
2. ...
3. ...
关键结论：...

核心体验：用户感觉在和一个"能做事"的助手对话，而不是一个只会聊天的 ChatBot。

---

2️⃣ 技术架构 🏗

💡 一句话理解

Telegram 负责收发消息，你的服务负责 Agent 逻辑。 两者通过 Webhook 或 Polling 连接。

完整数据流

┌──────────┐     Webhook POST      ┌──────────────────┐
│ Telegram │ ──────────────────────→│   FastAPI 服务    │
│  用户端   │                        │                  │
│          │ ←─────────────────────│  ┌────────────┐  │
└──────────┘   Bot API 回复消息     │  │ ReAct Agent│  │
                                    │  │            │  │
                                    │  │ Tools:     │  │
                                    │  │ - 天气查询  │  │
                                    │  │ - 网页摘要  │  │
                                    │  │ - 计算器    │  │
                                    │  └────────────┘  │
                                    │                  │
                                    │  ┌────────────┐  │
                                    │  │ 对话历史    │  │
                                    │  │ (Redis/内存)│  │
                                    │  └────────────┘  │
                                    └──────────────────┘

两种接收消息的方式

方式	原理	优点	缺点	适合
Polling	你的代码每隔 N 秒问 Telegram"有新消息吗"	不需要公网 IP、不需要 HTTPS	有延迟、浪费资源	本地开发
Webhook	Telegram 主动把新消息 POST 到你的 URL	实时、省资源	需要公网 HTTPS URL	生产部署

开发阶段用 Polling，上线后切 Webhook。

---

3️⃣ 完整代码：从零到能跑 💻

Step 0：创建 Bot

1. 在 Telegram 搜索 @BotFather
2. 发送 /newbot
3. 按提示设置名字和用户名
4. 拿到 Bot Token（形如 123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11）

Step 1：项目结构

telegram-agent-bot/
├── src/
│   ├── bot.py              # Telegram 消息处理
│   ├── agent.py            # ReAct Agent 核心
│   ├── tools.py            # 工具定义
│   └── memory.py           # 对话历史管理
├── pyproject.toml
└── .env                    # BOT_TOKEN + OPENAI_API_KEY

Step 2：工具定义

# src/tools.py
import httpx
from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气预报。返回温度、天气状况、建议。"""
    # 用 OpenWeatherMap 免费 API
    api_key = "YOUR_OPENWEATHER_KEY"  # 免费注册即可
    url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric&lang=zh_cn"

    resp = httpx.get(url, timeout=10)
    if resp.status_code != 200:
        return f"查询失败：{resp.status_code}"

    data = resp.json()
    temp = data["main"]["temp"]
    desc = data["weather"][0]["description"]
    humidity = data["main"]["humidity"]

    return f"{city}：{desc}，温度 {temp}°C，湿度 {humidity}%"


@tool
def summarize_url(url: str) -> str:
    """读取网页内容并生成摘要。输入一个 URL，返回文章的核心要点。"""
    try:
        resp = httpx.get(url, timeout=15, follow_redirects=True)
        # 简单提取正文（生产环境用 newspaper3k 或 trafilatura）
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(resp.text, "html.parser")

        # 移除 script/style
        for tag in soup(["script", "style", "nav", "footer"]):
            tag.decompose()

        text = soup.get_text(separator="\n", strip=True)
        # 截断到 3000 字（避免 Token 爆炸）
        return text[:3000]
    except Exception as e:
        return f"无法读取该网页：{str(e)}"


@tool
def calculate(expression: str) -> str:
    """计算数学表达式。支持加减乘除、幂运算、括号。"""
    try:
        # 安全计算（不用 eval）
        import ast
        tree = ast.parse(expression, mode='eval')
        # 只允许数字和运算符
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                return "不支持函数调用，只支持基础运算"
        result = eval(compile(tree, '<string>', 'eval'))
        return f"{expression} = {result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

Step 3：Agent 核心

# src/agent.py
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from src.tools import get_weather, summarize_url, calculate
from src.memory import get_history, save_message

SYSTEM_PROMPT = """你是一个 Telegram 上的个人助手 Bot。你能帮用户：
1. 查天气（任何城市）
2. 总结网页文章
3. 做数学计算

回复规则：
- 用中文回复
- 简洁明了，不要废话
- 适当用 emoji 让回复更友好
- 如果用户的请求超出你的能力范围，诚实说明
"""

tools = [get_weather, summarize_url, calculate]
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)

agent = create_react_agent(llm, tools, prompt=SYSTEM_PROMPT)


async def handle_user_message(user_id: str, message: str) -> str:
    """处理用户消息，返回 Agent 回复"""
    # 加载对话历史（保持上下文）
    history = get_history(user_id, max_turns=5)

    messages = history + [("user", message)]

    try:
        result = agent.invoke({"messages": messages})
        reply = result["messages"][-1].content

        # 保存到历史
        save_message(user_id, "user", message)
        save_message(user_id, "assistant", reply)

        return reply
    except Exception as e:
        return f"处理出错了，请稍后再试 🙏\n错误信息：{type(e).__name__}"

Step 4：对话历史管理

# src/memory.py
from collections import defaultdict

# 简单内存存储（生产环境换 Redis）
_history: dict[str, list[tuple[str, str]]] = defaultdict(list)
MAX_HISTORY = 20  # 每个用户最多保留 20 条


def get_history(user_id: str, max_turns: int = 5) -> list[tuple[str, str]]:
    """获取用户最近 N 轮对话"""
    return _history[user_id][-(max_turns * 2):]


def save_message(user_id: str, role: str, content: str):
    """保存一条消息"""
    _history[user_id].append((role, content))
    # 超过上限就裁剪
    if len(_history[user_id]) > MAX_HISTORY:
        _history[user_id] = _history[user_id][-MAX_HISTORY:]


def clear_history(user_id: str):
    """清空用户对话历史"""
    _history[user_id] = []

Step 5：Telegram Bot 主程序

# src/bot.py
import asyncio
import logging
from telegram import Update
from telegram.ext import (
    Application, CommandHandler, MessageHandler, filters
)
from src.agent import handle_user_message
from src.memory import clear_history

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

BOT_TOKEN = "YOUR_BOT_TOKEN"  # 从 .env 读取


async def start(update: Update, context):
    """处理 /start 命令"""
    await update.message.reply_text(
        "👋 你好！我是 AI 助手 Bot。\n\n"
        "我能帮你：\n"
        "🌤 查天气（发"北京天气"）\n"
        "📖 总结文章（发一个链接）\n"
        "🧮 做计算（发算式）\n\n"
        "直接发消息就行，不用加前缀。\n"
        "发 /clear 清空对话记录。"
    )


async def clear(update: Update, context):
    """处理 /clear 命令"""
    user_id = str(update.effective_user.id)
    clear_history(user_id)
    await update.message.reply_text("✅ 对话记录已清空")


async def handle_message(update: Update, context):
    """处理普通文本消息"""
    user_id = str(update.effective_user.id)
    message = update.message.text

    logger.info(f"User {user_id}: {message}")

    # 发送"正在输入"状态
    await update.message.chat.send_action("typing")

    # 调用 Agent
    reply = await asyncio.to_thread(
        handle_user_message, user_id, message
    )

    logger.info(f"Bot reply to {user_id}: {reply[:100]}...")
    await update.message.reply_text(reply)


def main():
    app = Application.builder().token(BOT_TOKEN).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("clear", clear))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    logger.info("Bot started!")
    app.run_polling()


if __name__ == "__main__":
    main()

Step 6：依赖和环境

# pyproject.toml
[project]
name = "telegram-agent-bot"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "python-telegram-bot>=21.0",
    "langchain>=0.3.0",
    "langchain-openai>=0.2.0",
    "langgraph>=0.2.0",
    "httpx>=0.27.0",
    "beautifulsoup4>=4.12.0",
    "python-dotenv>=1.0.0",
]

# 启动
uv venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
uv pip install -e .
python -m src.bot

---

4️⃣ 生产化：从"能跑"到"能用" 🔧

必须加的 5 个补丁

补丁	为什么	实现
限流	防止单用户刷爆你的 API 额度	每用户每分钟最多 5 条消息
超时处理	Agent 思考太久用户会以为 Bot 挂了	30 秒超时 → 先回"处理中" → 异步等结果
错误兜底	Agent 报错不能把 traceback 发给用户	catch all → 友好提示
Token 预算	单次对话不能无限消耗 Token	限制 max_tokens + 历史裁剪
日志监控	出问题时能查到是哪一步挂了	记录 user_id + message + latency + tokens

限流实现

import time
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_calls: int = 5, window_seconds: int = 60):
        self.max_calls = max_calls
        self.window = window_seconds
        self._calls: dict[str, list[float]] = defaultdict(list)

    def is_allowed(self, user_id: str) -> bool:
        now = time.time()
        # 清理过期记录
        self._calls[user_id] = [
            t for t in self._calls[user_id]
            if now - t < self.window
        ]
        if len(self._calls[user_id]) >= self.max_calls:
            return False
        self._calls[user_id].append(now)
        return True

rate_limiter = RateLimiter(max_calls=5, window_seconds=60)

# 在 handle_message 里加：
if not rate_limiter.is_allowed(user_id):
    await update.message.reply_text("⏳ 你发得太快了，请稍等一分钟再试")
    return

超时处理

async def handle_message_with_timeout(update: Update, context):
    user_id = str(update.effective_user.id)
    message = update.message.text

    await update.message.chat.send_action("typing")

    try:
        reply = await asyncio.wait_for(
            asyncio.to_thread(handle_user_message, user_id, message),
            timeout=30.0,
        )
    except asyncio.TimeoutError:
        reply = "⏱ 这个问题比较复杂，处理超时了。请简化一下问题再试？"

    await update.message.reply_text(reply)

---

5️⃣ 部署：让 Bot 24 小时在线 🌐

方案对比

方案	成本	难度	适合
本地电脑	免费	极低	开发测试
Railway/Render	免费~$5/月	低	个人项目
阿里云/腾讯云 ECS	¥30-100/月	中	长期运行
Vercel Serverless	免费	中	轻量 Bot

Railway 一键部署（推荐新手）

# 1. 安装 Railway CLI
npm install -g @railway/cli

# 2. 登录
railway login

# 3. 初始化项目
railway init

# 4. 设置环境变量
railway variables set BOT_TOKEN=your_token
railway variables set OPENAI_API_KEY=your_key

# 5. 部署
railway up

切换到 Webhook 模式（生产推荐）

# 部署后把 Polling 换成 Webhook
from fastapi import FastAPI, Request
from telegram import Update

app = FastAPI()

WEBHOOK_URL = "https://your-app.railway.app/webhook"

@app.on_event("startup")
async def set_webhook():
    bot = Bot(token=BOT_TOKEN)
    await bot.set_webhook(WEBHOOK_URL)

@app.post("/webhook")
async def webhook(request: Request):
    data = await request.json()
    update = Update.de_json(data, bot)
    await application.process_update(update)
    return {"ok": True}

---

6️⃣ 简历怎么写这个项目 📝

简历表达模板

项目名称：AI Agent Telegram Bot

技术栈：Python / LangGraph / OpenAI API / Telegram Bot API / FastAPI

项目描述：

• 设计并部署了一个基于 ReAct 范式的 Telegram Agent Bot，支持多轮对话、工具调用（天气查询/网页摘要/计算）
• 实现了用户级对话历史管理、请求限流、超时兜底等生产化特性
• 日均处理 50+ 次对话，平均响应时间 < 3 秒，可用率 99.5%

亮点：

• 从 0 到上线仅 1 天，验证了 Agent 快速落地的可行性
• 通过限流 + Token 预算控制，月均 API 成本控制在 $5 以内

面试官可能追问的问题：

1. "你的 Agent 和普通 ChatBot 有什么区别？" → 答：能调用外部工具获取实时信息，不是纯靠模型内部知识
2. "对话历史怎么管理的？为什么不全部传给模型？" → 答：Token 成本 + Context Window 限制，只保留最近 5 轮
3. "如果用户量增长 10 倍怎么办？" → 答：内存存储换 Redis，单进程换多 Worker，加消息队列

---

7️⃣ 常见坑 🚨

坑	症状	修复
Webhook 收不到消息	Bot 无响应	检查 SSL 证书（Telegram 要求 HTTPS）；或先用 Polling
用户发超长消息	Token 爆炸 / 超时	截断用户输入 > 2000 字
Bot 被恶意刷消息	API 费用暴涨	加 user_id 级别 rate limit
免费云服务器休眠	Bot 隔一会儿不响应	用 UptimeRobot 每 5 分钟 ping
asyncio 和同步代码冲突	报错 "cannot run nested event loop"	Agent 调用包在 asyncio.to_thread() 里
多用户并发时对话串了	A 的回复发给了 B	确保 user_id 隔离，不要用全局变量存状态

---

🎓 总结

🎯 一句话总结

Telegram Bot = Agent 开发的 Hello World。Telegram 帮你做了前端和消息推送，你只需要写 Agent 逻辑。Polling 开发、Webhook 上线、Railway 部署——一天从零到有真实用户。简历上写"端到端 Agent 项目"，面试时能讲清楚 ReAct 循环和生产化补丁。

---

💡 自测题

1. Polling 和 Webhook 的区别是什么？各适合什么阶段？
2. 为什么对话历史要限制轮数？不限制会怎样？
3. 限流为什么按 user_id 而不是按 IP？
4. Agent 超时 30 秒还没回复，应该怎么处理？
5. 这个项目在面试时，面试官最可能追问什么？你怎么答？

答不上来的，回头再读一遍对应章节 ✍️

---

📌 转载声明

本文为🍋AI小柠檬原创。 转载请注明来源，付费产品包内禁止转载。