第十三节：私有化部署与校内 GPU API

nvidia-smi

python -c "import torch; print('CUDA 可用：', torch.cuda.is_available()); print('GPU：', torch.cuda.get_device_name(0) if torch.cuda.is_available() else '未找到')"

python -m pip install -U modelscope transformers accelerate torch
python -c "import modelscope, transformers, torch; print('modelscope:', modelscope.__version__); print('torch:', torch.__version__)"

from modelscope import snapshot_download

model_id = "Qwen/Qwen2.5-7B-Instruct"  # 以当前模型卡显示的 ID 为准
model_dir = snapshot_download(model_id=model_id)

print("模型下载或缓存目录：")
print(model_dir)

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM

if not torch.cuda.is_available():
    raise RuntimeError(
        "当前 Python 没有可用 CUDA GPU。请改用校内 GPU API 或云端路线。"
    )

tokenizer = AutoTokenizer.from_pretrained(model_dir)
model = AutoModelForCausalLM.from_pretrained(
    model_dir,
    torch_dtype="auto",
    device_map="auto",
)

messages = [
    {"role": "system", "content": "你是课程助理，回答必须简洁、客观。"},
    {"role": "user", "content": "为什么时间戳对事件驱动分析很重要？"},
]
text = tokenizer.apply_chat_template(
    messages, tokenize=False, add_generation_prompt=True
)
inputs = tokenizer([text], return_tensors="pt").to(model.device)

with torch.no_grad():
    generated = model.generate(**inputs, max_new_tokens=120, temperature=0.2)

new_tokens = generated[:, inputs.input_ids.shape[1]:]
answer = tokenizer.batch_decode(new_tokens, skip_special_tokens=True)[0]
print(answer)

school-finance-agent/
├── server.py          # FastAPI 服务端
├── .env               # 模型 ID 与访问 token，只留在服务器
├── requirements.txt   # 依赖版本
└── logs/              # 只保存脱敏的运行日志（如课程需要）

TEACHING_API_TOKEN=请生成一串足够长的随机 token
MODEL_ID=Qwen/Qwen2.5-7B-Instruct

import os
import torch
from dotenv import load_dotenv
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
from transformers import AutoTokenizer, AutoModelForCausalLM

load_dotenv()
ACCESS_TOKEN = os.environ["TEACHING_API_TOKEN"]
MODEL_ID = os.environ["MODEL_ID"]

app = FastAPI(
    title="Teaching Finance Event Agent",
    description="仅限课程内网教学使用；不提供投资建议。",
)

class AgentRequest(BaseModel):
    prompt: str = Field(min_length=20, max_length=12000)

@app.get("/health")
def health():
    return {
        "status": "ok",
        "model": MODEL_ID,
        "cuda_available": torch.cuda.is_available(),
    }

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
model = AutoModelForCausalLM.from_pretrained(
    MODEL_ID, torch_dtype="auto", device_map="auto"
)

def require_token(authorization: str | None):
    if authorization != f"Bearer {ACCESS_TOKEN}":
        raise HTTPException(status_code=401, detail="invalid token")

@app.post("/finance-agent")
def finance_agent(
    body: AgentRequest,
    authorization: str | None = Header(default=None),
):
    require_token(authorization)

    messages = [
        {
            "role": "system",
            "content": "你是课程金融信息整理助手。不得提供投资建议。"
        },
        {"role": "user", "content": body.prompt},
    ]
    text = tokenizer.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    inputs = tokenizer([text], return_tensors="pt").to(model.device)

    with torch.no_grad():
        outputs = model.generate(
            **inputs, max_new_tokens=350, temperature=0.2
        )

    new_tokens = outputs[:, inputs.input_ids.shape[1]:]
    report = tokenizer.batch_decode(
        new_tokens, skip_special_tokens=True
    )[0].strip()

    return {"analysis_report": report, "model": MODEL_ID}

python -m pip install -U "fastapi[standard]" python-dotenv transformers torch
fastapi dev server.py

TEACHING_AGENT_URL=http://教师公布的校内地址:8000
TEACHING_API_TOKEN=教师单独提供的访问 token

from dotenv import load_dotenv
import os
import requests

load_dotenv()
base_url = os.environ["TEACHING_AGENT_URL"].rstrip("/")

health_response = requests.get(f"{base_url}/health", timeout=10)
print("状态码：", health_response.status_code)
health_response.raise_for_status()
print(health_response.json())

access_token = os.environ["TEACHING_API_TOKEN"]

headers = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json",
}
body = {"prompt": user_prompt}  # user_prompt 来自第十二节的四段式任务

response = requests.post(
    f"{base_url}/finance-agent",
    headers=headers,
    json=body,
    timeout=120,
)

print("状态码：", response.status_code)
response.raise_for_status()
result = response.json()

print("服务器使用的模型：", result["model"])
print("分析报告：")
print(result["analysis_report"])