303 lines
9.3 KiB
Python
303 lines
9.3 KiB
Python
#!/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())
|