企业官网建设流程全解析

在实际 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 langchain==1.3.11 langchain-community==0.3.6

如果遇到版本冲突，可以先尝试安装最新版本，再根据错误信息调整。

2. 环境准备与基础配置

开始前需要准备 Python 环境和大模型访问权限。本文将使用 OpenAI GPT-4 作为示例模型，但 LangChain 支持多种模型提供商。

2.1 环境要求与依赖安装

确保 Python 版本 ≥ 3.8，然后安装核心依赖：

# 基础包 pip install langchain==1.3.11 langchain-community==0.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_KEY=sk-your-openai-key DASHSCOPE_API_KEY=your-dashscope-key

在代码中安全加载配置：

from dotenv import load_dotenv import os load_dotenv('config/.env') # 配置 OpenAI 模型 from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4", temperature=0.7, api_key=os.getenv("OPENAI_API_KEY") ) # 或者配置通义千问 from langchain_community.llms import Tongyi llm_tongyi = Tongyi( model="qwen-max", dashscope_api_key=os.getenv("DASHSCOPE_API_KEY") )

关键参数说明：

• temperature：控制输出随机性（0-1），值越大回答越多样。
• max_tokens：限制单次响应长度。
• model：指定模型版本，不同版本能力和价格差异很大。

注意：生产环境不要将密钥硬编码在代码中。使用环境变量或专业的密钥管理服务。

3. Prompt 提示词工程实战

Prompt 是与大模型交互的核心。好的提示词能显著提升模型输出质量，而糟糕的提示词会导致无关或错误的回答。

3.1 基础 Prompt 模板

直接拼接字符串的方式难以维护：

# 不推荐：硬编码提示词 user_query = "什么是机器学习" prompt = f"请用中文简单解释一下：{user_query}"

使用 LangChain 的PromptTemplate：

from langchain.prompts import PromptTemplate # 创建可复用的模板 template = """你是一个专业的AI助手，请用简洁的中文回答用户问题。 问题：{question} 回答：""" prompt_template = PromptTemplate( input_variables=["question"], template=template ) # 使用模板 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"], template=example_template ) # 创建少量示例提示词 few_shot_prompt = FewShotPromptTemplate( examples=examples, example_prompt=example_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=[], template=template) # 会报错 # 正确：匹配所有变量 prompt = PromptTemplate(input_variables=["name"], template=template)

错误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( llm=llm, prompt=prompt_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"], template=outline_template ) outline_chain = LLMChain(llm=llm, prompt=outline_prompt) # 第二步：根据大纲写文章 article_template = """根据以下大纲写一篇详细文章： 大纲：{outline} 文章：""" article_prompt = PromptTemplate( input_variables=["outline"], template=article_template ) article_chain = LLMChain(llm=llm, prompt=article_prompt) # 连接两个链 overall_chain = SimpleSequentialChain( chains=[outline_chain, article_chain], verbose=True # 显示执行过程 ) 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( llm=llm, prompt=PromptTemplate.from_template(physics_template) ) math_chain = LLMChain( llm=llm, prompt=PromptTemplate.from_template(math_template) ) general_chain = LLMChain( llm=llm, prompt=PromptTemplate.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]

创建 Agent：

from langchain.agents import initialize_agent, AgentType from langchain.schema import SystemMessage # 创建带有系统消息的LLM system_message = SystemMessage( content="""你是一个有帮助的助手，可以使用工具解决问题。 使用工具时请清晰说明你的思考过程。 如果不需要工具就能直接回答，请直接回答。""" ) # 初始化Agent agent = initialize_agent( tools=tools, llm=llm, agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, verbose=True, handle_parsing_errors=True # 处理解析错误 )

测试 Agent：

# 简单问题：直接回答 result1 = agent.invoke("什么是人工智能？") print(result1["output"]) # 需要工具的问题 result2 = agent.invoke("计算半径为5的圆的面积") print(result2["output"]) result3 = agent.invoke("现在是什么时间？顺便搜索一下机器学习") print(result3["output"])

5.3 处理复杂任务：ReAct 模式

ReAct（Reasoning + 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( agent=react_agent, tools=tools, verbose=True, max_iterations=5 # 限制最大迭代次数，防止无限循环 ) # 执行复杂任务 complex_task = """ 首先获取当前时间，然后计算半径为10的圆面积， 最后搜索一下圆周率的历史发展。 请按步骤执行并总结结果。 """ result = agent_executor.invoke({"input": complex_task})

5.4 Agent 常见问题排查

问题1：工具调用失败

现象：Agent 反复尝试同一个工具但失败。

排查步骤：

1. 检查工具函数参数类型是否匹配
2. 验证工具函数本身是否能正常工作
3. 查看 verbose 日志，确认模型是否正确解析了工具输入

问题2：无限循环

现象：Agent 在不同工具间来回切换，无法完成任务。

解决方案：

• 设置max_iterations限制最大尝试次数
• 在系统消息中明确任务边界
• 提供更清晰的示例演示何时应该停止

问题3：工具选择错误

现象：Agent 选择了不合适的工具处理任务。

改进方法：

• 优化工具描述，使其更准确具体
• 在提示词中提供工具选择示例
• 使用更先进的 Agent 类型（如OPENAI_FUNCTIONS）

6. RAG 检索增强生成实战

RAG（Retrieval-Augmented Generation）通过检索外部知识来增强模型回答，解决模型知识陈旧和幻觉问题。

6.1 RAG 工作流程

1. 文档加载：从各种来源加载文档（PDF、网页、数据库等）
2. 文本切分：将长文档切分成适合检索的片段
3. 向量化：将文本转换为向量表示
4. 检索：根据查询找到最相关的文本片段
5. 生成：将检索结果作为上下文，生成最终回答

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_size=1000, # 每个片段约1000字符 chunk_overlap=200, # 片段间重叠200字符 length_function=len ) 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( model="text-embedding-3-small", api_key=os.getenv("OPENAI_API_KEY") ) # 创建向量数据库 vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" ) # 创建检索器 retriever = vectorstore.as_retriever( search_type="similarity", 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( template=rag_prompt_template, input_variables=["context", "question"] ) # 创建RAG链 rag_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 简单地将所有上下文 stuffed 到提示词中 retriever=retriever, chain_type_kwargs={"prompt": rag_prompt}, return_source_documents=True ) # 测试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_size=5000, chunk_overlap=500 ) chapter_chunks = chapter_splitter.split_documents(documents) # 第二层：段落级检索（小块） paragraph_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200 ) # 建立层次化检索逻辑 def hierarchical_retrieval(query, top_k_chapters=2, top_k_paragraphs=3): # 先找相关章节 chapter_results = vectorstore.similarity_search(query, k=top_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, k=top_k_paragraphs) return final_results

7. 生产环境最佳实践

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(host='localhost', port=6379, db=0) 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( level=logging.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(f"RAG 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注入 r"<script>", # 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 result

8. 常见问题深度排查

8.1 LangChain 版本兼容性问题

症状：导入错误或运行时异常，提示缺少模块或属性。

排查步骤：

1. 检查版本匹配：

pip list | grep langchain

确保langchain和langchain-community版本兼容。

2. 查看官方文档的版本说明，确认使用的类或函数在当前版本中可用。

3. 如果从旧版本迁移，注意导入路径变化：

# 旧版本（0.x） from langchain.llms import OpenAI # 新版本（1.x） from langchain_openai import ChatOpenAI

8.2 Agent 工具调用失败

症状：Agent 反复尝试工具但失败，或错误选择工具。

解决方案：

1. 验证工具函数独立性：

# 单独测试工具 result = calculate_circle_area(5) print(f"工具测试结果: {result}")

2. 检查工具描述是否清晰：

@tool def search_database(query: str) -> str: """搜索产品数据库：输入产品名称或ID，返回库存和价格信息""" # 实现...

3. 使用更详细的Agent类型：

agent = initialize_agent( tools=tools, llm=llm, agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION, # 提供结构化思考 verbose=True )

8.3 RAG 检索效果不佳

症状：检索到不相关文档，或遗漏关键信息。

优化方法：

1. 调整文本切分策略：

text_splitter = RecursiveCharacterTextSplitter( chunk_size=800, # 减小块大小 chunk_overlap=150, separators=["\n\n", "\n", "。", "！", "？", "．"] # 中文友好分隔符 )

2. 改进检索参数：

retriever = vectorstore.as_retriever( search_type="mmr", # 最大边际相关性，平衡相关性和多样性 search_kwargs={"k": 5, "lambda_mult": 0.7} )

3. 添加查询扩展：

from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=retriever )

8.4 内存使用过多

症状：处理长文档或大量对话时内存快速增长。

优化策略：

1. 使用更高效的向量数据库：

# 使用FAISS替代Chroma（更节省内存） from langchain_community.vectorstores import FAISS vectorstore = FAISS.from_documents(chunks, embeddings)

2. 实现对话历史摘要：

from langchain.memory import ConversationSummaryMemory memory = ConversationSummaryMemory( llm=llm, return_messages=True, memory_key="chat_history" )

3. 分批处理大型文档：

def process_large_document_in_batches(documents, batch_size=50): for i in range(0, len(documents), batch_size): batch = documents[i:i+batch_size] # 处理批次 vectorstore.add_documents(batch)

实际项目中，LangChain 应用的稳定性既取决于框架的正确使用，也依赖于对业务场景的深入理解。建议从简单用例开始，逐步增加复杂度，在每个阶段都建立完整的测试和监控机制。