meeting-room/backend/command_interpreter.py

#!/usr/bin/env python3
"""
龙虾议事厅 - 自然语言命令解析器
让 AI 理解人类的自然语言命令
"""

import re
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum


class CommandType(Enum):
    """命令类型"""
    CREATE_MEETING = "create_meeting"
    SEND_MESSAGE = "send_message"
    LIST_MEETINGS = "list_meetings"
    JOIN_MEETING = "join_meeting"
    GET_MESSAGES = "get_messages"
    UNKNOWN = "unknown"


@dataclass
class ParsedCommand:
    """解析后的命令"""
    type: CommandType
    params: Dict[str, Any]
    confidence: float  # 置信度 0-1


class CommandInterpreter:
    """
    自然语言命令解析器
    
    支持的命令：
    - "创建一个会议，主题是 XXX"
    - "发送消息：XXX"
    - "查看会议列表"
    - "加入会议，邀请码 XXX"
    - "查看消息"
    """
    
    def __init__(self):
        # 命令模式匹配（简化版）
        self.patterns = {
            CommandType.CREATE_MEETING: [
                r'创建.*会议',
                r'新建.*会议',
                r'开.*会',
                r'组织.*会议',
            ],
            CommandType.SEND_MESSAGE: [
                r'发送.*消息',
                r'发消息',
                r'说.*',
                r'告诉.*',
            ],
            CommandType.LIST_MEETINGS: [
                r'查看.*列表',
                r'显示.*会议',
                r'有.*会议',
                r'我的会议',
            ],
            CommandType.JOIN_MEETING: [
                r'加入.*',
                r'进入.*',
                r'参加会议',
            ],
            CommandType.GET_MESSAGES: [
                r'查看.*消息',
                r'显示.*消息',
                r'有.*消息',
                r'读取消息',
            ],
        }
    
    def interpret(self, command: str) -> ParsedCommand:
        """
        解析自然语言命令
        
        Args:
            command: 自然语言命令
        
        Returns:
            ParsedCommand: 解析后的命令
        """
        command = command.strip()  # 不要转小写，保留中文
        
        # 尝试匹配各种命令模式
        for cmd_type, patterns in self.patterns.items():
            for pattern in patterns:
                if re.search(pattern, command, re.IGNORECASE):
                    params = self._extract_params(cmd_type, command)
                    return ParsedCommand(
                        type=cmd_type,
                        params=params,
                        confidence=0.8
                    )
        
        # 未匹配到任何模式
        return ParsedCommand(
            type=CommandType.UNKNOWN,
            params={'raw': command},
            confidence=0.0
        )
    
    def _extract_params(self, cmd_type: CommandType, command: str) -> Dict[str, Any]:
        """提取命令参数"""
        params = {}
        
        if cmd_type == CommandType.CREATE_MEETING:
            # 提取会议主题
            topic_match = re.search(r'主题是 (.+?)(?:，|,|。|$)', command)
            if topic_match:
                params['topic'] = topic_match.group(1).strip()
            else:
                # 尝试其他模式
                topic_match = re.search(r'创建.*?会议 (?:，|,)?(.+?)(?:。|$)', command)
                if topic_match:
                    params['topic'] = topic_match.group(1).strip()
                else:
                    params['topic'] = '新会议'
        
        elif cmd_type == CommandType.SEND_MESSAGE:
            # 提取消息内容
            content_match = re.search(r'(?::|：|说 | 告诉)(.+?)(?:。|$)', command)
            if content_match:
                params['content'] = content_match.group(1).strip()
            else:
                params['content'] = command
        
        elif cmd_type == CommandType.JOIN_MEETING:
            # 提取邀请码
            code_match = re.search(r'(?:邀请码 | 码 |code)[:：\s]?([A-Z0-9]+)', command, re.IGNORECASE)
            if code_match:
                params['invite_code'] = code_match.group(1).upper()
        
        return params


class AutonomousMeetingAgent:
    """
    自主会议 Agent
    结合命令解析和 AI SDK，实现自主操作
    """
    
    def __init__(self, api_base: str = 'http://localhost:8000'):
        from meeting_ai_sdk import MeetingAIOperations
        self.api = MeetingAIOperations(api_base)
        self.interpreter = CommandInterpreter()
        self.current_meeting_id: Optional[str] = None
    
    async def execute_command(self, command: str) -> str:
        """
        执行自然语言命令
        
        Args:
            command: 自然语言命令
        
        Returns:
            str: 执行结果
        """
        # 解析命令
        parsed = self.interpreter.interpret(command)
        
        if parsed.type == CommandType.UNKNOWN:
            return f"❌ 无法理解命令：{command}"
        
        # 执行对应操作
        if parsed.type == CommandType.CREATE_MEETING:
            return await self._create_meeting(parsed.params)
        
        elif parsed.type == CommandType.SEND_MESSAGE:
            return await self._send_message(parsed.params)
        
        elif parsed.type == CommandType.LIST_MEETINGS:
            return await self._list_meetings()
        
        elif parsed.type == CommandType.JOIN_MEETING:
            return await self._join_meeting(parsed.params)
        
        elif parsed.type == CommandType.GET_MESSAGES:
            return await self._get_messages()
        
        return "❌ 未知命令"
    
    async def _create_meeting(self, params: Dict) -> str:
        """创建会议"""
        topic = params.get('topic', '新会议')
        
        try:
            await self.api.login('test', 'test123')
            meeting = await self.api.create_meeting(topic)
            self.current_meeting_id = meeting.id
            return (
                f"✅ 会议已创建！\n"
                f"   主题：{meeting.topic}\n"
                f"   ID: {meeting.id}\n"
                f"   邀请码：{meeting.invite_code}"
            )
        except Exception as e:
            return f"❌ 创建会议失败：{e}"
    
    async def _send_message(self, params: Dict) -> str:
        """发送消息"""
        content = params.get('content', '')
        
        if not self.current_meeting_id:
            return "❌ 请先创建或加入会议"
        
        try:
            message = await self.api.send_message(self.current_meeting_id, content)
            return f"✅ 消息已发送：{message.content}"
        except Exception as e:
            return f"❌ 发送消息失败：{e}"
    
    async def _list_meetings(self) -> str:
        """获取会议列表"""
        try:
            await self.api.login('test', 'test123')
            meetings = await self.api.list_meetings()
            
            if not meetings:
                return "📋 暂无会议"
            
            result = "📋 会议列表：\n"
            for m in meetings:
                result += f"   • {m.topic} (邀请码：{m.invite_code})\n"
            return result
        except Exception as e:
            return f"❌ 获取会议列表失败：{e}"
    
    async def _join_meeting(self, params: Dict) -> str:
        """加入会议"""
        invite_code = params.get('invite_code')
        
        if not invite_code:
            return "❌ 请提供邀请码"
        
        try:
            await self.api.login('test', 'test123')
            meetings = await self.api.list_meetings()
            
            for m in meetings:
                if m.invite_code == invite_code:
                    self.current_meeting_id = m.id
                    return f"✅ 已加入会议：{m.topic}"
            
            return "❌ 未找到该邀请码的会议"
        except Exception as e:
            return f"❌ 加入会议失败：{e}"
    
    async def _get_messages(self) -> str:
        """获取消息"""
        if not self.current_meeting_id:
            return "❌ 请先创建或加入会议"
        
        try:
            messages = await self.api.get_messages(self.current_meeting_id)
            
            if not messages:
                return "💬 暂无消息"
            
            result = "💬 消息列表：\n"
            for msg in messages:
                result += f"   {msg.sender_name}: {msg.content}\n"
            return result
        except Exception as e:
            return f"❌ 获取消息失败：{e}"


# ============ 演示 ============

async def demo():
    """演示自然语言命令"""
    agent = AutonomousMeetingAgent('http://localhost:8000')
    
    print("=" * 60)
    print("🤖 自主会议 Agent - 自然语言命令演示")
    print("=" * 60)
    
    # 测试命令
    commands = [
        "创建一个会议，主题是 Q2 计划讨论",
        "查看会议列表",
        "发送消息：大家好！这是 AI 自动发送的消息",
        "查看消息",
    ]
    
    for cmd in commands:
        print(f"\n🗣️  人类命令：{cmd}")
        result = await agent.execute_command(cmd)
        print(f"🤖 AI 执行：{result}")
    
    print("\n" + "=" * 60)
    print("✅ 演示完成！")
    print("=" * 60)


if __name__ == '__main__':
    import asyncio
    asyncio.run(demo())