开源模型应用落地-FastAPI-助力模型交互-WebSocket篇(六)

 一、前言

    使用 FastAPI 可以帮助我们更简单高效地部署 AI 交互业务。FastAPI 提供了快速构建 API 的能力,开发者可以轻松地定义模型需要的输入和输出格式,并编写好相应的业务逻辑。

    FastAPI 的异步高性能架构,可以有效支持大量并发的预测请求,为用户提供流畅的交互体验。此外,FastAPI 还提供了容器化部署能力,开发者可以轻松打包 AI 模型为 Docker 镜像,实现跨环境的部署和扩展。

    总之,使用 FastAPI 可以大大提高 AI 应用程序的开发效率和用户体验,为 AI 模型的部署和交互提供全方位的支持。

    本篇在开源模型应用落地-FastAPI-助力模型交互-WebSocket篇(五)基础上,学习如何集成Tool获取实时数据,并以流式方式返回


二、术语

2.1.Tool

    Tool(工具)是为了增强其语言模型的功能和实用性而设计的一系列辅助手段,用于扩展模型的能力。例如代码解释器(Code Interpreter)和知识检索(Knowledge Retrieval)等都属于其工具。

2.2.langchain预置的tools

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

   基本这些工具能满足大部分需求,具体使用参见:

2.3.LangChain支持流式输出的方法

  • stream:基本的流式传输方式,能逐步给出代理的动作和观察结果。
  • astream:异步的流式传输,用于异步处理需求的情况。
  • astream_events:更细致的流式传输,能流式传输代理的每个具体事件,如工具调用和结束、模型启动和结束等,便于深入了解和监控代理执行的详细过程。

2.4.langchainhub

    是 LangChain 相关工具的集合中心,其作用在于方便开发者发现和共享常用的提示(Prompt)、链、代理等。

    它受 Hugging Face Hub 启发,促进社区交流与协作,推动 LangChain 生态发展。当前,它在新架构中被置于 LangSmith 里,主要聚焦于 Prompt。

2.5.asyncio

    是一个用于编写并发代码的标准库,它提供了构建异步应用程序的基础框架。


三、前置条件

3.1. 创建虚拟环境&安装依赖

  增加Google Search以及langchainhub的依赖包

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 langchainhub

3.2. 注册Google Search API账号

参见:开源模型应用落地-FastAPI-助力模型交互-WebSocket篇(五)

3.3. 生成Google Search API的KEY


四、技术实现

4.1. 使用Tool&流式输出

# -*- coding: utf-8 -*-
import asyncio
import os
from langchain.agents import  create_structured_chat_agent, AgentExecutor
from langchain_community.utilities.serpapi import SerpAPIWrapper
from langchain_core.prompts import SystemMessagePromptTemplate, HumanMessagePromptTemplate, ChatPromptTemplate
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"] = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"llm = ChatOpenAI(model="gpt-3.5-turbo",temperature=0,max_tokens=512)@tool
def search(query:str):"""只有需要了解实时信息或不知道的事情的时候才会使用这个工具,需要传入要搜索的内容。"""serp = SerpAPIWrapper()result = serp.run(query)print("实时搜索结果:", result)return resulttools = [search]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])print(prompt)agent = create_structured_chat_agent(llm, tools, prompt
)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)async def chat(params):events = agent_executor.astream_events(params,version="v2")async for event in events:type = event['event']if 'on_chat_model_stream' == type:data = event['data']chunk =  data['chunk']content =  chunk.contentif content and len(content) > 0:print(content)asyncio.run(chat({"input": "广州现在天气如何?"}))

调用结果:

说明:

流式输出的数据结构为:

{'event': 'on_chat_model_stream', 'data': {'chunk': AIMessageChunk(content='天', id='run-92515b63-4b86-4af8-8515-2f84def9dfab')}, 'run_id': '92515b63-4b86-4af8-8515-2f84def9dfab', 'name': 'ChatOpenAI', 'tags': ['seq:step:3'], 'metadata': {'ls_provider': 'openai', 'ls_model_name': 'gpt-3.5-turbo', 'ls_model_type': 'chat', 'ls_temperature': 0.0, 'ls_max_tokens': 512, 'ls_stop': ['\nObservation']}}
type: on_chat_model_stream
{'event': 'on_chat_model_stream', 'data': {'chunk': AIMessageChunk(content='气', id='run-92515b63-4b86-4af8-8515-2f84def9dfab')}, 'run_id': '92515b63-4b86-4af8-8515-2f84def9dfab', 'name': 'ChatOpenAI', 'tags': ['seq:step:3'], 'metadata': {'ls_provider': 'openai', 'ls_model_name': 'gpt-3.5-turbo', 'ls_model_type': 'chat', 'ls_temperature': 0.0, 'ls_max_tokens': 512, 'ls_stop': ['\nObservation']}}

4.2. 通过langchainhub使用公共prompt

   在4.1使用Tool&流式输出的代码基础上进行调整

# -*- coding: utf-8 -*-
import asyncio
import os
from langchain.agents import  create_structured_chat_agent, AgentExecutor
from langchain_community.utilities.serpapi import SerpAPIWrapper
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"] = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"from langchain import hubllm = ChatOpenAI(model="gpt-3.5-turbo",temperature=0,max_tokens=512)@tool
def search(query:str):"""只有需要了解实时信息或不知道的事情的时候才会使用这个工具,需要传入要搜索的内容。"""serp = SerpAPIWrapper()result = serp.run(query)print("实时搜索结果:", result)return resulttools = [search]prompt = hub.pull("hwchase17/structured-chat-agent")print(prompt)agent = create_structured_chat_agent(llm, tools, prompt
)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)async def chat(params):events = agent_executor.astream_events(params,version="v2")async for event in events:type = event['event']if 'on_chat_model_stream' == type:data = event['data']chunk =  data['chunk']content =  chunk.contentif content and len(content) > 0:print(content)asyncio.run(chat({"input": "广州现在天气如何?"}))

调用结果:

4.3. 整合代码

在开源模型应用落地-FastAPI-助力模型交互-WebSocket篇(五)的代码基础上进行调整

import uvicorn
import osfrom typing import Annotated
from fastapi import (Depends,FastAPI,WebSocket,WebSocketException,WebSocketDisconnect,status,
)
from langchain import hub
from langchain.agents import create_structured_chat_agent, AgentExecutor
from langchain_community.utilities import SerpAPIWrapperfrom langchain_core.tools import tool
from langchain_openai import ChatOpenAIos.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'  # 你的Open AI Key
os.environ["SERPAPI_API_KEY"] = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"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():prompt = hub.pull("hwchase17/structured-chat-agent")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)events = agent_executor.astream_events({"input": query}, version="v1")async for event in events:type = event['event']if 'on_chat_model_stream' == type:data = event['data']chunk = data['chunk']content = chunk.contentif content and len(content) > 0:print(content)yield content@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>

调用结果:

用户输入:你好

不需要触发工具调用

模型输出:

用户输入:广州现在天气如何?

需要调用工具

模型输出:

```
Action:
```
{"action": "Final Answer","action_input": "广州现在的天气是多云,温度为87华氏度,降水概率为7%,湿度为76%,风力为7英里/小时。"
}
```

PS:

1. 上面仅用于演示流式输出的效果,里面包含一些冗余的信息,例如:"action": "Final Answer",要根据实际情况过滤。

2. 页面输出的样式可以根据实际需要进行调整,此处仅用于演示效果。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/367000.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【JS】纯web端使用ffmpeg实现的视频编辑器-视频合并

纯前端实现的视频合并 接上篇ffmpeg文章 【JS】纯web端使用ffmpeg实现的视频编辑器 这次主要添加了一个函数&#xff0c;实现了视频合并的操作。 static mergeArgs(timelineList) {const cmd []console.log(时间轴数据,timelineList)console.log("文件1",this.readD…

MatLab三维图形绘制基础

三维图形绘制 三维曲线 plot3 螺旋图绘制 % %三维图像:螺旋图绘制 clear; clc; t [0:0.1:10*pi];% 向量 x sin(t) t.*cos(t);%t是向量&#xff0c;用点乘 y cos(t) - t.*sin(t); z t; plot3(x,y,z); grid on;plot3 绘制同型矩阵 %% % plot3绘制同型矩阵 t [0:0.1:10*…

游戏AI的创造思路-技术基础-tanh函数详解

又来搞事情&#xff0c;总想着把sigmoid函数替换成其他函数作为激活函数&#xff0c;或者找到更合适某一段训练的函数&#xff0c;所以今天来聊聊tanh函数&#xff08;谁让咱当年差点去了数学系&#xff0c;结果还是在数学系转过去计算机的&#xff09; 目录 3.9. tanh函数详解…

【Rust基础入门】Hello Cargo

文章目录 前言Cargo是什么&#xff1f;Cargo的作用查看cargo版本使用cargo创建项目Cargo.toml文件cargo build命令cargo runcargo check为发布构建 总结 前言 在Rust编程中&#xff0c;Cargo扮演着至关重要的角色。它是Rust的包管理器&#xff0c;负责处理许多任务&#xff0c…

uniapp入门

一、新建项目 进入到主界面&#xff0c;左上角点击新建——1.项目 输入项目名称&#xff0c;Vue版本选择3 二、创建页面 选中左侧文件目录里的pages文件夹&#xff0c;右键&#xff0c;选择新建页面 1输入名称 2选中“创建同名目录” 3选择模板&…

昇思25天学习打卡营第7天|网络构建

网络构建 神经网络模型由tensor操作和神经网络层构成。 MIndSporezhong&#xff0c;Cell是构建所有网络的基类&#xff0c;也是网络的基本单元。cell也由子cell构成。 定义模型类 # 继承nn.Cell类 class Network(nn.Cell):def __init__(self):super().__init__()self.flatte…

ElasticSearch 和 MySQL的区别

MySQLElasticSearch 数据库&#xff08;database&#xff09;索引&#xff08;index&#xff09;数据表&#xff08;table&#xff09; 类型&#xff08;type&#xff09; 记录文档&#xff08;document&#xff0c;json格式&#xff09; 一、ES基础命令 1. ES cat查询命令 2.…

分布式技术专题 | TCP在分布式网络中的通信机制与底层实现

深入解析分布式网络中的TCP通信协议实现 跨地域通信与资源共享网络节点与主机的定义网络技术通信机制TCP/IP协议模型TCP/IP分层机制TCP的Socket链接处理控制TCP的优势和特性自动差错控制正确性和有序性 TCP的Socket使用端口在应用程序间通信TCP的Socket使用端口套接字操作 跨地…

uniapp封装虚拟列表滚动组件

uniapp封装虚拟列表滚动组件 这里用到一个列表&#xff0c;然后数据可能有很多很多…&#xff0c;一次性全部渲染到dom上会卡顿&#xff0c;很废性能&#xff0c;于是用了这个虚拟列表就变丝滑很多很多。 组件mosoweInventedList 代码&#xff1a; <!-- 虚拟滚动列表组件&a…

GPT-4o首次引入!全新图像自动评估基准发布!

目录 01 什么是DreamBench&#xff1f; 02 与人类对齐的自动化评估 03 更全面的个性化数据集 04 实验结果 面对层出不穷的个性化图像生成技术&#xff0c;一个新问题摆在眼前&#xff1a;缺乏统一标准来衡量这些生成的图片是否符合人们的喜好。 对此&#xff0c;来自清华大…

mysql主键自增连续新增时报错主键重复-Duplicate entry “x” for key PRIMARY

mysql主键自增连续新增时报错主键重复 1、mysql数据库设置数据库主键自增的规律 id -- AUTO_INCREMENT2、可视化工具查看自增没问题 3、问题描述 新增第一个时操作成功&#xff0c;新增第二个时候操作失败报错&#xff1a; Duplicate entry “x” for key PRIMARY4、分析&a…

千益畅行,旅游卡,如何赚钱?

​ 赚钱这件事情&#xff0c;只有自己努力执行才会有结果。生活中没有幸运二字&#xff0c;每个光鲜亮丽的背后&#xff0c;都是不为人知的付出&#xff01; #旅游卡服务#

简过网:考公务员难度大吗,一般考几年才上岸!

先调查一下&#xff0c;大家考公务员都是几年才上岸的&#xff1f; 最近有网友私信小编&#xff0c;普通人考公要准备多久才能上岸&#xff0c;其实对于多久能上岸这个问题&#xff0c;并没有一个准确的数字&#xff0c;众所周知&#xff0c;公务员考试是有一定难度的&#xf…

Git入门 本地仓库 远端仓库 多分支

Git入门 Git入门本地git初始化git仓库初始化 创建远端仓库githubgitee 指定远端仓库推送至远端多分支将feature分支合并至dev分支 其他开发者 Git入门 本地git初始化 git仓库初始化 mkdir myrepo # 创建仓库文件夹 cd myrepo/ # 进入目录 git init # 初始化git仓库 (创建.g…

C - Popcorn(abs358)

题意&#xff1a;有n个摊子&#xff0c;m个爆米花&#xff0c;想花费最少去的店铺买到所有的口味的爆米花&#xff0c;找到每一列都为‘o’的最少行数。 分析&#xff1a;用dfs寻找最少路径 #include<bits/stdc.h> using namespace std; typedef long long ll; char x;…

CAN通信波形【示波器抓取】

在测试bms系统过程中&#xff0c;在上位机发现无法读取CAN通信&#xff0c;尝试使用示波器抓取CAN通信波形&#xff0c;&#xff0c;去确定CAN通信是否正常。 做一想要从车上测出can总线上的数据还不太容易。 于是我首先使用示波器&#xff08;我使用的示波器型号是TDS 220&am…

第十四届蓝桥杯省赛C++B组E题【接龙数列】题解(AC)

需求分析 题目要求最少删掉多少个数后&#xff0c;使得数列变为接龙数列。 相当于题目要求求出数组中的最长接龙子序列。 题目分析 对于一个数能不能放到接龙数列中&#xff0c;只关系到这个数的第一位和最后一位&#xff0c;所以我们可以先对数组进行预处理&#xff0c;将…

opencascade AIS_InteractiveContext源码学习7 debug visualization

AIS_InteractiveContext 前言 交互上下文&#xff08;Interactive Context&#xff09;允许您在一个或多个视图器中管理交互对象的图形行为和选择。类方法使这一操作非常透明。需要记住的是&#xff0c;对于已经被交互上下文识别的交互对象&#xff0c;必须使用上下文方法进行…

JVM专题十一:JVM 中的收集器一

上一篇JVM专题十&#xff1a;JVM中的垃圾回收机制专题中&#xff0c;我们主要介绍了Java的垃圾机制&#xff0c;包括垃圾回收基本概念&#xff0c;重点介绍了垃圾回收机制中自动内存管理与垃圾收集算法。如果说收集算法是内存回收的方法论&#xff0c;那么垃圾收集器就是内存回…

OBS 免费的录屏软件

一、下载 obs 【OBS】OBS Studio 的安装、参数设置和录屏、摄像头使用教程-CSDN博客 二、使用 obs & 输出无黑屏 【OBS任意指定区域录屏的方法-哔哩哔哩】 https://b23.tv/aM0hj8A OBS任意指定区域录屏的方法_哔哩哔哩_bilibili 步骤&#xff1a; 1&#xff09;获取区域…