ollama_request.utils
1import traceback 2 3from ollama import AsyncClient 4from ollama_request.conventions.definitions import OllamaResponse, OllamaResponseStatus, OllamaRequestType, \ 5 OllamaRequest 6 7 8async def get_response(client: AsyncClient, request: OllamaRequest) -> OllamaResponse: 9 is_error = False 10 match request.request_type: 11 case OllamaRequestType.GENERATE: 12 try: 13 response = await client.generate(**request.to_generate_kwargs()) 14 except Exception as e: 15 response = traceback.format_exc() 16 is_error = True 17 case OllamaRequestType.CHAT: 18 try: 19 response = await client.chat(**request.to_chat_kwargs()) 20 except Exception as e: 21 response = traceback.format_exc() 22 is_error = True 23 case _: 24 response = f"Unknown request type: {request.request_type}" 25 is_error = True 26 return OllamaResponse( 27 request_id=request.request_id, 28 response=response, 29 status=OllamaResponseStatus.ERROR if is_error else OllamaResponseStatus.SUCCESS 30 )
async def
get_response( client: ollama._client.AsyncClient, request: ollama_request.conventions.definitions.OllamaRequest) -> ollama_request.conventions.definitions.OllamaResponse:
9async def get_response(client: AsyncClient, request: OllamaRequest) -> OllamaResponse: 10 is_error = False 11 match request.request_type: 12 case OllamaRequestType.GENERATE: 13 try: 14 response = await client.generate(**request.to_generate_kwargs()) 15 except Exception as e: 16 response = traceback.format_exc() 17 is_error = True 18 case OllamaRequestType.CHAT: 19 try: 20 response = await client.chat(**request.to_chat_kwargs()) 21 except Exception as e: 22 response = traceback.format_exc() 23 is_error = True 24 case _: 25 response = f"Unknown request type: {request.request_type}" 26 is_error = True 27 return OllamaResponse( 28 request_id=request.request_id, 29 response=response, 30 status=OllamaResponseStatus.ERROR if is_error else OllamaResponseStatus.SUCCESS 31 )