"""
ERNIE-4.5 心理健康机器人命令行聊天界面
支持与本地部署的 ERNIE-4.5 模型进行交互，包含心理健康微调功能
"""
import json
import requests
import sys
import os
import time
from datetime import datetime
import argparse
import signal
import random
import re
from typing import List, Dict, Any
class PsychologyFineTuner:
"""心理健康微调模块"""
def __init__(self):
self.psychology_patterns = {
'anxiety': ['焦虑', '紧张', '担心', '不安', '恐惧', '害怕', '慌张'],
'depression': ['抑郁', '沮丧', '绝望', '无助', '悲伤', '难过', '低落'],
'stress': ['压力', '疲惫', '累', '烦躁', '急躁', '烦恼', '困扰'],
'anger': ['愤怒', '生气', '气愤', '恼火', '暴躁', '发火', '怒'],
'loneliness': ['孤独', '寂寞', '独自', '一个人', '没朋友', '孤单'],
'confusion': ['困惑', '迷茫', '不知道', '混乱', '不明白', '茫然']
}
self.empathy_responses = {
'anxiety': [
"我理解你现在感到焦虑，这种感觉很不舒服。",
"焦虑是一种很常见的情绪反应，你并不孤单。",
"我能感受到你的紧张，让我们一起来面对这种感觉。"
],
'depression': [
"我能理解你现在的低落情绪，这一定很难受。",
"感到悲伤是正常的，请不要因此责备自己。",
"虽然现在很困难，但请相信这种感觉会过去的。"
],
'stress': [
"我理解你现在承受着很大的压力。",
"压力确实会让人感到疲惫，这是很正常的反应。",
"让我们一起寻找缓解压力的方法。"
],
'anger': [
"我理解你现在很愤怒，这种情绪是可以理解的。",
"愤怒是一种正常的情绪，重要的是如何处理它。",
"我能感受到你的愤怒，让我们谈谈是什么让你有这种感觉。"
],
'loneliness': [
"我理解孤独感是很难受的，你现在并不是一个人。",
"感到孤独是很多人都会经历的情感。",
"虽然你感到孤独，但请记住总有人关心你。"
],
'confusion': [
"我理解你现在感到困惑，这种不确定感确实不好受。",
"困惑是成长过程中很正常的一部分。",
"让我们一起梳理一下你的想法，或许能找到一些方向。"
]
}
self.therapeutic_techniques = {
'breathing': "试试这个简单的呼吸练习：慢慢吸气 4 秒，憋气 4 秒，然后慢慢呼气 6 秒。重复几次。",
'grounding': "试试 5-4-3-2-1 技巧：说出你能看到的 5 样东西，能听到的 4 个声音，能摸到的 3 样东西，能闻到的 2 种气味，能尝到的 1 种味道。",
'reframing': "让我们试着从另一个角度看待这个问题。有没有其他的方式来理解这种情况？",
'validation': "你的感受是完全有效的，任何人在这种情况下都可能有类似的感受。",
'self_care': "记得照顾好自己：保证充足的睡眠，规律的饮食，适当的运动，这些都很重要。"
}
def detect_emotion(self, text: str) -> List[str]:
"""检测文本中的情绪"""
detected_emotions = []
text_lower = text.lower()
for emotion, keywords in self.psychology_patterns.items():
for keyword in keywords:
if keyword in text_lower:
detected_emotions.append(emotion)
break
return detected_emotions
def generate_empathy_response(self, emotions: List[str]) -> str:
"""生成共情回应"""
if not emotions:
return ""
primary_emotion = emotions[0]
responses = self.empathy_responses.get(primary_emotion, [])
if responses:
return random.choice(responses)
return ""
def suggest_technique(self, emotions: List[str]) -> str:
"""根据情绪建议应对技巧"""
if not emotions:
return ""
technique_map = {
'anxiety': 'breathing',
'stress': 'breathing',
'anger': 'grounding',
'depression': 'self_care',
'loneliness': 'validation',
'confusion': 'reframing'
}
primary_emotion = emotions[0]
technique = technique_map.get(primary_emotion, 'validation')
return self.therapeutic_techniques.get(technique, "")
def enhance_prompt(self, user_input: str, system_prompt: str) -> str:
"""增强系统提示词"""
emotions = self.detect_emotion(user_input)
psychology_context = """作为一个心理健康助手，请遵循以下原则：
1. 保持共情和理解的态度
2. 不要给出医学诊断或治疗建议
3. 鼓励用户在需要时寻求专业帮助
4. 使用积极、支持性的语言
5. 提供实用的应对策略和技巧
6. 尊重用户的感受和经历"""
enhanced_prompt = system_prompt + "\n\n" + psychology_context
if emotions:
emotion_context = f"\n用户可能正在经历：{', '.join(emotions)}相关的情绪。请给予适当的共情和支持。"
enhanced_prompt += emotion_context
return enhanced_prompt
class ERNIEChatCLI:
def __init__(self, base_url="http://localhost:8180", model_name="baidu/ERNIE-4.5-21B-A3B-Base-Paddle"):
self.base_url = base_url
self.model_name = model_name
self.session = requests.Session()
self.conversation_history = []
self.system_prompt = "你是一个专业的心理健康助手，致力于为用户提供情感支持和心理健康指导。"
self.psychology_tuner = PsychologyFineTuner()
self.psychology_mode = True
self.crisis_keywords = ['自杀', '自杀念头', '想死', '不想活', '结束生命', '伤害自己']
self.professional_help_keywords = ['专业帮助', '心理咨询', '治疗师', '心理医生']
def check_server_status(self):
"""检查服务器状态"""
try:
response = self.session.get(f"{self.base_url}/health", timeout=5)
return response.status_code == 200
except:
return False
def get_models(self):
"""获取可用模型列表"""
try:
response = self.session.get(f"{self.base_url}/v1/models", timeout=10)
if response.status_code == 200:
return response.json()
return None
except:
return None
def check_crisis_content(self, text: str) -> bool:
"""检查是否包含危机内容"""
text_lower = text.lower()
for keyword in self.crisis_keywords:
if keyword in text_lower:
return True
return False
def handle_crisis_response(self) -> str:
"""处理危机情况的回应"""
return """ 🚨 重要提醒：
如果你正在经历自杀念头或极度痛苦，请立即寻求专业帮助：
• 全国心理危机干预热线：400-161-9995
• 北京危机干预热线：400-161-9995
• 上海心理援助热线：021-64383562
• 或拨打当地精神卫生中心电话
你的生命很宝贵，请不要独自承受这些痛苦。专业的心理健康工作者可以为你提供更好的帮助。"""
def chat_completion(self, messages, temperature=0.7, max_tokens=2048, stream=False):
"""发送聊天请求"""
payload = {
"model": self.model_name,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream
}
try:
response = self.session.post(
f"{self.base_url}/v1/chat/completions",
json=payload,
timeout=60,
stream=stream
)
if stream:
return response
else:
if response.status_code == 200:
return response.json()
else:
return {"error": f"HTTP {response.status_code}: {response.text}"}
except Exception as e:
return {"error": str(e)}
def stream_response(self, response):
"""处理流式响应"""
try:
content = ""
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data = line[6:]
if data.strip() == '[DONE]':
break
try:
json_data = json.loads(data)
if 'choices' in json_data and len(json_data['choices']) > 0:
delta = json_data['choices'][0].get('delta', {})
if 'content' in delta:
chunk = delta['content']
content += chunk
print(chunk, flush=True)
except json.JSONDecodeError:
continue
return content
except KeyboardInterrupt:
print("\n[中断]")
return content
def format_message(self, role, content):
"""格式化消息"""
timestamp = datetime.now().strftime("%H:%M:%S")
if role == "user":
return f"\033[36m[{timestamp}] 你：\033[0m{content}"
else:
return f"\033[32m[{timestamp}] 心理助手：\033[0m{content}"
def show_help(self):
"""显示帮助信息"""
help_text = """\033[1m=== 心理健康助手命令 ===\033[0m
/help - 显示帮助信息
/clear - 清除对话历史
/history - 显示对话历史
/system - 设置系统提示词
/psychology - 切换心理健康模式 (当前：{'开启' if self.psychology_mode else '关闭'})
/emotion - 分析上一条消息的情绪
/technique - 获取应对技巧建议
/crisis - 显示危机干预信息
/models - 显示可用模型
/status - 检查服务器状态
/temp <n> - 设置温度参数 (0.0-2.0)
/tokens <n> - 设置最大 token 数
/stream - 切换流式输出模式
/save - 保存对话到文件
/load - 从文件加载对话
/exit - 退出程序
\033[1m心理健康功能:\033[0m
• 自动情绪识别和共情回应
• 危机内容检测和干预
• 心理应对技巧建议
• 专业帮助引导
\033[1m快捷键:\033[0m
Ctrl+C - 中断当前响应
Ctrl+D - 退出程序"""
print(help_text)
def save_conversation(self, filename=None):
"""保存对话到文件"""
if not filename:
filename = f"psychology_chat_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
try:
with open(filename, 'w', encoding='utf-8') as f:
json.dump({
"system_prompt": self.system_prompt,
"conversation": self.conversation_history,
"psychology_mode": self.psychology_mode
}, f, ensure_ascii=False, indent=2)
print(f"对话已保存到：{filename}")
except Exception as e:
print(f"保存失败：{e}")
def load_conversation(self, filename):
"""从文件加载对话"""
try:
with open(filename, 'r', encoding='utf-8') as f:
data = json.load(f)
self.system_prompt = data.get("system_prompt", self.system_prompt)
self.conversation_history = data.get("conversation", [])
self.psychology_mode = data.get("psychology_mode", True)
print(f"对话已从 {filename} 加载")
except Exception as e:
print(f"加载失败：{e}")
def analyze_emotion(self, text: str):
"""分析文本情绪"""
emotions = self.psychology_tuner.detect_emotion(text)
if emotions:
print(f"检测到的情绪：{', '.join(emotions)}")
empathy = self.psychology_tuner.generate_empathy_response(emotions)
if empathy:
print(f"共情回应：{empathy}")
technique = self.psychology_tuner.suggest_technique(emotions)
if technique:
print(f"建议技巧：{technique}")
else:
print("未检测到特定情绪")
def run(self):
"""运行聊天界面"""
print("\033[1m=== ERNIE-4.5 心理健康助手 ===\033[0m")
print(f"模型：{self.model_name}")
print(f"服务器：{self.base_url}")
print(f"心理健康模式：{'🟢 开启' if self.psychology_mode else '🔴 关闭'}")
if not self.check_server_status():
print(f"\033[31m 错误：无法连接到服务器 {self.base_url}\033[0m")
print("请确保服务器正在运行并且端口正确")
return
print("\033[32m 服务器连接正常\033[0m")
print("我是你的心理健康助手，随时为你提供情感支持和心理健康指导。")
print("输入 /help 查看帮助信息，输入 /exit 退出")
print("-" * 50)
temperature = 0.7
max_tokens = 2048
stream_mode = True
while True:
try:
user_input = input("\033[36m> \033[0m").strip()
if not user_input:
continue
if user_input.startswith('/'):
cmd_parts = user_input.split()
cmd = cmd_parts[0].lower()
if cmd == '/help':
self.show_help()
elif cmd == '/exit':
print("记得照顾好自己，再见！💚")
break
elif cmd == '/clear':
self.conversation_history.clear()
print("对话历史已清除")
elif cmd == '/history':
if not self.conversation_history:
print("暂无对话历史")
else:
for msg in self.conversation_history:
print(self.format_message(msg['role'], msg['content']))
elif cmd == '/psychology':
self.psychology_mode = not self.psychology_mode
print(f"心理健康模式：{'🟢 开启' if self.psychology_mode else '🔴 关闭'}")
elif cmd == '/emotion':
if self.conversation_history:
last_user_msg = None
for msg in reversed(self.conversation_history):
if msg['role'] == 'user':
last_user_msg = msg['content']
break
if last_user_msg:
self.analyze_emotion(last_user_msg)
else:
print("没有找到用户消息")
else:
print("暂无对话历史")
elif cmd == '/technique':
techniques = list(self.psychology_tuner.therapeutic_techniques.values())
print(f"💡 建议技巧：{random.choice(techniques)}")
elif cmd == '/crisis':
print(self.handle_crisis_response())
elif cmd == '/system':
if len(cmd_parts) > 1:
self.system_prompt = ' '.join(cmd_parts[1:])
print(f"系统提示词已设置为：{self.system_prompt}")
else:
print(f"当前系统提示词：{self.system_prompt}")
elif cmd == '/models':
models = self.get_models()
if models:
print("可用模型:")
for model in models.get('data', []):
print(f" - {model.get('id', 'N/A')}")
else:
print("无法获取模型列表")
elif cmd == '/status':
if self.check_server_status():
print("\033[32m 服务器状态：正常\033[0m")
else:
print("\033[31m 服务器状态：异常\033[0m")
elif cmd == '/temp':
if len(cmd_parts) > 1:
try:
temperature = float(cmd_parts[1])
temperature = max(0.0, min(2.0, temperature))
print(f"温度参数设置为：{temperature}")
except ValueError:
print("无效的温度值")
else:
print(f"当前温度：{temperature}")
elif cmd == '/tokens':
if len(cmd_parts) > 1:
try:
max_tokens = int(cmd_parts[1])
max_tokens = max(1, min(32768, max_tokens))
print(f"最大 token 数设置为：{max_tokens}")
except ValueError:
print("无效的 token 数")
else:
print(f"当前最大 token 数：{max_tokens}")
elif cmd == '/stream':
stream_mode = not stream_mode
print(f"流式输出模式：{'开启' if stream_mode else '关闭'}")
elif cmd == '/save':
filename = cmd_parts[1] if len(cmd_parts) > 1 else None
self.save_conversation(filename)
elif cmd == '/load':
if len(cmd_parts) > 1:
self.load_conversation(cmd_parts[1])
else:
print("请指定文件名")
else:
print(f"未知命令：{cmd}")
continue
if self.check_crisis_content(user_input):
print(self.handle_crisis_response())
continue
current_system_prompt = self.system_prompt
if self.psychology_mode:
current_system_prompt = self.psychology_tuner.enhance_prompt(
user_input, self.system_prompt
)
messages = [{"role": "system", "content": current_system_prompt}]
messages.extend(self.conversation_history)
messages.append({"role": "user", "content": user_input})
print(self.format_message("user", user_input))
if self.psychology_mode:
emotions = self.psychology_tuner.detect_emotion(user_input)
if emotions:
empathy = self.psychology_tuner.generate_empathy_response(emotions)
if empathy:
print(f"\033[33m💝 {empathy}\033[0m")
print(f"\033[32m[{datetime.now().strftime('%H:%M:%S')}] 心理助手：\033[0m", flush=True)
if stream_mode:
response = self.chat_completion(messages, temperature, max_tokens, stream=True)
if hasattr(response, 'iter_lines'):
ai_response = self.stream_response(response)
print()
else:
ai_response = "连接错误"
print(ai_response)
else:
result = self.chat_completion(messages, temperature, max_tokens, stream=False)
if 'error' in result:
ai_response = f"错误：{result['error']}"
else:
ai_response = result['choices'][0]['message']['content']
print(ai_response)
if self.psychology_mode:
emotions = self.psychology_tuner.detect_emotion(user_input)
if emotions:
technique = self.psychology_tuner.suggest_technique(emotions)
if technique:
print(f"\033[35m💡 应对建议：{technique}\033[0m")
self.conversation_history.append({"role": "user", "content": user_input})
self.conversation_history.append({"role": "assistant", "content": ai_response})
if len(self.conversation_history) > 20:
self.conversation_history = self.conversation_history[-20:]
except KeyboardInterrupt:
print("\n使用 /exit 退出程序")
continue
except EOFError:
print("\n记得照顾好自己，再见！💚")
break
except Exception as e:
print(f"\n错误：{e}")
continue
def main():
parser = argparse.ArgumentParser(description='ERNIE-4.5 心理健康助手命令行界面')
parser.add_argument('--url', default='http://localhost:8180', help='服务器 URL')
parser.add_argument('--model', default='baidu/ERNIE-4.5-21B-A3B-Base-Paddle', help='模型名称')
parser.add_argument('--no-psychology', action='store_true', help='禁用心理健康模式')
args = parser.parse_args()
def signal_handler(sig, frame):
print('\n记得照顾好自己，正在退出...')
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
cli = ERNIEChatCLI(args.url, args.model)
if args.no_psychology:
cli.psychology_mode = False
cli.run()
if __name__ == "__main__":
main()
