必备知识：Python基础、HTTP基础和FastAPI（了解）

实例仅支持本机：127.0.0.1 是“回环地址”，改成0.0.0.0即可实现局域网其他设备访问

环境安装：

httpx>=0.27      # 发 HTTP 请求调用大模型 API
python-dotenv>=1.0  # 读取 .env 文件里的配置
fastapi>=0.115,<1
uvicorn>=0.30,<1

APIkey准备：直接使用deepseek即可，写一个.env文件，apikey不进代码，

本地部署建议显卡配置大于等于5070ti，就不需要apikey

#  把下面等号右边换成你自己的 key
LLM_API_KEY=

# ---- 服务地址：告诉程序去哪里调用大模型 ----

LLM_BASE_URL=https://api.deepseek.com
LLM_MODEL=deepseek-flash

# ---- 可选参数（可以先不动）----
# 温度：0~1，越低回答越稳定，越高越有创造性（和上个项目的 temperature 同一个概念）
LLM_TEMPERATURE=0.7

# 最多保留多少"轮"对话历史（一问一答算一轮），防止历史太长 token 爆炸
MAX_HISTORY_ROUNDS=10

技术栈：

技术	负责什么
Python	实现后端业务、模型调用、数据库操作
FastAPI	提供聊天和会话接口，校验输入，返回结果或流式事件
Uvicorn	运行 FastAPI，监听本机端口，收发 HTTP 请求
Pydantic	定义请求和响应结构，校验消息等字段
HTTPX	后端向模型服务发送普通请求和异步流式请求
SQLite／sqlite3	保存会话、完整消息和 Token 用量
python-dotenv	从 .env 加载模型地址、密钥和参数
logging	记录运行事件、调用耗时、用量和错误，支持日志轮转
HTML＋CSS＋JavaScript	构建聊天界面、处理按钮和输入、显示回答
Fetch＋ReadableStream＋SSE	在网页持续接收并展示流式文字

项目结构

llm_learning/
├── run_web.py          启动入口
├── backend/            后端
├── static/             网页前端（暂不提供）
├── data/               会话数据库
├── logs/               运行日志
├── .env                本地配置和密钥
├── requirements.txt    依赖清单
└── README.md           使用及流程说明

当前效果

用户在网页发送消息
    ↓
app.js 发出 HTTP 请求
    ↓
Uvicorn 接收 → FastAPI 路由 → Pydantic 校验
    ↓
ChatService 读取 SQLite 历史，构建上下文
    ↓
HTTPX 向模型服务发起流式请求
    ↓
模型文字片段 → 后端 SSE → stream.js → 网页逐段显示
    ↓
正常结束后，SQLite 保存完整问答和用量
    ↓
网页更新会话列表和统计

run_web.py  

import uvicorn

if __name__ == '__main__':
    #导入 backend/api.py，找到其中名为 app 的 FastAPI 对象,启动服务器
    uvicorn.run('backend.api:app', host='127.0.0.1', port=8000, workers=1)

backend/api.py

"""本机 FastAPI 服务。启动：python run_web.py。"""
import json
import logging
import mimetypes
from contextlib import asynccontextmanager, aclosing
from pathlib import Path
from urllib.parse import urlsplit

from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from .schemas import History, MessageInput, Reply, Session
from starlette.middleware.trustedhost import TrustedHostMiddleware

from .chat_service import ChatError, ChatService
from .config import MODEL, check_config, setup_logging
from .storage import SessionStore, StorageError

logger = logging.getLogger(__name__)
STATIC = Path(__file__).resolve().parent.parent / 'static'
# Windows 的注册表可能把 .js 识别为 text/plain；显式声明脚本类型。
mimetypes.init()
mimetypes.add_type('text/javascript', '.js')
mimetypes.add_type('text/css', '.css')


def create_app(store=None, llm=None, stream_llm=None):
    @asynccontextmanager
    async def lifespan(app):
        closer = None
        streaming_call = stream_llm
        if llm is None:
            check_config()
            setup_logging()
            from . import llm_client
            if llm_client.client.is_closed:
                import httpx
                llm_client.client = httpx.Client(timeout=60, trust_env=False)
            model_call = llm_client.call_llm
            streaming_call = stream_llm or llm_client.stream_llm
            closer = llm_client.client.close
        else:
            model_call = llm
        app.state.chat = ChatService(store if store is not None else SessionStore(), model_call, streaming_call)
        try:
            yield
        finally:
            if closer:
                closer()

    app = FastAPI(title='Chatbot 本机接口', version='1.0', lifespan=lifespan)
    app.add_middleware(TrustedHostMiddleware, allowed_hosts=['127.0.0.1', 'localhost', 'testserver'])

    @app.middleware('http')
    async def local_browser_access(request: Request, call_next):
        # 同源页面无需 CORS。拒绝其他网站从浏览器调用本机聊天接口。
        origin = request.headers.get('origin')
        if origin:
            actual = urlsplit(str(request.url))
            supplied = urlsplit(origin)
            if (supplied.scheme, supplied.netloc) != (actual.scheme, actual.netloc):
                return JSONResponse({'error': {'kind': 'origin', 'message': '仅允许本机同源页面访问。'}}, status_code=403)
        response = await call_next(request)
        response.headers['Cache-Control'] = 'no-store'
        response.headers['X-Content-Type-Options'] = 'nosniff'
        response.headers['Content-Security-Policy'] = "default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self'; img-src 'self' data:; frame-ancestors 'none'" if request.url.path != '/docs' else "frame-ancestors 'none'"
        return response

    @app.exception_handler(ChatError)
    async def chat_error(request, error):
        return JSONResponse({'error': {'kind': error.kind, 'message': str(error)}}, status_code=error.status)

    @app.exception_handler(StorageError)
    async def storage_error(request, error):
        logger.error('数据库操作失败', exc_info=error)
        return JSONResponse({'error': {'kind': 'storage', 'message': '数据库暂时无法访问，请稍后重试。'}}, status_code=503)

    @app.exception_handler(RequestValidationError)
    async def validation_error(request, error):
        return JSONResponse({'error': {'kind': 'validation', 'message': '请输入 1～20000 个字符的消息。'}}, status_code=422)

    @app.exception_handler(Exception)
    async def unexpected_error(request, error):
        logger.error('请求失败', exc_info=error)
        kind = getattr(error, 'kind', None)
        if kind in {'timeout', 'network', 'auth', 'ratelimit', 'http', 'parse', 'empty'}:
            messages = {'timeout': '模型响应超时，请稍后重试。', 'auth': '模型服务鉴权失败，请检查后端密钥配置。',
                        'ratelimit': '模型服务限流或额度不足，请稍后再试或检查账户。'}
            status = 504 if kind == 'timeout' else 502
            return JSONResponse({'error': {'kind': kind, 'message': messages.get(kind, '模型服务暂时无法返回有效回答，本轮未保存。')}}, status_code=status)
        return JSONResponse({'error': {'kind': 'internal', 'message': '处理失败，请检查运行日志。本轮可能尚未保存，请刷新历史确认。'}}, status_code=500)

    @app.get('/', include_in_schema=False)
    def index():
        return FileResponse(STATIC / 'index.html')

    @app.get('/health')
    def health():
        return {'status': 'ok', 'model': MODEL, 'streaming': True}

    @app.get('/sessions', response_model=list[Session])
    def sessions(request: Request):
        return request.app.state.chat.store.list_sessions()

    @app.post('/sessions', response_model=Session, status_code=201)
    def create_session(request: Request):
        return request.app.state.chat.create()

    @app.get('/sessions/{session_id}/messages', response_model=History)
    def history(session_id: str, request: Request):
        session, messages = request.app.state.chat.load(session_id)
        return {'session': session, 'messages': messages}

    @app.post('/sessions/{session_id}/messages', response_model=Reply)
    def send(session_id: str, body: MessageInput, request: Request):
        return request.app.state.chat.send(session_id, body.content)

    @app.post('/sessions/{session_id}/messages/stream')
    def stream(session_id: str, body: MessageInput, request: Request):
        service = request.app.state.chat
        service.load(session_id)  # 流开始前返回正常的 404/503。

        async def events():
            async with aclosing(service.stream(session_id, body.content)) as source:
                async for event in source:
                    yield f"event: {event['type']}\ndata: {json.dumps(event, ensure_ascii=False)}\n\n"

        return StreamingResponse(events(), media_type='text/event-stream',
                                 headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'})

    app.mount('/static' , StaticFiles(directory=STATIC), name='static')
    return app


app = create_app()

backend/chat_service.py

"""网页聊天业务：读取历史、限制上下文、调用模型并保存完整问答。"""
from threading import Lock
from contextlib import aclosing
import logging

from .config import MAX_HISTORY_ROUNDS, SYSTEM_PROMPT
from .storage import StorageError


class ChatError(Exception):
    def __init__(self, message, status=500, kind="internal"):
        super().__init__(message)
        self.status, self.kind = status, kind


class ChatService:
    def __init__(self, store, llm, stream_llm=None):
        self.store, self.llm = store, llm
        self.stream_llm = stream_llm
        self._locks, self._guard = {}, Lock()

    def load(self, session_id):
        # 使用现有存储接口；区分不存在与数据库读取失败。
        if not any(s['id'] == session_id for s in self.store.list_sessions()):
            raise ChatError("找不到这段对话，请选择其他会话。", 404, "not_found")
        return self.store.load_session(session_id)

    def create(self):
        return self.load(self.store.create_session(SYSTEM_PROMPT))[0]

    def send(self, session_id, content):
        with self._guard:
            lock = self._locks.setdefault(session_id, Lock())
        if not lock.acquire(blocking=False):
            raise ChatError("这段对话正在生成回答，请等待完成。", 409, "busy")
        try:
            session, history = self.load(session_id)
            count = max(1, MAX_HISTORY_ROUNDS) * 2
            context = history[:1] + history[1:][-count:]
            context.append({'role': 'user', 'content': content})
            reply, usage = self.llm(context)
            if not isinstance(reply, str) or not reply.strip() or not isinstance(usage, dict):
                raise ChatError("模型响应格式异常，本轮未保存。", 502, "parse")
            try:
                self.store.save_turn(session_id, content, reply, usage, session['rounds'])
            except StorageError as e:
                raise ChatError("回答已生成，但保存失败。本轮未加入历史；模型可能已产生用量。请刷新会话后再试。", 409, "save") from e
            return {'session': self.load(session_id)[0], 'reply': reply, 'usage': usage}
        finally:
            lock.release()


    async def stream(self, session_id, content):
        """部分内容只送往页面，正常结束后才提交一整轮。取消时释放锁和上游。"""
        with self._guard:
            lock = self._locks.setdefault(session_id, Lock())
        if not lock.acquire(blocking=False):
            yield {'type': 'error', 'kind': 'busy', 'message': '这段对话正在生成回答，请稍候。', 'saved': False}
            return
        saved = False
        try:
            if self.stream_llm is None:
                raise ChatError('当前未配置流式模型', 503, 'configuration')
            session, history = self.load(session_id)
            context = history[:1] + history[1:][-max(1, MAX_HISTORY_ROUNDS)*2:]
            context.append({'role': 'user', 'content': content})
            parts, usage, completed = [], {}, False
            async with aclosing(self.stream_llm(context)) as upstream:
                async for event in upstream:
                    if event['type'] == 'delta':
                        parts.append(event['text'])
                        yield event
                    elif event['type'] == 'complete':
                        usage = event.get('usage', {})
                        completed = True
            reply = ''.join(parts)
            if not completed:
                raise ChatError('回答中断，本轮未保存。', 502, 'incomplete')
            if not reply.strip():
                raise ChatError('模型返回空回复，本轮未保存。', 502, 'empty')
            # 本地提交期间没有 await，避免在事务提交与 saved 标记之间被取消。
            self.store.save_turn(session_id, content, reply, usage, session['rounds'])
            saved = True
            result = dict(session)
            result.update(rounds=session['rounds']+1,
                          total_tokens=session['total_tokens']+usage.get('total_tokens', 0))
            if session['rounds'] == 0:
                result['title'] = content[:40]
            yield {'type': 'done', 'session': result, 'usage': usage,
                   'usage_available': 'total_tokens' in usage, 'saved': True}
        except Exception as e:
            logging.getLogger(__name__).exception('流式对话失败 session_id=%s', session_id)
            kind = getattr(e, 'kind', 'storage' if isinstance(e, StorageError) else 'internal')
            message = {'timeout':'模型响应超时', 'auth':'模型鉴权失败，请检查后端密钥',
                       'ratelimit':'模型限流或额度不足', 'incomplete':'回答中断或被截断',
                       'storage':'保存失败，请刷新历史', 'empty':'模型返回空回复',
                       'network':'模型网络连接中断'}.get(kind, '流式生成失败，请检查日志')
            yield {'type':'error', 'kind':kind, 'message':message+'；本轮未保存，模型可能已产生用量。', 'saved':saved}
        finally:
            lock.release()

backend/config.py

"""
config.py —— 配置中心：把 .env 文件里的配置装进 Python 变量

执行流程：
    .env 文件 --(load_dotenv 读取)--> 进程环境变量 --(os.getenv 取出)--> 常量

两个知识点：
  1. 环境变量的优先级：os.getenv 先看"真正的环境变量"，没有才用 .env 里的值。
     所以 `export LLM_MODEL`（临时设置环境变量）能覆盖 .env 的配置
     ——测试/切换时不用改文件，很常用。
  2. 配置和代码分离：本文件是唯一"知道配置存在"的地方，
     后端模块只管从这里 import，以后加配置也只改这里。
"""

import logging
import logging.handlers
import os
from pathlib import Path

# python-dotenv：把 .env 文件里的 KEY=VALUE 加载进环境变量
# （这个库已是 fastapi 的老朋友，之前安装依赖时自动装好了）
from dotenv import load_dotenv

# 无论从哪个工作目录启动，都读取项目根目录下的 .env。
PROJECT_ROOT = Path(__file__).resolve().parent.parent
load_dotenv(PROJECT_ROOT / ".env")

# ------------------------------------------------------------------
# 必填项
# ------------------------------------------------------------------
# API Key：调用大模型的"身份证"，服务器靠它识别你是谁、计费归谁
API_KEY: str = os.getenv("LLM_API_KEY", "")

# 服务地址：所有请求都发到这个"根地址"下
# 真正的接口地址 = BASE_URL + "/chat/completions"（在 llm_client.py 里拼接）
BASE_URL: str = os.getenv("LLM_BASE_URL", "https://open.bigmodel.cn/api/paas/v4")

# 模型名：同一家服务商通常有多个模型（快而便宜的 / 强而贵的）
MODEL: str = os.getenv("LLM_MODEL", "glm-4-flash")

# ------------------------------------------------------------------
# 可调参数
# ------------------------------------------------------------------
TEMPERATURE: float = float(os.getenv("LLM_TEMPERATURE", "0.7"))

MAX_HISTORY_ROUNDS: int = int(os.getenv("MAX_HISTORY_ROUNDS", "20"))

# ------------------------------------------------------------------
# System Prompt —— 给 AI 的"人设/规则说明书"
#
# 三种角色一条消息的区别（本项目核心知识点）：
#   system    全局设定："你是谁、怎么说话"（用户一般看不到，但影响所有回答）
#   user      用户说的话
#   assistant AI 的回答
# System Prompt 放在 messages 的第一条，全程不变。
# 想换个性格？改下面的字符串就行，试试看效果差别。
# ------------------------------------------------------------------
SYSTEM_PROMPT: str = (
    "你是一位耐心的中文技术助教，擅长用通俗的比喻解释编程概念。"
    "回答遵循：先给结论，再给解释，必要时给一个小例子。"
    "不确定的事情明确说不确定，不要编造。"
)


def check_config() -> None:
    """启动时做一次配置体检，把"配置错误"拦在运行之前，报错信息友好一点。"""
    problems = []
    if not API_KEY or "请把这里" in API_KEY:
        problems.append("还没有配置 API Key！请打开 llm_learning/.env，"
                        "把 LLM_API_KEY 换成你自己的 key"
                        "（智谱免费注册：https://open.bigmodel.cn/）")
    if not BASE_URL.startswith("http"):
        problems.append(f"LLM_BASE_URL 格式不对：{BASE_URL}")
    if problems:
        print("× 配置检查未通过：")
        for p in problems:
            print(f"  - {p}")
        raise SystemExit(1)   # 直接终止程序，退出码 1 表示"有错误"


# ==================================================================
# logging 配置 —— 和 agent_backend/core/logging_config.py 同一套思路
#
# 网页负责展示对话；logging 负责记录请求耗时、Token 和错误等运行事件。
# logging 的解法：一次 logger.info() 调用，多个 handler 各取所需——
#
#   控制台 handler → 只显示 WARNING 以上（平时安静，出错必现形）
#   文件 handler   → 记录 INFO 以上（logs/chat.log 留下完整审计轨迹）
#
# 这就是 logging 比 print 强大的本质：print 是"喊一嗓子谁都听见"，
# logging 是"按级别分流到不同目的地"。
# ==================================================================
LOG_DIR = str(PROJECT_ROOT / "logs")
LOG_FORMAT = "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s"


def setup_logging(level: int = logging.INFO) -> None:
    """初始化日志系统：整个程序启动时调用一次。"""
    os.makedirs(LOG_DIR, exist_ok=True)

    root = logging.getLogger()
    if root.handlers:                      # 清掉旧 handler，防止热重载时日志重复
        for h in root.handlers:
            root.removeHandler(h)
    root.setLevel(level)

    formatter = logging.Formatter(LOG_FORMAT, datefmt="%Y-%m-%d %H:%M:%S")

    # handler ①：控制台——只放行 WARNING 以上，聊天界面不被刷屏
    console = logging.StreamHandler()      # 默认输出到 stderr（和 print 的 stdout 分开）
    console.setLevel(logging.WARNING)
    console.setFormatter(formatter)

    # handler ②：滚动文件——INFO 全记录，1MB 轮转，最多留 3 份历史
    file_handler = logging.handlers.RotatingFileHandler(
        os.path.join(LOG_DIR, "chat.log"),
        maxBytes=1_000_000,
        backupCount=3,
        encoding="utf-8",
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(formatter)

    root.addHandler(console)
    root.addHandler(file_handler)

# 默认请求流式用量；不兼容此参数的服务可在 .env 中关闭。
STREAM_INCLUDE_USAGE = os.getenv("LLM_STREAM_INCLUDE_USAGE", "true").lower() in {"true", "1", "yes"}

backend/llm_client.py

"""
llm_client.py —— 模型调用模块：负责调用模型、处理接口错误、解析响应

================================================================
设计思想：为什么要拆出这个文件（"分层/解耦"）
================================================================
api.py / chat_service.py 只关心"聊天"，不应该关心 HTTP 细节（状态码、JSON 结构、网络异常）。
本模块负责把"一次大模型调用"的【所有失败方式】统统翻译成同一种错误
LLMError 再抛出去——这就是"统一错误类型"：

    api.py / chat_service.py 视角的世界只有两种结果：
        call_llm(messages)  →  成功：拿到 (回复, usage)
                            →  失败：抛 LLMError（看 .kind 知道是哪类错）
    它完全不需要知道 httpx 的存在。

================================================================
一次调用可能失败的全部方式（本模块逐一处理，缺一不可）
================================================================
  ┌─ 发送阶段（网络/HTTP 层）────────────────────────────────┐
  │ 连不上服务器                        → kind="network"     │
  │ 请求超时（60秒无响应）               → kind="timeout"     │
  │ 其他传输异常（连接中途断开等）        → kind="network"     │
  │ 401 key 无效                        → kind="auth"        │
  │ 429 限流/额度不足                   → kind="ratelimit"   │
  │ 其他 4xx/5xx                        → kind="http"        │
  ├─ 解析阶段（数据格式层）──────────────────────────────────┤
  │ 响应不是 JSON（如网关返回 HTML 错误页）→ kind="parse"     │
  │ choices 为空列表                    → kind="parse"       │
  │ message/content 字段缺失             → kind="parse"       │
  │ 回复内容为空（None 或空字符串）       → kind="empty"       │
  └─────────────────────────────────────────────────────────┘
"""

import logging
import time

import httpx

from .config import API_KEY, BASE_URL, MODEL, TEMPERATURE

logger = logging.getLogger(__name__)

# ------------------------------------------------------------------
# 统一错误类型
#
# 为什么自定义异常类，而不直接用 RuntimeError？
#   1. 精确捕获：api.py / chat_service.py 里 except LLMError 只接住"模型调用错误"，
#      不会误伤程序其他部分的异常
#   2. 自带分类：.kind 属性让调用方能按类别应对
#      （比如 ratelimit 时自动等几秒重试——后续可按 kind 实现重试）
# ------------------------------------------------------------------


class LLMError(Exception):
    """所有 LLM 调用相关的错误。kind: network/timeout/auth/ratelimit/http/parse/empty"""

    def __init__(self, message: str, kind: str = "unknown"):
        super().__init__(message)
        self.kind = kind


# ------------------------------------------------------------------
# 全局 HTTP 客户端（从 api.py / chat_service.py 迁移过来）
#   1. 连接复用：多轮对话反复请求同一服务器，复用 TCP 连接更快
#   2. trust_env=False：不读系统代理，避免代理软件拦截导致 502
# ------------------------------------------------------------------
client = httpx.Client(timeout=60, trust_env=False)


# ==================================================================
# 公共入口：api.py / chat_service.py 只需要认识这一个函数
# ==================================================================
def call_llm(messages: list) -> tuple:
    """
    调用一次大模型（OpenAI 兼容接口）。

    成功：返回 (回复文本, usage字典)
    失败：抛出 LLMError（绝不让 httpx/json 的原生异常漏出去）
    """
    start = time.perf_counter()

    data = _post_messages(messages)       # 阶段①：拿到合法的响应 JSON（dict）
    reply, usage = _extract_reply(data)   # 阶段②：从中安全取出回复

    # 成功才记 INFO：消息数逐轮+2、耗时、token——都能在日志里观察
    logger.info(
        "LLM调用成功 model=%s 发送消息数=%d 耗时=%.2fs prompt_tokens=%s completion_tokens=%s",
        MODEL, len(messages), time.perf_counter() - start,
        usage.get("prompt_tokens", "?"), usage.get("completion_tokens", "?"),
    )
    return reply, usage


# ==================================================================
# 阶段①：发送请求，处理 网络层/HTTP状态码 层 的错误
# ==================================================================
def _post_messages(messages: list) -> dict:
    """发出 POST 并保证返回一个合法 JSON 字典，否则抛 LLMError。"""

    url = f"{BASE_URL}/chat/completions"
    headers = {
        # Authorization：HTTP 标准鉴权头，服务器靠它识别你的账号
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": MODEL,
        "messages": messages,
        "temperature": TEMPERATURE,
    }

    # ---- 网络层错误：请求根本没成功送达/没得到完整响应 ----
    try:
        resp = client.post(url, json=payload, headers=headers)
    except httpx.TimeoutException as e:
        # TimeoutException 是各种超时（连接/读/写/池）的基类
        logger.error("请求超时：%s", url)
        raise LLMError("请求超时（60秒无响应），可稍后重试", kind="timeout")
    except httpx.TransportError as e:
        # TransportError 是所有传输层异常的基类：连接被拒、DNS 解析失败、
        # 对方中途断开等——最后一道网络层防线，兜住漏网之鱼
        logger.error("网络异常 %s: %s", type(e).__name__, e)
        raise LLMError(f"网络异常（{type(e).__name__}），请检查网络或 BASE_URL", kind="network")

    # ---- HTTP 状态码层错误：请求送到了，但服务器说"不行" ----
    if resp.status_code == 401:
        logger.error("API Key 无效（401），鉴权被服务器拒绝")
        raise LLMError("API Key 无效（401），请检查 .env 里的 LLM_API_KEY", kind="auth")
    if resp.status_code == 429:
        logger.error("被限流（429）：调用过于频繁或额度不足")
        raise LLMError("请求被限流（429），请稍等几秒再试", kind="ratelimit")
    if resp.status_code != 200:
        logger.error("请求失败 HTTP %d: %s", resp.status_code, resp.text[:200])
        raise LLMError(f"请求失败 HTTP {resp.status_code}", kind="http")

    # ---- 响应体是 JSON 吗？----
    # 现实中很常见：网关/代理出故障时返回 HTML 错误页（<html>502...</html>），
    # 此时 resp.json() 会抛 JSONDecodeError，不接住就是程序崩溃
    try:
        return resp.json()
    except ValueError:   # json.JSONDecodeError 是 ValueError 的子类
        snippet = resp.text[:100].replace("\n", " ")
        logger.error("响应不是合法 JSON：%r", snippet)
        raise LLMError(f"服务器返回了非 JSON 内容（可能是网关故障）：{snippet}...", kind="parse")


# ==================================================================
# 阶段②：解析响应体，处理 结构缺失/空回复 层 的错误
#
# 目标 JSON 结构（每一层都可能缺）：
#   {"choices": [ {"message": {"role":..., "content": "回复"}} ],
#    "usage": {...}}
#   choices 缺失/空列表 → choices[0] 缺 message → message.content 缺失 → content 为空
# ==================================================================
def _extract_reply(data: dict) -> tuple:
    """从响应字典里安全取出回复文本和 usage，任何结构问题都抛 LLMError。"""

    # 防1：choices 字段缺失，或存在但是空列表 []
    # （部分服务在异常时返回 200 + 空 choices，状态码看不出问题！）
    choices = data.get("choices")
    if not choices:
        logger.error("响应中 choices 为空：%s", str(data)[:200])
        raise LLMError("服务器返回的 choices 为空，没有生成任何内容", kind="parse")

    # 防2：message 或 content 字段缺失（结构不完整）
    message = choices[0].get("message") or {}
    content = message.get("content")
    if content is None:
        # content 键不存在（结构缺失）和值为 null（部分推理模型只返回
        # reasoning_content 而不带最终答案）都走这里
        logger.error("响应缺少 message.content：%s", str(data)[:200])
        raise LLMError("响应结构异常：缺少 message.content 字段", kind="parse")

    # 防3：空回复——content 是 "" 或全空白。
    # 严格说"请求成功了"，但拿到的回复没法用，单独归类为 empty
    if not content.strip():
        logger.error("模型返回了空回复")
        raise LLMError("模型返回了空回复，请换个问法试试", kind="empty")

    usage = data.get("usage") or {}   # usage 缺失不致命，给空字典继续跑
    return content, usage

# 流式入口：异步读取 SSE，HTTPX 负责跨网络块的 UTF-8 与行解码。
async def stream_llm(messages: list, http_client=None):
    """依次产生 delta 和 complete；只有明确正常结束才产生 complete。"""
    import json
    from contextlib import asynccontextmanager
    from .config import STREAM_INCLUDE_USAGE

    @asynccontextmanager
    async def connection():
        if http_client is not None:
            yield http_client
        else:
            async with httpx.AsyncClient(timeout=httpx.Timeout(60, connect=10), trust_env=False) as session:
                yield session

    async def events(response):
        lines = []
        size = 0
        async for line in response.aiter_lines():
            if not line:
                if lines:
                    yield '\n'.join(lines)
                    lines, size = [], 0
            elif line.startswith('data:'):
                part = line[5:].removeprefix(' ')
                size += len(part)
                if size > 1_000_000:
                    raise LLMError('流式事件过大', kind='parse')
                lines.append(part)
        if lines:
            yield '\n'.join(lines)

    payload = {'model': MODEL, 'messages': messages, 'temperature': TEMPERATURE, 'stream': True}
    if STREAM_INCLUDE_USAGE:
        payload['stream_options'] = {'include_usage': True}
    usage, finish = {}, None
    start = time.perf_counter()
    try:
        async with connection() as session:
            async with session.stream('POST', f"{BASE_URL.rstrip('/')}/chat/completions",
                                      headers={'Authorization': f'Bearer {API_KEY}'}, json=payload) as response:
                if response.status_code != 200:
                    kind = {401:'auth', 429:'ratelimit'}.get(response.status_code, 'http')
                    raise LLMError(f'模型服务返回 HTTP {response.status_code}', kind=kind)
                if 'text/event-stream' not in response.headers.get('content-type', '').lower():
                    raise LLMError('模型服务没有返回 SSE 流式响应', kind='parse')
                async for event in events(response):
                    if event.strip() == '[DONE]':
                        break
                    try:
                        chunk = json.loads(event)
                    except ValueError as e:
                        raise LLMError('流式事件不是合法 JSON', kind='parse') from e
                    if not isinstance(chunk, dict) or chunk.get('error'):
                        raise LLMError('模型返回异常流式事件', kind='parse')
                    incoming_usage = chunk.get('usage')
                    if incoming_usage is not None:
                        if not isinstance(incoming_usage, dict):
                            raise LLMError('流式用量格式错误', kind='parse')
                        for key in ('total_tokens', 'prompt_tokens', 'completion_tokens'):
                            if key in incoming_usage and (type(incoming_usage[key]) is not int or incoming_usage[key] < 0):
                                raise LLMError('流式用量数值错误', kind='parse')
                        usage = incoming_usage
                    choices = chunk.get('choices', [])
                    if not isinstance(choices, list):
                        raise LLMError('流式 choices 格式错误', kind='parse')
                    if not choices:  # 独立 usage 事件允许空 choices。
                        continue
                    choice = choices[0]
                    if not isinstance(choice, dict):
                        raise LLMError('流式 choice 格式错误', kind='parse')
                    delta = choice.get('delta') or {}
                    if not isinstance(delta, dict) or delta.get('tool_calls'):
                        raise LLMError('当前聊天不支持此流式内容', kind='parse')
                    content = delta.get('content')
                    if content is not None:
                        if not isinstance(content, str) or (finish is not None and content):
                            raise LLMError('流式文本格式或顺序错误', kind='parse')
                        if content:
                            yield {'type': 'delta', 'text': content}
                    reason = choice.get('finish_reason')
                    if reason is not None:
                        finish = reason
                        if reason != 'stop':
                            raise LLMError('回答被截断或未正常结束，本轮未保存', kind='incomplete')
        if finish != 'stop':
            raise LLMError('流式连接提前结束，本轮未保存', kind='incomplete')
        logger.info('流式调用完成 model=%s 耗时=%.2fs total_tokens=%s', MODEL,
                    time.perf_counter()-start, usage.get('total_tokens', '?'))
        yield {'type': 'complete', 'usage': usage}
    except httpx.TimeoutException as e:
        raise LLMError('流式请求超时，本轮未保存', kind='timeout') from e
    except httpx.TransportError as e:
        raise LLMError('流式网络中断，本轮未保存', kind='network') from e

backend/schemas.py

"""HTTP 请求及响应数据结构。"""
from pydantic import BaseModel, ConfigDict, Field, field_validator


class MessageInput(BaseModel):
    model_config = ConfigDict(extra='forbid')
    content: str = Field(min_length=1, max_length=20000)

    @field_validator('content')
    @classmethod
    def nonempty(cls, value):
        value = value.strip()
        if not value:
            raise ValueError('请输入消息')
        return value


class Message(BaseModel):
    role: str
    content: str


class Session(BaseModel):
    id: str
    title: str
    created_at: str
    updated_at: str
    rounds: int
    total_tokens: int


class History(BaseModel):
    session: Session
    messages: list[Message]


class Reply(BaseModel):
    session: Session
    reply: str
    usage: dict

backend/storage.py

"""SQLite 会话存储。完整历史与模型请求的裁剪窗口相互独立。"""
import json
import sqlite3
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from uuid import uuid4

DB_PATH = Path(__file__).resolve().parent.parent / "data" / "chatbot.db"


class StorageError(Exception):
    """数据库操作失败，调用方应保留当前会话状态。"""


class SessionStore:
    def __init__(self, path=DB_PATH):
        self.path = Path(path)
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            with closing(self._connect()) as db, db:
                db.executescript("""
                    CREATE TABLE IF NOT EXISTS sessions (
                        id TEXT PRIMARY KEY,
                        title TEXT NOT NULL,
                        created_at TEXT NOT NULL,
                        updated_at TEXT NOT NULL,
                        rounds INTEGER NOT NULL DEFAULT 0,
                        total_tokens INTEGER NOT NULL DEFAULT 0
                    );
                    CREATE TABLE IF NOT EXISTS messages (
                        id INTEGER PRIMARY KEY AUTOINCREMENT,
                        session_id TEXT NOT NULL REFERENCES sessions(id),
                        role TEXT NOT NULL CHECK(role IN ('system','user','assistant')),
                        content TEXT NOT NULL,
                        created_at TEXT NOT NULL,
                        usage_json TEXT
                    );
                    CREATE INDEX IF NOT EXISTS messages_session
                        ON messages(session_id, id);
                """)
        except (OSError, sqlite3.Error) as e:
            raise StorageError("无法初始化会话数据库") from e

    def _connect(self):
        db = sqlite3.connect(self.path, timeout=5)
        db.row_factory = sqlite3.Row
        db.execute("PRAGMA foreign_keys=ON")
        return db

    @staticmethod
    def _now():
        return datetime.now(timezone.utc).isoformat(timespec="microseconds")

    def create_session(self, system_prompt):
        session_id, now = uuid4().hex, self._now()
        try:
            with closing(self._connect()) as db, db:
                db.execute("INSERT INTO sessions(id,title,created_at,updated_at) VALUES(?,?,?,?)",
                           (session_id, "新会话", now, now))
                db.execute("INSERT INTO messages(session_id,role,content,created_at) VALUES(?,?,?,?)",
                           (session_id, "system", system_prompt, now))
        except sqlite3.Error as e:
            raise StorageError("创建会话失败") from e
        return session_id

    def list_sessions(self):
        try:
            with closing(self._connect()) as db:
                return [dict(row) for row in db.execute(
                    "SELECT * FROM sessions ORDER BY updated_at DESC, id")]
        except sqlite3.Error as e:
            raise StorageError("读取会话列表失败") from e

    def load_session(self, session_id):
        """只接受完整 ID；返回元数据及按写入顺序排列的完整消息。"""
        try:
            with closing(self._connect()) as db, db:
                session = db.execute("SELECT * FROM sessions WHERE id=?", (session_id,)).fetchone()
                if session is None:
                    raise StorageError("找不到该会话，请刷新网页并选择已有会话")
                messages = [dict(row) for row in db.execute(
                    "SELECT role,content FROM messages WHERE session_id=? ORDER BY id", (session_id,))]
                return dict(session), messages
        except sqlite3.Error as e:
            raise StorageError("加载会话失败") from e

    def save_turn(self, session_id, user_content, assistant_content, usage, expected_rounds):
        """问答及统计一起提交；轮数检查防止两个进程覆盖同一会话。"""
        used = usage.get("total_tokens", 0)
        if type(used) is not int or used < 0:
            raise StorageError("total_tokens 必须是非负整数，本轮尚未保存")
        try:
            usage_json = json.dumps(usage, ensure_ascii=False)
            now = self._now()
            with closing(self._connect()) as db, db:
                result = db.execute("""
                    UPDATE sessions SET rounds=rounds+1, total_tokens=total_tokens+?,
                        updated_at=?, title=CASE WHEN rounds=0 THEN ? ELSE title END
                    WHERE id=? AND rounds=?
                """, (used, now, user_content[:40], session_id, expected_rounds))
                if result.rowcount != 1:
                    raise StorageError("会话不存在或已被另一进程更新，请刷新会话后再试")
                db.executemany("""
                    INSERT INTO messages(session_id,role,content,created_at,usage_json)
                    VALUES(?,?,?,?,?)
                """, [(session_id, "user", user_content, now, None),
                      (session_id, "assistant", assistant_content, now, usage_json)])
        except (sqlite3.Error, TypeError, ValueError) as e:
            raise StorageError("保存问答失败，本轮未写入数据库") from e