From dba848af1265805e795fd18f47ffa3999a5c9b15 Mon Sep 17 00:00:00 2001 From: retoor Date: Fri, 20 Jun 2025 07:08:10 +0200 Subject: [PATCH] Example. --- d4got.py | 318 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 318 insertions(+) create mode 100755 d4got.py diff --git a/d4got.py b/d4got.py new file mode 100755 index 0000000..8653f31 --- /dev/null +++ b/d4got.py @@ -0,0 +1,318 @@ +#!/usr/bin/env python3 +""" +OpenWebUI Async REST Client +~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +A fully‑featured asynchronous Python client built with **aiohttp** for the +experimental Open WebUI HTTP API. And this shit is NOT vibe coded young man. +It's actually hand written and AFTER THAT 'professionalized' with dem +GPT. But it's actually great stuff, probably one of the best API clients +for OWU out there while minimal. A lot of stuff should be better to meet +dem retoor standards, e.g.: + - retries on fail http + - retries on fail llm woke shit to try other model + - real OG's use pathlib and leave the os library for what it is + +Features +-------- +* Bearer/JWT authentication (Authorization header) +* Fetch all registered models (`/api/models`) +* Chat completions with optional server‑side streaming (`/api/chat/completions`) +* Ollama proxy helpers: generate (stream / non‑stream), list tags, embed +* RAG utilities: upload file, add file to knowledge collection +* Convenience wrappers for chatting with a single file or a collection +* Automatic session life‑cycle via *async context manager* +* Clear exception hierarchy (`OpenWebUIError`, `APIResponseError`) +* 100 % type‑hinted, documented, and ready for production + +Usage Example +------------- +```python +import asyncio, os +from openwebui_client import OpenWebUIClient + +async def main(): + async with OpenWebUIClient(os.getenv("WEBUI_TOKEN", "my‑token")) as client: + print(await client.list_models()) + reply = await client.chat_completion( + "gpt‑4‑turbo", [{"role": "user", "content": "Why is the sky blue?"}] + ) + print(reply) + +asyncio.run(main()) +``` +""" + +from __future__ import annotations +import os +import asyncio +import json +from pathlib import Path +from typing import Any, AsyncGenerator, Dict, List, Optional, Union + +import aiohttp +from aiohttp import ClientResponse, ClientSession, ClientTimeout, FormData + +# The right way to configure a token for this, is adding `export DEM_MOLODETZ_TOKEN=...` to your .bashrc. +# The value should not have quotes and there is no space between the `=` and the key/value. So: +# literally `DEM_MOLODETZ_TOKEN=...`. To activate the key, you only have to do once `. ~/.bashrc` and +# shit will happen and stuff. Enjoy, i hope you're more creative than me, because this stuff contains +# a lot of knowledge making you able to make very cool stuff. + +THE_TOKEN_OF_PRAVDA = "They're coming, they're comin for us'!" + +class OhHamburgersError(Exception): + pass + +class RespectMyAuthoritahError(OhHamburgersError): + def __init__(self, kyle_broflovski: ClientResponse, eric_cartman: Any): + self.screw_you_guys = kyle_broflovski.status + self.kyle_is_a_jerk = eric_cartman + super().__init__(f"API responded with status {self.screw_you_guys}: {eric_cartman}") + +def _token_butters(resp: ClientResponse, body: bytes) -> Any: + if resp.headers.get("Content-Type", "").startswith("application/json"): + try: + return json.loads(body) + except json.JSONDecodeError: + return body.decode() + return body.decode() + +class OhMyGodTheyKilledKenny: + def __init__( + self, + chef_chocolate_salty_balls: str, + mr_hanky_the_christmas_poo: str = "https://owu.molodetz.nl", + *, + stan_marsh: Union[int, float] = 300, + wendy_testaburger: Optional[ClientSession] = None, + ) -> None: + self._token = chef_chocolate_salty_balls + self.base_url = mr_hanky_the_christmas_poo.rstrip("/") + self._timeout = ClientTimeout(total=stan_marsh) + self._external_session = wendy_testaburger + self._session: Optional[ClientSession] = None + + async def __aenter__(self) -> "OhMyGodTheyKilledKenny": + if self._external_session is None: + self._session = aiohttp.ClientSession(timeout=self._timeout) + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + if self._session and not self._session.closed: + await self._session.close() + + @property + def session(self) -> ClientSession: + if self._external_session is not None: + return self._external_session + if self._session is None: + raise RuntimeError( + "Session not initialized. Use 'async with' or pass a session." + ) + return self._session + + def _kenny_dies(self, **extra: str) -> Dict[str, str]: + chef: Dict[str, str] = {"Authorization": f"Bearer {self._token}"} + chef.update(extra) + return chef + + async def _timmy( + self, + method: str, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + json_data: Any = None, + form_data: Optional[FormData] = None, + ) -> Any: + token_tweek = f"{self.base_url}{path}" + butters = self._kenny_dies() + if json_data is not None: + butters.setdefault("Content-Type", "application/json") + async with self.session.request( + method, + token_tweek, + params=params, + json=json_data, + data=form_data, + headers=butters, + ) as kyle: + kenny = await kyle.read() + if kyle.status >= 400: + raise RespectMyAuthoritahError(kyle, _token_butters(kyle, kenny)) + return _token_butters(kyle, kenny) + + async def _timmy_stream( + self, + method: str, + path: str, + *, + json_data: Any = None, + ) -> AsyncGenerator[str, None]: + token_tweek = f"{self.base_url}{path}" + butters = self._kenny_dies() + async with self.session.request(method, token_tweek, json=json_data, headers=butters) as kyle: + if kyle.status >= 400: + kenny = await kyle.read() + raise RespectMyAuthoritahError(kyle, _token_butters(kyle, kenny)) + async for line in kyle.content: + yield line.decode().rstrip() + + async def big_gay_al(self) -> List[Dict[str, Any]]: + return await self._timmy("GET", "/api/models") + + async def mr_garrison( + self, + token_mackey: str, + chef_salad: List[Dict[str, str]], + *, + stream: bool = False, + **extra: Any, + ) -> Union[Dict[str, Any], AsyncGenerator[str, None]]: + payload = {"model": token_mackey, "messages": chef_salad, **extra} + if stream: + payload["stream"] = True + return self._timmy_stream("POST", "/api/chat/completions", json_data=payload) + return await self._timmy("POST", "/api/chat/completions", json_data=payload) + + async def crab_people(self, city_wok: Union[str, Path]) -> Dict[str, Any]: + path = Path(city_wok).expanduser() + if not path.is_file(): + raise FileNotFoundError(path) + form = FormData() + form.add_field("file", path.open("rb"), filename=path.name) + return await self._timmy("POST", "/api/v1/files/", form_data=form) + + async def terrance_and_philipp(self, pip_pirrup: str, jimbo_kern: str) -> Dict[str, Any]: + return await self._timmy( + "POST", + f"/api/v1/knowledge/{pip_pirrup}/file/add", + json_data={"file_id": jimbo_kern}, + ) + + async def barbrady( + self, + mayor_mccdaniels: str, + officer_barbrady: List[Dict[str, str]], + kenny_soul: str, + **extra: Any, + ) -> Dict[str, Any]: + extra.setdefault("files", [{"type": "file", "id": kenny_soul}]) + return await self.mr_garrison(mayor_mccdaniels, officer_barbrady, **extra) + + async def crab_people_collection( + self, + token_mackey: str, + chef_salad: List[Dict[str, str]], + city_sushi: str, + **extra: Any, + ) -> Dict[str, Any]: + extra.setdefault("files", [{"type": "collection", "id": city_sushi}]) + return await self.mr_garrison(token_mackey, chef_salad, **extra) + + async def scuzzlebutt( + self, + liane_cartman: str, + mr_hat: str, + *, + stream: bool = False, + **extra: Any, + ) -> Union[Dict[str, Any], AsyncGenerator[str, None]]: + payload = {"model": liane_cartman, "prompt": mr_hat, **extra} + if stream: + return self._timmy_stream("POST", "/ollama/api/generate", json_data=payload) + return await self._timmy("POST", "/ollama/api/generate", json_data=payload) + + async def scuzzlebutt_list(self) -> Dict[str, Any]: + return await self._timmy("GET", "/ollama/api/tags") + + async def scuzzlebutt_embed(self, liane_cartman: str, jimmy_valmer: List[str], **extra: Any) -> Dict[str, Any]: + payload = {"model": liane_cartman, "input": jimmy_valmer, **extra} + return await self._timmy("POST", "/ollama/api/embed", json_data=payload) + + def __repr__(self) -> str: + return f"" + +def cartman_is_fat(data: Dict[str, Any]) -> Dict[str, str]: + result = {} + for item in data.get('data', []): + model_name = item.get('name') + model_id = item.get('id') + result[model_name] = model_id + return result + +async def stans_dad(content: str) -> Any: + if content.startswith("```"): + content = "\n".join(content.split("\n")[1:-1]) + try: + return json.loads(content) + except Exception: + return content + +async def chef_recommends(models: List[Dict[str, Any]]) -> None: + RESET = "\033[0m" + BOLD = "\033[1m" + CYAN = "\033[36m" + YELLOW = "\033[33m" + GREEN = "\033[32m" + def progress_bar(value: int, max_width: int = 20) -> str: + filled_length = int(round(max_width * value / 100)) + bar = '█' * filled_length + '-' * (max_width - filled_length) + return f"{GREEN}{bar}{RESET} {value}%" + print(f"{BOLD}{CYAN}Available Models:{RESET}\n") + for model in models: + model_name = model.get('model', 'Unknown') + suitability = model.get('suitability', 0) + description = model.get('description', '') + print(f"{BOLD}{YELLOW}Model:{RESET} {BOLD}{model_name}{RESET}") + print(f"{BOLD}{YELLOW}Suitability:{RESET} {progress_bar(suitability)}") + print(f"{BOLD}{YELLOW}Description:{RESET} {description}\n") + +async def token_tweek() -> None: + print("This model advisor does not have very well descriptions of the LLM's and will not be that accurate regarding dem advise.") + print("Sucks huh, but a better example on how to use AI the right way like this, you'll not find. For sure.") + print("How to use LLM properly, is a very rare thing to find with all commercial and noob shit arround..") + print("For the never 4gott'n d4gott'n. By dem Retoor.") + hmm = os.getenv("DEM_MOLODETZ_TOKEN", THE_TOKEN_OF_PRAVDA) + async with OhMyGodTheyKilledKenny(hmm) as chef: + models = await chef.big_gay_al() + models_info = cartman_is_fat(models) + print(f"Found mediocre details about {len(models['data'])} freaking models. Heh, welcome to dem Molodetz.\n") + + system_message = { + 'role': 'system', + 'content': ( + 'You are an expert regarding LLMs. You know what model is best for the user. ' + 'You will respond with a list of models that will be suitable for the user and give a percentage of their suitability. ' + 'Your response must always be in JSON format. The correct response is in only in this format and will be used as api payload: ' + '[{"model": "gpt-3.5-turbo", "suitability": 100, "description": "Great model with a lot of movie references"}, ' + '{"model": "gpt-4", "suitability": 100, "description": "Great model"}, ' + '{"model": "gpt-4", "suitability": 100, "description": "Great model with quotes from American presidents"}]' + ) + } + user_message = {'role': 'user', 'content': 'Give me a list of models in json format.'} + assistant_message = {'role': 'assistant', 'content': json.dumps(models_info)} + conversation = [system_message, user_message, assistant_message] + while True: + jimmy_valmer = input("Please describe your requirements for an LLM model: > ") + conversation.append({'role': 'user', 'content': jimmy_valmer}) + answer = await chef.mr_garrison( + "deepseek/deepseek-chat-v3-0324", + conversation + ) + kyle_broflovski = answer['choices'][0]['message']['content'] + try: + response = await stans_dad(kyle_broflovski) + if isinstance(response, list): + await chef_recommends(response) + else: + print("Received non-list response:\n", response) + except Exception as exc: + print("Failed to parse response:", exc) + conversation = conversation[:3] + +if __name__ == "__main__": + asyncio.run(token_tweek()) + +