import re from django.conf import settings from openai import AsyncOpenAI SYSTEM_PROMPT = """你是一个网页生成助手。根据用户的需求描述,生成网页代码。 规则: 1. 使用一个 ```html 代码块返回所有代码 2. HTML 代码只需要 body 内的内容,不需要完整的 HTML 文档结构 3. CSS 样式写在 ```""" GUIDANCE_SYSTEM_PROMPT = """你是一个提示词写作教练,帮助学生写出清晰、具体的网页需求描述。 判断标准——满足以下条件视为"够好": - 有明确主题(例如:登录页、计时器、商品卡片列表) - 至少包含以下一项具体描述:颜色/布局/风格、交互行为(按钮点击、动画等)、页面内容(文字、数量、图标) 判断为"不够好"的典型情况: - 目标太泛,如"做一个好看的页面" - 只有主题,完全没有任何视觉、交互或内容描述 规则: 1. 如果提示词不够好,用 1-2 个启发性问题引导学生补充细节,不要直接给出答案 2. 如果提示词已经够好,以 [READY] 开头回复,简短夸奖学生并说明可以生成了 3. 用中文回复,语气鼓励,简洁明了 4. 使用 Markdown 语法高亮关键词,优先突出 **主题**、**视觉**、**交互**、**内容**、**可以生成** 等重点 5. 如果回复以 [READY] 开头,[READY] 不要加粗,必须保持原始文本 6. 不要生成任何代码""" DEFAULT_MODEL = "deepseek-v4-flash" DEEPSEEK_THINKING_MODEL = "deepseek-v4-flash-thinking" MODEL_ALIASES = { DEEPSEEK_THINKING_MODEL: DEFAULT_MODEL, } NON_THINKING_MODELS = {"deepseek-v4-flash"} NON_THINKING_EXTRA_BODY = {"thinking": {"type": "disabled"}} # Models served by the ARK (Volcengine) endpoint ARK_MODELS = {"doubao-seed-2-0-lite-260215"} def _get_client(model: str) -> tuple[AsyncOpenAI, str]: """Return (client, model_id) for the given model name.""" requested_model = model or DEFAULT_MODEL resolved_model = MODEL_ALIASES.get(requested_model, requested_model) if resolved_model in ARK_MODELS: return ( AsyncOpenAI( api_key=settings.ARK_API_KEY, base_url=settings.ARK_BASE_URL, timeout=120.0, ), resolved_model, ) return ( AsyncOpenAI( api_key=settings.LLM_API_KEY, base_url=settings.LLM_BASE_URL, timeout=120.0, ), resolved_model, ) def _should_disable_thinking(requested_model: str, resolved_model: str) -> bool: return ( resolved_model in NON_THINKING_MODELS and requested_model not in MODEL_ALIASES ) def _chat_completion_kwargs( requested_model: str, resolved_model: str, messages: list[dict], stream: bool, ) -> dict: kwargs = { "model": resolved_model, "messages": messages, "stream": stream, } if _should_disable_thinking(requested_model, resolved_model): kwargs["extra_body"] = NON_THINKING_EXTRA_BODY return kwargs async def _stream_completion(messages: list[dict], model: str = ""): client, resolved_model = _get_client(model) requested_model = model or DEFAULT_MODEL async with client as c: stream = await c.chat.completions.create( **_chat_completion_kwargs(requested_model, resolved_model, messages, stream=True), ) async for chunk in stream: delta = chunk.choices[0].delta if delta.content: yield delta.content async def stream_chat(history: list[dict], model: str = ""): """Stream chat completion from the LLM. Yields content chunks.""" messages = [{"role": "system", "content": SYSTEM_PROMPT}, *history] async for chunk in _stream_completion(messages, model): yield chunk def extract_code(text: str) -> dict: """Extract code from AI response. Supports single HTML block (new) or separate html/css/js blocks (legacy).""" result = {"html": None, "css": None, "js": None} pattern = r"```(html|css|js|javascript|typescript|ts|jsx|tsx)\s*\n(.*?)```" matches = re.findall(pattern, text, re.DOTALL | re.IGNORECASE) for lang, code in matches: lang = lang.lower() if lang in ("javascript", "typescript", "ts", "jsx", "tsx"): lang = "js" if lang in result and result[lang] is None: result[lang] = code.strip() # Single HTML block: extract ", html, re.DOTALL | re.IGNORECASE) if style_match: result["css"] = style_match.group(1).strip() html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) script_match = re.search(r"]*>(.*?)", html, re.DOTALL | re.IGNORECASE) if script_match: result["js"] = script_match.group(1).strip() html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) result["html"] = html.strip() return result def parse_guidance_response(full_response: str) -> tuple[str, bool]: if full_response.startswith("[READY]"): return full_response[len("[READY]"):].lstrip("\n"), True return full_response, False async def stream_guidance(history: list[dict]): """Stream guidance coaching response. Yields content chunks.""" messages = [{"role": "system", "content": GUIDANCE_SYSTEM_PROMPT}, *history] async for chunk in _stream_completion(messages): yield chunk