diff --git a/ahserver/llmProcessor.py b/ahserver/llmProcessor.py new file mode 100755 index 0000000..0456b33 --- /dev/null +++ b/ahserver/llmProcessor.py @@ -0,0 +1,32 @@ +import aiohttp +from aiohttp import web, BasicAuth +from aiohttp import client +from appPublic.dictObject import DictObject +from .llm_client import LlmProxy +from .baseProcessor import * + +class LlmProcessor(BaseProcessor): + @classmethod + def isMe(self,name): + return name=='llm' + + async def path_call(self, request, params={}): + await self.set_run_env(request) + path = self.path + url = self.resource.entireUrl(request, path) + ns = self.run_ns + ns.update(params) + te = self.run_ns['tmpl_engine'] + txt = await te.render(url,**ns) + data = json.loads(txt) + return DictObject(**data) + + async def datahandle(self,request): + chunk_size = 40960 + d = await self.path_call(request) + llm = LlmProxy(self, d) + self.retResponse = await llm(request, self.run_ns.params_kw) + + def setheaders(self): + pass + diff --git a/ahserver/llm_client.py b/ahserver/llm_client.py new file mode 100644 index 0000000..f3d06be --- /dev/null +++ b/ahserver/llm_client.py @@ -0,0 +1,103 @@ +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] +