如果你正在学习 LangChain却感觉概念太多、无从下手如果你尝试搭建 AI 应用却总是被幻觉问题困扰如果你想知道 Agent 和 RAG 到底有什么区别以及它们在实际项目中如何配合使用——那么这篇文章就是为你准备的。2026 年的 AI 开发格局已经非常清晰单纯调用大模型 API 的时代已经过去真正有价值的应用都需要框架级的工程化能力。LangChain 作为目前最成熟的 AI 应用开发框架正在成为连接创意与落地的关键桥梁。但很多开发者反映LangChain 的学习曲线陡峭官方文档虽然全面但缺乏实战视角网上教程又往往只讲片段不讲体系。本文将从零开始通过完整的代码示例带你掌握 LangChain 的核心组件、Agent 智能体搭建和 RAG 知识库构建三大核心能力。不同于其他教程我们不会停留在概念层面而是通过真实的开发场景让你理解每个设计决策背后的原因以及在实际项目中如何避免常见陷阱。1. LangChain 真正解决了什么问题在深入代码之前我们需要先理解 LangChain 存在的价值。很多初学者误以为 LangChain 只是另一个大模型包装库但实际上它解决的是 AI 应用工程化的核心痛点。1.1 大模型原生能力的局限性当你直接使用 OpenAI API 或其他大模型服务时会遇到几个典型问题幻觉问题模型会自信地给出错误答案特别是在专业领域或私有数据相关的问题上。# 直接调用 API 的典型风险 response client.chat.completions.create( modelgpt-4, messages[{role: user, content: 我们公司的最新产品特性是什么}] ) # 模型可能会编造根本不存在的产品特性上下文限制即使是最新的大模型其上下文窗口也是有限的通常 128K-200K tokens无法处理大量文档。缺乏状态管理多轮对话中模型无法记住之前的交互历史需要开发者手动维护对话上下文。工具调用能力缺失原生模型无法直接操作数据库、调用 API 或执行代码。1.2 LangChain 的架构价值LangChain 通过模块化设计将 AI 应用开发分解为可组合的组件Models统一的多模型接口支持 OpenAI、Anthropic、本地模型等Prompts模板化提示词管理支持动态内容注入Indexes文档加载、分割、向量化检索Chains任务流水线组装实现复杂逻辑Agents工具调用和决策能力Memory对话状态持久化这种架构让开发者可以像搭积木一样构建应用而不是每次都从零开始。2. 环境准备与基础配置2.1 开发环境要求本文使用 Python 环境进行演示这是 LangChain 最成熟的生态系统。系统要求Python 3.8至少 8GB 内存用于运行本地模型和向量计算稳定的网络连接用于访问云端模型推荐开发工具VS Code 或 PyCharmJupyter Notebook用于实验性代码Git版本控制2.2 依赖安装创建新的项目目录并安装核心依赖# 创建项目目录 mkdir langchain-tutorial cd langchain-tutorial # 创建虚拟环境推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langchain langchain-community langchain-openai pip install python-dotenv # 环境变量管理2.3 API 密钥配置创建.env文件管理敏感信息# .env 文件内容 OPENAI_API_KEYsk-your-actual-api-key-here # 可选其他模型服务配置 ANTHROPIC_API_KEYyour-anthropic-key SERPAPI_API_KEYyour-serpapi-key # 用于搜索工具在代码中安全加载配置# config.py import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) MODEL_NAME gpt-4o-mini # 可根据需要调整 classmethod def validate_config(cls): if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY 未配置请在 .env 文件中设置)3. LangChain 核心组件深度解析3.1 模型层Models统一接口的价值LangChain 最大的优势之一是为不同模型提供商提供了统一接口# model_demo.py from langchain_openai import ChatOpenAI from langchain_community.llms import Ollama # 本地模型 from langchain.schema import HumanMessage def demonstrate_model_unification(): # 使用 OpenAI 模型 openai_llm ChatOpenAI( model_namegpt-4o-mini, temperature0.7 ) # 使用本地模型如通过 Ollama 部署 local_llm Ollama(modelllama3.1:8b) # 相同的调用方式 message HumanMessage(content请用一句话介绍人工智能) # 选择其中一个模型进行测试 response openai_llm.invoke([message]) print(fOpenAI 响应: {response.content}) # 本地模型调用需要先启动 Ollama 服务 # local_response local_llm.invoke([message]) # print(f本地模型响应: {local_response.content}) if __name__ __main__: demonstrate_model_unification()这种设计让开发者可以轻松切换模型而无需重写业务逻辑。3.2 提示词模板Prompt Templates提示词工程是 AI 应用的关键LangChain 提供了强大的模板管理# prompt_templates.py from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate from langchain.schema import SystemMessage def create_customer_service_template(): # 系统角色设定 system_template SystemMessage( content你是一个专业的客服助手回答要准确、友好、简洁。 ) # 用户消息模板 human_template HumanMessagePromptTemplate.from_template( 请回答以下用户问题参考已知信息\n已知信息{context}\n用户问题{question} ) # 组合成完整提示词 chat_prompt ChatPromptTemplate.from_messages([ system_template, human_template ]) # 使用模板 formatted_prompt chat_prompt.format_messages( context我们的营业时间是周一至周五 9:00-18:00, question你们周末营业吗 ) return formatted_prompt # 测试模板 prompt_messages create_customer_service_template() for msg in prompt_messages: print(f{msg.type}: {msg.content})3.3 链Chains任务流水线Chain 是 LangChain 的核心抽象将多个组件连接成完整的工作流# simple_chain.py from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI def create_sentiment_analysis_chain(): # 定义提示词模板 prompt PromptTemplate( input_variables[text], template分析以下文本的情感倾向正面/负面/中性并简要说明理由{text} ) # 创建链 llm ChatOpenAI(temperature0) chain LLMChain(llmllm, promptprompt) return chain def demonstrate_chain_usage(): chain create_sentiment_analysis_chain() test_texts [ 这个产品真是太棒了完全超出了我的预期, 服务态度很差再也不会来了。, 今天天气不错。 ] for text in test_texts: result chain.run(texttext) print(f文本: {text}) print(f分析: {result}\n) if __name__ __main__: demonstrate_chain_usage()4. RAG 知识库实战从零搭建企业级系统4.1 RAG 核心概念与架构RAGRetrieval-Augmented Generation是目前最实用的知识增强方案其核心流程文档处理加载、分割、向量化文档检索根据问题查找相关文档片段生成基于检索结果生成答案# rag_core_concepts.py import os from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma class SimpleRAGSystem: def __init__(self, data_directorydata): self.data_directory data_directory self.embeddings OpenAIEmbeddings() self.vector_store None def load_and_process_documents(self): 加载并处理文档 documents [] # 加载所有文本文件 for filename in os.listdir(self.data_directory): if filename.endswith(.txt): loader TextLoader(os.path.join(self.data_directory, filename)) documents.extend(loader.load()) # 文档分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) splits text_splitter.split_documents(documents) # 创建向量存储 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embeddings, persist_directory./chroma_db ) return len(splits) def query(self, question, k3): 查询知识库 if self.vector_store is None: raise ValueError(请先调用 load_and_process_documents() 初始化知识库) # 检索相关文档 docs self.vector_store.similarity_search(question, kk) # 构建上下文 context \n\n.join([doc.page_content for doc in docs]) return { context: context, source_documents: docs } # 使用示例 def demonstrate_rag_workflow(): rag SimpleRAGSystem() # 确保有数据目录和示例文件 os.makedirs(data, exist_okTrue) with open(data/company_info.txt, w, encodingutf-8) as f: f.write(公司成立于2020年专注于人工智能技术研发。 主要产品包括智能客服系统、文档分析工具和预测分析平台。 公司价值观客户第一、技术创新、团队合作。 总部位于北京在深圳、上海设有分公司。) document_count rag.load_and_process_documents() print(f处理了 {document_count} 个文档片段) result rag.query(公司的主要产品有哪些) print(检索到的上下文) print(result[context]) if __name__ __main__: demonstrate_rag_workflow()4.2 高级 RAG 特性实现基础 RAG 可以工作但生产环境需要更多优化# advanced_rag.py from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import EmbeddingsFilter from langchain.retrievers.ensemble import EnsembleRetriever from langchain_community.retrievers import BM25Retriever from langchain.retrievers import MultiQueryRetriever class AdvancedRAGSystem: def __init__(self, vector_store): self.vector_store vector_store self.embeddings OpenAIEmbeddings() def setup_hybrid_retrieval(self): 设置混合检索向量 关键词 # 向量检索器 vector_retriever self.vector_store.as_retriever(search_kwargs{k: 5}) # 关键词检索器需要文档文本 # 注意这里需要实际的文档内容简化演示 # bm25_retriever BM25Retriever.from_documents(documents) # bm25_retriever.k 5 # 组合检索器 # ensemble_retriever EnsembleRetriever( # retrievers[vector_retriever, bm25_retriever], # weights[0.5, 0.5] # ) return vector_retriever # 简化版本 def setup_compression_retriever(self, base_retriever): 设置重排序压缩 compressor EmbeddingsFilter( embeddingsself.embeddings, similarity_threshold0.76 ) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) return compression_retriever def multi_query_retrieval(self, question): 多查询检索提升召回率 base_retriever self.vector_store.as_retriever() multi_retriever MultiQueryRetriever.from_llm( retrieverbase_retriever, llmChatOpenAI(temperature0) ) return multi_retriever.get_relevant_documents(question)5. Agent 智能体开发让 AI 拥有行动能力5.1 Agent 基础概念Agent 与普通 Chain 的关键区别在于工具调用能力。Agent 可以决定何时使用何种工具来解决问题。# basic_agent.py from langchain.agents import initialize_agent, AgentType from langchain.agents import load_tools from langchain_openai import ChatOpenAI def create_calculator_agent(): 创建简单的计算器 Agent llm ChatOpenAI(temperature0, modelgpt-4o-mini) # 加载工具 tools load_tools([llm-math], llmllm) # 创建 Agent agent initialize_agent( tools, llm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue # 显示详细思考过程 ) return agent def demonstrate_agent_capabilities(): agent create_calculator_agent() # 测试数学问题 questions [ 123的平方是多少, 1000除以37的结果保留两位小数, 圆周率π的前10位数字是多少 ] for question in questions: print(f\n问题: {question}) try: result agent.run(question) print(f答案: {result}) except Exception as e: print(f错误: {e}) if __name__ __main__: demonstrate_agent_capabilities()5.2 自定义工具开发真正的 Agent 能力来自于自定义工具# custom_tools.py from langchain.agents import tool from datetime import datetime import requests tool def get_current_time(timezone: str Asia/Shanghai) - str: 获取指定时区的当前时间 now datetime.now() if timezone Asia/Shanghai: return now.strftime(%Y-%m-%d %H:%M:%S 北京时间) else: return f{now.strftime(%Y-%m-%d %H:%M:%S)} 时区: {timezone} tool def get_weather(city: str) - str: 获取指定城市的天气信息模拟 # 实际项目中这里会调用天气API weather_data { 北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C, 杭州: 阴16°C } return weather_data.get(city, f未找到{city}的天气信息) tool def calculate_bmi(weight: float, height: float) - dict: 计算BMI指数 height_in_meters height / 100 # 厘米转米 bmi weight / (height_in_meters ** 2) if bmi 18.5: category 偏瘦 elif bmi 24: category 正常 elif bmi 28: category 超重 else: category 肥胖 return { bmi: round(bmi, 1), category: category, advice: f建议{category}请注意保持健康生活方式 } def create_custom_agent(): 创建带有自定义工具的Agent llm ChatOpenAI(temperature0, modelgpt-4o-mini) custom_tools [get_current_time, get_weather, calculate_bmi] agent initialize_agent( custom_tools, llm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) return agent # 测试自定义工具 def test_custom_tools(): agent create_custom_agent() test_queries [ 现在北京的时间是多少, 上海的天气怎么样, 我的体重70公斤身高175厘米BMI是多少, 先获取杭州的天气然后告诉我现在的时间 ] for query in test_queries: print(f\n{*50}) print(f查询: {query}) print(*50) try: result agent.run(query) print(f结果: {result}) except Exception as e: print(f执行错误: {e}) if __name__ __main__: test_custom_tools()6. 完整项目实战智能技术文档助手现在我们将所有组件组合起来构建一个完整的技术文档问答系统# tech_doc_assistant.py import os from langchain import hub from langchain.agents import AgentExecutor, create_react_agent from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader from langchain_community.vectorstores import FAISS from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import ChatOpenAI, OpenAIEmbeddings class TechDocAssistant: def __init__(self, docs_directory./technical_docs): self.docs_directory docs_directory self.vector_store None self.retriever None self.agent None def setup_knowledge_base(self): 设置知识库 if not os.path.exists(self.docs_directory): os.makedirs(self.docs_directory) print(f请将技术文档放入 {self.docs_directory} 目录) return False # 加载文档 loader DirectoryLoader( self.docs_directory, glob**/*.pdf, loader_clsPyPDFLoader ) documents loader.load() if not documents: print(未找到PDF文档) return False # 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) splits text_splitter.split_documents(documents) # 创建向量存储 embeddings OpenAIEmbeddings() self.vector_store FAISS.from_documents(splits, embeddings) self.retriever self.vector_store.as_retriever(search_kwargs{k: 4}) print(f知识库初始化完成共处理 {len(splits)} 个文档片段) return True def create_retrieval_tool(self): 创建检索工具 from langchain.agents import tool tool def search_technical_docs(query: str) - str: 在技术文档中搜索相关信息 docs self.retriever.get_relevant_documents(query) if not docs: return 未找到相关技术文档 context \n\n.join([f来源: {doc.metadata.get(source, 未知)}\n内容: {doc.page_content} for doc in docs]) return f找到 {len(docs)} 个相关文档片段\n\n{context} return search_technical_docs def setup_assistant(self): 设置助手 if not self.setup_knowledge_base(): return False # 创建工具 retrieval_tool self.create_retrieval_tool() tools [retrieval_tool] # 获取prompt模板 prompt hub.pull(hwchase17/react-chat) # 创建LLM llm ChatOpenAI( modelgpt-4o-mini, temperature0.3 ) # 创建agent self.agent create_react_agent(llm, tools, prompt) self.agent_executor AgentExecutor( agentself.agent, toolstools, verboseTrue, handle_parsing_errorsTrue ) return True def chat(self, question): 与助手对话 if not self.agent_executor: return 助手未初始化请先调用 setup_assistant() try: response self.agent_executor.invoke({input: question}) return response[output] except Exception as e: return f对话过程中出现错误: {str(e)} # 使用示例 def demo_tech_assistant(): assistant TechDocAssistant() if assistant.setup_assistant(): questions [ 系统架构的主要组件有哪些, 如何配置数据库连接, 安全认证是如何实现的 ] for question in questions: print(f\n用户: {question}) response assistant.chat(question) print(f助手: {response}) if __name__ __main__: demo_tech_assistant()7. 性能优化与生产环境部署7.1 缓存策略优化# optimization.py from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache import hashlib def setup_caching(strategysqlite): 设置缓存策略 if strategy sqlite: cache SQLiteCache(database_path./.langchain.db) else: cache InMemoryCache() set_llm_cache(cache) print(f已启用 {strategy} 缓存) def content_based_cache_key(*args, **kwargs): 基于内容的自定义缓存键 key_parts [] # 处理prompt内容 for arg in args: if hasattr(arg, content): key_parts.append(arg.content) else: key_parts.append(str(arg)) # 处理关键字参数 for k, v in kwargs.items(): key_parts.append(f{k}{v}) # 生成哈希键 content |.join(key_parts) return hashlib.md5(content.encode()).hexdigest() # 使用缓存示例 setup_caching(sqlite)7.2 错误处理与重试机制# error_handling.py from tenacity import retry, stop_after_attempt, wait_exponential from langchain.schema import OutputParserException class RobustLangChainService: def __init__(self): self.llm ChatOpenAI( modelgpt-4o-mini, temperature0, max_retries3 ) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) def robust_invoke(self, prompt): 带重试的调用 try: return self.llm.invoke(prompt) except Exception as e: print(f调用失败: {e}, 进行重试...) raise def safe_parse_response(self, response, parser): 安全解析响应 try: return parser.parse(response.content) except OutputParserException as e: print(f解析失败: {e}) # 返回默认值或进行修复 return {error: 解析失败, original_response: response.content}8. 常见问题与解决方案8.1 安装与配置问题问题现象可能原因解决方案导入 LangChain 报错版本冲突或依赖缺失使用虚拟环境确保安装正确版本的包API 密钥错误环境变量未正确设置检查 .env 文件格式和路径内存不足文档过大或向量维度太高使用更小的嵌入模型或分批处理8.2 运行时问题问题现象可能原因解决方案检索结果不相关文档分割策略不当调整 chunk_size 和 chunk_overlapAgent 陷入循环工具设计或提示词问题添加最大迭代次数限制响应速度慢网络延迟或模型负载启用缓存使用更轻量模型8.3 性能优化问题问题现象可能原因解决方案内存使用过高向量数据库未持久化使用外部向量数据库响应时间波动模型API不稳定实现重试机制和降级策略9. 最佳实践与工程建议9.1 项目结构规范my-langchain-project/ ├── src/ │ ├── agents/ # Agent 定义 │ ├── chains/ # Chain 定义 │ ├── tools/ # 自定义工具 │ ├── prompts/ # 提示词模板 │ └── utils/ # 工具函数 ├── data/ # 文档数据 ├── tests/ # 测试用例 ├── config/ # 配置文件 └── requirements.txt # 依赖管理9.2 提示词工程规范# prompt_best_practices.py def create_effective_prompt_template(): 创建高效的提示词模板 template 请根据以下上下文信息回答问题。请遵循以下规则 1. 如果上下文包含答案请基于上下文回答 2. 如果上下文不包含相关信息请明确说明根据现有信息无法回答 3. 保持回答专业、准确、简洁 4. 如果问题涉及多个方面请分点回答 上下文信息 {context} 问题{question} 请回答 return template9.3 监控与日志# monitoring.py import logging from datetime import datetime def setup_logging(): 设置结构化日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(flangchain_app_{datetime.now().strftime(%Y%m%d)}.log), logging.StreamHandler() ] ) class UsageTracker: 使用情况跟踪 def __init__(self): self.usage_data [] def track_invocation(self, component, input_size, output_size, duration): 跟踪调用信息 record { timestamp: datetime.now(), component: component, input_size: input_size, output_size: output_size, duration: duration } self.usage_data.append(record)通过本文的完整学习路径你应该已经掌握了 LangChain 的核心概念和实战技能。从基础组件到高级 Agent 开发从简单的 RAG 系统到完整的生产级应用这些知识将帮助你在实际项目中构建可靠的 AI 应用。记住LangChain 的真正价值不在于框架本身而在于它提供的工程化思维和最佳实践。在实际项目中始终从用户需求出发选择合适的组件组合并持续优化系统性能。
LangChain实战：从Agent智能体到RAG知识库的AI应用开发指南
发布时间：2026/7/14 2:43:16
如果你正在学习 LangChain却感觉概念太多、无从下手如果你尝试搭建 AI 应用却总是被幻觉问题困扰如果你想知道 Agent 和 RAG 到底有什么区别以及它们在实际项目中如何配合使用——那么这篇文章就是为你准备的。2026 年的 AI 开发格局已经非常清晰单纯调用大模型 API 的时代已经过去真正有价值的应用都需要框架级的工程化能力。LangChain 作为目前最成熟的 AI 应用开发框架正在成为连接创意与落地的关键桥梁。但很多开发者反映LangChain 的学习曲线陡峭官方文档虽然全面但缺乏实战视角网上教程又往往只讲片段不讲体系。本文将从零开始通过完整的代码示例带你掌握 LangChain 的核心组件、Agent 智能体搭建和 RAG 知识库构建三大核心能力。不同于其他教程我们不会停留在概念层面而是通过真实的开发场景让你理解每个设计决策背后的原因以及在实际项目中如何避免常见陷阱。1. LangChain 真正解决了什么问题在深入代码之前我们需要先理解 LangChain 存在的价值。很多初学者误以为 LangChain 只是另一个大模型包装库但实际上它解决的是 AI 应用工程化的核心痛点。1.1 大模型原生能力的局限性当你直接使用 OpenAI API 或其他大模型服务时会遇到几个典型问题幻觉问题模型会自信地给出错误答案特别是在专业领域或私有数据相关的问题上。# 直接调用 API 的典型风险 response client.chat.completions.create( modelgpt-4, messages[{role: user, content: 我们公司的最新产品特性是什么}] ) # 模型可能会编造根本不存在的产品特性上下文限制即使是最新的大模型其上下文窗口也是有限的通常 128K-200K tokens无法处理大量文档。缺乏状态管理多轮对话中模型无法记住之前的交互历史需要开发者手动维护对话上下文。工具调用能力缺失原生模型无法直接操作数据库、调用 API 或执行代码。1.2 LangChain 的架构价值LangChain 通过模块化设计将 AI 应用开发分解为可组合的组件Models统一的多模型接口支持 OpenAI、Anthropic、本地模型等Prompts模板化提示词管理支持动态内容注入Indexes文档加载、分割、向量化检索Chains任务流水线组装实现复杂逻辑Agents工具调用和决策能力Memory对话状态持久化这种架构让开发者可以像搭积木一样构建应用而不是每次都从零开始。2. 环境准备与基础配置2.1 开发环境要求本文使用 Python 环境进行演示这是 LangChain 最成熟的生态系统。系统要求Python 3.8至少 8GB 内存用于运行本地模型和向量计算稳定的网络连接用于访问云端模型推荐开发工具VS Code 或 PyCharmJupyter Notebook用于实验性代码Git版本控制2.2 依赖安装创建新的项目目录并安装核心依赖# 创建项目目录 mkdir langchain-tutorial cd langchain-tutorial # 创建虚拟环境推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langchain langchain-community langchain-openai pip install python-dotenv # 环境变量管理2.3 API 密钥配置创建.env文件管理敏感信息# .env 文件内容 OPENAI_API_KEYsk-your-actual-api-key-here # 可选其他模型服务配置 ANTHROPIC_API_KEYyour-anthropic-key SERPAPI_API_KEYyour-serpapi-key # 用于搜索工具在代码中安全加载配置# config.py import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) MODEL_NAME gpt-4o-mini # 可根据需要调整 classmethod def validate_config(cls): if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY 未配置请在 .env 文件中设置)3. LangChain 核心组件深度解析3.1 模型层Models统一接口的价值LangChain 最大的优势之一是为不同模型提供商提供了统一接口# model_demo.py from langchain_openai import ChatOpenAI from langchain_community.llms import Ollama # 本地模型 from langchain.schema import HumanMessage def demonstrate_model_unification(): # 使用 OpenAI 模型 openai_llm ChatOpenAI( model_namegpt-4o-mini, temperature0.7 ) # 使用本地模型如通过 Ollama 部署 local_llm Ollama(modelllama3.1:8b) # 相同的调用方式 message HumanMessage(content请用一句话介绍人工智能) # 选择其中一个模型进行测试 response openai_llm.invoke([message]) print(fOpenAI 响应: {response.content}) # 本地模型调用需要先启动 Ollama 服务 # local_response local_llm.invoke([message]) # print(f本地模型响应: {local_response.content}) if __name__ __main__: demonstrate_model_unification()这种设计让开发者可以轻松切换模型而无需重写业务逻辑。3.2 提示词模板Prompt Templates提示词工程是 AI 应用的关键LangChain 提供了强大的模板管理# prompt_templates.py from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate from langchain.schema import SystemMessage def create_customer_service_template(): # 系统角色设定 system_template SystemMessage( content你是一个专业的客服助手回答要准确、友好、简洁。 ) # 用户消息模板 human_template HumanMessagePromptTemplate.from_template( 请回答以下用户问题参考已知信息\n已知信息{context}\n用户问题{question} ) # 组合成完整提示词 chat_prompt ChatPromptTemplate.from_messages([ system_template, human_template ]) # 使用模板 formatted_prompt chat_prompt.format_messages( context我们的营业时间是周一至周五 9:00-18:00, question你们周末营业吗 ) return formatted_prompt # 测试模板 prompt_messages create_customer_service_template() for msg in prompt_messages: print(f{msg.type}: {msg.content})3.3 链Chains任务流水线Chain 是 LangChain 的核心抽象将多个组件连接成完整的工作流# simple_chain.py from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI def create_sentiment_analysis_chain(): # 定义提示词模板 prompt PromptTemplate( input_variables[text], template分析以下文本的情感倾向正面/负面/中性并简要说明理由{text} ) # 创建链 llm ChatOpenAI(temperature0) chain LLMChain(llmllm, promptprompt) return chain def demonstrate_chain_usage(): chain create_sentiment_analysis_chain() test_texts [ 这个产品真是太棒了完全超出了我的预期, 服务态度很差再也不会来了。, 今天天气不错。 ] for text in test_texts: result chain.run(texttext) print(f文本: {text}) print(f分析: {result}\n) if __name__ __main__: demonstrate_chain_usage()4. RAG 知识库实战从零搭建企业级系统4.1 RAG 核心概念与架构RAGRetrieval-Augmented Generation是目前最实用的知识增强方案其核心流程文档处理加载、分割、向量化文档检索根据问题查找相关文档片段生成基于检索结果生成答案# rag_core_concepts.py import os from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma class SimpleRAGSystem: def __init__(self, data_directorydata): self.data_directory data_directory self.embeddings OpenAIEmbeddings() self.vector_store None def load_and_process_documents(self): 加载并处理文档 documents [] # 加载所有文本文件 for filename in os.listdir(self.data_directory): if filename.endswith(.txt): loader TextLoader(os.path.join(self.data_directory, filename)) documents.extend(loader.load()) # 文档分割 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) splits text_splitter.split_documents(documents) # 创建向量存储 self.vector_store Chroma.from_documents( documentssplits, embeddingself.embeddings, persist_directory./chroma_db ) return len(splits) def query(self, question, k3): 查询知识库 if self.vector_store is None: raise ValueError(请先调用 load_and_process_documents() 初始化知识库) # 检索相关文档 docs self.vector_store.similarity_search(question, kk) # 构建上下文 context \n\n.join([doc.page_content for doc in docs]) return { context: context, source_documents: docs } # 使用示例 def demonstrate_rag_workflow(): rag SimpleRAGSystem() # 确保有数据目录和示例文件 os.makedirs(data, exist_okTrue) with open(data/company_info.txt, w, encodingutf-8) as f: f.write(公司成立于2020年专注于人工智能技术研发。 主要产品包括智能客服系统、文档分析工具和预测分析平台。 公司价值观客户第一、技术创新、团队合作。 总部位于北京在深圳、上海设有分公司。) document_count rag.load_and_process_documents() print(f处理了 {document_count} 个文档片段) result rag.query(公司的主要产品有哪些) print(检索到的上下文) print(result[context]) if __name__ __main__: demonstrate_rag_workflow()4.2 高级 RAG 特性实现基础 RAG 可以工作但生产环境需要更多优化# advanced_rag.py from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import EmbeddingsFilter from langchain.retrievers.ensemble import EnsembleRetriever from langchain_community.retrievers import BM25Retriever from langchain.retrievers import MultiQueryRetriever class AdvancedRAGSystem: def __init__(self, vector_store): self.vector_store vector_store self.embeddings OpenAIEmbeddings() def setup_hybrid_retrieval(self): 设置混合检索向量 关键词 # 向量检索器 vector_retriever self.vector_store.as_retriever(search_kwargs{k: 5}) # 关键词检索器需要文档文本 # 注意这里需要实际的文档内容简化演示 # bm25_retriever BM25Retriever.from_documents(documents) # bm25_retriever.k 5 # 组合检索器 # ensemble_retriever EnsembleRetriever( # retrievers[vector_retriever, bm25_retriever], # weights[0.5, 0.5] # ) return vector_retriever # 简化版本 def setup_compression_retriever(self, base_retriever): 设置重排序压缩 compressor EmbeddingsFilter( embeddingsself.embeddings, similarity_threshold0.76 ) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverbase_retriever ) return compression_retriever def multi_query_retrieval(self, question): 多查询检索提升召回率 base_retriever self.vector_store.as_retriever() multi_retriever MultiQueryRetriever.from_llm( retrieverbase_retriever, llmChatOpenAI(temperature0) ) return multi_retriever.get_relevant_documents(question)5. Agent 智能体开发让 AI 拥有行动能力5.1 Agent 基础概念Agent 与普通 Chain 的关键区别在于工具调用能力。Agent 可以决定何时使用何种工具来解决问题。# basic_agent.py from langchain.agents import initialize_agent, AgentType from langchain.agents import load_tools from langchain_openai import ChatOpenAI def create_calculator_agent(): 创建简单的计算器 Agent llm ChatOpenAI(temperature0, modelgpt-4o-mini) # 加载工具 tools load_tools([llm-math], llmllm) # 创建 Agent agent initialize_agent( tools, llm, agentAgentType.ZERO_SHOT_REACT_DESCRIPTION, verboseTrue # 显示详细思考过程 ) return agent def demonstrate_agent_capabilities(): agent create_calculator_agent() # 测试数学问题 questions [ 123的平方是多少, 1000除以37的结果保留两位小数, 圆周率π的前10位数字是多少 ] for question in questions: print(f\n问题: {question}) try: result agent.run(question) print(f答案: {result}) except Exception as e: print(f错误: {e}) if __name__ __main__: demonstrate_agent_capabilities()5.2 自定义工具开发真正的 Agent 能力来自于自定义工具# custom_tools.py from langchain.agents import tool from datetime import datetime import requests tool def get_current_time(timezone: str Asia/Shanghai) - str: 获取指定时区的当前时间 now datetime.now() if timezone Asia/Shanghai: return now.strftime(%Y-%m-%d %H:%M:%S 北京时间) else: return f{now.strftime(%Y-%m-%d %H:%M:%S)} 时区: {timezone} tool def get_weather(city: str) - str: 获取指定城市的天气信息模拟 # 实际项目中这里会调用天气API weather_data { 北京: 晴15°C, 上海: 多云18°C, 深圳: 阵雨22°C, 杭州: 阴16°C } return weather_data.get(city, f未找到{city}的天气信息) tool def calculate_bmi(weight: float, height: float) - dict: 计算BMI指数 height_in_meters height / 100 # 厘米转米 bmi weight / (height_in_meters ** 2) if bmi 18.5: category 偏瘦 elif bmi 24: category 正常 elif bmi 28: category 超重 else: category 肥胖 return { bmi: round(bmi, 1), category: category, advice: f建议{category}请注意保持健康生活方式 } def create_custom_agent(): 创建带有自定义工具的Agent llm ChatOpenAI(temperature0, modelgpt-4o-mini) custom_tools [get_current_time, get_weather, calculate_bmi] agent initialize_agent( custom_tools, llm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue ) return agent # 测试自定义工具 def test_custom_tools(): agent create_custom_agent() test_queries [ 现在北京的时间是多少, 上海的天气怎么样, 我的体重70公斤身高175厘米BMI是多少, 先获取杭州的天气然后告诉我现在的时间 ] for query in test_queries: print(f\n{*50}) print(f查询: {query}) print(*50) try: result agent.run(query) print(f结果: {result}) except Exception as e: print(f执行错误: {e}) if __name__ __main__: test_custom_tools()6. 完整项目实战智能技术文档助手现在我们将所有组件组合起来构建一个完整的技术文档问答系统# tech_doc_assistant.py import os from langchain import hub from langchain.agents import AgentExecutor, create_react_agent from langchain_community.document_loaders import DirectoryLoader, PyPDFLoader from langchain_community.vectorstores import FAISS from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_openai import ChatOpenAI, OpenAIEmbeddings class TechDocAssistant: def __init__(self, docs_directory./technical_docs): self.docs_directory docs_directory self.vector_store None self.retriever None self.agent None def setup_knowledge_base(self): 设置知识库 if not os.path.exists(self.docs_directory): os.makedirs(self.docs_directory) print(f请将技术文档放入 {self.docs_directory} 目录) return False # 加载文档 loader DirectoryLoader( self.docs_directory, glob**/*.pdf, loader_clsPyPDFLoader ) documents loader.load() if not documents: print(未找到PDF文档) return False # 分割文档 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) splits text_splitter.split_documents(documents) # 创建向量存储 embeddings OpenAIEmbeddings() self.vector_store FAISS.from_documents(splits, embeddings) self.retriever self.vector_store.as_retriever(search_kwargs{k: 4}) print(f知识库初始化完成共处理 {len(splits)} 个文档片段) return True def create_retrieval_tool(self): 创建检索工具 from langchain.agents import tool tool def search_technical_docs(query: str) - str: 在技术文档中搜索相关信息 docs self.retriever.get_relevant_documents(query) if not docs: return 未找到相关技术文档 context \n\n.join([f来源: {doc.metadata.get(source, 未知)}\n内容: {doc.page_content} for doc in docs]) return f找到 {len(docs)} 个相关文档片段\n\n{context} return search_technical_docs def setup_assistant(self): 设置助手 if not self.setup_knowledge_base(): return False # 创建工具 retrieval_tool self.create_retrieval_tool() tools [retrieval_tool] # 获取prompt模板 prompt hub.pull(hwchase17/react-chat) # 创建LLM llm ChatOpenAI( modelgpt-4o-mini, temperature0.3 ) # 创建agent self.agent create_react_agent(llm, tools, prompt) self.agent_executor AgentExecutor( agentself.agent, toolstools, verboseTrue, handle_parsing_errorsTrue ) return True def chat(self, question): 与助手对话 if not self.agent_executor: return 助手未初始化请先调用 setup_assistant() try: response self.agent_executor.invoke({input: question}) return response[output] except Exception as e: return f对话过程中出现错误: {str(e)} # 使用示例 def demo_tech_assistant(): assistant TechDocAssistant() if assistant.setup_assistant(): questions [ 系统架构的主要组件有哪些, 如何配置数据库连接, 安全认证是如何实现的 ] for question in questions: print(f\n用户: {question}) response assistant.chat(question) print(f助手: {response}) if __name__ __main__: demo_tech_assistant()7. 性能优化与生产环境部署7.1 缓存策略优化# optimization.py from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache, SQLiteCache import hashlib def setup_caching(strategysqlite): 设置缓存策略 if strategy sqlite: cache SQLiteCache(database_path./.langchain.db) else: cache InMemoryCache() set_llm_cache(cache) print(f已启用 {strategy} 缓存) def content_based_cache_key(*args, **kwargs): 基于内容的自定义缓存键 key_parts [] # 处理prompt内容 for arg in args: if hasattr(arg, content): key_parts.append(arg.content) else: key_parts.append(str(arg)) # 处理关键字参数 for k, v in kwargs.items(): key_parts.append(f{k}{v}) # 生成哈希键 content |.join(key_parts) return hashlib.md5(content.encode()).hexdigest() # 使用缓存示例 setup_caching(sqlite)7.2 错误处理与重试机制# error_handling.py from tenacity import retry, stop_after_attempt, wait_exponential from langchain.schema import OutputParserException class RobustLangChainService: def __init__(self): self.llm ChatOpenAI( modelgpt-4o-mini, temperature0, max_retries3 ) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) def robust_invoke(self, prompt): 带重试的调用 try: return self.llm.invoke(prompt) except Exception as e: print(f调用失败: {e}, 进行重试...) raise def safe_parse_response(self, response, parser): 安全解析响应 try: return parser.parse(response.content) except OutputParserException as e: print(f解析失败: {e}) # 返回默认值或进行修复 return {error: 解析失败, original_response: response.content}8. 常见问题与解决方案8.1 安装与配置问题问题现象可能原因解决方案导入 LangChain 报错版本冲突或依赖缺失使用虚拟环境确保安装正确版本的包API 密钥错误环境变量未正确设置检查 .env 文件格式和路径内存不足文档过大或向量维度太高使用更小的嵌入模型或分批处理8.2 运行时问题问题现象可能原因解决方案检索结果不相关文档分割策略不当调整 chunk_size 和 chunk_overlapAgent 陷入循环工具设计或提示词问题添加最大迭代次数限制响应速度慢网络延迟或模型负载启用缓存使用更轻量模型8.3 性能优化问题问题现象可能原因解决方案内存使用过高向量数据库未持久化使用外部向量数据库响应时间波动模型API不稳定实现重试机制和降级策略9. 最佳实践与工程建议9.1 项目结构规范my-langchain-project/ ├── src/ │ ├── agents/ # Agent 定义 │ ├── chains/ # Chain 定义 │ ├── tools/ # 自定义工具 │ ├── prompts/ # 提示词模板 │ └── utils/ # 工具函数 ├── data/ # 文档数据 ├── tests/ # 测试用例 ├── config/ # 配置文件 └── requirements.txt # 依赖管理9.2 提示词工程规范# prompt_best_practices.py def create_effective_prompt_template(): 创建高效的提示词模板 template 请根据以下上下文信息回答问题。请遵循以下规则 1. 如果上下文包含答案请基于上下文回答 2. 如果上下文不包含相关信息请明确说明根据现有信息无法回答 3. 保持回答专业、准确、简洁 4. 如果问题涉及多个方面请分点回答 上下文信息 {context} 问题{question} 请回答 return template9.3 监控与日志# monitoring.py import logging from datetime import datetime def setup_logging(): 设置结构化日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(flangchain_app_{datetime.now().strftime(%Y%m%d)}.log), logging.StreamHandler() ] ) class UsageTracker: 使用情况跟踪 def __init__(self): self.usage_data [] def track_invocation(self, component, input_size, output_size, duration): 跟踪调用信息 record { timestamp: datetime.now(), component: component, input_size: input_size, output_size: output_size, duration: duration } self.usage_data.append(record)通过本文的完整学习路径你应该已经掌握了 LangChain 的核心概念和实战技能。从基础组件到高级 Agent 开发从简单的 RAG 系统到完整的生产级应用这些知识将帮助你在实际项目中构建可靠的 AI 应用。记住LangChain 的真正价值不在于框架本身而在于它提供的工程化思维和最佳实践。在实际项目中始终从用户需求出发选择合适的组件组合并持续优化系统性能。
)
