在实际 AI 应用开发中直接调用大模型 API 往往只能解决简单问答。真正要把 AI 能力集成到业务系统里你需要处理上下文管理、工具调用、知识检索、流程控制等一系列工程问题。LangChain 正是为此而生的框架它把大模型应用开发中的常见模式抽象成可复用的组件。本文基于 LangChain 1.3 版本从基础概念到进阶实战完整演示如何构建具备记忆、工具使用和知识检索能力的 AI 应用。适合有一定 Python 基础希望系统掌握 LangChain 开发模式的开发者。1. 理解 LangChain 的核心设计理念LangChain 不是另一个大模型而是连接大模型与实际应用的“胶水层”。它的核心价值在于提供了标准化的接口和组件让开发者能快速构建可维护的 AI 应用。1.1 为什么需要 LangChain直接使用大模型 API 会遇到几个典型问题上下文长度限制当对话或文档超过模型限制时需要自己处理文本切分、摘要和上下文选择。工具集成困难要让模型能查询数据库、调用 API 或执行代码需要设计复杂的交互协议。状态管理复杂多轮对话中需要维护对话历史、用户状态和会话数据。知识更新滞后模型训练数据有截止日期无法直接获取最新信息。LangChain 通过模块化设计解决了这些问题。它的核心组件包括Models统一接口调用不同的大模型OpenAI、通义千问、本地模型等。Prompts模板化提示词管理支持变量注入和少量示例。Chains将多个组件串联成执行流程。Agents让模型自主选择工具完成复杂任务。Memory管理对话历史和状态。Indexes处理文档加载、切分、检索和向量化。1.2 LangChain 1.3 的关键更新1.3 版本在稳定性和功能完整性上有显著提升更清晰的模块划分将社区贡献组件分离到langchain-community包核心框架更轻量。改进的 Agent 执行器提供更可靠的错误处理和状态管理。增强的 RAG 支持优化检索器接口和向量存储集成。更好的类型提示提升开发时的代码补全和错误检测能力。对于新项目建议直接使用 1.3.x 版本。配套的langchain-community版本需要与核心包匹配一般安装最新版本即可pip install langchain1.3.11 langchain-community0.3.6如果遇到版本冲突可以先尝试安装最新版本再根据错误信息调整。2. 环境准备与基础配置开始前需要准备 Python 环境和大模型访问权限。本文将使用 OpenAI GPT-4 作为示例模型但 LangChain 支持多种模型提供商。2.1 环境要求与依赖安装确保 Python 版本 ≥ 3.8然后安装核心依赖# 基础包 pip install langchain1.3.11 langchain-community0.3.6 # 可选但常用的扩展 pip install openai tiktoken chromadb pypdf python-dotenv # 如果使用通义千问等国内模型 pip install dashscope创建项目目录结构langchain-project/ ├── config/ │ └── .env ├── data/ │ └── documents/ ├── src/ │ ├── chains/ │ ├── agents/ │ └── rag/ └── tests/2.2 模型配置与密钥管理在config/.env中配置模型密钥OPENAI_API_KEYsk-your-openai-key DASHSCOPE_API_KEYyour-dashscope-key在代码中安全加载配置from dotenv import load_dotenv import os load_dotenv(config/.env) # 配置 OpenAI 模型 from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4, temperature0.7, api_keyos.getenv(OPENAI_API_KEY) ) # 或者配置通义千问 from langchain_community.llms import Tongyi llm_tongyi Tongyi( modelqwen-max, dashscope_api_keyos.getenv(DASHSCOPE_API_KEY) )关键参数说明temperature控制输出随机性0-1值越大回答越多样。max_tokens限制单次响应长度。model指定模型版本不同版本能力和价格差异很大。注意生产环境不要将密钥硬编码在代码中。使用环境变量或专业的密钥管理服务。3. Prompt 提示词工程实战Prompt 是与大模型交互的核心。好的提示词能显著提升模型输出质量而糟糕的提示词会导致无关或错误的回答。3.1 基础 Prompt 模板直接拼接字符串的方式难以维护# 不推荐硬编码提示词 user_query 什么是机器学习 prompt f请用中文简单解释一下{user_query}使用 LangChain 的PromptTemplatefrom langchain.prompts import PromptTemplate # 创建可复用的模板 template 你是一个专业的AI助手请用简洁的中文回答用户问题。 问题{question} 回答 prompt_template PromptTemplate( input_variables[question], templatetemplate ) # 使用模板 formatted_prompt prompt_template.format(question什么是机器学习) response llm.invoke(formatted_prompt) print(response.content)3.2 少量示例学习Few-shot Learning对于复杂任务提供示例能帮助模型理解期望的输出格式from langchain.prompts import FewShotPromptTemplate # 定义示例 examples [ { input: 这个产品的价格是多少, output: 我需要查询产品数据库来获取最新价格信息。 }, { input: 最近的销售数据怎么样, output: 我可以帮您生成销售报表请告诉我需要哪个时间段的數據。 } ] # 创建示例模板 example_template 用户{input} 助手{output} example_prompt PromptTemplate( input_variables[input, output], templateexample_template ) # 创建少量示例提示词 few_shot_prompt FewShotPromptTemplate( examplesexamples, example_promptexample_prompt, prefix你是一个客户服务助手根据用户问题判断是否需要查询外部系统。, suffix用户{input}\n助手, input_variables[input], example_separator\n\n ) # 使用 result few_shot_prompt.format(input库存情况如何) response llm.invoke(result)3.3 常见 Prompt 错误与调试错误1提示词验证失败ValueError: Prompt outputs failed validation: checkpointloadersimple: - value not in list这通常是因为模板变量与输入不匹配。检查input_variables是否正确定义# 错误模板中有 {name}但 input_variables 未包含 template Hello {name} prompt PromptTemplate(input_variables[], templatetemplate) # 会报错 # 正确匹配所有变量 prompt PromptTemplate(input_variables[name], templatetemplate)错误2系统消息位置错误API Error: 400 failed to build prompt: system message must be at the beginning在使用聊天模型时系统消息必须在对话开始from langchain.schema import SystemMessage, HumanMessage # 错误系统消息不在开头 messages [ HumanMessage(你好), SystemMessage(你是一个助手) # 这会报错 ] # 正确系统消息优先 messages [ SystemMessage(你是一个专业的AI助手), HumanMessage(请解释机器学习) ]错误3提示词无输出当提示词过于模糊或矛盾时模型可能无法生成有效输出。确保提示词任务要求明确具体输出格式有清晰指示没有相互矛盾的指令4. Chain 链式调用实战Chain 是 LangChain 的核心抽象它将多个组件连接成可复用的工作流。4.1 基础 LLMChain最简单的链将提示词模板与 LLM 连接from langchain.chains import LLMChain # 创建链 llm_chain LLMChain( llmllm, promptprompt_template ) # 执行链 result llm_chain.invoke({question: Python 的优缺点是什么}) print(result[text])4.2 顺序链SequentialChain处理多个步骤的任务前一个步骤的输出作为后一个步骤的输入from langchain.chains import SimpleSequentialChain # 第一步生成文章大纲 outline_template 为以下主题生成文章大纲 主题{topic} 大纲 outline_prompt PromptTemplate( input_variables[topic], templateoutline_template ) outline_chain LLMChain(llmllm, promptoutline_prompt) # 第二步根据大纲写文章 article_template 根据以下大纲写一篇详细文章 大纲{outline} 文章 article_prompt PromptTemplate( input_variables[outline], templatearticle_template ) article_chain LLMChain(llmllm, promptarticle_prompt) # 连接两个链 overall_chain SimpleSequentialChain( chains[outline_chain, article_chain], verboseTrue # 显示执行过程 ) result overall_chain.invoke(人工智能在教育领域的应用)4.3 路由链RouterChain根据输入内容选择不同的处理分支from langchain.chains import RouterChain from langchain.chains.llm import LLMChain # 定义不同专业的提示词 physics_template 你是一个物理专家用专业术语回答物理问题 问题{input} 回答 math_template 你是一个数学专家专注于数学问题的解决 问题{input} 回答 general_template 你是一个通用助手回答一般性问题 问题{input} 回答 # 创建多个链 physics_chain LLMChain( llmllm, promptPromptTemplate.from_template(physics_template) ) math_chain LLMChain( llmllm, promptPromptTemplate.from_template(math_template) ) general_chain LLMChain( llmllm, promptPromptTemplate.from_template(general_template) ) # 在实际项目中需要使用 MultiRouteChain 或 LLMRouterChain # 这里简化演示概念 def route_question(question): if 物理 in question or 力学 in question: return physics_chain elif 数学 in question or 计算 in question: return math_chain else: return general_chain # 使用 question 解释牛顿第二定律 chain route_question(question) result chain.invoke({input: question})5. Agent 智能体开发实战Agent 是 LangChain 最强大的功能之一它让大模型能够自主使用工具完成任务。5.1 Agent 核心概念Agent LLM 工具 决策逻辑LLM负责思考和分析工具外部能力接口搜索、计算、数据库等决策逻辑ReAct 等框架指导模型如何思考和使用工具5.2 基础 Agent 实现首先定义工具函数from langchain.agents import tool import math from datetime import datetime tool def calculate_circle_area(radius: float) - float: 计算圆的面积输入半径返回面积 return math.pi * radius * radius tool def get_current_time() - str: 获取当前日期和时间 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) tool def search_wikipedia(query: str) - str: 搜索维基百科摘要简化版实际需要API # 实际项目中这里调用维基百科API return f关于{query}的搜索结果这是模拟的搜索结果内容。 # 工具列表 tools [calculate_circle_area, get_current_time, search_wikipedia]创建 Agentfrom langchain.agents import initialize_agent, AgentType from langchain.schema import SystemMessage # 创建带有系统消息的LLM system_message SystemMessage( content你是一个有帮助的助手可以使用工具解决问题。 使用工具时请清晰说明你的思考过程。 如果不需要工具就能直接回答请直接回答。 ) # 初始化Agent agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verboseTrue, handle_parsing_errorsTrue # 处理解析错误 )测试 Agent# 简单问题直接回答 result1 agent.invoke(什么是人工智能) print(result1[output]) # 需要工具的问题 result2 agent.invoke(计算半径为5的圆的面积) print(result2[output]) result3 agent.invoke(现在是什么时间顺便搜索一下机器学习) print(result3[output])5.3 处理复杂任务ReAct 模式ReActReasoning Acting是 Agent 的核心模式让模型先推理再行动from langchain import hub from langchain.agents import AgentExecutor, create_react_agent # 从LangChain Hub获取优化过的ReAct提示词 react_prompt hub.pull(hwchase17/react) # 创建ReAct Agent react_agent create_react_agent(llm, tools, react_prompt) agent_executor AgentExecutor( agentreact_agent, toolstools, verboseTrue, max_iterations5 # 限制最大迭代次数防止无限循环 ) # 执行复杂任务 complex_task 首先获取当前时间然后计算半径为10的圆面积 最后搜索一下圆周率的历史发展。 请按步骤执行并总结结果。 result agent_executor.invoke({input: complex_task})5.4 Agent 常见问题排查问题1工具调用失败现象Agent 反复尝试同一个工具但失败。排查步骤检查工具函数参数类型是否匹配验证工具函数本身是否能正常工作查看 verbose 日志确认模型是否正确解析了工具输入问题2无限循环现象Agent 在不同工具间来回切换无法完成任务。解决方案设置max_iterations限制最大尝试次数在系统消息中明确任务边界提供更清晰的示例演示何时应该停止问题3工具选择错误现象Agent 选择了不合适的工具处理任务。改进方法优化工具描述使其更准确具体在提示词中提供工具选择示例使用更先进的 Agent 类型如OPENAI_FUNCTIONS6. RAG 检索增强生成实战RAGRetrieval-Augmented Generation通过检索外部知识来增强模型回答解决模型知识陈旧和幻觉问题。6.1 RAG 工作流程文档加载从各种来源加载文档PDF、网页、数据库等文本切分将长文档切分成适合检索的片段向量化将文本转换为向量表示检索根据查询找到最相关的文本片段生成将检索结果作为上下文生成最终回答6.2 构建企业知识库 RAG 系统6.2.1 文档加载与处理from langchain_community.document_loaders import PyPDFLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 加载文档 def load_documents(directory_path): documents [] for file_path in os.listdir(directory_path): full_path os.path.join(directory_path, file_path) if file_path.endswith(.pdf): loader PyPDFLoader(full_path) elif file_path.endswith(.txt): loader TextLoader(full_path) else: continue documents.extend(loader.load()) return documents # 加载并切分文档 documents load_documents(data/documents/) text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 每个片段约1000字符 chunk_overlap200, # 片段间重叠200字符 length_functionlen ) chunks text_splitter.split_documents(documents) print(f共切分得到 {len(chunks)} 个文本片段)6.2.2 向量存储与检索from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings # 初始化嵌入模型 embeddings OpenAIEmbeddings( modeltext-embedding-3-small, api_keyos.getenv(OPENAI_API_KEY) ) # 创建向量数据库 vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db ) # 创建检索器 retriever vectorstore.as_retriever( search_typesimilarity, search_kwargs{k: 3} # 返回最相关的3个片段 )6.2.3 构建 RAG 链from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate # 自定义提示词模板 rag_prompt_template 使用以下上下文信息回答用户问题。 如果你不知道答案就说不知道不要编造信息。 上下文 {context} 问题{question} 回答 rag_prompt PromptTemplate( templaterag_prompt_template, input_variables[context, question] ) # 创建RAG链 rag_chain RetrievalQA.from_chain_type( llmllm, chain_typestuff, # 简单地将所有上下文 stuffed 到提示词中 retrieverretriever, chain_type_kwargs{prompt: rag_prompt}, return_source_documentsTrue ) # 测试RAG系统 question 我们公司的最新产品政策是什么 result rag_chain.invoke({query: question}) print(答案, result[result]) print(\n来源文档) for doc in result[source_documents][:2]: # 显示前2个来源 print(f- {doc.metadata.get(source, 未知)}: {doc.page_content[:200]}...)6.3 RAG 性能优化技巧6.3.1 改进检索质量多向量检索同时使用多种检索方式提升召回率from langchain.retrievers import BM25Retriever, EnsembleRetriever # BM25检索器关键词匹配 from langchain_community.retrievers import BM25Retriever bm25_retriever BM25Retriever.from_documents(chunks) bm25_retriever.k 2 # 向量检索器语义匹配 vector_retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 集成检索器 ensemble_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.4, 0.6] # 权重可调整 )重排序对检索结果进行二次排序from langchain_community.document_transformers import LongContextReorder reorder LongContextReorder() reordered_docs reorder.transform_documents(retrieved_docs)6.3.2 处理长文档挑战层次化检索先检索章节再检索具体内容# 第一层章节级检索大块 chapter_splitter RecursiveCharacterTextSplitter( chunk_size5000, chunk_overlap500 ) chapter_chunks chapter_splitter.split_documents(documents) # 第二层段落级检索小块 paragraph_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) # 建立层次化检索逻辑 def hierarchical_retrieval(query, top_k_chapters2, top_k_paragraphs3): # 先找相关章节 chapter_results vectorstore.similarity_search(query, ktop_k_chapters) # 在每个相关章节中找具体段落 all_paragraphs [] for chapter in chapter_results: paragraphs paragraph_splitter.split_documents([chapter]) all_paragraphs.extend(paragraphs) # 对段落进行向量化检索 paragraph_vectorstore Chroma.from_documents( all_paragraphs, embeddings ) final_results paragraph_vectorstore.similarity_search(query, ktop_k_paragraphs) return final_results7. 生产环境最佳实践7.1 性能优化缓存机制减少重复的LLM调用from langchain.globals import set_llm_cache from langchain.cache import InMemoryCache # 内存缓存开发环境 set_llm_cache(InMemoryCache()) # Redis缓存生产环境 from langchain.cache import RedisCache import redis redis_client redis.Redis(hostlocalhost, port6379, db0) set_llm_cache(RedisCache(redis_client))异步处理提高并发性能import asyncio async def process_questions_async(questions): tasks [rag_chain.ainvoke({query: q}) for q in questions] results await asyncio.gather(*tasks) return results # 使用 questions [问题1, 问题2, 问题3] results asyncio.run(process_questions_async(questions))7.2 监控与日志结构化日志import logging import json from datetime import datetime def setup_logging(): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(logs/application.log), logging.StreamHandler() ] ) def log_rag_interaction(question, answer, sources, latency): interaction_log { timestamp: datetime.now().isoformat(), question: question, answer_length: len(answer), sources_count: len(sources), latency_ms: latency, sources: [str(source.metadata) for source in sources] } logging.info(fRAG Interaction: {json.dumps(interaction_log)})性能监控import time from functools import wraps def monitor_latency(func): wraps(func) def wrapper(*args, **kwargs): start_time time.time() result func(*args, **kwargs) latency (time.time() - start_time) * 1000 # 毫秒 # 记录延迟 if hasattr(result, get) and query in kwargs: log_rag_interaction( kwargs[query], result.get(result, ), result.get(source_documents, []), latency ) return result return wrapper # 装饰RAG链 rag_chain.invoke monitor_latency(rag_chain.invoke)7.3 安全考虑输入验证import re def validate_input(query: str) - bool: 验证用户输入的安全性 # 检查长度 if len(query) 1000: return False # 检查潜在恶意模式 malicious_patterns [ r\.\./, # 路径遍历 r;\s*(DROP|DELETE|INSERT), # SQL注入 rscript, # XSS ] for pattern in malicious_patterns: if re.search(pattern, query, re.IGNORECASE): return False return True def safe_rag_invoke(query: str): if not validate_input(query): return {result: 输入验证失败请重新输入问题} return rag_chain.invoke({query: query})内容过滤def content_filter(text: str) - bool: 简单的内容过滤 sensitive_keywords [敏感词1, 敏感词2] # 实际项目中使用更复杂的列表 for keyword in sensitive_keywords: if keyword in text: return False return True def filtered_rag_invoke(query: str): result rag_chain.invoke({query: query}) if not content_filter(result[result]): result[result] 根据内容策略无法回答该问题 return result8. 常见问题深度排查8.1 LangChain 版本兼容性问题症状导入错误或运行时异常提示缺少模块或属性。排查步骤检查版本匹配pip list | grep langchain确保langchain和langchain-community版本兼容。查看官方文档的版本说明确认使用的类或函数在当前版本中可用。如果从旧版本迁移注意导入路径变化# 旧版本0.x from langchain.llms import OpenAI # 新版本1.x from langchain_openai import ChatOpenAI8.2 Agent 工具调用失败症状Agent 反复尝试工具但失败或错误选择工具。解决方案验证工具函数独立性# 单独测试工具 result calculate_circle_area(5) print(f工具测试结果: {result})检查工具描述是否清晰tool def search_database(query: str) - str: 搜索产品数据库输入产品名称或ID返回库存和价格信息 # 实现...使用更详细的Agent类型agent initialize_agent( toolstools, llmllm, agentAgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, # 提供结构化思考 verboseTrue )8.3 RAG 检索效果不佳症状检索到不相关文档或遗漏关键信息。优化方法调整文本切分策略text_splitter RecursiveCharacterTextSplitter( chunk_size800, # 减小块大小 chunk_overlap150, separators[\n\n, \n, 。, , , ] # 中文友好分隔符 )改进检索参数retriever vectorstore.as_retriever( search_typemmr, # 最大边际相关性平衡相关性和多样性 search_kwargs{k: 5, lambda_mult: 0.7} )添加查询扩展from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor compressor LLMChainExtractor.from_llm(llm) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrieverretriever )8.4 内存使用过多症状处理长文档或大量对话时内存快速增长。优化策略使用更高效的向量数据库# 使用FAISS替代Chroma更节省内存 from langchain_community.vectorstores import FAISS vectorstore FAISS.from_documents(chunks, embeddings)实现对话历史摘要from langchain.memory import ConversationSummaryMemory memory ConversationSummaryMemory( llmllm, return_messagesTrue, memory_keychat_history )分批处理大型文档def process_large_document_in_batches(documents, batch_size50): for i in range(0, len(documents), batch_size): batch documents[i:ibatch_size] # 处理批次 vectorstore.add_documents(batch)实际项目中LangChain 应用的稳定性既取决于框架的正确使用也依赖于对业务场景的深入理解。建议从简单用例开始逐步增加复杂度在每个阶段都建立完整的测试和监控机制。
