大模型应用：大模型多线程推理：并发请求的处理与资源隔离实践.77

# 单线程大模型推理基础代码
from transformers import AutoTokenizer, AutoModelForCausalLM
import time
from modelscope import snapshot_download

# 配置信息（选择轻量级模型，适合初学者运行，无需高端GPU）
MODEL_NAME = "qwen/Qwen1.5-1.8B-Chat"  # Qwen1.5-1.8B-Chat模型
CACHE_DIR = "D:\\modelscope\\hub"  # 本地模型缓存目录
DEVICE = "cpu"  # 支持cpu/gpu，无GPU直接用cpu即可

def load_model_and_tokenizer():
    """加载模型和分词器（单线程/多线程都只需要加载一次，节省资源）"""
    print("正在加载模型和分词器...（首次运行会下载模型，稍等片刻）")
    local_model_path = snapshot_download(MODEL_NAME, cache_dir=CACHE_DIR)
    tokenizer = AutoTokenizer.from_pretrained(local_model_path)
    model = AutoModelForCausalLM.from_pretrained(local_model_path).to(DEVICE).eval()  # eval()切换为推理模式，禁用训练相关层
    print("模型和分词器加载完成！")
    return tokenizer, model

def single_thread_inference(prompt, tokenizer, model):
    """单线程推理函数：处理单个请求"""
    start_time = time.time()
    try:
        # 步骤1：编码Prompt（转为模型可识别的输入）
        inputs = tokenizer(
            prompt,
            return_tensors="pt",  # 返回PyTorch张量
            truncation=True,     # 截断过长输入
            max_length=512       # 最大输入长度限制
        ).to(DEVICE)  # 把输入移到指定设备（cpu/gpu）

        # 步骤2：模型推理生成（核心计算步骤）
        outputs = model.generate(
            **inputs,
            max_new_tokens=100,  # 生成的最大新token数
            temperature=0.7,     # 生成随机性（0=确定性，1=高随机性）
            do_sample=True,      # 采样生成（避免重复输出）
            pad_token_id=tokenizer.eos_token_id  # 填充token id
        )

        # 步骤3：解码输出（转为自然语言）
        result = tokenizer.decode(outputs[0], skip_special_tokens=True)
        end_time = time.time()
        cost_time = round(end_time - start_time, 2)

        # 整理返回结果
        return {
            "prompt": prompt,
            "result": result,
            "cost_time": cost_time,
            "status": "success"
        }
    except Exception as e:
        end_time = time.time()
        cost_time = round(end_time - start_time, 2)
        return {
            "prompt": prompt,
            "result": str(e),
            "cost_time": cost_time,
            "status": "failed"
        }

# 主函数：单线程测试
if __name__ == "__main__":
    # 加载模型和分词器（只加载一次）
    tokenizer, model = load_model_and_tokenizer()

    # 定义测试请求
    test_prompts = [
        "请介绍一下人工智能的应用场景",
        "请写一句关于春天的诗句",
        "请解释什么是多线程",
        "请写一个简单的Python Hello World程序",
        "请介绍一下大模型的推理过程"
    ]

    # 串行处理多个请求
    print("\n开始串行处理请求...")
    total_start_time = time.time()
    for idx, prompt in enumerate(test_prompts, 1):
        print(f"\n===== 处理第{idx}个请求 =====")
        result = single_thread_inference(prompt, tokenizer, model)
        print(f"请求内容：{result['prompt']}")
        print(f"处理结果：{result['result']}")
        print(f"耗时：{result['cost_time']}秒")

    total_end_time = time.time()
    total_cost_time = round(total_end_time - total_start_time, 2)
    print(f"\n===== 所有请求处理完成 =====")
    print(f"总耗时：{total_cost_time}秒")

# 大模型多线程推理完整代码（含资源隔离）
from modelscope import AutoTokenizer, AutoModelForCausalLM
import time
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from modelscope import snapshot_download
# ===================== 配置信息（资源隔离核心配置）=====================
MODEL_NAME = "qwen/Qwen1.5-1.8B-Chat"  # Qwen1.5-1.8B-Chat模型
CACHE_DIR = "D:\\modelscope\\hub"  # 本地模型缓存目录
DEVICE = "cpu"  # 支持cpu/gpu，无GPU直接用cpu即可
MAX_THREADS = 3  # 最大线程数（资源隔离：根据硬件配置调整，4核CPU建议设为4-8）
MAX_NEW_TOKENS = 100  # 单个请求生成的最大token数（资源隔离：避免生成过长文本）
MAX_INPUT_LENGTH = 512  # 单个请求的最大输入长度（资源隔离：避免输入过长占用过多内存）
INFERENCE_TIMEOUT = 30  # 单个请求的最大推理超时时间（资源隔离：避免请求卡死占用线程）

# ===================== 全局变量（模型和分词器，所有线程共享）=====================
global_tokenizer = None
global_model = None

def load_global_model_and_tokenizer():
    """加载全局模型和分词器（所有线程共享，只加载一次，节省资源）"""
    global global_tokenizer, global_model
    print("正在加载模型和分词器...（首次运行会下载模型，稍等片刻）")
    start_load_time = time.time()
    local_model_path = snapshot_download(MODEL_NAME, cache_dir=CACHE_DIR)
    # 加载分词器
    global_tokenizer = AutoTokenizer.from_pretrained(local_model_path, trust_remote_code=True)
    # 加载模型并切换为推理模式
    global_model = AutoModelForCausalLM.from_pretrained(local_model_path, trust_remote_code=True).to(DEVICE).eval()
    
    load_cost_time = round(time.time() - start_load_time, 2)
    print(f"模型和分词器加载完成！耗时：{load_cost_time}秒")

def multi_thread_inference_task(prompt, task_id):
    """多线程推理任务函数（每个线程执行该函数，处理单个请求，自带资源隔离）"""
    thread_name = threading.current_thread().name
    start_time = time.time()
    try:
        # 资源隔离：检查请求是否超时（简单实现，进阶可使用signal模块）
        if time.time() - start_time > INFERENCE_TIMEOUT:
            raise TimeoutError(f"请求超时（超过{INFERENCE_TIMEOUT}秒）")
        
        # 步骤1：编码Prompt（转为模型可识别的输入，独立缓存，不共享）
        inputs = global_tokenizer(
            prompt,
            return_tensors="pt",
            truncation=True,
            max_length=MAX_INPUT_LENGTH
        ).to(DEVICE)
        
        # 步骤2：模型推理生成（核心计算，线程安全，Transformers库已优化）
        outputs = global_model.generate(
            **inputs,
            max_new_tokens=MAX_NEW_TOKENS,
            temperature=0.7,
            do_sample=True,
            pad_token_id=global_tokenizer.eos_token_id
        )
        
        # 步骤3：解码输出（转为自然语言，独立缓存，不共享）
        result = global_tokenizer.decode(outputs[0], skip_special_tokens=True)
        cost_time = round(time.time() - start_time, 2)
        
        # 整理成功结果
        return {
            "task_id": task_id,
            "thread_name": thread_name,
            "prompt": prompt,
            "result": result,
            "cost_time": cost_time,
            "status": "success"
        }
    
    except Exception as e:
        cost_time = round(time.time() - start_time, 2)
        # 整理失败结果
        return {
            "task_id": task_id,
            "thread_name": thread_name,
            "prompt": prompt,
            "result": str(e),
            "cost_time": cost_time,
            "status": "failed"
        }

def run_multi_thread_inference(prompts):
    """运行多线程推理（核心入口，使用线程池管理线程）"""
    print(f"\n开始多线程处理请求...最大线程数：{MAX_THREADS}")
    total_start_time = time.time()
    results = []
    
    # 创建线程池（资源隔离：限制最大线程数，自动管理线程生命周期）
    with ThreadPoolExecutor(max_workers=MAX_THREADS, thread_name_prefix="InferenceThread") as executor:
        # 提交所有任务到线程池
        future_to_task = {
            executor.submit(multi_thread_inference_task, prompt, idx): (prompt, idx)
            for idx, prompt in enumerate(prompts, 1)
        }
        
        # 收集所有任务的执行结果
        for future in as_completed(future_to_task):
            task_result = future.result()
            results.append(task_result)
            
            # 打印单个任务结果
            print(f"\n===== 任务{task_result['task_id']}（{task_result['thread_name']}）=====")
            print(f"请求内容：{task_result['prompt']}")
            print(f"处理状态：{task_result['status']}")
            print(f"耗时：{task_result['cost_time']}秒")
            if task_result['status'] == "success":
                print(f"处理结果：{task_result['result']}")
            else:
                print(f"错误信息：{task_result['result']}")
    
    # 汇总结果
    total_end_time = time.time()
    total_cost_time = round(total_end_time - total_start_time, 2)
    success_count = sum(1 for r in results if r['status'] == "success")
    failed_count = len(results) - success_count
    
    print(f"\n===== 所有请求处理完成 =====")
    print(f"总任务数：{len(prompts)} | 成功：{success_count} | 失败：{failed_count}")
    print(f"总耗时：{total_cost_time}秒（对比单线程，多线程耗时更短）")
    return results

# ===================== 主函数：测试多线程推理 =====================
if __name__ == "__main__":
    # 1. 加载全局模型和分词器（只加载一次）
    load_global_model_and_tokenizer()
    
    # 2. 定义测试请求列表
    test_prompts = [
        "请介绍一下人工智能的应用场景",
        "请写一句关于春天的诗句",
        "请解释什么是多线程",
        "请写一个简单的Python Hello World程序",
        "请介绍一下大模型的推理过程"
    ]
    
    # 3. 运行多线程推理
    run_multi_thread_inference(test_prompts)