ai-video-fullstack/backend/services/brains/fastgpt_llm.py

"""FastGPT 作为 pipecat LLM 槽位。

与普通 LLM 的关键不同:context / 知识库 / 工具全在 FastGPT 服务端,靠 chatId
维持会话。所以本服务只发「最后一条 user 文本」+ 稳定 chatId,把流式 answer
事件转成 LLMTextFrame 交给下游 TTS;不消费/不依赖本地 LLMContext 的历史。
"""

from __future__ import annotations

from typing import Any

from fastgpt_client import AsyncChatClient, aiter_stream_events
from loguru import logger
from models import AssistantConfig

from pipecat.frames.frames import (
    Frame,
    LLMContextFrame,
    LLMFullResponseEndFrame,
    LLMFullResponseStartFrame,
    LLMTextFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.settings import LLMSettings

# 承载回复文本的事件种类。detail=False 时 FastGPT 走 OpenAI 兼容流,文本以裸
# data: 块下发(无 event 名 → kind="data");detail=True / 旧版则用 answer/fastAnswer。
_ANSWER_KINDS = {"data", "answer", "fastAnswer"}

# SDK 会自动在 base_url 后拼 /api/v1/chat/completions(并去掉末尾 /api)。
# 用户常把「完整接口地址」填进 api_url,这里剥掉这些后缀,归一成主机根地址,
# 避免路径重复导致 404。
_ENDPOINT_SUFFIXES = (
    "/api/v1/chat/completions",
    "/v1/chat/completions",
    "/chat/completions",
)


def normalize_base_url(url: str) -> str:
    base = (url or "").strip().rstrip("/")
    for suffix in _ENDPOINT_SUFFIXES:
        if base.endswith(suffix):
            base = base[: -len(suffix)]
            break
    return base or "http://localhost:3000"


def _last_user_text(messages: list[dict]) -> str:
    """取最近一条 user 消息的纯文本(兼容多模态分片)。"""
    for message in reversed(messages or []):
        if message.get("role") != "user":
            continue
        content = message.get("content")
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            return "".join(
                str(part.get("text") or "")
                for part in content
                if isinstance(part, dict)
            )
    return ""


def _event_text(data: Any) -> str:
    """从一个流事件里取增量文本。

    兼容两种形态(对齐 SDK examples 的解析):
      - 直接 text 字段(answer/fastAnswer 详情流);
      - OpenAI 兼容块 choices[0].delta.content / message.content(detail=False)。
    """
    if not isinstance(data, dict):
        return ""

    text = data.get("text")
    if isinstance(text, str) and text:
        return text

    choices = data.get("choices")
    if not isinstance(choices, list) or not choices:
        return ""
    first = choices[0] if isinstance(choices[0], dict) else {}

    delta = first.get("delta")
    if isinstance(delta, dict):
        content = delta.get("content")
        if isinstance(content, str) and content:
            return content

    message = first.get("message")
    if isinstance(message, dict):
        content = message.get("content")
        if isinstance(content, str) and content:
            return content

    return ""


class FastGPTLLMService(LLMService):
    """包 FastGPT OpenAPI 的伪 LLM 服务。"""

    def __init__(self, cfg: AssistantConfig, chat_id: str):
        # FastGPT 自管 model/温度等参数,这里把所有 LLM 设置初始化为 None,
        # 满足基类 validate_complete(否则启动期会报 NOT_GIVEN)。
        super().__init__(
            settings=LLMSettings(
                model=None,
                system_instruction=None,
                temperature=None,
                max_tokens=None,
                top_p=None,
                top_k=None,
                frequency_penalty=None,
                presence_penalty=None,
                seed=None,
                filter_incomplete_user_turns=None,
                user_turn_completion_config=None,
            )
        )
        self._chat_id = chat_id
        self._base_url = normalize_base_url(cfg.fastgpt_api_url)
        self._client = AsyncChatClient(
            api_key=cfg.fastgpt_api_key,
            base_url=self._base_url,
        )

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)

        if not isinstance(frame, LLMContextFrame):
            await self.push_frame(frame, direction)
            return

        user_text = _last_user_text(frame.context.get_messages())
        if not user_text:
            return

        await self.push_frame(LLMFullResponseStartFrame())
        try:
            response = await self._client.create_chat_completion(
                messages=[{"role": "user", "content": user_text}],
                stream=True,
                chatId=self._chat_id,
                detail=False,
            )
            async for event in aiter_stream_events(response):
                if event.kind in _ANSWER_KINDS:
                    text = _event_text(event.data)
                    if text:
                        await self.push_frame(LLMTextFrame(text))
                elif event.kind == "error":
                    logger.error(f"FastGPT 流式错误: {event.data}")
        except Exception as exc:  # noqa: BLE001 - 单轮失败不应中断通话
            logger.error(
                f"FastGPT 调用失败: {exc} "
                f"(base_url={self._base_url},拼接后应为 {self._base_url}/api/v1/chat/completions)"
            )
        finally:
            await self.push_frame(LLMFullResponseEndFrame())