from pydantic import BaseModel, Field import requests class Pipe: class Valves(BaseModel): NAME_PREFIX: str = Field( default="OPENAI/", description="添加在模型名称前的前缀 / Prefix to be added before model names", ) OPENAI_API_BASE_URL: str = Field( default="https://open.bigmodel.cn/api/paas/v4/chat/completions", description="访问 OpenAI API 端点的基础 URL / Base URL for accessing OpenAI API endpoints", ) OPENAI_API_KEY: str = Field( default="36caeeae64d64955b357e0ced47141e4.SM9HpIYr72z6Uumu", description="用于验证 OpenAI API 请求的 API 密钥 / API key for authenticating requests to the OpenAI API", ) def __init__(self): self.valves = self.Valves() def pipes(self): if self.valves.OPENAI_API_KEY: try: headers = { "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}", "Content-Type": "application/json", } r = requests.get( f"{self.valves.OPENAI_API_BASE_URL}/models", headers=headers ) models = r.json() return [ { "id": model["id"], "name": f'{self.valves.NAME_PREFIX}{model.get("name", model["id"])}', } for model in models["data"] if "gpt" in model["id"] ] except Exception as e: return [ { "id": "error", "name": "获取模型时出错。请检查你的 API 密钥。", }, ] else: return [ { "id": "error", "name": "未提供 API 密钥。", }, ] def pipe(self, body: dict, __user__: dict): print(f"pipe:{__name__}") headers = { "Authorization": f"Bearer {self.valves.OPENAI_API_KEY}", "Content-Type": "application/json", } # 从模型名称中提取模型 ID / Extract model ID from model name model_id = body["model"][body["model"].find(".") + 1 :] # 在主体中更新模型 ID / Update model ID in the body payload = {**body, "model": model_id} try: r = requests.post( url=f"{self.valves.OPENAI_API_BASE_URL}/chat/completions", json=payload, headers=headers, stream=True, ) r.raise_for_status() if body.get("stream", False): return r.iter_lines() else: return r.json() except Exception as e: return f"错误：{e}"