三分钟快速搭建基于FastAPI的AI Agent应用!

本文涉及的产品
视觉智能开放平台,视频资源包5000点
视觉智能开放平台,分割抠图1万点
NLP自然语言处理_基础版,每接口每天50万次
简介: 【10月更文挑战第1天】
fastapi==0.108.0
langchain_core==0.1.28
langchain_openai == 0.0.5
langchain_community==0.0.25
langchain==0.1.10
redis==7.2.0
qdrant_client == 1.7.1
uvicorn==0.23.2
pip install -r requirements.txt

想检查某依赖是否安装完毕:

pip show fastapi

那就先引入 fastapi。

# 这是一个使用 FastAPI 框架编写的简单应用程序的示例。
# 导入FastAPI模块
from fastapi import FastAPI

# 创建一个FastAPI应用实例
app = FastAPI()


# 定义一个路由,当访问'/'时会被触发
@app.get("/")
# 定义一个函数,返回一个字典,key为"Hello",value为"World"
def read_root():
    return {
   "Hello": "World"}


# 如果主程序为 __main__,则启动服务器
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8090)

如何运行呢?

直接点击它:

直达API文档

新增一个 chat 接口:

# 这是一个使用 FastAPI 框架编写的简单应用程序的示例。
# 导入FastAPI模块
from fastapi import FastAPI, BackgroundTasks

# 创建一个FastAPI应用实例
app = FastAPI()


# 定义一个路由,当访问'/'时会被触发
@app.get("/")
# 定义一个函数,返回一个字典,key为"Hello",value为"World"
def read_root():
    return {
   "Hello": "World"}


@app.post("/chat")
def chat():
    return {
   "response": "I am a chat bot!"}


# 如果主程序为 __main__,则启动服务器
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8090)

API文档立即更新:

同理,我们编写ws函数:

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        print("Connection closed")
        await websocket.close()

使用 postman 构造 websocket 请求:

先点击 connect,再输入要发送的消息:你好。点击 send 即请求,响应了你好!

完整代码

# 这是一个使用 FastAPI 框架编写的简单应用程序的示例。
# 导入FastAPI模块
import os

from dotenv import load_dotenv, find_dotenv
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, BackgroundTasks
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor, tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import StrOutputParser
from langchain.memory import ConversationTokenBufferMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
import os
import asyncio
import uuid
from langchain_community.vectorstores import Qdrant
from qdrant_client import QdrantClient
from Mytools import *

# 设置 API 密钥
DASHSCOPE_API_KEY = "xxx"
load_dotenv(find_dotenv())
os.environ["DASHSCOPE_API_KEY"] = DASHSCOPE_API_KEY
os.environ["SERPAPI_API_KEY"] = "xxx"

# 创建一个FastAPI应用实例
app = FastAPI()


# 定义一个工具函数
@tool
def test():
    """ Test tool"""""
    return "test"


# 定义一个Master类
class Master:
    def __init__(self):
        # 初始化ChatOpenAI模型
        self.chatmodel = ChatOpenAI(
            api_key=os.getenv("DASHSCOPE_API_KEY"),
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
            model="qwen-plus",
            temperature=0,
            streaming=True,
        )
        # 设置记忆存储键名
        self.MEMORY_KEY = "chat_history"
        # 初始化系统提示模板
        self.SYSTEMPL = ""
        self.prompt = ChatPromptTemplate.from_messages(
            [
                (
                    "system",
                    "你是一个助手"
                ),
                (
                    "user",
                    "{input}"
                ),
                MessagesPlaceholder(variable_name="agent_scratchpad"),
            ],
        )
        # 初始化记忆存储
        self.memory = ""
        # 初始化工具列表
        tools = [test]
        # 创建OpenAI工具代理
        agent = create_openai_tools_agent(
            self.chatmodel,
            tools=tools,
            prompt=self.prompt,
        )
        # 创建代理执行器
        self.agent_executor = AgentExecutor(
            agent=agent,
            tools=tools,
            verbose=True,
        )

    # 定义运行方法
    def run(self, query):
        # 调用代理执行器并获取结果
        result = self.agent_executor.invoke({
   "input": query})
        # 返回执行器的响应
        return result


# 定义根路由
@app.get("/")
# 定义根路由处理函数,返回一个包含"Hello"和"World"的字典
def read_root():
    return {
   "Hello": "World"}


# 定义聊天路由
@app.post("/chat")
# 定义聊天路由处理函数,接收一个字符串查询并调用Master类的run方法进行处理
def chat(query: str):
    master = Master()  # 初始化Master对象
    return master.run(query)


# 定义添加PDF路由
@app.post("/add_pdfs")
# 定义添加PDF路由处理函数,返回一个包含"response"键和"PDFs added!"值的字典
def add_pdfs():
    return {
   "response": "PDFs added!"}


# 定义添加文本路由
@app.post("add_texts")
# 定义添加文本路由处理函数,返回一个包含"response"键和"Texts added!"值的字典
def add_texts():
    return {
   "response": "Texts added!"}


# 定义WebSocket路由
@app.websocket("/ws")
# 定义WebSocket路由处理函数,接收一个WebSocket连接并启动一个无限循环
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        print("Connection closed")
        await websocket.close()


# 如果主程序为 __main__,则启动服务器
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="localhost", port=8090)

fastapi 请求:

postman 请求:

PyCharm 命令行记录:

相关实践学习
阿里云百炼xAnalyticDB PostgreSQL构建AIGC应用
通过该实验体验在阿里云百炼中构建企业专属知识库构建及应用全流程。同时体验使用ADB-PG向量检索引擎提供专属安全存储,保障企业数据隐私安全。
AnalyticDB PostgreSQL 企业智能数据中台:一站式管理数据服务资产
企业在数据仓库之上可构建丰富的数据服务用以支持数据应用及业务场景;ADB PG推出全新企业智能数据平台,用以帮助用户一站式的管理企业数据服务资产,包括创建, 管理,探索, 监控等; 助力企业在现有平台之上快速构建起数据服务资产体系
目录
相关文章
|
3天前
|
机器学习/深度学习 人工智能 供应链
AI技术在医疗领域的应用与未来展望###
本文深入探讨了人工智能(AI)技术在医疗领域的多种应用及其带来的革命性变化,从疾病诊断、治疗方案优化到患者管理等方面进行了详细阐述。通过具体案例和数据分析,展示了AI如何提高医疗服务效率、降低成本并改善患者体验。同时,文章也讨论了AI技术在医疗领域面临的挑战和未来发展趋势,为行业从业者和研究人员提供参考。 ###
|
3天前
|
机器学习/深度学习 人工智能 算法
AI技术在医疗领域的应用与挑战
【10月更文挑战第21天】 本文探讨了人工智能(AI)在医疗领域的多种应用,包括疾病诊断、治疗方案推荐、药物研发和患者管理等。通过分析这些应用案例,我们可以看到AI技术如何提高医疗服务的效率和准确性。然而,AI在医疗领域的广泛应用也面临诸多挑战,如数据隐私保护、算法透明度和伦理问题。本文旨在为读者提供一个全面的视角,了解AI技术在医疗领域的潜力和面临的困难。
|
1天前
|
存储 Serverless API
基于百炼平台构建智能体应用——十分钟构造能主动提问的导购智能体
本文介绍了如何使用阿里云百炼大模型服务平台构建一个多智能体的智能导购应用,并将其部署到钉钉。通过百炼的Assistant API,您可以快速构建一个包含规划助理、手机导购、冰箱导购和电视导购的智能导购系统。文章详细讲解了从创建函数计算应用、访问网站、验证智能导购效果到将商品检索应用集成到智能导购中的全过程,帮助您快速实现智能导购功能。
基于百炼平台构建智能体应用——十分钟构造能主动提问的导购智能体
|
3天前
|
机器学习/深度学习 人工智能 搜索推荐
AI在医疗健康领域的应用与前景
随着科技的不断进步,人工智能(AI)技术已经深入到我们生活的方方面面,特别是在医疗健康领域。本文将探讨AI在医疗健康领域的应用现状、面临的挑战以及未来的发展前景。
|
4天前
|
人工智能 自然语言处理 监控
AI技术在文本情感分析中的应用
【10月更文挑战第22天】本文将探讨人工智能(AI)如何改变我们对文本情感分析的理解和应用。我们将通过实际的代码示例,深入了解AI如何帮助我们识别和理解文本中的情感。无论你是AI新手还是有经验的开发者,这篇文章都将为你提供有价值的信息。让我们一起探索AI的奇妙世界吧!
13 3
|
3天前
|
人工智能 分布式计算 数据可视化
大模型私有化部署全攻略:硬件需求、数据隐私、可解释性与维护成本挑战及解决方案详解,附示例代码助你轻松实现企业内部AI应用
【10月更文挑战第23天】随着人工智能技术的发展,企业越来越关注大模型的私有化部署。本文详细探讨了硬件资源需求、数据隐私保护、模型可解释性、模型更新和维护等方面的挑战及解决方案,并提供了示例代码,帮助企业高效、安全地实现大模型的内部部署。
9 1
|
3天前
|
人工智能 分布式计算 数据可视化
大模型私有化部署全攻略:硬件需求、数据隐私、可解释性与维护成本挑战及解决方案详解,附示例代码助你轻松实现企业内部AI应用
【10月更文挑战第23天】随着人工智能技术的发展,大模型在各领域的应用日益广泛。然而,将其私有化部署到企业内部面临诸多挑战,如硬件资源需求高、数据隐私保护、模型可解释性差、更新维护成本高等。本文探讨了这些挑战,并提出了优化硬件配置、数据加密、可视化工具、自动化更新机制等解决方案,帮助企业顺利实现大模型的私有化部署。
10 1
|
4天前
|
人工智能 边缘计算 监控
边缘AI计算技术应用-实训解决方案
《边缘AI计算技术应用-实训解决方案》提供完整的实训体系,面向高校和科研机构的AI人才培养需求。方案包括云原生AI平台、百度AIBOX边缘计算硬件,以及8门计算机视觉实训课程与2门大模型课程。AI平台支持大规模分布式训练、超参数搜索、标注及自动化数据管理等功能,显著提升AI训练与推理效率。硬件涵盖多规格AIBOX服务器,支持多种推理算法及灵活部署。课程涵盖从计算机视觉基础到大模型微调的完整路径,通过真实商业项目实操,帮助学员掌握前沿AI技术和产业应用。
18 2
|
5天前
|
机器学习/深度学习 人工智能 自动驾驶
2024.10|AI/大模型在机器人/自动驾驶/智能驾舱领域的最新应用和深度洞察
本文介绍了AI和大模型在机器人、自动驾驶和智能座舱领域的最新应用和技术进展。涵盖多模态大语言模型在机器人控制中的应用、移动机器人(AMRs)的规模化部署、协作机器人的智能与安全性提升、AR/VR技术在机器人培训中的应用、数字孪生技术的优化作用、Rust语言在机器人编程中的崛起,以及大模型在自动驾驶中的核心地位、端到端自动驾驶解决方案、全球自动驾驶的前沿进展、智能座舱的核心技术演变和未来发展趋势。
15 2
|
6天前
|
人工智能 资源调度 数据可视化
【AI应用落地实战】智能文档处理本地部署——可视化文档解析前端TextIn ParseX实践
2024长沙·中国1024程序员节以“智能应用新生态”为主题,吸引了众多技术大咖。合合信息展示了“智能文档处理百宝箱”的三大工具:可视化文档解析前端TextIn ParseX、向量化acge-embedding模型和文档解析测评工具markdown_tester,助力智能文档处理与知识管理。