diff --git a/ahserver/llm_client.py b/ahserver/llm_client.py index 6024f38..3dffd0c 100644 --- a/ahserver/llm_client.py +++ b/ahserver/llm_client.py @@ -1,10 +1,12 @@ import re import base64 import json -from traceback import print_exc +from traceback import format_exc from aiohttp import web from appPublic.dictObject import DictObject +from appPublic.log import debug, info, exception, error from appPublic.httpclient import HttpClient, RESPONSE_TEXT, RESPONSE_JSON, RESPONSE_BIN,RESPONSE_FILE, RESPONSE_STREAM +from appPublic.registerfunction import RegisterFunction from appPublic.argsConvert import ArgsConvert def encode_imagefile(fn): @@ -55,15 +57,14 @@ class StreamLlmProxy: f = opfuncs.get(op) if f and f(v,v1): j[self.api.chunk_filter.field] = '' - print('filtered j=', j) jstr = json.dumps(j) + '\n' bin = jstr.encode('utf-8') await self.resp.write(bin) await self.resp.drain() except Exception as e: - print(f'Error:Write_chunk(),{l=} error:{e=}') - print_exc() + tb = format_exc() + exception(f'Error:Write_chunk(),{l=} error:{e=}{tb}') async def stream_handle(self, chunk): print('chunk=', chunk) @@ -84,6 +85,18 @@ class StreamLlmProxy: return await f(apiname, self.user) raise Exception('get_llm_user_apikey() function not found in ServerEnv') + async def get_apidata(self, parts, params={}): + ret = {} + for d in parts or []: + v = d['value'] + if params != {}: + v = self.datalize(v, params) + if d.get('convertor'): + rf = RegisterFunction() + v = await rf.exe(d['convertor'], v) + ret[d['name']] = v + return ret + async def do_auth(self, request): d = self.desc.auth self.data = self.get_data(self.name) @@ -92,21 +105,12 @@ class StreamLlmProxy: self.data = await self.get_apikey(self.name) if self.data is None: raise Exception(f'user({self.user}) do not has a apikey for {self.name}') + params = self.data 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') + _headers = self.get_apidata(d.get('headers', []), params) + _data = self.get_apidata(d.get('data', []), params) + _params = self.get_apidata(d.get('params',[]), params) 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, @@ -153,23 +157,18 @@ class StreamLlmProxy: self.data = await self.get_apikey(self.name) assert d.get('url') - method = d.get('method', 'POST') - headers = {} - for h in d.get('headers',{}): - headers[h.get('name')] = h.get('value') - mydata = {} - for p in d.get('data', {}): - mydata[p.get('name')] = p.get('value') - myparams = {} - for p in d.get('params', {}): - myparams[p.get('name')] = p.get('value') url = d.get('url') - _params = self.datalize(myparams, params) - _headers = self.datalize(headers, params) - _data = self.datalize(mydata, params) + method = d.get('method', 'POST') + params1 = self.data + params1.update(params) + params = params1 + method = d.get('method', 'POST') + _headers = self.get_apidata(d.get('headers', []), params) + _data = self.get_apidata(d.get('data', []), params) + _params = self.get_apidata(d.get('params',[]), params) response_type = RESPONSE_STREAM hc = HttpClient() - print(f'{url=},{method=},{_params=},{_data=},{_headers=}') + debug(f'{url=},{method=},{_params=},{_data=},{_headers=}') resp_data = await hc.request(url, method, response_type=response_type, params=_params, data=None if _data == {} else json.dumps(_data), @@ -203,36 +202,33 @@ class SyncLlmProxy(StreamLlmProxy): assert d.get('url') method = d.get('method', 'POST') - headers = {} - for h in d.get('headers',{}): - headers[h.get('name')] = h.get('value') - mydata = {} - for p in d.get('data', {}): - mydata[p.get('name')] = p.get('value') - myparams = {} - for p in d.get('params', {}): - myparams[p.get('name')] = p.get('value') url = d.get('url') - _params = self.datalize(myparams, params) - _headers = self.datalize(headers, params) - _data = self.datalize(mydata, params) + params1 = self.data + params1.update(params) + params = params1 + method = d.get('method', 'POST') + _headers = self.get_apidata(d.get('headers', []), params) + _data = self.get_apidata(d.get('data', []), params) + _params = self.get_apidata(d.get('params',[]), params) response_type = RESPONSE_JSON hc = HttpClient() - print(f'{url=},{method=},{_params=},{_data=},{_headers=}') + debug(f'{url=},{method=},{_params=},{_data=},{_headers=}') resp_data = await hc.request(url, method, response_type=response_type, params=_params, data=None if _data == {} else json.dumps(_data), headers=_headers) - print(f'{resp_data=}') - resp_data = DictObject(resp_data) + debug(f'{resp_data=}') if resp_data is None: return { "status":"Error", "message":f'{mapi} not defined' } + resp_data = DictObject(resp_data) return self.convert_resp(resp_data) def convert_resp(self, resp): + if self.api.resp is None: + return resp j = {} for r in self.api.resp or []: j[r.name] = resp.get_data_by_keys(r.value); @@ -240,8 +236,6 @@ class SyncLlmProxy(StreamLlmProxy): class AsyncLlmProxy(StreamLlmProxy): - pass -class AsyncLlmProxy: async def __call__(self, request, params): self.user = await self.processor.run_ns.get_user() mapi = params.mapi @@ -262,23 +256,18 @@ class AsyncLlmProxy: self.data = await self.get_apikey(self.name) assert d.get('url') - method = d.get('method', 'POST') - headers = {} - for h in d.get('headers',{}): - headers[h.get('name')] = h.get('value') - mydata = {} - for p in d.get('data', {}): - mydata[p.get('name')] = p.get('value') - myparams = {} - for p in d.get('params', {}): - myparams[p.get('name')] = p.get('value') url = d.get('url') - _params = self.datalize(myparams, params) - _headers = self.datalize(headers, params) - _data = self.datalize(mydata, params) + method = d.get('method', 'POST') + params1 = self.data + params1.update(params) + params = params1 + method = d.get('method', 'POST') + _headers = self.get_apidata(d.get('headers', []), params) + _data = self.get_apidata(d.get('data', []), params) + _params = self.get_apidata(d.get('params',[]), params) response_type = RESPONSE_JSON hc = HttpClient() - print(f'{url=},{method=},{_params=},{_data=},{_headers=}') + debug(f'{url=},{method=},{_params=},{_data=},{_headers=}') resp_data = await hc.request(url, method, response_type=response_type, params=_params, data=None if _data == {} else json.dumps(_data), diff --git a/requirements.txt b/requirements.txt index f6a16e0..6353db8 100755 --- a/requirements.txt +++ b/requirements.txt @@ -2,7 +2,7 @@ asyncio aiofiles aiodns # cchardet -aiohttp +aiohttp==3.10.10 aiohttp_session aiohttp_auth_autz aiohttp-cors