From 4685924ec5e3b4276e2b715346e841917b099a88 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 9 Apr 2024 18:10:52 +0800 Subject: [PATCH] bugfix --- ahserver/baseProcessor.py | 1 + ahserver/configuredServer.py | 15 ++++++- ahserver/globalEnv.py | 78 +++++++++++++++++++++++++++++++++ ahserver/llm_client.py | 85 +++++++++++++++++++++++++++++------- 4 files changed, 162 insertions(+), 17 deletions(-) diff --git a/ahserver/baseProcessor.py b/ahserver/baseProcessor.py index 395dac7..a66fefe 100755 --- a/ahserver/baseProcessor.py +++ b/ahserver/baseProcessor.py @@ -72,6 +72,7 @@ class BaseProcessor(AppLogger): self.run_ns.update(g) self.run_ns.update(self.resource.y_env) self.run_ns['request'] = request + self.run_ns['app'] = request.app kw = await self.run_ns['request2ns']() kw.update(params) self.run_ns['params_kw'] = kw diff --git a/ahserver/configuredServer.py b/ahserver/configuredServer.py index b61ae74..8269335 100755 --- a/ahserver/configuredServer.py +++ b/ahserver/configuredServer.py @@ -6,7 +6,7 @@ from socket import * from aiohttp import web from appPublic.folderUtils import ProgramPath -from appPublic.background import Background +from appPublic.dictObject import DictObject from appPublic.jsonConfig import getConfig from appPublic.app_logger import AppLogger @@ -19,6 +19,17 @@ from .globalEnv import initEnv from .filestorage import TmpFileRecord from .loadplugins import load_plugins +class AHApp(web.Application): + def __init__(self, *args, **kw): + super().__init__(*args, **kw) + self.data = DictObject() + + def set_data(self, k, v): + self.data[k] = v + + def get_data(self, k): + return self.data.get(k, DictObject()) + class ConfiguredServer(AppLogger): def __init__(self, auth_klass=AuthAPI, workdir=None): self.auth_klass = auth_klass @@ -40,7 +51,7 @@ class ConfiguredServer(AppLogger): client_max_size = config.website.client_max_size load_plugins(self.workdir) - self.app = web.Application(client_max_size=client_max_size) + self.app = AHApp(client_max_size=client_max_size) async def init_auth(self): auth = self.auth_klass() diff --git a/ahserver/globalEnv.py b/ahserver/globalEnv.py index 1853799..59f138c 100755 --- a/ahserver/globalEnv.py +++ b/ahserver/globalEnv.py @@ -147,10 +147,13 @@ def initEnv(): g.visualcoding = visualcoding g.uriop = URIOp g.isNone = isNone + g.len = len + g.print = print g.json = json g.int = int g.str = str g.float = float + g.complex = complex g.type = type g.ArgsConvert = ArgsConvert g.time = time @@ -187,3 +190,78 @@ def initEnv(): g.get_config_value = get_config_value g.get_definition = get_definition +def set_builtins(): + all_f="""abs +aiter +all +anext +any +ascii +bin +bool +breakpoint +bytearray +bytes +callable +chr +classmethod +compile +complex +delattr +dict +dir +divmod +enumerate +eval +exec +filter +float +format +frozenset +getattr +globals +hasattr +hash +help +hex +id +input +int +isinstance +issubclass +iter +len +list +locals +map +max +memoryview +min +next +object +oct +open +ord +pow +print +property +range +repr +reversed +round +set +setattr +slice +sorted +staticmethod +str +sum +super +tuple +type +vars +zip""" + g = ServerEnv() + for l in all_f.aplit('\n'): + if l != '': + g[l] = l diff --git a/ahserver/llm_client.py b/ahserver/llm_client.py index f3d06be..5d5cb63 100644 --- a/ahserver/llm_client.py +++ b/ahserver/llm_client.py @@ -1,3 +1,4 @@ +import re import json from aiohttp import web from appPublic.dictObject import DictObject @@ -6,18 +7,26 @@ from appPublic.argsConvert import ArgsConvert class LlmProxy: def __init__(self, processor, desc): - assert desc.get('data') assert desc.name + self.name = desc.name self.processor = processor self.auth_api = desc.auth self.desc = desc - self.data = desc.data self.api_name = desc.name self.ac = ArgsConvert('${', '}') - async def write_chunk(self, l, finished=False): + def line_chunk_match(self, l): + if self.chunk_match: + match = re.search(self.chunk_match, l) + if match: + return match.group(1) + return l + + async def write_chunk(self, l): + l = self.line_chunk_match(l) d = DictObject(** json.loads(l)) content = d.get_data_by_keys(self.api.content_keys) + finished = d.get_data_by_keys(self.api.finished_keys) tokens = {} if finished: tokens = d.get_data_by_keys(self.api.tokens_keys) @@ -39,17 +48,69 @@ class LlmProxy: self.remain_str = lines[-1] ls = lines[:-1] for l in ls: + if l == '': + continue await self.write_chunk(l) async def get_apikey(self, apiname, user): - return None + f = self.processor.run_ns.get_llm_user_apikey + if f: + # return a DictObject instance + return await f(apiname, user) + raise Exception('get_llm_user_apikey() function not found in ServerEnv') async def do_auth(self, request): - pass + d = self.desc.auth + user = await self.processor.run_ns.get_user() + self.data = self.get_data(self.name, user) + if self.data.authed: + return + self.data = await self.get_apikey(self.name, user) + if self.data is None: + raise Exception(f'user({user}) do not has a apikey for {self.name}') + method = d.get('method', 'POST') + headers = {} + for h in d.get('headers',{}): + headers[h.get('name')] = h.get('value') + mydata = {} + for p in d.data or []: + mydata[p.get('name')] = p.get('value') + myparams = {} + for p in d.params or []: + myparams[p.get('name')] = p.get('value') + url = d.get('url') + params = {} + _params = self.datalize(myparams, params) + _headers = self.datalize(headers, params) + _data = self.datalize(mydata, params) + hc = HttpClient() + resp_data = await hc.request(url, method, response_type=RESPONSE_JSON, + params=_params, + data=None if _data == {} else json.dumps(_data), + headers=_headers) + resp_data = DictObject(**resp_data) + for sd in d.set_data: + self.data[sd.name] = resp_data.get_data_by_keys(sd.field) + self.data.authed = True + self.set_data(self.name, user, self.data) + def data_key(self, apiname, user): + if user is None: + user = 'anoumous' + return apiname + '_a_' + user + + def set_data(self, apiname, user, data): + request = self.processor.run_ns.request + app = request.app + app.set_data(self.data_key(apiname, user), data) + + def get_data(self, apiname, user): + request = self.processor.run_ns.request + app = request.app + return app.get_data(self.data_key(apiname, user)) + async def __call__(self, request, params): mapi = params.mapi - print(f'{params=}') stream = params.stream self.resp = web.StreamResponse() await self.resp.prepare(request) @@ -60,8 +121,9 @@ class LlmProxy: raise Exception(f'{mapi} not defined') d = self.desc[mapi] self.api = d + self.chunk_match = d.chunk_match if self.api.need_auth and self.auth_api: - await self.do_auth() + await self.do_auth(request) assert d.get('url') method = d.get('method', 'POST') headers = {} @@ -85,7 +147,7 @@ class LlmProxy: stream_func=self.stream_handle, headers=_headers) if self.remain_str != '': - await self.write_chunk(self.remain_str, finished=True) + await self.write_chunk(self.remain_str) return self.resp def datalize(self, dic, data={}): @@ -94,10 +156,3 @@ class LlmProxy: s1 = self.ac.convert(dic, mydata) return s1 - def set_data(self, resp_data, data_desc): - for dd in data_desc: - f = dd['field'] - n = dd['name'] - if resp_data.get(f): - self.data[n] = resp_data[f] -