LangChain 实战：构建生产级 LLM 应用的完整工程指南
LangChain 是当前最主流的 LLM 应用开发框架。本文不讲基础入门，直接聚焦生产级开发中的关键工程实践，包括性能优化、错误处理、可观测性和成本控制。
LangChain 是当前最主流的 LLM 应用开发框架。本文不讲基础入门，直接聚焦生产级开发中的关键工程实践，包括性能优化、错误处理、可观测性和成本控制。
LangChain 核心抽象├── Models（模型）│ ├── LLMs（文本补全）│ └── ChatModels（对话）├── Prompts（提示）│ ├── PromptTemplate│ └── ChatPromptTemplate├── Chains（链）│ ├── LLMChain（简单链）│ ├── SequentialChain（顺序链）│ └── LCEL（新一代声明式接口）├── Retrievers（检索器）│ └── VectorStoreRetriever 等├── Agents（智能体）│ ├── ReAct Agent│ └── OpenAI Functions Agent└── Memory（记忆） └── 对话历史管理—## 二、LCEL（LangChain 表达式语言）：生产推荐写法LCEL 是 LangChain 的现代化接口，相比旧版 Chain 有显著优势：pythonfrom langchain_openai import ChatOpenAIfrom langchain.prompts import ChatPromptTemplatefrom langchain.schema.output_parser import StrOutputParser# LCEL 声明式链构建（用管道符 | 连接）prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的{domain}专家"), ("human", "{question}")])model = ChatOpenAI(model="gpt-4o", temperature=0)output_parser = StrOutputParser()# 链式组合chain = prompt | model | output_parser# 同步调用result = chain.invoke({"domain": "数据科学", "question": "解释什么是梯度下降"})# 流式输出for chunk in chain.stream({"domain": "数据科学", "question": "解释什么是梯度下降"}): print(chunk, end="", flush=True)# 并发批处理results = chain.batch([ {"domain": "法律", "question": "合同无效的情形有哪些"}, {"domain": "医学", "question": "高血压的危险因素是什么"},])LCEL 的核心优势：- 支持流式输出- 支持异步/批量操作- 内置 LangSmith 追踪集成- 声明式语法，可读性强—## 三、生产级 RAG 实现### 完整 RAG Pipelinepythonfrom langchain_openai import OpenAIEmbeddings, ChatOpenAIfrom langchain_community.vectorstores import Qdrantfrom langchain.text_splitter import RecursiveCharacterTextSplitterfrom langchain.schema import Documentfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthrough# 1. 文档处理def load_and_split_documents(file_paths: list[str]) -> list[Document]: splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, separators=["\n\n", "\n", "。", "！", "？", " "] ) docs = [] for path in file_paths: # 根据文件类型使用不同 Loader loader = get_loader(path) docs.extend(loader.load()) return splitter.split_documents(docs)# 2. 建立向量数据库embeddings = OpenAIEmbeddings(model="text-embedding-3-small")chunks = load_and_split_documents(["./docs/knowledge_base.pdf"])vectorstore = Qdrant.from_documents( documents=chunks, embedding=embeddings, url="http://localhost:6333", collection_name="knowledge_base")# 3. 构建 RAG Chainretriever = vectorstore.as_retriever( search_type="mmr", # 最大边际相关性，减少重复 search_kwargs={"k": 5, "fetch_k": 20})rag_prompt = ChatPromptTemplate.from_template("""你是一个专业的知识库助手。请基于以下检索到的文档内容回答用户问题。如果文档内容不足以回答问题，请明确说明"根据当前知识库，无法回答此问题"，不要猜测。检索文档：{context}用户问题：{question}回答：""")llm = ChatOpenAI(model="gpt-4o", temperature=0)def format_docs(docs): return "\n\n".join([ f"[来源：{doc.metadata.get('source', '未知')}]\n{doc.page_content}" for doc in docs ])# LCEL RAG Chainrag_chain = ( {"context": retriever | format_docs, "question": RunnablePassthrough()} | rag_prompt | llm | StrOutputParser())# 调用answer = rag_chain.invoke("如何申请公司差旅报销？")—## 四、对话历史管理多轮对话是实际应用的标配，需要正确管理对话历史：pythonfrom langchain.memory import ConversationSummaryBufferMemoryfrom langchain_core.chat_history import BaseChatMessageHistoryfrom langchain_core.runnables.history import RunnableWithMessageHistoryfrom langchain_community.chat_message_histories import RedisChatMessageHistory# 生产环境：使用 Redis 持久化对话历史def get_session_history(session_id: str) -> BaseChatMessageHistory: return RedisChatMessageHistory( session_id=session_id, url="redis://localhost:6379" )# 创建带历史记录的 Chainprompt_with_history = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的客服助手"), ("placeholder", "{chat_history}"), # 历史记录插入位置 ("human", "{question}")])base_chain = prompt_with_history | llm | StrOutputParser()chain_with_history = RunnableWithMessageHistory( base_chain, get_session_history, input_messages_key="question", history_messages_key="chat_history")# 使用 session_id 区分不同用户response = chain_with_history.invoke( {"question": "你好，我想咨询退款政策"}, config={"configurable": {"session_id": "user_123"}})—## 五、错误处理与容错机制生产环境必须处理各种异常：pythonfrom langchain_core.runnables import RunnableWithFallbacksfrom langchain_openai import ChatOpenAIfrom tenacity import retry, stop_after_attempt, wait_exponential# 方案一：Fallback 降级链primary_llm = ChatOpenAI(model="gpt-4o")fallback_llm = ChatOpenAI(model="gpt-4o-mini") # 降级到更便宜的模型chain_with_fallback = (prompt | primary_llm).with_fallbacks( [prompt | fallback_llm])# 方案二：自定义重试逻辑@retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30))async def invoke_with_retry(chain, inputs): try: return await chain.ainvoke(inputs) except Exception as e: logger.warning(f"调用失败，准备重试: {e}") raise# 方案三：超时控制import asyncioasync def invoke_with_timeout(chain, inputs, timeout=30): try: return await asyncio.wait_for( chain.ainvoke(inputs), timeout=timeout ) except asyncio.TimeoutError: return "处理超时，请稍后重试"—## 六、可观测性：LangSmith 集成生产环境必须能追踪每次 LLM 调用：pythonimport os# 开启 LangSmith 追踪（仅需设置环境变量）os.environ["LANGCHAIN_TRACING_V2"] = "true"os.environ["LANGCHAIN_API_KEY"] = "your-api-key"os.environ["LANGCHAIN_PROJECT"] = "production-rag"# 之后所有 LangChain 调用都会自动追踪，包括：# - 输入/输出# - 延迟时间# - Token 消耗# - 错误信息# - 子链调用关系自定义追踪标签：pythonfrom langchain.callbacks.tracers import LangChainTracertracer = LangChainTracer(project_name="my-project")result = chain.invoke( {"question": "user query"}, config={ "callbacks": [tracer], "metadata": { "user_id": "user_123", "feature": "customer_service" } })—## 七、成本控制LLM API 成本管理是生产级应用的必要考量：pythonfrom langchain_community.callbacks import get_openai_callback# 精确统计每次调用的 Token 消耗with get_openai_callback() as cb: result = chain.invoke({"question": "your query"}) print(f"Prompt Tokens: {cb.prompt_tokens}") print(f"Completion Tokens: {cb.completion_tokens}") print(f"Total Cost: ${cb.total_cost:.4f}")成本优化策略：| 策略 | 实现方式 | 节省幅度 ||------|----------|----------|| 模型降级 | 简单任务用 GPT-4o-mini | 60-90% || 输出缓存 | 相同查询复用结果 | 20-40% || Prompt 压缩 | 精简系统提示词 | 10-30% || 批处理 | 合并多个小请求 | 15-25% || 流量分层 | 高价值用户用高端模型 | 整体 30-50% |—## 八、总结LangChain 生产级开发的核心要点：1.用 LCEL 代替旧版 Chain：更简洁，更灵活，流式原生支持2.会话历史用 Redis 持久化：不要用内存存储3.必须有 Fallback 机制：LLM 服务不是100%可靠的4.接入 LangSmith：没有可观测性，无法优化5.成本意识：Token 消耗要定期审计和优化—标签：LangChain | LCEL | RAG | LangSmith | 生产部署 | LLM应用开发