87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
import re
|
||
import time
|
||
import logging
|
||
from uuid import UUID
|
||
|
||
from django.conf import settings
|
||
from openai import OpenAI
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CLASSIFY_SYSTEM_PROMPT = """你是一个教育评估专家。根据布鲁姆认知分类学，分析以下学生在前端学习中发送给AI助手的一条提示词，判断该提示词所体现的认知层级。
|
||
|
||
层级定义：
|
||
- L1 记忆：能背诵HTML标签语法（例："帮我写一个按钮"）
|
||
- L2 理解：能解释flex布局原理（例："为什么这里不居中？"）
|
||
- L3 应用：能独立搭建页面结构（例："用flex做导航栏，间距16px"）
|
||
- L4 分析：能定位跨浏览器兼容性bug（例："Safari中margin失效，原因？"）
|
||
- L5 评价：能对比并选择方案（例："对比Grid与Flex方案优劣"）
|
||
- L6 创造：能设计并实现原创交互作品（例："设计夜间/日间切换效果"）
|
||
|
||
只返回一个数字（1-6），不要解释。"""
|
||
|
||
|
||
def _call_llm(content: str) -> int | None:
|
||
"""Call LLM to classify a single message content. Returns level 1-6 or None."""
|
||
try:
|
||
client = OpenAI(
|
||
api_key=settings.LLM_API_KEY,
|
||
base_url=settings.LLM_BASE_URL,
|
||
timeout=30.0,
|
||
)
|
||
response = client.chat.completions.create(
|
||
model=settings.LLM_MODEL,
|
||
messages=[
|
||
{"role": "system", "content": CLASSIFY_SYSTEM_PROMPT},
|
||
{"role": "user", "content": content},
|
||
],
|
||
max_tokens=10,
|
||
stream=False,
|
||
)
|
||
text = response.choices[0].message.content or ""
|
||
match = re.search(r"[1-6]", text)
|
||
if not match:
|
||
logger.warning("classify: unexpected LLM response '%s'", text)
|
||
return None
|
||
return int(match.group())
|
||
except Exception as e:
|
||
logger.error("classify LLM call failed: %s", e)
|
||
return None
|
||
|
||
|
||
def classify_message(message_id: int) -> int | None:
|
||
"""Classify a single user Message by ID. Returns level or None."""
|
||
from prompt.models import Message
|
||
|
||
try:
|
||
msg = Message.objects.get(id=message_id, role="user")
|
||
except Message.DoesNotExist:
|
||
return None
|
||
|
||
level = _call_llm(msg.content)
|
||
if level is not None:
|
||
Message.objects.filter(id=message_id).update(prompt_level=level)
|
||
return level
|
||
|
||
|
||
def classify_conversation_messages(conversation_id: UUID, force: bool = False) -> None:
|
||
"""Classify all user messages in a conversation."""
|
||
from prompt.models import Message
|
||
|
||
qs = Message.objects.filter(conversation_id=conversation_id, role="user")
|
||
if not force:
|
||
qs = qs.filter(prompt_level__isnull=True)
|
||
|
||
for msg in qs.order_by("created"):
|
||
level = _call_llm(msg.content)
|
||
if level is not None:
|
||
Message.objects.filter(id=msg.id).update(prompt_level=level)
|
||
time.sleep(0.3)
|
||
|
||
|
||
def classify_messages_batch(message_ids: list) -> None:
|
||
"""Classify a list of messages by ID."""
|
||
for mid in message_ids:
|
||
classify_message(mid)
|
||
time.sleep(0.5)
|