#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Agent模式Telegram Bot - 使用Anthropic SDK实现工具调用和决策循环 100% 虚拟机运行，使用Sonnet 4.5 """ import os import json import logging import asyncio from datetime import datetime from typing import Dict, List, Any, Optional from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, filters, ContextTypes import anthropic from pyrogram import Client # 日志配置 logging.basicConfig( format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO, handlers=[ logging.FileHandler('agent_bot.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # ===== 配置 ===== TELEGRAM_TOKEN = "8426529617:AAHAxzohSMFBAxInzbAVJsZfkB5bHnOyFC4" SEARCH_BOT_USERNAME = "openaiw_bot" # Claude API配置 try: CLAUDE_CLIENT = anthropic.Anthropic( auth_token=os.environ.get('ANTHROPIC_AUTH_TOKEN'), base_url=os.environ.get('ANTHROPIC_BASE_URL', 'https://api.anthropic.com') ) logger.info("✅ Claude Agent客户端初始化成功") except Exception as e: logger.error(f"❌ Claude客户端初始化失败: {e}") CLAUDE_CLIENT = None # ===== 工具定义 ===== TOOLS = [ { "name": "search_telegram_groups", "description": "在Telegram中搜索群组。当用户想要查找群组、频道或者需要搜索特定关键词时使用此工具。", "input_schema": { "type": "object", "properties": { "keyword": { "type": "string", "description": "搜索关键词，例如 'AI'、'翻译'、'编程' 等" }, "search_type": { "type": "string", "enum": ["groups", "text", "human", "topchat"], "description": "搜索类型：groups=群组名称，text=讨论内容，human=用户，topchat=热门分类", "default": "groups" } }, "required": ["keyword"] } }, { "name": "get_cached_results", "description": "从数据库获取已缓存的搜索结果。用于快速返回之前搜索过的内容。", "input_schema": { "type": "object", "properties": { "command": { "type": "string", "description": "搜索命令，如 'search'、'text' 等" }, "keyword": { "type": "string", "description": "搜索关键词" } }, "required": ["command", "keyword"] } } ] # ===== Agent决策引擎 ===== class ClaudeAgent: """Claude Agent - 带工具调用和决策循环""" def __init__(self): self.client = CLAUDE_CLIENT self.model = "claude-sonnet-4-5-20250929" self.max_tokens = 2048 self.conversations: Dict[int, List[Dict]] = {} # 用户对话历史 self.max_history = 10 logger.info("✅ Claude Agent引擎初始化完成") def get_history(self, user_id: int) -> List[Dict]: """获取用户对话历史""" if user_id not in self.conversations: self.conversations[user_id] = [] return self.conversations[user_id][-self.max_history:] def add_to_history(self, user_id: int, role: str, content: Any): """添加到对话历史""" if user_id not in self.conversations: self.conversations[user_id] = [] self.conversations[user_id].append({"role": role, "content": content}) async def think_and_act(self, user_id: int, user_message: str) -> Dict[str, Any]: """ 决策循环：思考 -> 选择工具 -> 执行 -> 返回结果 返回: { "response": "AI回复文本", "tools_used": [{"name": "tool_name", "input": {...}, "result": ...}], "buttons": [{"text": "...", "callback_data": "..."}] } """ logger.info(f"[Agent] 用户 {user_id} 发起对话: {user_message}") # 构建消息历史 history = self.get_history(user_id) messages = history + [{"role": "user", "content": user_message}] try: # 第一轮：调用Claude获取决策 logger.info(f"[Agent] 调用Claude API（带工具）") response = self.client.messages.create( model=self.model, max_tokens=self.max_tokens, tools=TOOLS, messages=messages ) logger.info(f"[Agent] Claude响应类型: {response.stop_reason}") # 处理工具调用 tools_used = [] final_text = "" if response.stop_reason == "tool_use": # Claude决定使用工具 logger.info(f"[Agent] Claude决定使用工具") # 提取工具调用和文本 tool_results = [] for block in response.content: if block.type == "text": final_text += block.text elif block.type == "tool_use": logger.info(f"[Agent] 工具调用: {block.name} - {block.input}") # 执行工具 tool_result = await self._execute_tool(block.name, block.input) tools_used.append({ "name": block.name, "input": block.input, "result": tool_result }) # 准备工具结果给Claude tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(tool_result, ensure_ascii=False) }) # 第二轮：将工具结果返回给Claude if tool_results: logger.info(f"[Agent] 将工具结果返回给Claude") messages.append({"role": "assistant", "content": response.content}) messages.append({"role": "user", "content": tool_results}) # 再次调用Claude获取最终回复 final_response = self.client.messages.create( model=self.model, max_tokens=self.max_tokens, tools=TOOLS, messages=messages ) # 提取最终文本 for block in final_response.content: if block.type == "text": final_text += block.text else: # 直接回复，无需工具 for block in response.content: if block.type == "text": final_text += block.text # 保存对话历史 self.add_to_history(user_id, "user", user_message) self.add_to_history(user_id, "assistant", final_text) # 提取按钮 buttons = self._extract_buttons(final_text) logger.info(f"[Agent] ✅ 完成决策循环，使用了 {len(tools_used)} 个工具") return { "response": final_text, "tools_used": tools_used, "buttons": buttons } except Exception as e: logger.error(f"[Agent] ❌ 决策失败: {e}") return { "response": f"抱歉，我遇到了一些问题：{str(e)}", "tools_used": [], "buttons": [] } async def _execute_tool(self, tool_name: str, tool_input: Dict) -> Any: """执行工具调用""" logger.info(f"[工具执行] {tool_name}({tool_input})") if tool_name == "search_telegram_groups": keyword = tool_input.get("keyword", "") search_type = tool_input.get("search_type", "groups") # 调用实际搜索（通过Pyrogram镜像） result = await self._perform_telegram_search(keyword, search_type) return result elif tool_name == "get_cached_results": command = tool_input.get("command", "") keyword = tool_input.get("keyword", "") # 从数据库获取缓存 # TODO: 实际连接数据库 return { "status": "success", "cached": True, "results": [] } return {"status": "unknown_tool"} async def _perform_telegram_search(self, keyword: str, search_type: str) -> Dict: """执行Telegram搜索（镜像openaiw_bot）""" # TODO: 实际通过Pyrogram发送搜索命令 logger.info(f"[搜索] 类型={search_type}, 关键词={keyword}") # 模拟返回结果 return { "status": "success", "keyword": keyword, "search_type": search_type, "results_count": 5, "message": f"搜索 '{keyword}' 完成" } def _extract_buttons(self, text: str) -> List[Dict[str, str]]: """从AI回复中提取可点击按钮""" buttons = [] # 提取命令格式：/search xxx, /text xxx import re patterns = [ r'/search\s+(\S+)', r'/text\s+(\S+)', r'/human\s+(\S+)', r'/topchat' ] for pattern in patterns: matches = re.findall(pattern, text) for match in matches: if pattern == r'/topchat': buttons.append({ "text": "🔥 热门分类", "callback_data": "cmd_topchat" }) else: cmd = pattern.split('\\s')[0].replace('/', '') buttons.append({ "text": f"🔍 {cmd} {match}", "callback_data": f"cmd_{cmd}_{match}"[:64] }) return buttons # ===== Bot处理器 ===== class AgentBot: """Agent模式Telegram Bot""" def __init__(self): self.agent = ClaudeAgent() self.app = None logger.info("✅ Agent Bot初始化完成") async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """处理 /start 命令""" user_id = update.effective_user.id logger.info(f"[命令] 用户 {user_id} 启动Bot") welcome = ( "👋 你好！我是AI Agent Bot\n\n" "💡 我可以帮你：\n" "- 🔍 智能搜索Telegram群组\n" "- 💬 自然语言对话\n" "- 🤖 自动选择合适的工具\n\n" "直接告诉我你想做什么吧！" ) await update.message.reply_text(welcome) async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """处理用户消息 - Agent决策入口""" user_id = update.effective_user.id user_message = update.message.text logger.info(f"[消息] 用户 {user_id}: {user_message}") # 调用Agent决策循环 result = await self.agent.think_and_act(user_id, user_message) # 构建回复 response_text = result["response"] buttons = result["buttons"] tools_used = result["tools_used"] # 添加工具使用信息 if tools_used: tool_info = "\n\n🔧 使用的工具:\n" for tool in tools_used: tool_info += f"- {tool['name']}\n" response_text += tool_info # 发送回复（带按钮） if buttons: keyboard = [[InlineKeyboardButton(btn["text"], callback_data=btn["callback_data"])] for btn in buttons] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text( response_text, reply_markup=reply_markup ) logger.info(f"[回复] 已发送（带 {len(buttons)} 个按钮）") else: await update.message.reply_text(response_text) logger.info(f"[回复] 已发送") async def handle_button(self, update: Update, context: ContextTypes.DEFAULT_TYPE): """处理按钮点击""" query = update.callback_query await query.answer() callback_data = query.data logger.info(f"[按钮] 用户点击: {callback_data}") # 解析按钮命令 if callback_data.startswith("cmd_"): parts = callback_data[4:].split("_") command = parts[0] keyword = "_".join(parts[1:]) if len(parts) > 1 else "" # 将按钮点击转换为消息，重新进入Agent决策 user_message = f"/{command} {keyword}".strip() user_id = query.from_user.id logger.info(f"[按钮->命令] 转换为消息: {user_message}") result = await self.agent.think_and_act(user_id, user_message) await query.message.reply_text(result["response"]) def run(self): """启动Bot""" logger.info("🚀 启动Agent Bot...") # 创建Application self.app = Application.builder().token(TELEGRAM_TOKEN).build() # 注册处理器 self.app.add_handler(CommandHandler("start", self.start_command)) self.app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_message)) self.app.add_handler(CallbackQueryHandler(self.handle_button)) # 启动轮询 logger.info("✅ Agent Bot已启动，等待用户消息...") self.app.run_polling(allowed_updates=Update.ALL_TYPES) # ===== 主入口 ===== if __name__ == "__main__": logger.info("=" * 60) logger.info("🤖 Claude Agent Bot - 启动中") logger.info(f"📅 时间: {datetime.now()}") logger.info(f"🔑 Auth Token: {os.environ.get('ANTHROPIC_AUTH_TOKEN', 'NOT SET')[:20]}...") logger.info(f"🌐 Base URL: {os.environ.get('ANTHROPIC_BASE_URL', 'NOT SET')}") logger.info("=" * 60) if not CLAUDE_CLIENT: logger.error("❌ Claude客户端未初始化，无法启动") exit(1) bot = AgentBot() bot.run()