nezdali/queenbee-agent

# queenbee-agent **一个运行在 Telegram 上的可自我扩展的 LLM agent，内置 runtime 和经过安全审查的** **工具工厂。** 大多数“Telegram + GPT”机器人只是一个简单的封装：接收消息，然后返回 LLM 回复。`queenbee-agent` 在结构上完全不同。该机器人是基于任何兼容 OpenAI 的模型构建的完整 agent loop，并带有模型可以调用的工具注册表。与众不同的是，**注册表不是固定的**： 终端用户可以用简单的英语描述一项新功能（`>> 描述该工具应执行的操作`），机器人就会 编写 Python 代码，通过禁止模式和 LLM 的安全审查，将其交给管理员进行 批准/拒绝，将其持久化为一等公民工具，并使其在下一轮对话中即可调用。 ## 它的功能 - **Telegram 上的 LLM agent loop** — 支持用户级对话历史记录、流式 响应、工具调用、图片输入、群聊 `@mention` 处理。 - **兼容 OpenAI** — 支持 OpenAI、Anthropic Claude（通过 OpenAI 兼容）、Google Gemini、Mistral、Groq、Together、OpenRouter 或本地 Ollama 实例。通过 `LLM_MODEL` / `OPENAI_BASE_URL` 选择模型。 - **Queen Bee 工具工厂** — 从简单的英语描述中在 runtime 生成新工具： 1. 对描述进行禁止意图正则表达式检查（文件列表、shell 执行、机密提取 — 管理员可绕过）。 2. codex 模型编写一个 `async def run(context) -> str` Python 模块 以及包含名称/描述/触发关键字/ 所需权限的 JSON 清单。 3. 扫描生成的代码中是否包含禁止模式 （`subprocess`、`os.system`、`eval`、文件删除等）。 4. 对于非管理员用户，将运行异步 LLM **安全审查**，管理员将获得批准/拒绝按钮。 5. 批准的工具会存入 `tools/.py`，并注册到 支持 RBAC 的工具注册表中，可在下一轮对话中调用。 - **四种调度工具的方式** 1. **LLM 函数调用** — 由模型在聊天过程中决定。 2. **`/runtool [args…]`** — 显式调用。 3. **触发关键字自动调度** — 用户消息的第一个词匹配 工具的 `trigger_keywords`。 4. **智能意图调度** — 如果消息不匹配关键字但 看起来像工具请求，机器人会提供带有 确认/拒绝按钮的工具建议。 - **基于用户的 RBAC** — 每个工具都有一个 `permission` 标签（`public`、 `finance`、`email`、`admin` 等）。用户只能查看和运行其角色允许的 工具。管理员可以看到所有内容。 - **计划任务 / 监控** — 工具可以在类似 cron 的计划上运行 （APScheduler）并将结果推送回给用户。 - **迭代循环** — `/qbtest`、`/qbfix `、`/qbsave`、 `/qbdiscard`、`/edittool `，用于在不离开 Telegram 的情况下完善 工具。 - **操作强化** — 使用 Azure Key Vault 加载器管理机密（包含 AWS Secrets Manager / GCP Secret Manager 方案），对工具生成进行基于用户的速率 限制、输出净化器、群聊允许列表。 ## 内置约 30 个示例工具 CoinGecko BTC 价格、天气、爱沙尼亚燃料/酒精/化妆品价格、 Amazon.de 搜索、Google Shopping 搜索、Booking.com / IKEA / JYSK 抓取器、Spotify 查找、Duolingo 个人资料统计、斯诺克 / FIFA / 足球 状态、Euribor 利率、Cozytouch / eWeLink / Huum 智能家居控制、 yt-dlp 媒体下载器、TLDR 摘要器等等 — 所有这些都在 `tools/` 中， 每个都是一个独立的 `async def run(context) -> str` 模块。 ## 前置条件 在开始之前，请准备好以下三样东西： ### 1. LLM API key 该机器人可与任何**兼容 OpenAI 的 Chat Completions endpoint** 通信，因此您 可以选择您偏好的任何提供商： | 提供商 | 获取密钥地址 | 需设置的 `OPENAI_BASE_URL` | |---|---|---| | **OpenAI**（默认） | | 保持未设置状态（默认为 `https://api.openai.com/v1`） | | **Anthropic Claude** | | `https://api.anthropic.com/v1/`（使用 [Anthropic 兼容 OpenAI 的 endpoint](https://docs.anthropic.com/en/api/openai-sdk)) | | **Google Gemini** | | `https://generativelanguage.googleapis.com/v1beta/openai/` | | **Mistral / Groq / Together / OpenRouter / 本地 Ollama** | 它们各自的仪表板 | 它们兼容 OpenAI 的 base URL | 无论您选择哪种提供商，请将密钥放入 `OPENAI_API_KEY` 中，并（如有需要） 在 `.env` 中设置 `OPENAI_BASE_URL`。您还可以使用 `OPENAI_DEFAULT_MODEL` 覆盖默认模型（例如 `gpt-4o-mini`、`claude-3-5-sonnet-latest`、 `gemini-2.0-flash`、`llama-3.3-70b-versatile` 等）。 ### 2. Telegram bot token + 您的 Telegram 用户 ID - 从 [@BotFather](https://t.me/BotFather) 获取 bot token （`/newbot` → 按照提示操作 → 复制 token）。 - 从 [@userinfobot](https://t.me/userinfobot) 获取您自己的数字 Telegram ID。 ### 3. 一台运行机器人的机器 该机器人是一个长时间运行的 Python 进程 — 它必须保持在线才能接收 Telegram 更新。对于本地测试，您的笔记本电脑即可；对于生产环境，您 需要一个小型始终在线的虚拟机（VM）。 任何小型 Linux VM 都可以（1 vCPU / 1 GB RAM 足矣）。最便宜的选项： | 提供商 | 最小实例层 | 大约费用 | 快速创建 | |---|---|---|---| | **Azure** | `Standard_B1s` Ubuntu 22.04 | 约 $8/月 | `az vm create -g rg -n tgbot --image Ubuntu2204 --size Standard_B1s --admin-username azureuser --generate-ssh-keys` | | **AWS EC2** | `t4g.nano` Ubuntu 22.04 (arm64) | 约 $3/月 | `aws ec2 run-instances --image-id ami-0... --instance-type t4g.nano --key-name mykey` | | **GCP Compute Engine** | `e2-micro` Ubuntu 22.04 | 符合免费层条件（美国区域） | `gcloud compute instances create tgbot --machine-type=e2-micro --image-family=ubuntu-2204-lts --image-project=ubuntu-os-cloud` | | **Hetzner / DigitalOcean / Vultr / Linode** | 共享 CPU 1 GB | 约 $4-5/月 | 使用它们的 Web 控制台 | | **Raspberry Pi / 家庭服务器** | 任意 | 如果您已经拥有则为 $0 | `ssh pi@` | VM 启动后： ``` ssh user@ sudo apt update && sudo apt install -y python3.11 python3.11-venv git git clone cd telegram_chatbot python3.11 -m venv .venv source .venv/bin/activate pip install -r requirements.txt cp .env.example .env && nano .env # paste your tokens python bot.py # smoke-test ``` 为了在您注销后保持机器人运行，请创建一个 systemd unit （`/etc/systemd/system/telegram-bot.service`）： ``` [Unit] Description=Telegram Chatbot After=network.target [Service] Type=simple User=youruser WorkingDirectory=/home/youruser/telegram_chatbot EnvironmentFile=/home/youruser/telegram_chatbot/.env ExecStart=/home/youruser/telegram_chatbot/.venv/bin/python bot.py Restart=always RestartSec=5 [Install] WantedBy=multi-user.target ``` ``` sudo systemctl daemon-reload sudo systemctl enable --now telegram-bot journalctl -u telegram-bot -f # follow logs ``` ## 快速开始 ``` git clone cd telegram_chatbot python -m venv .venv source .venv/bin/activate # or .venv\Scripts\activate on Windows pip install -r requirements.txt cp .env.example .env # 编辑 .env：设置 TELEGRAM_BOT_TOKEN, OPENAI_API_KEY, QB_ADMIN_USER_ID python bot.py ``` 从 [@BotFather](https://t.me/BotFather) 获取 bot token。从 [@userinfobot](https://t.me/userinfobot) 获取您自己的 Telegram 数字用户 ID，并 将其放入 `QB_ADMIN_USER_ID` 中。 ## 架构 ``` bot.py — entry point: registers handlers, starts polling config.py — env-var + (optional) Azure Key Vault loading tool_utils.py — fetch_json/text/html/rendered helpers for tools core/ tool_factory.py — Queen Bee: generate / edit / run tools tool_match.py — keyword scoring for auto-dispatch agent_tools.py — built-in LLM tools (list_tools, fetch_url, run_tool) conversation.py — per-user history + model preference output_safety.py — global output sanitiser tool_registry.py — RBAC-aware tool registry user_roles.py — persistent JSON user→roles store handlers/ core.py — /start, /help, /clear, /model, handle_message queen_bee.py — >>, /tools, /runtool, /qbtest, /qbsave, ... admin_users.py — /adduser, /deluser, /listusers, /setrole, ... tool_dispatch.py — thinking phrases + translation helper strava_handlers.py — /stravaconnect, /stravaauth, /stravahelp group.py — auto-leave non-allowed groups utils.py — shared helpers, ConversationManager services/ llm_service.py — OpenAI API wrapper (streaming + tool-calling) model_router.py — picks chat / codex / vision model per task strava_service.py — Strava OAuth + REST wrapper (per-user tokens) tools/ — 29 pre-built tools + your own generated ones (see tools/README.md for the list) ``` ## Queen Bee 工具工厂 在与机器人的私聊中： ``` >> fetch the current Bitcoin price from CoinGecko and format it nicely ``` 机器人将执行以下操作： 1. 根据禁止意图正则表达式列表（文件列表、 shell 执行、机密提取等）验证请求 — 管理员用户可绕过此步骤。 2. 要求 LLM 生成一个 `async def run(context: dict) -> str` Python 模块。 3. 扫描生成的代码中是否包含禁止模式（`subprocess`、`os.system`、 `eval`、文件删除等） — 管理员用户可绕过此步骤。 4. 对于非管理员用户：将工具保存为 `pending_review`，触发异步 LLM 安全审查，并向管理员提供**批准 / 拒绝**按钮。 5. 一旦批准，该工具即可通过以下方式调用： - `/runtool ` — 显式调用 - 首词关键字匹配（每个工具都有 `trigger_keywords`） - LLM 本身，通过 `run_tool` 工具（按用户进行 RBAC 过滤） ### 管理工具 | 命令 | 用途 | |---|---| | `/tools` | 列出已保存的工具 | | `/toolhelp` | 全面的操作指南 + 每个工具的描述 | | `/toolhelp ` | 某个工具的完整详情 | | `/newtool` | 关于如何创建新工具的详细步骤 | | `/runtool ` | 显式运行工具 | | `/deltool ` (管理员) | 删除工具 | | `/edittool ` | 要求 LLM 修改现有工具 | | `/qbtest ` | 生成 + 运行工具但不保存 | | `/qbsave` | 保存上次测试生成的工具 | | `/qbdiscard` | 将其丢弃 | | `/qbfix ` | 要求 LLM 修复上次的测试工具 | | `/qbdebug` | 显示上次生成的代码 | ## RBAC 角色存在于 `config.ROLE_PERMISSIONS` 中： ``` ROLE_PERMISSIONS = { "admin": ["*"], # everything "public": ["public"], # default } ``` 添加您自己的角色（`finance`、`email`、`ops` 等），并通过以下方式分配给用户： ``` /adduser 123456789 finance,email @alice /setrole my_tool finance # only users with the `finance` role may run it ``` 内置工具（`list_tools`、`fetch_url`、`run_tool`）均为 `public`。 生成的工具默认为 `public`；管理员可以通过 `/setrole` 提升其权限。 ## 添加您自己的领域工具 要将自定义工具（例如银行 API）接入 LLM 协调器，请在 `core/tool_registry.py` 中注册它，包含 schema + handler + 权限标签： ``` # 在你的模块的 init 中 import core.tool_registry as tr tr.register( "my_tool", schema={"type": "function", "function": {"name": "my_tool", "description": "...", "parameters": {...}}}, handler=async_handler, permission="finance", # only users with a role mapping to 'finance' see it ) ``` LLM 只能看到调用用户拥有权限的工具。 ## 安全 - 生成的工具通过 `importlib` 在进程内运行；它们没有被沙箱化。 防御措施包括： - 针对非管理员用户的意图正则表达式（LLM 前） + 代码正则表达式（LLM 后） - 后台 LLM 安全审查与管理员批准 - `pending_review` 和 `rejected` 状态会阻止执行 - 每次工具调用有 90 秒的硬性超时 - 禁止列表位于 `core/tool_factory.py` 中 （`_FORBIDDEN_INTENT_PATTERNS`、`_FORBIDDEN_PATTERNS`）。 - 切勿将 `QB_ADMIN_USER_ID` 设置为您不完全信任的用户 — 管理员可绕过 所有检查，且其生成的代码将未经核实直接运行。 - Telegram token、OpenAI 密钥等仅从环境中读取。 `.gitignore` 排除了 `.env`、所有 `*token*.json`、所有 `*credentials*.json` 和所有 `*.pem`/`*.key` 文件。 ## 可选集成 该仓库内置了一些与第三方服务通信的工具和处理程序。**所有这些功能默认关闭** — 将相应的环境变量留空，集成将被静默跳过。设置它们以启用。 | 集成 | 您将获得的功能 | 环境变量 | | --- | --- | --- | | Strava | `/stravaconnect`、`/stravaauth`，以及 4 个 LLM 工具（运动员、活动、单次活动区间、kudos） | `STRAVA_CLIENT_ID`, `STRAVA_CLIENT_SECRET` | | Spotify | `spotify_lookup_tool`（曲目 / 艺术家 / 专辑查找） | `SPOTIFY_CLIENT_ID`, `SPOTIFY_CLIENT_SECRET`, `SPOTIFY_REDIRECT_URI` | | Duolingo | `duolingo_stats_tool`（公开个人资料查找 — 无需认证） | — | | eWeLink | `ewelink_smart_home` 工具（控制 Sonoff 设备） | `EWELINK_EMAIL`, `EWELINK_PASSWORD`, `EWELINK_REGION` | | Huum 桑拿 | `huum_sauna` 工具（状态 / 开启加热 / 关闭加热） | `HUUM_USERNAME`, `HUUM_PASSWORD` | | Cozytouch 热泵 | `cozytouchpump` 工具（温度和模式） | `COZYTOUCH_USERNAME`, `COZYTOUCH_PASSWORD` | | TMDB | `movie_search` 工具（电影元数据） | `TMDB_API_KEY` | | SerpAPI | `amazon_search`、`google_shopping_search`、`booking_com_search` | `SERPAPI_KEY` | | OpenAI 使用情况 | `openai_api_usage_status` 工具（费用报告） | `OPENAI_ADMIN_API_KEY` | | ElevenLabs | `media_creator` 中的可选 TTS | `ELEVENLABS_API_KEY` | | Yandex 翻译 | 可选的翻译助手 | `YANDEX_API_KEY` | ### Strava — 连接用户 1. 访问 → **Create & Manage Your App**。 2. 将 **Authorization Callback Domain** 设置为 `localhost`。 3. 将您的 Client ID + Client Secret 复制到 `.env` 中： STRAVA_CLIENT_ID=12345 STRAVA_CLIENT_SECRET=... 4. 重启机器人并在 Telegram 中运行 `/stravaconnect`。 5. **在外部浏览器中打开链接**（Telegram 的应用内浏览器会被 Strava 阻止）。授权，然后从生成的 URL 中复制 `code=` 的值并运行： /stravaauth 6. token 将保存为 `strava_token_.json`（位置可通过 `STRAVA_TOKEN_DIR` 覆盖）。 连接后，您可以向 LLM 询问诸如*“我上一次骑行怎么样？”*或 *“我昨天在功率区间 5 花了多少时间？”*之类的问题。LLM 会自动选择 正确的 Strava 工具。 要让用户询问彼此的情况（*“Alice 的上一次骑行怎么样？”*），请填写 `KNOWN_USERS` 环境变量： ``` KNOWN_USERS=alice:123456789;al:123456789;bob:987654321 ``` ### Spotify 1. 在 创建一个应用。 2. 添加重定向 URI（例如 `http://localhost:8888/callback`）。 3. 填写： SPOTIFY_CLIENT_ID=... SPOTIFY_CLIENT_SECRET=... SPOTIFY_REDIRECT_URI=http://localhost:8888/callback 4. 运行一次 `spotify_lookup_tool` — 它将打印一个授权 URL 并引导您粘贴回重定向后的 URL。 ### eWeLink (Sonoff) 使用您用于 eWeLink 移动应用的相同电子邮件/密码： ``` EWELINK_EMAIL=you@example.com EWELINK_PASSWORD=... EWELINK_REGION=eu # one of: us | eu | as | cn # 可选：使用手机登录代替邮箱 # EWELINK_PHONE=+3725551234 ``` ### Huum 桑拿 ``` HUUM_USERNAME=you@example.com HUUM_PASSWORD=... ``` 该工具查询 Huum 的 REST API；凭据与 Huum 移动应用相同。 ### Atlantic / Cozytouch 热泵 ``` COZYTOUCH_USERNAME=you@example.com COZYTOUCH_PASSWORD=... ``` 使用 Atlantic / Cozytouch IO 门户；凭据与移动应用相同。 ### Duolingo 无需认证 — 该工具使用 Duolingo 的公开个人资料 API： ``` duolingo ``` ### TMDB (movie_search 工具) 在 获取免费的 v3 API key 并设置： ``` TMDB_API_KEY=... ``` 如果 `yt-dlp` 在 `$PATH` 中，YouTube 预告片提取将使用它；可以选择将 `youtube_cookies.txt` 文件放在 `bot.py` 旁边以绕过机器人检测。 ### OpenAI 使用报告 在 生成一个单独的 **Admin API key**（组织范围， `/v1/organization/usage/*` endpoint 必需），并设置： ``` OPENAI_ADMIN_API_KEY=sk-admin-... ``` ## 可选：密钥管理器（Azure / AWS / GCP） 对于生产环境，您通常不希望在 VM 上的 `.env` 中存放明文 API 密钥。选择以下云密钥管理器之一，将 `config._KV_SECRET_MAP` 中列出的每个值存储在那里，并授予您的 VM 身份读取权限。 ### Azure Key Vault（内置） 设置 `AZURE_KEYVAULT_URL=https://your-vault.vault.azure.net/`，机器人将在 启动时使用 `DefaultAzureCredential`（在生产环境中为 Managed Identity，在 本地为 `az login`）从您的保险库中提取列在 `config._KV_SECRET_MAP` 中的环境变量。保持未设置状态则仅使用 `.env`。 创建保险库： ``` az group create -n tgbot-rg -l northeurope az keyvault create -n tgbot-kv -g tgbot-rg -l northeurope az keyvault secret set --vault-name tgbot-kv --name TELEGRAM-BOT-TOKEN --value '...' az keyvault secret set --vault-name tgbot-kv --name OPENAI-API-KEY --value 'sk-...' # 授予 VM 的 managed identity 读取权限 az vm identity assign -g tgbot-rg -n tgbot VM_PRINCIPAL=$(az vm show -g tgbot-rg -n tgbot --query identity.principalId -o tsv) az role assignment create --role "Key Vault Secrets User" \ --assignee "$VM_PRINCIPAL" \ --scope $(az keyvault show -n tgbot-kv --query id -o tsv) ``` 机器人期望保险库密钥名称带有短划线（例如 `OPENAI-API-KEY`）；它们将在启动时映射到带有下划线的环境变量。 ### AWS Secrets Manager（DIY） 该仓库内置了 Azure 加载器，但相同的模式也适用于 AWS — 您只需 将一个小型加载器插入 `config.py`。 创建密钥库： ``` # 1. 创建一个包含所有 key/value 的 JSON secret aws secretsmanager create-secret \ --name tgbot/prod \ --secret-string '{"TELEGRAM_BOT_TOKEN":"...","OPENAI_API_KEY":"sk-...","QB_ADMIN_USER_ID":"123456"}' # 2. 创建一个可以读取它的 IAM role 并附加到你的 EC2 instance ROLE_ARN=$(aws iam create-role --role-name tgbot-secrets-reader \ --assume-role-policy-document file://trust-policy.json --query Role.Arn -o tsv) aws iam put-role-policy --role-name tgbot-secrets-reader \ --policy-name read-tgbot-secret \ --policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":"secretsmanager:GetSecretValue","Resource":"arn:aws:secretsmanager:*:*:secret:tgbot/prod-*"}]}' aws ec2 associate-iam-instance-profile --instance-id i-... \ --iam-instance-profile Name=tgbot-secrets-reader ``` 然后在 Azure 加载器附近扩展 `config.py`，例如： ``` if aws_secret := os.getenv("AWS_SECRET_ID"): import boto3, json as _json blob = boto3.client("secretsmanager").get_secret_value(SecretId=aws_secret) for k, v in _json.loads(blob["SecretString"]).items(): os.environ.setdefault(k, v) ``` ### Google Cloud Secret Manager（DIY） 相同思路，GCP 风格： ``` # 1. 启用 API 并为每个 key/value 创建一个 secret gcloud services enable secretmanager.googleapis.com printf 'sk-...' | gcloud secrets create OPENAI_API_KEY --data-file=- printf '...' | gcloud secrets create TELEGRAM_BOT_TOKEN --data-file=- # 2. 给予 VM 的 service account 读取权限 SA=$(gcloud compute instances describe tgbot --zone=europe-north1-a \ --format='value(serviceAccounts[0].email)') gcloud secrets add-iam-policy-binding OPENAI_API_KEY \ --member="serviceAccount:$SA" --role="roles/secretmanager.secretAccessor" gcloud secrets add-iam-policy-binding TELEGRAM_BOT_TOKEN \ --member="serviceAccount:$SA" --role="roles/secretmanager.secretAccessor" ``` 然后扩展 `config.py`： ``` if gcp_project := os.getenv("GCP_PROJECT_ID"): from google.cloud import secretmanager client = secretmanager.SecretManagerServiceClient() for key in ("OPENAI_API_KEY", "TELEGRAM_BOT_TOKEN", ...): name = f"projects/{gcp_project}/secrets/{key}/versions/latest" os.environ.setdefault(key, client.access_secret_version(name=name).payload.data.decode()) ``` 使用上述三种方式中的任何一种，VM 都永远不会看到明文 `.env`，轮换 密钥只需在保险库中更新，然后执行 `systemctl restart telegram-bot`。 ## 许可证 MIT — 查看 [LICENSE](./LICENSE)。