当前位置： 首页 > news >正文

Python-LLMChat

news 2025/9/10 4:27:09

Python-LLMChat

使用Python进行大模型对话，分別使用【request】【FastAPI】【Flask】【Langchain】进行实现，附带POSTMAN请求示例，代码仓库实现地址：https://gitee.com/enzoism/python_llm_chat

pip install -r requirements.txt

1-request原生请求

postman都可以实现的那种请求方式

import jsonimport requestsurl = "https://api.siliconflow.cn/v1/chat/completions"
API_KEY = "sk-rnqurwrjarpxjetwkcphjymkcrbxpoyhpueaxwtkevmdwhlw"payload = {"model": "Qwen/QwQ-32B","messages": [{"role": "user","content": "给我讲一个笑话?"}]
}
headers = {"Authorization": f"Bearer {API_KEY}","Content-Type": "application/json"
}response = requests.post(url, json=payload, headers=headers)
# 假设 response 是你通过 requests 得到的响应
json_data = response.json()print(json_data)# 打印标准格式的 JSON（带缩进，中文正常显示）
print(json.dumps(json_data, indent=4, ensure_ascii=False))

2-request原生请求-添加会话历史-FastAPI

from typing import List, Dictimport requests
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModelapp = FastAPI(title="聊天历史管理API", version="1.0.0")# API配置
API_URL = "https://api.siliconflow.cn/v1/chat/completions"
API_KEY = "sk-rnqurwrjarpxjetwkcphjymkcrbxpoyhpueaxwtkevmdwhlw"# 模拟数据库存储用户聊天历史
user_chat_history: Dict[str, List[Dict[str, str]]] = {}class ChatRequest(BaseModel):user_id: strmessage: strclass ChatResponse(BaseModel):user_id: strresponse: strchat_history: List[Dict[str, str]]def get_ai_response(messages: List[Dict[str, str]]) -> str:"""调用AI模型获取回复"""payload = {"model": "Qwen/QwQ-32B","messages": messages}headers = {"Authorization": f"Bearer {API_KEY}","Content-Type": "application/json"}try:response = requests.post(API_URL, json=payload, headers=headers)response.raise_for_status()return response.json()['choices'][0]['message']['content']except requests.exceptions.RequestException as e:raise HTTPException(status_code=500, detail=f"AI模型调用失败: {str(e)}")@app.post("/chat", response_model=ChatResponse)
async def chat_endpoint(chat_request: ChatRequest):"""处理用户聊天请求"""user_id = chat_request.user_iduser_message = chat_request.message# 获取用户历史记录（如果没有则创建）if user_id not in user_chat_history:user_chat_history[user_id] = []# 添加用户消息到历史记录user_chat_history[user_id].append({"role": "user","content": user_message})# 获取AI回复ai_response = get_ai_response(user_chat_history[user_id])# 添加AI回复到历史记录user_chat_history[user_id].append({"role": "assistant","content": ai_response})return ChatResponse(user_id=user_id,response=ai_response,chat_history=user_chat_history[user_id])@app.get("/history/{user_id}", response_model=List[Dict[str, str]])
async def get_chat_history(user_id: str):"""获取用户聊天历史"""if user_id not in user_chat_history:return []return user_chat_history[user_id]@app.delete("/history/{user_id}")
async def clear_chat_history(user_id: str):"""清除用户聊天历史"""if user_id in user_chat_history:user_chat_history[user_id] = []return {"message": f"用户 {user_id} 的聊天历史已清除"}return {"message": f"用户 {user_id} 没有聊天历史"}@app.get("/")
async def root():return {"message": "聊天历史管理API正在运行", "docs": "/docs"}if __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=8000)

3-request原生请求-添加会话历史-Flask版本

from typing import List, Dictimport requests
from flask import Flask, request, jsonifyapp = Flask(__name__)# API配置
API_URL = "https://api.siliconflow.cn/v1/chat/completions"
API_KEY = "sk-rnqurwrjarpxjetwkcphjymkcrbxpoyhpueaxwtkevmdwhlw"# 模拟数据库存储用户聊天历史
user_chat_history: Dict[str, List[Dict[str, str]]] = {}def get_ai_response(messages: List[Dict[str, str]]) -> str:"""调用AI模型获取回复"""payload = {"model": "Qwen/QwQ-32B","messages": messages}headers = {"Authorization": f"Bearer {API_KEY}","Content-Type": "application/json"}try:response = requests.post(API_URL, json=payload, headers=headers)response.raise_for_status()return response.json()['choices'][0]['message']['content']except requests.exceptions.RequestException as e:return f"AI 模型调用失败: {str(e)}"@app.route("/chat", methods=["POST"])
def chat():"""处理用户聊天请求"""data = request.get_json()user_id = data.get("user_id")user_message = data.get("message")if not user_id or not user_message:return jsonify({"error": "缺少 user_id 或 message"}), 400# 初始化用户历史if user_id not in user_chat_history:user_chat_history[user_id] = []# 添加用户消息user_chat_history[user_id].append({"role": "user", "content": user_message})# 获取AI回复ai_response = get_ai_response(user_chat_history[user_id])# 添加AI回复到历史记录user_chat_history[user_id].append({"role": "assistant", "content": ai_response})return jsonify({"user_id": user_id,"response": ai_response,"chat_history": user_chat_history[user_id]})@app.route("/history/<user_id>", methods=["GET"])
def get_history(user_id):"""获取用户聊天历史"""history = user_chat_history.get(user_id, [])return jsonify(history)@app.route("/history/<user_id>", methods=["DELETE"])
def clear_history(user_id):"""清除用户聊天历史"""if user_id in user_chat_history:user_chat_history[user_id] = []return jsonify({"message": f"用户 {user_id} 的聊天历史已清除"})return jsonify({"message": f"用户 {user_id} 没有聊天历史"})@app.route("/")
def index():return jsonify({"message": "Flask 聊天历史管理API正在运行", "endpoints": ["/chat", "/history/<user_id>"]})if __name__ == "__main__":app.run(debug=True, host="0.0.0.0", port=8000)

4-request原生请求-Langchain版本

import osfrom flask import Flask, request, jsonify
from langchain.schema import HumanMessage
from langchain_openai import ChatOpenAIapp = Flask(__name__)# ---------- 关键配置 ----------
# SiliconFlow硅基流动 兼容 OpenAI 接口
os.environ["OPENAI_API_KEY"] = "sk-rnqurwrjarpxjetwkcphjymkcrbxpoyhpueaxwtkevmdwhlw"
os.environ["OPENAI_API_BASE"] = "https://api.siliconflow.cn/v1"MODEL_NAME = "Qwen/QwQ-32B"# LangChain 客户端
llm = ChatOpenAI(model_name=MODEL_NAME,temperature=0.7,max_tokens=512
)# ---------- 内存历史 ----------
user_histories: dict[str, list] = {}  # list[HumanMessage | AIMessage]def get_history_list(user_id: str):"""返回 LangChain 消息对象列表"""return user_histories.setdefault(user_id, [])@app.route("/chat", methods=["POST"])
def chat():data = request.get_json()user_id = data.get("user_id")user_text = data.get("message")if not user_id or not user_text:return jsonify({"error": "缺少 user_id 或 message"}), 400history = get_history_list(user_id)history.append(HumanMessage(content=user_text))# 调用模型ai_msg = llm.invoke(history)  # 返回 AIMessagehistory.append(ai_msg)# 转成普通 dict 给前端serializable = [{"role": "user" if isinstance(msg, HumanMessage) else "assistant","content": msg.content}for msg in history]return jsonify({"user_id": user_id,"response": ai_msg.content,"chat_history": serializable})@app.route("/history/<user_id>", methods=["GET"])
def get_history(user_id):history = get_history_list(user_id)serializable = [{"role": "user" if isinstance(msg, HumanMessage) else "assistant","content": msg.content}for msg in history]return jsonify(serializable)@app.route("/history/<user_id>", methods=["DELETE"])
def clear_history(user_id):user_histories.pop(user_id, None)return jsonify({"message": f"用户 {user_id} 的聊天历史已清除"})@app.route("/")
def index():return jsonify({"message": "Flask + LangChain 聊天服务已启动","endpoints": ["/chat", "/history/<user_id>"]})if __name__ == "__main__":app.run(debug=True, host="0.0.0.0", port=8000)

文章转载自：

http://hjUdR9GN.pLhhd.cn
http://cnaBiB90.pLhhd.cn
http://0bwlusxs.pLhhd.cn
http://YvYR0VNf.pLhhd.cn
http://cL4bcfot.pLhhd.cn
http://t414R3Rf.pLhhd.cn
http://Kbc1BUa2.pLhhd.cn
http://qwCS9nyi.pLhhd.cn
http://LbpEm60Z.pLhhd.cn
http://QhlR9OXg.pLhhd.cn
http://8l7Fs62t.pLhhd.cn
http://aX46wYz6.pLhhd.cn
http://jDn1L3HU.pLhhd.cn
http://yNLtTc7k.pLhhd.cn
http://TibxhOih.pLhhd.cn
http://d93uCj6H.pLhhd.cn
http://FoC1524k.pLhhd.cn
http://eTFfRBou.pLhhd.cn
http://uyzXZovl.pLhhd.cn
http://evqidUME.pLhhd.cn
http://9w8WNK3z.pLhhd.cn
http://Xn9XqGOa.pLhhd.cn
http://1xYSOk0U.pLhhd.cn
http://mzNhZEl3.pLhhd.cn
http://DuHI2WLx.pLhhd.cn
http://lpek62Px.pLhhd.cn
http://an5JFA0Q.pLhhd.cn
http://CU0HQ0bL.pLhhd.cn
http://HYYcTWTl.pLhhd.cn
http://qbmfgazk.pLhhd.cn
http://www.dtcms.com/a/371554.html

相关文章：

  • 【C++】C++入门—(下)
  • 大数据毕业设计选题推荐-基于大数据的国家基站整点数据分析系统-Hadoop-Spark-数据可视化-BigData
  • 如何编写ICT模拟功能测试
  • 【C++】类与对象（下）
  • 在Ubuntu中如何使用PM2来运行一个编译好的Vue项目
  • Mysql数据库——第一阶段
  • 10 qml教程-自定义属性
  • 万字详解网络编程之TCP/IP协议与UDP协议
  • Gitlab 配置自定义 clone 地址
  • 408考研——循环队列代码题常见套路总结
  • 「日拱一码」081 机器学习——梯度增强特征选择GBFS
  • 阿里云镜像地址获取，并安装 docker的mysql和nginx等服务，java,python,ffmpeg,go等环境
  • IPSec综合配置实验
  • 实现滚动到页面指定位置
  • Linux 系统监控 + 邮件告警实战：CPU、内存、IO、流量全覆盖
  • HarmonyOS 应用开发新范式：深入剖析 Stage 模型与 ArkTS 状态管理
  • Elasticsearch面试精讲 Day 11：索引模板与动态映射
  • 5G NR PDCCH之信号调制
  • Android --- AOSP下载及编译
  • C#中的托管资源与非托管资源介绍
  • 初识Vue
  • JSP到Tomcat特详细教程
  • 滑动窗口与双指针（1）——定长
  • Lua ＞ OpenResty Lua Module
  • [LeetCode 热题 100] 32. 最长有效括号
  • Python IO编程——文件读写
  • fps：游戏玩法
  • S 4.1深度学习--自然语言处理NLP--理论
  • [NCTF2019]Fake XML cookbook
  • ARM体系结构学习②