341 lines
12 KiB
Python
341 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Django management command for AI chat with readline support
|
||
Usage: python manage.py chat
|
||
"""
|
||
|
||
import os
|
||
import readline
|
||
import atexit
|
||
from django.core.management.base import BaseCommand
|
||
from products.aichat import AIChatService
|
||
|
||
|
||
class ReadlineConfig:
|
||
"""Readline配置类"""
|
||
|
||
def __init__(self, history_file: str = None):
|
||
self.history_file = history_file or os.path.expanduser("~/.aichat_history")
|
||
self.commands = ['help', 'history', 'clear', 'readline-help', 'stats', 'quit', 'exit', 'q']
|
||
self.setup_readline()
|
||
|
||
def setup_readline(self):
|
||
"""设置readline功能"""
|
||
# 设置历史文件
|
||
try:
|
||
readline.read_history_file(self.history_file)
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
# 设置历史文件大小
|
||
readline.set_history_length(1000)
|
||
|
||
# 注册退出时保存历史
|
||
atexit.register(self.save_history)
|
||
|
||
# 设置tab补全
|
||
readline.parse_and_bind('tab: complete')
|
||
|
||
# 设置其他有用的绑定
|
||
readline.parse_and_bind('set editing-mode emacs') # 使用emacs编辑模式
|
||
readline.parse_and_bind('set completion-ignore-case on') # 忽略大小写补全
|
||
readline.parse_and_bind('set show-all-if-ambiguous on') # 显示所有可能的补全
|
||
|
||
# 设置自定义补全器
|
||
readline.set_completer(self.completer)
|
||
|
||
# 设置补全分隔符
|
||
readline.set_completer_delims(' \t\n`!@#$%^&*()=+[{]}\\|;:\'",<>?')
|
||
|
||
def completer(self, text, state):
|
||
"""自定义补全器"""
|
||
options = [i for i in self.commands if i.startswith(text)]
|
||
if state < len(options):
|
||
return options[state]
|
||
return None
|
||
|
||
def save_history(self):
|
||
"""保存历史到文件"""
|
||
try:
|
||
readline.write_history_file(self.history_file)
|
||
except Exception as e:
|
||
print(f"⚠️ 保存历史记录失败: {e}")
|
||
|
||
def add_to_history(self, line: str):
|
||
"""添加行到历史"""
|
||
if line.strip():
|
||
readline.add_history(line.strip())
|
||
|
||
def get_history_stats(self):
|
||
"""获取历史统计信息"""
|
||
try:
|
||
history_length = readline.get_current_history_length()
|
||
return {
|
||
'current_length': history_length,
|
||
'max_length': readline.get_history_length(),
|
||
'history_file': self.history_file
|
||
}
|
||
except Exception:
|
||
return {'error': '无法获取历史统计信息'}
|
||
|
||
def clear_history(self):
|
||
"""清空readline历史"""
|
||
try:
|
||
readline.clear_history()
|
||
print("🗑️ Readline历史已清空")
|
||
except Exception as e:
|
||
print(f"⚠️ 清空历史失败: {e}")
|
||
|
||
|
||
class Command(BaseCommand):
|
||
"""AI聊天管理命令"""
|
||
|
||
help = '启动AI聊天系统，支持readline功能'
|
||
|
||
def add_arguments(self, parser):
|
||
"""添加命令行参数"""
|
||
parser.add_argument(
|
||
'--history-file',
|
||
type=str,
|
||
help='指定历史文件路径 (默认: ~/.aichat_history)'
|
||
)
|
||
parser.add_argument(
|
||
'--chat-type',
|
||
type=str,
|
||
choices=['platform', 'product'],
|
||
default='platform',
|
||
help='聊天类型: platform(平台客服) 或 product(产品客服) (默认: platform)'
|
||
)
|
||
parser.add_argument(
|
||
'--product-id',
|
||
type=int,
|
||
help='产品ID，当chat-type为product时必须提供'
|
||
)
|
||
|
||
def handle(self, *args, **options):
|
||
"""执行命令"""
|
||
# 参数验证
|
||
chat_type = options.get('chat_type')
|
||
product_id = options.get('product_id')
|
||
|
||
if chat_type == 'product' and not product_id:
|
||
self.stderr.write("❌ 错误: 当chat-type为product时，必须提供--product-id参数")
|
||
self.stderr.write("示例: python manage.py chat --chat-type product --product-id 12345")
|
||
return
|
||
|
||
# 显示系统信息
|
||
self.stdout.write("🚀 徵象防伪验证平台 AI客服系统")
|
||
self.stdout.write("=" * 50)
|
||
self.stdout.write(f"聊天类型: {chat_type.upper()}")
|
||
if product_id:
|
||
self.stdout.write(f"产品ID: {product_id}")
|
||
self.stdout.write("输入 'help' 查看帮助信息")
|
||
self.stdout.write("输入 'history' 查看对话历史")
|
||
self.stdout.write("输入 'clear' 清空对话历史")
|
||
self.stdout.write("输入 'test-rag' 测试知识库搜索功能")
|
||
self.stdout.write("输入 'quit' 或 'exit' 退出系统")
|
||
self.stdout.write("=" * 50)
|
||
self.stdout.write("💡 提示: 使用 ↑↓ 键浏览历史，Tab 键补全，Ctrl+A/E 跳转行首/尾")
|
||
self.stdout.write("=" * 50)
|
||
|
||
try:
|
||
# 初始化readline
|
||
readline_config = ReadlineConfig(options.get('history_file'))
|
||
|
||
# 初始化AI服务，传递聊天类型和产品ID
|
||
ai_service = AIChatService(
|
||
chat_type=chat_type,
|
||
product_id=product_id
|
||
)
|
||
self.stdout.write("✅ AI服务初始化成功！")
|
||
self.stdout.write("")
|
||
|
||
# 启动聊天循环
|
||
self._chat_loop(ai_service, readline_config)
|
||
|
||
except Exception as e:
|
||
self.stderr.write(f"❌ 系统初始化失败: {str(e)}")
|
||
self.stderr.write("请检查API密钥配置")
|
||
|
||
def _chat_loop(self, ai_service, readline_config):
|
||
"""聊天主循环"""
|
||
while True:
|
||
try:
|
||
# 获取用户输入
|
||
user_input = input("👤 您: ").strip()
|
||
|
||
if not user_input:
|
||
continue
|
||
|
||
# 添加到readline历史
|
||
readline_config.add_to_history(user_input)
|
||
|
||
# 处理特殊命令
|
||
if user_input.lower() in ['quit', 'exit', 'q']:
|
||
self.stdout.write("👋 感谢使用，再见！")
|
||
break
|
||
elif user_input.lower() == 'help':
|
||
self._print_help()
|
||
continue
|
||
elif user_input.lower() == 'history':
|
||
ai_service.print_history()
|
||
continue
|
||
elif user_input.lower() == 'clear':
|
||
ai_service.clear_history()
|
||
self.stdout.write("🗑️ 对话历史已清空")
|
||
continue
|
||
elif user_input.lower() == 'readline-help':
|
||
self._print_readline_help()
|
||
continue
|
||
elif user_input.lower() == 'stats':
|
||
self._print_readline_stats(readline_config)
|
||
continue
|
||
elif user_input.lower() == 'test-rag':
|
||
self._test_rag_function(ai_service)
|
||
continue
|
||
|
||
# 发送消息给AI
|
||
self.stdout.write("🤖 AI助手正在思考...")
|
||
response = ai_service.chat(user_input)
|
||
|
||
self.stdout.write(f"🤖 AI助手: {response}")
|
||
self.stdout.write("")
|
||
|
||
except KeyboardInterrupt:
|
||
self.stdout.write("\n\n👋 感谢使用，再见！")
|
||
break
|
||
except EOFError:
|
||
self.stdout.write("\n👋 再见！")
|
||
break
|
||
except Exception as e:
|
||
self.stderr.write(f"❌ 发生错误: {str(e)}")
|
||
self.stderr.write("请重试或输入 'quit' 退出")
|
||
|
||
def _print_help(self):
|
||
"""打印帮助信息"""
|
||
help_text = """
|
||
📖 帮助信息
|
||
==========
|
||
|
||
基本命令:
|
||
- 直接输入文字与AI助手对话
|
||
- help: 显示此帮助信息
|
||
- history: 查看对话历史
|
||
- clear: 清空对话历史
|
||
- readline-help: 显示readline快捷键帮助
|
||
- stats: 显示readline统计信息
|
||
- quit/exit/q: 退出系统
|
||
|
||
功能特性:
|
||
- 智能客服问答
|
||
- 二维码扫描验证（输入 '扫描' 或 '验证' 触发）
|
||
- 知识库检索（RAG功能）
|
||
- 防伪平台信息咨询
|
||
- 多轮对话支持
|
||
- 持久化输入历史记录
|
||
|
||
使用提示:
|
||
1. 可以询问平台功能、防伪验证等问题
|
||
2. 需要验证产品时，AI会主动提供扫描功能
|
||
3. 支持中文对话，表达自然即可
|
||
4. 使用 ↑↓ 键快速访问历史输入
|
||
"""
|
||
self.stdout.write(help_text)
|
||
|
||
def _print_readline_help(self):
|
||
"""打印readline快捷键帮助"""
|
||
readline_help = """
|
||
⌨️ Readline 快捷键帮助
|
||
=======================
|
||
|
||
导航快捷键:
|
||
↑/↓ - 浏览历史记录
|
||
Ctrl+A - 跳转到行首
|
||
Ctrl+E - 跳转到行尾
|
||
Ctrl+B - 向后移动一个字符
|
||
Ctrl+F - 向前移动一个字符
|
||
Ctrl+P - 上一个历史记录 (同 ↑)
|
||
Ctrl+N - 下一个历史记录 (同 ↓)
|
||
|
||
编辑快捷键:
|
||
Ctrl+U - 删除从光标到行首的内容
|
||
Ctrl+K - 删除从光标到行尾的内容
|
||
Ctrl+W - 删除光标前的一个单词
|
||
Ctrl+D - 删除光标后的一个字符
|
||
Ctrl+H - 删除光标前的一个字符 (同 Backspace)
|
||
|
||
历史搜索:
|
||
Ctrl+R - 反向搜索历史记录
|
||
Ctrl+S - 正向搜索历史记录
|
||
|
||
补全功能:
|
||
Tab - 自动补全
|
||
双击Tab - 显示所有可能的补全
|
||
|
||
其他:
|
||
Ctrl+L - 清屏
|
||
Ctrl+C - 中断当前操作
|
||
Ctrl+D - 退出 (EOF)
|
||
|
||
💡 提示: 这些快捷键基于 Emacs 编辑模式，与大多数终端应用保持一致
|
||
"""
|
||
self.stdout.write(readline_help)
|
||
|
||
def _print_readline_stats(self, readline_config):
|
||
"""打印readline统计信息"""
|
||
stats = readline_config.get_history_stats()
|
||
|
||
if 'error' in stats:
|
||
self.stderr.write(f"❌ {stats['error']}")
|
||
return
|
||
|
||
self.stdout.write("\n📊 Readline 统计信息")
|
||
self.stdout.write("=" * 30)
|
||
self.stdout.write(f"当前历史记录数: {stats['current_length']}")
|
||
self.stdout.write(f"最大历史记录数: {stats['max_length']}")
|
||
self.stdout.write(f"历史文件路径: {stats['history_file']}")
|
||
|
||
# 检查历史文件大小
|
||
try:
|
||
if os.path.exists(stats['history_file']):
|
||
file_size = os.path.getsize(stats['history_file'])
|
||
self.stdout.write(f"历史文件大小: {file_size} 字节")
|
||
else:
|
||
self.stdout.write("历史文件: 尚未创建")
|
||
except Exception:
|
||
self.stdout.write("历史文件: 无法获取信息")
|
||
|
||
self.stdout.write("=" * 30)
|
||
|
||
def _test_rag_function(self, ai_service):
|
||
"""测试RAG知识库搜索功能"""
|
||
self.stdout.write("\n🧪 测试RAG知识库搜索功能")
|
||
self.stdout.write("=" * 50)
|
||
|
||
try:
|
||
# 测试查询
|
||
test_queries = [
|
||
"徵象平台有什么功能？",
|
||
"如何验证产品真伪？",
|
||
"防伪技术原理是什么？"
|
||
]
|
||
|
||
for i, query in enumerate(test_queries, 1):
|
||
self.stdout.write(f"\n🔍 测试查询 {i}: {query}")
|
||
self.stdout.write("-" * 30)
|
||
|
||
try:
|
||
response = ai_service.chat(query)
|
||
self.stdout.write(f"✅ AI回复: {response[:200]}...")
|
||
except Exception as e:
|
||
self.stdout.write(f"❌ 查询失败: {str(e)}")
|
||
|
||
self.stdout.write("\n" + "=" * 50)
|
||
self.stdout.write("🎯 RAG功能测试完成！")
|
||
self.stdout.write("💡 提示: 如果AI使用了知识库搜索工具，说明RAG功能正常工作")
|
||
|
||
except Exception as e:
|
||
self.stderr.write(f"❌ RAG测试失败: {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|