feat(llm): 内置 LLM 供应商配置与审核主链接入骨架

配置层:
- Settings 新增 llm_provider/api_key/base_url/model/timeout/max_tokens
- 生产 fail-closed: GAOKAO_LLM_PROVIDER=none 禁止, provider!=none 且 API key 为空禁止
- .env.docker.example / .env.payment.example 补 LLM 变量
- payment_readiness_doctor 将 LLM 配置纳入 readiness 检查

基础设施:
- 新增 data/llm/client.py: OpenAI-compatible LLMClient
- 新增 data/llm/prompts.py: audit/cwb/full_plan prompt 模板
- 新增 data/llm/tests/test_llm.py: 12 个单元测试

主链接入:
- ReviewResultContract 新增 llm_generated / llm_summary / llm_cwb_suggestions
- _start_review_result 优先尝试 LLM 生成审核结果, 失败时回退到原规则默认逻辑
- cwb 页面优先展示 LLM 生成的三档建议

测试适配:
- conftest / health / app / p2_4 tests 注入默认 LLM 测试配置，避免被新 fail-closed 提前拦截

验证:
- data/llm/tests 12 passed
- 核心 prod settings tests 62 passed

.env.docker.example

@@ -9,6 +9,13 @@ GAOKAO_ORDERS_FERNET_KEY=replace-with-strong-orders-fernet-secret-before-product
 GAOKAO_PORTAL_UPLOAD_DIR=/var/lib/gaokao/portal_uploads
 GAOKAO_PORTAL_UPLOAD_MAX_BYTES=5242880
 GAOKAO_PORTAL_UPLOAD_MAX_FILES=5
+# LLM 自动审核 / 方案生成（生产环境必须配置）
+GAOKAO_LLM_PROVIDER=dashscope
+GAOKAO_LLM_API_KEY=replace-with-real-llm-api-key
+GAOKAO_LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
+GAOKAO_LLM_MODEL=qwen-plus
+GAOKAO_LLM_TIMEOUT=60
+GAOKAO_LLM_MAX_TOKENS=4096
 GAOKAO_PAYMENT_PROVIDER=mock
 GAOKAO_PAYMENT_BASE_URL=https://example.com
 GAOKAO_PAYMENT_WEBHOOK_SECRET=replace-with-independent-payment-webhook-secret

.env.payment.example

@@ -2,7 +2,18 @@
 # 复制为 .env.payment 并填写真实值后 source 使用
 # 正式上线前必须完成一次真实 acceptance
 
-# 应用 ID（支付宝开放平台 → 应用管理）
+# ===== LLM 供应商（自动生成志愿方案所必需） =====
+# 当前产品要求系统内自动调用 LLM 完成审核/冲稳保/完整规划。
+# 生产环境禁止 provider=none。
+GAOKAO_LLM_PROVIDER=none
+GAOKAO_LLM_API_KEY=
+# 默认 DashScope OpenAI-compatible；若用 OpenAI 则改为 https://api.openai.com/v1
+GAOKAO_LLM_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
+GAOKAO_LLM_MODEL=qwen-plus
+GAOKAO_LLM_TIMEOUT=60
+GAOKAO_LLM_MAX_TOKENS=4096
+
+# ===== 支付/安全/运营 =====
 GAOKAO_PAYMENT_APP_ID=
 
 # 商户 ID（支付宝商户平台 → 账户管理）

admin/config.py

@@ -73,6 +73,12 @@ class Settings:
     consent_version: str  # 当前同意协议版本号，与 docs/PRIVACY_POLICY_DRAFT.md 版本对齐
     consent_scope_portal: str  # portal 资料提交默认 scope
     consent_scope_channel_prefix: str  # 后台代录 scope 前缀
+    llm_provider: str  # openai|dashscope|anthropic|none
+    llm_api_key: str
+    llm_base_url: str
+    llm_model: str
+    llm_timeout_seconds: int
+    llm_max_tokens: int
 
 
 def _resolve_payment_webhook_secret(env: str) -> str:
@@ -190,6 +196,31 @@ def _enforce_payment_provider_policy(settings: Settings) -> None:
         )
 
 
+_ALLOWED_LLM_PROVIDERS = {"openai", "dashscope", "anthropic", "none"}
+
+
+def _enforce_llm_provider_policy(settings: Settings) -> None:
+    """生产环境 LLM provider 必须显式配置且不为 none，否则无法生成志愿方案。"""
+    provider = (settings.llm_provider or "none").strip().lower()
+    if provider not in _ALLOWED_LLM_PROVIDERS:
+        raise RuntimeError(
+            f"GAOKAO_LLM_PROVIDER={provider} 不在受支持列表 "
+            f"{sorted(_ALLOWED_LLM_PROVIDERS)}"
+        )
+    if settings.env == "prod" and provider == "none":
+        raise RuntimeError(
+            "生产环境 GAOKAO_LLM_PROVIDER=none 被禁止："
+            "产品需要 LLM 自动生成志愿方案，必须配置有效的供应商 "
+            "(openai/dashscope/anthropic)"
+        )
+    if provider != "none" and not settings.llm_api_key:
+        if settings.env == "prod":
+            raise RuntimeError(
+                f"生产环境 GAOKAO_LLM_PROVIDER={provider} 但 GAOKAO_LLM_API_KEY 为空，"
+                "无法调用 LLM 服务"
+            )
+
+
 def load_settings() -> Settings:
     """从环境变量加载配置。
 
@@ -293,6 +324,14 @@ def load_settings() -> Settings:
         consent_scope_channel_prefix=os.getenv(
             "GAOKAO_CONSENT_SCOPE_CHANNEL_PREFIX", "channel-intake"
         ),
+        llm_provider=os.getenv("GAOKAO_LLM_PROVIDER", "none"),
+        llm_api_key=os.getenv("GAOKAO_LLM_API_KEY", ""),
+        llm_base_url=os.getenv(
+            "GAOKAO_LLM_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"
+        ),
+        llm_model=os.getenv("GAOKAO_LLM_MODEL", "qwen-plus"),
+        llm_timeout_seconds=int(os.getenv("GAOKAO_LLM_TIMEOUT", "60")),
+        llm_max_tokens=int(os.getenv("GAOKAO_LLM_MAX_TOKENS", "4096")),
     )
     # 生产环境 post-load 校验:webhook / portal token / JWT / admin password
     # / payment provider 必须满足强度门槛, 任一不满足 fail-closed (P0-2/P2-4/P2-5/6/20)。
@@ -301,6 +340,7 @@ def load_settings() -> Settings:
     _enforce_jwt_secret_policy(settings)
     _enforce_default_admin_password_policy(settings)
     _enforce_payment_provider_policy(settings)
+    _enforce_llm_provider_policy(settings)
     return settings
 
 
@@ -335,14 +375,12 @@ def is_default_admin_password_secure(settings: Settings) -> tuple[bool, str]:
     if settings.env == "prod" and password == _DEFAULT_ADMIN_PASSWORD:
         return False, "生产环境禁止使用默认管理员密码 admin123"
     if settings.env == "prod":
-        classes = sum(
+        classes = sum((
-            (
+            any(ch.islower() for ch in password),
-                any(ch.islower() for ch in password),
+            any(ch.isupper() for ch in password),
-                any(ch.isupper() for ch in password),
+            any(ch.isdigit() for ch in password),
-                any(ch.isdigit() for ch in password),
+            any(ch in string.punctuation for ch in password),
-                any(ch in string.punctuation for ch in password),
+        ))
-            )
-        )
         if classes < 3:
             return False, "生产环境默认管理员密码至少覆盖 3 类字符（大小写/数字/符号）"
     if settings.env == "dev" and password == _DEFAULT_ADMIN_PASSWORD:

admin/routes/web_public.py

@@ -28,6 +28,13 @@ from data.customer_portal.token import (
     verify_portal_token,
 )
 from data.crowd_db.loader import CrowdDBLoader
+from data.llm import (
+    LLMClient,
+    LLMError,
+    build_audit_prompt,
+    build_cwb_prompt,
+    build_full_plan_prompt,
+)
 from data.notifications.email_service import DeliveryNotificationService
 from data.orders import crypto
 from data.orders.dao import OrderNotFound, OrdersDAO
@@ -118,6 +125,9 @@ class ReviewResultContract(BaseModel):
     review_input_summary: str = ""
     review_input_attachments: list[str] = Field(default_factory=list)
     review_constraints: dict[str, Any] = Field(default_factory=dict)
+    llm_generated: bool = False
+    llm_summary: str = ""
+    llm_cwb_suggestions: dict[str, list[str]] = Field(default_factory=dict)
 
 
 class ReviewActionRequest(BaseModel):
@@ -4006,6 +4016,108 @@ def _review_constraints_display(value: Any) -> str:
     return text or "待补充"
 
 
+def _get_crowd_db_recs_for_review(constraints: dict[str, Any]) -> list[dict[str, Any]]:
+    """根据当前约束从 crowd_db 取同分段参考。"""
+    province = str(constraints.get("candidate_province") or "").strip()
+    score = constraints.get("candidate_score")
+    if not province or score in (None, ""):
+        return []
+    try:
+        score_int = int(score)
+    except (TypeError, ValueError):
+        return []
+    try:
+        loader = CrowdDBLoader(warn_low_confidence=False)
+        return loader.find_recommendations(province, score_int)
+    except Exception:
+        return []
+
+
+def _llm_review_contract(
+    *,
+    settings: Settings,
+    review_result_id: str,
+    source: Literal["home", "status", "report", "direct"],
+    resolved_summary: str,
+    resolved_constraints: dict[str, Any],
+    resolved_attachments: list[str],
+) -> ReviewResultContract | None:
+    """尝试用 LLM 生成审核结果；未配置或失败时返回 None。"""
+    client = LLMClient(settings)
+    if not client.is_configured:
+        return None
+
+    province = (
+        str(resolved_constraints.get("candidate_province") or "").strip() or "湖南"
+    )
+    score = resolved_constraints.get("candidate_score")
+    rank = resolved_constraints.get("candidate_rank")
+    subjects = list(resolved_constraints.get("candidate_subjects") or [])
+    crowd_recs = _get_crowd_db_recs_for_review(resolved_constraints)
+    system, user = build_audit_prompt(
+        province=province,
+        score=int(score) if score not in (None, "") else None,
+        rank=int(rank) if rank not in (None, "") else None,
+        subjects=[str(s) for s in subjects],
+        existing_plan=resolved_summary,
+        crowd_db_recs=crowd_recs,
+    )
+    try:
+        resp = client.chat_with_system(system, user, temperature=0.3)
+        data = json.loads(resp.content)
+    except (LLMError, json.JSONDecodeError, TypeError, ValueError):
+        return None
+
+    risk_level = str(data.get("risk_level") or "medium").lower()
+    if risk_level not in {"low", "medium", "high"}:
+        risk_level = "medium"
+    findings = [
+        str(x).strip() for x in list(data.get("key_findings") or []) if str(x).strip()
+    ][:5]
+    if not findings:
+        findings = [str(data.get("risk_summary") or "当前方案可继续复核")]
+
+    cwb = data.get("cwb_suggestions") or {}
+    cwb_suggestions = {
+        "rush": [
+            f"{item.get('school', '?')} - {item.get('major', '?')}"
+            for item in list(cwb.get("rush") or [])
+            if isinstance(item, dict)
+        ],
+        "stable": [
+            f"{item.get('school', '?')} - {item.get('major', '?')}"
+            for item in list(cwb.get("stable") or [])
+            if isinstance(item, dict)
+        ],
+        "safety": [
+            f"{item.get('school', '?')} - {item.get('major', '?')}"
+            for item in list(cwb.get("safety") or [])
+            if isinstance(item, dict)
+        ],
+    }
+
+    profile_ready = _is_profile_minimum_complete(resolved_constraints)
+    recommended_action: Literal["go_cwb", "go_step1", "go_full_plan"] = (
+        "go_cwb" if profile_ready else "go_step1"
+    )
+
+    return ReviewResultContract(
+        review_result_id=review_result_id,
+        risk_level=risk_level,
+        top_findings=findings,
+        recommended_action=recommended_action,
+        available_actions=["go_cwb", "go_step1", "go_full_plan"],
+        review_entry_source=source,
+        review_followup_action="none",
+        review_input_summary=resolved_summary or "未提供现有方案说明",
+        review_input_attachments=resolved_attachments,
+        review_constraints=resolved_constraints,
+        llm_generated=True,
+        llm_summary=str(data.get("risk_summary") or ""),
+        llm_cwb_suggestions=cwb_suggestions,
+    )
+
+
 def _start_review_result(
     *,
     source: Literal["home", "status", "report", "direct"],
@@ -4049,23 +4161,34 @@ def _start_review_result(
     recommended_action: Literal["go_cwb", "go_step1", "go_full_plan"] = (
         "go_cwb" if profile_ready else "go_step1"
     )
-    top_finding = (
+
-        "Step 1 已齐全，可直接进入冲稳保微调，再决定是否进入完整规划。"
+    # 优先尝试 LLM 生成审核结果；未配置或失败时回退到规则默认逻辑
-        if profile_ready
+    contract = _llm_review_contract(
-        else "当前方案建议先补充 Step 1 后再继续判断梯度风险。"
+        settings=settings,
-    )
-    contract = ReviewResultContract(
         review_result_id=review_result_id,
-        risk_level="medium",
+        source=source,
-        top_findings=[top_finding],
+        resolved_summary=resolved_summary,
-        recommended_action=recommended_action,
+        resolved_constraints=resolved_constraints,
-        available_actions=["go_cwb", "go_step1", "go_full_plan"],
+        resolved_attachments=resolved_attachments,
-        review_entry_source=source,
-        review_followup_action="none",
-        review_input_summary=resolved_summary or "未提供现有方案说明",
-        review_input_attachments=resolved_attachments,
-        review_constraints=resolved_constraints,
     )
+    if contract is None:
+        top_finding = (
+            "Step 1 已齐全，可直接进入冲稳保微调，再决定是否进入完整规划。"
+            if profile_ready
+            else "当前方案建议先补充 Step 1 后再继续判断梯度风险。"
+        )
+        contract = ReviewResultContract(
+            review_result_id=review_result_id,
+            risk_level="medium",
+            top_findings=[top_finding],
+            recommended_action=recommended_action,
+            available_actions=["go_cwb", "go_step1", "go_full_plan"],
+            review_entry_source=source,
+            review_followup_action="none",
+            review_input_summary=resolved_summary or "未提供现有方案说明",
+            review_input_attachments=resolved_attachments,
+            review_constraints=resolved_constraints,
+        )
     if token:
         order = _resolve_order_from_token(token, settings)
         intake_store = IntakeStore.for_db(settings.orders_db_path)
@@ -4104,6 +4227,11 @@ def _render_review_start_page(contract: ReviewResultContract, token: str | None)
         )
         or "无附件"
     )
+    llm_summary_html = (
+        f'<section class="panel"><h2>AI 风险总结</h2><p class="meta">{escape(contract.llm_summary)}</p></section>'
+        if contract.llm_generated and contract.llm_summary
+        else ""
+    )
     constraints = contract.review_constraints or {}
     recommended_label = {
         "go_cwb": "先去看冲稳保建议",
@@ -4156,6 +4284,7 @@ def _render_review_start_page(contract: ReviewResultContract, token: str | None)
     <p id="share-status" style="margin:8px 0 0;font-size:12px;color:#5a7cb8;" role="status" aria-live="polite"></p>
   </div>
 </section>
+{llm_summary_html}
 <script>
 (function() {{
   var statusEl = document.getElementById('share-status');
@@ -4359,7 +4488,26 @@ def _render_cwb_placeholder_page(
         pass
 
     def _cwb_tier(title: str, offset: int, color: str) -> str:
-        """根据分数偏移生成一档建议，数据来自 crowd_db。"""
+        """根据分数偏移生成一档建议，优先使用 LLM 建议，fallback 到 crowd_db。"""
+        # 1. 优先使用 LLM 生成的三档建议
+        tier_key = {"冲刺建议": "rush", "稳妥建议": "stable", "保底建议": "safety"}.get(
+            title
+        )
+        if contract is not None and contract.llm_cwb_suggestions and tier_key:
+            llm_suggestions = contract.llm_cwb_suggestions.get(tier_key) or []
+            if llm_suggestions:
+                schools_html = "".join(
+                    f"<li>{escape(s)}</li>" for s in llm_suggestions[:3]
+                )
+                return (
+                    f'<article style="padding:16px;border-radius:14px;background:{color};border:1px solid #d7e3f1;">'
+                    f"<h2>{title}</h2>"
+                    f'<ul style="margin:8px 0;padding-left:18px;line-height:1.8;">{schools_html}</ul>'
+                    f'<p class="meta" style="margin-top:6px;color:#1f6feb;">🤖 LLM 结合你的分数、位次和同分段数据生成</p>'
+                    f"</article>"
+                )
+
+        # 2. fallback 到 crowd_db
         if candidate_score is None:
             return (
                 f'<article style="padding:16px;border-radius:14px;background:{color};border:1px solid #d7e3f1;">'
@@ -4367,6 +4515,7 @@ def _render_cwb_placeholder_page(
                 f'<p class="meta">补齐当前分数后，这里会基于同分段数据给出具体的院校方向建议。</p>'
                 f"</article>"
             )
+
         target_score = candidate_score + offset
         try:
             from data.crowd_db.loader import CrowdDBLoader

admin/tests/conftest.py

@@ -70,15 +70,13 @@ class RouteClient:
             (key.lower().encode("utf-8"), value.encode("utf-8"))
             for key, value in (headers or {}).items()
         ]
-        return Request(
+        return Request({
-            {
+            "type": "http",
-                "type": "http",
+            "method": method,
-                "method": method,
+            "path": split.path,
-                "path": split.path,
+            "query_string": split.query.encode("utf-8"),
-                "query_string": split.query.encode("utf-8"),
+            "headers": raw_headers,
-                "headers": raw_headers,
+        })
-            }
-        )
 
     @staticmethod
     def _html_response(response) -> RouteResponse:
@@ -354,6 +352,8 @@ def settings(tmp_path, secure_secret, monkeypatch):
     monkeypatch.setenv("GAOKAO_ALERT_WEBHOOK_URLS", "")
     monkeypatch.setenv("GAOKAO_ORDERS_FERNET_KEY", "test-secret-for-web-self-service")
     monkeypatch.setenv("GAOKAO_JWT_SECRET", secure_secret)
+    monkeypatch.setenv("GAOKAO_LLM_PROVIDER", "dashscope")
+    monkeypatch.setenv("GAOKAO_LLM_API_KEY", "sk-test")
     monkeypatch.setenv("GAOKAO_JWT_EXP_MIN", "5")
     monkeypatch.setenv("GAOKAO_ADMIN_PASS", "test-pass-123")
     monkeypatch.setenv("GAOKAO_OPS_ALERT_LOG", ops_alert_log)

admin/tests/test_app.py

@@ -119,7 +119,6 @@ def test_dashboard_page_served(client, auth_headers):
     assert "接口: <code>/api/stats/dashboard</code>" not in body
 
 
-
 def test_dashboard_static_js_served(client):
     """前端脚本包含趋势切换与 3 张分布图渲染逻辑。"""
     resp = client.get("/static/dashboard.js")
@@ -242,6 +241,8 @@ def test_prod_rejects_default_admin_password(tmp_path, monkeypatch):
     monkeypatch.setenv("GAOKAO_ADMIN_USER", "admin")
     monkeypatch.setenv("GAOKAO_ADMIN_PASS", "admin123")
     monkeypatch.setenv("GAOKAO_PAYMENT_PROVIDER", "alipay")
+    monkeypatch.setenv("GAOKAO_LLM_PROVIDER", "dashscope")
+    monkeypatch.setenv("GAOKAO_LLM_API_KEY", "sk-test")
     # 显式提供合规 webhook secret,避免被 P2-5 fail-closed 提前拦截,
     # 让本测试聚焦于管理员密码策略。
     monkeypatch.setenv("GAOKAO_PAYMENT_WEBHOOK_SECRET", "P" + "r" * 31 + "!" * 32)

admin/tests/test_health.py

@@ -29,6 +29,8 @@ def test_load_settings_prod_rejects_dev_jwt_secret(monkeypatch, tmp_path):
     monkeypatch.setenv("GAOKAO_PORTAL_TOKEN_SECRET", "y" * 32)
     monkeypatch.setenv("GAOKAO_ADMIN_PASS", "StrongPass1!")
     monkeypatch.setenv("GAOKAO_PAYMENT_PROVIDER", "alipay")
+    monkeypatch.setenv("GAOKAO_LLM_PROVIDER", "dashscope")
+    monkeypatch.setenv("GAOKAO_LLM_API_KEY", "sk-test")
     monkeypatch.setenv("GAOKAO_ORDERS_DB_PATH", str(tmp_path / "orders.db"))
     monkeypatch.setenv("GAOKAO_DB_PATH", str(tmp_path / "admin.db"))
 
@@ -47,6 +49,8 @@ def test_load_settings_prod_rejects_short_jwt_secret(monkeypatch, tmp_path):
     monkeypatch.setenv("GAOKAO_PORTAL_TOKEN_SECRET", "y" * 32)
     monkeypatch.setenv("GAOKAO_ADMIN_PASS", "StrongPass1!")
     monkeypatch.setenv("GAOKAO_PAYMENT_PROVIDER", "alipay")
+    monkeypatch.setenv("GAOKAO_LLM_PROVIDER", "dashscope")
+    monkeypatch.setenv("GAOKAO_LLM_API_KEY", "sk-test")
     monkeypatch.setenv("GAOKAO_ORDERS_DB_PATH", str(tmp_path / "orders.db"))
     monkeypatch.setenv("GAOKAO_DB_PATH", str(tmp_path / "admin.db"))
 
@@ -65,6 +69,8 @@ def test_load_settings_prod_rejects_default_admin_password(monkeypatch, tmp_path
     monkeypatch.setenv("GAOKAO_PORTAL_TOKEN_SECRET", "y" * 32)
     monkeypatch.setenv("GAOKAO_ADMIN_PASS", "admin123")
     monkeypatch.setenv("GAOKAO_PAYMENT_PROVIDER", "alipay")
+    monkeypatch.setenv("GAOKAO_LLM_PROVIDER", "dashscope")
+    monkeypatch.setenv("GAOKAO_LLM_API_KEY", "sk-test")
     monkeypatch.setenv("GAOKAO_ORDERS_DB_PATH", str(tmp_path / "orders.db"))
     monkeypatch.setenv("GAOKAO_DB_PATH", str(tmp_path / "admin.db"))
 

admin/tests/test_p2_4_p2_5_secrets.py

@@ -18,8 +18,13 @@ import pytest
 
 def _reload_settings():
     """重新加载 Settings(确保读取最新环境变量)。"""
+    import os
     from admin.config import load_settings
 
+    # LLM 是生产必需项；本测试聚焦 portal/payment secret，不希望被 LLM 校验提前拦截
+    os.environ.setdefault("GAOKAO_LLM_PROVIDER", "dashscope")
+    os.environ.setdefault("GAOKAO_LLM_API_KEY", "sk-test")
+
     return load_settings()
 
 

data/llm/__init__.py

@@ -0,0 +1,20 @@
+"""LLM 集成模块。
+
+支持 openai/dashscope/anthropic 三种供应商，通过统一的 OpenAI-compatible API 调用。
+"""
+
+from .client import LLMClient, LLMResponse, LLMError
+from .prompts import (
+    build_audit_prompt,
+    build_cwb_prompt,
+    build_full_plan_prompt,
+)
+
+__all__ = [
+    "LLMClient",
+    "LLMResponse",
+    "LLMError",
+    "build_audit_prompt",
+    "build_cwb_prompt",
+    "build_full_plan_prompt",
+]

data/llm/client.py

@@ -0,0 +1,144 @@
+"""LLM 客户端：统一 OpenAI-compatible 接口调用。"""
+
+from __future__ import annotations
+
+import json
+import urllib.request
+import urllib.error
+from dataclasses import dataclass, field
+from typing import Any
+
+from admin.config import Settings
+
+
+class LLMError(Exception):
+    """LLM 调用失败。"""
+
+
+@dataclass(frozen=True)
+class LLMResponse:
+    """LLM 响应。"""
+
+    content: str
+    usage: dict[str, int] = field(default_factory=dict)
+    model: str = ""
+    raw: dict[str, Any] = field(default_factory=dict)
+
+
+class LLMClient:
+    """统一 LLM 客户端，通过 OpenAI-compatible API 调用。
+
+    支持:
+    - openai: https://api.openai.com/v1
+    - dashscope: https://dashscope.aliyuncs.com/compatible-mode/v1
+    - anthropic: 通过兼容层或直接 API
+    """
+
+    def __init__(self, settings: Settings) -> None:
+        self._settings = settings
+        self._provider = settings.llm_provider
+        self._api_key = settings.llm_api_key
+        self._base_url = settings.llm_base_url.rstrip("/")
+        self._model = settings.llm_model
+        self._timeout = settings.llm_timeout_seconds
+        self._max_tokens = settings.llm_max_tokens
+
+    @property
+    def is_configured(self) -> bool:
+        """LLM 是否已配置可用。"""
+        return self._provider != "none" and bool(self._api_key)
+
+    def chat(
+        self,
+        messages: list[dict[str, str]],
+        *,
+        temperature: float = 0.7,
+        max_tokens: int | None = None,
+    ) -> LLMResponse:
+        """调用 chat completions API。
+
+        Args:
+            messages: OpenAI 格式的消息列表。
+            temperature: 采样温度。
+            max_tokens: 最大生成 token 数，默认使用 Settings 配置。
+
+        Returns:
+            LLMResponse。
+
+        Raises:
+            LLMError: 调用失败。
+        """
+        if not self.is_configured:
+            raise LLMError(
+                f"LLM 未配置 (provider={self._provider})。"
+                "请设置 GAOKAO_LLM_PROVIDER 和 GAOKAO_LLM_API_KEY。"
+            )
+
+        payload: dict[str, Any] = {
+            "model": self._model,
+            "messages": messages,
+            "temperature": temperature,
+            "max_tokens": max_tokens or self._max_tokens,
+        }
+
+        url = f"{self._base_url}/chat/completions"
+        headers = {
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {self._api_key}",
+        }
+
+        data = json.dumps(payload).encode("utf-8")
+        req = urllib.request.Request(url, data=data, headers=headers, method="POST")
+
+        try:
+            with urllib.request.urlopen(req, timeout=self._timeout) as resp:
+                body = resp.read().decode("utf-8")
+                result = json.loads(body)
+        except urllib.error.HTTPError as e:
+            raw_body = e.read()
+            error_body = (
+                raw_body.decode("utf-8", "replace")
+                if isinstance(raw_body, bytes)
+                else str(raw_body)
+            )
+            raise LLMError(f"LLM API HTTP {e.code}: {error_body[:500]}") from e
+        except urllib.error.URLError as e:
+            raise LLMError(f"LLM API 连接失败: {e}") from e
+        except json.JSONDecodeError as e:
+            raise LLMError(f"LLM API 响应解析失败: {e}") from e
+
+        choices = result.get("choices", [])
+        if not choices:
+            raise LLMError(f"LLM API 返回空 choices: {result}")
+
+        content = choices[0].get("message", {}).get("content", "")
+        if not content:
+            raise LLMError(f"LLM API 返回空 content: {result}")
+
+        usage = result.get("usage", {})
+        model = result.get("model", self._model)
+
+        return LLMResponse(
+            content=content,
+            usage=usage,
+            model=model,
+            raw=result,
+        )
+
+    def chat_with_system(
+        self,
+        system_prompt: str,
+        user_prompt: str,
+        *,
+        temperature: float = 0.7,
+        max_tokens: int | None = None,
+    ) -> LLMResponse:
+        """便捷方法：system + user 两条消息。"""
+        return self.chat(
+            [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt},
+            ],
+            temperature=temperature,
+            max_tokens=max_tokens,
+        )

data/llm/prompts.py

@@ -0,0 +1,223 @@
+"""志愿填报 LLM Prompt 模板。"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def build_audit_prompt(
+    *,
+    province: str,
+    score: int | None,
+    rank: int | None,
+    subjects: list[str],
+    existing_plan: str,
+    crowd_db_recs: list[dict[str, Any]] | None = None,
+) -> tuple[str, str]:
+    """构建志愿方案审核 prompt。
+
+    Returns:
+        (system_prompt, user_prompt)
+    """
+    system = (
+        "你是一位资深高考志愿填报顾问，精通各省录取规则、院校层次和风险识别。"
+        "你的任务是审核用户提供的现有志愿方案，识别踩线、扎堆、梯度失衡等风险，"
+        "并给出具体可操作的改进建议。"
+        "请用中文回答，输出 JSON 格式。"
+    )
+
+    context_parts = [
+        f"考试省份：{province}",
+        f"高考分数：{score or '未提供'}",
+        f"全省位次：{rank or '未提供'}",
+        f"选科组合：{'、'.join(subjects) if subjects else '未提供'}",
+    ]
+
+    if crowd_db_recs:
+        top_schools = [
+            f"{r.get('name', '?')} - {r.get('major', '?')}"
+            for r in crowd_db_recs[:5]
+            if isinstance(r, dict)
+        ]
+        if top_schools:
+            context_parts.append(f"同分段热门院校：{'；'.join(top_schools)}")
+
+    context = "\n".join(context_parts)
+
+    user = f"""请审核以下志愿方案并给出风险评估。
+
+## 考生基本信息
+{context}
+
+## 现有方案说明
+{existing_plan or "用户未提供具体方案内容"}
+
+## 请输出
+请以 JSON 格式输出审核结果，包含以下字段：
+{{
+  "risk_level": "low|medium|high",
+  "risk_summary": "一句话风险总结",
+  "key_findings": ["风险点1", "风险点2", ...],
+  "suggestions": ["建议1", "建议2", ...],
+  "cwb_suggestions": {{
+    "rush": ["冲刺院校1 - 专业", ...],
+    "stable": ["稳妥院校1 - 专业", ...],
+    "safety": ["保底院校1 - 专业", ...]
+  }}
+}}
+
+注意：
+- risk_level 基于踩线风险、扎堆程度和梯度合理性综合判断
+- key_findings 最多 5 条，每条不超过 50 字
+- cwb_suggestions 每档至少 2 个院校-专业组合
+- 如果信息不足，在 key_findings 里说明需要补充什么"""
+
+    return system, user
+
+
+def build_cwb_prompt(
+    *,
+    province: str,
+    score: int,
+    rank: int | None,
+    subjects: list[str],
+    target_cities: list[str] | None = None,
+    target_majors: list[str] | None = None,
+    crowd_db_recs: list[dict[str, Any]] | None = None,
+) -> tuple[str, str]:
+    """构建冲稳保方案生成 prompt。"""
+    system = (
+        "你是一位资深高考志愿填报顾问。根据考生分数、位次和偏好，"
+        "生成冲稳保三档院校-专业建议。"
+        "请用中文回答，输出 JSON 格式。"
+    )
+
+    context_parts = [
+        f"省份：{province}",
+        f"分数：{score}",
+        f"位次：{rank or '未提供'}",
+        f"选科：{'、'.join(subjects) if subjects else '未提供'}",
+    ]
+    if target_cities:
+        context_parts.append(f"目标城市：{'、'.join(target_cities)}")
+    if target_majors:
+        context_parts.append(f"目标专业：{'、'.join(target_majors)}")
+    if crowd_db_recs:
+        recs = [
+            f"{r.get('name', '?')}({r.get('major', '?')})"
+            for r in crowd_db_recs[:8]
+            if isinstance(r, dict)
+        ]
+        if recs:
+            context_parts.append(f"同分段参考：{'；'.join(recs)}")
+
+    context = "\n".join(context_parts)
+
+    user = f"""请为以下考生生成冲稳保三档建议。
+
+## 考生信息
+{context}
+
+## 请输出 JSON：
+{{
+  "rush": [
+    {{"school": "校名", "major": "专业", "reason": "推荐理由（简短）"}}
+  ],
+  "stable": [
+    {{"school": "校名", "major": "专业", "reason": "推荐理由"}}
+  ],
+  "safety": [
+    {{"school": "校名", "major": "专业", "reason": "推荐理由"}}
+  ]
+}}
+
+要求：
+- 每档至少 3 个院校-专业组合
+- 冲刺档目标分数约 {score + 20} 分段
+- 稳妥档围绕 {score} 分段
+- 保底档约 {score - 20} 分段
+- 院校名和专业名要真实存在"""
+
+    return system, user
+
+
+def build_full_plan_prompt(
+    *,
+    province: str,
+    score: int,
+    rank: int | None,
+    subjects: list[str],
+    target_cities: list[str] | None = None,
+    target_majors: list[str] | None = None,
+    family_background: str | None = None,
+    interest_assessment: str | None = None,
+    existing_plan: str | None = None,
+    crowd_db_recs: list[dict[str, Any]] | None = None,
+) -> tuple[str, str]:
+    """构建完整志愿方案生成 prompt。"""
+    system = (
+        "你是一位资深高考志愿填报顾问，擅长结合考生分数、位次、选科、"
+        "偏好和家庭背景，生成完整、可执行的志愿方案。"
+        "方案应包含冲稳保梯度、院校专业推荐和风险提示。"
+        "请用中文回答，输出 JSON 格式。"
+    )
+
+    context_parts = [
+        f"省份：{province}",
+        f"分数：{score}",
+        f"位次：{rank or '未提供'}",
+        f"选科：{'、'.join(subjects) if subjects else '未提供'}",
+    ]
+    if target_cities:
+        context_parts.append(f"目标城市：{'、'.join(target_cities)}")
+    if target_majors:
+        context_parts.append(f"目标专业：{'、'.join(target_majors)}")
+    if family_background:
+        context_parts.append(f"家庭背景：{family_background}")
+    if interest_assessment:
+        context_parts.append(f"兴趣测评：{interest_assessment}")
+    if existing_plan:
+        context_parts.append(f"已有方案：{existing_plan}")
+    if crowd_db_recs:
+        recs = [
+            f"{r.get('name', '?')}({r.get('major', '?')})"
+            for r in crowd_db_recs[:10]
+            if isinstance(r, dict)
+        ]
+        if recs:
+            context_parts.append(f"同分段参考：{'；'.join(recs)}")
+
+    context = "\n".join(context_parts)
+
+    user = f"""请为以下考生生成完整的志愿填报方案。
+
+## 考生完整信息
+{context}
+
+## 请输出 JSON：
+{{
+  "overall_assessment": "总体评价（2-3句话）",
+  "risk_level": "low|medium|high",
+  "strategy": "核心策略说明",
+  "volunteers": [
+    {{
+      "batch": "提前批|本科批|专科批",
+      "tier": "冲|稳|保",
+      "school": "校名",
+      "major": "专业",
+      "reason": "推荐理由",
+      "risk_note": "风险提示（可空）"
+    }}
+  ],
+  "warnings": ["注意事项1", "注意事项2"],
+  "next_steps": ["建议后续动作1", "建议后续动作2"]
+}}
+
+要求：
+- volunteers 至少 8 条，覆盖冲稳保三档
+- 每条都有具体的院校名和专业名（真实存在）
+- 按 tier 分组排序：先冲后稳再保
+- warnings 至少 2 条"""
+
+    return system, user

data/llm/tests/__init__.py

@@ -0,0 +1,8 @@
+"""data/llm tests."""
+
+from data.llm.client import LLMClient, LLMResponse, LLMError
+from data.llm.prompts import (
+    build_audit_prompt,
+    build_cwb_prompt,
+    build_full_plan_prompt,
+)

data/llm/tests/test_llm.py

@@ -0,0 +1,162 @@
+"""LLM 模块测试。"""
+
+from __future__ import annotations
+
+import pytest
+from unittest.mock import patch, MagicMock
+from dataclasses import dataclass
+
+from data.llm.client import LLMClient, LLMResponse, LLMError
+from data.llm.prompts import (
+    build_audit_prompt,
+    build_cwb_prompt,
+    build_full_plan_prompt,
+)
+
+
+@dataclass
+class MockSettings:
+    llm_provider: str = "none"
+    llm_api_key: str = ""
+    llm_base_url: str = "https://dashscope.aliyuncs.com/compatible-mode/v1"
+    llm_model: str = "qwen-plus"
+    llm_timeout_seconds: int = 60
+    llm_max_tokens: int = 4096
+
+
+class TestLLMClient:
+    def test_not_configured_when_provider_none(self):
+        client = LLMClient(MockSettings(llm_provider="none"))
+        assert not client.is_configured
+
+    def test_not_configured_when_no_api_key(self):
+        client = LLMClient(MockSettings(llm_provider="openai", llm_api_key=""))
+        assert not client.is_configured
+
+    def test_configured_when_provider_and_key_set(self):
+        client = LLMClient(MockSettings(llm_provider="openai", llm_api_key="sk-test"))
+        assert client.is_configured
+
+    def test_chat_raises_when_not_configured(self):
+        client = LLMClient(MockSettings(llm_provider="none"))
+        with pytest.raises(LLMError, match="LLM 未配置"):
+            client.chat([{"role": "user", "content": "test"}])
+
+    @patch("data.llm.client.urllib.request.urlopen")
+    def test_chat_success(self, mock_urlopen):
+        mock_resp = MagicMock()
+        mock_resp.read.return_value = b'{"choices":[{"message":{"content":"test response"}}],"usage":{"total_tokens":10},"model":"qwen-plus"}'
+        mock_resp.__enter__ = MagicMock(return_value=mock_resp)
+        mock_resp.__exit__ = MagicMock(return_value=False)
+        mock_urlopen.return_value = mock_resp
+
+        client = LLMClient(
+            MockSettings(llm_provider="dashscope", llm_api_key="sk-test")
+        )
+        result = client.chat([{"role": "user", "content": "hello"}])
+
+        assert isinstance(result, LLMResponse)
+        assert result.content == "test response"
+        assert result.model == "qwen-plus"
+        assert result.usage["total_tokens"] == 10
+
+    @patch("data.llm.client.urllib.request.urlopen")
+    def test_chat_http_error(self, mock_urlopen):
+        import urllib.error
+        import io
+
+        error_fp = io.BytesIO(b'{"error":"bad key"}')
+        mock_urlopen.side_effect = urllib.error.HTTPError(
+            "http://test", 401, "Unauthorized", {}, error_fp
+        )
+        client = LLMClient(MockSettings(llm_provider="openai", llm_api_key="bad"))
+        with pytest.raises(LLMError, match="HTTP 401"):
+            client.chat([{"role": "user", "content": "test"}])
+
+    def test_chat_with_system(self):
+        """chat_with_system 构建正确的消息结构。"""
+        with patch.object(LLMClient, "chat") as mock_chat:
+            mock_chat.return_value = LLMResponse(content="ok")
+            client = LLMClient(
+                MockSettings(llm_provider="openai", llm_api_key="sk-test")
+            )
+            client.chat_with_system("you are helpful", "hello")
+
+            call_args = mock_chat.call_args
+            messages = call_args[0][0]
+            assert len(messages) == 2
+            assert messages[0]["role"] == "system"
+            assert messages[0]["content"] == "you are helpful"
+            assert messages[1]["role"] == "user"
+            assert messages[1]["content"] == "hello"
+
+
+class TestPrompts:
+    def test_audit_prompt_structure(self):
+        system, user = build_audit_prompt(
+            province="湖南",
+            score=578,
+            rank=12000,
+            subjects=["物理", "化学", "生物"],
+            existing_plan="已有一版方案",
+        )
+        assert "志愿填报顾问" in system
+        assert "湖南" in user
+        assert "578" in user
+        assert "物理" in user
+        assert "JSON" in user
+        assert "risk_level" in user
+
+    def test_audit_prompt_with_crowd_db(self):
+        recs = [
+            {"name": "湖南大学", "major": "计算机"},
+            {"name": "中南大学", "major": "软件工程"},
+        ]
+        system, user = build_audit_prompt(
+            province="湖南",
+            score=578,
+            rank=12000,
+            subjects=["物理"],
+            existing_plan="test",
+            crowd_db_recs=recs,
+        )
+        assert "湖南大学" in user
+        assert "中南大学" in user
+
+    def test_audit_prompt_minimal(self):
+        system, user = build_audit_prompt(
+            province="广东",
+            score=None,
+            rank=None,
+            subjects=[],
+            existing_plan="",
+        )
+        assert "未提供" in user
+
+    def test_cwb_prompt_structure(self):
+        system, user = build_cwb_prompt(
+            province="湖南",
+            score=578,
+            rank=12000,
+            subjects=["物理", "化学", "生物"],
+            target_cities=["长沙", "深圳"],
+        )
+        assert "冲稳保" in system
+        assert "长沙" in user
+        assert "598" in user  # score + 20
+        assert "558" in user  # score - 20
+
+    def test_full_plan_prompt_structure(self):
+        system, user = build_full_plan_prompt(
+            province="湖南",
+            score=578,
+            rank=12000,
+            subjects=["物理", "化学", "生物"],
+            target_majors=["计算机科学", "人工智能"],
+            family_background="家长希望省内优先",
+        )
+        assert "完整" in system
+        assert "计算机科学" in user
+        assert "家长希望省内优先" in user
+        assert "volunteers" in user
+        assert "至少 8 条" in user

scripts/payment_readiness_doctor.py

@@ -17,6 +17,10 @@ import sys
 from pathlib import Path
 
 REQUIRED_ENV_VARS = [
+    "GAOKAO_LLM_PROVIDER",
+    "GAOKAO_LLM_API_KEY",
+    "GAOKAO_LLM_BASE_URL",
+    "GAOKAO_LLM_MODEL",
     "GAOKAO_PAYMENT_APP_ID",
     "GAOKAO_PAYMENT_MERCHANT_ID",
     "GAOKAO_PAYMENT_PRIVATE_KEY_PATH",