""" title: GLM-4 Pipe version: 0.01 Author : Aa995 Description :https://bigmodel.cn/ offers free credits on thier models which are comparable to GPT4 """ import json import requests from typing import List, Union, Iterator from pydantic import BaseModel, Field from open_webui.utils.misc import ( pop_system_message, ) # Adjust based on your OpenWebUI structure # Set DEBUG to True to enable detailed logging DEBUG = True class Pipe: class Valves(BaseModel): GLM4_API_KEY: str = Field(default="", description="Your GLM-4 API key.") GLM4_ENDPOINT: str = Field( default="https://open.bigmodel.cn/api/paas/v4/chat/completions", description="GLM-4 synchronous API endpoint.", ) def __init__(self): self.type = "glm4" self.id = "glmfunction" self.name = "GLM-4 Models" # Temporarily hardcode the API key for testing self.valves = self.Valves( GLM4_API_KEY="YOUR API KEY GOES HERE" ) if DEBUG: if self.valves.GLM4_API_KEY: print("GLM4_API_KEY is set.") else: print("GLM4_API_KEY is NOT set.") self.models = [ "glm-4-plus", "glm-4-0520", "glm-4", "glm-4-air", "glm-4-airx", "glm-4-long", "glm-4-flash", ] def get_glm4_models(self) -> List[dict]: """ Returns a list of available GLM-4 models without the Pipe ID prefix. OpenWebUI will handle prefixing based on the Pipe's ID. """ return [ {"id": model, "name": model.replace("glm-4", "GLM-4 ")} for model in self.models ] def pipes(self) -> List[dict]: """ OpenWebUI expects a pipes() method that returns available models. """ return self.get_glm4_models() def pipe(self, body: dict) -> Union[str, Iterator[str]]: """ Handles incoming requests and routes them to the appropriate handler (synchronous or streaming) based on the 'stream' flag. """ model_id = body.get("model") if not model_id: return "Error: No model specified." if DEBUG: print(f"Original model_id: {model_id}") # Remove the function ID prefix if present function_id_with_dot = f"{self.id}." if model_id.startswith(function_id_with_dot): model_id = model_id[len(function_id_with_dot) :] if DEBUG: print( f"Removed prefix '{function_id_with_dot}' from model_id, new model_id: {model_id}" ) else: if DEBUG: print("No prefix to remove from model_id.") if DEBUG: print(f"Using model_id: {model_id}") if model_id not in self.models: return f"Error: Unknown GLM-4 model '{model_id}'. Available models are: {', '.join(self.models)}." # Process messages system_message, messages = pop_system_message(body.get("messages", [])) processed_messages = [] # Add system message if it exists if system_message: processed_messages.append( {"role": "system", "content": str(system_message)} ) # Process user and assistant messages for message in messages: content = message.get("content", "") if isinstance(content, list): # Process multimedia content content_parts = [] for item in content: if item["type"] == "text": content_parts.append(item["text"]) elif item["type"] == "image_url": # Currently, GLM-4 does not process images; skip or handle accordingly if DEBUG: print( "Image URLs are not processed by GLM-4. Skipping image." ) continue combined_content = " ".join(content_parts) if combined_content: processed_messages.append( {"role": message["role"], "content": combined_content} ) else: # Simple text content processed_messages.append({"role": message["role"], "content": content}) stream = body.get("stream", False) try: if stream: return self.call_glm4_stream(body, processed_messages, model_id) else: return self.call_glm4_sync(body, processed_messages, model_id) except Exception as e: if DEBUG: print(f"Error in GLM-4 pipe method: {e}") return f"Error: {e}" def call_glm4_sync(self, body: dict, messages: List[dict], model_id: str) -> str: """ Handles synchronous requests to GLM-4 models. """ api_key = self.valves.GLM4_API_KEY if not api_key: return "Error: GLM-4 API key is not set. Please configure it in OpenWebUI's settings." if DEBUG: print("GLM4_API_KEY retrieved successfully.") url = self.valves.GLM4_ENDPOINT headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", } if DEBUG: print(f"Authorization header set to: Bearer {'*' * len(api_key)}") print(f"Sending request to URL: {url}") payload = { "model": model_id, "messages": messages, "temperature": body.get("temperature", 0.95), "top_p": body.get("top_p", 0.7), "max_tokens": body.get("max_tokens", 1024), "do_sample": body.get("do_sample", True), "stream": False, } # Include optional parameters from the body optional_fields = ["request_id", "stop", "tools", "tool_choice", "user_id"] for field in optional_fields: if field in body: payload[field] = body[field] if DEBUG: print("GLM-4 Sync Request Payload:") print(json.dumps(payload, indent=2)) response = requests.post(url, headers=headers, json=payload, timeout=60) if response.status_code == 200: response_data = response.json() if DEBUG: print("GLM-4 Sync Response Data:") print(json.dumps(response_data, indent=2)) choices = response_data.get("choices", []) if choices: generated_text = choices[0].get("message", {}).get("content", "") return generated_text else: if DEBUG: print("No choices found in the GLM-4 response.") return "No response generated by the GLM-4 model." else: if DEBUG: print(f"GLM-4 Sync Error: {response.status_code}") print("Response Headers:", response.headers) print("Response Content:", response.text) return f"Error: {response.status_code} {response.text}" def call_glm4_stream( self, body: dict, messages: List[dict], model_id: str ) -> Iterator[str]: """ Handles streaming (SSE) requests to GLM-4 models. """ api_key = self.valves.GLM4_API_KEY if not api_key: yield "Error: GLM-4 API key is not set. Please configure it in OpenWebUI's settings." return url = self.valves.GLM4_ENDPOINT headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}", } payload = { "model": model_id, "messages": messages, "temperature": body.get("temperature", 0.95), "top_p": body.get("top_p", 0.7), "max_tokens": body.get("max_tokens", 1024), "do_sample": body.get("do_sample", True), "stream": True, } # Include optional parameters from the body optional_fields = ["request_id", "stop", "tools", "tool_choice", "user_id"] for field in optional_fields: if field in body: payload[field] = body[field] if DEBUG: print("GLM-4 Stream Request Payload:") print(json.dumps(payload, indent=2)) try: with requests.post( url, headers=headers, json=payload, timeout=60, stream=True ) as response: if response.status_code != 200: if DEBUG: print(f"GLM-4 Stream Error: {response.status_code}") print("Response Headers:", response.headers) print("Response Content:", response.text) yield f"Error: {response.status_code} {response.text}" return for line in response.iter_lines(): if line: decoded_line = line.decode("utf-8") if decoded_line.startswith("data:"): content = decoded_line[5:].strip() if content == "[DONE]": break try: data = json.loads(content) choices = data.get("choices", []) if choices: delta = choices[0].get("delta", {}) if "content" in delta: yield delta["content"] except json.JSONDecodeError: if DEBUG: print(f"Failed to parse JSON: {content}") except KeyError as e: if DEBUG: print(f"Unexpected data structure: {e}") except requests.exceptions.RequestException as e: if DEBUG: print(f"GLM-4 Stream Request Exception: {e}") yield f"Error: {e}"