AI大模型开发全栈实战：从本地部署到Agent工具链搭建

AI大模型开发全栈实战：从本地部署到Agent工具链搭建

前言

2026年，AI开发不再是算法工程师的专利。本文从零搭建AI开发环境，手把手带你完成大模型本地部署、MCP工具协议实战、RAG知识库搭建三大核心环节，全部代码可直接运行。

技术栈：Python 3.11 + PyTorch 2.x + Ollama + vLLM + LangChain + ChromaDB

一、2026年AI开发的三个核心问题

很多人入门AI开发时会遇到三个灵魂拷问：

本文就是围绕这三个问题，从零到一搭建一套完整的AI开发工具链。

二、环境搭建：Python + PyTorch + CUDA

2.1 基础环境

python-mvenv ai-dev-envsourceai-dev-env/bin/activate pipinstall--upgradepip

2.2 安装PyTorch

nvidia-smi pipinstalltorch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121

验证安装：

importtorchprint(f"PyTorch版本:{torch.__version__}")print(f"CUDA可用:{torch.cuda.is_available()}")print(f"GPU设备:{torch.cuda.get_device_name(0)iftorch.cuda.is_available()else'无'}")

2.3 安装AI开发核心依赖

pipinstalltransformers accelerate peft bitsandbytes pipinstalllangchain langchain-community chromadb pipinstallollama vllm sentence-transformers pipinstallfastapi uvicorn

三、大模型本地部署

3.1 使用Ollama快速部署

Ollama是最简单的本地大模型部署工具，一行命令即可启动。

curl-fsSLhttps://ollama.com/install.sh|shollama pull qwen2.5:7b ollama pull qwen2.5:14b ollama pull deepseek-r1:8b ollama run qwen2.5:7b"用Python写一个快速排序算法"

3.2 使用vLLM生产级部署

vLLM是高性能推理引擎，支持PagedAttention和连续批处理。

fromvllmimportLLM,SamplingParamsfromfastapiimportFastAPIfrompydanticimportBaseModelimportuvicorn app=FastAPI()llm=LLM(model="Qwen/Qwen2.5-7B-Instruct",tensor_parallel_size=1,gpu_memory_utilization=0.9,max_model_len=8192,)classGenerateRequest(BaseModel):prompt:strmax_tokens:int=512temperature:float=0.7@app.post("/generate")asyncdefgenerate(request:GenerateRequest):sampling_params=SamplingParams(temperature=request.temperature,max_tokens=request.max_tokens,)outputs=llm.generate([request.prompt],sampling_params)return{"text":outputs[0].outputs[0].text}if__name__=="__main__":uvicorn.run(app,host="0.0.0.0",port=8000)

3.3 使用LangChain统一调用

fromlangchain.llmsimportOllamafromlangchain.chat_modelsimportChatOpenAI local_llm=Ollama(model="qwen2.5:7b",temperature=0.7)cloud_llm=ChatOpenAI(model="gpt-4o",temperature=0)defask_llm(question:str,use_local:bool=True):llm=local_llmifuse_localelsecloud_llmreturnllm.invoke(question)

四、MCP工具协议实战

MCP（Model Context Protocol）是2026年大模型工具调用的统一标准。

4.1 创建MCP Server

importasynciofrommcp.serverimportServerfrommcp.server.stdioimportstdio_serverfrommcp.typesimportTool,TextContent server=Server("weather-service")@server.list_tools()asyncdeflist_tools()->list[Tool]:return[Tool(name="get_weather",description="获取指定城市的实时天气信息",inputSchema={"type":"object","properties":{"city":{"type":"string","description":"城市名称"}},"required":["city"],},),Tool(name="get_forecast",description="获取指定城市未来几天的天气预报",inputSchema={"type":"object","properties":{"city":{"type":"string"},"days":{"type":"integer","default":7},},"required":["city"],},),]@server.call_tool()asyncdefcall_tool(name:str,arguments:dict)->list[TextContent]:ifname=="get_weather":city=arguments["city"]weather=f"{city}天气：晴，温度25°C，湿度60%，风力3级"return[TextContent(type="text",text=weather)]elifname=="get_forecast":city=arguments["city"]days=arguments.get("days",7)forecast=f"{city}未来{days}天天气预报：\n周一：晴 22-30°C\n周二：多云 20-28°C"return[TextContent(type="text",text=forecast)]asyncdefmain():asyncwithstdio_server()as(read_stream,write_stream):awaitserver.run(read_stream,write_stream,server.create_initialization_options())if__name__=="__main__":asyncio.run(main())

4.2 创建MCP Client

importasynciofrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientasyncdefrun():server_params=StdioServerParameters(command="python",args=["weather_server.py"],)asyncwithstdio_client(server_params)as(read,write):asyncwithClientSession(read,write)assession:awaitsession.initialize()tools=awaitsession.list_tools()print(f"可用工具:{[t.namefortintools.tools]}")result=awaitsession.call_tool("get_weather",{"city":"北京"})print(f"结果:{result.content[0].text}")asyncio.run(run())

五、RAG知识库搭建

5.1 文档处理与向量化

fromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain.embeddingsimportHuggingFaceEmbeddingsfromlangchain.vectorstoresimportChromafromlangchain.document_loadersimportDirectoryLoader,TextLoader loader=DirectoryLoader("./documents/",glob="**/*.txt",loader_cls=TextLoader)documents=loader.load()print(f"加载了{len(documents)}个文档")text_splitter=RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200,separators=["\n\n","\n","。","！","？","，"," ",""],)chunks=text_splitter.split_documents(documents)print(f"分割为{len(chunks)}个文本块")embeddings=HuggingFaceEmbeddings(model_name="BAAI/bge-large-zh-v1.5",model_kwargs={"device":"cuda"},encode_kwargs={"normalize_embeddings":True},)vectorstore=Chroma.from_documents(documents=chunks,embedding=embeddings,persist_directory="./chroma_db",)vectorstore.persist()print("向量数据库已保存")

5.2 构建RAG问答链

fromlangchain.chainsimportRetrievalQAfromlangchain.promptsimportPromptTemplate prompt_template="""你是一个专业的知识问答助手。请基于以下参考资料回答问题。 如果参考资料中没有相关信息，请明确说"根据现有资料，我无法回答这个问题"。 参考资料： {context} 问题：{question} 答案："""PROMPT=PromptTemplate(template=prompt_template,input_variables=["context","question"])qa_chain=RetrievalQA.from_chain_type(llm=local_llm,chain_type="stuff",retriever=vectorstore.as_retriever(search_kwargs={"k":5}),chain_type_kwargs={"prompt":PROMPT},return_source_documents=True,)result=qa_chain({"query":"公司的年假政策是什么？"})print(f"答案:{result['result']}")

5.3 高级RAG：查询重写与混合检索

classAdvancedRAG:def__init__(self,vectorstore,llm):self.vectorstore=vectorstore self.llm=llmdefrewrite_query(self,query:str)->str:prompt=f"将以下问题重写为更适合文档检索的关键词组合：\n\n原始问题：{query}\n重写结果："returnself.llm.invoke(prompt)defhybrid_search(self,query:str,k:int=5):vector_results=self.vectorstore.similarity_search_with_score(query,k=k)fromrank_bm25importBM25Okapiimportjieba all_docs=self.vectorstore.get()["documents"]tokenized_docs=[list(jieba.cut(doc))fordocinall_docs]bm25=BM25Okapi(tokenized_docs)tokenized_query=list(jieba.cut(query))bm25_scores=bm25.get_scores(tokenized_query)returnself.reciprocal_rank_fusion(vector_results,bm25_scores,k)defreciprocal_rank_fusion(self,vector_results,bm25_scores,k):rrf_scores={}forrank,(doc,score)inenumerate(vector_results):rrf_scores[doc.page_content]=rrf_scores.get(doc.page_content,0)+1/(rank+60)foridx,scoreinenumerate(bm25_scores):doc_content=all_docs[idx]rrf_scores[doc_content]=rrf_scores.get(doc_content,0)+score sorted_docs=sorted(rrf_scores.items(),key=lambdax:x[1],reverse=True)returnsorted_docs[:k]

六、构建完整的AI应用

6.1 FastAPI Web服务

fromfastapiimportFastAPI,HTTPExceptionfromfastapi.middleware.corsimportCORSMiddlewarefrompydanticimportBaseModelfromtypingimportList,Optionalimportuvicorn app=FastAPI(title="AI知识库问答系统")app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_methods=["*"],allow_headers=["*"])rag=AdvancedRAG(vectorstore,local_llm)classQuestionRequest(BaseModel):question:struse_rewrite:bool=Truetop_k:int=5classQuestionResponse(BaseModel):answer:strsources:List[dict]rewritten_query:Optional[str]=None@app.post("/ask",response_model=QuestionResponse)asyncdefask(request:QuestionRequest):try:query=request.question rewritten=Noneifrequest.use_rewrite:rewritten=rag.rewrite_query(query)query=rewritten results=rag.hybrid_search(query,k=request.top_k)context="\n\n".join([docfordoc,_inresults])answer=local_llm.invoke(f"基于以下参考资料回答问题：\n\n{context}\n\n问题：{request.question}\n\n答案：")returnQuestionResponse(answer=answer,sources=[{"content":doc[:200],"score":score}fordoc,scoreinresults],rewritten_query=rewritten,)exceptExceptionase:raiseHTTPException(status_code=500,detail=str(e))@app.get("/health")asyncdefhealth():return{"status":"ok","model":"qwen2.5:7b"}if__name__=="__main__":uvicorn.run(app,host="0.0.0.0",port=8000)

6.2 Docker容器化部署

FROM python:3.11-slim WORKDIR /app RUN apt-get update && apt-get install -y build-essential curl && rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["python", "main.py"]

version:'3.8'services:ai-qa:build:.ports:-"8000:8000"volumes:-./documents:/app/documents-./chroma_db:/app/chroma_dbenvironment:-CUDA_VISIBLE_DEVICES=0deploy:resources:reservations:devices:-driver:nvidiacount:1capabilities:[gpu]

七、性能优化与最佳实践

7.1 推理加速

fromtransformersimportBitsAndBytesConfig quantization_config=BitsAndBytesConfig(load_in_4bit=True,bnb_4bit_compute_dtype=torch.bfloat16,bnb_4bit_use_double_quant=True,bnb_4bit_quant_type="nf4",)model=AutoModelForCausalLM.from_pretrained(model_name,quantization_config=quantization_config,device_map="auto",)

7.2 缓存策略

importhashlibclassCachedRAG:def__init__(self):self.cache={}def_get_cache_key(self,query:str)->str:returnhashlib.md5(query.encode()).hexdigest()defquery(self,question:str)->dict:cache_key=self._get_cache_key(question)ifcache_keyinself.cache:returnself.cache[cache_key]result=self._do_query(question)self.cache[cache_key]=resultreturnresult

7.3 并发处理

importasynciofromconcurrent.futuresimportThreadPoolExecutorclassConcurrentRAG:def__init__(self,max_workers:int=4):self.executor=ThreadPoolExecutor(max_workers=max_workers)asyncdefbatch_query(self,questions:List[str])->List[dict]:loop=asyncio.get_event_loop()tasks=[loop.run_in_executor(self.executor,self._query_single,q)forqinquestions]returnawaitasyncio.gather(*tasks)def_query_single(self,question:str)->dict:returnrag.query(question)

八、总结

本文从零搭建了一套完整的AI开发工具链，涵盖：

1. 环境搭建：Python + PyTorch + CUDA完整配置
2. 模型部署：Ollama快速启动 + vLLM生产级部署
3. 工具集成：MCP协议统一工具调用
4. 知识库：RAG检索增强生成完整实现
5. Web服务：FastAPI构建API + Docker容器化部署
6. 性能优化：量化推理、缓存策略、并发处理

掌握这些技能后，你将能够独立完成从模型部署到应用上线的全流程开发。2026年的AI开发，核心不在于"会用API"，而在于"能构建可靠的AI系统"。希望本文能帮助你迈出这关键的一步。