nanocode/weixin-nanobot-bridge/src/nanobot_adapter.py

#!/usr/bin/env python3
"""
Nanobot 调用适配器
负责将微信消息转换为 nanobot 调用
"""

import os
import subprocess
import tempfile
import logging
from typing import List, Dict

logger = logging.getLogger(__name__)


class NanobotAdapter:
    """nanobot 适配器"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.command = config.get('command', 'nanobot')
        self.timeout = config.get('timeout', 60)
        
    def chat(self, message: str, history: List[Dict], conversation_id: str) -> str:
        """
        调用 nanobot 处理消息
        
        Args:
            message: 当前消息
            history: 历史消息
            conversation_id: 会话ID
            
        Returns:
            nanobot 的回复
        """
        # 构建上下文提示
        context = self._build_context(history, conversation_id)
        
        # 构建完整提示
        prompt = self._build_prompt(context, message)
        
        logger.info(f"调用 nanobot，prompt长度: {len(prompt)}")
        
        try:
            # 调用 nanobot
            result = self._call_nanobot(prompt)
            return result
            
        except Exception as e:
            logger.error(f"调用 nanobot 失败: {e}")
            return f"处理消息时出错: {str(e)}"
    
    def _build_context(self, history: List[Dict], conversation_id: str) -> str:
        """构建上下文"""
        if not history:
            return ""
        
        lines = ["历史对话:"]
        for msg in history[-5:]:  # 最近5条
            role = "用户" if msg['role'] == 'user' else "助手"
            content = msg['content'][:100]  # 截断过长内容
            lines.append(f"{role}: {content}")
        
        return "\n".join(lines)
    
    def _build_prompt(self, context: str, message: str) -> str:
        """构建提示"""
        lines = [
            "你是一个 AI 助手，正在通过微信公众号为用户提供服务。",
            "",
            "规则:",
            "1. 保持回答简洁明了，适合微信聊天场景",
            "2. 如果需要执行复杂操作，先告知用户",
            "3. 不要执行危险命令（删除文件、格式化等）",
            "4. 回复长度控制在 4000 字符以内",
            "",
        ]
        
        if context:
            lines.extend([context, ""])
        
        lines.extend([
            "用户消息:",
            message,
            "",
            "请回复:"
        ])
        
        return "\n".join(lines)
    
    def _call_nanobot(self, prompt: str) -> str:
        """
        调用 nanobot agent 命令
        
        使用: nanobot agent -m "message" --session "id"
        """
        try:
            # 构建命令
            cmd = [
                self.command,
                "agent",
                "-m", prompt,
                "--session", f"wechat:{hash(prompt) % 10000}",
                "--no-markdown",
                "--no-logs"
            ]
            
            logger.info(f"执行命令: {' '.join(cmd[:5])}...")
            
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=self.timeout
            )
            
            if result.returncode == 0:
                return result.stdout.strip() or "处理完成"
            else:
                error_msg = result.stderr.strip() if result.stderr else "未知错误"
                return f"调用失败: {error_msg}"
                
        except subprocess.TimeoutExpired:
            return "处理超时，请稍后再试"
        except Exception as e:
            logger.error(f"调用异常: {e}")
            return f"系统错误: {str(e)}"
    
    def _parse_output(self, output: str) -> str:
        """解析 nanobot 输出"""
        lines = output.strip().split('\n')
        
        # 过滤掉日志和系统信息
        filtered = []
        for line in lines:
            # 跳过日志行
            if any(x in line for x in ['DEBUG', 'INFO', 'WARNING', 'ERROR']):
                continue
            # 跳过工具调用提示
            if line.startswith('>'):
                continue
            filtered.append(line)
        
        return '\n'.join(filtered).strip() or "处理完成，无返回内容"


class SimpleNanobotAdapter:
    """
    简化版适配器（用于测试）
    不依赖 nanobot 命令，直接返回模拟响应
    """
    
    def __init__(self, config: Dict):
        self.config = config
        self.response_templates = [
            "收到你的消息: {message}",
            "我理解你在问: {message}\n这是一个很有趣的问题！",
            "正在为你处理: {message}\n请稍候...",
        ]
    
    def chat(self, message: str, history: List[Dict], conversation_id: str) -> str:
        """模拟回复"""
        import random
        template = random.choice(self.response_templates)
        return template.format(message=message[:50])


# 兼容不同环境的适配器选择
try:
    # 检查 nanobot 是否可用
    result = subprocess.run(
        ['nanobot', '--version'],
        capture_output=True,
        timeout=5
    )
    NANOBOT_AVAILABLE = result.returncode == 0
except:
    NANOBOT_AVAILABLE = False

if not NANOBOT_AVAILABLE:
    logger.warning("nanobot 命令不可用，将使用模拟适配器")
    NanobotAdapter = SimpleNanobotAdapter