写在前面 在大学的时候,我一直对机器学习、人工智能之类的课程不怎么喜欢。一部分原因是授课老师讲的过于无聊,更重要的原因是ai存在明显的不确定性。当执行一段C语言代码时,再怎么样程序的功能都是可被理解的,你对于一行代码执行后会有什么样的表现拥有基本的预期,尽管存在多线程、内核抢占、编译优化、CPU乱序发射,底层硬件故障等一些不确定性,但你对它们的存在也是了解的,至少可以假装认为自己了解。但ai的执行结果就完全是个谜了,机器学习中的一系列参数的含义完全未知,为什么这个参数比较好,那个参数比较差,本身就很难彻底说明白,因此当时我们都叫搞ai的叫太白金星,整天炼丹调参。
之前的程序都是从原理上,从逻辑上,从结构上仔细地拆分,解决问题;但ai看起来是直接从结果上来解决问题。从一个强大的模型、神经网络开始,尝试解决这个问题并计算结果的准确程度,然后一直往更加准确的方向前进。但这种方式的问题是:即使ai对一个问题的准确度一直是100%,也无法确保下一次同类型问题同样正确,虽然非ai方法也不一定真的能保证完全正确,但由于对“底层原理”的偏执, 我还是更加喜欢非ai的范式,猜字谜”的行为模式本身并不能说服我。
但在2026年的这个时刻,不得不承认的事实就是:ai一定会改变世界!之前使用chatgpt、qwen等一些对话模型时,虽然能感觉到ai很强大,但没有那种惊叹的感觉(我好像非常喜欢用感觉这两个字)。直到我使用cursor生成前后端页面,我本来只是随便写一个简短的提示词试一试,后续再研究研究如何修改,毕竟在大学的时候我还是用过一段时间的vue,没想到它生成代码后就能执行运行,而且结果完全符合我的预期,这个时刻我就知道,我过几年就要被ai取代掉了,至少过几年后,我就完全不需要自己写代码了,是不是所有人到头来都是提示词工程师?最近的几个月, 我已经不怎么自己写代码了, 都是让ai生成后直接用(实际上没怎么review),这也带来很大的不确定性,但目前看起来没出什么问题,所以整个系统就是建立在不确定性之上的? 我从之前的如何写好一段代码变成了如何写好一段提示词,希望我后面能学会如何review一段ai代码,完全放飞也不太好(#^.^#)
计算机领域好像只有ai可以称得上改变世界,如果说“SSD改变了世界!存储改变了世界!Linux改变了世界!”都有点违和,但ai
与一个伟大的技术挂钩
与改变世界的技术挂钩
黑客
伦理
与人类的区别
冯洛伊曼架构
绝对正确的无聊
ai味
推荐阅读/观看
莱姆狂想曲—泥人十四
时空奇旅
机器人之梦(封面)
ps:使用小写的ai感觉更加亲切,随和,大写的AI就更加刻板,严肃,官方。一个小发现~
初见 RAG、提示词工程、上下文工程、token、mcp、fuction call、skills、ai agent、agi、训练与推理、上下文压缩与遗忘、kv cache等一些花里胡哨的词汇
第一步:购买云服务器部署一个模型进行一次对话
购买 ecs.gn6v-c8g1.4xlarge实例,选择Alibaba Cloud Linux 3.2104 LTS 64位 预装NVIDIA GPU驱动和 CUDA,配置密钥对后通过vscode远程登录
安装依赖包
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 yum install python3.11 sudo update-alternatives --config python [root@iZbp1bzvlb6th4j2vwt9taZ ~] Python 3.11.13 python -m venv qwen-envsource qwen-env/bin/activate pip install --upgrade pip pip install torch torchvision torchaudio \ --index-url https://mirrors.aliyun.com/pytorch-wheels/cu121/ \ --extra-index-url https://mirrors.aliyun.com/pypi/simple/ \ --trusted-host mirrors.aliyun.com pip install transformers accelerate bitsandbytes sentencepiece \ --index-url https://mirrors.aliyun.com/pypi/simple/ \ --trusted-host mirrors.aliyun.com
运行demo
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 "" " Qwen2.5-7B-Instruct 4-bit 量化推理示例(完整注释版) 功能:加载通义千问 7B 指令微调模型,进行一次对话生成 适用环境:Linux + CUDA + 双 Tesla V100(16GB×2) " "" from transformers import ( AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig ) import torch model_id = "Qwen/Qwen2.5-7B-Instruct" quantization_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16 ) tokenizer = AutoTokenizer.from_pretrained( model_id, trust_remote_code=True ) model = AutoModelForCausalLM.from_pretrained( model_id, quantization_config=quantization_config, device_map="auto" , trust_remote_code=True ) messages = [ {"role" : "system" , "content" : "You are a helpful assistant." }, {"role" : "user" , "content" : "你好!请介绍一下你自己。" } ] text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = tokenizer(text, return_tensors="pt" ) inputs = inputs.to(model.device) outputs = model.generate( **inputs, max_new_tokens=512, do_sample=True, temperature=0.7, top_p=0.9 ) response_ids = outputs[0][inputs.input_ids.shape[-1]:] response_text = tokenizer.decode(response_ids, skip_special_tokens=True)print (response_text)
第一次运行需要下载模型
1 2 3 4 5 6 7 8 pip install hf-transferexport HF_ENDPOINT=https://hf-mirror.comexport HF_HUB_ENABLE_HF_TRANSFER=1 python hello.py
模型下载太慢了,改成ModelScope
1 pip install modelscope[vision,nlp]
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 """ Qwen2.5-7B-Instruct 4-bit 量化推理示例 (ModelScope 魔搭版) 优势:下载速度快,兼容性好,无需担心远程代码安全警告 """ from modelscope import ( AutoTokenizer, AutoModelForCausalLM, )import torchfrom transformers import BitsAndBytesConfig model_id = "qwen/Qwen2.5-7B-Instruct" quantization_config = BitsAndBytesConfig( load_in_4bit=True , bnb_4bit_compute_dtype=torch.bfloat16 ) tokenizer = AutoTokenizer.from_pretrained( model_id, ) model = AutoModelForCausalLM.from_pretrained( model_id, quantization_config=quantization_config, device_map="auto" , ) messages = [ {"role" : "system" , "content" : "You are a helpful assistant." }, {"role" : "user" , "content" : "你好!请介绍一下你自己。" } ] text = tokenizer.apply_chat_template( messages, tokenize=False , add_generation_prompt=True ) inputs = tokenizer(text, return_tensors="pt" ).to(model.device) outputs = model.generate( **inputs, max_new_tokens=512 , do_sample=True , temperature=0.7 , top_p=0.9 ) response_ids = outputs[0 ][inputs.input_ids.shape[-1 ]:] response_text = tokenizer.decode(response_ids, skip_special_tokens=True )print (response_text)
第二步:运行一系列demo,了解一些相关概念 使用ai来学ai~
fastapi 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom transformers import AutoModelForCausalLM, AutoTokenizerimport torchimport gcfrom transformers import BitsAndBytesConfig app = FastAPI(title="Qwen2.5 FastAPI 服务" ) CONVERSATION_HISTORY = {} LOCAL_MODEL_PATH = "/root/.cache/modelscope/hub/models/qwen/Qwen2.5-7B-Instruct" MODEL_PATH = LOCAL_MODEL_PATH print ("正在加载分词器..." ) tokenizer = AutoTokenizer.from_pretrained( MODEL_PATH, trust_remote_code=True , local_files_only=True )print ("正在加载模型..." ) bnb_config = BitsAndBytesConfig( load_in_4bit=True , bnb_4bit_quant_type="nf4" , bnb_4bit_compute_dtype=torch.float16, ) model = AutoModelForCausalLM.from_pretrained( MODEL_PATH, quantization_config=bnb_config, device_map="auto" , trust_remote_code=True , )print ("模型加载完成!" )class ChatRequest (BaseModel ): session_id: str message: str class ChatResponse (BaseModel ): session_id: str response: str def generate_response (session_id: str , user_message: str ) -> str : if session_id not in CONVERSATION_HISTORY: CONVERSATION_HISTORY[session_id] = [] CONVERSATION_HISTORY[session_id].append({"role" : "user" , "content" : user_message}) messages = CONVERSATION_HISTORY[session_id] try : input_ids = tokenizer.apply_chat_template( messages, tokenize=True , return_tensors="pt" , add_generation_prompt=True ).to(model.device) terminators = [ tokenizer.eos_token_id, tokenizer.convert_tokens_to_ids("<|eot_id|>" ) ] eos_token_id = tokenizer.eos_token_id eot_token_id = tokenizer.convert_tokens_to_ids("<|eot_id|>" ) valid_eos_tokens = [eos_token_id] if eot_token_id is not None : valid_eos_tokens.append(eot_token_id) attention_mask = input_ids.ne(tokenizer.pad_token_id).long() outputs = model.generate( input_ids=input_ids, attention_mask=attention_mask, max_new_tokens=512 , do_sample=True , temperature=0.7 , top_p=0.9 , pad_token_id=tokenizer.eos_token_id, ) generated_ids = outputs[0 ][input_ids.shape[-1 ]:] response_text = tokenizer.decode(generated_ids, skip_special_tokens=True ) CONVERSATION_HISTORY[session_id].append({"role" : "assistant" , "content" : response_text}) return response_text except Exception as e: torch.cuda.empty_cache() gc.collect() raise HTTPException(status_code=500 , detail=str (e))@app.post("/v1/chat/completions" , response_model=ChatResponse ) async def chat_completions (request: ChatRequest ): try : response_text = generate_response(request.session_id, request.message) return ChatResponse(session_id=request.session_id, response=response_text) except Exception as e: import traceback print ("!!! 服务器内部错误详情:" ) traceback.print_exc() raise HTTPException(status_code=500 , detail=str (e))@app.get("/health" ) async def health_check (): return {"status" : "healthy" , "model" : "Qwen2.5-7B-Instruct" }
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 import requestsimport json BASE_URL = "http://localhost:8000" SESSION_ID = "user_123" def chat (): print ("欢迎使用 Qwen 聊天室!输入 'quit' 退出。" ) print ("-" * 50 ) while True : user_input = input ("\n👤 你: " ).strip() if user_input.lower() in ['quit' , 'exit' , 'bye' , '退出' ]: print ("👋 再见!" ) break if not user_input: continue try : response = requests.post( f"{BASE_URL} /v1/chat/completions" , json={ "session_id" : SESSION_ID, "message" : user_input } ) if response.status_code == 200 : bot_reply = response.json().get("response" , "(无响应)" ) print (f"\n🤖 Qwen: {bot_reply} " ) else : print (f" 请求失败: {response.status_code} " ) print (response.text) except requests.exceptions.ConnectionError: print (" 连接错误:请确保服务器 (uvicorn) 已经启动。" ) break except KeyboardInterrupt: break if __name__ == "__main__" : chat()
RAG
传统的问答模式是让大模型“凭记忆答题”,而 RAG 则是让大模型“开卷考试”。
**先检索 (Retrieval)**:当用户提出一个问题时,系统首先在外部知识库(如你的文档、数据库)中搜索与问题最相关的信息片段。
**再增强 (Augmented)**:将检索到的“参考资料”与用户的原始问题拼接在一起,形成一份包含“最新、特定外部知识”的增强提示(Prompt)。
**后生成 (Generation)**:将这个增强后的提示输入给大语言模型(LLM)。LLM 基于这些具体的参考资料来生成最终答案,而不是依赖其内部过时的参数知识。
通过这种方式,RAG 成功解决了大模型的两个主要痛点:
知识时效性差 :无需重新训练模型,只需更新知识库即可让模型掌握最新信息。
事实性错误(幻觉) :让模型的回答有据可依,大幅降低其“一本正经胡说八道”的概率。
⚙️ RAG 的具体实现流程
实现一个 RAG 系统主要分为两个阶段:离线构建知识库 (数据准备)和 在线问答 (推理服务)。
第一阶段:离线构建知识库
这是 RAG 的“备考”阶段,目的是将你的文档资料处理成计算机可以高效检索的格式。
加载文档 :从各种来源(如 TXT, PDF, Word, 数据库)读取原始文本数据。
文本分块 :将长文档切分成小的文本片段(Chunk)。这是为了适应大模型的上下文窗口,并保证检索的精度。例如,将一本长篇小说按章节或固定长度切分成几百个段落。
向量化存储 :使用嵌入模型(Embedding Model)将每个文本块转换成一串数字(向量),并存入向量数据库。这一步相当于将文字信息“翻译”成计算机可以理解的数学语言,并建立索引。
第二阶段:在线问答
这是 RAG 的“考试”阶段,当用户提问时,系统实时执行以下步骤。
检索 :用户输入问题后,系统使用相同的嵌入模型将问题也转换成向量。然后在向量数据库中进行相似度搜索(如余弦相似度),找出与问题最相关的前 K 个文本块(例如最相关的 3 个段落)。
构建 Prompt :将检索到的最相关文本块作为上下文,与用户的原始问题拼接在一起,形成一个新的 Prompt。
生成答案 :将这个包含上下文的 Prompt 发送给大语言模型(LLM)。模型阅读上下文后,生成最终的、准确的回答。
1 2 3 pip install langchain langchain_community langchain-core sentence-transformers faiss-cpu pip install langchain-huggingface
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 import osfrom langchain_community.document_loaders import TextLoaderfrom langchain_text_splitters import CharacterTextSplitterfrom langchain_huggingface import HuggingFaceEmbeddingsfrom langchain_community.vectorstores import FAISSfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_core.output_parsers import StrOutputParserimport requestsimport json DOC_PATH = "podcast.txt" LOCAL_MODEL_PATH = "/root/.cache/modelscope/hub/models/AI-ModelScope/bge-small-en-v1___5" QWEN_SERVER_URL = "http://localhost:8000/v1/chat/completions" class RAGChain : def __init__ (self ): print ("正在加载文档..." ) loader = TextLoader(DOC_PATH, encoding='utf-8' ) documents = loader.load() print ("正在切分文本..." ) text_splitter = CharacterTextSplitter(chunk_size=500 , chunk_overlap=50 ) splits = text_splitter.split_documents(documents) print ("正在生成向量库..." ) embedding_model = HuggingFaceEmbeddings(model_name=LOCAL_MODEL_PATH) self.vectorstore = FAISS.from_documents(splits, embedding_model) self.retriever = self.vectorstore.as_retriever(search_kwargs={"k" : 2 }) print ("RAG 系统初始化完成!" ) def call_qwen_server (self, prompt ): """调用本地 Qwen 服务""" try : response = requests.post( QWEN_SERVER_URL, json={ "session_id" : "rag_user" , "message" : prompt } ) if response.status_code == 200 : return response.json()["response" ] else : return f"API Error: {response.status_code} " except Exception as e: return f"Connection Error: {e} " def invoke (self, question: str ): retrieved_docs = self.retriever.invoke(question) context = "\n" .join([doc.page_content for doc in retrieved_docs]) final_prompt = f"""你是一个问答助手。请根据以下检索到的信息回答问题。 检索到的信息: {context} 问题:{question} 请用中文回答:""" result = self.call_qwen_server(final_prompt) return resultif __name__ == "__main__" : rag = RAGChain() print ("\n" + "=" *50 ) print ("RAG Demo 已启动!输入 'quit' 退出。" ) print ("=" *50 ) while True : query = input ("\n❓ 你的问题: " ).strip() if query.lower() in ['quit' , 'exit' ]: break if not query: continue answer = rag.invoke(query) print (f"\n✅ 最终答案:\n{answer} " )
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45