参考:
https://fastapi.tiangolo.com/zh/tutorial/first-steps/https://fastapi.tiangolo.com/zh/tutorial/first-steps/
FastAPI 用于基于标准 Python 类型提示使用 Python 构建 API,使用 ASGI 的标准来构建 Python Web 框架和服务器。所有简单理解:FastAPI 是一个 ASGI Web 框架。
最简单的 FastAPI 文件可能像下面这样:
from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}
将其复制到 main.py
文件中。
python解释文件启动方案
python解释文件启动方案
跨域请求
前端(如 React/Vue/Angular)和后端服务(如 FastAPI)运行在不同的域名或端口时(如前端运行在 http://localhost:3000
,后端运行在 http://localhost:8000
),需要解决跨域问题。
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)
项目完整代码
from typing import Union
from fastapi import FastAPI
from server.chat.chat import chat
import uvicorn
import argparse
from fastapi.middleware.cors import CORSMiddlewaredef create_app(run_mode: str = None):app = FastAPI(title="fufan-chat-api API Server",)# Add CORS middleware to allow all origins# 在config.py中设置OPEN_DOMAIN=True,允许跨域# set OPEN_DOMAIN=True in config.py to allow cross-domainapp.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)mount_app_routes(app)return appdef mount_app_routes(app: FastAPI, run_mode: str = None):# 这是路由挂载的两种方式之一# 大模型对话接口# 将函数 chat 绑定为该接口的处理逻辑。# 当客户端向 /api/chat 路径发送 POST 请求时,FastAPI 会调用 chat 函数来处理请求。app.post("/api/chat",tags=["Chat"],summary="大模型对话交互接口",)(chat)# **kwargs 让变量以key=value的形式传递
def run_api(host, port, **kwargs):if kwargs.get("ssl_keyfile") and kwargs.get("ssl_certfile"):uvicorn.run(app,host=host,port=port,ssl_keyfile=kwargs.get("ssl_keyfile"),ssl_certfile=kwargs.get("ssl_certfile"),)else:uvicorn.run(app, host=host, port=port)if __name__ == '__main__':parser = argparse.ArgumentParser()parser.add_argument("--host", type=str, default="192.168.110.131")parser.add_argument("--port", type=int, default=8000)parser.add_argument("--ssl_keyfile", type=str)parser.add_argument("--ssl_certfile", type=str)# 初始化消息args = parser.parse_args()args_dict = vars(args)app = create_app()run_api(host=args.host,port=args.port,ssl_keyfile=args.ssl_keyfile,ssl_certfile=args.ssl_certfile,)
创建数据模型¶
把数据模型声明为继承 BaseModel
的类。
使用 Python 标准类型声明所有属性:
参数的数据类型:
| 是联合类型的意思,相当于 union 意思为类型为几个中的一个
from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: str# 带=None 说明 该字段可以不填,不带则默认为必填项description: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.post("/items/")
async def create_item(item: Item):return item
使用省略号(...
)声明必需参数
参数以 Body
的形式定义,表示这些数据会从请求体(HTTP Request Body)中获取。
默认值:Body(...)
表示这是一个必填参数,用户必须在请求体中提供 query
字段,否则会报错。
async def chat(query: str = Body(..., description="用户输入", examples=["你好"]),user_id: str = Body("", description="用户ID"),conversation_id: str = Body("", description="对话框ID"),conversation_name: str = Body("", description="对话框名称"),history_len: int = Body(-1, description="从数据库中取历史消息的数量"),history: Union[int, List[History]] = Body([],description="历史对话,设为一个整数可以从数据库中读取历史消息",examples=[[{"role": "user","content": "我们来玩成语接龙,我先来,生龙活虎"},{"role": "assistant", "content": "虎头虎脑"}]]),stream: bool = Body(False, description="流式输出"),model_name: str = Body(LLM_MODELS[0], description="LLM 模型名称。"),temperature: float = Body(TEMPERATURE, description="LLM 采样温度", ge=0.0, le=2.0),max_tokens: Optional[int] = Body(None, description="限制LLM生成Token数量,默认None代表模型最大值"),prompt_name: str = Body("default", description="使用的prompt模板名称(在configs/prompt_config.py中配置)"),):
嵌套函数通常用于生成流式输出的迭代器,尤其是在需要逐步生成数据的异步函数中。
async def chat(query: str):async def stream_response():for char in query:yield char # 每次生成一个字符await asyncio.sleep(0.1) # 模拟延迟return stream_response()