LangChain实战：从零构建RAG应用与Agent智能体开发指南

这次我们来看一个完整的 LangChain 实战教程从零基础入门到构建 RAG 应用、部署 Ollama 本地大模型再到开发 Agent 智能体。这个教程的重点不是概念有多复杂而是能不能在普通开发环境下快速跑通全流程。如果你关心本地部署、显存占用、批量任务和接口调用这篇文章可以直接收藏。我们将覆盖 LangChain 的核心组件、RAG 知识库构建、Ollama 本地模型部署、Agent 开发实战以及如何将这些技术栈整合成可用的生产级应用。本文适合有一定 Python 基础希望快速上手大模型应用开发的读者。我们将从环境准备开始逐步演示每个环节的配置、代码实现和效果验证确保每一步都可复现。1. 核心能力速览能力项说明技术栈LangChain RAG Ollama Agent主要功能文档问答、知识检索、本地模型推理、智能体决策硬件需求CPU/GPU 均可GPU 推荐 8G 显存显存占用根据 Ollama 模型大小浮动7B 模型约 4-6G支持平台Windows/Linux/MacPython 3.8启动方式命令行启动 Web 服务API 支持是支持 RESTful 接口调用批量任务是支持文档批量处理和异步任务适合场景企业知识库、本地问答系统、智能助手开发2. 适用场景与使用边界LangChain 这套技术栈最适合需要处理私有文档、注重数据隐私、且希望控制推理成本的场景。比如企业内部知识库问答、学术资料检索、个人文档管理等都适用。但需要注意几个边界首先Ollama 部署的本地模型能力受限于模型大小7B 模型在复杂推理任务上可能不如云端大模型其次RAG 系统的效果高度依赖文档质量和检索精度最后Agent 开发需要清晰的任务边界避免陷入无限循环。在合规方面处理企业文档时务必确认数据授权涉及个人隐私的信息要做脱敏处理。如果用于生产环境建议先在小范围测试效果和稳定性。3. 环境准备与前置条件开始前需要确保开发环境满足以下要求操作系统Windows 10/11、LinuxUbuntu 20.04、macOS 12推荐使用 Linux 或 WSL2 以获得最佳兼容性Python 环境Python 3.8-3.113.12 可能存在兼容性问题建议使用 conda 或 venv 创建虚拟环境硬件要求内存16GB处理大文档时推荐 32GB存储至少 10GB 空闲空间用于模型和文档GPU可选但能显著提升推理速度NVIDIA 显卡推荐 8G 显存网络要求能正常访问 PyPI 和 GitHub下载依赖和模型如果网络受限需要提前配置镜像源检查环境是否就绪# 检查 Python 版本 python --version # 检查 pip 是否可用 pip --version # 检查 GPU 是否可用如果有 NVIDIA 显卡 nvidia-smi4. 安装部署与启动方式4.1 创建虚拟环境# 创建并激活虚拟环境 python -m venv langchain_env source langchain_env/bin/activate # Linux/macOS # 或 langchain_env\Scripts\activate # Windows4.2 安装核心依赖# 安装 LangChain 及相关组件 pip install langchain langchain-community langchain-core # 安装向量数据库以 Chroma 为例 pip install chromadb # 安装 Web 框架 pip install fastapi uvicorn # 安装 Ollama Python 客户端 pip install ollama4.3 安装 Ollama# Linux/macOS 安装 curl -fsSL https://ollama.ai/install.sh | sh # Windows 安装PowerShell winget install Ollama.Ollama # 启动 Ollama 服务 ollama serve4.4 下载模型# 下载常用的 7B 模型约 4GB ollama pull llama2:7b # 或下载更小的 3B 模型约 2GB ollama pull llama2:3b5. 功能测试与效果验证5.1 LangChain 基础功能测试先测试 LangChain 的基本链式操作from langchain.llms import Ollama from langchain.prompts import PromptTemplate from langchain.chains import LLMChain # 初始化 Ollama 本地模型 llm Ollama(modelllama2:7b) # 创建提示词模板 prompt PromptTemplate( input_variables[topic], template请用简单的话解释一下{topic}是什么 ) # 创建链 chain LLMChain(llmllm, promptprompt) # 测试链式调用 result chain.run(机器学习) print(回答:, result)预期结果模型应该能返回关于机器学习的基本解释。如果运行成功说明 LangChain 和 Ollama 的基础集成正常。5.2 RAG 知识库构建测试接下来测试 RAG 系统的文档加载和检索能力from langchain.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.embeddings import OllamaEmbeddings from langchain.vectorstores import Chroma # 1. 加载文档准备一个测试文档 example.txt loader TextLoader(example.txt) documents loader.load() # 2. 文档切分 text_splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50 ) texts text_splitter.split_documents(documents) # 3. 创建向量数据库 embeddings OllamaEmbeddings(modelllama2:7b) vectorstore Chroma.from_documents( documentstexts, embeddingembeddings, persist_directory./chroma_db ) # 4. 测试检索 retriever vectorstore.as_retriever() docs retriever.get_relevant_documents(文档中的关键主题) print(f检索到 {len(docs)} 个相关文档片段)成功标准能正常加载文档、完成切分、建立向量索引并能根据查询返回相关文档片段。5.3 Agent 智能体功能测试测试 Agent 的任务规划和工具调用能力from langchain.agents import initialize_agent, Tool from langchain.agents import AgentType from langchain.utilities import WikipediaAPIWrapper # 创建工具 wikipedia WikipediaAPIWrapper() tools [ Tool( nameWikipedia, funcwikipedia.run, description用于查询事实性信息 ) ] # 初始化 Agent agent initialize_agent( tools, llm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) # 测试 Agent result agent.run(请查一下人工智能的发展历史) print(Agent 回答:, result)成功标准Agent 应该能正确识别需要查询 Wikipedia并返回相关信息。6. 接口 API 与批量任务6.1 创建 FastAPI 服务将整个系统封装成 API 服务from fastapi import FastAPI from pydantic import BaseModel from langchain.chains import RetrievalQA app FastAPI() class QueryRequest(BaseModel): question: str top_k: int 3 # 初始化 RAG 系统 app.on_event(startup) async def startup_event(): global qa_chain # 这里复用之前创建的 vectorstore qa_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, retrievervectorstore.as_retriever() ) app.post(/query) async def query_knowledge(request: QueryRequest): result qa_chain.run(request.question) return {answer: result, top_k: request.top_k} app.post(/batch_query) async def batch_query(questions: list[str]): results [] for question in questions: result qa_chain.run(question) results.append({question: question, answer: result}) return {results: results}6.2 启动服务并测试# 启动服务 uvicorn main:app --host 0.0.0.0 --port 8000 --reload测试 API 调用# 单条查询测试 curl -X POST http://127.0.0.1:8000/query \ -H Content-Type: application/json \ -d {question: 什么是深度学习, top_k: 3} # 批量查询测试 curl -X POST http://127.0.0.1:8000/batch_query \ -H Content-Type: application/json \ -d [什么是机器学习, 神经网络如何工作]6.3 批量任务处理对于大量文档处理建议使用异步任务队列import asyncio from concurrent.futures import ThreadPoolExecutor class BatchProcessor: def __init__(self, max_workers3): self.executor ThreadPoolExecutor(max_workersmax_workers) async def process_batch(self, questions: list): loop asyncio.get_event_loop() tasks [] for question in questions: task loop.run_in_executor( self.executor, qa_chain.run, question ) tasks.append(task) results await asyncio.gather(*tasks) return results # 使用示例 processor BatchProcessor() questions [问题1, 问题2, 问题3, ...] # 大量问题 results asyncio.run(processor.process_batch(questions))7. 资源占用与性能观察7.1 显存占用监控使用 Ollama 时可以通过以下方式监控资源# 查看 Ollama 进程资源占用 ollama ps # 实时监控 GPU 使用情况如果有 NVIDIA 显卡 watch -n 1 nvidia-smi典型资源占用情况7B 模型推理时显存占用 4-6GB3B 模型推理时显存占用 2-3GBCPU 模式内存占用约为模型大小的 1.5 倍7.2 性能优化建议减少显存占用# 使用量化模型 llm Ollama(modelllama2:7b-q4) # 4-bit 量化 # 限制并发数 from langchain.callbacks.manager import CallbackManager from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler llm Ollama( modelllama2:7b, callback_managerCallbackManager([StreamingStdOutCallbackHandler()]), num_thread4 # 限制线程数 )提高检索速度# 使用更快的嵌入模型 embeddings OllamaEmbeddings(modelnomic-embed-text) # 优化检索参数 retriever vectorstore.as_retriever( search_typemmr, # 最大边际相关度 search_kwargs{k: 5, fetch_k: 10} )7.3 压力测试进行简单的压力测试观察系统稳定性import time import requests def stress_test(api_url, questions, concurrent3): import concurrent.futures def send_query(question): start time.time() response requests.post(api_url, json{question: question}) end time.time() return end - start, response.status_code with concurrent.futures.ThreadPoolExecutor(max_workersconcurrent) as executor: futures [executor.submit(send_query, q) for q in questions] results [f.result() for f in futures] avg_time sum(r[0] for r in results) / len(results) success_rate sum(1 for r in results if r[1] 200) / len(results) print(f平均响应时间: {avg_time:.2f}s) print(f成功率: {success_rate:.2%})8. 常见问题与排查方法问题现象可能原因排查方式解决方案Ollama 服务启动失败端口冲突或权限问题检查 11434 端口占用更换端口或重启服务模型下载缓慢网络连接问题检查网络状态使用国内镜像源显存不足模型太大或并发过多检查 nvidia-smi使用更小模型或量化版本检索效果差文档切分不合理检查 chunk_size 设置调整切分参数Agent 循环执行任务边界不清晰查看 verbose 日志设置最大执行步数API 调用超时推理时间过长检查模型响应时间增加超时设置8.1 详细排查步骤Ollama 服务问题# 检查服务状态 systemctl status ollama # Linux # 或 ollama serve # 直接启动查看日志 # 检查模型是否正常加载 ollama list向量数据库问题# 检查向量库是否正常创建 print(f向量库中文档数量: {vectorstore._collection.count()}) # 测试相似度搜索 test_embedding embeddings.embed_query(测试查询) similar_docs vectorstore.similarity_search_by_vector(test_embedding) print(f找到相似文档: {len(similar_docs)}个)LangChain 版本兼容性# 检查关键包版本 import langchain print(fLangChain: {langchain.__version__}) # 常见兼容性配置 from langchain_community.llms import Ollama # 新版本导入方式9. 最佳实践与使用建议9.1 项目结构规范建议按以下结构组织代码project/ ├── docs/ # 原始文档 ├── processed/ # 处理后的文档 ├── vector_db/ # 向量数据库 ├── src/ │ ├── config.py # 配置文件 │ ├── loader.py # 文档加载器 │ ├── rag.py # RAG 核心逻辑 │ └── api.py # API 服务 ├── tests/ # 测试用例 └── requirements.txt9.2 配置管理使用配置文件管理参数# config.py class Config: # 模型配置 MODEL_NAME llama2:7b EMBEDDING_MODEL nomic-embed-text # 向量数据库配置 CHROMA_PERSIST_DIR ./vector_db CHUNK_SIZE 500 CHUNK_OVERLAP 50 # API 配置 API_HOST 0.0.0.0 API_PORT 8000 # 性能配置 MAX_CONCURRENT 3 TIMEOUT 309.3 日志和监控添加完整的日志记录import logging from datetime import datetime # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(flogs/app_{datetime.now().strftime(%Y%m%d)}.log), logging.StreamHandler() ] ) logger logging.getLogger(__name__) # 在关键函数中添加日志 def query_with_logging(question: str): logger.info(f开始处理查询: {question}) start_time time.time() try: result qa_chain.run(question) elapsed time.time() - start_time logger.info(f查询完成耗时: {elapsed:.2f}s) return result except Exception as e: logger.error(f查询失败: {str(e)}) raise9.4 安全考虑API 安全from fastapi import Security, HTTPException from fastapi.security import APIKeyHeader api_key_header APIKeyHeader(nameX-API-Key) def verify_api_key(api_key: str Security(api_key_header)): if api_key ! your_secret_key: raise HTTPException(status_code403, detail无效的 API Key) return api_key app.post(/secure_query) async def secure_query(request: QueryRequest, api_key: str Security(verify_api_key)): # 安全验证后的处理逻辑 return await query_knowledge(request)数据隐私敏感文档处理前进行脱敏向量数据库加密存储定期清理临时文件API 访问日志不记录敏感信息10. 进阶扩展方向完成基础功能后可以考虑以下扩展10.1 多模态能力集成图像和文档处理from langchain.document_loaders import UnstructuredImageLoader from langchain.document_loaders import PyPDFLoader # 支持多种文档格式 loaders { .pdf: PyPDFLoader, .jpg: UnstructuredImageLoader, .png: UnstructuredImageLoader, .txt: TextLoader }10.2 缓存优化添加 Redis 缓存提升性能from langchain.cache import RedisCache import redis redis_client redis.Redis(hostlocalhost, port6379, db0) langchain.llm_cache RedisCache(redis_client)10.3 监控告警集成 Prometheus 监控from prometheus_client import Counter, Histogram, generate_latest # 定义指标 query_counter Counter(query_total, Total queries, [status]) query_duration Histogram(query_duration_seconds, Query duration) app.post(/monitored_query) query_duration.time() async def monitored_query(request: QueryRequest): try: result qa_chain.run(request.question) query_counter.labels(statussuccess).inc() return {answer: result} except Exception: query_counter.labels(statuserror).inc() raise这个完整的 LangChain 开发实战教程涵盖了从环境搭建到生产部署的全流程。最重要的是先确保基础流程能跑通然后再根据实际需求逐步优化扩展。建议按照文章顺序一步步实践每个环节都验证通过后再进入下一步。实际部署时可能会遇到环境差异问题重点观察日志输出大多数问题都能通过调整配置参数解决。如果需要在生产环境使用建议先进行充分的压力测试和效果评估。