Python 这个系列写到第二篇说明你已经过了“装环境、调通第一个 API、跑通一个 demo”的阶段。上一篇我们解决了“能不能跑起来”的问题这一篇我要集中解决另一个更现实的问题怎么让大语言模型从“能跑”变成“能用”。我见过太多人卡在这个阶段模型调用会写了一上真实业务就废——批量数据不会并发、参数靠硬编码、本地模型部署完不知道怎么接进代码、模型输出的 JSON 动不动解析失败。这篇文章就把这些“脏活”逐个拆开讲全部可以照着抄。1. 重装完 Python 之后先搞定这四件事再碰模型我默认你已经装好了 Python 3.9 或 3.10 以上的版本并且能跑通最基本的import torch或者import openai。如果还没到这个程度先把上一篇的基础过一遍否则下面内容你会卡在环境上。大语言模型项目对 Python 环境的要求非常苛刻它不像写爬虫那样随便装个 requests 就能跑。1.1 虚拟环境不是可选项是保命项大语言模型相关的库迭代速度极快。你上个月装好的 transformers 是这个版本下个月项目要求你装另一个版本如果所有库都混在同一个全局环境里最后一定会出现“上午装好 A 库下午 B 库崩掉”的局面。我一个朋友的真实经历项目里需要旧版 numpy他直接在全局环境里降级了 numpy结果把另一个项目里依赖新版 numpy 的 pandas 搞崩了最后花了半天时间排查版本冲突。所以第一步永远是创建虚拟环境python -m venv llm_env激活环境# Windows llm_env\Scripts\activate.bat # Linux / macOS source llm_env/bin/activateVSCode 用户注意激活终端之后按CtrlShiftP打开命令面板输入 “Python: Select Interpreter”选择llm_env下的解释器路径。很多人环境激活了代码还是跑在旧解释器上报错说“找不到某个包”但终端里明明显示已经激活了问题就出在这里。1.2 依赖锁版本requirements.txt 不是写完就完事每次环境里成功跑通一个模型后第一件事是导出依赖清单pip freeze requirements.txt这句话的意义在于把当前环境里所有包以及精确版本号记录下来。一个月后换电脑、换队友、上服务器部署用pip install -r requirements.txt就能还原出几乎一致的环境。注意我强调“精确版本号”。不要图省事只写transformers不写版本号因为 transformers 每个小版本之间的 API 变化都很大。之前有个项目别人给我的 requirements.txt 里写着transformers4.30结果跑代码时报错找不到某个模型类的属性最后发现是 transformers 新版本改了接口。锁版本锁的是可复现性不锁版本就是在给自己埋雷。1.3 解释器路径为什么坚持用 python -m pip很多人习惯直接pip install xxx但这条命令有个隐患它安装到的解释器不一定是你当前正在用的那个。Windows 上最容易出现这种问题——系统装了 Python 3.8又装了 3.11环境变量 PATH 里哪个在前面pip默认就对应哪个。我现在的习惯是统一用python -m pip install xxxpython -m pip的含义是“用当前被执行的 python 解释器跑来调用 pip”这样装到哪一版解释器上一目了然绝对不会装错位置。这个习惯在你机器上有多个 Python 版本时尤其重要。检查当前解释器路径# Windows where python # Linux / macOS which python如果python和python3指向不同路径就用python3 -m pip install或python -m pip install严格区分。1.4 ModuleNotFoundError 的定位思路跑模型时报ModuleNotFoundError: No module named torch别急着重装。先查两件事当前解释器是谁python -c import sys; print(sys.executable)包里有没有python -m pip list | findstr torchWindows或python -m pip list | grep torchLinux/macOS如果解释器路径和你激活的环境不一致那问题不在“没安装”而在“装错了地方”。这种情况我遇到不下十次每次都有人问我“为什么 pip list 里有但 import 报错”答案永远是解释器没对齐。2. 本地部署大语言模型从下载模型到第一次推理这个系列上一篇应该已经带你走通过云端 API 调用。这一篇重点说本地部署因为这是热搜词里出现频率最高的词也是很多人的心结网上教程满天飞但真正跑通本地模型的人少之又少。2.1 本地部署图什么本地部署大语言模型的核心动机是三个字可控性。API 调用很方便但有几类场景必须走本地数据敏感合同、病历、企业内部文档不能传到第三方 API。长期成本高频调用时API 按 token 收费本地部署是一次性硬件投入。离线环境某些内网环境连外网都不通API 完全不可用。我做过一个本地处理合同的项目客户明确要求所有文本不能出内网。最后就是一台 GPU 服务器 本地模型扛下来的。如果你没有 GPU也不用慌7B 以下的小模型用 CPU 也能跑就是速度慢些但几十条短文本的批处理完全能接受。2.2 推理框架选型三个主流方案怎么挑框架语言/生态硬件要求适合场景OllamaGo 写的API 极简一条命令拉起服务中低支持 CPU/GPU快速体验、本地服务、不想折腾配置llama.cppC 实现CPU 友好依赖少低纯 CPU 也可以低配机器、边缘设备Hugging Face TransformersPython 生态与 PyTorch 深度绑定高GPU 优先研究、微调、精细控制、与 Python 项目无缝集成我的建议是如果你是入门先用 Ollama。它把模型下载、量化、服务启动全部封装好了一条命令就能把模型跑成 HTTP 服务。如果你想做研究、微调或者模型和业务代码深度耦合再上 Transformers。llama.cpp 适合资源受限的部署环境代码里的集成难度高一些。2.3 拿到模型文件下载、格式与路径本地模型最常见的两种文件格式safetensorsPyTorch 生态的主流格式安全、加载快Transformer 项目默认使用。GGUFllama.cpp/Ollama 生态使用的量化格式把模型体积压缩到原来的四分之一左右换取更低的硬件门槛。下载模型推荐优先从ModelScope这类国内可直接访问的模型社区获取避免网络不稳定的问题。如果你在 HuggingFace 上找到了模型也可以用它的下载工具配合断点续传。下载完成后注意模型目录结构是否完整config.json、tokenizer.json、model.safetensors这些文件缺一不可。我给一个推荐的本地模型目录组织方式models/ └── Qwen2.5-1.5B-Instruct/ ├── config.json ├── tokenizer.json ├── model.safetensors └── generation_config.json把模型目录和代码目录分开后面写配置文件时路径管理会更清楚。2.4 第一次推理最容易出的三个错写一个最小推理代码建议先用小参数量模型比如 1.5B 级别把链路跑通再换大模型from transformers import AutoModelForCausalLM, AutoTokenizer import torch model_path ./models/Qwen2.5-1.5B-Instruct tokenizer AutoTokenizer.from_pretrained(model_path) model AutoModelForCausalLM.from_pretrained( model_path, device_mapauto, torch_dtypetorch.float16 ) messages [ {role: system, content: 你是一个数据分析助手 \ 回答要简洁只输出结论。}, {role: user, content: 用一句话解释什么是大语言模型。} ] text tokenizer.apply_chat_template( messages, tokenizeFalse, add_generation_promptTrue ) inputs tokenizer(text, return_tensorspt).to(model.device) outputs model.generate( **inputs, max_new_tokens256, do_sampleFalse ) result tokenizer.decode(outputs[0][inputs[input_ids].shape[1]:], skip_special_tokensTrue) print(result)三个最常见的坑显存不足CUDA out of memory报错后先看模型有多少参数、用了几倍显存。1.5B 模型 if16 大约占 3GB7B 模型大约 14GB。如果不够走量化方案GGUF或者把torch_dtype换成float32但速度会慢。2.device_mapauto有可能把不同层分配到不同设备这是好事CPUGPU 混合也能跑但如果你强制.to(cuda:0)就会报设备不匹配。第一次跑通前别乱加.to()。pad_token警告很多模型的 tokenizer 没有设置pad_token生成时会出现黄色警告不影响结果但最好在推理前补一句if tokenizer.pad_token_id is None: tokenizer.pad_token_id tokenizer.eos_token_id3. 为什么大语言模型项目几乎都用 YAML 管配置热搜词里有一条是“大语言模型是不是主流用 yaml 提供配置参数”。答案是是的而且这几乎是工程默认选择。这一章我把原因和具体用法一次讲透。3.1 YAML 比 JSON 和 Python 文件强在哪配置管理要解决三个问题可读、可改、可追溯。YAML 在这三方面都做得很好有注释项目里每个参数代表什么意思直接写在配置里。JSON 不支持注释看完一个 200 行的配置头都大了。层级清晰生成相关参数和API 相关参数用缩进做子级一眼能看出分组。JSON 要一层层找大括号。Diff 友好代码评审时YAML 的缩进结构让修改点一目了然。更关键的是整个生态已经习惯了这套习惯。Kubernetes 用 YAML 写编排配置CI/CD 用 YAML 写流水线大语言模型项目沿用这个惯例新成员接手成本最低。3.2 PyYAML 的读取套路与安全提醒先看一个典型的大语言模型项目配置# config.yaml model: name: Qwen2.5-1.5B-Instruct path: ./models/Qwen2.5-1.5B-Instruct dtype: float16 generation: temperature: 0.7 top_p: 0.9 max_new_tokens: 512 batch: concurrency: 4 output_dir: ./output读取的代码import yaml with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) model_name config[model][name] temperature config[generation][temperature] concurrency config[batch][concurrency]千万不要用yaml.load(f)——这个函数默认使用不安全的 Loader如果配置里被人塞进恶意对象有反序列化攻击风险。yaml.safe_load只解析基础类型是官方推荐的安全选项。3.3 一套配置多个场景环境切换怎么做实际项目里开发环境和生产环境的模型路径、并发数往往不同。我的做法是准备一个config.yaml作为基础配置再加一个config.prod.yaml做覆盖import os import yaml env os.getenv(LLM_ENV, dev) config_file fconfig.{env}.yaml if os.path.exists(fconfig.{env}.yaml) else config.yaml with open(config_file, r, encodingutf-8) as f: config yaml.safe_load(f)运行时用环境变量指定配置LLM_ENVprod python main.py代码不需要改一行。另外一个经验敏感信息不要进 YAML。API Key、数据库密码这类信息应该从环境变量或密钥管理服务里读否则一旦配置文件被提交到 Git 仓库密钥等于公开。3.4 哪些配置该进 YAML哪些不该进该进 YAML 的模型名称与路径、生成参数temperature、top_p、max_new_tokens、批处理并发数、输出目录。不该进 YAML 的API Key、数据库密码、任何和权限相关的内容动态变化的临时参数比如“本次只跑前 100 条数据”——这种一次性参数用命令行参数--limit 100更好。4. 让模型按你的要求说话流式输出与结构化返回模型输出的质量不仅取决于模型本身还取决于你怎么调用它。这一章讲两个产品级调用必备技巧流式输出和结构化返回。4.1 流式输出别让用户对着白屏等十秒大语言模型生成文本是一个 token 一个 token 产生的服务端完全可以边生成边推送。如果不做流式用户点击按钮后要等 5~10 秒才看到整段文字体验极差。做流式之后文字像打字一样逐字呈现。如果你用 Hugging Face Transformers可以用TextIteratorStreamerfrom transformers import TextIteratorStreamer from threading import Thread streamer TextIteratorStreamer(tokenizer, skip_promptTrue) generation_kwargs { input_ids: inputs[input_ids], max_new_tokens: 512, do_sample: False, streamer: streamer, } thread Thread(targetmodel.generate, kwargsgeneration_kwargs) thread.start() for new_text in streamer: print(new_text, end, flushTrue)注意TextIteratorStreamer必须配合多线程使用因为generate是阻塞调用如果你在主线程里等它跑完就看不到流式效果了。如果你用的是 APIOpenAI 兼容接口更快from openai import OpenAI client OpenAI(base_urlhttp://localhost:11434/v1, api_keynone) stream client.chat.completions.create( modelqwen2.5:1.5b, messages[{role: user, content: 写一段 500 字的产品介绍}], streamTrue ) for chunk in stream: delta chunk.choices[0].delta.content if delta: print(delta, end, flushTrue)4.2 让模型稳定吐 JSON 的三种手段模型返回文档字段下游程序要解析成结构化数据。如果模型返回的是散文代码就没法处理。所以你要用三种手段层层加固第一提示词明确要求。在 system 消息里写死格式约束你是一个文本分析工具。只输出 JSON 格式不允许输出其他文字。 JSON 格式示例{sentiment: positive, confidence: 0.95}第二如果用的是 OpenAI 兼容接口直接在请求参数里指定。OpenAI SDK 支持response_format{type: json_object}会强制模型输出 JSON。第三做解析兜底。本地模型即使提示词写得再好偶尔也会把 JSON 包在 markdown 代码块里或者其他地方加描述性语言。所以解析时不能直接json.loads(result)要提取import json import re def parse_json_from_text(text): # 优先找 json ... 代码块 match re.search(r(?:json)?\s*([\s\S]*?), text) if match: text match.group(1) # 再找第一个 { 到最后一个 } start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(f未找到 JSON 片段原文{text[:200]}) return json.loads(text[start:end1])这个函数是我在自己项目里踩过几次坑后写出来的。它先处理 markdown 包装再做首尾裁剪能解决 90% 的 JSON 解析问题。4.3 批量分析场景爬虫数据 提示词模板 结果规整实战场景用 Python 爬了一批商品评论下来文本存在 CSV 里现在想把每条评论做情感分类。伪代码思路import csv import json def build_prompt(comment): system_prompt 你是一个电商评论分析助手。只输出 JSON \ 格式为{\sentiment\: \positive|negative|neutral\, \summary\: \一句话总结\} user_prompt f请分析以下评论\n{comment} return [ {role: system, content: system_prompt}, {role: user, content: user_prompt} ] with open(comments.csv, r, encodingutf-8) as f: reader csv.DictReader(f) comments [row[comment] for row in reader] results [] for comment in comments[:20]: # 先跑 20 条做验证 response client.chat.completions.create( modelqwen2.5:1.5b, messagesbuild_prompt(comment), max_tokens128 ) try: parsed parse_json_from_text(response.choices[0].message.content) results.append({comment: comment, **parsed}) except ValueError as e: results.append({comment: comment, error: str(e)})注意我用的是for comment in comments[:20]。任何批量任务都要先在小样本上验证输出格式确认干净之后再全量跑。我见过有人一次性跑 5000 条模型输出的 JSON 格式有偏差解析失败率高达 30%又得重跑浪费时间也烧 token。5. 协程批量调用写一个不阻塞的并发客户端第 4 章的批量代码是同步 for 循环每条评论等模型生成完再下一条。数据量小看不出问题数据量一大就完全扛不住。这一章解决并发问题。5.1 同步调用慢在哪算一笔账200 条数据每条模型生成时间平均 2 秒同步跑就是 400 秒6 分多钟。如果其中有几条特别长还要更久。并发之后用 8 个并发理论耗时降到 50 秒左右。如果用的是 API 服务并发还能拉高几百条数据一分钟内就能跑完。5.2 asyncio 异步客户端的批量骨架用 OpenAI 异步客户端做这事最省事import asyncio from openai import AsyncOpenAI client AsyncOpenAI( base_urlhttp://localhost:11434/v1, api_keynone ) async def analyze_one(comment): response await client.chat.completions.create( modelqwen2.5:1.5b, messagesbuild_prompt(comment), max_tokens128 ) return comment, response.choices[0].message.content async def analyze_many(comments, limit8): results [] semaphore asyncio.Semaphore(limit) async def worker(comment): async with semaphore: return await analyze_one(comment) tasks [worker(c) for c in comments] return await asyncio.gather(*tasks) comments [...] # 读取自 CSV results asyncio.run(analyze_many(comments))关键点在asyncio.Semaphore(limit)。它限定了同时进行的请求数量。没有这个信号量如果本地模型服务一次只能同时处理一个请求你发 50 个并发请求过去大概率直接 OOM 或者排队严重速度反而更慢。信号量一定要加数值最好在4~8之间先调试。异步接口不是魔法。本地模型推理本身也是耗时的但是有了并发可以让多个请求排队进入 GPU 推理服务服务端批量处理效率远高于客户端串行等待。5.3 超时与重试把网络请求当合格公民对待并发起来之后单个请求失败率也会被放大。所以要给每个请求设超时时间并做重试。用 tenacity 库写重试最方便from tenacity import retry, stop_after_attempt, wait_exponential from openai import APIError retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retry(APIError, TimeoutError), reraiseTrue ) async def analyze_one_retry(comment): return await analyze_one(comment)规则最多重试 3 次第一次失败等 2 秒第二次等 4 秒第三次等 8 秒指数退避。这种策略比固定等 5 秒更科学因为如果服务端正在过载立刻重试往往还是会失败。5.4 本地 GPU 只有一个并发真的有效吗很多人会有疑问本地就一块 GPU并发不是自己抢自己吗两种场景下并发仍然有效GPU 显存够大推理服务可以 batch 处理多个请求推理总吞吐量翻倍。生成阶段和预填充阶段交织多个请求交替填充 GPU 的计算流水线让显存和算力都被喂饱。如果你发现并发加到一定程度后吞吐量不再提升甚至下降那就是到瓶颈了。这时候的解法不是在客户端加并发而是换一个支持动态批处理的推理框架比如 vLLM或者干脆换更大的显存。并发数要配合服务端能力调节不是越大越好。6. 完整实例本地模型 YAML 配置 协程批量情感分析前面五章讲的都是独立知识点这一章把它们拼成一个最小可用项目。这个项目可以直接改改就用读一个 CSV 里的短文本调用本地部署的模型做情感分析把结果写回新的 CSV。6.1 项目结构sentiment/ ├── config.yaml ├── client.py ├── analyzer.py ├── main.py └── comments.csv这个结构非常简单但保留了良好的分层配置归配置、接口归接口、业务归业务。6.2 config.yaml先把配置写明白model: base_url: http://localhost:11434/v1 name: qwen2.5:1.5b analysis: max_tokens: 128 concurrency: 4 system_prompt: 你是一个电商评论分析助手。只输出 JSON格式为 {\sentiment\: \positive|negative|neutral\, \summary\: \一句话总结\} io: input_file: ./comments.csv output_file: ./results.csv把system_prompt放在配置里是个好习惯——你想改提示词不需要改代码只改 YAML 再重新运行就行。6.3 各模块代码client.py封装异步调用带重试。import asyncio from openai import AsyncOpenAI from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_not_exception_type client None def init_client(base_url): global client client AsyncOpenAI(base_urlbase_url, api_keynone) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_not_exception_type(ValueError), reraiseTrue ) async def ask_model(messages, max_tokens128): response await client.chat.completions.create( modelconfig_model_name, messagesmessages, max_tokensmax_tokens ) return response.choices[0].message.contentanalyzer.py负责构造 prompt 和解析结果。import json import re from client import ask_model def build_messages(system_prompt, comment): return [ {role: system, content: system_prompt}, {role: user, content: f请分析以下评论\n{comment}} ] def parse_json_from_text(text): match re.search(r(?:json)?\s*([\s\S]*?), text) if match: text match.group(1) start text.find({) end text.rfind(}) if start -1 or end -1: raise ValueError(f未找到 JSON 片段) return json.loads(text[start:end 1]) async def analyze(comment, config): messages build_messages(config[analysis][system_prompt], comment) raw await ask_model(messages, max_tokensconfig[analysis][max_tokens]) return parse_json_from_text(raw)main.py读取配置、加载 CSV、并发调用、写回结果。import asyncio import csv import yaml import sys from client import init_client from analyzer import analyze async def main(): with open(config.yaml, r, encodingutf-8) as f: config yaml.safe_load(f) init_client(config[model][base_url]) with open(config[io][input_file], r, encodingutf-8) as f: reader csv.DictReader(f) comments [row[comment] for row in reader] semaphore asyncio.Semaphore(config[analysis][concurrency]) async def worker(comment): async with semaphore: try: result await analyze(comment, config) return {comment: comment, **result} except Exception as e: return {comment: comment, error: str(e)} results await asyncio.gather(*[worker(c) for c in comments]) with open(config[io][output_file], w, encodingutf-8, newline) as f: writer csv.DictWriter(f, fieldnamesresults[0].keys()) writer.writeheader() writer.writerows(results) print(f完成 {len(results)} 条输出至 {config[io][output_file]}) if __name__ __main__: asyncio.run(main())6.4 上线前检查清单这个项目跑通了再往生产方向走之前检查三件事模型路径和服务状态如果用 Ollama确认ollama serve在跑并且ollama list里有对应的模型 tag。这是新手最容易忽略的——代码没问题服务没起给你报连接拒绝。小样本测试先用comments[:5]跑一遍确认输出字段和预期一致再放开全量。这一步省下的调试时间不可估量。错误处理批量任务里有一条数据出错不要整体中断。我见过很多项目因为一条脏数据卡住整个流程。上面代码里每一条都用 try/except 捕获了异常并标记 error 字段这就是正确的容错思路。我在实际项目里这套代码逻辑已经帮助好几个团队把“本地模型 demo”升级成了能跑的批处理工具。核心不在于代码多复杂而在于环境隔离、配置外置、并发可控、解析兜底这四个习惯。你把这四条带到任何 Python 大语言模型项目里都能少踩大半的坑。