From 543bf5a62a25564a4ee190a5c1c5a3f7180fc136 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Tue, 9 Apr 2024 11:57:18 +0800
Subject: [PATCH] bugfic

---
 ahserver/llmProcessor.py |  32 ++++++++++++
 ahserver/llm_client.py   | 103 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 135 insertions(+)
 create mode 100755 ahserver/llmProcessor.py
 create mode 100644 ahserver/llm_client.py

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]
+