智能体测开Day59

demo03

''' 采购管理 采购单号 全局唯一 采购日期 商品名称 商品数量 采购总金额 状态：待审核、已下单、已入库、已取消 增 post json格式的参数 /purchase_add 删 post json格式的参数 /purchase_delete 改 post json格式的参数 /purchase_modify 全量查 get /purchase_list 根据采购单号查询 get 依赖注入： 采购增删改查的功能，血药登录后才能使用，每个接口需要校验用户是否登录。 将是否登录提取出来作为一个公共的函数，其他接口中使用 采购增删改查的功能，需要连接数据库，使用完成后断开连接， 将数据库连接提取出来作为一个公共的函数，其他的 ''' import uuid import pymysql import uvicorn from fastapi import FastAPI, Form, Header from fastapi.params import Depends from langgraph_sdk.auth.exceptions import HTTPException from pydantic import BaseModel, Field from starlette.responses import HTMLResponse app = FastAPI() # 创建一个实例 class PurchaseMode(BaseModel): id: str = Field(...,min_length=16,max_length=16,description="采购单号")# ... 必填项 date: str product: str = Field(...,min_length=2,max_length=16,description="商品名称")# ... 必填项 num: int = Field(gt=0,description="数量") totalAmount: float status: str class DeleteMode(BaseModel): id: str = Field(...,min_length=16,max_length=16,description="采购单号") DB = { "host": "127.0.0.1", "port": 3306, "user": "root", "password": "123456", "charset": "utf8" } class UserModel(BaseModel): phone:str = Field(...,min_length=11,max_length=11,description="手机号码") pwd:str = Field(...,min_length=6,max_length=18,description="密码") def init_db(): try: db = pymysql.connect(**DB) # 解包 key=value,key=value # db = pymysql.connect(host="127.0.0.1",port= 3306,user= "root",password= "123456",) cursor = db.cursor() cursor.execute(f"create database if not exists fastapi38 character set utf8;") db.commit() db.select_db("fastapi38") cursor.execute(f"create table if not exists purchase (id varchar(16) primary key, date date," f"product varchar(16) not null, num int, totalAmount float, status varchar(16));") cursor.execute("create table if not exists user(id int primary key auto_increment, phone varchar(11)," "pwd varchar(20), token varchar(32));") db.commit() cursor.close() db.close() print("初始化数据库完成") except Exception as e: print(f"初始化数据库异常{e}") #####依赖1：数据库连接 def get_db(): db = pymysql.connect(**DB) db.select_db("fastapi38") yield db db.close() ##### 依赖2：是否登录 def check_login(token: str = Header(None)): if token is None: return {"msg": "请先登录","code": 20101} db = pymysql.connect(**DB) db.select_db("fastapi38") cursor = db.cursor() sql = "select * from user where token = %s" cursor.execute(sql,(token,)) a = cursor.fetchone() if a is not None: return True else: raise HTTPException(status_code=401,detail="登录状态已经失效") @app.post("/user/register") def register(data:UserModel, db: pymysql.Connection = Depends(get_db)): cursor = db.cursor() sql = "select * from user where phone = %s" cursor.execute(sql, (data.phone,)) a = cursor.fetchone() if a is not None: return {"msg": "手机号已被注册", "code": 20101} else: sql = "insert into user (phone, pwd) values (%s, %s)" cursor.execute(sql, (data.phone, data.pwd)) db.commit() return {"msg": "注册成功", "code": 10001} @app.post("/user/login") def login(data:UserModel, db: pymysql.Connection = Depends(get_db)): cursor = db.cursor() sql = "select * from user where phone = %s and pwd = %s" cursor.execute(sql, (data.phone,data.pwd,)) a = cursor.fetchone() if a is not None: token = str(uuid.uuid4()).replace("-","") # 生成一个token,表示用户登录状态 print("token",token) cursor.execute("update user set token = %s where phone = %s and pwd = %s",(token,data.phone,data.pwd)) db.commit() return {"msg": "登录成功", "code": 10001,"token": token} else: return {"msg": "手机号或密码错误", "code": 20101} @app.post("/purchase_add",dependencies=[Depends(check_login)]) def add(data: PurchaseMode,db: pymysql.Connection = Depends(get_db)): cursor = db.cursor() # 查询结果是元祖 sql = "select * from purchase where id = %s" cursor.execute(sql,(data.id,)) a = cursor.fetchone() if a is not None: return {"msg": "采购单号重复", "code": 20101} else: sql = "insert into purchase (id, date, product, num, totalAmount, status) values (%s, %s, %s, %s, %s, %s)" cursor.execute(sql,(data.id,data.date,data.product,data.num,data.totalAmount,data.status)) db.commit() cursor.close() return {"msg": "添加成功", "code": 10001} @app.post("/purchase_delete", dependencies=[Depends(check_login)]) def delete(data: DeleteMode, db: pymysql.Connection = Depends(get_db)): cursor = db.cursor() sql = "select * from purchase where id = %s" cursor.execute(sql, (data.id,)) a = cursor.fetchone() if a is None: cursor.close() db.close() return {"msg": "采购单号不存在", "code": 20102} else: sql1 = "delete from purchase where id=%s" cursor.execute(sql1,(data.id,)) db.commit() cursor.close() return {"msg": "删除成功", "code": 10001} @app.post("/purchase_modify", dependencies=[Depends(check_login)]) def modify(data: PurchaseMode, db: pymysql.Connection = Depends(get_db)): cursor = db.cursor() cursor.execute("select * from purchase where id=%s", (data.id,)) a = cursor.fetchone() if a is None: cursor.close() db.close() return {"msg":"单据不存在，无法修改","code":20103} else: sql = "update purchase set date=%s, product=%s, num=%s, totalAmount=%s, status=%s where id=%s" cursor.execute(sql,(data.date,data.product,data.num,data.totalAmount,data.status,data.id)) db.commit() cursor.close() return {"msg":"修改成功","code":10001} @app.get("/purchase_list", dependencies=[Depends(check_login)]) def purchase_list(db: pymysql.Connection = Depends(get_db)): cursor = db.cursor(pymysql.cursors.DictCursor) sql = "select * from purchase" cursor.execute(sql) a = cursor.fetchall() cursor.close() return {"msg": "查询成功", "code": 10001, "data": a} @app.get("/purchase_get",dependencies=[Depends(check_login)]) def get_one(id:str, db: pymysql.Connection = Depends(get_db)): cursor = db.cursor(pymysql.cursors.DictCursor) sql = "select * from purchase where id = %s" cursor.execute(sql, (id,)) a = cursor.fetchone() cursor.close() if a is None: return {"msg": "查询失败,无该采购单号", "code": 20104} else: return {"msg": "查询成功", "code": 10001, "data":a} # 页面路由：读取static文件夹html文件 @app.get("/register.html", response_class=HTMLResponse) def page_register(): with open("static/register.html", "r", encoding="utf-8") as f: return f.read() @app.get("/login.html", response_class=HTMLResponse) def page_login(): with open("static/login.html", "r", encoding="utf-8") as f: return f.read() @app.get("/purchase.html",response_class=HTMLResponse) def purchase_page(): with open('static/purchase.html', 'r', encoding='utf-8') as f: return f.read() if __name__ == '__main__': init_db() # 初始化数据库 uvicorn.run("demo03:app",port=8000,reload=True,log_level="debug")

test_demo03

''' pip install faker 构造测试数据的一个库 ''' import random import requests from faker import Faker def test_001(): f = Faker('zh-CN') phone = f.phone_number() pwd = f.password() print("手机号",phone) print("密码",pwd) url = "http://127.0.0.1:8000/user/register" data = {"phone": phone, "pwd": pwd} r = requests.post(url, json=data) print(r.text) assert r.status_code == 200 assert r.json()["code"] == 10001 assert r.json()["msg"] == "注册成功" url = "http://127.0.0.1:8000/user/login" r = requests.post(url, json=data) print(r.text) token = r.json()['token'] print(token) assert r.status_code == 200 assert r.json()["code"] == 10001 assert r.json()["msg"] == "登录成功" url = "http://127.0.0.1:8000/purchase_add" data = { "id": str(random.randint(1000, 9999)) * 4, "date": "2026-08-21", "product": "办公打印机", "num": 2, "totalAmount": 5600.00, "status": "待审核" } headers = {"token": token} r = requests.post(url, json=data, headers=headers) print(r.text) assert r.status_code == 200 assert r.json()["code"] == 10001 assert r.json()["msg"] == "添加成功"

demo01_调用大模型

''' pip install langchain langchain-openai python-dotenv langchain 用来开发基于大模型的应用程序 api-key 1. 可以写死代码中 2. 写到env文件中-我们使用这种 3. 写到环境变量 ''' import os import dotenv from langchain.chat_models import init_chat_model from langchain_community.chat_models.tongyi import ChatTongyi from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url = os.getenv("BASE_URL") api_key = os.getenv("API_KEY") model_name = os.getenv("MODEL_NAME") # 方式1：创建一个大模型的实例，任何大模型都可以使用这种方式。 model = init_chat_model( model = model_name, api_key = api_key, base_url = base_url, model_provider = "openai", temperature = 0.9, # 数字越高越有创意 max_tokens = 300, # 回答的最大的token数量 max_retries = 3 # 最大重试的次数 ) # 调用大模型 r = model.invoke('你好，你是谁？') print(r) print(type(r)) # <class 'langchain_core.messages.ai.AIMessage'> print(r.content) # 方式2：针对OpenAI的模型 model = ChatOpenAI( model = model_name, api_key = api_key, base_url = base_url ) r = model.invoke('你好，你是谁？') print(r.content) # 方式3：针对通义千问、deepseek等 model = ChatTongyi(model = model_name,base_url = base_url,dashscope_api_key = api_key) r = model.invoke('你好，你是谁？') print(r.content)

demo02_消息的几种方式

''' 1. 字符串 适用于不用保留历史会话的场景。 2. AIMessage、HumanMessage、SystemMessage，定义消息列表 使用于保留历史会话的场景。 3. 字典列表 [{"role": "", "content": ""},{"role": "", "content": ""}] ''' import os import dotenv from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url = os.getenv("BASE_URL") api_key = os.getenv("API_KEY") model_name = os.getenv("MODEL_NAME","qwen3.7-plus") # 方式1：创建一个大模型的实例，任何大模型都可以使用这种方式。 model = ChatOpenAI( model = model_name, api_key = api_key, base_url = base_url, ) # 1. 字符串 # r = model.invoke('你好，你是谁？') # print(r.content) # 2. 消息列表 msgs = [ SystemMessage(content="你是一个人工智能的助手，擅长帮用户解决数字问题，你的名字叫小A"), AIMessage(content="好的，我叫小A"), HumanMessage(content="你是谁？") ] r = model.invoke(msgs) print(r.content) # 小A # 3. 字典列表 msgs = [ {"role": "system","content":"你是一个人工智能的助手，擅长帮用户解决数字问题，你的名字叫小A"}, {"role": "ai","content": "好的，我叫小A"}, {"role": "human","content": "你是谁？"}, ] r = model.invoke(msgs) print(r.content) # 小A msgs = [ ("system","你是一个人工智能的助手，擅长帮用户解决数字问题，你的名字叫小A"), ("ai","好的，我叫小A"), ("human","你是谁？"), ] r = model.invoke(msgs) print(r.content) # 小A

demo03_调用大模型的几种方式

''' 1. invoke 非流式输出，大模型一次性把回答返回 2. stream 流式输出，大模型逐字逐句的返回 3. batch 批量调用 4. ainvoke 异步调用 ''' import os import dotenv from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_openai import ChatOpenAI # 加载环境文件 dotenv.load_dotenv() base_url = os.getenv("BASE_URL") api_key = os.getenv("API_KEY") model_name = os.getenv("MODEL_NAME","qwen3.7-plus") model = ChatOpenAI( model = model_name, api_key = api_key, base_url = base_url, ) ## 2.stream流式输出。大模型逐字逐句的返回 r = model.stream('给我一首李白的诗') print(type(r)) # <class 'generator'> 生成器 # 列表推导式： [i for i in range(10)] 生成器（i for i in range(10)） # 普通函数加yield，生成器 for item in r: print("====",item.content)