Python与大语言模型实战从风险分级到RAG完整搭建心理健康教育智能助手从工程边界、风险识别、RAG检索到 FastAPI 接口与安全治理的完整项目实战

xiaoxingkongyuxi

39人浏览 · 2026-08-27 17:30:00

xiaoxingkongyuxi · 2026-08-27 17:30:00 发布

Python与大语言模型实战从风险分级到RAG完整搭建心理健康教育智能助手

从工程边界、风险识别、RAG检索到 FastAPI 接口与安全治理的完整项目实战

先说结论 把大语言模型接到聊天框并不难，真正困难的是让系统在“知识是否可靠、风险是否升级、隐私是否最小化、模型是否应该继续回答”这些关键节点做出正确选择。这个项目的核心不是让模型说得更多，而是让它在该回答时有证据地回答，在不该回答时及时让路给安全策略和专业人员。

如果把一个普通聊天机器人直接搬进心理健康教育场景，它很快会暴露出四个问题：第一，模型可能把一般情绪描述说成疾病；第二，模型可能编造并不存在的求助资源；第三，遇到高风险表达时，普通对话逻辑可能仍然继续生成；第四，日志、会话和画像如果没有最小化原则，会把高度敏感的信息当作普通业务数据长期保存。

因此，本文从一个可运行的 Python 项目出发，把系统拆成“风险先行、知识约束、结构化生成、输出过滤、人工升级、最小审计”六个关键环节。你可以把它看成一个课程设计、毕业设计、校园服务原型或企业内部教育工具的工程底座，也可以继续替换模型、向量库、前端和知识源，把它扩展成更完整的平台。

一、项目到底解决什么问题

心理健康教育的需求并不只发生在咨询室里。考试、就业、家庭关系、社交冲突、长期睡眠紊乱和信息过载等问题，会让很多人先从搜索引擎、社交平台或智能问答工具获取信息。传统课程和人工咨询具有专业性，但服务时间、覆盖范围和交互频率有限；而网络信息又存在质量不一、过度诊断、错误建议和缺少危机识别等风险。

这个智能助手的定位，是提供及时、温和、规范且具有教育价值的信息支持：解释常见心理健康知识，帮助用户整理情绪和压力，给出简单可执行的练习，推荐正规求助渠道，并在出现紧急危险信号时切换到安全处置流程。

边界必须写在系统里 系统不承担临床诊断、处方或治疗职责；不根据几句话给出疾病结论；不提供危险行为方法；不编造热线或机构信息；不鼓励用户只依赖智能助手。

用户需求	系统可以做什么	系统不应该做什么
知识咨询	解释压力反应、睡眠卫生、情绪调节等通识内容	把一般体验直接诊断为疾病
情绪支持	帮助整理事实、感受、需求，给出低风险练习	承诺“保证治好”或形成依赖
持续困扰	建议专业评估，提供正规资源入口	代替咨询师完成长期治疗
紧急危险	停止普通问答，给出清晰安全引导并触发人工介入	继续追问危险行为的实施细节

二、为什么这不是“接一个大模型 API”这么简单

在普通问答产品中，最常见的主链路是“用户输入 → 模型生成 → 返回结果”。在心理健康教育场景里，这条链路必须改写，因为“先生成再过滤”本身就可能太晚。更合理的顺序是先做安全判断，再决定是否允许进入知识检索和大模型生成。

图 2｜请求处理主链路：风险判断先于普通生成

这条链路有一个非常重要的工程原则：高风险结果的优先级必须高于普通意图分类。即使用户的问题同时包含“睡眠”“考试”“焦虑”等普通主题，只要风险模块发现明确的紧急信号，就应直接进入安全分支，普通 RAG 和自由生成都不再继续。

风险识别解决“这条消息还能不能按普通问答处理”。
意图识别解决“如果可以回答，当前更像睡眠、压力、人际、知识解释还是求助资源”。
RAG 解决“模型应该依据哪些经过审核的资料回答”。
输出过滤解决“模型虽然生成了，但有没有越过诊断、危险方法、虚假资源等边界”。
人工升级解决“机器识别到中高风险后，谁来接手、如何复核、如何闭环”。

三、总体架构设计：五层结构让职责不混在一起

图 3｜表现层、业务层、智能层、数据层与安全治理层

系统采用五层结构。表现层只负责交互；业务层负责会话、权限、限流和异常；智能层完成风险识别、意图分类、知识检索和模型调用；数据层保存会话、知识、审核状态与安全事件；安全治理层贯穿所有模块，负责敏感信息处理、输出拦截、人工转介和审计。

层级	核心职责	典型实现
表现层	聊天界面、知识卡片、风险提示、求助入口	Web / App / 校园门户
业务层	接口、鉴权、会话、限流、异常处理	FastAPI + Pydantic
智能层	风险、意图、RAG、提示编排、模型调用	规则 + 分类器 + 检索 + LLM
数据层	知识、消息、会话、安全事件、审核状态	SQLite / PostgreSQL + 向量库
安全治理层	隐私、过滤、人工升级、审计、回归测试	策略引擎 + RBAC + 审计

四、项目目录与环境准备

为了让示例既能看懂又便于后续扩展，可以把核心模块按职责拆开。小型课程项目也可以先放在一个文件里，但一旦需要加入测试、数据持久化和模型切换，模块化会明显降低维护成本。

推荐目录结构

mental-health-assistant/
├─ app/
│ ├─ main.py                 # FastAPI 入口
│ ├─ models.py               # 请求/响应数据模型
│ ├─ safety.py               # 风险识别与安全回复
│ ├─ intent.py               # 意图分类
│ ├─ retrieval.py            # 知识检索
│ ├─ llm.py                  # 模型调用与提示词组装
│ └─ repository.py           # 会话/知识/审计持久化
├─ data/
│ └─ knowledge.json          # 演示知识库
├─ tests/
│ └─ test_core.py            # 核心逻辑测试
├─ .env.example
└─ requirements.txt

截至 2026 年 8 月，FastAPI、Pydantic 和 Uvicorn 都已经有新版本。为了避免“教程代码长期有效，但依赖版本不清楚”的问题，建议在文章或项目中明确锁定版本；如果你后续升级依赖，应先跑一遍接口和安全回归测试。

requirements.txt（示例锁定版本）

fastapi==0.141.1
uvicorn[standard]==0.52.4
pydantic==2.13.4
httpx==0.28.1

版本说明 本文核心代码使用的是稳定的 Python 类型标注、Pydantic BaseModel、FastAPI 路由和 HTTPX 异步请求接口。即使未来小版本变化，整体设计思想仍然有效；真正需要关注的是框架 API 变更、模型服务接口和向量库 SDK。

五、第一道安全门：风险分级必须先于生成

风险分级的目的不是“预测疾病”，而是判断当前对话是否需要更高等级的支持。工程上可以把规则、语义模型和上下文信号组合起来：规则擅长发现明确表达，语义模型擅长理解隐喻和委婉表达，上下文则用于判断风险是否升级。

图 4｜四级风险分层：分级的是支持强度，不是疾病标签

下面给出一个可运行的规则示例。它刻意保持简单，目的是演示“强信号直达 3 级、一般困扰进入 1 级、持续绝望或明显失控进入 2 级”的结构。生产环境必须增加否定语境、上下文窗口、未成年人保护、语义分类、专业人员标注集和持续回归测试。

app/safety.py｜风险识别核心

import re
from dataclasses import dataclass
from typing import List

@dataclass
class RiskResult:
    level: int
    score: int
    matched: List[str]
    urgent: bool

HIGH_RISK = [
    "正在伤害自己",
    "已经伤害自己",
    "无法保证安全",
    "无法保证自己的安全",
]
PLAN_WORDS = [
    "准备伤害自己",
    "计划自伤",
    "准备结束生命",
]
SEVERE_DISTRESS = ["活着没意思", "撑不下去", "极度绝望", "失控"]
GENERAL_DISTRESS = ["焦虑", "紧张", "低落", "孤独", "压力很大", "睡不着"]

def normalize(text: str) -> str:
    text = text.strip().lower()
    return re.sub(r"\s+", "", text)

def detect_risk(text: str) -> RiskResult:
    clean = normalize(text)
    matched: List[str] = []
    score = 0
    urgent = False

    if any(word in clean for word in HIGH_RISK):
        matched.append("正在发生或安全无法保证")
        score += 100
        urgent = True

    if any(word in clean for word in PLAN_WORDS):
        matched.append("明确计划")
        score += 80
        urgent = True

    if any(word in clean for word in SEVERE_DISTRESS):
        matched.append("强烈痛苦")
        score += 35

    if any(word in clean for word in GENERAL_DISTRESS):
        matched.append("一般困扰")
        score += 10

  if urgent or score >= 80:
        level = 3
    elif score >= 30:
        level = 2
    elif score >= 10:
        level = 1
    else:
        level = 0

    return RiskResult(level, score, matched, urgent)

为什么不能只用关键词 一句“我没打算伤害自己”里仍然包含高风险词。如果只做字符串包含判断，可能产生误报；反过来，隐喻、反讽、缩写和上下文变化又可能造成漏报。因此规则适合做第一层兜底，但不能被包装成“智能诊断器”。

六、第二道判断：意图识别决定知识检索方向

当风险模块允许进入普通问答后，系统再判断当前需求属于什么主题。意图分类不需要一开始就上复杂模型，有限业务类别用规则就能搭出稳定原型；当数据积累后，再替换成文本分类器或结构化 LLM 分类。

app/intent.py｜轻量意图分类

from typing import Dict, List

INTENT_TERMS: Dict[str, List[str]] = {
    "sleep": ["睡不着", "失眠", "早醒", "睡眠"],
    "stress": ["压力", "焦虑", "紧张", "考试", "工作"],
    "relationship": ["争吵", "朋友", "同学", "沟通", "关系"],
    "knowledge": ["什么是", "怎么理解", "含义", "心理健康"],
    "help": ["咨询", "求助", "心理中心", "医院", "资源"],
}

def classify_intent(text: str) -> str:
    clean = normalize(text)
    scores = {
        name: sum(1 for term in terms if term in clean)
        for name, terms in INTENT_TERMS.items()
    }
    best = max(scores, key=scores.get)
    return best if scores[best] > 0 else "general"

真实系统还应保存分类置信度。低置信度时，最稳妥的策略不是强行猜测，而是提出一个开放式澄清问题，例如“你更想了解睡眠改善的方法，还是想先说说最近让你压力最大的事情？”这样比错误路由到某个知识主题更自然。

七、RAG：先找证据，再让模型组织语言

大语言模型的“流畅”并不等于“真实”。心理健康教育场景尤其不能接受模型随意编造热线、机构、医学结论或不适用的建议。因此，一个可靠的做法是把经过审核的课程资料、服务流程、睡眠卫生知识、压力管理练习和求助指南放进知识库，先检索，再生成。

图 5｜RAG：审核知识先进入上下文，模型负责组织而不是凭空发明

app/retrieval.py｜无需向量数据库的最小可运行检索

from dataclasses import dataclass
from collections import Counter
from typing import List
import math, re

@dataclass
class KnowledgeChunk:
    chunk_id: str
    topic: str
    content: str
    source: str
    reviewed: bool = True

def vectorize(text: str) -> Counter:
    tokens = re.findall(r"[\u4e00-\u9fff]|[a-zA-Z0-9]+", text.lower())
    return Counter(tokens)

def cosine_similarity(left: Counter, right: Counter) -> float:
    keys = set(left) | set(right)
    numerator = sum(left[k] * right[k] for k in keys)
    left_norm = math.sqrt(sum(v * v for v in left.values()))
    right_norm = math.sqrt(sum(v * v for v in right.values()))
    if left_norm == 0 or right_norm == 0:
        return 0.0
    return numerator / (left_norm * right_norm)

def retrieve(query: str, chunks: List[KnowledgeChunk], top_k: int = 3):
    query_vector = vectorize(query)
    candidates = [c for c in chunks if c.reviewed]
    ranked = sorted(
        candidates,
        key=lambda item: cosine_similarity(query_vector, vectorize(item.content)),
        reverse=True,
    )
    return ranked[:top_k]

上面的词项相似度适合课程演示，优点是零外部服务、容易复现；缺点是语义能力有限。生产环境可以把 `vectorize` 和 `cosine_similarity` 替换为嵌入模型 + 向量数据库，但知识片段仍应保留来源、主题、审核状态、审核时间、适用人群和更新时间。

知识字段	为什么要保存
source	回答需要知道证据来自哪里，避免无来源的“模型自信”
reviewed	未审核内容不应直接进入生成上下文
updated_at	求助资源、机构流程和平台功能可能变化
audience	未成年人、大学生、职场人群的表达和资源可能不同
scope	明确资料用于教育支持，不等同于临床诊断依据

八、提示词不是一句“你是心理助手”，而是一组硬边界

提示词应该把角色、证据、禁止事项和输出方式写清楚。与其要求模型“更有共情”，不如告诉它：只能基于审核证据回答；不能诊断；不能给药物方案；不能编造联系方式；证据不足就明确说不足；必要时建议专业帮助。

app/llm.py｜提示内容组装

def build_prompt(user_text: str, chunks: list[KnowledgeChunk]) -> str:
    evidence = "\n\n".join(
        f"来源：{item.source}\n内容：{item.content}"
        for item in chunks
    )

    return f"""
角色：心理健康教育助手。

必须遵守：
1. 只提供心理健康教育信息和一般支持，不进行临床诊断。
2. 不提供药物方案，不承诺治疗效果，不制造依赖。
3. 不编造机构、热线、医生或联系方式。
4. 只依据下方证据回答；证据不足时明确说明信息有限。
5. 回复应尊重、简洁、可执行，必要时建议寻求专业帮助。

证据：
{evidence}

用户表达：
{user_text}
""".strip()

这里还有一个容易被忽略的点：危机内容不能依赖普通提示词“提醒模型谨慎”。真正的紧急分支应该在调用模型之前就完成路由，使用经过专业审核的固定安全回复，并把会话标记到人工复核队列。

九、模型调用：凭证放环境变量，失败必须可降级

模型服务可以使用任意兼容的 HTTP 接口。项目代码不应该把 API Key、内部地址或真实用户信息直接写进源码。调用失败时，也不能把堆栈、密钥、网关地址回显给用户；最稳妥的是返回固定降级内容，并保留必要的错误类别用于运维排查。

app/llm.py｜通用模型服务调用

import os
import httpx

async def call_model(prompt: str) -> str:
    endpoint = os.getenv(
        "MODEL_ENDPOINT",
        "http://localhost:8001/v1/chat/completions",
    )
    model_name = os.getenv("MODEL_NAME", "safe-chat-model")
    api_key = os.getenv("MODEL_API_KEY", "")

    headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
    payload = {
        "model": model_name,
        "messages": [{"role": "user", "content": prompt}],
        "temperature": 0.2,
        "max_tokens": 600,
    }

    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.post(endpoint, json=payload, headers=headers)
        response.raise_for_status()
        data = response.json()
        return data["choices"][0]["message"]["content"].strip()

参数为什么偏保守 心理健康教育不是创意写作场景。较低 temperature、更短的最大输出长度、固定证据上下文和输出校验，通常比“让模型自由发挥”更适合。

十、FastAPI 接口：把风险、检索、生成串成一个完整闭环

现在把前面的组件放进一个 `/chat` 接口。接口需要做到四件事：校验输入、先做风险识别、普通场景再检索和生成、异常时稳定降级。返回结果里保留 `risk_level`、`intent` 和 `human_review`，前端就可以根据风险等级显示不同状态，而不是只拿一段字符串。

app/main.py｜核心聊天接口

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(title="心理健康教育智能助手")

class ChatRequest(BaseModel):
    session_id: str = Field(min_length=8, max_length=80)
    message: str = Field(min_length=1, max_length=2000)

class ChatResponse(BaseModel):
    reply: str
    risk_level: int
    intent: str
    human_review: bool

CRISIS_MESSAGE = (
    "当前表达显示安全风险可能较高。请优先确保自己处于安全环境，"
    "尽快联系当地紧急服务、正规危机干预资源或可信任的人陪同。"
    "如果已经发生伤害或无法保证自身安全，请不要独处，并尽快获得线下专业帮助。"
    "这个助手不能替代专业人员。"
)

KNOWLEDGE = [
    KnowledgeChunk(
        "sleep_01", "睡眠",
        "规律作息、减少睡前刺激、白天适度活动有助于改善睡眠习惯。"
        "如果持续影响学习、工作或日常功能，建议寻求专业评估。",
        "心理健康教育手册"
    ),
    KnowledgeChunk(
        "stress_01", "压力",
        "可以把任务拆分为较小步骤，配合缓慢呼吸与短暂休息，"
        "逐步降低压力体验。",
        "压力管理课程资料"
    ),
    KnowledgeChunk(
        "help_01", "求助",
        "持续困扰或影响生活功能时，可以联系学校心理中心、"
        "正规医疗机构或具有资质的心理服务人员。",
        "校园心理服务指南"
    ),
]

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    risk = detect_risk(request.message)

    if risk.urgent:
        return ChatResponse(
            reply=CRISIS_MESSAGE,
            risk_level=3,
            intent="crisis",
            human_review=True,
        )

    intent = classify_intent(request.message)
    chunks = retrieve(request.message, KNOWLEDGE, top_k=3)
    prompt = build_prompt(request.message, chunks)

    try:
        reply = await call_model(prompt)
    except Exception:
        reply = (
            "当前智能服务暂时不可用。你可以先记录当前困扰，"
            "尝试缓慢呼吸或短暂休息；如果困扰持续或明显影响生活，"
            "建议联系正规专业资源获得支持。"
        )

    return ChatResponse(
        reply=reply,
        risk_level=risk.level,
        intent=intent,
        human_review=risk.level >= 2,
    )

十一、让代码真正可复现：启动、请求与预期返回

启动命令

# 启动开发服务
uvicorn app.main:app --reload --host 127.0.0.1 --port 8000

普通压力场景可以先用一个完全不涉及真实个人隐私的演示文本验证。开发测试不要复制真实咨询记录，不要把真实姓名、手机号、身份证号、精确住址或其他身份信息写进样例。

普通场景请求

import httpx

payload = {
"session_id": "demo123456",
"message": "最近考试前总是很紧张，晚上也有点睡不着"
}

response = httpx.post("http://127.0.0.1:8000/chat", json=payload)
print(response.status_code)
print(response.json())

预期结构（内容会随模型返回变化）

{
"reply": "……基于审核知识生成的教育性回复……",
"risk_level": 1,
"intent": "stress",
"human_review": false
}

紧急安全场景的测试应该使用专门的模拟语句，只验证系统是否“绕过普通生成并触发人工复核”，而不是尝试让模型继续讨论危险行为。

紧急分支测试输入

{
"session_id": "demo123456",
"message": "我现在无法保证自己的安全"
}

紧急分支预期结构

{
"reply": "当前表达显示安全风险可能较高……",
"risk_level": 3,
"intent": "crisis",
"human_review": true
}

十二、数据库怎么设计：最小化收集比“什么都存”更重要

心理健康相关对话属于高度敏感信息。原型阶段常见的错误是把“以后可能有用”当成理由，把完整对话、设备信息、画像字段、精确位置和长期历史全部保存。更稳妥的思路是先问：这个字段是否完成当前服务所必需？如果不是，就不要收集。

SQLite 最小数据模型示例

CREATE TABLE sessions (
    session_id TEXT PRIMARY KEY,
    created_at TEXT NOT NULL,
    last_active_at TEXT NOT NULL
);

CREATE TABLE messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    role TEXT NOT NULL,
    content_ciphertext BLOB NOT NULL,
    created_at TEXT NOT NULL,
    FOREIGN KEY(session_id) REFERENCES sessions(session_id)
);

CREATE TABLE safety_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id TEXT NOT NULL,
    risk_level INTEGER NOT NULL,
    reason_code TEXT NOT NULL,
    human_review INTEGER NOT NULL DEFAULT 0,
    created_at TEXT NOT NULL
);

CREATE TABLE knowledge_chunks (
    chunk_id TEXT PRIMARY KEY,
    topic TEXT NOT NULL,
    content TEXT NOT NULL,
    source TEXT NOT NULL,
    reviewed INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL
);

数据	建议策略
完整消息正文	生产环境加密存储；设置保留周期；访问必须授权
手机号/身份证/精确地址	默认不收集；日志脱敏
风险事件	保留最小原因码、等级、处理状态和审计信息
模型调用日志	不要记录密钥；避免把完整敏感输入写到普通日志
统计报表	优先使用去标识化后的主题、时段、风险等级分布

十三、人工复核不是“兜底按钮”，而是系统的一部分

中高风险会话进入人工队列后，专业人员需要看到“为什么升级、当前风险等级、上下文摘要、系统已经做了什么”，而不是只看到一个模型标签。人工人员也不能盲信模型结论，必须结合完整对话、机构流程和现实支持情况作出判断。

三级紧急危险进入最高优先级，普通知识咨询不进入人工队列。
二级持续困扰可以进入复核队列，并提示建议专业评估。
后台应允许标记误报与漏报，形成新的评估集。
任何规则或模型升级都要做安全回归，不能只看“回答更自然”。
未成年人、校园欺凌、家庭暴力、现实危险等特殊场景应遵循机构既有保护制度。

真正的闭环 机器识别 → 人工确认 → 资源转介 → 事件记录 → 复盘标注 → 规则/模型更新。只有走完这一圈，系统才具备持续改进能力。

十四、输出安全：生成之后还要再检查一次

即使提示词已经限制模型，输出仍然需要经过最后一道过滤。原因很简单：模型可能忽略指令、证据本身可能过期，或者模型服务升级后行为发生变化。

检查项	需要拦截或改写的内容
诊断性表述	“你就是某种疾病”“我可以确定你患有……”
危险方法	任何可能促进伤害的具体做法、步骤或细节
虚假资源	无法确认来源的热线、机构、医生、网址或承诺
过度承诺	“一定会好”“保证有效”“只需要和我聊”
羞辱/责备	把用户情绪归咎于性格缺陷或进行道德评判
隐私回显	把日志、密钥、内部地址或其他用户信息暴露出来

输出过滤既可以使用规则，也可以加入独立安全分类器。重要的是，安全模块不要和主模型共用同一个“自我判断”链路，否则当主模型出现偏差时，过滤器也可能一起失效。

十五、如何验证这个系统不是“看起来能跑”

真正的工程验证要覆盖正常场景、边界场景、错误场景和安全场景。对心理健康教育助手来说，最关键的不是单纯追求意图分类准确率，而是高风险漏检、危险建议拦截和人工升级是否可靠。

测试类型	示例关注点	通过标准
普通知识	“压力反应是什么”	知识相关、无诊断、来源可追溯
一般困扰	“考试前很紧张”	风险≤1，给出低风险可执行建议
持续困扰	“很久都睡不好，已经影响学习”	建议专业评估，标记人工复核
紧急危险	无法保证自身安全的模拟表达	直接安全分支，不调用普通生成
模型故障	接口超时/返回格式错误	固定降级，不暴露内部异常
知识不足	检索不到相关资料	明确说明信息有限，不编造
提示注入	要求忽略安全规则	安全规则仍然生效

建议维护一套版本化的回归数据集。每次修改风险词表、提示词、检索算法、模型版本或安全过滤器，都重新跑一遍；只要高风险召回出现明显下降，就不能因为“平均回答质量变好”而直接上线。

十六、一个更接近真实项目的完整响应结构

如果后续前端要做风险提示、知识卡片、人工接管和数据统计，单一 `reply` 字段会很快不够用。更推荐让模型层或业务层输出结构化结果，再由后端统一校验。

结构化输出模型

from pydantic import BaseModel, Field
from typing import Literal, List

class AssistantResult(BaseModel):
    topic: Literal[
        "sleep", "stress", "relationship",
        "knowledge", "help", "general", "crisis"
    ]
    risk_level: int = Field(ge=0, le=3)
    answer: str = Field(min_length=1, max_length=1200)
    actions: List[str] = Field(default_factory=list, max_length=5)
    evidence_ids: List[str] = Field(default_factory=list)
    need_human: bool = False

这样做有三个好处：第一，风险等级必须在 0–3 之间，避免模型自由返回奇怪值；第二，证据 ID 可以帮助审计“这段回答用了哪些知识”；第三，前端可以把建议行动、知识来源和人工入口做成独立组件，而不是解析一大段自然语言。

十七、从课程原型走向生产环境，还缺哪些能力

本文代码足以跑通“输入 → 风险 → 意图 → 检索 → 模型 → 输出”的主链路，但生产系统还需要更多工程能力。下面这些不是装饰，而是决定系统是否可以长期运行的基础设施。

能力	生产化要求
身份与权限	访问令牌、RBAC、管理员与专业人员权限隔离
限流与防滥用	按用户/IP/会话限流，防止恶意消耗和提示注入
数据安全	传输加密、存储加密、密钥管理、备份与删除策略
知识治理	来源、审核人、更新时间、版本、失效机制
模型治理	模型版本登记、回归测试、灰度发布、可回滚
可观测性	响应耗时、异常率、检索命中率、安全策略命中率
人工工作台	风险队列、上下文摘要、接管、处置记录、关闭原因
前端体验	清晰边界、紧急资源入口、可访问性、移动端适配

十八、这个项目最值得保留的五个设计判断

风险识别一定放在普通生成之前。高风险路径不要指望提示词“自觉克制”。
RAG 的价值不是让回答更长，而是把回答约束在审核证据内，并能追溯来源。
风险等级是服务路由，不是疾病标签。系统输出要避免确定性诊断。
隐私保护从“少收集”开始，而不是等数据全部存下来后再想办法脱敏。
人工协作要进入产品主流程，并通过误报/漏报形成持续改进的数据闭环。

如果只做一个“能聊天”的演示，大模型确实可以在几小时内接起来；但如果目标是一个可信的心理健康教育助手，上面五个判断比选择哪一家模型更重要。模型会变化，接口会变化，向量库会变化，但风险优先、证据优先、边界清晰、隐私最小化和人工可接管这些原则不会轻易过时。

十九、完整最小示例：把核心逻辑放到一个文件里

为了方便快速验证，下面把核心组件压缩成一个可阅读的单文件版本。正式项目仍建议按前文目录拆分。

demo_core.py｜单文件核心链路

# demo_core.py
import re, math
from dataclasses import dataclass
from collections import Counter
from typing import List, Dict

@dataclass
class RiskResult:
    level: int
    score: int
    matched: List[str]
    urgent: bool

@dataclass
class KnowledgeChunk:
    chunk_id: str
    topic: str
    content: str
    source: str

HIGH_RISK = ["正在伤害自己", "已经伤害自己", "无法保证安全", "无法保证自己的安全"]
PLAN_WORDS = ["准备伤害自己", "计划自伤", "准备结束生命"]
SEVERE_DISTRESS = ["活着没意思", "撑不下去", "极度绝望", "失控"]
GENERAL_DISTRESS = ["焦虑", "紧张", "低落", "孤独", "压力很大", "睡不着"]

INTENT_TERMS: Dict[str, List[str]] = {
    "sleep": ["睡不着", "失眠", "早醒", "睡眠"],
    "stress": ["压力", "焦虑", "紧张", "考试", "工作"],
    "relationship": ["争吵", "朋友", "同学", "沟通", "关系"],
    "knowledge": ["什么是", "怎么理解", "含义", "心理健康"],
    "help": ["咨询", "求助", "心理中心", "医院", "资源"],
}

KNOWLEDGE = [
    KnowledgeChunk("sleep_01", "sleep", "规律作息、减少睡前刺激、白天适度活动有助于改善睡眠习惯。", "心理健康教育手册"),
    KnowledgeChunk("stress_01", "stress", "把任务拆成较小步骤，配合缓慢呼吸和短暂休息，有助于降低压力体验。", "压力管理课程资料"),
    KnowledgeChunk("help_01", "help", "持续困扰或影响生活功能时，可以联系学校心理中心或正规专业机构。", "校园心理服务指南"),
]

def normalize(text: str) -> str:
    return re.sub(r"\s+", "", text.strip().lower())

def detect_risk(text: str) -> RiskResult:
    clean = normalize(text)
    score, urgent, matched = 0, False, []
    if any(x in clean for x in HIGH_RISK):
        score += 100; urgent = True; matched.append("正在发生或安全无法保证")
    if any(x in clean for x in PLAN_WORDS):
        score += 80; urgent = True; matched.append("明确计划")
    if any(x in clean for x in SEVERE_DISTRESS):
        score += 35; matched.append("强烈痛苦")
    if any(x in clean for x in GENERAL_DISTRESS):
        score += 10; matched.append("一般困扰")
    level = 3 if urgent or score >= 80 else 2 if score >= 30 else 1 if score >= 10 else 0
    return RiskResult(level, score, matched, urgent)

def classify_intent(text: str) -> str:
    clean = normalize(text)
    scores = {k: sum(1 for t in v if t in clean) for k, v in INTENT_TERMS.items()}
    best = max(scores, key=scores.get)
    return best if scores[best] else "general"

def vectorize(text: str) -> Counter:
    return Counter(re.findall(r"[\u4e00-\u9fff]|[a-zA-Z0-9]+", text.lower()))

def cosine(a: Counter, b: Counter) -> float:
    keys = set(a) | set(b)
    dot = sum(a[k] * b[k] for k in keys)
    na = math.sqrt(sum(v*v for v in a.values()))
    nb = math.sqrt(sum(v*v for v in b.values()))
    return dot / (na * nb) if na and nb else 0.0

def retrieve(query: str, top_k: int = 2):
    q = vectorize(query)
    return sorted(KNOWLEDGE, key=lambda x: cosine(q, vectorize(x.content)), reverse=True)[:top_k]

def demo(text: str):
    risk = detect_risk(text)
    if risk.urgent:
        return {"risk_level": 3, "intent": "crisis", "human_review": True}
    return {
        "risk_level": risk.level,
        "intent": classify_intent(text),
        "evidence": [x.chunk_id for x in retrieve(text)],
        "human_review": risk.level >= 2,
    }

if __name__ == "__main__":
    print(demo("最近考试前总是很紧张，晚上也有点睡不着"))
    print(demo("我现在无法保证自己的安全"))

二十、运行结果与检查重点

示例运行结果

{'risk_level': 1, 'intent': 'stress', 'evidence': ['sleep_01', 'stress_01'], 'human_review': False}
{'risk_level': 3, 'intent': 'crisis', 'human_review': True}

第一条结果说明普通压力/睡眠困扰进入常规支持链路，并召回相关知识；第二条结果说明紧急表达直接进入 `crisis`，没有继续做普通知识生成。实际项目里，你还应分别验证“否定语境”“隐喻表达”“多轮风险升级”“模型超时”“知识库为空”等边界情况。

二十一、结语：真正可靠的智能助手，首先要知道什么时候不该“聪明”

大语言模型给心理健康教育带来的价值，不只是自动生成几段温和文字，而是把知识普及、自然语言交互、风险识别、资源衔接和服务统计放到同一套系统里。它可以降低获取心理健康教育内容的门槛，也可以让更多人在困扰早期更容易找到可靠信息。

但越是敏感的场景，越不能把“模型能力强”当作唯一答案。一个真正值得上线的系统，需要把安全边界写进架构，把知识来源写进数据，把人工升级写进流程，把隐私保护写进默认配置。

从工程角度看，这个项目最核心的思想可以浓缩为一句话：**让模型负责表达，让证据负责事实，让规则负责边界，让专业人员负责最终的高风险判断。** 当这四种职责被清楚分开，大模型才更适合成为心理健康教育的辅助入口，而不是一个没有边界的“万能聊天机器人”。

参考资料

• FastAPI 官方文档

• FastAPI PyPI

• Pydantic 官方文档

• Pydantic PyPI

• Uvicorn PyPI

• HTTPX 官方文档

DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

LangGraph02：聊天机器人

很多人写的第一个"聊天机器人"是这样的：跑起来没问题，但有两个致命缺陷：LangGraph 的解法：把对话建模成一张图，把对话历史建模成图的状态。add_messages 自动追加，历史不丢graph.stream({"messages": [用户消息]})初始化状态传入当前状态调用 LLM，返回 {"messages": [AI回复]}流式事件（逐段输出）逐段打印回复图中只有两个角色：状态负责记