開源模型應用落地-FastAPI-助力模型交互-WebSocket篇（五）

一、前言

? ? 使用 FastAPI 可以幫助我們更簡單高效地部署 AI 交互業務。FastAPI 提供了快速構建 API 的能力,開發者可以輕松地定義模型需要的輸入和輸出格式,并編寫好相應的業務邏輯。

? ? FastAPI 的異步高性能架構,可以有效支持大量并發的預測請求,為用戶提供流暢的交互體驗。此外,FastAPI 還提供了容器化部署能力,開發者可以輕松打包 AI 模型為 Docker 鏡像,實現跨環境的部署和擴展。

? ? 總之,使用 FastAPI 可以大大提高 AI 應用程序的開發效率和用戶體驗,為 AI 模型的部署和交互提供全方位的支持。

? ? LangChain基礎入門：開源模型應用落地-FastAPI-助力模型交互-WebSocket篇（一），本篇學習如何集成LangChain進行模型交互，并使用工具獲取實時信息

二、術語

2.1.FastAPI

? ? FastAPI 是一個用于構建 API 的現代、快速(高性能)的 Python Web 框架。它是基于標準 Python 類型注釋的 ASGI (Asynchronous Server Gateway Interface) 框架。

FastAPI 具有以下主要特點:

快速: FastAPI 使用 ASGI 服務器和 Starlette 框架,在性能測試中表現出色。它可以與 Uvicorn 一起使用,提供非常高的性能。
簡單: FastAPI 利用 Python 類型注釋,使 API 定義變得簡單且直觀。開發人員只需要定義輸入和輸出模型,FastAPI 會自動生成 API 文檔。
現代: FastAPI 支持 OpenAPI 標準,可以自動生成 API 文檔和交互式文檔。它還支持 JSON Schema 和數據驗證。
全功能: FastAPI 提供了路由、依賴注入、數據驗證、安全性、測試等功能,是一個功能齊全的 Web 框架。
可擴展: FastAPI 被設計為可擴展的。開發人員可以輕松地集成其他庫和組件,如數據庫、身份驗證等。

2.2.WebSocket

? ? 是一種計算機通信協議,它提供了在單個 TCP 連接上進行全雙工通信的機制。它是 HTML5 一個重要的組成部分。

WebSocket 協議主要有以下特點:

全雙工通信：WebSocket 允許客戶端和服務器之間進行雙向實時通信,即數據可以同時在兩個方向上流動。這與傳統的 HTTP 請求-響應模型不同,HTTP 中數據只能單向流動。
持久性連接：WebSocket 連接是一種持久性的連接,一旦建立就會一直保持,直到客戶端或服務器主動關閉連接。這與 HTTP 的連接是短暫的不同。
低開銷：相比 HTTP 請求-響應模型,WebSocket 在建立連接時需要較少的數據交換,因此網絡開銷較小。
實時性：由于 WebSocket 連接是持久性的,且數據可以雙向流動,因此 WebSocket 非常適用于需要實時、低延遲數據交互的應用場景,如聊天應用、實時游戲、股票行情等。

2.3.Tool

? ? Tool（工具）是為了增強其語言模型的功能和實用性而設計的一系列輔助手段，用于擴展模型的能力。例如代碼解釋器（Code Interpreter）和知識檢索（Knowledge Retrieval）等都屬于其工具。

2.4.langchain預置的tools

? ? https://github.com/langchain-ai/langchain/tree/v0.1.16/docs/docs/integrations/tools

? ?基本這些工具能滿足大部分需求，具體使用參見：

三、前置條件

3.1. 創建虛擬環境&安裝依賴

? 增加Google Search的依賴包

conda create -n fastapi_test python=3.10
conda activate fastapi_test
pip install fastapi websockets uvicorn
pip install --quiet  langchain-core langchain-community langchain-openai
pip install google-search-results

3.2. 注冊Google Search API賬號

1. 輸入注冊信息

可以使用Google賬號登錄，但仍要執行下面的認證操作

2. 需要認證郵箱

3. 需要認證手機

使用接碼平臺：https://sms-activate.io/cn

選擇Google服務

注意

1. 購買的虛擬號碼要與IP（科學上網）所在地一致，此處選擇美國

2. 使用支付寶充值，最低需要充值2美金

4. 認證成功

3.3. 生成Google Search API的KEY

四、技術實現

4.1. Google Search小試

# -*- coding: utf-8 -*-
import osfrom langchain_community.utilities.serpapi import SerpAPIWrapperos.environ["SERPAPI_API_KEY"] = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
serp = SerpAPIWrapper()
result = serp.run("廣州的實時氣溫如何？")
print("實時搜索結果:", result)

調用結果：

4.2. 非流式輸出

? ? 本章代碼將開源模型應用落地-FastAPI-助力模型交互-WebSocket篇（三）基礎上進行拓展

服務端：

import uvicorn
import osfrom typing import Annotated
from fastapi import (Depends,FastAPI,WebSocket,WebSocketException,WebSocketDisconnect,status,
)
from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain_community.utilities import SerpAPIWrapperfrom langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain_core.tools import tool
from langchain_openai import ChatOpenAIos.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'    #你的Open AI Key
os.environ["SERPAPI_API_KEY"] = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"class ConnectionManager:def __init__(self):self.active_connections: list[WebSocket] = []async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)async def send_personal_message(self, message: str, websocket: WebSocket):await websocket.send_text(message)async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager = ConnectionManager()app = FastAPI()async def authenticate(websocket: WebSocket,userid: str,secret: str,
):if userid is None or secret is None:raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)print(f'userid: {userid},secret: {secret}')if '12345' == userid and 'xxxxxxxxxxxxxxxxxxxxxxxxxx' == secret:return 'pass'else:return 'fail'@tool
def search(query:str):"""只有需要了解實時信息或不知道的事情的時候才會使用這個工具，需要傳入要搜索的內容。"""serp = SerpAPIWrapper()result = serp.run(query)print("實時搜索結果:", result)return resultdef get_prompt():template='''Respond to the human as helpfully and accurately as possible. You have access to the following tools:{tools}Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).Valid "action" values: "Final Answer" or {tool_names}Provide only ONE action per $JSON_BLOB, as shown:```{{"action": $TOOL_NAME,"action_input": $INPUT}}```Follow this format:Question: input question to answerThought: consider previous and subsequent stepsAction:```$JSON_BLOB```Observation: action result... (repeat Thought/Action/Observation N times)Thought: I know what to respondAction:```{{"action": "Final Answer","action_input": "Final response to human"}}Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Format is Action:```$JSON_BLOB```then Observation'''system_message_prompt = SystemMessagePromptTemplate.from_template(template)human_template='''{input}{agent_scratchpad}(reminder to respond in a JSON blob no matter what)'''human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])return promptasync def chat(query):global llm,toolsagent = create_structured_chat_agent(llm, tools, get_prompt())agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)result = agent_executor.invoke({"input": query})print(result['output'])yield result['output']@app.websocket("/ws")
async def websocket_endpoint(*,websocket: WebSocket,userid: str,permission: Annotated[str, Depends(authenticate)],):await manager.connect(websocket)try:while True:text = await websocket.receive_text()if 'fail' == permission:await manager.send_personal_message(f"authentication failed", websocket)else:if text is not None and len(text) > 0:async for msg in chat(text):await manager.send_personal_message(msg, websocket)except WebSocketDisconnect:manager.disconnect(websocket)print(f"Client #{userid} left the chat")await manager.broadcast(f"Client #{userid} left the chat")if __name__ == '__main__':tools = [search]llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, max_tokens=512)uvicorn.run(app, host='0.0.0.0',port=7777)

客戶端：

<!DOCTYPE html>
<html><head><title>Chat</title></head><body><h1>WebSocket Chat</h1><form action="" onsubmit="sendMessage(event)"><label>USERID: <input type="text" id="userid" autocomplete="off" value="12345"/></label><label>SECRET: <input type="text" id="secret" autocomplete="off" value="xxxxxxxxxxxxxxxxxxxxxxxxxx"/></label><br/><button onclick="connect(event)">Connect</button><hr><label>Message: <input type="text" id="messageText" autocomplete="off"/></label><button>Send</button></form><ul id='messages'></ul><script>var ws = null;function connect(event) {var userid = document.getElementById("userid")var secret = document.getElementById("secret")ws = new WebSocket("ws://localhost:7777/ws?userid="+userid.value+"&secret=" + secret.value);ws.onmessage = function(event) {var messages = document.getElementById('messages')var message = document.createElement('li')var content = document.createTextNode(event.data)message.appendChild(content)messages.appendChild(message)};event.preventDefault()}function sendMessage(event) {var input = document.getElementById("messageText")ws.send(input.value)input.value = ''event.preventDefault()}</script></body>
</html>

調用結果：

用戶輸入：你好

不需要觸發工具調用

模型輸出：你好！有什么我可以幫忙的嗎？

用戶輸入：廣州現在天氣如何？

需要調用工具

模型輸出：The current weather in Guangzhou is partly cloudy with a temperature of 95°F, 66% chance of precipitation, 58% humidity, and wind speed of 16 mph. This information was last updated on Monday at 1:00 PM.

?

PS:

1. 在AI交互中，LangChain框架并不是必須引入，此處引用僅用于簡化Openai的交互流程。

2. 頁面輸出的樣式可以根據實際需要進行調整，此處僅用于演示效果。

3. 目前還遺留兩個問題，一是如何實現流式輸出，二是如何更好維護prompt模版，篇幅有限，下回分解

五、附帶說明

5.1. 如何避免模型用英文回復

在提示詞模版加入：Remember to answer in Chinese.? 暗示模型一定要以中文進行回復。

修改后的提示語為：

Respond to the human as helpfully and accurately as possible. You have access to the following tools:{tools}Use a json blob to specify a tool by providing an action key (tool name) and an action_input key (tool input).Valid "action" values: "Final Answer" or {tool_names}Provide only ONE action per $JSON_BLOB, as shown:```{{"action": $TOOL_NAME,"action_input": $INPUT}}```Follow this format:Question: input question to answerThought: consider previous and subsequent stepsAction:```$JSON_BLOB```Observation: action result... (repeat Thought/Action/Observation N times)Thought: I know what to respondAction:```{{"action": "Final Answer","action_input": "Final response to human"}}Begin! Reminder to ALWAYS respond with a valid json blob of a single action. Use tools if necessary. Respond directly if appropriate. Remember to answer in Chinese.Format is Action:```$JSON_BLOB```then Observation