pytorch 加载大模型

```
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Optional
import time
from transformers import AutoModelForCausalLM, AutoTokenizer

app = FastAPI(title="智能对话API", description="基于Qwen3-8B模型的对话API服务")

# 加载模型和分词器
model_path = "./models--Qwen--Qwen3-8B/snapshots/2069b3fae1114555f3c020c81410e51fa0f656f2"
# tokenizer = AutoTokenizer.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype="auto",
    device_map="auto",
    trust_remote_code=True  # 加载本地模型时可能需要
)

# 系统初始提示词
'''
SYSTEM_PROMPT = ("你是一个能够执行交通场景巡检任务的智能机器人，目前能够巡检的区域为彩虹科技大厦，能够执行的巡检任务为车辆违停和道路施工。"
                 "请你根据用户的要求回答问题或者执行巡检任务。如果需要执行巡检任务，请你简要并明确给出要执行的任务，比如车辆违停或者道路施工。"
                 "如果你发现客户是要做巡检任务，但是不确定客户具体的巡检意图，你可以跟客户确认。但是请你不要在客户没有问你的时候主动说明自己的能力，"
                 "也不要试图去引导客户发起巡检任务要求。当你确定了要做的任务时，请输出'1、车辆违停'或者'2、道路施工'")
'''
SYSTEM_PROMPT = ("你是道通科技旗下道和通泰机器人公司研发的一个交通巡检智能体，可以自主调度可用的无人机执行交通场景的巡检任务。目前能够巡检的区域为彩虹科技大厦，能够执行的巡检任务为车辆违停和道路施工，可以使用的无人机有EVO Max。请你根据用户的要求回答问题或者执行巡检任务。"
          "如果你发现客户是要做巡检任务，你需要先跟用户确认清楚具体的巡检区域、巡检任务和要使用的无人机。如果用户没有明确说明巡检区域、巡检任务和要使用的无人机，你需要跟用户确认清楚每一项。如果确认清楚了，那就明确返回‘好的，我现在就立即执行’。"
          "但是请你不要在客户没有问你的时候主动说明自己的能力，也不要试图去引导客户发起巡检任务要求。并且请不要在没有明确清楚巡检区域、巡检任务和使用的无人机的时候去开始执行任务。")

# 数据模型
class ChatMessage(BaseModel):
    role: str
    content: str

class ChatRequest(BaseModel):
    messages: List[ChatMessage]

class ChatResponse(BaseModel):
    response: str
    thinking: Optional[str] = None

# 存储会话历史的字典（简单内存存储，不用数据库）
conversation_history = {}

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    messages = [{"role": msg.role, "content": msg.content} for msg in request.messages]

# 如果是新会话，添加系统提示
    if not any(msg["role"] == "system" for msg in messages):
        messages.insert(0, {"role": "system", "content": SYSTEM_PROMPT})

# 准备模型输入
    text = tokenizer.apply_chat_template(messages,
                                         tokenize=False,
                                         add_generation_prompt=True,
                                         enable_thinking=False)

model_inputs = tokenizer([text], return_tensors="pt").to(model.device)

# 生成回复
    start_time = time.time()
    generated_ids = model.generate(**model_inputs, max_new_tokens=32768)
    output_ids = generated_ids[0][len(model_inputs.input_ids[0]):].tolist()

# 解析思考内容
    try:
        index = len(output_ids) - output_ids[::-1].index(151668)
    except ValueError:
        index = 0

thinking_content = tokenizer.decode(output_ids[:index], skip_special_tokens=True).strip("\n")
    content = tokenizer.decode(output_ids[index:], skip_special_tokens=True).strip("\n")

# 计算响应时间
    response_time = time.time() - start_time
    print(f"响应时间: {response_time:.2f}秒")

return ChatResponse(response=content, thinking=thinking_content)

@app.get("/")
async def root():
    return {"message": "智能对话API服务正在运行，请使用 /chat 端点进行对话"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8060)
```