""" title: FastGPT Pipe Author: Trae CN Orginal_authors: LB7666 Updated_author: silenceroom author_url: https://github.com/acking-you funding_url: https://github.com/open-webui version: 0.1 required_open_webui_version: 0.5.6 license: MIT """ import os import requests import json import time import uuid from typing import List, Union, Generator, Iterator from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message class Pipe: class Valves(BaseModel): API_KEY: str = Field(default="") def __init__(self): self.type = "manifold" self.id = "fastgpt" self.name = "fastgpt/" self.valves = self.Valves(**{"API_KEY": os.getenv("FASTGPT_API_KEY", "")}) self.data_prefix = "data:" def get_fastgpt_models(self): return [ {"id": "fastgpt-pro", "name": "FastGPT Pro"}, {"id": "fastgpt-standard", "name": "FastGPT Standard"}, ] def pipes(self) -> List[dict]: return self.get_fastgpt_models() def pipe(self, body: dict) -> Union[str, Generator, Iterator]: system_message, messages = pop_system_message(body["messages"]) processed_messages = [] for message in messages: processed_content = message.get("content", "") if isinstance(processed_content, list): processed_content = " ".join( [ item["text"] for item in processed_content if item["type"] == "text" ] ) processed_messages.append( {"role": message["role"], "content": processed_content} ) payload = { "chatId": body.get("chatId", str(uuid.uuid4())), "stream": body.get("stream", False), "detail": False, # 根据 API 文档固定为 False "messages": processed_messages, } headers = { "Authorization": f"Bearer {self.valves.API_KEY}", "Content-Type": "application/json", } url = "https://api.fastgpt.in/api/v1/chat/completions" try: if body.get("stream"): return self.stream_response(url, headers, payload) else: return self.non_stream_response(url, headers, payload) except Exception as e: return f"Error: {str(e)}" def stream_response(self, url, headers, payload): try: response = requests.post( url=url, json=payload, headers=headers, stream=True ) if response.status_code != 200: raise Exception(f"HTTP Error {response.status_code}: {response.text}") for chunk in response.iter_content(chunk_size=None): if chunk: decoded_chunk = chunk.decode("utf-8") for line in decoded_chunk.splitlines(): if line.startswith(self.data_prefix): json_str = line[len(self.data_prefix) :] if json_str.strip() == "[DONE]": continue try: data = json.loads(json_str) content = data["choices"][0]["delta"].get("content", "") if content: yield content except json.JSONDecodeError: continue except Exception as e: yield f"Error: {str(e)}" def non_stream_response(self, url, headers, payload): try: response = requests.post(url=url, headers=headers, json=payload, timeout=10) if response.status_code != 200: raise Exception(f"HTTP Error {response.status_code}: {response.text}") return response.json()["choices"][0]["message"]["content"] except Exception as e: return f"Error: {str(e)}"