From 2edcb51805a63d8f464fa0057b8415d1dfa7ead0 Mon Sep 17 00:00:00 2001 From: Xin Wang Date: Mon, 25 May 2026 08:45:30 +0800 Subject: [PATCH] Add chat_id, variables, detail, timeout_sec, and send_system_prompt to LLMConfig; update create_llm_service to accept new parameters and handle FastGPT integration. Modify pipeline to utilize chat_id and adjust message handling based on provider settings. --- config/fastgpt.example.json | 49 ++++++ engine/config.py | 13 +- engine/fastgpt_llm.py | 301 ++++++++++++++++++++++++++++++++++++ engine/pipeline.py | 24 ++- engine/services.py | 25 ++- 5 files changed, 406 insertions(+), 6 deletions(-) create mode 100644 config/fastgpt.example.json create mode 100644 engine/fastgpt_llm.py diff --git a/config/fastgpt.example.json b/config/fastgpt.example.json new file mode 100644 index 0000000..78a8ce8 --- /dev/null +++ b/config/fastgpt.example.json @@ -0,0 +1,49 @@ +{ + "server": { + "host": "0.0.0.0", + "port": 8000, + "cors_origins": ["*"] + }, + "audio": { + "sample_rate_hz": 16000, + "channels": 1, + "frame_ms": 20 + }, + "session": { + "inactivity_timeout_sec": 60 + }, + "agent": { + "system_prompt": "FastGPT app owns the system prompt when send_system_prompt is false.", + "greeting": "你好", + "greeting_mode": "generated" + }, + "services": { + "stt": { + "provider": "openai", + "api_key": "YOUR_STT_KEY", + "base_url": "https://api.openai.com/v1", + "model": "gpt-4o-mini-transcribe", + "language": "zh" + }, + "llm": { + "provider": "fastgpt", + "api_key": "fastgpt-xxxxx", + "base_url": "http://localhost:3000", + "model": "my-voice-app", + "chat_id": null, + "variables": { + "user_name": "访客" + }, + "detail": false, + "timeout_sec": 60.0, + "send_system_prompt": false + }, + "tts": { + "provider": "openai", + "api_key": "YOUR_TTS_KEY", + "base_url": "https://api.openai.com/v1", + "model": "gpt-4o-mini-tts", + "voice": "alloy" + } + } +} diff --git a/engine/config.py b/engine/config.py index 962180b..4971323 100644 --- a/engine/config.py +++ b/engine/config.py @@ -104,6 +104,11 @@ class LLMConfig: base_url: str | None = None model: str = "gpt-4o-mini" temperature: float | None = 0.7 + chat_id: str | None = None + variables: dict[str, str] = field(default_factory=dict) + detail: bool = False + timeout_sec: float = 60.0 + send_system_prompt: bool = False @dataclass(frozen=True) @@ -180,6 +185,12 @@ def config_from_dict(data: dict) -> EngineConfig: if stt.get("language") == "": stt["language"] = None + llm = _dict(services.get("llm")) + if llm.get("chat_id") == "": + llm["chat_id"] = None + if not isinstance(llm.get("variables"), dict): + llm["variables"] = {} + turn = _dict(data.get("turn")) vad = _dict(turn.get("vad")) @@ -207,7 +218,7 @@ def config_from_dict(data: dict) -> EngineConfig: ), agent=AgentConfig(**agent), services=ServicesConfig( - llm=LLMConfig(**_dict(services.get("llm"))), + llm=LLMConfig(**llm), stt=STTConfig(**stt), tts=TTSConfig(**_dict(services.get("tts"))), ), diff --git a/engine/fastgpt_llm.py b/engine/fastgpt_llm.py new file mode 100644 index 0000000..862091e --- /dev/null +++ b/engine/fastgpt_llm.py @@ -0,0 +1,301 @@ +from __future__ import annotations + +import uuid +from dataclasses import dataclass, field +from typing import Any + +import httpx +from fastgpt_client import AsyncChatClient, FastGPTInteractiveEvent, aiter_stream_events +from fastgpt_client.exceptions import FastGPTError +from loguru import logger + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, + OutputTransportMessageFrame, +) +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.llm_service import LLMService +from pipecat.services.settings import LLMSettings + + +def _extract_text_from_event(kind: str, payload: Any) -> str: + if not isinstance(payload, dict): + return "" + + if kind in {"answer", "fastAnswer"}: + text = payload.get("text") + if isinstance(text, str) and text: + return text + + choices = payload.get("choices") if isinstance(payload.get("choices"), list) else [] + if not choices: + return str(payload.get("text") or "") + + first_choice = choices[0] if isinstance(choices[0], dict) else {} + delta = first_choice.get("delta") if isinstance(first_choice.get("delta"), dict) else {} + content = delta.get("content") + if isinstance(content, str) and content: + return content + + message = first_choice.get("message") if isinstance(first_choice.get("message"), dict) else {} + message_content = message.get("content") + if isinstance(message_content, str) and message_content: + return message_content + + return "" + + +def _message_text(message: dict[str, Any]) -> str: + content = message.get("content") + if isinstance(content, str): + return content.strip() + if isinstance(content, list): + parts: list[str] = [] + for part in content: + if isinstance(part, dict) and part.get("type") == "text": + text = part.get("text") + if isinstance(text, str) and text.strip(): + parts.append(text.strip()) + return " ".join(parts) + return "" + + +def _first_nonempty_text(*values: Any) -> str: + for value in values: + if isinstance(value, str): + text = value.strip() + if text: + return text + return "" + + +def _interactive_spoken_prompt(event: FastGPTInteractiveEvent) -> str: + payload = event.data if isinstance(event.data, dict) else {} + params = payload.get("params") if isinstance(payload.get("params"), dict) else {} + + prompt = _first_nonempty_text( + payload.get("opener"), + params.get("opener"), + payload.get("prompt"), + params.get("prompt"), + payload.get("text"), + params.get("text"), + payload.get("title"), + params.get("title"), + payload.get("description"), + params.get("description"), + ) + if prompt: + return prompt + + if event.interaction_type == "userSelect": + raw_options = ( + params.get("userSelectOptions") + if isinstance(params.get("userSelectOptions"), list) + else [] + ) + labels: list[str] = [] + for index, raw in enumerate(raw_options, start=1): + if isinstance(raw, str) and raw.strip(): + labels.append(f"{index}. {raw.strip()}") + elif isinstance(raw, dict): + label = _first_nonempty_text(raw.get("label"), raw.get("value")) + if label: + labels.append(f"{index}. {label}") + if labels: + return "请选择：" + "，".join(labels) + return "请选择一个选项。" + + if event.interaction_type == "userInput": + input_form = params.get("inputForm") if isinstance(params.get("inputForm"), list) else [] + labels = [ + _first_nonempty_text(field.get("label"), field.get("name")) + for field in input_form + if isinstance(field, dict) + ] + labels = [label for label in labels if label] + if labels: + return "请提供以下信息：" + "，".join(labels) + return "请补充所需信息。" + + return "请继续。" + + +@dataclass +class FastGPTLLMSettings(LLMSettings): + variables: dict[str, Any] = field(default_factory=dict) + detail: bool = False + + +class FastGPTLLMService(LLMService): + """FastGPT LLM service using chatId server-side memory and workflow variables.""" + + Settings = FastGPTLLMSettings + + def __init__( + self, + *, + api_key: str, + base_url: str, + chat_id: str | None = None, + send_system_prompt: bool = False, + greeting_prompt: str | None = None, + timeout: float = 60.0, + settings: FastGPTLLMSettings | None = None, + **kwargs, + ) -> None: + default_settings = self.Settings(model="fastgpt") + if settings is not None: + default_settings.apply_update(settings) + super().__init__(settings=default_settings, **kwargs) + + self._chat_id = chat_id or f"voice_{uuid.uuid4().hex[:16]}" + self._send_system_prompt = send_system_prompt + self._greeting_prompt = (greeting_prompt or "你好").strip() or "你好" + self._client = AsyncChatClient( + api_key=api_key, + base_url=base_url, + timeout=timeout, + ) + self._active_response = None + + @property + def chat_id(self) -> str: + return self._chat_id + + def set_variables(self, variables: dict[str, Any]) -> None: + merged = dict(self._settings.variables) + merged.update(variables) + self._settings.variables = merged + + async def stop(self, frame: EndFrame) -> None: + await self._close_active_response() + await self._client.close() + await super().stop(frame) + + async def cancel(self, frame: CancelFrame) -> None: + await self._close_active_response() + await super().cancel(frame) + + async def _close_active_response(self) -> None: + response = self._active_response + self._active_response = None + if response is not None: + await response.aclose() + + def _build_fastgpt_messages(self, context: LLMContext) -> list[dict[str, str]]: + raw_messages = context.get_messages() + messages: list[dict[str, str]] = [] + + if self._send_system_prompt: + for message in raw_messages: + if not isinstance(message, dict) or message.get("role") != "system": + continue + text = _message_text(message) + if text: + messages.append({"role": "system", "content": text}) + + for message in reversed(raw_messages): + if not isinstance(message, dict) or message.get("role") != "user": + continue + text = _message_text(message) + if text: + messages.append({"role": "user", "content": text}) + return messages + + messages.append({"role": "user", "content": self._greeting_prompt}) + return messages + + async def _process_context(self, context: LLMContext) -> None: + messages = self._build_fastgpt_messages(context) + variables = self._settings.variables or None + + await self.start_ttfb_metrics() + + try: + response = await self._client.create_chat_completion( + messages=messages, + stream=True, + chatId=self._chat_id, + variables=variables, + detail=self._settings.detail, + ) + except FastGPTError as exc: + await self.push_error(error_msg=f"FastGPT request failed: {exc}", exception=exc) + return + except httpx.HTTPError as exc: + await self.push_error(error_msg=f"FastGPT HTTP error: {exc}", exception=exc) + return + + self._active_response = response + + try: + async for event in aiter_stream_events(response): + if event.kind in {"data", "answer", "fastAnswer"}: + text = _extract_text_from_event(event.kind, event.data) + if text: + await self.stop_ttfb_metrics() + await self.push_frame(LLMTextFrame(text)) + continue + + if event.kind == "interactive" and isinstance(event, FastGPTInteractiveEvent): + await self._handle_interactive(event) + break + + if event.kind == "error": + payload = event.data if isinstance(event.data, dict) else {} + message = _first_nonempty_text( + payload.get("message"), + payload.get("error"), + ) or "FastGPT stream error" + await self.push_error(error_msg=message) + break + + if event.kind == "done": + break + finally: + self._active_response = None + await response.aclose() + + async def _handle_interactive(self, event: FastGPTInteractiveEvent) -> None: + prompt = _interactive_spoken_prompt(event) + if prompt: + await self.stop_ttfb_metrics() + await self.push_frame(LLMTextFrame(prompt)) + + await self.push_frame( + OutputTransportMessageFrame( + message={ + "type": "response.interactive", + "interaction_type": event.interaction_type, + "data": event.data, + } + ), + FrameDirection.DOWNSTREAM, + ) + + async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: + await super().process_frame(frame, direction) + + if isinstance(frame, LLMContextFrame): + try: + await self.push_frame(LLMFullResponseStartFrame()) + await self.start_processing_metrics() + await self._process_context(frame.context) + except httpx.TimeoutException as exc: + await self._call_event_handler("on_completion_timeout") + await self.push_error(error_msg="FastGPT completion timeout", exception=exc) + except Exception as exc: + await self.push_error(error_msg=f"FastGPT completion error: {exc}", exception=exc) + finally: + await self.stop_processing_metrics() + await self.push_frame(LLMFullResponseEndFrame()) + else: + await self.push_frame(frame, direction) diff --git a/engine/pipeline.py b/engine/pipeline.py index 1dcb6b7..d32f2f7 100644 --- a/engine/pipeline.py +++ b/engine/pipeline.py @@ -1,5 +1,7 @@ from __future__ import annotations +import uuid + from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer @@ -82,12 +84,26 @@ async def run_pipeline_with_serializer( ) stt = create_stt_service(config.services.stt, config.audio) - llm = create_llm_service(config.services.llm) + + llm_config = config.services.llm + chat_id = llm_config.chat_id or f"voice_{uuid.uuid4().hex[:16]}" + llm = create_llm_service( + llm_config, + chat_id=chat_id, + session_variables={"session_id": chat_id, "channel": "voice"}, + greeting_prompt=config.agent.greeting, + ) + if llm_config.provider == "fastgpt": + logger.info(f"FastGPT chatId={chat_id}") + tts = create_tts_service(config.services.tts, config.audio) - messages = [{"role": "system", "content": config.agent.system_prompt}] - if config.agent.greeting and config.agent.greeting_mode == "generated": - messages.append({"role": "system", "content": config.agent.greeting}) + use_fastgpt = llm_config.provider == "fastgpt" and not llm_config.send_system_prompt + messages: list[dict[str, str]] = [] + if not use_fastgpt: + messages = [{"role": "system", "content": config.agent.system_prompt}] + if config.agent.greeting and config.agent.greeting_mode == "generated": + messages.append({"role": "system", "content": config.agent.greeting}) context = LLMContext(messages) diff --git a/engine/services.py b/engine/services.py index b322be6..4272003 100644 --- a/engine/services.py +++ b/engine/services.py @@ -13,6 +13,7 @@ from pipecat.services.openai.tts import VALID_VOICES, OpenAITTSService from pipecat.transcriptions.language import Language from .config import AudioConfig, LLMConfig, STTConfig, TTSConfig +from .fastgpt_llm import FastGPTLLMService, FastGPTLLMSettings from .xfyun_asr import DEFAULT_XFYUN_ASR_URL, XfyunASRService from .xfyun_tts import DEFAULT_XFYUN_TTS_URL, XfyunTTSService @@ -46,7 +47,29 @@ def create_stt_service(config: STTConfig, audio: AudioConfig | None = None): ) -def create_llm_service(config: LLMConfig): +def create_llm_service( + config: LLMConfig, + *, + chat_id: str | None = None, + session_variables: dict | None = None, + greeting_prompt: str | None = None, +): + if config.provider == "fastgpt": + variables = {**config.variables, **(session_variables or {})} + return FastGPTLLMService( + api_key=config.api_key, + base_url=config.base_url or "http://localhost:3000", + chat_id=chat_id or config.chat_id, + send_system_prompt=config.send_system_prompt, + greeting_prompt=greeting_prompt, + timeout=config.timeout_sec, + settings=FastGPTLLMSettings( + model=config.model or "fastgpt", + variables=variables, + detail=config.detail, + ), + ) + _require_provider(config.provider, "openai", "llm") return OpenAILLMService( api_key=config.api_key or None,