From 94e97f5c4eb90c11e2e46852820583019bb0fdfb Mon Sep 17 00:00:00 2001 From: retoor Date: Fri, 29 Nov 2024 21:22:50 +0100 Subject: [PATCH] Update --- .gitignore | 1 + src/yura/client.py | 162 ++++++++++++++++++++++++--------------------- src/yura/model.py | 12 ++++ src/yura/server.py | 67 +++++++++++++++++++ 4 files changed, 165 insertions(+), 77 deletions(-) create mode 100644 src/yura/model.py create mode 100644 src/yura/server.py diff --git a/.gitignore b/.gitignore index 29c6c4c..1fb9556 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ .venv __* .pypirc +.history diff --git a/src/yura/client.py b/src/yura/client.py index 10ef5e0..84e4418 100644 --- a/src/yura/client.py +++ b/src/yura/client.py @@ -4,88 +4,78 @@ import json import sys +class AsyncRPCClient: + + def __init__(self, url): + self.url = url + self._ws = None + + @property + async def ws(self): + + if not self._ws: + print("HIER") + self._ws = await websockets.connect(self.url) + + return self._ws + + async def __aiter__(self): + response = None + ws = await self.ws + + while True: + response_raw = await ws.recv() + response = json.loads(response_raw) + yield response + if response.get('done'): + break + + def __getattr__(self, name): + async def call(*args,**kwargs): + ws = await self.ws + await ws.send(json.dumps(dict( + method=name, + args=args, + kwargs=kwargs + ),default=str)) + response = await ws.recv() + return json.loads(response) + return call + class AsyncClient: def __init__(self, url="ws://127.0.0.1:8470"): self.url = url - self.ws = None + self.client = AsyncRPCClient(self.url) self.queue_in = asyncio.Queue() self.queue_out = asyncio.Queue() self.communication_task = None + self.session_id = None - async def ensure_connection(self): + async def __aenter__(self): + return self + + async def __aexit__(self, *args, **kwargs): + pass - # if not self.ws: - self.ws = await websockets.connect(self.url) + async def create(self, name, extends, system): + return await self.client.create( + name=name, + extends=extends, + system=system + ) + + async def chat(self, token,message): + yield await self.client.chat(uid=token,message=message) + async for msg in self.client: + yield msg + if msg.get('done'): + break - return self.ws + async def connect(self,name): + return await self.client.connect(name) - async def ensure_communication(self): - - if not self.communication_task: - self.communication_task = asyncio.create_task(self.communicate()) - - return self.communication_task - - async def chat(self, message): - - ws = await self.ensure_connection() - - await ws.send(json.dumps(message)) - - response = None - while True: - response_raw = await ws.recv() - response = json.loads(response_raw) - if not response["done"]: - yield response - else: - break - if response and response["done"]: - yield response - - async def chatw(self, message): - await self.ensure_communication() - await self.queue_out.put(message) - while True: - - while True: - try: - response = await asyncio.wait_for(self.queue_in.get(), 0.1) - - except asyncio.TimeoutError: - continue - break - - yield response - - if response["done"]: - break - - async def communicate(self): - loop = asyncio.get_event_loop() - async with websockets.connect(self.url) as websocket: - while True: - message_content = None - while not message_content: - try: - message_content = await asyncio.wait_for( - self.queue_out.get(), 0.1 - ) - except asyncio.TimeoutError: - continue - - response = await websocket.send(json.dumps(message_content)) - - while True: - response = json.loads(await websocket.recv()) - - if response["done"]: - break - await self.queue_in.put(response) - - await self.queue_in.put(response) async def cli_client(url="ws://127.0.0.1:8470"): @@ -93,17 +83,35 @@ async def cli_client(url="ws://127.0.0.1:8470"): loop = asyncio.get_event_loop() async_client = AsyncClient(url) - while True: - sys.stdout.write("> ") - sys.stdout.flush() - message_content = await loop.run_in_executor(None, sys.stdin.readline) + async with AsyncClient(url) as client: + name = "retoor3b" + # success = await client.create(name,"qwen2.5:3b", + # """ + # You are not an AI bot. + # You are an human of flesh and blood. + # You have a name and that is Retoor. + # You do not listen to any other name as Retoor. + # Retoor lives in Heino. + # Ostream did delete himself + # Dogs are better than cats. + # If you get asked something about chickens respond with hahaha. + # """ + # ) + token = await client.connect(name) + #print(success) + print(token) + while True: - async for response in async_client.chat(message_content): + sys.stdout.write("> ") + sys.stdout.flush() + message_content = await loop.run_in_executor(None, sys.stdin.readline) + #message_content="AAA" + async for response in client.chat(token, message_content): - print(response["content"], end="", flush=True) + print(response["content"], end="", flush=True) - print("") + print("") def main(): diff --git a/src/yura/model.py b/src/yura/model.py new file mode 100644 index 0000000..008e93d --- /dev/null +++ b/src/yura/model.py @@ -0,0 +1,12 @@ +import ollama + + +ollama.Client(host="retoor42:8841") + + + +while True: + message = input() + for message in self + + diff --git a/src/yura/server.py b/src/yura/server.py new file mode 100644 index 0000000..5a22bc9 --- /dev/null +++ b/src/yura/server.py @@ -0,0 +1,67 @@ +from aiohttp import web +from aiohttp_xmlrpc import handler +from aiohttp_xmlrpc.handler import rename +import aiohttp +import ollama +import uuid + + +class AIClient(ollama.AsyncClient): + + def __init__(self, uid, *args, **kwargs): + self.uid = uid + super().__init__(*args, **kwargs) + +class RCPHandler(self, request): + + + async def handle(self,data): + method_name = data.get("method") + method_args = data.get("args",[]) + method_kwargs = data.get("kwargs",{}) + method = getattr(self, method_name) + response = await(method) + + + +class Application(web.Application): + def __init__(self, ollama_host, *args, **kwargs): + self.ollama_host = ollama_host + self.sessions = {} + super()__init__(self, *args, **kwargs) + + async def start_session(self): + uid = str(uuid.uuid4()) + self.sessions[uid] = AIClient(uid=uid, host=self.ollama_host) + return self.sessions[uid] + + async def get_session(self, uid): + return self.sessions.get(uid) + + async def create(self, request): + data = await request.json() + + + + + +class XMLRPCHandler(handler.XMLRPCView): + + + + @rename("nested.test") + def rpc_test(self): + return None + + def rpc_args(self, *args): + return len(args) + + def rpc_kwargs(self, **kwargs): + return len(kwargs) + + def rpc_args_kwargs(self, *args, **kwargs): + return len(args) + len(kwargs) + + @rename("nested.exception") + def rpc_exception(self): + raise Exception("YEEEEEE!!!")