代碼實現

下面使用<a href="https://mp.weixin.qq.com/s/zjejh78IHXx3EJJtvnFJVg">FastAPI</a>的同步和異步兩種方式實現http請求轉發。  

其中services是在本網關中定義的服務名稱，path是服務的路徑，網關將把這個路徑轉發到后端服務。  

例如：前端使用URL:http://127.0.0.1:8000/translation/trans/v1訪問本網關時，@api_route將會把server的值填充為translation，將path的值填充為trans/v1，這樣API網關調用的后端服務的API地址為：http://127.0.0.1:5001/trans/v1。

同步轉發

from fastapi import FastAPI, Request,HTTPException

import requests



# 創建一個FastAPI實例

app = FastAPI()



# 定義服務

services = {

    "translation": "http://127.0.0.1:5001",

    # 可以在這里添加其它服務地址

}

'''

services的key是服務名稱，客戶端在請求時傳入服務名稱，本網關再根據服務名稱找到對應的服務地址

'''



# 接收客戶端請求并轉發到后端服務

@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])

async def gateway(service: str, path: str, request: Request):

    '''

    !注意：網關并未將header轉發給后端服務，這樣比較簡單。

    '''



    if service not in services:

        raise HTTPException(status_code=401, detail="未找到該服務")



    #  根據服務名稱找到對應的服務地址

    service_url = services[service]

    body = await request.json() if request.method in ["POST", "PUT", "PATCH"] else None   

    url = f"{service_url}/{path}"



    # 同步調用，這里會阻塞

    response = requests.post(url, json = body)

    return response.json()    



# 啟動網關

if __name__ == "__main__":

    import uvicorn



    # 交互式API文檔地址：

    # http://127.0.0.1:8000/docs/ 

    # http://127.0.0.1:8000/redoc/

    uvicorn.run(app, host="0.0.0.0", port=8000)

異步轉發

from fastapi import FastAPI, Request,HTTPException

import httpx



#  定義超時時間，單位：秒

time_out = 30



# 創建一個FastAPI實例

app = FastAPI()



# 定義服務

services = {

    "translation": "http://127.0.0.1:5001",

    # 可以在這里添加其它服務地址

}

'''

services的key是服務名稱，客戶端在請求時傳入服務名稱，本網關再根據服務名稱找到對應的服務地址

'''



# 接收客戶端請求并轉發到后端服務

@app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])

async def gateway(service: str, path: str, request: Request):

    '''

    !注意：網關并未將header轉發給后端服務，這樣比較簡單。

    '''



    if service not in services:

        raise HTTPException(status_code=401, detail="未找到該服務")



    #headers = dict(request.headers)



    # 從客戶端請求中獲取數據

    client_request_data = await request.json()



    service_url = services[service]

    url = f"{service_url}/{path}"   



    # 使用 httpx 將請求轉發到后端服務，非阻塞，不過在我的配置一般的開發機上沒有發現和阻塞式調用在性能上有多少區別。

    async with httpx.AsyncClient() as client:

        '''

        !注意：httpx.AsyncClient默認的timeout為5秒，在調用基于大模型的后端服務時經常超時，所以這里設置超時時間為30秒

        '''

        response = await client.post(url=url, json=client_request_data,timeout=time_out)

        #print(response)

        return response.json()



# 啟動網關

if __name__ == "__main__":

    import uvicorn



    # 交互式API文檔地址：

    # http://127.0.0.1:8000/docs/ 

    # http://127.0.0.1:8000/redoc/

    uvicorn.run(app, host="0.0.0.0", port=8000)

按理說在訪問量大的時候，后面的異步轉發性能應該更好一些。不過在我的開發機簡單測試時，發現在訪問量不大的情況下，二者的性能差別并不明顯。

測試網關

啟動翻譯服務

在我的開發環境中，可以用命令行啟動：

# 激活venv

.venv\Scripts\activate



# 啟動翻譯服務

python "06-2.langchain api with fastapi.py"

翻譯服務的代碼可以去本文末尾的gitee或者github中下載。

啟動網關

# 激活venv

.venv\Scripts\activate



# 啟動網關服務

python "07-1.fastapi gateway.py"

測試網關??

使用Apifox發送請求。

下載源代碼

-[gitee](https://gitee.com/liupras/langchain-llama3-Chroma-RAG-demo)

-[github](https://github.com/liupras/langchain-llama3-Chroma-RAG-demo)

文章轉自微信公眾號@AI很有趣

上一篇：

使用FastAPI和langchain做本地大模型的API

下一篇：

手把手教你用Python和Flask創建REST API
#你可能也喜歡這些API文章!

我們有何不同？

API服務商零注冊

多API并行試用

數據驅動選型，提升決策效率

查看全部API→
??

熱門場景實測，選對API

#AI文本生成大模型API

對比大模型API的內容創意新穎性、情感共鳴力、商業轉化潛力

25個渠道
一鍵對比試用API 限時免費

#AI深度推理大模型API

對比大模型API的邏輯推理準確性、分析深度、可視化建議合理性

10個渠道
一鍵對比試用API 限時免費