#!/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())