关于AI_Qwen的几点记录

# Python 环境
conda create -n qwen python=3.10
conda activate qwen

# PyTorch（CUDA 11.8）
pip install torch==2.0.1+cu118 torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118

# Transformers 版本
pip install transformers==4.32.0
pip install accelerate==0.24.0
pip install bitsandbytes==0.41.0

# 其他依赖
pip install protobuf==3.20.0
pip install sentencepiece

from transformers import AutoModelForCausalLM, AutoTokenizer

model_name = "Qwen/Qwen-7B-Chat-Int8"

tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",
    trust_remote_code=True
)

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

def chat_with_qwen(prompt, model, tokenizer):
    response, history = model.chat(tokenizer, prompt, history=None)
    return response

# 使用示例
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen-7B-Chat-Int8",
    device_map="auto",
    trust_remote_code=True
).eval()

tokenizer = AutoTokenizer.from_pretrained(
    "Qwen/Qwen-7B-Chat-Int8",
    trust_remote_code=True
)

response = chat_with_qwen("Python中如何用装饰器实现单例模式？", model, tokenizer)
print(response)

from typing import List

def batch_chat(prompts: List[str], model, tokenizer, batch_size=4):
    """批量处理聊天请求"""
    results = []
    for i in range(0, len(prompts), batch_size):
        batch = prompts[i:i+batch_size]
        with torch.no_grad():
            batch_responses = []
            for prompt in batch:
                response, _ = model.chat(tokenizer, prompt, history=None)
                batch_responses.append(response)
        results.extend(batch_responses)
    return results

# 使用示例
questions = [
    "什么是闭包？",
    "解释一下 Python 的 GIL",
    "如何优化数据库查询？",
    "微服务和单体架构的区别？"
]

answers = batch_chat(questions, model, tokenizer)

# 使用梯度检查点
model.gradient_checkpointing_enable()

# 混合精度推理
from torch.cuda.amp import autocast

with autocast():
    response = model.chat(tokenizer, prompt, history=None)

from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
import asyncio

app = FastAPI()

@app.post("/chat")
async def chat_endpoint(prompt: str):
    response = await run_in_threadpool(
        chat_with_qwen, prompt, model, tokenizer
    )
    return {"response": response}

def chat_with_qwen(prompt, model, tokenizer):
    system_prompt = "你是一个专业的中文技术助手，请用清晰、准确的语言回答技术问题。"
    full_prompt = f"{system_prompt}\n用户：{prompt}\n助手："

    with torch.no_grad():
        response = model.chat(
            tokenizer,
            full_prompt,
            history=None,
            temperature=0.7,  # 降低随机性
            top_p=0.8         # 控制采样范围
        )
    return response

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

class RAGQwen:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.embedder = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
        self.index = None
        self.documents = []

    def add_documents(self, docs):
        """添加文档到知识库"""
        self.documents.extend(docs)
        embeddings = self.embedder.encode(docs)
        if self.index is None:
            self.index = faiss.IndexFlatL2(embeddings.shape[1])
        self.index.add(embeddings.astype('float32'))

    def search(self, query, k=3):
        """检索相关文档"""
        query_embedding = self.embedder.encode([query])
        distances, indices = self.index.search(query_embedding.astype('float32'), k)
        return [self.documents[i] for i in indices[0]]

    def chat_with_rag(self, query):
        """结合检索结果进行聊天"""
        relevant_docs = self.search(query)
        context = "\n".join(relevant_docs)
        prompt = f"基于以下信息回答问题：\n{context}\n\n问题：{query}"

        response, _ = self.model.chat(self.tokenizer, prompt, history=None)
        return response