参考:
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 FastAPI
app = 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 CORSMiddleware
def 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-domain
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
mount_app_routes(app)
return app
def 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 BaseModel
class Item(BaseModel):
name: str
# 带=None 说明 该字段可以不填,不带则默认为必填项
description: str | None = None
price: float
tax: float | None = None
app = 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()