001
# src/fastapi/api/stream/deepseek_sse.py
002
"""使用 .env 中的 DeepSeek 配置调用真实模型的 SSE 接口。"""
003

004
import asyncio
005
import logging
006
from collections.abc import AsyncIterator
007
from typing import Annotated
008
from uuid import uuid4
009

010
from fastapi import APIRouter, Depends, HTTPException
011
from fastapi.sse import EventSourceResponse, ServerSentEvent
012
from langchain_openai import ChatOpenAI
013

014
from app.core.config import Settings, get_settings
015
from app.model import create_model
016

017
from .common import ConversationRequest
018

019
logger = logging.getLogger(__name__)
020
router = APIRouter()
021

022

023
def get_deepseek_model(
024
    settings: Annotated[Settings, Depends(get_settings)],
025
) -> ChatOpenAI:
026
    """在流启动前创建模型，缺少密钥时返回普通 HTTP 错误。"""
027

028
    try:
029
        return create_model(settings)
030
    except RuntimeError as exc:
031
        raise HTTPException(
032
            status_code=503,
033
            detail="DeepSeek 模型配置不完整",
034
        ) from exc
035

036

037
# POST /api/v1/stream/deepseek
038
@router.post(
039
    "/deepseek",
040
    response_class=EventSourceResponse,
041
    summary="DeepSeek 流式聊天",
042
)
043
async def stream_deepseek(
044
    payload: ConversationRequest,
045
    model: Annotated[ChatOpenAI, Depends(get_deepseek_model)],
046
) -> AsyncIterator[ServerSentEvent]:
047
    """调用 DeepSeek，并把 LangChain 消息块转换成 SSE 生命周期事件。"""
048

049
    message_id = f"msg_{uuid4().hex}"
050
    sequence = 0
051
    usage: dict[str, object] | None = None
052

053
    yield ServerSentEvent(
054
        event="message_start",
055
        id=f"{message_id}:{sequence}",
056
        retry=3_000,
057
        data={
058
            "message_id": message_id,
059
            "model": model.model_name,
060
        },
061
    )
062

063
    try:
064
        async for chunk in model.astream(payload.to_langchain_messages()):
065
            if chunk.usage_metadata is not None:
066
                usage = dict(chunk.usage_metadata)
067

068
            # ``text`` 会从字符串或标准文本内容块中提取可展示文本。
069
            if not chunk.text:
070
                continue
071

072
            sequence += 1
073
            yield ServerSentEvent(
074
                event="token",
075
                id=f"{message_id}:{sequence}",
076
                data={"text": chunk.text},
077
            )
078
    except asyncio.CancelledError:
079
        # 客户端断开时让取消信号继续传播，以便终止上游读取。
080
        raise
081
    except Exception:
082
        logger.exception("DeepSeek streaming request failed")
083
        sequence += 1
084
        yield ServerSentEvent(
085
            event="stream_error",
086
            id=f"{message_id}:{sequence}",
087
            data={
088
                "code": "model_unavailable",
089
                "message": "模型暂时不可用，请稍后重试",
090
            },
091
        )
092
        return
093

094
    sequence += 1
095
    yield ServerSentEvent(
096
        event="message_end",
097
        id=f"{message_id}:{sequence}",
098
        data={
099
            "finish_reason": "stop",
100
            "usage": usage,
101
        },
102
    )