🌞欢迎来到人工智能的世界
🌈博客主页：卿云阁💌欢迎关注🎉点赞👍收藏⭐️留言📝
📆首发时间：🌹2026年5月30日🌹
✉️希望可以和大家一起完成进阶之路！
🙏作者水平很有限，如果发现错误，请留言轰炸哦！万分感谢！
目录
医疗问答系统的RAG增强
普通的调用本地大模型的例子
数据集
ChromaDB
RAG实现公司HR制度智能问答系统
读取 Word 文档并切块
准备 Embedding 函数
带缓存的 Embedding 封装
建立 Chroma 向量数据库
医疗问答系统的RAG增强
普通的调用本地大模型的例子
# -*- coding: utf-8 -*- """ 使用 ModelScope 加载 Qwen3-VL-2B-Instruct 多模态模型,进行「纯文本」对话推理的完整示例。 整体流程: 加载模型 / 处理器 -> 构造对话消息 -> 套用对话模板并分词 -> 模型生成 -> 截取并解码新生成的文本 注意:Qwen3-VL 本身是「视觉-语言」多模态模型,但同样可以只用来做纯文本对话(本例就是)。 """ # ===================== 导入依赖 ===================== # 这里没有从 transformers 导入,而是从 modelscope(魔搭社区)导入同名的类。 # modelscope 提供了与 HuggingFace transformers 高度兼容的接口,区别在于: # 调用 from_pretrained 时,它会自动从「魔搭社区」下载模型权重(国内访问更快、更稳定), # 而不是从 HuggingFace Hub 下载。 # - Qwen3VLForConditionalGeneration:Qwen3-VL 系列(视觉-语言多模态模型)的生成式模型类。 # "ForConditionalGeneration" 表示它用于「条件生成」,即根据输入生成输出文本。 # - AutoProcessor:自动处理器,负责把文本(以及图像 / 视频)预处理成模型能接受的张量。 # 它内部通常封装了一个 tokenizer(分词器)和一个 image processor(图像处理器)。 from modelscope import Qwen3VLForConditionalGeneration, AutoProcessor # ===================== 加载模型 ===================== # 从预训练权重加载模型。"Qwen/Qwen3-VL-2B-Instruct" 是模型在魔搭社区上的仓库名: # 2B —— 约 20 亿参数; # Instruct —— 经过指令微调、可直接拿来对话的版本(相对的是只做续写的 Base 版)。 model = Qwen3VLForConditionalGeneration.from_pretrained( "Qwen/Qwen3-VL-2B-Instruct", dtype="auto", # 数据精度自动选择:让框架根据模型配置和硬件自动挑选合适精度 #（如 bfloat16 / float16），以节省显存、加快推理。 # 注意:较老版本的 transformers 中,这个参数名叫 torch_dtype。 device_map="auto" # 设备自动分配:自动把模型各层放到可用设备上(优先 GPU); # 显存不够时还能把部分层放到 CPU 或多张卡上(需要安装 accelerate 库)。 ) # 加载与模型配套的处理器。务必使用与模型相同的仓库名, # 这样分词方式、特殊 token、对话模板才能和模型训练时保持一致,否则结果会乱。 processor = AutoProcessor.from_pretrained("Qwen/Qwen3-VL-2B-Instruct") # ===================== 纯文本对话 ===================== # 构造对话消息。由于 Qwen-VL 是多模态模型,即使只发纯文本, # content 也采用「列表 + 字典」这种多模态格式(每个元素用 type 声明自己的类型)。 # 这样做的好处是格式统一:将来要加图片时,只需在 content 列表里再追加一个 # {"type": "image", "image": "图片路径或URL"} 即可,无需改动整体结构。 messages = [ { "role": "user", # 角色:user 表示这是用户发出的消息 #（其他常见角色:system 系统设定、assistant 模型回复) "content": [ # content 是一个列表,可同时包含多个不同类型的内容块 { "type": "text", # 声明这一块的类型为文本 "text": "请介绍一下什么是RAG，并举一个医疗问答系统中的例子。" # 实际的提问内容 } ], } ] # 套用对话模板,把上面结构化的 messages 转换成模型真正需要的输入。 # 这一步会:① 按 Qwen 的对话格式把消息拼成带特殊标记的字符串; # ② 将字符串分词成 token id;③ 打包成张量。 inputs = processor.apply_chat_template( messages, tokenize=True, # True 表示直接分词(返回 token id),而非只返回拼接好的字符串 add_generation_prompt=True, # 在末尾追加「轮到助手回答」的提示标记,引导模型开始生成回复; # 若不加,模型可能不知道该接着说话 return_dict=True, # 返回字典(含 input_ids、attention_mask 等), # 方便后面用 ** 解包传给 generate return_tensors="pt" # 返回 PyTorch 张量("pt" = PyTorch) ) # 把输入张量搬到与模型相同的设备上(例如 GPU)。 # 如果模型在 GPU 而数据在 CPU(或反之),前向计算时会因设备不一致而报错。 inputs = inputs.to(model.device) # 调用 generate 进行文本生成:模型会自回归地一个 token 接一个 token 地输出。 generated_ids = model.generate( **inputs, # 解包前面的字典:把 input_ids、attention_mask 等作为关键字参数传入 max_new_tokens=512, # 最多「新生成」512 个 token(不含输入部分),用来限制回答长度上限 temperature=0.7, # 温度,控制随机性:值越高输出越发散有创意,越低越确定保守; # 0.7 是较常用的折中值 do_sample=True # True 表示采样生成(此时 temperature 才会生效); # 若为 False 则是贪心 / 确定性解码,temperature 将不起作用 ) # generate 返回的 generated_ids 里,同时包含了「原始输入的 token」和「新生成的 token」。 # 下面这一步把每条序列开头的「输入部分」切掉,只保留模型新生成的内容。 generated_ids_trimmed = [ out_ids[len(in_ids):] # 以输入长度 len(in_ids) 作为切片起点,取其后(即新生成)的部分 for in_ids, out_ids in zip(inputs.input_ids, generated_ids) # 逐条配对:输入序列 与 对应输出序列 ] # 把 token id 解码回人类可读的文本。batch_decode 可一次处理一批(这里 batch 大小为 1)。 output_text = processor.batch_decode( generated_ids_trimmed, skip_special_tokens=True, # 跳过 <|im_end|> 之类的特殊标记,只保留正文 clean_up_tokenization_spaces=False # 不额外清理分词产生的空格(中文场景通常设 False,避免误删) ) # batch_decode 返回的是一个列表(每条输入对应一个字符串),取第 0 条打印出来。 print(output_text[0])数据集
train_zh.json 是JSONL 格式，也就是“每一行都是一个独立的 JSON 对象”。
里面每条数据大概长这样：
{ "instruction": "牙齿黄，怎么办，想做漂白，但是不知道有哪些危害。还是什么牙齿贴面", "input": "", "output": "牙齿黄的主要原因是牙齿表面的牙釉质变薄..." }三个字段的含义：
instruction：用户问题 / 指令 input：补充输入，很多数据里是空字符串 output：参考答案 / 医生回答ChromaDB
专门给 RAG 用的“向量数据库”。
普通数据库存的是文字、数字、表格，比如：
id: 1 question: 牙齿黄怎么办？ answer: 建议注意口腔卫生，可以咨询牙医...而 ChromaDB 主要存的是这种东西：
id: 1 document: 牙齿黄怎么办？ embedding: [0.12, -0.03, 0.88, ...] metadata: {"answer": "..."}它的作用是： 存储向量 + 根据相似度检索资料
ChromaDB 在 RAG 里的位置分成了两条线：
资料入库：资料 -> 文本片段 -> embedding -> 向量 -> 存进 ChromaDB
用户提问：问题 -> embedding -> ChromaDB 检索 -> 找资料 -> 拼 Prompt -> 大模型回答
读取 train_zh.json
def load_data(path=DATA_PATH, limit=50000): questions, answers = [], [] with open(path, "r", encoding="utf-8") as f: for i, line in enumerate(f): if limit and i >= limit: break item = json.loads(line) questions.append(item["instruction"]) answers.append(item["output"]) return questions, answers questions, answers = load_data(limit=50000) len(questions), questions[0], answers[0][:80]原始数据格式
{"instruction": "牙齿黄怎么办？", "input": "", "output": "建议去看牙医..."}整理成两个列表的格式
questions = [ "这段时间去上厕所本来想小便的可是每次都会拉大便", "医生呀！我刚被查出得了白癜风...", ... ] answers = [ "这可能是因为你的饮食习惯或者消化系统的问题导致的...", "白癜风的治疗费用因个体差异...", ... ]定义 embedding 函数
def embed_texts(texts, batch_size=10): embeddings = [] for start in range(0, len(texts), batch_size): batch = texts[start:start + batch_size] response = embedding_client.embeddings.create( model=EMBED_MODEL, input=batch, ) embeddings.extend([item.embedding for item in response.data]) return embeddings输入：
embed_texts(["牙齿黄怎么办？", "前列腺肥大怎么办？"])返回大概是：
[ [0.012, -0.034, 0.128, ...], [0.044, 0.081, -0.023, ...] ]建立 ChromaDB 向量库
client = chromadb.PersistentClient(path=CHROMA_DIR) collection = client.get_or_create_collection(COLLECTION_NAME) if collection.count() == 0: batch_size = 1000 for start in range(0, len(questions), batch_size): end = start + batch_size batch_questions = questions[start:end] batch_answers = answers[start:end] collection.add( ids=[str(i) for i in range(start, min(end, len(questions)))], documents=batch_questions, metadatas=[{"answer": answer} for answer in batch_answers], embeddings=embed_texts(batch_questions), ) collection.count()这里的向量实际上是问题的向量
{ "ids": ["0", "1", "2"], "documents": [ "牙齿黄怎么办？", "前列腺肥大怎么办？", "白癜风初期治疗多少钱？" ], "metadatas": [ {"answer": "牙齿发黄可能和饮食有关..."}, {"answer": "前列腺肥大建议就医..."}, {"answer": "白癜风费用因方案不同..."} ], "embeddings": None }检索相似病例
def retrieve(query, top_k=3): result = collection.query( query_embeddings=embed_texts([query]), n_results=top_k, ) contexts = [] for question, metadata, distance in zip( result["documents"][0], result["metadatas"][0], result["distances"][0], ): contexts.append({ "score": 1 / (1 + float(distance)), "question": question, "answer": metadata["answer"], }) return contexts query = "牙齿黄怎么办？" contexts = retrieve(query, top_k=3) for i, item in enumerate(contexts, 1): print(f"{i}. score={item['score']:.3f} | {item['question']}")假设
query = "牙齿黄怎么办？" top_k = 3ChromaDB 返回的 result 是什么
{ "ids": [["265", "20231", "433043"]], "documents": [[ "牙齿黄，怎么办，想做漂白，但是不知道有哪些危害。还是什么牙齿贴面", "牙齿发黄怎么办，美白可以吗，要怎么变白呢", "牙齿黄，牙龈肿..." ]], "metadatas": [[ {"answer": "牙齿黄的主要原因是..."}, {"answer": "牙齿发黄可能是由于..."}, {"answer": "牙齿黄和牙龈肿可能与..."} ]], "distances": [[0.12, 0.18, 0.25]] }最终 contexts 大概是：
[ { "score": 0.89, "question": "牙齿黄，怎么办，想做漂白...", "answer": "牙齿黄的主要原因是..." }, { "score": 0.84, "question": "牙齿发黄怎么办，美白可以吗...", "answer": "牙齿发黄可能是由于..." }, { "score": 0.80, "question": "牙齿黄，牙龈肿...", "answer": "牙齿黄和牙龈肿可能与..." } ]构造 RAG Prompt
def build_prompt(query, contexts): context_text = "\n\n".join( f"相似问题：{item['question']}\n参考回答：{item['answer']}" for item in contexts ) return f"""你是一个医疗问答助手，请严格根据参考资料回答问题。 如果参考资料不足，请回答“根据已有资料无法确定，建议咨询医生”。 不要编造诊断，不要替代医生给出最终医疗结论。 参考资料： {context_text} 用户问题： {query} 请用中文回答：""" prompt = build_prompt(query, contexts) print(prompt)最简单的 RAG 回答
先不调用大模型，直接返回最相似病例的参考答案。
answer = contexts[0]["answer"] print(answer)contexts = [ { "score": 0.92, "question": "牙齿发黄怎么办，美白可以吗？", "answer": "牙齿发黄可能和饮食、吸烟、喝茶有关..." }, { "score": 0.88, "question": "牙齿黄，想做漂白，有什么危害？", "answer": "漂白可能造成牙齿敏感..." } ]可选：调用 Qwen3-VL 纯文本生成
如果你的服务器已经安装模型依赖，并且模型能正常下载/加载，可以运行下面的 cell。
def qwen_generate(prompt, model_name="Qwen/Qwen3-VL-2B-Instruct"): import torch from modelscope import AutoProcessor, Qwen3VLForConditionalGeneration model = Qwen3VLForConditionalGeneration.from_pretrained( model_name, dtype="auto", device_map="auto" ) processor = AutoProcessor.from_pretrained(model_name) messages = [ { "role": "user", "content": [{"type": "text", "text": prompt}], } ] inputs = processor.apply_chat_template( messages, tokenize=True, add_generation_prompt=True, return_dict=True, return_tensors="pt", ).to(model.device) with torch.no_grad(): output_ids = model.generate(**inputs, max_new_tokens=512) new_ids = [ out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, output_ids) ] return processor.batch_decode( new_ids, skip_special_tokens=True, clean_up_tokenization_spaces=False, )[0] qwen_answer = qwen_generate(prompt) print(qwen_answer)RAG实现公司HR制度智能问答系统
读取 Word 文档并切块
def extract_docx_by_paragraph(filename: Path, min_line_length: int = 1) -> List[str]: doc = Document(filename) paragraphs = [] for para in doc.paragraphs: text = para.text.strip() if len(text) >= min_line_length: paragraphs.append(text) # Word 表格中的制度内容也需要读出来 for table in doc.tables: for row in table.rows: cells = [cell.text.strip() for cell in row.cells if cell.text.strip()] if cells: paragraphs.append(" | ".join(cells)) return paragraphs def extract_docx_with_overlap( filename: Path, chunk_size: int = 500, overlap: int = 100, min_line_length: int = 10 ) -> List[str]: paragraphs = extract_docx_by_paragraph(filename, min_line_length=min_line_length) full_text = "\n".join(paragraphs) if chunk_size <= overlap: raise ValueError("chunk_size 必须大于 overlap") chunks = [] step = chunk_size - overlap for start in range(0, len(full_text), step): chunk = full_text[start:start + chunk_size].strip() if len(chunk) >= min_line_length: chunks.append(chunk) return chunks chunks = extract_docx_with_overlap(DOCX_PATH, chunk_size=500, overlap=100, min_line_length=10) print(f"切分得到 {len(chunks)} 个文本块") print(chunks[0][:500])准备 Embedding 函数
def raw_embed(texts: List[str]) -> List[List[float]]: response = embedding_client.embeddings.create( input=texts, model="text-embedding-v3" ) return [item.embedding for item in response.data]输入：
raw_embed([ "请假流程是什么？", "离职需要提前多久申请？" ])输出：
[ [0.0123, -0.0456, 0.0789, ...], [0.0345, 0.0112, -0.0891, ...] ]带缓存的 Embedding 封装
def get_embeddings(texts: List[str], batch_size: int = 10, cache_dir: Path = CACHE_DIR) -> List[List[float]]: cache_dir.mkdir(exist_ok=True) text_hash = hashlib.md5("\n".join(texts).encode("utf-8")).hexdigest() backend = EMBEDDING_BACKEND cache_file = cache_dir / f"{text_hash}_{backend}.pkl" if cache_file.exists(): print(f"从缓存读取向量：{cache_file}") with cache_file.open("rb") as f: return pickle.load(f) all_embeddings = [] for start in range(0, len(texts), batch_size): batch = texts[start:start + batch_size] all_embeddings.extend(raw_embed(batch)) print(f"已处理 {min(start + batch_size, len(texts))}/{len(texts)}") with cache_file.open("wb") as f: pickle.dump(all_embeddings, f) return all_embeddings
