litserve openapi schema 处理简单说明

litserve 在处理openai schema 的时候实际上还是利用了fastapi 的能力，只是litserve 抽象了api 开发的处理，对于openapi request 以及response 的类型基于了动态获取（通过python 的inspect 模块）

参考处理

• server.py

register_endpoints

def register_endpoints(self):self._register_internal_endpoints()for lit_api in self.litapi_connector:# 获取lit_api 几个方法的签名，包含了request 以及response，decode_request 处理request 类型，encode_response 处理response 类型decode_request_signature = inspect.signature(lit_api.decode_request)encode_response_signature = inspect.signature(lit_api.encode_response)request_type = decode_request_signature.parameters["request"].annotationif request_type == decode_request_signature.empty:request_type = Requestresponse_type = encode_response_signature.return_annotationif response_type == encode_response_signature.empty:response_type = Responseself._register_api_endpoints(lit_api, request_type, response_type)

_register_api_endpoints 内部实际就是标准的fastapi 动态注册路由，此处会判断是spec 还是普通的litapi

def _register_api_endpoints(self, lit_api: LitAPI, request_type, response_type):"""Register endpoint routes for the FastAPI app."""self._callback_runner.trigger_event(EventTypes.ON_SERVER_START.value, litserver=self)# Create handlershandler = StreamingRequestHandler(lit_api, self) if lit_api.stream else RegularRequestHandler(lit_api, self)# Create endpoint functionasync def endpoint_handler(request: request_type) -> response_type:return await handler.handle_request(request, request_type)# Register endpointif not lit_api.spec:self.app.add_api_route(lit_api.api_path,endpoint_handler,methods=["POST"],dependencies=[Depends(self.setup_auth())],)# Handle specsself._register_spec_endpoints(lit_api)# Register middlewareself._register_middleware()

说明

litserve 的内部实际上是利用了inspect 模块以及动态路由注册的模式实现openapi schema 处理的，了解此机制还是比较重要的，可以编写更好的方法

参考资料

src/litserve/server.py

https://docs.python.org/zh-cn/3.9/library/inspect.html