import json from aiohttp import web from appPublic.dictObject import DictObject from appPublic.httpclient import HttpClient, RESPONSE_TEXT, RESPONSE_JSON, RESPONSE_BIN,RESPONSE_FILE, RESPONSE_STREAM from appPublic.argsConvert import ArgsConvert class LlmProxy: def __init__(self, processor, desc): assert desc.get('data') assert 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): d = DictObject(** json.loads(l)) content = d.get_data_by_keys(self.api.content_keys) tokens = {} if finished: tokens = d.get_data_by_keys(self.api.tokens_keys) j = { 'content':content, 'finish':False, 'token':tokens } jstr = json.dumps(j) + '\n' bin = jstr.encode('utf-8') await self.resp.write(bin) async def stream_handle(self, chunk): chunk = chunk.decode('utf-8') print(f'{chunk=}') chunk = self.remain_str + chunk lines = chunk.split('\n') self.remain_str = lines[-1] ls = lines[:-1] for l in ls: await self.write_chunk(l) async def get_apikey(self, apiname, user): return None async def do_auth(self, request): pass async def __call__(self, request, params): mapi = params.mapi print(f'{params=}') stream = params.stream self.resp = web.StreamResponse() await self.resp.prepare(request) if stream is None: stream = True self.remain_str = '' if not self.desc[mapi]: raise Exception(f'{mapi} not defined') d = self.desc[mapi] self.api = d if self.api.need_auth and self.auth_api: await self.do_auth() 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) response_type = RESPONSE_STREAM hc = HttpClient() resp_data = await hc.request(url, method, response_type=response_type, params=_params, data=None if _data == {} else json.dumps(_data), stream_func=self.stream_handle, headers=_headers) if self.remain_str != '': await self.write_chunk(self.remain_str, finished=True) return self.resp def datalize(self, dic, data={}): mydata = self.data.copy() mydata.update(data) 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]