import requests import json import base64 from typing import List, Optional, Dict # 配置模型服务地址和参数 API_BASE_URL = "http://localhost:8000/v1" # 你的vLLM服务地址 MODEL_NAME = "Qwen3-VL-4B-Instruct" # 和启动时--served-model-name一致 TIMEOUT = 30 # 请求超时时间（秒） class Qwen3VLClient: """Qwen3-VL-4B-Instruct 模型调用客户端""" def __init__(self, api_base_url: str, model_name: str): self.api_base_url = api_base_url self.model_name = model_name self.headers = { "Content-Type": "application/json", "Accept": "application/json", } def chat_completions( self, messages: List[Dict], temperature: float = 0.7, max_tokens: int = 1024, top_p: float = 0.8, ) -> Optional[str]: """ 调用模型生成回复 :param messages: 对话消息列表，格式参考OpenAI规范 :param temperature: 生成随机性，0-1之间，值越大越随机 :param max_tokens: 最大生成token数 :param top_p: 采样阈值，0-1之间 :return: 模型回复文本，失败返回None """ # 构造请求体 payload = { "model": self.model_name, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, "top_p": top_p, "stream": False, # 非流式输出 } try: # 发送POST请求 response = requests.post( url=f"{self.api_base_url}/chat/completions", headers=self.headers, data=json.dumps(payload, ensure_ascii=False), # 避免中文乱码 timeout=TIMEOUT, ) # 检查响应状态 response.raise_for_status() # 解析响应结果 result = response.json() if result.get("choices") and len(result["choices"]) > 0: return result["choices"][0]["message"]["content"] else: print("模型未返回有效回复") return None except requests.exceptions.RequestException as e: print(f"请求失败：{str(e)}") if hasattr(e, "response") and e.response is not None: print(f"错误响应详情：{e.response.json()}") # 打印服务端具体错误 return None def get_image_base64(image_path: str) -> str: """将本地图片转换为base64编码（修复Windows路径问题）""" try: # 统一路径分隔符（兼容Windows） image_path = image_path.replace("\\", "/") print(f"正在读取图片：{image_path}") with open(image_path, "rb") as f: base64_data = base64.b64encode(f.read()).decode("utf-8") return base64_data except FileNotFoundError: print(f"错误：找不到图片文件 {image_path}，请检查路径是否正确") return "" except Exception as e: print(f"图片转base64失败：{str(e)}") return "" # ------------------------------ # 示例1：纯文本对话调用（验证服务可用性） # ------------------------------ def test_text_chat(): print("===== 纯文本对话测试 =====") client = Qwen3VLClient(API_BASE_URL, MODEL_NAME) messages = [{"role": "user", "content": "你好，请介绍一下你自己"}] response = client.chat_completions(messages, temperature=0.7, max_tokens=500) if response: print(f"模型回复：\n{response}\n") # ------------------------------ # 示例2：多模态（图片+文本）调用（修复路径问题） # ------------------------------ def test_multimodal_chat(): print("===== 多模态（图片+文本）测试 =====") client = Qwen3VLClient(API_BASE_URL, MODEL_NAME) # ********** 重点：修改为你的图片实际路径 ********** # 方式1：Windows 原始路径（需双反斜杠） # image_path = "D:\\phpstudy_pro\\python\\.vscode\\练习\\ollama\\test.jpg" # 方式2：推荐用正斜杠（无需转义） image_path = "D:/phpstudy_pro/python/.vscode/练习/ollama/test.jpg" # 转换图片为base64 img_base64 = get_image_base64(image_path) if not img_base64: print("图片base64编码失败，跳过多模态测试") return # 构造合法的多模态消息体 messages = [ { "role": "user", "content": [ {"type": "text", "text": "请详细描述这张图片的内容"}, { "type": "image_url", "image_url": { "url": f"data:image/jpeg;base64,{img_base64}" # 确保base64非空 }, }, ], } ] # 调用模型 response = client.chat_completions(messages, temperature=0.1, max_tokens=1024) if response: print(f"模型回复：\n{response}\n") # ------------------------------ # 主函数：运行测试 # ------------------------------ if __name__ == "__main__": # 先测试纯文本（确保服务正常） # test_text_chat() # 测试多模态（已修复路径问题） test_multimodal_chat()
vLLM运行模型显存不足的原因及解决方法在使用vLLM部署大语言模型时，显存不足是最常见的报错（如提示“Free memory on device cuda:0 is less than desired GPU memory utilization”），其核心是GPU显存无法满足模型运行、优化策略及任务负载的综合需求。以下将详细分析显存不足的原因，并提供按优先级排序的可操作解决方法，适配各类常见场景（尤其针对8GB等小显存设备）。一、显存不足的核心原因vLLM显存占用主要分为「模型本身占用」「vLLM优化策略占用」「任务负载占用」三类，任一环节超出GPU显存上限，都会导致启动或运行失败，具体原因如下：1. 模型本身占用超出显存上限（最常见）这是最核心的原因，模型的参数规模、精度直接决定基础显存占用：参数规模过大：如13B、34B级别的模型，原生FP16精度下，单卡（如8GB显存）无法容纳（13B模型FP16精度约占用26GB显存），即使是7B模型，原生精度也需约14GB显存，8GB显存设备直接超出上限。模型精度过高：vLLM默认使用FP16精度运行模型，相比FP32精度虽已减半显存占用，但对小显存设备（8GB及以下）仍不友好；若未开启量化，原生精度的显存占用会直接超出设备承载能力。2. vLLM优化策略的显存预留过高vLLM为提升推理速度，会默认预留一定比例的显存用于优化（如CUDA图、PagedAttention分页机制），其中最关键的是「GPU显存利用率」参数：默认显存利用率过高：vLLM默认gpu_memory_utilization=0.85，即会尝试占用GPU 85%的显存用于模型加载和优化。例如8GB显存设备，默认需要占用约6.8GB显存，若设备本身可用显存不足（如被其他进程占用后剩余6.7GB），就会直接报错。CUDA图优化的显存预留：vLLM默认启用CUDA图优化，会额外预留部分显存用于加速推理，对于8GB等小显存设备，这部分预留会进一步挤压模型可用显存。3. 任务负载及环境干扰除模型和优化策略外，外部环境和任务设置也会占用显存，导致可用显存不足：其他进程占用显存：GPU上同时运行其他程序（如Jupyter内核、其他Python进程、PyTorch缓存），会占用部分显存，导致vLLM可使用的显存减少（如你之前的报错中，8GB显存仅剩余6.7GB可用）。任务参数设置不合理：如单批次token数（max-num-batched-tokens）设置过高、并发请求过多，会导致推理过程中显存占用临时飙升，超出预留显存。分布式配置错误：若误设置tensor-parallel-size（张量并行数）大于1，单卡会尝试加载部分模型参数+通信开销，额外增加显存占用，单卡部署时此参数必须为1。4. 次要原因：环境配置或版本冲突vLLM与PyTorch、CUDA版本不兼容，或未清理显存缓存，会导致显存泄露、占用异常，间接引发显存不足（如你之前遇到的NCCL进程组未销毁警告，虽不直接导致显存不足，但会造成显存泄露，长期运行会加剧问题）。二、显存不足的解决方法（按优先级从易到难）解决思路：先快速释放可用显存、降低显存预留，再通过量化、调整参数进一步优化，最后排查环境和模型问题，确保每一步都可落地、见效快。1. 快速释放显存（立即见效，优先操作）针对“其他进程占用显存”“显存缓存”问题，快速释放可用显存，无需修改模型或配置：关闭占用GPU的其他进程查看GPU进程：执行命令 nvidia-smi，查看占用GPU的进程ID（PID）。杀死进程：执行命令 kill -9 <PID>（替换<PID>为实际进程号，如Jupyter、多余Python进程）。清理PyTorch显存缓存：若之前运行过PyTorch程序，执行以下代码清理缓存（可在启动vLLM前运行）：`import torchtorch.cuda.empty_cache() # 清理未使用的显存缓存torch.cuda.reset_max_memory_allocated() # 重置显存统计`重启GPU相关服务：若显存泄露严重，可重启电脑或GPU驱动（执行 nvidia-smi -r，部分系统需sudo权限）。2. 降低vLLM显存预留（适配小显存设备，核心操作）针对“显存利用率过高”“CUDA图预留”问题，修改vLLM启动参数，直接减少显存占用，适配8GB等小显存设备：降低GPU显存利用率（最关键）启动vLLM时，通过 --gpu-memory-utilization 参数降低显存占用比例，建议8GB显存设备设置为0.7~0.8（根据可用显存调整）。示例启动命令（OpenAI兼容服务器）：`python -m vllm.entrypoints.openai.api_server \ --model 你的模型名 \--gpu-memory-utilization 0.75 # 8GB显存推荐0.7~0.8，减少预留显存 --tensor-parallel-size 1 # 单卡必须设为1，避免额外显存开销`禁用CUDA图优化 vLLM默认启用CUDA图加速推理，会预留部分显存，禁用后可释放约500MB~1GB显存，适合小显存设备：`python -m vllm.entrypoints.openai.api_server \--model 你的模型名 \--gpu-memory-utilization 0.75 \ --disable-cuda-graphs # 禁用CUDA图，减少显存预留 --enforce-eager # 可选：强制即时执行，进一步降低显存占用（牺牲少量速度）`3. 量化模型（大幅减少显存占用，进阶操作）若降低显存预留后仍不足，通过模型量化将显存占用减少50%~75%，是小显存设备运行大模型的核心方案，vLLM支持AWQ、GPTQ、SqueezeLLM等量化格式：使用量化模型（推荐） 直接下载已量化的模型（如Hugging Face上的AWQ/GPTQ量化版模型，标注“4-bit”“8-bit”），启动时指定量化格式即可：`# 8-bit AWQ量化（兼容性好，显存占用减少约50%）-m vllm.entrypoints.openai.api_server \--model 模型名（如TheBloke/Llama-2-7B-Chat-AWQ） \ --gpu-memory-utilization 0.75 \ --quantization awq \ --tensor-parallel-size 1`注意：量化格式需与模型匹配（AWQ模型对应--quantization awq，GPTQ模型对应--quantization gptq）。原生模型量化（若没有量化版模型） 若只有原生模型，可使用vLLM自带的量化工具，或借助第三方工具（如auto-gptq）先量化模型，再启动vLLM（步骤略复杂，适合有一定技术基础的用户）。 4. 调整任务参数（减少运行时显存波动）即使模型加载成功，运行时若任务参数设置不合理，也会出现显存飙升、报错的情况，需调整以下参数：降低单批次token数：通过 --max-num-batched-tokens 限制单批次处理的token总数，减少推理时的显存占用（8GB显存推荐设为512~1024）： `--max-num-batched-tokens 512` 限制并发请求数：若通过API部署，可通过 --max-num-seqs 限制同时处理的请求数，避免并发过高导致显存叠加： `--max-num-seqs 4 # 8GB显存推荐1~4，根据请求复杂度调整` 缩短生成token长度：在调用模型时，限制 max_tokens（生成的最大token数），避免生成过长文本导致显存飙升（如设为128~256）。5. 排查环境与版本（解决隐性显存问题）若以上方法均无效，需排查环境配置，避免版本冲突或显存泄露：适配vLLM、PyTorch、CUDA版本：vLLM对版本要求较高，推荐组合：CUDA：11.8+（推荐12.1，兼容性最好）PyTorch：2.0+vLLM：0.4.0+（最新版本修复了部分显存泄露问题）解决NCCL进程组警告：若出现“destroy_process_group() was not called”警告，需在自定义代码中添加进程销毁逻辑，避免显存泄露：`import torch.distributed as dist 在代码结束前添加（仅当使用分布式部署时）if dist.is_initialized():dist.destroy_process_group()`注：若为单卡部署，此警告多为模型启动失败导致，解决显存问题后会自动消失。 6. 终极方案：更换模型或硬件若以上方法均无法解决，说明当前硬件或模型不匹配，可选择：更换更小的模型：如8GB显存设备，优先选择7B量化版模型（4-bit量化后仅需3~4GB显存），避免使用13B及以上规模模型。升级GPU硬件：若需运行更大模型（如13B、34B），建议更换16GB及以上显存的GPU（如RTX 3090、RTX 4090）。三、常见场景适配示例（直接套用）场景1：8GB显存，运行7B原生模型（报错“显存不足”）# 推荐启动命令（降低显存利用率+禁用CUDA图） python -m vllm.entrypoints.openai.api_server \ --model 你的7B原生模型名 \ --gpu-memory-utilization 0.75 \ --disable-cuda-graphs \ --enforce-eager \ --tensor-parallel-size 1 \ --max-num-batched-tokens 512场景2：8GB显存，运行7B AWQ量化模型（仍报错）# 推荐启动命令（进一步降低显存利用率+限制并发） python -m vllm.entrypoints.openai.api_server \ --model TheBloke/Llama-2-7B-Chat-AWQ \ --gpu-memory-utilization 0.7 \ --quantization awq \ --tensor-parallel-size 1 \ --max-num-batched-tokens 512 \ --max-num-seqs 2四、注意事项优先级：先释放显存、降低显存利用率，再尝试量化，最后调整任务参数，避免盲目操作。权衡速度与显存：降低显存利用率、禁用CUDA图、量化模型，都会牺牲少量推理速度，需根据需求平衡（如追求速度可适当提高显存利用率，追求稳定性则降低）。模型兼容性：量化模型需与vLLM支持的量化格式匹配，否则会报错；部分小众模型可能不支持量化，需更换模型。总结：vLLM显存不足的核心是“显存需求＞设备可用显存”，解决的关键的是“减少显存需求”（降低预留、量化模型、调整参数）和“增加可用显存”（清理进程、清理缓存），结合实际硬件和模型选择对应方法，即可快速解决问题。（注：文档部分内容可能由 AI 生成）
Hugging Face 下载vLLm使用的模型前往Hugging Face 寻找自己需要的模型例如：下载这个模型 https://huggingface.co/Qwen/Qwen3-VL-4B-Instruct复制模型名称 Qwen/Qwen3-VL-4B-Instruct下载模型使用 huggingface-cli 命令行工具（推荐）pip install huggingface_hub登录 Hugging Face（如果是私有模型或需要避免速率限制）公共模型可以不用huggingface-cli login然后输入你的 Access Token (https://huggingface.co/settings/tokens)下载模型：export HF_ENDPOINT=https://hf-mirror.comhf download Qwen/Qwen3-VL-4B-Instruct --local-dir Qwen3-VL-4B普通版本的模型会默认占用90%的显存，可能会出现显存不够的问题，可以选择量化版本的模型进行下载mac 安装vllm 下载模型流程#创建虚拟环境 conda create --name vllm_dev python=3.12 #使用环境 conda activate vllm_dev #安装huggingface_hub pip install huggingface_hub #下载模型 export HF_ENDPOINT=https://hf-mirror.com hf download Qwen/Qwen3-Embedding-0.6B
vllm安装运行本地大模型 - windows篇前言目前 vLLM 对 Windows 的原生支持（Native Windows Support）非常有限且处于早期实验阶段，而在 WSL2 环境下则非常成熟和稳定，所以在windows电脑上运行vllm,尽可能的需要安装wsl2WSL安装及相关命令官网地址 https://learn.microsoft.com/zh-cn/windows/wsl/install一、WSL 基础安装与版本管理命令功能说明wsl --install一键安装 WSL（默认安装 Ubuntu，自动启用所需系统组件，无需手动配置）wsl --install -d <发行版名称>安装指定发行版（示例：wsl --install -d Debian，可通过 wsl --list --online 查看支持的发行版）wsl --list --online / wsl -l -o列出微软官方支持的所有 WSL 发行版，方便选择安装wsl --set-default-version <版本号>设置 WSL 默认版本（仅支持 1 或 2，示例：wsl --set-default-version 2，建议优先选择 WSL2，性能更优）wsl --update一键更新 WSL 内核和相关组件，修复已知兼容性问题wsl --status查看当前 WSL 状态，包括默认版本、默认发行版、内核版本等关键信息二、WSL 实例管理命令功能说明wsl --list --verbose / wsl -l -v列出已安装的所有 WSL 实例，清晰显示实例名称、状态（运行/停止）、WSL 版本wsl / wsl.exe快速启动默认 WSL 发行版，直接进入 Linux 终端环境wsl -d <发行版名称>启动指定 WSL 发行版（示例：wsl -d Ubuntu-22.04，需与已安装实例名称完全一致）wsl -u <用户名> / wsl --user <用户名>以指定用户身份启动 WSL（示例：wsl -u root，即 root 管理员身份，方便执行高权限操作）wsl --set-default <发行版名称>设置默认启动的 WSL 发行版，后续输入 wsl 可直接启动该实例wsl --shutdown关闭所有运行中的 WSL 实例和 WSL 虚拟机，释放系统资源，无需逐个终止wsl -t <发行版名称> / wsl --terminate <发行版名称>终止指定的 WSL 实例（强制关闭，适用于单个实例卡死或需单独停止的场景）三、WSL 版本切换与迁移命令功能说明wsl --set-version <发行版名称> <版本号>将指定发行版切换为 WSL1 或 WSL2（示例：wsl --set-version Ubuntu 2，切换过程可能需要几分钟）wsl --export <发行版名称> <导出路径>导出 WSL 实例为备份文件（示例：wsl --export Ubuntu D:\ubuntu-backup.tar，用于数据备份或迁移）wsl --import <新实例名称> <安装路径> <备份文件路径> --version <版本号>导入备份的 WSL 实例（示例：wsl --import Ubuntu-22.04 D:\WSL\Ubuntu D:\ubuntu-backup.tar --version 2，恢复备份或迁移到其他路径）四、WSL 卸载与重置命令功能说明wsl --unregister <发行版名称>卸载指定 WSL 实例（删除该实例所有数据，不可恢复，示例：wsl --unregister Ubuntu）wsl --cleanup清理 WSL 无用资源（如未使用的内核、临时文件），释放磁盘空间五、WSL 高级操作命令功能说明wsl <Linux 命令>在 PowerShell 中直接执行单个 Linux 命令（示例：wsl ls -l /home，无需进入 WSL 终端）wsl --mount <磁盘路径>挂载 Windows 磁盘到 WSL（示例：wsl --mount \\.\PhysicalDrive1，需管理员权限，方便在 WSL 中访问 Windows 磁盘文件）wsl --unmount <磁盘路径>卸载已挂载的 Windows 磁盘（示例：wsl --unmount \\.\PhysicalDrive1，避免磁盘占用）Get-ChildItem HKCU:\Software\Microsoft\Windows\CurrentVersion\Lxss\查看 WSL 实例的注册表配置（高级排查用，用于解决实例启动失败等异常问题）vLLm安装WSL 安装Ubuntuwsl -d Ubuntu WSL- Ubuntu安装vLLm创建虚拟环境// 1. 备份原有源 sudo cp /etc/apt/sources.list /etc/apt/sources.list.bak // 2. 替换为清华大学镜像源或阿里源 sudo sed -i 's/archive.ubuntu.com/mirrors.tuna.tsinghua.edu.cn/g' /etc/apt/sources.list sudo sed -i 's/security.ubuntu.com/mirrors.tuna.tsinghua.edu.cn/g' /etc/apt/sources.list //更新 sudo apt update && sudo apt upgrade //下载python3 sudo apt upgrade python3 //下载pip sudo apt install python3-pip //创建虚拟环境文件夹 mkdir vllm_dev //创建虚拟环境 cd vllm_dev/ python3 -m venv .venv //启动虚拟环境 source .venv/bin/activate 下载vLLmpip更换下载源 92 mkdir -p ~/.pip 93 # 写入清华源配置 94 cat > ~/.pip/pip.conf <<EOF 95 [global] 96 index-url = https://pypi.tuna.tsinghua.edu.cn/simple 97 trusted-host = pypi.tuna.tsinghua.edu.cn 98 [install] 99 trusted-host = pypi.tuna.tsinghua.edu.cn 100 EOF 下载vllm//查看是否安转cuda nvidia-smi //安装vLLm pip install vllm //安装openai pip install openai //安装huggingface_hub pip install huggingface_hub //下载模型 hf download cyankiwi/Qwen3-VL-4B-Instruct-AWQ-4bit --local-dir ./Qwen3-VL-4B-Instruct/ 启动vllmpython -m vllm.entrypoints.openai.api_server \ --model /home/t-mac/code/vllm_dev/models/Qwen3-VL-4B-Instruct/Qwen3-VL-4B-Instruct \ --host 0.0.0.0 \ --port 8000 \ --gpu-memory-utilization 0.85 \ --max-model-len 4096 \ --trust-remote-code \ --limit-mm-per-prompt '{"image": 4}' \ --served-model-name Qwen3-VL-4B-Instruct # 新增：指定对外暴露的模型名（必填）
LangChain 1.0 + LangGraph 1.0 全栈学习指南本指南面向有基础Python编程能力、希望系统掌握LangChain/LangGraph 1.0+生态的开发者，所有示例均基于2026年最新稳定版API，摒弃所有已废弃的旧写法（如langchain.chains、旧版Graph类），聚焦LCEL和StateGraph核心范式，兼顾理论精简性与实战可操作性，帮助开发者快速从入门到精通，重点掌握基于LangGraph的状态机和多代理编排能力。第一章：核心概念重构（LangChain 1.0+ 与 LangGraph 1.0+ 重大变更）LangChain 1.0+ 和 LangGraph 1.0+ 并非简单的版本迭代，而是对核心架构、API设计和开发范式的全面重构，核心目标是提升类型安全、异步性能和可扩展性，彻底解决旧版本“灵活但混乱”的痛点。以下是最影响开发的3个重大变更：1.1 包结构与API精简（彻底废弃旧版Chain）旧版本中langchain.chains模块包含大量冗余实现（如LLMChain、SequentialChain），1.0+ 版本已完全废弃该模块，统一采用 LCEL（LangChain Expression Language） 作为核心编排范式。同时包结构大幅精简，核心模块如下（重点记忆）：langchain_core：核心抽象（如BaseLLM、PromptTemplate、Tool、LCEL表达式），是所有模块的基础。langchain_llms/langchain_chat_models：LLM/聊天模型集成（如OpenAI、Anthropic），替代旧版langchain.llms。langchain_tools：工具集成（如搜索、数据库、文件操作），替代旧版langchain.tools。langgraph：独立拆分的状态机/多代理框架，1.0+ 版本完全基于StateGraph重构，不再依赖LangChain核心包的内部逻辑。关键提醒：1.0+ 版本中，任何基于langchain.chains的代码都会报错，必须用LCEL表达式替代；LangGraph不再是LangChain的子模块，需单独安装（pip install langgraph）。1.2 异步支持增强（原生async/await，性能翻倍）旧版本异步API（如async_run）为兼容写法，性能有限；1.0+ 版本全面拥抱Python原生异步，所有核心组件（LLM调用、工具调用、LCEL编排、LangGraph执行）均提供原生async/await支持，无需额外封装。核心优势：异步执行可同时处理多个LLM/工具调用，在多代理、多任务场景下，性能较同步执行提升50%以上，是生产环境必备写法。1.3 类型安全升级（Pydantic v2 深度集成）1.0+ 版本强制要求类型注解，全面集成 Pydantic v2，替代旧版的TypedDict和简单类型提示。无论是LLM输出解析、状态定义，还是工具输入输出，都需通过Pydantic模型进行类型校验，大幅减少运行时错误。核心变化：LangChain 1.0+ 所有核心类（如PromptTemplate、Tool）均继承自Pydantic v2的BaseModel，支持模型验证、序列化/反序列化，与LangGraph的状态管理无缝衔接。第二章：LangChain 1.0 基础（LCEL核心范式）LangChain 1.0+ 的核心是 LCEL，它是一种声明式的表达式语言，用于编排LLM、Prompt、Parser、Tool等组件，替代旧版的Chain写法。本节快速回顾基础组件的最新定义方式，所有示例均为1.0+ 标准写法。2.1 环境准备与依赖安装首先安装1.0+ 版本的核心依赖（指定版本以确保兼容性）：# 安装LangChain核心包（1.0+） pip install langchain-core==1.0.10 langchain-openai==0.1.0 langchain-tools==0.1.0 # 安装LangGraph（1.0+） pip install langgraph==1.0.5 # 安装Pydantic v2（LangChain 1.0+ 依赖） pip install pydantic==2.6.1注：这里以OpenAI模型为例，其他模型（如Anthropic、本地化模型）安装对应包即可（如langchain-anthropic）。2.2 LLM/聊天模型调用（最新写法）LangChain 1.0+ 将LLM和聊天模型分开管理，分别对应langchain_llms和langchain_chat_models，支持同步/异步调用，且必须指定类型注解。示例1：聊天模型（最常用，如gpt-4o）from langchain_chat_models import ChatOpenAI from pydantic import BaseModel, Field from typing import Optional # 1. 初始化聊天模型（1.0+ 写法，支持Pydantic配置） chat_model = ChatOpenAI( model="gpt-4o", temperature=0.7, api_key="your-openai-api-key" # 生产环境建议用环境变量加载 ) # 2. 同步调用（基础写法） response = chat_model.invoke("请简要介绍LangChain 1.0的核心变化") print("同步调用结果：", response.content) # 3. 异步调用（生产环境推荐，需用async/await） async def async_llm_call(): response = await chat_model.ainvoke("请简要介绍LangChain 1.0的核心变化") return response.content # 执行异步函数（需在async环境中运行） # import asyncio # print("异步调用结果：", asyncio.run(async_llm_call()))2.3 Prompt Template（模板定义与渲染）1.0+ 版本的PromptTemplate继承自Pydantic v2，支持动态参数、模板验证，且可直接与LCEL表达式结合。from langchain_core.prompts import PromptTemplate # 1. 定义Prompt模板（支持参数校验，如required=True） prompt = PromptTemplate( input_variables=["topic", "version"], # 模板参数 template="请详细介绍{topic}在LangChain {version}版本中的使用方法，要求简洁、实战，重点突出核心用法。", ) # 2. 渲染模板（生成最终Prompt文本） rendered_prompt = prompt.invoke({"topic": "PromptTemplate", "version": "1.0+"}) print("渲染后的Prompt：", rendered_prompt.to_string()) # 3. 与LLM结合（LCEL表达式写法，用 | 连接） chain = prompt | chat_model # LCEL核心：组件串联 response = chain.invoke({"topic": "PromptTemplate", "version": "1.0+"}) print("Prompt+LLM调用结果：", response.content)关键提醒：LCEL的核心是“组件串联”，用|符号将多个组件（Prompt、LLM、Parser等）连接成一个可执行的“链”，替代旧版的LLMChain。2.4 Output Parser（输出解析，Pydantic v2 集成）1.0+ 版本推荐使用Pydantic v2模型作为输出解析器，替代旧版的StrOutputParser、JsonOutputParser，实现类型安全的输出提取。from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import List # 1. 定义Pydantic模型（指定输出格式和类型） class LLMResponse(BaseModel): core_points: List[str] = Field(description="核心要点，至少3条，每条不超过50字") summary: str = Field(description="简要总结，不超过200字") # 2. 初始化输出解析器 parser = PydanticOutputParser(pydantic_object=LLMResponse) # 3. 构建LCEL链（Prompt + LLM + Parser） # 注意：Prompt中需加入解析器的格式说明 prompt_with_parser = PromptTemplate( input_variables=["topic"], template="请介绍{topic}的核心用法，输出格式必须符合以下要求：\n{format_instructions}", partial_variables={"format_instructions": parser.get_format_instructions()}, # 注入格式说明 ) # 串联组件（LCEL核心写法） chain = prompt_with_parser | chat_model | parser # 调用链，直接得到Pydantic模型实例（类型安全） response = chain.invoke({"topic": "LangChain 1.0 LCEL"}) print("核心要点：", response.core_points) print("总结：", response.summary) print("类型验证：", isinstance(response, LLMResponse)) # 输出True2.5 Tools（工具定义与调用）1.0+ 版本的Tool同样基于Pydantic v2，支持异步调用、输入验证，且可直接集成到LCEL链或LangGraph中。以下以“搜索工具”为例（需安装langchain-tools[serpapi]）。from langchain_core.tools import tool from pydantic import BaseModel, Field import serpapi # 需安装：pip install serpapi # 1. 用装饰器定义工具（1.0+ 推荐写法，自动生成工具描述和输入验证） class SearchInput(BaseModel): query: str = Field(description="搜索关键词，必须具体、明确，如'LangChain 1.0 LCEL 用法'") @tool(args_schema=SearchInput, return_direct=False) def search_tool(query: str) -> str: """用于搜索互联网最新信息，适用于获取实时数据、最新技术文档等场景。""" # 模拟搜索（实际使用时替换为真实搜索逻辑，如SerpAPI） try: # 这里用占位符模拟搜索结果，实际可替换为SerpAPI调用 return f"关于'{query}'的搜索结果：LangChain 1.0 LCEL是声明式编排语言，支持组件串联，替代旧版Chain..." except Exception as e: return f"搜索失败：{str(e)}" # 2. 同步调用工具 search_result = search_tool.invoke({"query": "LangChain 1.0 LCEL 核心用法"}) print("工具调用结果：", search_result) # 3. 异步工具调用（需用@atool装饰器） from langchain_core.tools import atool @atool(args_schema=SearchInput, return_direct=False) async def async_search_tool(query: str) -> str: """异步搜索工具，适用于多任务并行场景。""" return await search_tool.ainvoke({"query": query}) # 异步调用 # 4. 工具与LLM结合（LCEL链） # 定义Prompt，让LLM根据问题决定是否调用工具（后续LangGraph会更灵活） tool_prompt = PromptTemplate( input_variables=["question"], template="请判断是否需要调用搜索工具回答以下问题：{question}\n如果需要，输出搜索关键词；如果不需要，直接回答问题。", ) tool_chain = tool_prompt | chat_model | parser # 结合解析器，提取搜索关键词 response = tool_chain.invoke({"question": "LangChain 1.0 与 0.1 版本的核心区别是什么？"}) print("LLM决策结果：", response)第三章：LangGraph 1.0 核心机制（状态机与多代理基础）LangGraph 1.0+ 是独立于LangChain核心的状态机框架，核心是 StateGraph（状态图），用于编排多节点（Nodes）、定义状态流转（Edges），实现复杂的多代理逻辑、循环执行、条件路由等场景。相比旧版LangGraph，1.0+ 版本更轻量、更灵活，且与Pydantic v2深度集成。核心逻辑：将整个任务拆解为多个“节点”（如搜索、总结、反思），定义“状态”（任务的中间结果、上下文），通过“边”（状态流转规则）连接节点，形成一个可循环、可回溯的状态机。3.1 核心概念拆解State（状态）：整个状态机的核心数据载体，存储任务的所有中间结果（如搜索结果、总结内容、反思意见），1.0+ 版本推荐用Pydantic v2模型定义（替代旧版TypedDict），支持类型校验和数据持久化。Node（节点）：状态机的执行单元，本质是一个函数（同步/异步），接收当前状态，执行具体逻辑（如调用工具、LLM总结），并返回更新后的状态。Edge（边）：节点之间的流转规则，分为“直接边”（固定流转，如A→B）和“条件边”（根据状态判断流转方向，如A→B或A→C）。StateGraph（状态图）：LangGraph的核心类，用于组装节点、定义边、管理状态流转，最终通过compile()方法生成可执行的状态机。Checkpointer（检查点）：用于持久化状态机的中间状态，支持“断点续跑”“时间旅行”（回溯历史状态），是生产环境必备组件。3.2 State 定义（Pydantic v2 实战）1.0+ 版本中，State必须是可序列化的，推荐使用Pydantic v2的BaseModel定义，明确每个字段的类型和描述，确保状态流转的类型安全。from pydantic import BaseModel, Field from typing import List, Optional # 定义状态模型（Pydantic v2，替代旧版TypedDict） class ResearchState(BaseModel): """研究代理的状态模型，存储所有中间结果和任务信息""" topic: str = Field(description="用户输入的研究主题，不可修改") search_queries: List[str] = Field(default_factory=list, description="生成的搜索关键词列表") search_results: List[str] = Field(default_factory=list, description="搜索工具返回的结果列表") summary: Optional[str] = Field(default=None, description="对搜索结果的总结内容") reflection: Optional[str] = Field(default=None, description="对总结内容的反思意见（是否需要进一步搜索）") is_complete: bool = Field(default=False, description="任务是否完成，完成则终止状态机") # 示例：初始化状态 initial_state = ResearchState(topic="LangGraph 1.0 状态机用法") print("初始状态：", initial_state.model_dump()) # Pydantic序列化，方便持久化3.3 Node 定义（同步/异步节点）节点是状态机的执行单元，函数接收当前状态（Pydantic模型实例），返回更新后的状态。支持同步节点（def）和异步节点（async def），1.0+ 版本推荐使用异步节点以提升性能。from langchain_chat_models import ChatOpenAI # 初始化LLM（复用之前的聊天模型） chat_model = ChatOpenAI(model="gpt-4o", temperature=0.7) # 1. 同步节点：生成搜索关键词（示例） def generate_search_queries(state: ResearchState) -> ResearchState: """生成搜索关键词的节点：根据研究主题，生成3-5个具体的搜索关键词""" prompt = PromptTemplate( input_variables=["topic"], template="请为研究主题'{topic}'生成3-5个具体的搜索关键词，用于获取最新、最相关的信息，关键词用逗号分隔，不要多余内容。", ) # 构建LCEL链，生成搜索关键词 chain = prompt | chat_model | lambda x: x.content.split(",") # 简单解析 queries = chain.invoke({"topic": state.topic}) # 更新状态：添加搜索关键词 state.search_queries = [q.strip() for q in queries if q.strip()] return state # 2. 异步节点：执行搜索（推荐，支持并行） async def execute_search(state: ResearchState) -> ResearchState: """执行搜索的节点：根据搜索关键词，调用异步搜索工具，获取搜索结果""" # 遍历所有搜索关键词，异步调用搜索工具（并行执行，提升效率） search_tasks = [async_search_tool.ainvoke({"query": q}) for q in state.search_queries] # 等待所有搜索任务完成 search_results = await asyncio.gather(*search_tasks) # 更新状态：添加搜索结果 state.search_results.extend(search_results) return state # 3. 异步节点：总结搜索结果 async def summarize_results(state: ResearchState) -> ResearchState: """总结节点：将所有搜索结果汇总，生成简洁、全面的总结""" prompt = PromptTemplate( input_variables=["topic", "search_results"], template="请根据以下搜索结果，为研究主题'{topic}'生成一份简洁、全面的总结，要求逻辑清晰，重点突出，不超过500字：\n{search_results}", ) # 构建异步LCEL链（用ainvoke） chain = prompt | chat_model summary = await chain.ainvoke({"topic": state.topic, "search_results": "\n".join(state.search_results)}) # 更新状态：添加总结内容 state.summary = summary.content return state3.4 Edge 定义（直接边与条件边）边定义节点之间的流转规则，1.0+ 版本通过add_edge()（直接边）和add_conditional_edges()（条件边）两种方法定义，其中条件边是实现复杂逻辑的核心。from langgraph import StateGraph # 1. 初始化状态图（指定状态模型） graph = StateGraph(ResearchState) # 2. 添加节点（指定节点名称和对应的函数） graph.add_node("generate_queries", generate_search_queries) # 生成搜索关键词节点 graph.add_node("execute_search", execute_search) # 执行搜索节点 graph.add_node("summarize", summarize_results) # 总结节点 graph.add_node("reflect", reflect_on_summary) # 反思节点（后续定义） # 3. 定义直接边（固定流转） # 生成搜索关键词 → 执行搜索 → 总结 graph.add_edge("generate_queries", "execute_search") graph.add_edge("execute_search", "summarize") # 4. 定义条件边（根据状态判断流转方向） # 总结节点之后，进入反思节点，反思后判断是否需要重新搜索 def should_continue(state: ResearchState) -> str: """条件路由函数：根据反思结果，判断流转方向""" # 反思内容中包含"需要进一步搜索"则返回"generate_queries"，否则返回"end" if state.reflection and "需要进一步搜索" in state.reflection: return "generate_queries" # 重新生成搜索关键词，循环执行 else: state.is_complete = True # 标记任务完成 return "end" # 终止状态机 # 添加条件边：总结 → 反思 → （条件判断）生成关键词/终止 graph.add_edge("summarize", "reflect") # 条件边：从reflect节点出发，根据should_continue的返回值，流转到指定节点 graph.add_conditional_edges( source="reflect", # 源节点 path=should_continue, # 条件路由函数 dests=["generate_queries", "end"] # 可能的目标节点（与路由函数返回值对应） ) # 5. 定义起始节点和终止节点 graph.set_entry_point("generate_queries") # 起始节点：生成搜索关键词 graph.set_finish_point("end") # 终止节点：任务完成3.5 Checkpointer（持久化检查点）Checkpointer用于持久化状态机的中间状态，支持断点续跑、时间旅行（回溯历史状态），LangGraph 1.0+ 提供多种Checkpointer实现，最常用的是MemorySaver（内存持久化，适合测试）和FileSystemCheckpointer（文件持久化，适合生产）。from langgraph.checkpoint.memory import MemorySaver from langgraph.checkpoint.filesystem import FileSystemCheckpointer # 1. 内存Checkpointer（适合测试，程序退出后状态丢失） memory_checkpointer = MemorySaver() # 2. 文件Checkpointer（适合生产，状态持久化到本地文件） file_checkpointer = FileSystemCheckpointer("./checkpoints") # 状态存储路径 # 3. 编译状态机（指定Checkpointer） # 测试用内存Checkpointer，生产用文件Checkpointer graph_compiled = graph.compile(checkpointer=memory_checkpointer) # 4. 执行状态机（带检查点，支持断点续跑） # 初始化状态 initial_state = ResearchState(topic="LangGraph 1.0 状态机用法") # 执行状态机（返回迭代器，可逐步获取每个节点的执行结果） async for state in graph_compiled.astream(initial_state, config={"configurable": {"thread_id": "research_1"}}): # thread_id：用于区分不同的任务实例，确保检查点隔离 print(f"当前节点：{state['__step__']['node']}") # 获取当前执行的节点名称 print(f"当前状态：{state.model_dump(exclude_none=True)}") # 打印当前状态（排除None字段） print("-" * 50) # 5. 断点续跑（无需重新执行所有节点，从上次中断的节点继续） # async for state in graph_compiled.astream(None, config={"configurable": {"thread_id": "research_1"}}): # print(f"续跑节点：{state['__step__']['node']}") # print(f"续跑状态：{state.model_dump(exclude_none=True)}")关键提醒：thread_id 是检查点的核心，用于区分不同的任务实例，同一个thread_id 可以实现断点续跑，不同thread_id 状态相互隔离。第四章：实战项目 - 构建一个自我修正的研究代理本节基于前面的知识点，构建一个完整的自我修正研究代理，实现“用户输入研究主题 → 生成搜索关键词 → 执行搜索 → 总结内容 → 反思修正 → 循环优化 → 输出最终报告”的全流程，包含完整的多节点图、异步执行和持久化检查点。4.1 项目需求与架构设计需求说明用户输入一个研究主题（如“LangGraph 1.0 多代理编排实践”）。代理自动生成搜索关键词，调用搜索工具获取最新信息。对搜索结果进行总结，然后自我反思（判断总结是否全面、是否需要补充搜索）。如果需要补充搜索，重新生成搜索关键词，循环执行；如果不需要，输出最终报告。支持断点续跑（通过Checkpointer），支持异步执行以提升效率。节点架构整个状态机包含5个核心节点，流转逻辑如下：起始 → 生成搜索关键词 → 执行搜索 → 总结结果 → 反思修正 → （条件判断）重新生成关键词/输出报告 → 终止4.2 完整代码实现（生产级，含注释）import asyncio from pydantic import BaseModel, Field from typing import List, Optional from langchain_core.prompts import PromptTemplate from langchain_chat_models import ChatOpenAI from langchain_core.tools import atool from langgraph import StateGraph from langgraph.checkpoint.filesystem import FileSystemCheckpointer # -------------------------- 1. 环境初始化 -------------------------- # 初始化聊天模型（异步支持） chat_model = ChatOpenAI( model="gpt-4o", temperature=0.7, api_key="your-openai-api-key" # 生产环境建议用环境变量：os.getenv("OPENAI_API_KEY") ) # 初始化文件检查点（持久化状态，支持断点续跑） checkpointer = FileSystemCheckpointer("./research_agent_checkpoints") # -------------------------- 2. 状态定义（Pydantic v2） -------------------------- class ResearchState(BaseModel): """研究代理的状态模型，存储所有中间结果和任务状态""" # 核心输入（不可修改） topic: str = Field(description="用户输入的研究主题，贯穿整个任务周期") # 中间结果 search_queries: List[str] = Field(default_factory=list, description="生成的搜索关键词列表，用于获取相关信息") search_results: List[str] = Field(default_factory=list, description="搜索工具返回的结果列表，每条为一个搜索结果") summary: Optional[str] = Field(default=None, description="对搜索结果的汇总总结，逐步优化") reflection: Optional[str] = Field(default=None, description="对当前总结的反思意见，判断是否需要补充搜索") # 任务状态 is_complete: bool = Field(default=False, description="任务是否完成，True则终止状态机") report: Optional[str] = Field(default=None, description="最终输出的研究报告") # -------------------------- 3. 工具定义（异步搜索工具） -------------------------- class SearchInput(BaseModel): """搜索工具的输入模型，用于类型校验""" query: str = Field(description="搜索关键词，必须具体、明确，避免模糊表述，如'LangGraph 1.0 子图用法'") @atool(args_schema=SearchInput, return_direct=False) async def async_search_tool(query: str) -> str: """异步搜索工具，模拟调用互联网搜索接口（实际可替换为SerpAPI、Google Search等）""" try: # 模拟搜索逻辑（实际生产环境替换为真实API调用） # 这里用占位符返回搜索结果，实际可根据query调用SerpAPI获取实时数据 search_result = ( f"关于'{query}'的搜索结果：\n" "1. LangGraph 1.0 支持子图（Subgraph）和并行执行，可将复杂任务拆解为子任务编排。\n" "2. 异步节点是LangGraph 1.0的核心特性，可通过async/await实现多任务并行，提升执行效率。\n" "3. Checkpointer支持文件持久化，可实现断点续跑和时间旅行，适合长期运行的任务。" ) return search_result except Exception as e: return f"搜索失败：{str(e)}（请检查搜索关键词是否合理，或网络是否正常）" # -------------------------- 4. 节点定义（异步节点，提升性能） -------------------------- async def generate_search_queries(state: ResearchState) -> ResearchState: """节点1：生成搜索关键词""" # 定义Prompt，让LLM根据研究主题生成3-5个具体的搜索关键词 prompt = PromptTemplate( input_variables=["topic", "summary", "reflection"], template=""" 你需要为研究主题'{topic}'生成3-5个具体的搜索关键词，用于获取最新、最相关的信息。 注意以下要求： 1. 关键词必须具体，避免模糊（如不要只写'LangGraph'，要写'LangGraph 1.0 子图用法'）。 2. 如果有之前的总结（{summary}）和反思（{reflection}），请结合反思意见，补充相关关键词（如反思指出缺少并行执行内容，就生成相关关键词）。 3. 关键词用逗号分隔，不要多余内容，不要解释。 """, ) # 异步调用LLM，生成搜索关键词 chain = prompt | chat_model response = await chain.ainvoke({ "topic": state.topic, "summary": state.summary or "无", "reflection": state.reflection or "无" }) # 解析关键词，去重、去空 queries = [q.strip() for q in response.content.split(",") if q.strip()] # 去重（避免重复搜索） state.search_queries = list(set(queries))[:5] # 最多保留5个关键词 print(f"生成搜索关键词：{state.search_queries}") return state async def execute_search(state: ResearchState) -> ResearchState: """节点2：执行搜索（并行调用搜索工具）""" if not state.search_queries: state.search_results.append("未生成有效搜索关键词，无法执行搜索") return state # 并行调用异步搜索工具，提升效率（同时搜索所有关键词） search_tasks = [async_search_tool.ainvoke({"query": q}) for q in state.search_queries] # 等待所有搜索任务完成，获取结果 search_results = await asyncio.gather(*search_tasks) # 将搜索结果添加到状态中（去重） state.search_results = list(set(state.search_results + search_results)) print(f"执行搜索完成，共获取{len(state.search_results)}条结果") return state async def summarize_results(state: ResearchState) -> ResearchState: """节点3：总结搜索结果""" if not state.search_results: state.summary = f"研究主题'{state.topic}'：未获取到有效搜索结果，无法生成总结。" return state # 定义Prompt，让LLM汇总搜索结果，生成简洁、全面的总结 prompt = PromptTemplate( input_variables=["topic", "search_results"], template=""" 请根据以下搜索结果，为研究主题'{topic}'生成一份总结，要求： 1. 逻辑清晰，分点阐述核心内容（用数字序号），重点突出LangGraph 1.0+ 的特性。 2. 语言简洁，避免冗余，总字数不超过800字。 3. 保留关键信息，去掉无关内容，确保总结的准确性和实用性。 搜索结果： {search_results} """, ) # 异步调用LLM，生成总结 chain = prompt | chat_model response = await chain.ainvoke({ "topic": state.topic, "search_results": "\n\n".join(state.search_results) }) state.summary = response.content print(f"生成总结完成，总结长度：{len(state.summary)}字") return state async def reflect_on_summary(state: ResearchState) -> ResearchState: """节点4：反思总结，判断是否需要补充搜索""" if not state.summary: state.reflection = "当前无总结内容，需要重新执行搜索和总结。" return state # 定义Prompt，让LLM反思总结的完整性、准确性，判断是否需要补充搜索 prompt = PromptTemplate( input_variables=["topic", "summary"], template=""" 请作为一名资深AI架构师，反思以下关于'{topic}'的总结是否完整、准确： 总结内容： {summary} 反思要求： 1. 判断总结是否覆盖了该主题的核心知识点（如LangGraph 1.0的核心特性、用法、实战场景）。 2. 判断是否有遗漏的重要信息，是否需要进一步搜索补充。 3. 输出反思意见，明确说明“需要进一步搜索”或“无需进一步搜索”，并简要说明原因。 4. 若需要进一步搜索，简要提示需要补充的关键词方向。 """, ) # 异步调用LLM，生成反思意见 chain = prompt | chat_model response = await chain.ainvoke({ "topic": state.topic, "summary": state.summary }) state.reflection = response.content print(f"反思完成：{state.reflection[:100]}...") # 打印前100字，避免过长 return state async def generate_report(state: ResearchState) -> ResearchState: """节点5：生成最终报告""" if not state.summary: state.report = f"研究报告：{state.topic}\n\n未获取到有效信息，无法生成报告。" state.is_complete = True return state # 定义Prompt，将总结优化为正式的研究报告 prompt = PromptTemplate( input_variables=["topic", "summary", "reflection"], template=""" 请根据以下总结和反思意见，为研究主题'{topic}'生成一份正式的研究报告，要求： 1. 结构清晰：包含“研究主题”“核心总结”“反思与优化”三个部分。 2. 语言正式、专业，适合技术人员阅读，总字数不超过1000字。 3. 结合反思意见，说明总结的完整性，若有优化空间，简要提及。 总结内容： {summary} 反思意见： {reflection} """, ) # 异步调用LLM，生成最终报告 chain = prompt | chat_model response = await chain.ainvoke({ "topic": state.topic, "summary": state.summary, "reflection": state.reflection or "无" }) state.report = response.content state.is_complete = True # 标记任务完成 print("最终报告生成完成") return state # -------------------------- 5. 构建状态图（StateGraph） -------------------------- # 1. 初始化状态图，指定状态模型 graph = StateGraph(ResearchState) # 2. 添加所有节点（节点名称与函数对应） graph.add_node("generate_queries", generate_search_queries) # 生成搜索关键词 graph.add_node("execute_search", execute_search) # 执行搜索 graph.add_node("summarize", summarize_results) # 总结结果 graph.add_node("reflect", reflect_on_summary) # 反思修正 graph.add_node("generate_report", generate_report) # 生成最终报告 # 3. 定义直接边（固定流转） graph.add_edge("generate_queries", "execute_search") # 生成关键词 → 执行搜索 graph.add_edge("execute_search", "summarize") # 执行搜索 → 总结结果 graph.add_edge("reflect", "generate_report") # 反思 → 生成报告（默认流转） # 4. 定义条件边（反思后判断是否需要重新搜索） def should_continue(state: ResearchState) -> str: """条件路由函数：根据反思意见，判断流转方向""" if state.reflection and "需要进一步搜索" in state.reflection: # 需要补充搜索，重新生成关键词（清空之前的搜索结果，避免冗余） state.search_results = [] return "generate_queries" else: # 无需补充搜索，进入生成报告节点 return "generate_report" # 添加条件边：反思节点 → （条件判断）生成关键词/生成报告 graph.add_conditional_edges( source="reflect", path=should_continue, dests=["generate_queries", "generate_report"] ) # 5. 定义起始节点和终止节点 graph.set_entry_point("generate_queries") # 起始节点：生成搜索关键词 graph.set_finish_point("generate_report") # 终止节点：生成最终报告 # -------------------------- 6. 编译状态机并执行 -------------------------- # 编译状态机（指定检查点，支持断点续跑） graph_compiled = graph.compile(checkpointer=checkpointer) # 定义执行函数 async def run_research_agent(topic: str, thread_id: str = "research_task_1"): """运行研究代理，返回最终报告""" # 初始化状态 initial_state = ResearchState(topic=topic) print(f"开始研究任务：{topic}") print("=" * 60) # 异步执行状态机（流式获取每个节点的执行结果） async for state in graph_compiled.astream( initial_state, config={"configurable": {"thread_id": thread_id}} # 线程ID，用于检查点隔离 ): # 打印当前执行的节点和状态摘要 current_node = state["__step__"]["node"] print(f"【当前节点】：{current_node}") if current_node == "generate_queries": print(f"【关键词】：{state.search_queries}") elif current_node == "execute_search": print(f"【搜索结果数】：{len(state.search_results)}") elif current_node == "summarize": print(f"【总结摘要】：{state.summary[:50]}...") elif current_node == "reflect": print(f"【反思摘要】：{state.reflection[:50]}...") elif current_node == "generate_report": print(f"【最终报告】：\n{state.report}") print("-" * 60) # 返回最终状态（包含报告） final_state = await graph_compiled.get_state(config={"configurable": {"thread_id": thread_id}}) return final_state.report # -------------------------- 7. 运行测试 -------------------------- if __name__ == "__main__": # 运行研究代理，研究主题：LangGraph 1.0 多代理编排实践 report = asyncio.run(run_research_agent(topic="LangGraph 1.0 多代理编排实践")) # 保存报告到文件 with open("research_report.txt", "w", encoding="utf-8") as f: f.write(report) print("报告已保存到 research_report.txt")4.3 代码说明与测试要点异步优化：所有核心节点均为异步函数，搜索工具采用异步调用，支持多关键词并行搜索，大幅提升执行效率。状态管理：通过Pydantic v2模型定义状态，确保类型安全，所有中间结果均存储在状态中，便于调试和回溯。循环逻辑：通过条件边实现“反思→重新搜索”的循环，直到总结满足要求，实现自我修正。持久化：使用FileSystemCheckpointer将状态持久化到本地文件，支持断点续跑（如程序中断后，重新运行可从上次中断的节点继续）。测试方法：替换async_search_tool中的模拟逻辑为真实搜索API（如SerpAPI），即可获取实时数据；修改topic参数，可测试不同研究主题。第五章：高级特性（LangGraph 1.0+）LangGraph 1.0+ 提供了多种高级特性，用于解决复杂场景下的多代理编排问题，本节重点讲解最常用的4个高级特性：人类介入、时间旅行、子图、并行执行。5.1 人类介入（Human-in-the-loop）在多代理任务中，有时需要人类手动干预（如确认决策、补充信息），LangGraph 1.0+ 提供HumanMessage和HumanNode支持人类介入，暂停状态机执行，等待人类输入后继续。from langchain_core.messages import HumanMessage from langgraph.graph import END # 定义人类介入节点 async def human_in_the_loop(state: ResearchState) -> ResearchState: """人类介入节点：暂停状态机，等待人类输入反馈""" print("\n【人类介入】请查看当前总结，输入反馈（如'继续'、'需要补充搜索XX内容'）：") # 打印当前总结 print(f"当前总结：{state.summary}") # 获取人类输入 human_feedback = input("请输入反馈：") # 将人类反馈添加到状态中 state.reflection = f"人类反馈：{human_feedback}\n" + (state.reflection or "") return state # 将人类介入节点添加到状态图中 graph.add_node("human_in_loop", human_in_the_loop) # 修改流转逻辑：总结 → 人类介入 → 反思 graph.add_edge("summarize", "human_in_loop") graph.add_edge("human_in_loop", "reflect") # 重新编译状态机 graph_compiled = graph.compile(checkpointer=checkpointer)核心作用：在关键节点（如总结后）暂停状态机，让人类确认内容准确性，避免代理生成错误信息，适用于对准确性要求高的场景（如学术研究、商业决策）。5.2 时间旅行（Time Travel）时间旅行是Checkpointer的高级特性，支持回溯到状态机的任意历史状态，重新执行后续节点，用于调试、优化或修正错误。LangGraph 1.0+ 通过get_state_history()获取历史状态，通过astream()重新执行。async def time_travel_demo(thread_id: str): """时间旅行示例：回溯到历史状态，重新执行""" # 1. 获取该线程的所有历史状态（按执行顺序排列） history = await graph_compiled.get_state_history(config={"configurable": {"thread_id": thread_id}}) # 2. 打印所有历史状态的节点和时间 print("历史状态列表：") for i, state in enumerate(history): node = state["__step__"]["node"] timestamp = state["__step__"]["timestamp"] print(f"{i+1}. 节点：{node}，时间：{timestamp}") # 3. 选择一个历史状态（如第2个状态，即执行搜索后的状态） if len(history) >= 2: historical_state = history[1] # 索引从0开始，1表示第2个状态 print(f"\n回溯到状态：{historical_state['__step__']['node']}") # 4. 从该历史状态重新执行后续节点 async for state in graph_compiled.astream( historical_state, config={"configurable": {"thread_id": thread_id + "_time_travel"}} # 新线程ID，避免覆盖原状态 ): print(f"时间旅行-当前节点：{state['__step__']['node']}") # 运行时间旅行示例 # asyncio.run(time_travel_demo(thread_id="research_task_1"))5.3 子图（Subgraphs）当状态机的节点过多、逻辑复杂时，可将部分节点拆分为“子图”（Subgraph），实现模块化管理，提升代码可读性和可维护性。LangGraph 1.0+ 支持将子图作为一个独立的节点，嵌入到主图中。from langgraph import StateGraph # 1. 定义子图（搜索+总结子图） subgraph = StateGraph(ResearchState) # 子图节点：生成关键词 → 执行搜索 → 总结 subgraph.add_node("sub_generate_queries", generate_search_queries) subgraph.add_node("sub_execute_search", execute_search) subgraph.add_node("sub_summarize", summarize_results) # 子图流转逻辑 subgraph.add_edge("sub_generate_queries", "sub_execute_search") subgraph.add_edge("sub_execute_search", "sub_summarize") # 子图起始和终止节点 subgraph.set_entry_point("sub_generate_queries") subgraph.set_finish_point("sub_summarize") # 编译子图 subgraph_compiled = subgraph.compile() # 2. 将子图作为节点，嵌入主图 main_graph = StateGraph(ResearchState) # 主图节点：子图 → 反思 → 生成报告 main_graph.add_node("search_summarize_subgraph", subgraph_compiled) # 子图作为节点 main_graph.add_node("reflect", reflect_on_summary) main_graph.add_node("generate_report", generate_report) # 主图流转逻辑 main_graph.add_edge("search_summarize_subgraph", "reflect") main_graph.add_edge("reflect", "generate_report") # 主图起始和终止节点 main_graph.set_entry_point("search_summarize_subgraph") main_graph.set_finish_point("generate_report") # 编译主图 main_graph_compiled = main_graph.compile(checkpointer=checkpointer)核心优势：将复杂逻辑拆分为子图，便于单独调试、复用，适合大型多代理系统（如包含10+节点的状态机）。5.4 并行执行（Parallel Execution）LangGraph 1.0+ 支持并行执行多个节点（如同时调用多个工具、同时生成多个总结），通过parallel()方法实现，大幅提升任务执行效率。from langgraph.graph import parallel # 定义两个并行节点（示例：同时生成两个不同角度的总结） async def summarize_technical(state: ResearchState) -> ResearchState: """技术角度总结节点""" prompt = PromptTemplate( input_variables=["topic", "search_results"], template="请从技术实现角度，总结'{topic}'的核心用法，重点突出代码示例和API细节：\n{search_results}", ) response = await (prompt | chat_model).ainvoke({"topic": state.topic, "search_results": "\n".join(state.search_results)}) state.summary = f"【技术角度总结】\n{response.content}\n" + (state.summary or "") return state async def summarize_practical(state: ResearchState) -> ResearchState: """实战角度总结节点""" prompt = PromptTemplate( input_variables=["topic", "search_results"], template="请从实战应用角度，总结'{topic}'的使用场景和最佳实践：\n{search_results}", ) response = await (prompt | chat_model).ainvoke({"topic": state.topic, "search_results": "\n".join(state.search_results)}) state.summary += f"【实战角度总结】\n{response.content}" return state # 构建状态图，添加并行节点 graph = StateGraph(ResearchState) graph.add_node("execute_search", execute_search) # 并行执行两个总结节点 graph.add_node("parallel_summarize", parallel([summarize_technical, summarize_practical])) graph.add_node("generate_report", generate_report) # 流转逻辑：执行搜索 → 并行总结 → 生成报告 graph.add_edge("execute_search", "parallel_summarize") graph.add_edge("parallel_summarize", "generate_report") # 编译状态机 graph_compiled = graph.compile(checkpointer=checkpointer)关键说明：并行节点接收当前状态，同时执行多个节点函数，执行完成后将所有节点返回的状态合并（若多个节点修改同一字段，以最后执行完成的节点结果为准）。并行执行仅适用于“无依赖关系”的节点，如多个独立的总结、多个不相关的工具调用，可最大程度利用资源，缩短任务耗时。第六章：常见问题与排错指南（LangChain/LangGraph 1.0+）本节汇总开发者在学习和使用LangChain 1.0+、LangGraph 1.0+ 过程中最常见的问题、报错原因及解决方案，覆盖环境配置、API调用、状态机执行、类型校验等核心场景，帮助开发者快速排错，提升开发效率。6.1 环境配置类问题问题1：安装后导入模块报错（如“No module named 'langchain_core'”）报错原因：1. 未安装对应模块或版本不兼容；2. 虚拟环境未激活，模块安装到了全局环境；3. 安装过程中出现中断，模块未安装完整。解决方案：确认已激活虚拟环境（如conda activate langchain-env），避免全局环境冲突。卸载已安装的相关包，重新执行安装命令（指定稳定版本，参考2.1节）：`pip uninstall langchain-core langchain-openai langgraph pydantic pip install langchain-core==1.0.10 langchain-openai==0.1.0 langgraph==1.0.5 pydantic==2.6.1`若仍报错，检查Python版本（要求Python 3.8+，推荐3.10+），过低版本不支持LangChain 1.0+ 的部分特性。问题2：安装LangGraph后，调用graph.compile()报错“Checkpointer must be a Checkpointer instance”报错原因：LangGraph 1.0+ 对Checkpointer的类型要求严格，传入的Checkpointer不是langgraph.checkpoint模块下的实例，或版本不兼容。解决方案：确认导入的Checkpointer正确，需从langgraph.checkpoint相关模块导入（如MemorySaver、FileSystemCheckpointer），而非自定义类。检查LangGraph版本，确保为1.0+，旧版本的Checkpointer不兼容1.0+ 的compile()方法，执行pip install --upgrade langgraph==1.0.5更新。若无需持久化，可省略Checkpointer参数（默认不使用），即graph.compile()，但生产环境不推荐。6.2 API调用类问题问题1：调用ChatOpenAI.invoke()报错“API key not provided”报错原因：未配置OpenAI API密钥，或密钥配置错误、过期；未正确加载环境变量。解决方案：获取有效的OpenAI API密钥（登录OpenAI官网创建），确保密钥未过期、未被封禁。两种配置方式（推荐第二种，更安全）： 直接传入api_key参数（测试用）：chat_model = ChatOpenAI(model="gpt-4o", api_key="your-api-key")通过环境变量加载（生产用）：`import osfrom langchain_chat_models import ChatOpenAIchat_model = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))` （配置环境变量：Windows用set OPENAI_API_KEY=your-api-key，Linux/Mac用export OPENAI_API_KEY=your-api-key）问题2：异步调用ainvoke()报错“cannot be called from a running event loop”报错原因：在异步环境中重复调用异步函数，或未通过asyncio.run()启动异步事件循环；Jupyter Notebook等环境自带事件循环，易出现此问题。解决方案：普通Python脚本：通过asyncio.run()包裹异步函数，如asyncio.run(run_research_agent(topic="xxx"))。Jupyter Notebook环境：在异步函数前添加%autoawait魔法命令，或使用async with语句，避免重复启动事件循环：`%autoawait await run_research_agent(topic="LangGraph 1.0 多代理编排实践")`避免在异步函数内部直接调用asyncio.run()，需通过await调用其他异步函数。6.3 状态机与节点类问题问题1：状态机执行报错“State must be a subclass of BaseModel”报错原因：LangGraph 1.0+ 要求State必须是Pydantic v2 BaseModel的子类，使用了旧版的TypedDict或普通类定义状态。解决方案：将状态类修改为继承自pydantic.BaseModel，参考3.2节的ResearchState定义，确保所有字段都有明确的类型注解和Field描述。问题2：节点函数返回报错“Node must return an instance of the state type”报错原因：节点函数未返回状态模型实例，或返回的实例类型与StateGraph指定的状态类型不一致；节点函数中修改了状态但未返回。解决方案：确保所有节点函数（同步/异步）的返回值是StateGraph初始化时指定的状态模型实例（如ResearchState）。节点函数中修改状态后，必须return state，避免忘记返回（如def node(state: ResearchState) -> ResearchState: state.is_complete = True; return state）。检查节点函数的返回类型注解，确保与状态类型一致，避免返回None或其他类型。问题3：条件边路由函数报错“Path function must return a destination node name”报错原因：条件路由函数（add_conditional_edges的path参数）返回值不是字符串，或返回的节点名称不在dests列表中；返回值为空。解决方案：确保路由函数的返回值是字符串，且该字符串是add_conditional_edges中dests列表中的节点名称（如return "generate_queries"）。避免路由函数返回None、数字等非字符串类型，可添加默认返回值（如else: return "end"）。检查dests列表中的节点名称是否与状态图中添加的节点名称一致（大小写敏感）。6.4 类型校验类问题问题1：Pydantic模型初始化报错“Field required [type=missing, input_value={}, input_type=dict]”报错原因：初始化Pydantic状态模型时，未传入必填字段（Field未指定default或default_factory）；传入的字段类型与注解类型不一致。解决方案：检查状态模型的字段定义，必填字段（无默认值）在初始化时必须传入，如ResearchState(topic="xxx")（topic是必填字段）。确保传入的字段类型与注解一致，如search_queries是List[str]类型，不可传入字符串或数字。若字段可选，可通过default=None或default_factory=list（列表类型）设置默认值，避免必填报错。问题2：LLM输出解析报错“Could not parse output into Pydantic model”报错原因：LLM返回的内容不符合Pydantic输出解析器的格式要求；Prompt中未注入解析器的格式说明；LLM输出存在冗余内容（如多余的解释、换行）。解决方案：在Prompt中必须注入解析器的格式说明，即partial_variables={"format_instructions": parser.get_format_instructions()}，参考2.4节。优化Prompt，明确要求LLM严格按照格式说明输出，避免多余内容（如“仅输出符合格式的JSON，不要添加任何解释”）。若LLM输出存在轻微格式错误，可使用PydanticOutputParser的partial_parse=True参数，允许部分解析（需LangChain 1.0.8+ 版本）。第七章：学习资源与进阶方向掌握LangChain 1.0+ 和 LangGraph 1.0+ 的基础用法后，可通过以下资源进一步提升，聚焦实战落地和高级场景，成为多代理系统开发专家。7.1 官方资源（最权威，优先参考）LangChain 1.0 官方文档：https://python.langchain.com/docs/get_started/introduction，包含核心概念、API详解、实战示例，实时更新。LangGraph 1.0 官方文档：https://langchain.com/docs/langgraph，重点关注StateGraph、Checkpointer、子图等核心特性的详细说明。LangChain GitHub 仓库：https://github.com/langchain-ai/langchain，查看源码、提交记录，了解底层实现逻辑。LangGraph GitHub 仓库：https://github.com/langchain-ai/langgraph，包含官方示例、测试用例，可直接复用代码。7.2 实战进阶资源LangChain 1.0 实战项目：官方示例仓库中的“agent”“graphs”目录，包含多代理、状态机的完整实战代码，可直接运行和修改。SerpAPI 文档：https://serpapi.com/docs，学习如何将真实搜索接口集成到工具中，替代模拟搜索逻辑。Pydantic v2 官方文档：https://docs.pydantic.dev/latest/，深入学习类型校验、模型序列化、嵌套模型等高级用法，提升状态管理能力。7.3 进阶学习方向多代理系统优化：学习节点调度策略、并行执行优化、资源限制（如LLM调用频率限制），提升系统稳定性和效率。复杂状态管理：结合Redis、PostgreSQL等数据库，实现分布式Checkpointer，支持多实例部署和大规模任务调度。LangGraph 与其他框架集成：如与FastAPI集成，构建多代理API服务；与LangSmith集成，实现任务监控、日志分析和调试。本地化部署：将LLM（如Llama 3、Qwen）本地化部署，结合LangChain/LangGraph，构建离线可用的多代理系统。总结本指南从LangChain 1.0+ 和 LangGraph 1.0+ 的核心变更出发，逐步讲解了LCEL核心范式、状态机机制、实战项目、高级特性及排错指南，覆盖了从入门到进阶的全流程知识点。LangChain 1.0+ 的核心是“类型安全、异步高效、简洁可扩展”，而LangGraph 1.0+ 则通过StateGraph实现了复杂多代理逻辑的灵活编排，两者结合可快速构建生产级的AI代理系统。学习过程中，建议重点掌握LCEL组件串联、Pydantic状态定义、节点与边的配置，多动手运行示例代码，结合实战项目巩固知识点。遇到问题时，优先参考官方文档和本指南的排错部分，逐步提升问题解决能力。随着AI技术的发展，LangChain/LangGraph生态会持续更新，建议保持关注官方动态，不断学习新特性，将技术落地到实际项目中。（注：文档部分内容可能由 AI 生成）
vLLM 大模型开发全栈学习指南本指南面向具备 Python 基础和大模型核心概念（Transformer、Attention、Quantization）的开发人员，旨在帮助大家从零开始掌握 vLLM 框架，实现高性能推理服务的搭建、二次开发与深度调优，全程注重实操性与实用性，避免冗余的学术化表述，助力快速落地生产级 LLM 服务。1. 核心原理解析 (Deep Dive)vLLM 之所以能成为当前最主流的高性能 LLM 推理框架之一，核心在于其创新的架构设计，解决了传统推理中吞吐量低、内存利用率低的关键痛点。以下从开发者视角，通俗解析其核心技术原理。1.1 PagedAttention 工作原理及优势在传统 Attention 机制中，模型推理时会为每个请求分配一块连续的 GPU 内存，用于存储该请求的 Key（K）和 Value（V）张量（即 KV Cache）。这种方式存在两个致命问题：一是内存碎片化，当请求长度不一、生命周期不同时，会产生大量空闲的内存碎片，无法被有效利用；二是内存浪费，即使请求只需要少量内存，也需分配连续块，导致 GPU 内存利用率极低。vLLM 提出的 PagedAttention（分页注意力）机制，借鉴了操作系统中的“内存分页”思想，彻底解决了上述问题，其核心逻辑如下：将 KV Cache 分割成固定大小的“页”（Page），每个页的大小统一（如 16 个 Token），页是内存分配的最小单元；为每个请求分配若干个页，用于存储其 KV Cache，这些页无需连续，可分散在 GPU 内存的不同位置；通过“页表”（Page Table）记录每个请求的 KV Cache 对应的页位置，推理时通过页表快速索引、拼接所需的 KV 数据，不影响计算逻辑。相比传统 Attention，PagedAttention 的核心优势的是：内存利用率提升 3-5 倍：彻底消除内存碎片化，空闲页可被新请求复用，避免“大材小用”；支持动态请求长度：无需提前预估请求长度，可灵活处理长短不一的推理任务（如对话、长文本生成）；降低 OOM 概率：通过精细化内存分配，最大化利用 GPU 显存，减少因连续内存不足导致的模型加载失败。1.2 Continuous Batching (迭代级调度) 提升吞吐量的原理传统 LLM 推理采用“静态批处理”（Static Batching）：将多个请求打包成一个批次，一次性送入模型计算，直到所有请求都完成生成（即生成到最大长度或触发停止条件），才能处理下一批请求。这种方式的弊端很明显：不同请求的生成时间差异大（如有的请求生成 10 个 Token，有的生成 100 个 Token），批次中“快请求”会等待“慢请求”完成，导致 GPU 空闲时间长，吞吐量极低。Continuous Batching（迭代级调度，也叫动态批处理） 打破了这种“批次绑定”的限制，核心逻辑是：以 Token 生成为迭代单位，而非以请求批次为单位，具体流程如下：初始化一个请求队列，将待处理的请求放入队列；每次迭代（生成一个 Token）时，从队列中选取尽可能多的请求（不超过 GPU 内存限制），组成一个临时批次，送入模型计算，生成每个请求的下一个 Token；计算完成后，检查每个请求是否达到停止条件（如生成停止符、达到最大长度）：若达到，则将其移出队列，释放其占用的内存；若未达到，则保留在队列中，等待下一次迭代；重复步骤 2-3，直到队列中所有请求都处理完成。这种调度方式的核心优势是：GPU 始终在处理“可生成 Token 的请求”，没有空闲等待时间，尤其在多请求、长短不一的场景下，吞吐量可提升 10-100 倍（具体取决于请求分布），是 vLLM 高性能的核心保障之一。1.3 vLLM 的内存管理模型vLLM 的内存管理模型围绕“高效利用 GPU 显存”设计，分为三个核心部分，协同工作实现内存的精细化管控：KV Cache 内存管理：基于 PagedAttention，将 KV Cache 分页管理，通过页表索引，实现内存的高效分配与复用，这是内存管理的核心；模型权重内存管理：支持模型权重的分片存储（如张量并行 Tensor Parallelism），将模型权重分散到多个 GPU 上，突破单 GPU 显存限制；同时支持量化（如 FP8、AWQ），将权重从 FP16 压缩到更低精度，减少显存占用；动态内存池：vLLM 会预先分配一块固定大小的 GPU 内存作为“内存池”，用于存储 KV Cache 页和临时计算数据，避免频繁的内存申请/释放（GPU 内存申请释放开销大），进一步提升性能；当内存池不足时，会自动淘汰长期未使用的 KV 页（基于 LRU 策略），保障新请求的正常处理。简单总结：vLLM 的内存管理，本质是“分页复用 + 动态调度 + 量化压缩”，最大化利用每一寸 GPU 显存，同时降低内存操作开销。2. 环境搭建与快速启动本节提供两种主流的 vLLM 安装方案（Docker + Pip），区分 NVIDIA GPU（主流场景）和 AMD/TPU 硬件，同时提供可直接运行的“Hello World”和 API Server 示例，帮助快速上手。2.1 安装方案（区分硬件）vLLM 对 NVIDIA GPU 的支持最完善（依赖 CUDA），AMD/TPU 需依赖额外插件，以下分别说明。2.1.1 NVIDIA GPU 安装（推荐）前提条件：GPU 需支持 CUDA 11.8 及以上版本，建议 CUDA 12.1+（性能更优），已安装 NVIDIA 驱动。方案 1：Pip 安装（简单快捷，适合开发调试）# 安装基础版 vLLM（支持主流模型，如 Llama 3、Qwen） pip install vllm # 若需支持量化格式（AWQ、GPTQ），安装带扩展的版本 pip install vllm[awq,gptq] # 验证安装（查看 vLLM 版本，确认 CUDA 可用） python -c "from vllm import LLM; print('vLLM 版本:', LLM.__version__)"方案 2：Docker 安装（推荐生产环境，避免环境冲突）# 拉取官方镜像（已预装 CUDA、vLLM 及依赖） docker pull vllm/vllm-openai:latest # 启动容器（映射端口，挂载本地模型目录，分配 GPU） # 说明：--gpus all 表示使用所有 GPU，-v 映射本地模型路径到容器内 docker run -it --gpus all -p 8000:8000 -v /path/to/your/models:/models vllm/vllm-openai:latest2.1.2 AMD/TPU 安装（实验性支持）vLLM 对 AMD GPU（依赖 ROCm）和 TPU 的支持处于实验阶段，功能不如 NVIDIA GPU 完善，适合尝鲜。# AMD GPU（依赖 ROCm 5.6+） pip install vllm[rocm] # TPU（依赖 JAX 框架） pip install vllm[tpu] # 注意：AMD/TPU 目前不支持部分量化格式和高级特性，建议优先使用 NVIDIA GPU 用于生产2.2 Hello World 示例（Python API 加载模型）以加载开源模型 Llama 3-8B-Instruct 为例（也可替换为 Qwen-7B、ChatGLM4-7B 等），实现简单的文本生成，代码可直接运行。from vllm import LLM, SamplingParams # 1. 配置采样参数（控制生成效果，新手可直接使用默认值） sampling_params = SamplingParams( temperature=0.7, # 温度，越低生成越确定，越高越随机（0-1） top_p=0.9, # 核采样，只从概率前 90% 的 Token 中选择，避免生成无意义内容 max_tokens=100, # 最大生成 Token 数，防止生成过长 stop=[""] # 停止符，当生成到该 Token 时停止 ) # 2. 初始化 LLM 实例（加载模型） # 说明：model 可填 Hugging Face 模型名（自动下载）或本地模型路径 # tensor_parallel_size：张量并行数，若有多个 GPU，可设置为 GPU 数量（如 2） llm = LLM( model="meta-llama/Llama-3-8B-Instruct", tensor_parallel_size=1, # 单 GPU 设为 1 gpu_memory_utilization=0.8 # 利用 80% 的 GPU 显存，留有余地 ) # 3. 输入提示词（遵循模型的 Prompt 格式，Llama 3 需用 <|begin_of_text|> 和 <|user|> 标签） prompts = [ "<|begin_of_text|><|user|>请简要介绍 vLLM 框架<|assistant|>" ] # 4. 生成文本（批量处理，可传入多个 prompts） outputs = llm.generate(prompts, sampling_params) # 5. 打印结果 for output in outputs: prompt = output.prompt generated_text = output.outputs[0].text print(f"输入：{prompt}") print(f"输出：{generated_text}\n")运行说明：首次运行会自动下载 Llama 3-8B-Instruct 模型（约 16GB，FP16 精度），若显存不足，可添加 quantization="fp8" 参数（需 CUDA 12.0+），将模型量化为 FP8 精度，显存占用可降至 8GB 左右。2.3 API Server 示例（兼容 OpenAI 格式）vLLM 支持启动兼容 OpenAI API 格式的服务器，可直接使用 OpenAI 的 SDK 或 curl 调用，无需修改代码，适合集成到现有系统中。步骤 1：启动 API Server# 启动服务器，加载 Llama 3-8B-Instruct 模型 # --model：模型名/路径 # --port：端口号 # --gpu-memory-utilization：显存利用率 vllm serve meta-llama/Llama-3-8B-Instruct --port 8000 --gpu-memory-utilization 0.8 # 若需支持量化，添加 --quantization 参数 # vllm serve meta-llama/Llama-3-8B-Instruct --port 8000 --quantization fp8步骤 2：调用 API（curl 方式）# 发送生成请求，格式与 OpenAI API 完全一致 curl http://localhost:8000/v1/completions \ -H "Content-Type: application/json" \ -d '{ "model": "meta-llama/Llama-3-8B-Instruct", "prompt": "<|begin_of_text|><|user|>解释什么是 PagedAttention<|assistant|>", "temperature": 0.7, "max_tokens": 150, "top_p": 0.9 }'步骤 3：调用 API（Python 方式，使用 OpenAI SDK）from openai import OpenAI # 连接本地 vLLM 服务器（替换 OpenAI 官方接口） client = OpenAI( base_url="http://localhost:8000/v1", # 本地 vLLM 服务器地址 api_key="dummy-key" # 占位符，vLLM 暂不强制验证 API Key ) # 发送生成请求 response = client.completions.create( model="meta-llama/Llama-3-8B-Instruct", prompt="<|begin_of_text|><|user|>如何快速上手 vLLM<|assistant|>", temperature=0.7, max_tokens=120, top_p=0.9 ) # 打印结果 print("生成结果：", response.choices[0].text.strip())说明：vLLM 的 API Server 还支持 Chat Completions（对话接口），用法与 OpenAI Chat API 一致，只需将请求地址改为 /v1/chat/completions，并调整 prompt 格式为对话格式即可。3. 高级配置与性能调优 (关键部分)vLLM 的性能表现高度依赖配置参数，本节详细解析核心启动参数、不同场景的最佳实践，以及量化格式的使用方法，帮助大家根据自身需求（高吞吐/低延迟）优化服务性能。3.1 关键启动参数详解无论是通过 Python API 初始化 LLM，还是通过命令行启动 API Server，以下参数都是最核心、最常用的，直接影响性能和稳定性。3.1.1 --gpu-memory-utilization（显存利用率）作用：设置 vLLM 可使用的 GPU 显存比例（0-1 之间），例如 0.8 表示使用 80% 的 GPU 显存。原理：vLLM 会根据该比例预先分配内存池，用于存储 KV Cache 和模型权重，预留部分显存用于临时计算，避免 OOM。最佳实践：单模型单 GPU 场景：设置为 0.8-0.9，预留 10%-20% 显存，防止突发请求导致 OOM；多模型或张量并行场景：设置为 0.7-0.8，预留更多显存用于模型分片和通信；显存紧张时（如 16GB GPU 加载 13B 模型）：设置为 0.95，最大化利用显存，同时启用量化。3.1.2 --max-model-len（最大模型长度）作用：设置模型支持的最大 Token 长度（输入 + 输出），例如 4096 表示输入 + 输出的总 Token 数不超过 4096。注意事项：不能超过模型本身的最大长度（如 Llama 3-8B 最大长度为 8192）；设置过大：会占用更多 KV Cache 内存（即使实际请求很短），导致显存浪费；设置过小：会导致长请求无法处理，报错“exceeds max model length”。最佳实践：根据实际业务场景设置，例如对话场景设置为 2048-4096，长文本生成场景设置为 4096-8192。3.1.3 --tensor-parallel-size（张量并行数）作用：将模型权重分片到多个 GPU 上，实现多 GPU 并行推理，突破单 GPU 显存限制。原理：将 Transformer 层的权重（如 Attention 矩阵、全连接层）分割成多个部分，每个 GPU 负责一部分计算，通过通信同步结果，提升显存容量和计算速度。最佳实践：张量并行数必须等于 GPU 数量（如 2 个 GPU，设置为 2）；小模型（≤13B）：单 GPU 即可，无需开启张量并行（开启后会有通信开销，反而降低性能）；大模型（≥34B）：必须开启张量并行，例如 34B 模型用 4 个 GPU，设置为 4。3.1.4 --quantization（量化格式）作用：将模型权重从 FP16（默认）压缩到更低精度，减少显存占用，提升推理速度（部分量化格式）。支持的格式：AWQ、GPTQ、FP8、INT8（实验性），具体说明见 3.3 节。注意事项：量化会有轻微的精度损失（一般不影响实际使用），不同量化格式对硬件有要求（如 FP8 需 CUDA 12.0+）。3.2 不同场景的参数配置最佳实践vLLM 的核心优势是兼顾高吞吐和低延迟，不同场景（离线批处理 vs 在线服务）的参数配置差异较大，以下表格汇总了最佳实践（基于 NVIDIA A100 GPU，Llama 3-8B 模型）。场景类型核心目标关键参数配置补充说明高吞吐离线批处理（如文本生成、数据标注）最大化单位时间处理请求数，容忍一定延迟1. gpu_memory_utilization=0.92. max_model_len=40963. quantization=fp8（可选）4. 批量传入 prompts（一次传入 32-64 个）5. tensor_parallel_size=GPU 数量适合离线任务，可通过增大批量大小进一步提升吞吐量；若显存充足，可关闭量化以保证精度低延迟在线服务（如对话机器人、实时问答）最小化单次请求响应时间（TTFT/TPOT），保证用户体验1. gpu_memory_utilization=0.852. max_model_len=2048（根据对话长度调整）3. quantization=fp8（优先保证速度）4. 启用流式输出（stream=True）5. tensor_parallel_size=1（单 GPU 减少通信延迟）TTFT（Time To First Token）：首 Token 生成时间；TPOT（Time Per Output Token）：后续每个 Token 生成时间，需重点监控补充：TTFT 是在线服务的核心指标（用户感知的“响应速度”），可通过减少 max_model_len、启用量化、优化 GPU 算力等方式降低 TTFT。3.3 常用量化格式支持及使用方法量化是平衡显存占用、推理速度和精度的关键手段，vLLM 支持多种主流量化格式，以下介绍最常用的 3 种，包括使用方法和适用场景。3.3.1 FP8 量化（推荐，平衡速度与精度）特点：将模型权重从 FP16（16 位）压缩到 FP8（8 位），显存占用减少 50%（如 8B 模型从 16GB 降至 8GB），推理速度提升 20%-30%，精度损失极小（几乎不影响生成质量）。硬件要求：NVIDIA GPU 支持 FP8 张量核心（如 A100、H100、RTX 4090 等），CUDA 12.0+。使用方法： `# 命令行启动 API Server 时启用serve meta-llama/Llama-3-8B-Instruct --quantization fp8 --gpu-memory-utilization 0.9Python API 中启用llm = LLM(model="meta-llama/Llama-3-8B-Instruct", quantization="fp8", gpu_memory_utilization=0.9)`3.3.2 AWQ 量化（显存占用最低，适合小显存 GPU）特点：基于 Activation-aware Weight Quantization 算法，将权重量化到 4 位（INT4），显存占用减少 75%（如 8B 模型从 16GB 降至 4GB），推理速度较快，精度损失略高于 FP8，但满足大部分场景需求。硬件要求：NVIDIA GPU（CUDA 11.8+），无需 FP8 张量核心（适合 RTX 3090、A10 等中端 GPU）。使用方法： `# 先安装 AWQ 依赖（已安装 vllm[awq] 可跳过）install vllm[awq]启动时指定量化格式为 awq，需使用 AWQ 量化后的模型注：不能直接使用原始 FP16 模型，需先下载量化好的模型（如 TheBloke/Llama-3-8B-Instruct-AWQ）vllm serve TheBloke/Llama-3-8B-Instruct-AWQ --quantization awqPython API 使用llm = LLM(model="TheBloke/Llama-3-8B-Instruct-AWQ", quantization="awq")`3.3.3 GPTQ 量化（兼容广泛，社区资源丰富）特点：最主流的 4 位量化格式之一，社区资源丰富（大量预量化模型），显存占用与 AWQ 相当，推理速度略低于 AWQ 和 FP8，精度表现良好。硬件要求：NVIDIA GPU（CUDA 11.8+），支持大部分中端及以上 GPU。使用方法： `# 安装 GPTQ 依赖install vllm[gptq]使用预量化的 GPTQ 模型（如 TheBloke/Llama-3-8B-Instruct-GPTQ-4bit）vllm serve TheBloke/Llama-3-8B-Instruct-GPTQ-4bit --quantization gptqPython API 使用llm = LLM(model="TheBloke/Llama-3-8B-Instruct-GPTQ-4bit", quantization="gptq")`总结：优先选择 FP8（性能最佳）；显存紧张（如 16GB GPU 加载 13B 模型）选择 AWQ；若需要兼容更多模型，选择 GPTQ。4. 进阶开发与集成本节介绍 vLLM 的进阶用法，包括自定义采样参数、集成 LoRA 适配器实现多租户/多模型切换，以及自定义 CUDA Kernel 的简要思路，帮助大家实现二次开发，适配复杂业务场景。4.1 自定义 Sampling 参数Sampling 参数决定了文本生成的随机性、多样性和准确性，vLLM 支持多种采样参数，可根据业务需求（如精准回答、创意生成）自定义配置，以下是常用参数及用法。4.1.1 核心 Sampling 参数详解from vllm import SamplingParams # 自定义采样参数，适用于 Python API 生成 custom_sampling = SamplingParams( temperature=0.3, # 低温度（0.1-0.5）：生成内容更确定、更精准，适合问答、摘要场景 # temperature=0.9：高温度（0.7-1.0）：生成内容更多样、更有创意，适合文案、故事生成 top_p=0.85, # 核采样，过滤低概率 Token，0.8-0.95 较为合适，避免生成无意义内容 top_k=50, # _top_k 采样，只从概率前 50 的 Token 中选择，与 top_p 二选一即可 max_tokens=200, # 最大生成 Token 数 stop=["", "###"], # 多停止符，当生成到其中任意一个时停止 presence_penalty=0.1, # 存在惩罚，减少重复出现的 Token（0-2），避免生成冗余内容 frequency_penalty=0.1, # 频率惩罚，减少高频 Token 的出现，进一步降低冗余 logit_bias={32000: 2.0}, # 日志偏差，提高指定 Token 的生成概率（32000 是 Llama 3 的 <|user|> Token ID） skip_special_tokens=True # 跳过特殊 Token（如 ），使生成结果更简洁 ) # 使用自定义采样参数生成文本 llm = LLM(model="meta-llama/Llama-3-8B-Instruct") prompts = ["<|begin_of_text|><|user|>写一篇关于 vLLM 性能优化的短文<|assistant|>"] outputs = llm.generate(prompts, custom_sampling) print(outputs[0].outputs[0].text)4.1.2 采样参数最佳实践精准场景（问答、代码生成）：temperature=0.1-0.3，top_p=0.8-0.9，开启 presence_penalty=0.1-0.2；创意场景（文案、故事）：temperature=0.7-0.9，top_p=0.9-0.95，关闭惩罚参数或设为 0；避免重复：开启 presence_penalty 和 frequency_penalty，取值 0.1-0.3，不宜过高（否则会导致生成不连贯）。4.2 集成 LoRA 适配器实现多租户/多模型切换在多租户场景中，不同用户可能需要不同的模型微调效果（如不同行业的问答、不同风格的生成），若为每个租户部署独立模型，会占用大量显存。vLLM 支持集成 LoRA（Low-Rank Adaptation）适配器，可在一个基础模型上加载多个 LoRA 权重，实现多租户/多模型切换，大幅节省显存。4.2.1 前提条件已训练好 LoRA 适配器（基于基础模型，如 Llama 3-8B），LoRA 权重格式为 Hugging Face 格式；安装 peft 库（用于加载 LoRA 权重）：pip install peft。4.2.2 集成 LoRA 的代码示例from vllm import LLM, SamplingParams from peft import PeftModel from transformers import AutoModelForCausalLM, AutoTokenizer # 1. 加载基础模型和 Tokenizer（与 LoRA 训练时的基础模型一致） base_model_name = "meta-llama/Llama-3-8B-Instruct" tokenizer = AutoTokenizer.from_pretrained(base_model_name) base_model = AutoModelForCausalLM.from_pretrained( base_model_name, torch_dtype="auto", device_map="auto" ) # 2. 加载多个 LoRA 适配器（示例：两个不同租户的 LoRA 权重） lora_model_1 = PeftModel.from_pretrained(base_model, "path/to/tenant1-lora") # 租户 1 的 LoRA lora_model_2 = PeftModel.from_pretrained(base_model, "path/to/tenant2-lora") # 租户 2 的 LoRA # 3. 将 LoRA 模型转换为 vLLM 可加载的格式（合并 LoRA 权重到基础模型） # 注意：vLLM 目前不支持动态加载 LoRA，需合并后加载，可通过循环切换模型实现多租户 def load_lora_model(lora_model): # 合并 LoRA 权重（不修改基础模型，避免污染） merged_model = lora_model.merge_and_unload() # 初始化 vLLM LLM 实例 return LLM( model=merged_model, tokenizer=tokenizer, gpu_memory_utilization=0.85 ) # 4. 多租户切换示例 sampling_params = SamplingParams(temperature=0.7, max_tokens=150) prompts = ["请回答关于金融行业的问题：什么是私募基金？"] # 租户 1（金融行业 LoRA） llm_tenant1 = load_lora_model(lora_model_1) output1 = llm_tenant1.generate(prompts, sampling_params) print("租户 1 输出：", output1[0].outputs[0].text) # 租户 2（教育行业 LoRA） llm_tenant2 = load_lora_model(lora_model_2) output2 = llm_tenant2.generate(prompts, sampling_params) print("租户 2 输出：", output2[0].outputs[0].text)4.2.3 注意事项vLLM 目前不支持动态加载/卸载 LoRA 权重，需合并后加载，切换租户时需重新初始化 LLM 实例（开销较小）；LoRA 权重体积小（通常几 MB 到几十 MB），合并后模型体积与基础模型一致，不会额外占用大量显存；建议为每个租户预加载合并后的模型，避免实时合并带来的延迟。4.3 自定义 CUDA Kernel 或使用扩展接口vLLM 的核心计算逻辑（如 PagedAttention）是基于 CUDA Kernel 实现的，若需要进一步优化性能（如适配特定硬件、自定义 Attention 逻辑），可通过以下两种方式扩展。4.3.1 自定义 CUDA Kernel（高级）vLLM 的 CUDA Kernel 代码位于 vllm/cuda_ops/ 目录下，主要包括 Attention、线性层等核心计算模块。自定义 CUDA Kernel 的步骤如下：基于 vLLM 现有 Kernel 代码，修改或新增 CUDA 函数（如优化 Attention 计算逻辑，适配特定 GPU 架构）；编写 C++ 包装代码，将 CUDA Kernel 暴露给 Python 接口；重新编译 vLLM（修改 setup.py，添加自定义 Kernel 的编译配置）；在 Python 代码中调用自定义 Kernel，替换原有计算逻辑。注意：自定义 CUDA Kernel 需要具备扎实的 CUDA 编程基础，建议参考 vLLM 官方文档的扩展指南，避免破坏原有架构。4.3.2 使用 vLLM 的扩展接口vLLM 提供了简单的扩展接口，允许开发者自定义 Tokenizer、采样逻辑等，无需修改底层 CUDA 代码，适合大多数二次开发场景。from vllm import LLM, SamplingParams from vllm.tokenizers import Tokenizer # 示例 1：自定义 Tokenizer（适配自定义模型的 Tokenizer） class CustomTokenizer(Tokenizer): def __init__(self, tokenizer_path): super().__init__() # 加载自定义 Tokenizer（如基于 SentencePiece 训练的 Tokenizer） from transformers import AutoTokenizer self.tokenizer = AutoTokenizer.from_pretrained(tokenizer_path) def encode(self, prompt, **kwargs): # 自定义编码逻辑 return self.tokenizer.encode(prompt, **kwargs) def decode(self, tokens, **kwargs): # 自定义解码逻辑 return self.tokenizer.decode(tokens, **kwargs) # 使用自定义 Tokenizer 初始化 LLM custom_tokenizer = CustomTokenizer("path/to/custom-tokenizer") llm = LLM( model="path/to/custom-model", tokenizer=custom_tokenizer ) # 示例 2：自定义采样逻辑（继承 SamplingParams，重写采样方法） class CustomSampling(SamplingParams): def __init__(self): super().__init__(temperature=0.7, max_tokens=100) def sample(self, logits): # 自定义采样逻辑（如基于业务规则过滤 Token） # 这里简化示例，返回概率最高的 Token return logits.argmax(dim=-1)5. 常见问题与调试 (Troubleshooting)本节汇总新手最容易遇到的报错及解决方案，同时提供性能监控建议，帮助大家快速定位和解决问题，保障服务稳定运行。5.1 常见报错及解决方案报错 1：OutOfMemoryError (OOM)：CUDA out of memory原因：GPU 显存不足，可能是模型体积过大、max_model_len 设置过高、显存利用率设置过高，或同时运行其他占用显存的程序。解决方案：启用量化：添加 --quantization fp8/awq/gptq，减少显存占用；降低显存利用率：将 --gpu-memory-utilization 从 0.9 降至 0.8 或更低；减小 max_model_len：根据实际场景设置（如从 8192 降至 4096）；使用张量并行：若有多个 GPU，设置 --tensor-parallel-size 为 GPU 数量，将模型分片；关闭其他占用显存的程序（如 PyTorch 进程、其他模型服务）。报错 2：CUDA error: invalid device function / CUDA version mismatch原因：vLLM 编译时使用的 CUDA 版本与当前 GPU 驱动支持的 CUDA 版本不匹配（如 vLLM 用 CUDA 12.1 编译，而 GPU 驱动只支持 CUDA 11.8）。解决方案：1. 查看 GPU 驱动支持的最高 CUDA 版本：`nvidia-smi`（右上角显示 CUDA Version）； 2. 安装对应版本的 CUDA（如驱动支持 CUDA 12.1，就安装 CUDA 12.1）； 3. 重新安装 vLLM，确保编译时使用对应版本的 CUDA：`pip install vllm --no-cache-dir`； 4. 推荐使用 Docker 安装，避免 CUDA 版本冲突。 报错 3：Model loading failed: Could not load model ...原因：模型路径错误、模型文件损坏、模型格式不支持，或量化模型与指定的量化格式不匹配（如用 AWQ 格式的模型，却指定 --quantization gptq）。解决方案：检查模型路径：确保路径正确，若使用 Hugging Face 模型名，确认网络可访问（或手动下载后指定本地路径）；检查模型完整性：若手动下载，确认所有文件（如 pytorch_model-xxx.bin）都已下载完成，无损坏；匹配量化格式：AWQ 模型对应 --quantization awq，GPTQ 模型对应 --quantization gptq，不可混用；确认模型支持：vLLM 支持大多数 Hugging Face 格式的 Transformer 模型，若为自定义模型，需确保模型结构与 vLLM 兼容。报错 4：Request exceeds max model length原因：输入 prompt 的 Token 数 + 生成的 Token 数，超过了 --max-model-len 设置的值。解决方案：增大 --max-model-len（不超过模型本身的最大长度）；缩短输入 prompt 的长度，删除无关内容；减少 --max_tokens（生成的 Token 数），避免生成过长内容。报错 5：API Server 启动失败：Address already in use原因：指定的端口（如 8000）已被其他程序占用。解决方案：查看占用端口的程序：netstat -tuln | grep 8000（Linux/Mac）或 netstat -ano | findstr 8000（Windows）；关闭占用端口的程序，或更换端口（如 --port 8001）。5.2 性能监控建议vLLM 服务的性能监控核心是跟踪“吞吐量”“延迟”和“显存占用”，以下是常用的监控工具和关键指标。5.2.1 关键监控指标TTFT（Time To First Token）：首 Token 生成时间，在线服务的核心延迟指标，建议控制在 100ms 以内（GPU 性能足够的情况下）；TPOT（Time Per Output Token）：后续每个 Token 的生成时间，建议控制在 10ms 以内，反映推理速度；吞吐量（Throughput）：单位时间内处理的请求数（或生成的 Token 数），离线批处理场景重点关注；显存占用（GPU Memory Usage）：实时监控 GPU 显存使用情况，避免 OOM，建议长期保持在 80%-90% 以内；请求队列长度：在线服务场景，若队列长度持续增长，说明服务压力过大，需扩容 GPU 或优化参数。5.2.2 常用监控工具nvidia-smi：最基础的 GPU 监控工具，实时查看显存占用、GPU 利用率，命令：nvidia-smi -l 1（每秒刷新一次）；vLLM 内置监控：启动 API Server 时添加 --enable-metrics 参数，会暴露 Prometheus 监控指标（如 vllm_requests_total、vllm_ttft_seconds），可通过 Prometheus + Grafana 可视化；tqdm：Python API 中可结合 tqdm 监控批量生成的进度和速度，示例： `from tqdm import tqdm prompts = [f"生成第 {i} 段文本" for i in range(100)]sampling_params = SamplingParams(max_tokens=50)llm = LLM(model="meta-llama/Llama-3-8B-Instruct")批量生成并监控进度for prompt in tqdm(prompts, desc="生成进度"):outputs = llm.generate([prompt], sampling_params)` 6. 实战演练项目本节设计一个小型实战任务：构建一个支持流式输出、带有简单鉴权的高并发问答服务 Demo，整合前面所学的知识（环境搭建、API Server、高级配置、性能调优），实现可直接部署的服务。6.1 项目需求功能：支持用户发送问答请求，返回流式输出（逐 Token 生成，提升用户体验）；鉴权：简单的 API Key 鉴权，防止未授权访问；性能：支持高并发，低延迟（TTFT ≤ 100ms，TPOT ≤ 10ms）；模型：使用 Llama 3-8B-Instruct（FP8 量化，节省显存）；部署：基于 vLLM API Server，可直接启动并对外提供服务。6.2 项目实现步骤步骤 1：环境准备# 安装依赖 pip install vllm[fp8] openai fastapi uvicorn python-multipart # 说明：vllm[fp8] 用于FP8量化支持，openai用于API调用，fastapi+uvicorn用于鉴权服务搭建，python-multipart处理请求数据步骤 2：搭建简单鉴权服务（基于FastAPI）为API Server添加API Key鉴权，防止未授权访问，同时作为vLLM API Server的代理层，转发请求并验证权限。创建文件auth_server.py，代码如下：from fastapi import FastAPI, Request, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse import requests import uvicorn # 初始化FastAPI应用 app = FastAPI(title="vLLM 鉴权代理服务", version="1.0") # 允许跨域请求（前端调用需开启） app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境需替换为具体前端域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 配置合法API Key（生产环境建议从配置文件或环境变量读取） VALID_API_KEYS = {"vllm-demo-2024", "test-key-123"} # 鉴权依赖：验证请求头中的API Key def verify_api_key(request: Request): api_key = request.headers.get("X-API-Key") if not api_key or api_key not in VALID_API_KEYS: raise HTTPException(status_code=401, detail="Invalid or missing API Key") return api_key # 代理vLLM的流式对话接口（核心接口） @app.post("/v1/chat/completions") async def chat_completions(request: Request, api_key: str = Depends(verify_api_key)): # 读取请求体（保持与OpenAI API格式一致） request_body = await request.json() # 转发请求到本地vLLM API Server vllm_url = "http://localhost:8000/v1/chat/completions" try: # 流式转发响应（逐Token返回，提升用户体验） response = requests.post( vllm_url, json=request_body, stream=True # 开启流式响应 ) # 验证vLLM返回状态 if response.status_code != 200: raise HTTPException(status_code=response.status_code, detail=response.text) # 流式返回响应给前端 def stream_generator(): for chunk in response.iter_content(chunk_size=1024): if chunk: yield chunk return StreamingResponse(stream_generator(), media_type="text/event-stream") except Exception as e: raise HTTPException(status_code=500, detail=f"Service error: {str(e)}") # 启动鉴权服务 if __name__ == "__main__": uvicorn.run( app="auth_server:app", host="0.0.0.0", # 允许外部访问 port=8001, # 鉴权服务端口（与vLLM API Server区分） workers=4 # 启动4个工作进程，支持高并发 )步骤 3：启动vLLM API Server（带FP8量化）启动vLLM服务，加载Llama 3-8B-Instruct模型，启用FP8量化节省显存，配置合理的显存利用率和最大模型长度，适配问答场景：# 启动vLLM API Server，监听8000端口 vllm serve meta-llama/Llama-3-8B-Instruct \ --port 8000 \ --gpu-memory-utilization 0.85 \ --max-model-len 2048 \ --quantization fp8 \ --enable-metrics # 开启监控指标，便于后续性能监控 # 说明：--max-model-len=2048适配对话场景，--quantization=fp8将显存占用降至8GB左右，适合中端GPU步骤 4：启动鉴权服务在另一个终端启动鉴权代理服务，监听8001端口，实现API Key验证和请求转发：# 启动鉴权服务 python auth_server.py # 启动成功后，终端会显示：Uvicorn running on http://0.0.0.0:8001 (Press CTRL+C to quit)步骤 5：测试服务（流式输出+鉴权）使用curl或Python代码测试服务，验证鉴权功能和流式输出效果，以下提供两种测试方式：方式1：curl测试（流式输出）# 发送流式对话请求，携带合法API Key（X-API-Key） curl http://localhost:8001/v1/chat/completions \ -H "Content-Type: application/json" \ -H "X-API-Key: vllm-demo-2024" \ -d '{ "model": "meta-llama/Llama-3-8B-Instruct", "messages": [ {"role": "user", "content": "请详细说明vLLM中PagedAttention和Continuous Batching的协同作用"} ], "temperature": 0.7, "max_tokens": 300, "stream": true # 开启流式输出 }' # 预期结果：逐段返回生成内容，终端会持续输出响应片段，直到生成完成方式2：Python代码测试（验证鉴权+流式）from openai import OpenAI # 连接鉴权服务（而非直接连接vLLM） client = OpenAI( base_url="http://localhost:8001/v1", api_key="dummy-key" # 此处api_key无效，鉴权由X-API-Key控制 ) # 发送流式请求，携带合法API Key stream = client.chat.completions.create( model="meta-llama/Llama-3-8B-Instruct", messages=[ {"role": "user", "content": "如何优化vLLM的TTFT指标？"} ], temperature=0.7, max_tokens=200, stream=True, headers={"X-API-Key": "vllm-demo-2024"} # 关键：携带合法API Key ) # 逐Token打印流式输出 print("流式生成结果：") for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="", flush=True) # 测试无效API Key（预期报错） try: stream_invalid = client.chat.completions.create( model="meta-llama/Llama-3-8B-Instruct", messages=[{"role": "user", "content": "测试无效API Key"}], stream=True, headers={"X-API-Key": "invalid-key"} ) except Exception as e: print(f"\n\n无效API Key测试结果：{e}") # 预期输出401错误步骤 6：性能优化与监控结合前面所学的性能调优知识，对Demo服务进行优化，并监控核心指标：性能优化：显存优化：已启用FP8量化，若显存仍紧张，可改为AWQ量化（需下载AWQ量化模型）； 并发优化：鉴权服务启动4个worker，vLLM可通过--max-num-batched-tokens调整批处理大小（如设置为1024），提升吞吐量；延迟优化：确保单GPU部署（tensor_parallel_size=1），减少通信延迟，TTFT可控制在100ms以内。性能监控：监控GPU显存：执行`nvidia-smi -l 1`，查看显存占用（应稳定在80%-85%）； 监控vLLM指标：访问http://localhost:8000/metrics，查看TTFT、TPOT等指标；并发测试：使用工具（如locust）模拟多用户请求，验证服务高并发能力，确保无OOM和请求阻塞。6.3 项目部署说明（生产环境参考）本Demo可直接用于测试和小型部署，生产环境需补充以下优化：鉴权优化：将API Key存储在环境变量或配置文件（如.env），避免硬编码；添加API Key过期、权限分级功能；模型部署：使用Docker容器化部署vLLM和鉴权服务，通过Docker Compose管理多服务，避免环境冲突；高可用：多GPU部署（启用张量并行），配置负载均衡，避免单点故障；监控告警：结合Prometheus+Grafana搭建监控面板，对OOM、高延迟、请求失败等异常设置告警；日志管理：添加日志记录（如使用logging模块），记录请求日志、错误日志，便于问题排查。6.4 项目总结本实战项目整合了vLLM环境搭建、API Server启动、量化配置、鉴权服务搭建和流式输出等核心知识点，实现了一个可直接部署的高并发问答服务Demo。通过本项目，可掌握：vLLM API Server的高级配置（量化、显存利用率、最大模型长度）；如何为vLLM服务添加鉴权，保障服务安全；流式输出的实现方式，提升用户体验；vLLM服务的性能优化和监控方法。可根据实际业务需求，扩展功能（如多模型切换、请求限流、前端界面），实现生产级别的LLM推理服务。7. 总结与进阶方向7.1 核心知识点总结本指南从核心原理、环境搭建、高级配置、进阶开发、问题调试到实战演练，全面覆盖了vLLM开发的全流程，核心要点如下：核心原理：PagedAttention解决内存碎片化问题，Continuous Batching提升吞吐量，两者协同构成vLLM高性能的核心；环境搭建：优先选择NVIDIA GPU（CUDA 12.0+），推荐Docker安装避免环境冲突，Pip安装适合开发调试；性能调优：核心是“量化压缩+参数配置+场景适配”，FP8量化平衡速度与精度，AWQ适合小显存场景，参数配置需结合离线/在线场景调整；进阶开发：可通过自定义Sampling参数、集成LoRA、扩展接口等方式，适配复杂业务场景；问题调试：常见报错（OOM、CUDA版本不匹配、模型加载失败）均有明确解决方案，重点关注显存和参数配置；实战落地：结合鉴权、流式输出、性能监控，可快速搭建可部署的高并发LLM推理服务。7.2 进阶学习方向掌握本指南内容后，可从以下方向进一步深入学习，提升vLLM开发能力：底层源码解析：深入研究vLLM的PagedAttention CUDA Kernel实现、Continuous Batching调度逻辑，理解高性能的底层原理；多模型与多GPU部署：学习vLLM的多模型部署（如模型并行）、多GPU张量并行/流水线并行配置，突破单GPU性能和显存限制；自定义优化：开发自定义CUDA Kernel、优化Sampling逻辑，适配特定硬件（如H100）和业务场景（如长文本推理）；生态集成：将vLLM与LangChain、LlamaIndex等框架集成，实现复杂的LLM应用（如RAG检索增强生成）；最新特性跟踪：关注vLLM官方更新，学习新特性（如动态批处理优化、新量化格式支持），保持技术同步。7.3 参考资源vLLM官方文档：https://docs.vllm.ai/（最权威的学习资源，包含最新特性和API说明）；vLLM GitHub仓库：https://github.com/vllm-project/vllm（源码、示例和问题排查）；Hugging Face模型库：https://huggingface.co/（获取预量化模型、基础模型）；vLLM性能优化实战：官方博客和技术分享，了解工业界落地案例。本指南旨在帮助开发者快速上手vLLM并落地生产级服务，后续可结合实际业务场景，灵活运用所学知识，实现高性能、高可用的LLM推理服务。（注：文档部分内容可能由 AI 生成）
Ollama 本地大模型学习与实战指南目标读者：有一定技术基础（了解终端/命令行、基本编程概念）但从未接触过 Ollama 的开发者1. 简介1.1 什么是 Ollama？Ollama 是一款轻量级的开源工具，专为在本地环境中轻松运行、管理和定制大型语言模型（LLM）而设计。它将复杂的大模型部署流程（如模型编译、显存管理、依赖配置）封装为简单的命令行操作，让开发者无需深入理解底层技术细节，就能快速在本地运行 Llama 3、Phi、Gemma、Qwen 等主流开源大模型，实现本地推理、私有部署等需求。1.2 核心优势轻量级：核心程序体积小，支持模型量化（如 4-bit、8-bit 量化），资源占用可控，低配电脑（8GB 内存+独立显卡）、小型服务器也能流畅运行。隐私保护：所有数据（输入prompt、推理结果）和计算过程都在本地完成，无需将敏感信息（如企业内部数据、个人隐私数据）发送到云端，完全符合数据合规要求。易用性：极简的命令行交互，无需复杂的环境配置（如配置 CUDA、依赖库），开箱即用，一行命令即可拉取、运行模型。可定制化：支持通过 Modelfile 自定义模型的系统提示、生成参数、对话模板等，适配不同业务场景（如编程助手、知识库问答）。API 支持：内置 REST API，无需额外部署服务，即可轻松集成到 Python、Java 等各类应用程序中，实现二次开发。跨平台：完美支持 macOS、Linux、Windows 三大主流操作系统，部署环境灵活无限制。2. 环境安装安装前准备：确保设备满足基础要求——CPU 支持 AVX2 指令集（大多数 2017 年后的 CPU 均支持）；若需 GPU 加速，NVIDIA 显卡需支持 CUDA 11.7+，AMD 显卡需支持 Metal（macOS）或 ROCm（Linux）。2.1 macOS 安装方式 1：使用 Homebrew（推荐，适合已安装 Homebrew 的用户）# 安装 Ollama（自动处理依赖，配置环境变量） brew install ollama # 启动 Ollama 服务（设置开机自启，无需手动重启） brew services start ollama方式 2：手动安装（适合未安装 Homebrew 的用户）# 下载并运行官方安装脚本，自动完成安装和服务配置 curl -fsSL https://ollama.com/install.sh | sh2.2 Linux 安装支持 Ubuntu、CentOS、Debian 等主流发行版，脚本会自动适配系统架构（x86_64、ARM64）。# 通用安装脚本（自动安装依赖、配置服务） curl -fsSL https://ollama.com/install.sh | sh # 验证服务状态（systemd 系统，如 Ubuntu 18.04+、CentOS 7+） systemctl status ollama # 若服务未启动，手动启动并设置开机自启 systemctl start ollama systemctl enable ollama2.3 Windows 安装方式 1：官方安装包（推荐，操作最简单）访问 Ollama 官方下载页：https://ollama.com/download/windows下载 Windows 安装包（.exe 文件，约 100MB 左右）双击运行安装包，按照向导完成安装（默认安装路径为 C:Program FilesOllama）安装完成后，Ollama 会自动注册为系统服务并启动，无需手动操作。方式 2：使用 Winget（适合习惯命令行的用户）# 通过 Windows 包管理器（Winget）安装，自动完成配置 winget install Ollama.Ollama2.4 验证安装是否成功安装完成后，打开终端（macOS/Linux 终端、Windows PowerShell/命令提示符），执行以下命令验证：# 1. 检查 Ollama 版本（验证安装是否成功） ollama --version # 2. 运行内置的 llama3 模型（首次运行会自动拉取模型，约 4.7GB） ollama run llama3验证成功标志：执行 ollama --version 能输出版本号（如 ollama version 0.1.48），无报错。执行 ollama run llama3 后，终端会进入交互模式（显示 >>>），输入问题（如“什么是 Ollama？”）能得到模型的有效回答。若验证失败：检查终端是否以管理员身份运行；Windows 系统需重启终端刷新环境变量；Linux/macOS 系统可重启 Ollama 服务（brew services restart ollama 或 systemctl restart ollama）。3. 基础命令速查Ollama 的核心操作均通过命令行完成，以下是开发者最常用的基础命令，每个命令均提供可直接复制执行的代码示例，标注关键参数说明。3.1 拉取模型 (pull)用于从 Ollama 官方模型仓库（https://ollama.com/library）下载预训练模型到本地，首次运行模型时也会自动拉取，可手动指定模型版本和量化等级。# 1. 拉取指定模型（默认最新版本，如 Llama 3 8B 基础版） ollama pull llama3 # 2. 拉取特定版本/量化等级的模型（推荐，节省资源） # 格式：ollama pull 模型名:版本-量化等级 ollama pull llama3:70b-q4_0 # Llama 3 70B 模型，4-bit 量化（约 39GB） ollama pull phi3:mini-q8_0 # Phi-3 Mini 模型，8-bit 量化（约 1.1GB，适合低配设备） # 3. 拉取其他主流开源模型（常用推荐） ollama pull gemma:7b # Google Gemma 7B 模型（适合通用场景） ollama pull qwen:14b # 通义千问 14B 模型（中文支持优秀） ollama pull mistral:7b # Mistral 7B 模型（推理速度快） ollama pull llama3-chinese:8b # 中文优化版 Llama 3 8B 模型 # 4. 查看拉取进度（后台拉取时，避免终端卡死） ollama pull --progress plain llama3关键说明：模型量化等级越低（如 q4_0），体积越小、资源占用越少，但推理精度会略有下降；q8_0 兼顾精度和资源，适合大多数开发者。3.2 运行模型 (run)启动模型的交互式对话界面，直接与模型进行问答交互，也支持非交互式一次性提问（适合脚本调用）。# 1. 基础用法：运行已下载的模型（进入交互模式） ollama run llama3 # 2. 运行时临时调整生成参数（无需修改配置，仅本次生效） ollama run llama3 --temperature 0.8 --num_ctx 2048 # 3. 非交互式运行（一次性提问，直接输出结果，适合脚本调用） ollama run llama3 "请解释一下 REST API 的核心概念，用简洁的语言说明" # 4. 交互模式退出方法（3种可选） # 方法1：输入 /bye 或 /exit（推荐，正常退出） # 方法2：按 Ctrl + D（终端快捷键，强制退出） # 方法3：按 Ctrl + C（中断进程，紧急退出）关键说明：交互模式下，模型会记住上下文（如连续提问“它的特点是什么”，会关联上一轮的“REST API”），退出后上下文清空。3.3 列出/删除模型 (list/rm)列出本地所有模型查看本地已下载的所有模型，包括模型名称、ID、体积、修改时间，方便管理。# 查看本地模型列表（简洁版） ollama list # 输出示例： # NAME ID SIZE MODIFIED # llama3:latest 78e26419b446 4.7 GB 2 minutes ago # phi3:latest a89989523527 2.3 GB 1 hour ago # qwen:14b 9f7a3999298c 13.8 GB 1 day ago删除本地模型当本地存储空间不足时，可删除不常用的模型，释放资源。# 1. 删除指定模型（通过模型名删除，默认删除 latest 版本） ollama rm llama3:latest # 2. 删除指定版本的模型（精准删除，避免误删其他版本） ollama rm phi3:mini-q8_0 # 3. 批量删除多个模型（用空格分隔模型名） ollama rm gemma:7b qwen:14b注意：删除模型后无法恢复，需重新拉取，删除前确认是否需要保留。3.4 查看模型信息 (show)查看模型的详细配置，包括基础信息、系统提示、参数配置、Modelfile 等，帮助开发者了解模型的默认设置。# 1. 查看模型的基本信息（名称、ID、体积、参数等） ollama show llama3 # 2. 查看模型的 Modelfile（自定义配置文件，后续进阶用法会详细讲解） ollama show llama3 --modelfile # 3. 查看模型的默认系统提示（定义模型的角色和行为） ollama show llama3 --system # 4. 查看模型的默认生成参数（temperature、num_ctx 等） ollama show llama3 --parameters3.5 其他常用基础命令# 1. 复制模型（创建模型副本，用于自定义修改，避免修改原模型） ollama cp llama3 my-llama3:v1 # 复制 llama3 为 my-llama3:v1 # 2. 停止 Ollama 服务（macOS/Linux，需重启服务时使用） ollama stop # 3. 重启 Ollama 服务（macOS Homebrew 安装方式） brew services restart ollama # 4. 重启 Ollama 服务（Linux systemd 安装方式） systemctl restart ollama4. 进阶用法基础命令仅能满足简单的本地交互需求，进阶用法可实现模型定制、API 集成、参数优化等，适配开发者的二次开发和业务场景需求。4.1 如何创建自定义 ModelfileModelfile 是 Ollama 用于定义模型行为的配置文件，本质是纯文本文件（无后缀），通过一系列指令自定义基础模型、系统提示、生成参数、对话模板等，最终可构建属于自己的定制化模型。4.1.1 Modelfile 核心指令详解以下是 Modelfile 最常用的指令，每个指令均标注用途、格式和示例，所有指令不区分大小写，但推荐大写书写以区分配置。FROM（必选）：指定基础模型，即基于哪个预训练模型进行定制，必须是本地已下载或能通过 Ollama 拉取的模型。格式：FROM 模型名:版本示例：FROM llama3:latest（基于 Llama 3 最新版定制）、FROM qwen:14b-q4_0（基于通义千问 14B 4-bit 量化版定制）SYSTEM（可选）：设置模型的系统提示，定义模型的角色、行为准则、回答风格，相当于给模型“定规矩”，会影响所有对话的输出。格式：SYSTEM "提示内容"（多行提示用三个双引号包裹）示例：见下方完整 Modelfile 示例PARAMETER（可选）：设置模型的生成参数，覆盖基础模型的默认参数，优化推理效果，常用参数后续会详细讲解。格式：PARAMETER 参数名 参数值示例：PARAMETER temperature 0.7、PARAMETER num_ctx 4096PROMPT（可选）：设置默认的用户提示，即模型启动后自动加载的初始提问，适合固定场景（如固定任务指令）。格式：PROMPT "初始提问内容"示例：PROMPT "请作为一名 Python 编程助手，帮我调试以下代码中的错误"TEMPLATE（可选）：自定义对话模板，控制输入输出的格式（如添加用户/助手标识、调整换行），适合集成到应用时统一格式。格式：TEMPLATE "{{ .System }}\n用户：{{ .Prompt }}\n助手："示例：自定义对话格式为“用户：XXXn助手：XXX”4.1.2 完整 Modelfile 示例（Python 编程助手）创建一个基于 Llama 3 的定制化模型，专门用于 Python 编程指导，步骤如下：# 1. 指定基础模型（本地已下载 llama3:latest） FROM llama3:latest # 2. 设置系统提示（定义 Python 编程助手的角色和规则） SYSTEM """ 你是一名专业的 Python 编程导师，擅长用简洁易懂的语言解释复杂的编程概念，帮助开发者解决 Python 开发中的问题。 回答必须严格遵循以下规则： 1. 所有代码示例必须包含详细注释，注释清晰易懂，说明关键步骤的作用； 2. 优先提供最佳实践方案，避免过时语法（兼容 Python 3.8+）； 3. 遇到错误排查类问题，先分析错误原因，再给出解决方案，最后提供完整可运行的代码； 4. 避免使用过于专业的术语，必要时给出术语解释； 5. 回答简洁明了，不冗余，重点突出核心解决方案。 """ # 3. 设置生成参数（优化编程指导的输出效果） PARAMETER temperature 0.6 # 降低随机性，让回答更严谨 PARAMETER top_p 0.9 # 控制生成内容的多样性，避免偏离主题 PARAMETER num_ctx 4096 # 增大上下文窗口，支持更长的代码和提问 PARAMETER max_tokens 2048 # 限制单次回答的最大长度，避免输出过长 PARAMETER stop "用户：" # 当出现“用户：”时停止生成，避免模型模拟用户提问 # 4. 设置默认提示（启动模型后自动加载的初始提问） PROMPT "请告诉我你能提供哪些 Python 编程帮助，用 bullet points 列出。"4.1.3 基于 Modelfile 构建自定义模型创建好 Modelfile 后，执行以下命令构建自定义模型（假设 Modelfile 保存在当前目录）：# 格式：ollama create 自定义模型名:版本 -f Modelfile路径 # 示例：构建名为 python-assistant、版本为 v1 的自定义模型，Modelfile 在当前目录 ollama create python-assistant:v1 -f ./Modelfile # 构建成功后，查看本地模型列表，确认自定义模型已存在 ollama list # 运行自定义模型，验证效果 ollama run python-assistant:v1关键说明：构建模型时，基础模型必须已在本地下载；若 Modelfile 中修改了系统提示或参数，需重新执行 create 命令更新自定义模型。4.2 如何通过 API 调用本地模型Ollama 内置 REST API（默认端口 11434），无需额外部署服务，启动 Ollama 服务后即可直接调用，支持交互对话、一次性提问、模型管理等操作，以下提供 curl 示例（快速测试）和 Python 代码示例（实际开发常用）。4.2.1 API 基础信息默认 API 地址：http://localhost:11434/api常用接口：一次性提问：POST /api/generate交互对话（保留上下文）：POST /api/chat模型管理（拉取、删除、查看）：POST /api/pull、DELETE /api/delete 等请求格式：JSON；响应格式：JSON（支持流式响应，适合长文本生成）4.2.2 curl 示例（快速测试 API）适合在终端快速测试 API 可用性，无需编写代码。# 示例 1：一次性提问（无上下文，调用 llama3 模型） curl http://localhost:11434/api/generate \ -H "Content-Type: application/json" \ -d '{ "model": "llama3", "prompt": "用 Python 写一个简单的冒泡排序算法，包含注释", "temperature": 0.6, "max_tokens": 1000 }' # 示例 2：交互对话（保留上下文，调用自定义模型 python-assistant:v1） # 第一次提问（询问冒泡排序） curl http://localhost:11434/api/chat \ -H "Content-Type: application/json" \ -d '{ "model": "python-assistant:v1", "messages": [ {"role": "user", "content": "用 Python 写一个冒泡排序算法"} ] }' # 第二次提问（追问优化方法，上下文关联上一轮） curl http://localhost:11434/api/chat \ -H "Content-Type: application/json" \ -d '{ "model": "python-assistant:v1", "messages": [ {"role": "user", "content": "用 Python 写一个冒泡排序算法"}, {"role": "assistant", "content": "（上一轮模型的回答）"}, {"role": "user", "content": "如何优化这个算法，提升效率？"} ] }'4.2.3 Python 代码示例（实际开发集成）适合集成到 Python 应用中，支持流式输出（实时显示模型回答，避免等待）和非流式输出，以下是完整可运行代码，需先安装 requests 库（pip install requests）。import requests import json # API 基础地址 BASE_URL = "http://localhost:11434/api" def generate_response(model, prompt, temperature=0.7, max_tokens=1000, stream=False): """ 一次性提问（无上下文） :param model: 模型名（如 llama3、python-assistant:v1） :param prompt: 用户提问内容 :param temperature: 生成随机性（0-1） :param max_tokens: 单次回答最大长度 :param stream: 是否流式输出（实时显示回答） :return: 模型回答内容 """ url = f"{BASE_URL}/generate" payload = { "model": model, "prompt": prompt, "temperature": temperature, "max_tokens": max_tokens, "stream": stream } # 流式输出（推荐，适合长文本） if stream: response = requests.post(url, json=payload, stream=True) response.raise_for_status() # 抛出请求错误 full_response = "" for line in response.iter_lines(): if line: data = json.loads(line.decode("utf-8")) # 拼接回答内容，直到生成结束 if "response" in data: full_response += data["response"] print(data["response"], end="", flush=True) # 实时打印 # 生成结束（done 为 true） if data.get("done", False): print("\n") return full_response # 非流式输出（等待完整回答后返回） else: response = requests.post(url, json=payload) response.raise_for_status() data = response.json() return data.get("response", "") def chat_with_model(model, messages, stream=False): """ 交互对话（保留上下文） :param model: 模型名 :param messages: 对话历史列表，格式：[{"role": "user/assistant", "content": "内容"}] :param stream: 是否流式输出 :return: 模型回答内容 """ url = f"{BASE_URL}/chat" payload = { "model": model, "messages": messages, "stream": stream } if stream: response = requests.post(url, json=payload, stream=True) response.raise_for_status() full_response = "" for line in response.iter_lines(): if line: data = json.loads(line.decode("utf-8")) if "message" in data and data["message"]["role"] == "assistant": content = data["message"]["content"] full_response += content print(content, end="", flush=True) if data.get("done", False): print("\n") return full_response else: response = requests.post(url, json=payload) response.raise_for_status() data = response.json() return data.get("message", {}).get("content", "") # ------------------- 测试代码 ------------------- if __name__ == "__main__": # 1. 测试一次性提问（流式输出） print("=== 一次性提问（流式输出） ===") generate_response( model="llama3", prompt="用 Python 写一个简单的文件读写工具，包含异常处理", temperature=0.6, stream=True ) # 2. 测试交互对话（保留上下文） print("=== 交互对话（保留上下文） ===") # 对话历史，初始为空，逐步添加 messages = [ {"role": "user", "content": "什么是 Python 装饰器？用简单的例子说明"} ] # 第一次对话 first_answer = chat_with_model(model="python-assistant:v1", messages=messages, stream=True) # 添加模型回答到对话历史，用于下一轮关联 messages.append({"role": "assistant", "content": first_answer}) # 第二次对话（追问） messages.append({"role": "user", "content": "如何自己写一个装饰器，实现日志打印功能？"}) chat_with_model(model="python-assistant:v1", messages=messages, stream=True)关键说明：运行代码前，需确保 Ollama 服务已启动（brew services start ollama 或系统服务启动）；若需远程调用，需修改 BASE_URL 为远程设备的 IP+端口（如 http://192.168.1.100:11434/api），并确保远程设备防火墙开放 11434 端口。4.3 如何调整参数（及其对生成的影响）Ollama 支持多种生成参数，可通过命令行（run 命令）、Modelfile（PARAMETER 指令）、API 调用（请求参数）三种方式调整，核心参数用于控制生成内容的随机性、长度、上下文范围等，以下是最常用的参数详解。4.3.1 核心参数详解（必掌握）参数名取值范围作用说明对生成的影响temperature（温度）0 ~ 1（默认 0.7）控制生成内容的随机性，值越高，随机性越强；值越低，回答越严谨、确定。• 高值（0.8~1）：适合创意生成（如写文案、编故事），但可能出现逻辑混乱；• 低值（0.2~0.6）：适合事实性问答、编程指导、严谨内容生成，回答更精准。top_p0 ~ 1（默认 0.9）控制生成内容的多样性，基于概率分布筛选候选词，值越低，筛选越严格，内容越集中。与 temperature 配合使用，一般不单独调整；若 temperature 较高，可适当降低 top_p，避免内容过于发散。num_ctx正整数（默认 2048，最大取决于模型）上下文窗口大小，即模型能记住的最大输入+输出长度（单位：token）。• 增大值（如 4096、8192）：支持更长的对话、更长的代码/文档输入，但会占用更多显存/内存；• 减小值：节省资源，但可能导致上下文丢失（如长对话中忘记前面的提问）。max_tokens正整数（默认 1000）限制单次生成的最大 token 数，避免输出过长。根据需求调整，如编程指导可设为 2000，短问答可设为 500；若生成内容被截断，需增大该值。stop字符串或字符串数组设置停止生成的触发词，当模型生成到触发词时，立即停止。适合控制输出格式，如设置 stop: "用户："，避免模型模拟用户提问；设置 stop: "nn"，避免输出过多冗余内容。4.3.2 其他常用参数（可选）num_predict：与 max_tokens 功能一致，用于兼容旧版本，优先使用 max_tokens。top_k：取值范围 0~100（默认 40），控制候选词的数量，值越低，生成内容越集中，一般无需调整。repeat_penalty：取值范围 1.0~2.0（默认 1.1），控制重复内容的生成，值越高，越能避免重复（如避免连续重复同一句话）。4.3.3 参数调整示例# 1. 命令行运行时调整参数（单次生效） ollama run llama3 --temperature 0.5 --num_ctx 4096 --max_tokens 2000 # 2. Modelfile 中设置参数（永久生效，针对自定义模型） PARAMETER temperature 0.6 PARAMETER num_ctx 4096 PARAMETER stop "用户：" # 3. API 调用时调整参数（单次生效） curl http://localhost:11434/api/generate \ -H "Content-Type: application/json" \ -d '{ "model": "llama3", "prompt": "写一篇关于本地大模型的技术短文", "temperature": 0.8, "top_p": 0.85, "max_tokens": 1500 }'优化建议：参数调整无固定标准，需根据场景测试优化——事实性问答（temperature 0.3~0.6，num_ctx 2048~4096）；创意生成（temperature 0.7~0.9，top_p 0.9）；长文档处理（num_ctx 4096~8192，max_tokens 2000~4000）。5. 实战场景结合前面的基础命令和进阶用法，以下三个实战场景覆盖开发者最常用的需求，每个场景提供清晰的实现思路和关键步骤，可直接落地。5.1 场景一：搭建本地私有知识库问答核心需求：将企业内部文档、个人笔记等私有数据导入，让本地模型基于这些数据进行问答，确保数据隐私不泄露（无需上传云端）。实现思路（简洁易懂，适合新手）准备私有数据：整理需要导入的文档（支持 TXT、PDF、Word 等格式），将文档转换为纯文本（避免格式干扰），可使用工具（如 PyPDF2 库）批量提取 PDF 文本。数据处理（关键步骤）：将纯文本分割为小块（每块 500~1000 token，避免超过模型的 num_ctx），并为每块添加简单的标题或标签（方便模型定位相关内容）。构建提示模板：创建自定义 Modelfile，在 SYSTEM 提示中加入“仅基于提供的私有数据回答，若数据中没有相关信息，直接说明‘未找到相关内容’，不编造答案”。实现问答交互：通过 API 调用模型，将用户提问与分割后的私有数据块拼接为 prompt，传递给模型，让模型基于私有数据生成回答。优化体验：可搭建简单的 Web 界面（如用 Flask/Django），实现“上传文档→提问→查看回答”的可视化操作，无需手动拼接 prompt。关键工具推荐：数据处理用 PyPDF2（提取 PDF 文本）、LangChain（分割文本、构建知识库）；Web 界面用 Flask（轻量易上手）。5.2 场景二：结合 LangChain 或 LlamaIndex 进行开发核心需求：LangChain 和 LlamaIndex 是大模型开发框架，可简化知识库构建、多模型集成、工具调用等流程，结合 Ollama 可快速实现复杂的本地大模型应用（如智能问答、文档分析）。示例：结合 LangChain 搭建本地知识库问答（关键代码）前提：安装依赖库 pip install langchain ollama pypdffrom langchain.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OllamaEmbeddings from langchain.vectorstores import FAISS from langchain.chains import RetrievalQA from langchain.llms import Ollama # 1. 加载私有 PDF 文档（可替换为自己的文档路径） loader = PyPDFLoader("private_knowledge.pdf") documents = loader.load() # 2. 分割文档（避免超过模型上下文窗口） text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 每块 500 token chunk_overlap=50 # 块之间重叠 50 token，保证上下文连贯 ) splits = text_splitter.split_documents(documents) # 3. 生成文档嵌入（用于后续检索），使用 Ollama 本地模型 embeddings = OllamaEmbeddings(model="llama3") # 4. 构建向量数据库（存储文档嵌入，方便快速检索相关内容） vectorstore = FAISS.from_documents(documents=splits, embedding=embeddings) # 5. 初始化 Ollama 本地模型 llm = Ollama(model="llama3", temperature=0.5) # 6. 构建检索式问答链（检索相关文档 + 模型生成回答） qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 简单直接，将检索到的文档拼接后传递给模型 retriever=vectorstore.as_retriever(k=3), # 每次检索 top3 相关文档 return_source_documents=True # 返回回答对应的原始文档（方便验证） ) # 7. 测试问答 query = "请基于文档内容，说明本地大模型的部署优势？" result = qa_chain({"query": query}) # 输出结果 print("回答：", result["result"]) print("\n相关文档：") for doc in result["source_documents"]: print("-", doc.page_content[:100], "...")关键说明：LlamaIndex 的实现思路类似，核心是“文档加载→分割→嵌入→检索→问答”，相比 LangChain 更专注于知识库构建，适合新手快速上手；可替换模型为自定义模型（如 python-assistant:v1），适配具体场景。5.3 场景三：在终端中作为智能助手使用核心需求：无需打开浏览器、无需编写代码，直接在终端中使用 Ollama 模型作为智能助手，快速解决问题（如编程调试、命令查询、知识问答）。实现步骤（简单易操作）创建自定义助手模型：编写 Modelfile，定义助手的角色（如“终端智能助手，擅长解答编程、命令行、技术问题，回答简洁，提供具体示例”），设置合适的参数（temperature 0.6，num_ctx 4096）。构建并运行自定义模型：`ollama create terminal-assistant:v1 -f ./Modelfilerun terminal-assistant:v1`常用场景示例：提问：“如何在 Linux 中查看进程占用的内存？” → 模型输出具体命令（top、ps aux）及使用说明。提问：“用 Bash 写一个批量重命名文件的脚本” → 模型输出可直接复制执行的脚本及注释。提问：“解释一下 Python 中的列表推导式” → 模型输出简洁的解释和示例。优化体验：可将 ollama run terminal-assistant:v1 封装为终端别名（如 alias ai='ollama run terminal-assistant:v1'），后续输入 ai 即可快速启动助手。6. 常见问题与优化 (FAQ)整理开发者使用 Ollama 时最常遇到的问题，提供具体的解决方案，帮助快速排查故障、优化性能。6.1 显存不足怎么办？核心原因：模型体积过大（如 70B 模型），超出显卡显存容量；或同时运行多个模型，占用过多显存。解决方案（按优先级排序）：使用量化版本的模型：优先选择 q4_0、q4_1 量化版（如 llama3:70b-q4_0），相比原始模型（fp16），显存占用可减少 50% 以上，是最有效的解决方案。减小 num_ctx 参数：降低上下文窗口大小（如从 4096 改为 2048），减少显存占用，但需注意避免上下文丢失。关闭其他占用显存的程序：关闭显卡加速的程序（如游戏、其他模型服务），释放显存资源。使用 CPU 推理（应急方案）：若没有独立显卡或显存不足，可强制使用 CPU 推理，命令如下： `ollama run llama3 --cpu-only`注意：CPU 推理速度较慢，适合小型模型（如 phi3:mini）。 升级硬件：长期使用建议升级显卡（选择显存 ≥ 16GB 的 NVIDIA 显卡，如 RTX 4070、RTX 3090）。6.2 如何量化模型以节省资源？模型量化是通过降低模型权重的精度（如从 16 位浮点型改为 4 位、8 位整数），在牺牲少量推理精度的前提下，大幅减少模型体积和资源占用，Ollama 支持两种量化方式。方式 1：直接拉取官方量化模型（推荐，新手首选）Ollama 官方仓库提供多种量化等级的模型，后缀格式为 -q4_0、-q4_1、-q8_0，直接拉取即可：# 4-bit 量化（最节省资源，精度略有下降） ollama pull llama3:8b-q4_0 # 8-bit 量化（兼顾精度和资源，推荐大多数场景） ollama pull llama3:8b-q8_0 # 16-bit 量化（原始精度，资源占用最高） ollama pull llama3:8b-fp16方式 2：手动量化本地模型（进阶，适合自定义模型）若自定义模型未量化，可使用 Ollama 自带的量化工具，步骤如下：# 1. 查看本地模型的 ID（通过 ollama list 获取） ollama list # 2. 手动量化模型（格式：ollama quantize 模型ID 输出模型名:版本 量化等级） # 量化等级：q4_0、q4_1、q8_0、fp16 ollama quantize 78e26419b446 my-llama3:q4_0 q4_0 # 3. 验证量化后的模型 ollama list # 查看量化后的模型是否存在 ollama run my-llama3:q4_0 # 测试运行6.3 网络连接问题处理常见问题：拉取模型时超时、无法连接 Ollama 官方仓库、API 调用失败。解决方案：拉取模型超时：原因：网络不稳定、官方仓库访问速度慢。解决：使用国内镜像源（如 ollama pull registry.cn-hangzhou.aliyuncs.com/ollama/llama3:latest）；或分段拉取，中断后重新执行 ollama pull 命令，会自动续传。无法连接官方仓库：原因：防火墙拦截、代理设置异常。解决：关闭本地防火墙（临时测试）；若使用代理，配置终端代理（如 export HTTP_PROXY=http://代理IP:端口）；检查网络是否能访问 https://ollama.com。API 调用失败：原因：Ollama 服务未启动、端口被占用、IP 或端口错误。解决：重启 Ollama 服务；检查 11434 端口是否被占用（netstat -tuln | grep 11434），若被占用，关闭占用端口的程序；确认 API 地址正确（http://localhost:11434/api）。6.4 其他常见问题模型运行速度慢：使用 GPU 推理（确保显卡支持 CUDA/Metal）；使用量化模型；减小 num_ctx 和 max_tokens 参数。模型回答偏离主题：降低 temperature 参数（如 0.5 以下）；调整 top_p 参数；在 SYSTEM 提示中明确回答规则。Windows 系统启动 Ollama 失败：以管理员身份运行终端；重启电脑；重新安装 Ollama 并检查环境变量。7. 学习资源推荐整理官方、社区、实战相关的学习资源，帮助开发者快速深入学习 Ollama，解决实际开发中的问题。7.1 官方资源（最权威，优先查看）Ollama 官方文档：https://ollama.com/docs（包含安装、命令、API、Modelfile 详解）Ollama 官方模型仓库：https://ollama.com/library（所有支持的预训练模型，可查看量化版本）Ollama GitHub 仓库（注：文档部分内容可能由 AI 生成）
import asyncioimport aiohttpimport timeimport osfrom anyio import Semaphorefrom openpyxl import Workbookfrom dotenv import load_dotenv1. 环境配置（复用之前的，无需修改）load_dotenv()LLM_API_KEY = os.getenv("LLM_API_KEY")if not LLM_API_KEY:raise ValueError("❌ 未设置 LLM_API_KEY，请检查 .env 文件") LLM_API_URL = ("https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation")2. 整合工具类（异步API+Excel保存）class AsyncClassifier:def __init__(self): self.wb = Workbook() self.ws = self.wb.active self.ws.title = "政务文本分类结果" self.headers = [ "文本ID", "原文", "分类标签", "置信度", "处理时间", "调用状态", "异常信息", ] self.ws.append(self.headers) self.text_id = 1 # 异步API调用（复用例子3） async def call_llm_async(self, text: str) -> dict: try: async with aiohttp.ClientSession() as session: categories = ["政策通知", "群众咨询", "投诉举报", "政务公告", "其他"] prompt = ( f"你是一个政务文本分类器。请将以下文本严格分类为：{', '.join(categories)}。\n" f"只输出类别名称，不要任何解释或标点。\n\n" f"文本：{text}\n" f"类别：" ) # 构造请求头和请求体 headers = { "Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json", } payload = { "model": "qwen-turbo-latest", # 或 qwen-max / qwen-plus "input": {"messages": [{"role": "user", "content": prompt}]}, "parameters": { "max_tokens": 10, "temperature": 0.1, # 降低随机性，提高稳定性 "seed": 12345, # 可选：固定种子 }, } async with session.post( url=LLM_API_URL, headers=headers, json=payload, timeout=30 ) as reponse: result = await reponse.json() # print(result, "\n") # 提取模型输出 output_text = result.get("output", {}).get("text", "").strip() # 简单置信度模拟（实际无法获取真实 confidence，可设为 1.0 或基于规则） confidence = 1.0 if output_text in categories else 0.0 return { "text": text, "category": output_text if confidence > 0 else "", "confidence": confidence, "status": "成功" if confidence > 0 else "失败", "error_msg": ( "" if confidence > 0 else f"模型返回无效类别: '{output_text}'" ), } except Exception as e: return { "text": text, "category": "", "confidence": 0.0, "status": "失败", "error_msg": str(e), } # 批量异步分类+保存结果（核心整合） async def batch_classify_and_save(self, texts: list[str], batch_size: int = 10): """异步批量分类，结果保存到Excel""" start_time = time.time() # 并发控制：限制同时发起10个API请求（避免被限流） semaphore = asyncio.Semaphore(10) # 带并发控制的异步任务（避免请求过多） async def bounded_task(text): async with semaphore: return await self.call_llm_async(text) # 批量添加任务（按批次拆分，可选，提升可控性） tasks = [] for i in range(0, len(texts), batch_size): batch_texts = texts[i : i + batch_size] for text in batch_texts: tasks.append(bounded_task(text)) # 执行任务，获取结果 results = await asyncio.gather(*tasks) # 保存结果到Excel self.save_results(results) total_time = round(time.time() - start_time, 2) print(f"批量分类完成！共{len(texts)}篇，耗时{total_time}秒") return results # Excel保存功能（复用例子1，优化异常处理） def save_results(self, results: list[dict]): try: for res in results: current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) row_data = [ self.text_id, res["text"], res["category"], res["confidence"], current_time, res["status"], res["error_msg"], ] self.ws.append(row_data) self.text_id += 1 self.wb.save("批量分类结果.xlsx") print("结果已保存到Excel") except Exception as e: print(f"保存失败：{str(e)}") self.wb.save("分类结果_备份.xlsx") 测试代码if name == "__main__":# 测试10条文本（模拟批量场景） test_texts = [ f"政务文本{i}：" + ( "政策通知内容" if i % 5 == 0 else "群众咨询内容" if i % 5 == 1 else "投诉举报内容" ) for i in range(1, 11) ] # 初始化工具类，执行批量分类 classifier = AsyncClassifier() asyncio.run(classifier.batch_classify_and_save(test_texts, batch_size=5))
异步调用API（关键优化，提升效率）import asynciofrom urllib import responseimport aiohttpimport osfrom dotenv import load_dotenv1. 环境配置（复用之前的，无需修改）load_dotenv()LLM_API_KEY = os.getenv("LLM_API_KEY")if not LLM_API_KEY:raise ValueError("❌ 未设置 LLM_API_KEY，请检查 .env 文件") LLM_API_URL = ("https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation")async def call_llm_async(text: str):try: """异步调用大模型API，单条文本分类""" async with aiohttp.ClientSession() as session: # 构造 Prompt（关键：让模型知道要做什么） categories = ["政策通知", "群众咨询", "投诉举报", "政务公告", "其他"] prompt = ( f"你是一个政务文本分类器。请将以下文本严格分类为：{', '.join(categories)}。\n" f"只输出类别名称，不要任何解释或标点。\n\n" f"文本：{text}\n" f"类别：" ) # 构造请求头和请求体 headers = { "Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json", } payload = { "model": "qwen-turbo-latest", # 或 qwen-max / qwen-plus "input": {"messages": [{"role": "user", "content": prompt}]}, "parameters": { "max_tokens": 10, "temperature": 0.1, # 降低随机性，提高稳定性 "seed": 12345, # 可选：固定种子 }, } async with session.post( url=LLM_API_URL, headers=headers, json=payload, timeout=30 ) as reponse: result = await reponse.json() # print(result, "\n") # 提取模型输出 output_text = result.get("output", {}).get("text", "").strip() # 简单置信度模拟（实际无法获取真实 confidence，可设为 1.0 或基于规则） confidence = 1.0 if output_text in categories else 0.0 return { "text": text, "category": output_text if confidence > 0 else "", "confidence": confidence, "status": "成功" if confidence > 0 else "失败", "error_msg": ( "" if confidence > 0 else f"模型返回无效类别: '{output_text}'" ), } except Exception as e: return { "text": text, "category": "", "confidence": 0.0, "status": "失败", "error_msg": str(e), } async def batch_async_classify(texts: list[str]):"""批量异步分类，同时发起多个API请求""" tasks = [call_llm_async(text) for text in texts] # 批量执行所有任务，等待全部完成（asyncio.gather是核心） results = await asyncio.gather(*tasks) return results if name == "__main__":# 测试3条文本，异步同时调用API test_texts = [ "济南市2024年民生政策通知", "群众咨询社保缴费流程", "投诉举报小区垃圾清运不及时", ] # 启动异步任务，获取结果 results = asyncio.run(batch_async_classify(test_texts)) # 打印结果 for res in results: print(res)
核心第一步——同步调用大模型API（理解API交互）import requestsimport osfrom dotenv import load_dotenv1. 环境配置load_dotenv()LLM_API_KEY = os.getenv("LLM_API_KEY")if not LLM_API_KEY:raise ValueError("❌ 未设置 LLM_API_KEY，请检查 .env 文件") LLM_API_URL = ("https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation")2. 同步调用APIdef call_llm_sync(text: str):"""同步调用 DashScope Qwen 模型进行文本分类""" try: # 构造 Prompt（关键：让模型知道要做什么） categories = ["政策通知", "群众咨询", "投诉举报", "政务公告", "其他"] prompt = ( f"你是一个政务文本分类器。请将以下文本严格分类为：{', '.join(categories)}。\n" f"只输出类别名称，不要任何解释或标点。\n\n" f"文本：{text}\n" f"类别：" ) # 构造请求头和请求体 headers = { "Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json", } payload = { "model": "qwen-turbo-latest", # 或 qwen-max / qwen-plus "input": {"messages": [{"role": "user", "content": prompt}]}, "parameters": { "max_tokens": 10, "temperature": 0.1, # 降低随机性，提高稳定性 "seed": 12345, # 可选：固定种子 }, } # 发送 POST 请求 response = requests.post( url=LLM_API_URL, headers=headers, json=payload, timeout=30 ) response.raise_for_status() # 抛出 HTTP 异常（如 401, 400, 500） result = response.json() # 提取模型输出 output_text = result.get("output", {}).get("text", "").strip() # 简单置信度模拟（实际无法获取真实 confidence，可设为 1.0 或基于规则） confidence = 1.0 if output_text in categories else 0.0 return { "text": text, "category": output_text if confidence > 0 else "", "confidence": confidence, "status": "成功" if confidence > 0 else "失败", "error_msg": "" if confidence > 0 else f"模型返回无效类别: '{output_text}'", } except Exception as e: return { "text": text, "category": "", "confidence": 0.0, "status": "失败", "error_msg": str(e), } 测试代码if name == "__main__":test_text = "群众投诉小区垃圾清运不及时，影响生活环境" result = call_llm_sync(test_text) print("API调用结果：", result)