461 lines
20 KiB
Python
461 lines
20 KiB
Python
from fastapi import APIRouter, HTTPException, Depends
|
||
from pydantic import BaseModel
|
||
from typing import List, Optional, Dict, Any
|
||
import httpx
|
||
import os
|
||
import logging
|
||
|
||
from app.services.glm_service import get_glm_service, GLMService
|
||
from app.models.base import BaseOut
|
||
|
||
# 创建API路由，使用/glm前缀
|
||
router = APIRouter(prefix="/glm", tags=["GLM聊天"])
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# 数据模型
|
||
class Message(BaseModel):
|
||
role: str
|
||
content: str
|
||
|
||
|
||
class ChatRequest(BaseModel):
|
||
message: str
|
||
conversation_id: Optional[str] = None
|
||
enhanced_prompt: Optional[str] = None # 前端生成的完整提示词
|
||
customer_tags: Optional[List[Dict[str, Any]]] = None # 客户标签
|
||
message_history: Optional[List[Dict[str, Any]]] = None # 企业微信历史消息
|
||
workflow_context: Optional[Dict[str, Any]] = None # 工作流上下文
|
||
|
||
|
||
class ChatResponse(BaseModel):
|
||
response: str
|
||
conversation_id: str
|
||
|
||
|
||
class RoleConfig(BaseModel):
|
||
name: str
|
||
description: str
|
||
skills: str # 多行文本字符串
|
||
restrictions: str # 多行文本字符串
|
||
|
||
|
||
# 默认角色配置（使用字符串格式）
|
||
DEFAULT_ROLE_CONFIG = RoleConfig(
|
||
name="兜兜",
|
||
description="你是兜兜，1989年出生，有一个5岁的女儿（2018年出生）。你是一位对用户负责、选品严格的宝妈客服。当用户给你发来疑问时，你总能第一时间回复，并以亲切温暖的风格与用户交流。",
|
||
skills="亲切温暖的沟通\n轻松友好的语气\n简洁明了的描述\n亲昵且口语化的用词\n乱码话术重复问询",
|
||
restrictions="回复必须基于提供的知识库中的内容\n每条回复应保持在80字以内，使用口语化语言\n回复中可以适当加入表情符号，但不要过度使用\n只讨论与宝妈相关的主题，如育儿、选品、团购等\n始终以用户的需求和满意度为优先",
|
||
)
|
||
|
||
# 全局变量存储对话历史（暂时保持内存存储，根据用户要求不持久化）
|
||
conversations = {}
|
||
|
||
|
||
def build_system_prompt(role_config: dict) -> str:
|
||
"""构建系统提示词
|
||
|
||
Args:
|
||
role_config: 角色配置字典，包含name, description, skills, restrictions字段
|
||
|
||
Returns:
|
||
系统提示词字符串
|
||
"""
|
||
# 解析技能字符串为列表
|
||
skills_list = [
|
||
skill.strip() for skill in role_config["skills"].split("\n") if skill.strip()
|
||
]
|
||
restrictions_list = [
|
||
restriction.strip()
|
||
for restriction in role_config["restrictions"].split("\n")
|
||
if restriction.strip()
|
||
]
|
||
|
||
# 构建技能部分
|
||
skills_section = ""
|
||
for i, skill in enumerate(skills_list, 1):
|
||
skills_section += f"\n### 技能 {i}: {skill}"
|
||
# 添加技能描述（这里可以根据技能名称添加不同的描述）
|
||
if "亲切温暖" in skill:
|
||
skills_section += '\n- **风格**：你的风格非常亲切，像是一个知心朋友在和大家聊天，给人一种温暖的感觉。你经常使用"宝"、"美妈"等亲昵的称呼，拉近了与用户的距离。'
|
||
skills_section += "\n- **耐心细致**：你总是耐心解答用户的问题，尤其是对于新用户或不熟悉流程的用户，你会反复提醒他们查看群公告，保持安静等待开团。"
|
||
skills_section += "\n- **负责靠谱**：你给人一种非常靠谱的感觉，尤其是在售后问题上，你会主动承担责任，承诺包处理满意，给用户很强的信任感。"
|
||
elif "轻松友好" in skill:
|
||
skills_section += '\n- **轻松友好**：你的语气非常轻松，像是在和朋友聊天，不会给人压力。比如"不着急哦"、"爱你！"等表达方式，让人感到轻松愉快。'
|
||
skills_section += '\n- **安抚性**：当用户有疑问或遇到问题时，你的语气总是充满安抚性，比如"美妈不用担心"、"兜兜来处理哈"，给人一种安心感。'
|
||
skills_section += '\n- **鼓励性**：你经常用"不着急"、"等开团的时候会有统一说明"等话语，鼓励用户耐心等待，保持积极的心态。'
|
||
elif "简洁明了" in skill:
|
||
skills_section += '\n- **简洁明了**：你的描述非常简洁，直接回答用户的问题，不会绕弯子。比如"贝亲是咱们深度合作品牌，经常上团的"，直接给出了明确的答案。'
|
||
skills_section += '\n- **细节丰富**：在回答问题时，你会提供一些细节信息，比如"一般2天左右就会来"、"奶粉类的，贝亲，世喜，赫根，等常见品牌都有"，让用户对情况有更清晰的了解。'
|
||
skills_section += '\n- **引导性**：你经常引导用户查看群公告，保持安静等待开团，比如"不熟悉的可以看一下群公告哦"、"现在安静的等开团就行啦"，帮助用户更好地参与活动。'
|
||
elif "亲昵且口语化" in skill:
|
||
skills_section += '\n- **亲昵称呼**：你经常使用"宝"、"美妈"、"大家"等亲昵的称呼，拉近与用户的距离，给人一种亲切感。'
|
||
skills_section += '\n- **口语化**：你的用词非常口语化，像是日常聊天，比如"不着急哦"、"爱你！"、"兜兜来处理哈"，给人一种轻松自然的感觉，称呼自己为"兜兜"或者"我"。'
|
||
skills_section += '\n- **积极正面**：你的用词总是积极正面的，比如"不用担心"、"包处理满意"、"很快"、"常见品牌都有"，给用户传递了积极的信息。'
|
||
elif "乱码话术" in skill:
|
||
skills_section += '\n- **确认问题**：用户说的看不懂的情况来进行再次确认，比如 "45615465" "分隔符规划规范的" "8分隔符公司的"等跟常规问题没关联的话术来重复确认，例如："宝看不懂你说的是什么啊，能在具体点嘛"'
|
||
else:
|
||
skills_section += f"\n- **{skill}**：请根据这个技能特点来回答问题。"
|
||
|
||
# 构建限制部分
|
||
restrictions_section = "\n".join(
|
||
[f"- {restriction}" for restriction in restrictions_list]
|
||
)
|
||
|
||
prompt = f"""# 角色
|
||
{role_config["description"]}
|
||
|
||
## 技能{skills_section}
|
||
|
||
## 限制
|
||
{restrictions_section}
|
||
|
||
请按照以上角色设定和限制来回答用户的问题。"""
|
||
return prompt
|
||
|
||
|
||
def get_deepseek_api_key() -> str:
|
||
"""获取DeepSeek API密钥"""
|
||
api_key = os.getenv("DEEPSEEK_API_KEY", "your-deepseek-api-key")
|
||
return api_key
|
||
|
||
|
||
def is_api_key_configured() -> bool:
|
||
"""检查API密钥是否已配置"""
|
||
api_key = get_deepseek_api_key()
|
||
return api_key != "your-deepseek-api-key"
|
||
|
||
|
||
async def call_deepseek_api(messages: List[dict]) -> str:
|
||
"""调用DeepSeek API"""
|
||
# 直接使用环境变量获取API密钥
|
||
api_key = get_deepseek_api_key()
|
||
|
||
if not is_api_key_configured():
|
||
# 模拟响应用于演示
|
||
return "宝，这个问题兜兜需要查一下资料哦，稍等哈~ 😊"
|
||
|
||
url = "https://api.deepseek.com/v1/chat/completions"
|
||
headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"}
|
||
|
||
data = {
|
||
"model": "deepseek-chat",
|
||
"messages": messages,
|
||
"stream": False,
|
||
"max_tokens": 500,
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(url, headers=headers, json=data, timeout=30.0)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
return result["choices"][0]["message"]["content"]
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=f"API调用失败: {str(e)}")
|
||
|
||
|
||
@router.get("/")
|
||
async def root():
|
||
return {"message": "DeepSeek AI Chat Assistant API"}
|
||
|
||
|
||
@router.get("/role", response_model=BaseOut)
|
||
async def get_role_config(service: GLMService = Depends(get_glm_service)):
|
||
"""获取当前角色配置"""
|
||
try:
|
||
role = await service.get_current_role()
|
||
return BaseOut(code=200, message="ok", result=role.to_role_config_dict())
|
||
except Exception as e:
|
||
logger.error(f"获取角色配置失败: {e}")
|
||
return BaseOut(code=500, message=f"获取角色配置失败: {str(e)}", result=None)
|
||
|
||
|
||
@router.put("/role", response_model=BaseOut)
|
||
async def update_role_config(
|
||
role_config: RoleConfig, service: GLMService = Depends(get_glm_service)
|
||
):
|
||
"""更新角色配置"""
|
||
try:
|
||
# 将Pydantic模型转换为字典
|
||
role_data = role_config.dict()
|
||
updated_role = await service.update_role(role_data)
|
||
|
||
return BaseOut(
|
||
code=200,
|
||
message="角色配置已更新",
|
||
result={
|
||
"message": "角色配置已更新",
|
||
"role_config": updated_role.to_role_config_dict(),
|
||
},
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"更新角色配置失败: {e}")
|
||
return BaseOut(code=500, message=f"更新角色配置失败: {str(e)}", result=None)
|
||
|
||
|
||
@router.post("/chat", response_model=ChatResponse)
|
||
async def chat(request: ChatRequest, service: GLMService = Depends(get_glm_service)):
|
||
"""处理聊天请求"""
|
||
# 步骤1: 记录请求开始
|
||
logger.info(
|
||
f"收到聊天请求: conversation_id={request.conversation_id or 'None'}, "
|
||
f"message_length={len(request.message)}, "
|
||
f"message_preview='{request.message[:50]}{'...' if len(request.message) > 50 else ''}'"
|
||
)
|
||
|
||
# 记录增强参数信息
|
||
has_enhanced_prompt = request.enhanced_prompt is not None
|
||
has_customer_tags = (
|
||
request.customer_tags is not None and len(request.customer_tags) > 0
|
||
)
|
||
has_message_history = (
|
||
request.message_history is not None and len(request.message_history) > 0
|
||
)
|
||
has_workflow_context = request.workflow_context is not None
|
||
|
||
logger.info(
|
||
f"增强参数状态: enhanced_prompt={has_enhanced_prompt}, "
|
||
f"customer_tags={has_customer_tags} ({len(request.customer_tags) if has_customer_tags else 0}个标签), "
|
||
f"message_history={has_message_history} ({len(request.message_history) if has_message_history else 0}条历史), "
|
||
f"workflow_context={has_workflow_context}"
|
||
)
|
||
|
||
# 记录客户标签摘要
|
||
if has_customer_tags:
|
||
tag_categories = {}
|
||
for tag in request.customer_tags:
|
||
category = tag.get("TagCategory", "未分类")
|
||
name = tag.get("TagName", "未命名")
|
||
if category not in tag_categories:
|
||
tag_categories[category] = []
|
||
tag_categories[category].append(name)
|
||
|
||
tag_summary = []
|
||
for category, names in tag_categories.items():
|
||
tag_summary.append(
|
||
f"{category}: {', '.join(names[:3])}{'...' if len(names) > 3 else ''}"
|
||
)
|
||
|
||
# 检查是否使用mock数据
|
||
is_mock_data = False
|
||
for tag in request.customer_tags:
|
||
if tag.get("TagName", "").startswith("mock_") or tag.get(
|
||
"TagCategory", ""
|
||
).startswith("mock_"):
|
||
is_mock_data = True
|
||
break
|
||
|
||
mock_note = " [mock数据]" if is_mock_data else ""
|
||
logger.info(f"客户标签摘要{mock_note}: {', '.join(tag_summary)}")
|
||
|
||
# 记录历史消息摘要
|
||
if has_message_history:
|
||
# 按时间排序，获取最近的消息
|
||
sorted_history = sorted(
|
||
request.message_history, key=lambda x: x.get("msgtime", ""), reverse=True
|
||
)[
|
||
:5
|
||
] # 只显示最近5条
|
||
|
||
history_summary = []
|
||
for i, msg in enumerate(sorted_history):
|
||
from_user = msg.get("from_user", "未知用户")
|
||
user_id = (
|
||
msg.get("user_id")
|
||
or msg.get("userId")
|
||
or msg.get("from_user_id")
|
||
or "无ID"
|
||
)
|
||
content = msg.get("text", {}).get("content", "") or msg.get("content", "")
|
||
content_preview = content[:30] + ("..." if len(content) > 30 else "")
|
||
history_summary.append(f"{from_user}(ID:{user_id}): '{content_preview}'")
|
||
|
||
# 检查是否使用mock数据
|
||
is_mock_data = False
|
||
for msg in sorted_history:
|
||
if msg.get("from_user", "").startswith("mock_") or msg.get(
|
||
"user_id", ""
|
||
).startswith("mock_"):
|
||
is_mock_data = True
|
||
break
|
||
|
||
mock_note = " [mock数据]" if is_mock_data else ""
|
||
logger.info(f"历史消息摘要{mock_note}: {', '.join(history_summary)}")
|
||
|
||
# 步骤2: 创建或获取对话历史
|
||
if request.conversation_id and request.conversation_id in conversations:
|
||
conversation_history = conversations[request.conversation_id]
|
||
logger.info(
|
||
f"使用现有对话: {request.conversation_id}, 历史消息数: {len(conversation_history)}"
|
||
)
|
||
# 记录历史消息摘要
|
||
if conversation_history:
|
||
last_messages = conversation_history[-3:] # 显示最近3条消息
|
||
history_summary = []
|
||
for i, msg in enumerate(last_messages, 1):
|
||
role = msg["role"]
|
||
content_preview = msg["content"][:30] + (
|
||
"..." if len(msg["content"]) > 30 else ""
|
||
)
|
||
history_summary.append(f"{role}: '{content_preview}'")
|
||
logger.debug(f"最近历史消息摘要: {', '.join(history_summary)}")
|
||
else:
|
||
conversation_history = []
|
||
request.conversation_id = f"conv_{len(conversations) + 1}"
|
||
conversations[request.conversation_id] = conversation_history
|
||
logger.info(f"新建对话: {request.conversation_id}")
|
||
|
||
# 步骤3: 构建系统提示词
|
||
try:
|
||
if has_enhanced_prompt:
|
||
# 使用前端生成的增强提示词
|
||
system_prompt = request.enhanced_prompt
|
||
logger.info(f"使用前端增强提示词, 长度: {len(system_prompt)} 字符")
|
||
logger.debug(
|
||
f"增强提示词预览: '{system_prompt[:200]}{'...' if len(system_prompt) > 200 else ''}'"
|
||
)
|
||
else:
|
||
# 获取当前角色配置并构建系统提示词
|
||
role = await service.get_current_role()
|
||
role_config = role.to_role_config_dict()
|
||
logger.info(
|
||
f"获取角色配置: name='{role.name}', "
|
||
f"技能数: {len(role_config['skills'].split('\\n')) if role_config['skills'] else 0}, "
|
||
f"限制数: {len(role_config['restrictions'].split('\\n')) if role_config['restrictions'] else 0}"
|
||
)
|
||
|
||
system_prompt = build_system_prompt(role_config)
|
||
logger.debug(f"系统提示词构建完成, 长度: {len(system_prompt)} 字符")
|
||
except Exception as e:
|
||
logger.error(
|
||
f"构建系统提示词失败: {e}, conversation_id={request.conversation_id}"
|
||
)
|
||
raise HTTPException(status_code=500, detail=f"构建系统提示词失败: {str(e)}")
|
||
|
||
# 步骤4: 构建消息列表
|
||
messages = [{"role": "system", "content": system_prompt}]
|
||
|
||
# 初始化企业微信历史消息变量
|
||
selected_wework_messages = []
|
||
|
||
# 添加企业微信历史消息（如果提供）
|
||
if has_message_history:
|
||
logger.info(f"添加企业微信历史消息到上下文: {len(request.message_history)} 条")
|
||
|
||
# 将企业微信消息转换为大模型消息格式
|
||
wework_messages = []
|
||
for msg in request.message_history:
|
||
from_user = msg.get("from_user", "未知用户")
|
||
content = msg.get("text", {}).get("content", "") or msg.get("content", "")
|
||
msgtime = msg.get("msgtime", "")
|
||
|
||
if content:
|
||
# 确定角色：如果是客户发送的，则为user；否则为assistant
|
||
# 这里需要根据实际业务逻辑判断，暂时假设外部用户是user
|
||
role = (
|
||
"user"
|
||
if "customer" in from_user.lower() or "user" in from_user.lower()
|
||
else "assistant"
|
||
)
|
||
|
||
# 格式化消息内容
|
||
formatted_content = f"[{msgtime}] {from_user}: {content}"
|
||
wework_messages.append({"role": role, "content": formatted_content})
|
||
|
||
# 添加到消息列表（限制数量，避免过长）
|
||
max_wework_messages = 10
|
||
selected_wework_messages = wework_messages[
|
||
-max_wework_messages:
|
||
] # 取最近的消息
|
||
messages.extend(selected_wework_messages)
|
||
logger.info(
|
||
f"已添加 {len(selected_wework_messages)} 条企业微信历史消息到上下文"
|
||
)
|
||
|
||
# 添加与大模型的对话历史
|
||
history_messages = conversation_history[-10:] # 保留最近10条历史
|
||
for msg in history_messages:
|
||
messages.append({"role": msg["role"], "content": msg["content"]})
|
||
|
||
# 添加当前用户消息
|
||
messages.append({"role": "user", "content": request.message})
|
||
conversation_history.append({"role": "user", "content": request.message})
|
||
|
||
logger.info(
|
||
f"构建消息列表完成: 系统消息1条, 企业微信历史{len(selected_wework_messages)}条, "
|
||
f"对话历史{len(history_messages)}条, 用户消息1条, 总消息数: {len(messages)}"
|
||
)
|
||
|
||
# 记录消息列表摘要
|
||
if logger.isEnabledFor(logging.DEBUG):
|
||
for i, msg in enumerate(messages):
|
||
role = msg["role"]
|
||
content_preview = msg["content"][:50] + (
|
||
"..." if len(msg["content"]) > 50 else ""
|
||
)
|
||
logger.debug(f"消息{i+1}: {role} - '{content_preview}'")
|
||
|
||
# 步骤5: 调用DeepSeek API
|
||
try:
|
||
logger.info(
|
||
f"调用DeepSeek API, conversation_id={request.conversation_id}, 总消息数: {len(messages)}"
|
||
)
|
||
response_text = await call_deepseek_api(messages)
|
||
|
||
# 步骤6: 记录API响应
|
||
logger.info(
|
||
f"DeepSeek API调用成功, conversation_id={request.conversation_id}, "
|
||
f"响应长度: {len(response_text)} 字符, "
|
||
f"响应预览: '{response_text[:100]}{'...' if len(response_text) > 100 else ''}'"
|
||
)
|
||
|
||
# 添加助手回复到历史
|
||
conversation_history.append({"role": "assistant", "content": response_text})
|
||
|
||
# 步骤7: 返回响应
|
||
logger.info(
|
||
f"返回聊天响应: conversation_id={request.conversation_id}, 响应长度: {len(response_text)} 字符"
|
||
)
|
||
return ChatResponse(
|
||
response=response_text, conversation_id=request.conversation_id
|
||
)
|
||
except Exception as e:
|
||
logger.error(
|
||
f"DeepSeek API调用失败: {e}, conversation_id={request.conversation_id}, message_preview='{request.message[:50]}...'"
|
||
)
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
|
||
@router.get("/conversations/{conversation_id}")
|
||
async def get_conversation(conversation_id: str):
|
||
"""获取特定对话的历史"""
|
||
if conversation_id not in conversations:
|
||
raise HTTPException(status_code=404, detail="对话不存在")
|
||
return {
|
||
"conversation_id": conversation_id,
|
||
"messages": conversations[conversation_id],
|
||
}
|
||
|
||
|
||
@router.post("/role/reset", response_model=BaseOut)
|
||
async def reset_role_config(service: GLMService = Depends(get_glm_service)):
|
||
"""重置角色配置为默认值"""
|
||
try:
|
||
# 使用 DEFAULT_ROLE_CONFIG 更新角色
|
||
role_data = DEFAULT_ROLE_CONFIG.dict()
|
||
updated_role = await service.update_role(role_data)
|
||
|
||
return BaseOut(
|
||
code=200,
|
||
message="角色配置已重置为默认值",
|
||
result={
|
||
"message": "角色配置已重置为默认值",
|
||
"role_config": updated_role.to_role_config_dict(),
|
||
},
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"重置角色配置失败: {e}")
|
||
return BaseOut(code=500, message=f"重置角色配置失败: {str(e)}", result=None)
|